-
Notifications
You must be signed in to change notification settings - Fork 7
/
otpprofiler.py
executable file
·378 lines (338 loc) · 13.8 KB
/
otpprofiler.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
#!/usr/bin/python
import urllib2, time, itertools, json
import subprocess, urllib, random
import simplejson
import pprint
from copy import copy
# python-requests no longer has first-class support for concurrent asynchronous HTTP requests
# the author has moved it to https://github.com/kennethreitz/grequests
# python-requests wraps urllib2 providing a much nicer API.
import grequests
DATE = '2014-12-30'
# split out base and specific endpoint
SHOW_PARAMS = False
SHOW_URL = False
SHOW_RESPONSE = False
# globals to store accumulated responses. one file for summaries, one file for full itineraries.
response_json = []
full_itins_json = []
pp = pprint.PrettyPrinter(indent=4)
n = 0 # number of responses received
N = 0 # total number of responses expected
t0 = 0 # time that search begins
def cycle(seq):
"A generator that loops over a sequence forever."
# Strangely, there seems to be no library function to do this.
# https://docs.python.org/3.4/library/itertools.html#itertools.cycle
# has to make a copy of every element in the iterator to obey the iterator interface.
while True:
for elem in seq:
yield elem
def pairs(iterable):
"A generator that takes items from a sequence two at a time. (s0, s1), (s2, s3), (s4, s5), ..."
it = iter(iterable)
for x in it:
yield (x, next(it, None))
def get_params(fast):
requests_json = simplejson.load(open("requests.json"))
requests = requests_json['requests']
endpoints = requests_json['endpoints']
# Make sure there is an even number of endpoints
if len(endpoints) % 2 != 0 :
endpoints = endpoints[:-1]
ret = []
# if fast :
# else :
for (request, (origin, target)) in zip(cycle(requests), pairs(endpoints)):
req = copy(request)
req['oid'] = origin['id']
req['tid'] = target['id']
req['fromPlace'] = "%s,%s"%(origin['lat'],origin['lon'])
req['toPlace'] = "%s,%s"%(target['lat'],target['lon'])
ret.append( req )
# TODO yield
return ret
def getServerInfo(host):
"""Get information about the server that is being profiled. Returns a tuple of:
(sha1, version, cpuName, nCores)
where sha1 is the hash of the HEAD commit, version is the output of 'git describe' (which
includes the last tag and how many commits have been made on top of that tag),
cpuName is the model of the cpu as reported by /proc/cpuinfo via the OTP serverinfo API,
and nCores is the number of (logical) cores reported by that same API, including hyperthreading.
"""
url_meta = "http://"+host+"/otp/"
try :
print "grabbing metadata from %s"%url_meta
req = urllib2.Request(url_meta)
req.add_header('Accept', 'application/json')
response = urllib2.urlopen(req)
if response.code != 200 :
print "Server metadata response was not 200"
return None
content = response.read()
objs = json.loads(content)
print "Server metadata response: ", objs
version = objs['serverVersion']['version']
sha1 = objs['serverVersion']['commit']
# would also be nice to have build date
cpuName = objs['cpuName']
nCores = objs['nCores']
except urllib2.URLError :
# This is a normal condition while waiting for server to come up, trap exception and return None.
# Any problem with decoding the server response will still blow up the program with an exception.
print "Error requesting metadata from OTP server. Is it running?"
return None
print "sha1 of commit is:", sha1
print "version of OTP is:", version
print "processor type is:", cpuName
print "number of logical cores is:", nCores
return (sha1, version, cpuName, nCores)
# summarize the result for an itinerary planner request
def summarize_plan (itinerary) :
routes = []
trips = []
waits = []
leg_modes = []
leg_times = []
n_vehicles = 0
n_legs = 0
for leg in itinerary['legs'] :
n_legs += 1
leg_modes.append(leg['mode'])
leg_times.append((leg['endTime'] - leg['startTime'])/1000)
if 'route' in leg and len(leg['route']) > 0 :
routes.append(leg['route'])
trips.append(leg['tripId'])
n_vehicles += 1
wait = (leg['from']['departure'] - leg['from']['arrival'])/1000
#print(' - wait = %d'%(wait))
waits.append(wait)
ret = {
'start_time' : time.asctime(time.gmtime(itinerary['startTime'] / 1000)) + ' GMT',
'duration' : '%d msec' % int(itinerary['duration']),
'n_legs' : n_legs,
'n_vehicles' : n_vehicles,
'walk_distance' : itinerary['walkDistance'],
'wait_time_sec' : itinerary['waitingTime'],
'ride_time_sec' : itinerary['transitTime'],
'routes' : routes,
'trips' : trips,
'waits' : waits,
'leg_modes' : leg_modes,
'leg_times' : leg_times
}
return ret
# summarize the result for a profile request
def summarize_profile (option) :
routes = []
trips = []
waits = []
leg_modes = []
leg_times = []
n_vehicles = 0
n_legs = 0
wait_time_sec = 0
ride_time_sec = 0
if 'transit' in option:
if 'access' in option:
n_legs += 1
summarize_profile_non_transit_leg(option['access'], leg_modes, leg_times)
for transit_leg in option['transit']:
n_legs += 1
n_vehicles += 1
leg_modes.append(transit_leg['mode'])
avg_wait_time = transit_leg['waitStats']['avg']
waits.append(avg_wait_time)
wait_time_sec += avg_wait_time
avg_ride_time = transit_leg['rideStats']['avg']
leg_times.append(avg_ride_time)
ride_time_sec += avg_ride_time
if len(transit_leg['routes']) == 1:
routes.append(transit_leg['routes'][0]['id'])
else:
route_ids = []
for route in transit_leg['routes']:
route_ids.append(route['id'])
routes.append(route_ids)
if 'egress' in option:
n_legs += 1
summarize_profile_non_transit_leg(option['egress'], leg_modes, leg_times)
else:
n_legs = 1
summarize_profile_non_transit_leg(option['access'], leg_modes, leg_times)
ret = {
'n_legs' : n_legs,
'n_vehicles' : n_vehicles,
'wait_time_sec' : wait_time_sec,
'ride_time_sec' : ride_time_sec,
'routes' : routes,
'waits' : waits,
'leg_modes' : leg_modes,
'leg_times' : leg_times
}
return ret
def summarize_profile_non_transit_leg (leg, leg_modes, leg_times) :
if len(leg) == 1:
leg_modes.append(leg[0]['mode'])
leg_times.append(leg[0]['time'])
else:
modes = []
times = []
for mode_option in leg:
modes.append(mode_option['mode'])
times.append(mode_option['time'])
leg_modes.append(modes)
leg_times.append(times)
# Generate a callback closure containing the unfinished row.
# We could potentially avoid this by only saving the URL or query parameters, and not passing in a row.
def response_callback_factory(row, profile) :
def handle_response(response, *args, **kwargs) :
global n # avoid "referenced before assignment" weirdness
n += 1
t = (time.time() - t0) / 60.0
T = (N * t) / n
print "Request %d/%d, time %0.2f min of %0.2f (estimated) received" % (n, N, t, T), response,
n_itin = 0
elapsed = 0
if response.status_code != 200 :
status = 'failed'
else :
row['itins'] = []
objs = response.json()
if profile:
row['query_type'] = 'profile'
if 'options' in objs:
options = objs['options']
n_itin = len(options)
# check response for timeout flag
status = 'complete'
# status = 'timed out'
else:
status = 'no paths'
else:
row['query_type'] = 'plan'
row['debug'] = objs['debugOutput']
elapsed = objs['debugOutput']['totalTime']
if 'plan' in objs:
itineraries = objs['plan']['itineraries']
n_itin = len(itineraries)
# check response for timeout flag
status = 'complete'
# status = 'timed out'
else:
status = 'no paths'
row['total_time'] = str(elapsed) + ' msec'
row['avg_time'] = None if n_itin == 0 else '%f msec' % (float(elapsed) / n_itin)
print status
response_id = len(response_json) # not threadsafe -- not atomic with following line
response_json.append( row )
row['status'] = response.status_code
row['response_id'] = response_id
# Create a row for each itinerary/option within this single trip planner result
if profile:
for (option_number, option) in enumerate(options) :
option_row = summarize_profile (option)
option_row['itinerary_number'] = option_number + 1
full_itin = {'response_id':response_id,'itinerary_number':option_number+1}
full_itin['body']=option
full_itins_json.append( full_itin )
row['itins'].append( option_row )
else:
for (itinerary_number, itinerary) in enumerate(itineraries) :
itin_row = summarize_plan (itinerary)
#itin_row['response_id'] = response_id
itin_row['itinerary_number'] = itinerary_number + 1
full_itin = {'response_id':response_id,'itinerary_number':itinerary_number+1}
full_itin['body']=itinerary
full_itins_json.append( full_itin )
row['itins'].append( itin_row )
if SHOW_RESPONSE :
pp.pprint(row)
# return the function definition, a closure for a specific instance of 'row'
return handle_response
def run(connect_args) :
"This is the principal function..."
notes = connect_args.pop('notes')
retry = connect_args.pop('retry')
fast = connect_args.pop('fast')
host = connect_args.pop('host')
profile = connect_args.pop('profile')
print("profile=%s"%profile)
info = getServerInfo(host)
while retry > 0 and info == None:
print "Failed to connect to OTP server. Waiting to retry (%d)." % retry
time.sleep(10)
info = getServerInfo(host)
retry -= 1
if info == None :
print "Failed to identify OTP version. Exiting."
exit(-2)
# Create a dict describing this particular run of the profiler, which will be output as JSON
run_time_id = int(time.time())
run_row = info + (notes, run_time_id)
run_json = dict(zip(('git_sha1','git_describe','cpu_name','cpu_cores','notes','id'), run_row))
all_params = get_params(fast)
random.shuffle(all_params)
global t0, N # HACK
t0 = time.time()
N = len(all_params)
reqs = []
for params in all_params :
params = dict(params) # TODO necessary?
request_id = params.pop('id')
oid = params.pop('oid')
tid = params.pop('tid')
params['date'] = DATE
if profile:
api_method = 'profile'
params['from'] = params.pop('fromPlace')
params['to'] = params.pop('toPlace')
params['modes'] = params.pop('mode')
params['limit'] = 3
else:
api_method = 'plan'
params['numItineraries'] = 3
qstring = urllib.urlencode(params)
url = "http://%s/otp/routers/default/%s?%s"%(host, api_method, qstring)
# Tomcat server + spaces in URLs -> HTTP 505 confusion
if SHOW_PARAMS :
pp.pprint(params)
if SHOW_URL :
print url
row = { 'url' : url,
'run_id' : run_time_id,
'request_id' : request_id,
'origin_id' : oid,
'target_id' : tid,
'id_tuple' : "%s-%s-%s"%(oid,tid,request_id),
'membytes' : None }
# You can't give arguments to the response callback, you have to make a factory function:
# "http://stackoverflow.com/questions/25115151/how-to-pass-parameters-to-hooks-in-python-grequests"
# Closures are created in Python by function calls.
response_callback = response_callback_factory(row, profile)
headers = {'Accept': 'application/json'}
req = grequests.get(url, headers=headers, hooks=dict(response=response_callback))
reqs.append(req)
# Set max number of concurrent requests to 20, OTP should throttle this via worker threads
grequests.map(reqs, size=20)
# Write out all results at the end. Really, this should probably be done in streaming fashion.
fpout = open("run_summary.%s.json"%run_time_id,"w")
run_json['responses'] = response_json
simplejson.dump(run_json, fpout, indent=2)
fpout.close()
fpout = open("full_itins.%s.json"%run_time_id,"w")
simplejson.dump( full_itins_json, fpout, indent=2 )
fpout.close()
import argparse
if __name__=="__main__":
import argparse # optparse is deprecated
parser = argparse.ArgumentParser(description='perform an otp profiler run')
parser.add_argument('host')
parser.add_argument('-f', '--fast', action='store_true', default=False)
parser.add_argument('-n', '--notes')
parser.add_argument('-r', '--retry', type=int, default=5)
parser.add_argument('-p', '--profile', action='store_true', default=False)
args = parser.parse_args()
# args is a non-iterable, non-mapping Namespace (allowing usage in the form args.name),
# so convert it to a dict before passing it into the run function.
run(vars(args))