Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
return x
amp.set_context("spawn")
with self.assertRaises(AttributeError):
p = amp.Worker(target=inline, args=(1,), name="test_inline")
p.start()
await p.join()
p = amp.Worker(target=two, name="test_global")
p.start()
await p.join()
values = list(range(10))
results = [await mapper(i) for i in values]
async with amp.Pool(2) as pool:
self.assertEqual(await pool.map(mapper, values), results)
self.assertEqual(p.result, 2)
async def test_pool_concurrency(self):
results = []
for sleep, tasks, processes, concurrency in PERF_SETS:
with Timer() as timer:
async with amp.Pool(processes, childconcurrency=concurrency) as pool:
await pool.map(sleepy, (sleep for _ in range(tasks)))
results.append((sleep, tasks, processes, concurrency, timer.result))
print()
for result in results:
print(*result)
async def makeurl(self,sta,end,limit):
urls=['http://img1.mm131.me/pic/'+str(i)+'/'+str(j)+'.jpg' for i in range(sta,end+1) for j in range(1,limit)]
return await Pool().map(self.async_get,urls)