def main():
- import io, pprint
+ import io
+ import pprint
# Initialize and populate our database.
conn = sqlite3.connect(":memory:")
# All rights reserved.
#
-import time, sys, multiprocessing, threading, queue, gc
+import time
+import sys
+import multiprocessing
+import threading
+import queue
+import gc
if sys.platform == 'win32':
_timer = time.clock
for i in range(iterations):
a = seq[5]
- elapsed = _timer()-t
+ elapsed = _timer() - t
print(iterations, 'iterations in', elapsed, 'seconds')
print('average number/sec:', iterations/elapsed)
l.acquire()
l.release()
- elapsed = _timer()-t
+ elapsed = _timer() - t
print(iterations, 'iterations in', elapsed, 'seconds')
print('average number/sec:', iterations/elapsed)
c.notify()
c.wait()
- elapsed = _timer()-t
+ elapsed = _timer() - t
c.release()
p.join()
return calculate(*args)
def mul(a, b):
- time.sleep(0.5*random.random())
+ time.sleep(0.5 * random.random())
return a * b
def plus(a, b):
- time.sleep(0.5*random.random())
+ time.sleep(0.5 * random.random())
return a + b
def f(x):
- return 1.0 / (x-5.0)
+ return 1.0 / (x - 5.0)
def pow3(x):
- return x**3
+ return x ** 3
def noop(x):
pass
# All rights reserved.
#
-import time, sys, random
+import time
+import sys
+import random
from queue import Empty
import multiprocessing # may get overwritten
multiprocessing = namespace
- for func in [ test_value, test_queue, test_condition,
- test_semaphore, test_join_timeout, test_event,
- test_sharedvalues ]:
+ for func in [test_value, test_queue, test_condition,
+ test_semaphore, test_join_timeout, test_event,
+ test_sharedvalues]:
print('\n\t######## %s\n' % func.__name__)
func()
def note(format, *args):
- sys.stderr.write('[%s]\t%s\n' % (current_process().name, format%args))
+ sys.stderr.write('[%s]\t%s\n' % (current_process().name, format % args))
class RequestHandler(SimpleHTTPRequestHandler):
server = HTTPServer(address, RequestHandler)
# create child processes to act as workers
- for i in range(number_of_processes-1):
+ for i in range(number_of_processes - 1):
Process(target=serve_forever, args=(server,)).start()
# main process also acts as a worker
import sqlite3
-import datetime, time
+import datetime
+import time
def adapt_datetime(ts):
return time.mktime(ts.timetuple())
"""Fixed offset in minutes east from UTC."""
def __init__(self, offset, name):
- self.__offset = timedelta(minutes = offset)
+ self.__offset = timedelta(minutes=offset)
self.__name = name
def utcoffset(self, dt):