File: asyncio_to_thread.py

package info (click to toggle)
python-eventlet 0.40.1-2
  • links: PTS, VCS
  • area: main
  • in suites: forky
  • size: 3,200 kB
  • sloc: python: 25,101; sh: 78; makefile: 31
file content (26 lines) | stat: -rw-r--r-- 505 bytes parent folder | download | duplicates (5) 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
 
import eventlet
eventlet.monkey_patch()

from eventlet.patcher import original
from eventlet.asyncio import spawn_for_awaitable

import asyncio


def in_thread():
    time = original("time")
    time.sleep(0.001)
    return 3 + 4


async def go():
    return await asyncio.gather(
        asyncio.to_thread(in_thread),
        asyncio.to_thread(in_thread),
        asyncio.to_thread(in_thread),
    )


if __name__ == "__main__":
    assert spawn_for_awaitable(go()).wait() == [7, 7, 7]
    print("pass")