Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def run_test(router, fp, s, context):
fp.seek(0, 2)
size = fp.tell()
print('Testing %s...' % (s,))
context.call(prepare)
t0 = mitogen.core.now()
context.call(transfer, router.myself(), fp.name)
t1 = mitogen.core.now()
print('%s took %.2f ms to transfer %.2f MiB, %.2f MiB/s' % (
s, 1000 * (t1 - t0), size / 1048576.0,
(size / (t1 - t0) / 1048576.0),
))
def run_test(router, fp, s, context):
fp.seek(0, 2)
size = fp.tell()
print('Testing %s...' % (s,))
context.call(prepare)
t0 = mitogen.core.now()
context.call(transfer, router.myself(), fp.name)
t1 = mitogen.core.now()
print('%s took %.2f ms to transfer %.2f MiB, %.2f MiB/s' % (
s, 1000 * (t1 - t0), size / 1048576.0,
(size / (t1 - t0) / 1048576.0),
))
@mitogen.main()
def main(router):
t0 = mitogen.core.now()
for x in xrange(200):
t = mitogen.core.now()
ctx = router.fork()
ctx.shutdown(wait=True)
print '++', 1000 * ((mitogen.core.now() - t0) / (1.0+x))
out.put(None)
ready = mitogen.core.Latch()
l1 = mitogen.core.Latch()
l2 = mitogen.core.Latch()
t1 = threading.Thread(target=flip_flop, args=(ready, l1, l2))
t2 = threading.Thread(target=flip_flop, args=(ready, l2, l1))
t1.start()
t2.start()
ready.get()
ready.get()
t0 = mitogen.core.now()
l1.put(None)
t1.join()
t2.join()
print('++', int(1e6 * ((mitogen.core.now() - t0) / (1.0+X))), 'usec')
def main(router):
c = router.fork()
n = 1048576 * 127
s = ' ' * n
print('bytes in %.2fMiB string...' % (n/1048576.0),)
t0 = mitogen.core.now()
for x in range(10):
tt0 = mitogen.core.now()
assert n == c.call(len, s)
print('took %dms' % (1000 * (mitogen.core.now() - tt0),))
t1 = mitogen.core.now()
print('total %dms / %dms avg / %.2fMiB/sec' % (
1000 * (t1 - t0),
(1000 * (t1 - t0)) / (x + 1),
((n * (x + 1)) / (t1 - t0)) / 1048576.0,
))
def main(router):
f = router.fork()
f.call(do_nothing)
t0 = mitogen.core.now()
for x in xrange(20000):
f.call(do_nothing)
print('++', int(1e6 * ((mitogen.core.now() - t0) / (1.0+x))), 'usec')
def main(router):
t0 = mitogen.core.now()
for x in range(100):
t = mitogen.core.now()
f = router.local()# debug=True)
tt = mitogen.core.now()
print(x, 1000 * (tt - t))
print('%.03f ms' % (1000 * (mitogen.core.now() - t0) / (1.0 + x)))
self.max_message_size = max_message_size
if python_path:
self.python_path = python_path
if connect_timeout:
self.connect_timeout = connect_timeout
if remote_name is None:
remote_name = get_default_remote_name()
if '/' in remote_name or '\\' in remote_name:
raise ValueError('remote_name= cannot contain slashes')
if remote_name:
self.remote_name = mitogen.core.to_text(remote_name)
self.debug = debug
self.profiling = profiling
self.unidirectional = unidirectional
self.max_message_size = max_message_size
self.connect_deadline = mitogen.core.now() + self.connect_timeout
self._cache[fullname] = tup
return tup
if source is None:
# TODO: make this .warning() or similar again once importer has its
# own logging category.
self._log.debug('could not find source for %r', fullname)
tup = self._make_negative_response(fullname)
self._cache[fullname] = tup
return tup
if self.minify_safe_re.search(source):
# If the module contains a magic marker, it's safe to minify.
t0 = mitogen.core.now()
source = mitogen.minify.minimize_source(source).encode('utf-8')
self.minify_secs += mitogen.core.now() - t0
if is_pkg:
pkg_present = get_child_modules(path)
self._log.debug('%s is a package at %s with submodules %r',
fullname, path, pkg_present)
else:
pkg_present = None
if fullname == '__main__':
source = self.neutralize_main(path, source)
compressed = mitogen.core.Blob(zlib.compress(source, 9))
related = [
to_text(name)
for name in self._finder.find_related(fullname)
if not mitogen.core.is_blacklisted_import(self, name)
]
def _install_timer(self, delay):
new = self._timer is None
self._timer = self.broker.timers.schedule(
when=mitogen.core.now() + delay,
func=self.reap,
)
if new:
mitogen.core.listen(self.broker, 'shutdown',
self._on_broker_shutdown)