Skip to content

Instantly share code, notes, and snippets.

@pwtail
Last active January 3, 2023 16:27
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save pwtail/e7c8bfacbe87c0cfcacc4a7477341468 to your computer and use it in GitHub Desktop.
Save pwtail/e7c8bfacbe87c0cfcacc4a7477341468 to your computer and use it in GitHub Desktop.
import asyncio
from functools import wraps
# это https://github.com/Bi-Coloured-Python-Rock-Snake/greenhack
from greenhack import exempt, as_async
def wrap_outer(co_func):
@wraps(co_func)
def wrapper(*args, **kwargs):
gen = co_func(*args, **kwargs)
val = None
while 1:
try:
val = gen.send(val)
except StopIteration as ex:
return ex.value
if val == 'start':
def fun():
# Выполняется в другом гринлете
yielded = gen.send('started')
assert yielded == 'end'
task = as_async(fun)()
task = asyncio.create_task(task)
yield from task
else:
yield val
return wrapper
class Start:
def __await__(self):
yielded = yield 'start'
assert yielded == 'started'
class End:
def __await__(self):
yielded = yield 'end'
assert yielded == 'start'
class Fibers:
def __aenter__(self):
return Start()
def __aexit__(self, exc_type, exc_val, exc_tb):
return End()
fibers = Fibers()
@exempt
async def sleep(sec):
await asyncio.sleep(sec)
return sec
async def main():
async with fibers:
x = sleep(0.5)
print(x)
await asyncio.sleep(0.5)
print(0.5)
if __name__ == '__main__':
@wrap_outer
async def outer():
await main()
asyncio.run(outer())
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment