Skip to content

Instantly share code, notes, and snippets.

@rsj217

rsj217/asyncio-as-complete.py

Created January 17, 2017 02:27
Download ZIP

Revisions

  1. rsj217 created this gist Jan 17, 2017.
    19 changes: 19 additions & 0 deletions asyncio-as-complete.py
    View file
    Original file line number Diff line number Diff line change
    @@ -0,0 +1,19 @@
    async def main():
    coroutine1 = do_some_work(1)
    coroutine2 = do_some_work(2)
    coroutine3 = do_some_work(4)

    tasks = [
    asyncio.ensure_future(coroutine1),
    asyncio.ensure_future(coroutine2),
    asyncio.ensure_future(coroutine3)
    ]
    for task in asyncio.as_completed(tasks):
    result = await task
    print('Task ret: {}'.format(result))

    start = now()

    loop = asyncio.get_event_loop()
    done = loop.run_until_complete(main())
    print('TIME: ', now() - start)
    21 changes: 21 additions & 0 deletions asyncio-await.py
    View file
    Original file line number Diff line number Diff line change
    @@ -0,0 +1,21 @@
    import asyncio
    import time

    now = lambda: time.time()

    async def do_some_work(x):
    print('Waiting: ', x)
    await
    asyncio.sleep(x)
    return 'Done after {}s'.format(x)


    start = now()

    coroutine = do_some_work(2)
    loop = asyncio.get_event_loop()
    task = asyncio.ensure_future(coroutine)
    loop.run_until_complete(task)

    print('Task ret: ', task.result())
    print('TIME: ', now() - start)
    32 changes: 32 additions & 0 deletions asyncio-concurency.py
    View file
    Original file line number Diff line number Diff line change
    @@ -0,0 +1,32 @@
    import asyncio

    import time

    now = lambda: time.time()

    async def do_some_work(x):
    print('Waiting: ', x)

    await asyncio.sleep(x)
    return 'Done after {}s'.format(x)


    start = now()

    coroutine1 = do_some_work(1)
    coroutine2 = do_some_work(2)
    coroutine3 = do_some_work(4)

    tasks = [
    asyncio.ensure_future(coroutine1),
    asyncio.ensure_future(coroutine2),
    asyncio.ensure_future(coroutine3)
    ]

    loop = asyncio.get_event_loop()
    loop.run_until_complete(asyncio.wait(tasks))

    for task in tasks:
    print('Task ret: ', task.result())

    print('TIME: ', now() - start)
    17 changes: 17 additions & 0 deletions asyncio-corutine.py
    View file
    Original file line number Diff line number Diff line change
    @@ -0,0 +1,17 @@
    import time
    import asyncio


    now = lambda : time.time()

    async def do_some_work(x):
    print('Waiting: ', x)

    start = now()

    coroutine = do_some_work(2)

    loop = asyncio.get_event_loop()
    loop.run_until_complete(coroutine)

    print('TIME: ', now() - start)
    43 changes: 43 additions & 0 deletions asyncio-crountine-while.py
    View file
    Original file line number Diff line number Diff line change
    @@ -0,0 +1,43 @@
    import time
    import asyncio
    import redis


    now = lambda : time.time()

    def get_redis():
    connection_pool = redis.ConnectionPool(host='127.0.0.1', db=3)
    return redis.Redis(connection_pool=connection_pool)

    rcon = get_redis()

    async def worker():
    print('Start worker')

    while True:
    start = now()
    task = rcon.rpop("queue")
    if not task:
    await asyncio.sleep(1)
    continue
    print('Wait ', int(task))
    await asyncio.sleep(int(task))
    print('Done ', task, now() - start)


    def main():
    asyncio.ensure_future(worker())
    asyncio.ensure_future(worker())

    loop = asyncio.get_event_loop()
    try:
    loop.run_forever()
    except KeyboardInterrupt as e:
    print(asyncio.gather(*asyncio.Task.all_tasks()).cancel())
    loop.stop()
    loop.run_forever()
    finally:
    loop.close()

    if __name__ == '__main__':
    main()
    19 changes: 19 additions & 0 deletions asyncio-futrue-result.py
    View file
    Original file line number Diff line number Diff line change
    @@ -0,0 +1,19 @@
    import time
    import asyncio


    now = lambda : time.time()

    async def do_some_work(x):
    print('Waiting {}'.format(x))
    return 'Done after {}s'.format(x)

    start = now()

    coroutine = do_some_work(2)
    loop = asyncio.get_event_loop()
    task = asyncio.ensure_future(coroutine)
    loop.run_until_complete(task)

    print('Task ret: {}'.format(task.result()))
    print('TIME: {}'.format(time.time() - start))
    55 changes: 55 additions & 0 deletions asyncio-master-worker-brpop.py
    View file
    Original file line number Diff line number Diff line change
    @@ -0,0 +1,55 @@

    import time
    import asyncio
    from threading import Thread
    import redis
    import aiohttp

    def get_redis():
    connection_pool = redis.ConnectionPool(host='127.0.0.1', db=3)
    return redis.Redis(connection_pool=connection_pool)

    rcon = get_redis()

    now = lambda : time.time()

    def start_loop(loop):
    asyncio.set_event_loop(loop)
    loop.run_forever()

    async def fetch(url):
    async with aiohttp.ClientSession() as session:
    async with session.get(url) as resp:
    print(resp.status)
    return await resp.text()

    async def do_some_work(x):
    # print('Waiting ', x)
    # await asyncio.sleep(x)
    # print('Done ', x)
    try:
    ret = await fetch(url='http://127.0.0.1:5000/{}'.format(x))
    print(ret)
    except Exception as e:
    try:
    print(await fetch(url='http://127.0.0.1:5000/error'))
    except Exception as e:
    print(e)
    else:
    print('Done {}'.format(x))


    new_loop = asyncio.new_event_loop()
    t = Thread(target=start_loop, args=(new_loop,))
    t.setDaemon(True)
    t.start()

    try:
    while True:
    _, task = rcon.brpop("queue")
    asyncio.run_coroutine_threadsafe(do_some_work(int(task)), new_loop)
    except Exception as e:
    print('error', e)
    new_loop.stop()
    finally:
    pass
    53 changes: 53 additions & 0 deletions asyncio-master-worker-rpop.py
    View file
    Original file line number Diff line number Diff line change
    @@ -0,0 +1,53 @@
    import time
    import asyncio
    from threading import Thread
    import redis
    import aiohttp

    def get_redis():
    connection_pool = redis.ConnectionPool(host='127.0.0.1', db=3)
    return redis.Redis(connection_pool=connection_pool)

    rcon = get_redis()

    def start_loop(loop):
    asyncio.set_event_loop(loop)
    loop.run_forever()

    async def fetch(url):
    async with aiohttp.ClientSession() as session:
    async with session.get(url) as resp:
    print(resp.status)
    return await resp.text()

    async def do_some_work(x):
    print('Waiting ', x)
    try:
    ret = await fetch(url='http://127.0.0.1:5000/{}'.format(x))
    print(ret)
    except Exception as e:
    try:
    print(await fetch(url='http://127.0.0.1:5000/error'))
    except Exception as e:
    print(e)
    else:
    print('Done {}'.format(x))


    new_loop = asyncio.new_event_loop()
    t = Thread(target=start_loop, args=(new_loop,))
    t.setDaemon(True)
    t.start()

    try:
    while True:
    task = rcon.rpop("queue")
    if not task:
    time.sleep(1)
    continue
    asyncio.run_coroutine_threadsafe(do_some_work(int(task)), new_loop)
    except Exception as e:
    print('error')
    new_loop.stop()
    finally:
    pass
    58 changes: 58 additions & 0 deletions asyncio-multi-task.py
    View file
    Original file line number Diff line number Diff line change
    @@ -0,0 +1,58 @@
    import asyncio

    import time

    now = lambda: time.time()

    async def do_some_work(x):
    print('Waiting: ', x)

    await asyncio.sleep(x)
    return 'Done after {}s'.format(x)


    async def main():
    coroutine1 = do_some_work(1)
    coroutine2 = do_some_work(2)
    coroutine3 = do_some_work(4)

    tasks = [
    asyncio.ensure_future(coroutine1),
    asyncio.ensure_future(coroutine2),
    asyncio.ensure_future(coroutine3)
    ]

    dones, pendings = await asyncio.wait(tasks)

    for task in dones:
    print('Task ret: ', task.result())

    start = now()

    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())

    print('TIME: ', now() - start)



    async def main():
    coroutine1 = do_some_work(1)
    coroutine2 = do_some_work(2)
    coroutine3 = do_some_work(2)

    tasks = [
    asyncio.ensure_future(coroutine1),
    asyncio.ensure_future(coroutine2),
    asyncio.ensure_future(coroutine3)
    ]

    return await asyncio.gather(*tasks)

    start = now()

    loop = asyncio.get_event_loop()
    results = loop.run_until_complete(main())

    for result in results:
    print('Task ret: ', result)
    41 changes: 41 additions & 0 deletions asyncio-stop-2.py
    View file
    Original file line number Diff line number Diff line change
    @@ -0,0 +1,41 @@
    import asyncio

    import time

    now = lambda: time.time()

    async def do_some_work(x):
    print('Waiting: ', x)

    await asyncio.sleep(x)
    return 'Done after {}s'.format(x)


    async def main():
    coroutine1 = do_some_work(1)
    coroutine2 = do_some_work(2)
    coroutine3 = do_some_work(2)

    tasks = [
    asyncio.ensure_future(coroutine1),
    asyncio.ensure_future(coroutine2),
    asyncio.ensure_future(coroutine3)
    ]
    done, pending = await asyncio.wait(tasks)
    for task in done:
    print('Task ret: ', task.result())


    start = now()

    loop = asyncio.get_event_loop()
    task = asyncio.ensure_future(main())
    try:
    loop.run_until_complete(task)
    except KeyboardInterrupt as e:
    print(asyncio.Task.all_tasks())
    print(asyncio.gather(*asyncio.Task.all_tasks()).cancel())
    loop.stop()
    loop.run_forever()
    finally:
    loop.close()
    38 changes: 38 additions & 0 deletions asyncio-stop.py
    View file
    Original file line number Diff line number Diff line change
    @@ -0,0 +1,38 @@
    import asyncio

    import time

    now = lambda: time.time()

    async def do_some_work(x):
    print('Waiting: ', x)

    await asyncio.sleep(x)
    return 'Done after {}s'.format(x)


    coroutine1 = do_some_work(1)
    coroutine2 = do_some_work(2)
    coroutine3 = do_some_work(2)

    tasks = [
    asyncio.ensure_future(coroutine1),
    asyncio.ensure_future(coroutine2),
    asyncio.ensure_future(coroutine3)
    ]

    start = now()

    loop = asyncio.get_event_loop()
    try:
    loop.run_until_complete(asyncio.wait(tasks))
    except KeyboardInterrupt as e:
    print(asyncio.Task.all_tasks())
    for task in asyncio.Task.all_tasks():
    print(task.cancel())
    loop.stop()
    loop.run_forever()
    finally:
    loop.close()

    print('TIME: ', now() - start)
    22 changes: 22 additions & 0 deletions asyncio-task-callback.py
    View file
    Original file line number Diff line number Diff line change
    @@ -0,0 +1,22 @@
    import time
    import asyncio


    now = lambda : time.time()

    async def do_some_work(x):
    print('Waiting: ', x)
    return 'Done after {}s'.format(x)

    def callback(future):
    print('Callback: ', future.result())

    start = now()

    coroutine = do_some_work(2)
    loop = asyncio.get_event_loop()
    task = asyncio.ensure_future(coroutine)
    task.add_done_callback(callback)
    loop.run_until_complete(task)

    print('TIME: ', now() - start)
    18 changes: 18 additions & 0 deletions asyncio-task.py
    View file
    Original file line number Diff line number Diff line change
    @@ -0,0 +1,18 @@
    import asyncio
    import time

    now = lambda : time.time()

    async def do_some_work(x):
    print('Waiting: ', x)

    start = now()

    coroutine = do_some_work(2)
    loop = asyncio.get_event_loop()
    # task = asyncio.ensure_future(coroutine)
    task = loop.create_task(coroutine)
    print(task)
    loop.run_until_complete(task)
    print(task)
    print('TIME: ', now() - start)
    22 changes: 22 additions & 0 deletions asyncio-thread-block.py
    View file
    Original file line number Diff line number Diff line change
    @@ -0,0 +1,22 @@
    from threading import Thread


    def start_loop(loop):
    asyncio.set_event_loop(loop)
    loop.run_forever()


    def more_work(x):
    print('More work {}'.format(x))
    time.sleep(x)
    print('Finished more work {}'.format(x))


    start = time.time()
    new_loop = asyncio.new_event_loop()
    t = Thread(target=start_loop, args=(new_loop,))
    t.start()
    print('TIME: {}'.format(time.time() - start))

    new_loop.call_soon_threadsafe(more_work, 6)
    new_loop.call_soon_threadsafe(more_work, 3)
    24 changes: 24 additions & 0 deletions asyncio-thread-unblock
    View file
    Original file line number Diff line number Diff line change
    @@ -0,0 +1,24 @@
    def start_loop(loop):
    asyncio.set_event_loop(loop)
    loop.run_forever()

    async def do_some_work(x):
    print('Waiting {}'.format(x))
    await asyncio.sleep(x)
    print('Done after {}s'.format(x))


    def more_work(x):
    print('More work {}'.format(x))
    time.sleep(x)
    print('Finished more work {}'.format(x))


    start = time.time()
    new_loop = asyncio.new_event_loop()
    t = Thread(target=start_loop, args=(new_loop,))
    t.start()
    print('TIME: {}'.format(time.time() - start))

    asyncio.run_coroutine_threadsafe(do_some_work(6), new_loop)
    asyncio.run_coroutine_threadsafe(do_some_work(4), new_loop)