Skip to content

Instantly share code, notes, and snippets.

@snaury
Last active February 10, 2023 05:22
  • Star 11 You must be signed in to star a gist
  • Fork 3 You must be signed in to fork a gist
Star You must be signed in to star a gist
Save snaury/202bf4f22c41ca34e56297bae5f33fef to your computer and use it in GitHub Desktop.
#!/usr/bin/python3
import sys
import asyncio
import greenlet
class AsyncIoGreenlet(greenlet.greenlet):
def __init__(self, driver, fn):
greenlet.greenlet.__init__(self, fn, driver)
self.driver = driver
async def green_spawn(fn, *args):
context = AsyncIoGreenlet(greenlet.getcurrent(), fn)
result = context.switch(*args)
while context:
try:
value = await result
except:
result = context.throw(*sys.exc_info())
else:
result = context.switch(value)
return result
def green_await(awaitable):
current = greenlet.getcurrent()
if not isinstance(current, AsyncIoGreenlet):
raise TypeError('Cannot use green_await outside of green_spawn target')
return current.driver.switch(awaitable)
if __name__ == '__main__':
import time
def green_sleeper(delay, count=1):
while count > 0:
print(f'Green sleeping for {delay} seconds...')
now = time.time()
green_await(asyncio.sleep(delay))
print(f'Green returned after {time.time()-now:.1f} seconds (expected {delay} seconds).')
count -= 1
async def asyncio_sleeper(delay, count=1):
while count > 0:
print(f'AsyncIo sleeping for {delay} seconds...')
now = time.time()
await asyncio.sleep(delay)
print(f'AsyncIo returned after {time.time()-now:.1f} seconds (expected {delay} seconds).')
count -= 1
async def main():
tasks = []
tasks.extend(green_spawn(green_sleeper, i + 1, 3) for i in range(2))
tasks.extend(asyncio_sleeper(i + 1, 3) for i in range(2))
await asyncio.gather(*tasks)
asyncio.run(main())
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment