Skip to content

Instantly share code, notes, and snippets.

@vxgmichel
Created April 3, 2019 12:07
Show Gist options
  • Save vxgmichel/fc394d139218ebbfd1b775ae392b26ec to your computer and use it in GitHub Desktop.
Save vxgmichel/fc394d139218ebbfd1b775ae392b26ec to your computer and use it in GitHub Desktop.
Patch asyncio handles to measure task runtime
import time
import asyncio
def task_timing_patch():
class TaskTimingHandle(asyncio.events.Handle):
def _add_runtime(self, value):
try:
instance = self._callback.__self__
except AttributeError:
return
if not isinstance(instance, asyncio.Task):
return
try:
runtime = instance.runtime
except AttributeError:
runtime = 0.
instance.runtime = runtime + value
def _run(self, *args, **kwargs):
start = time.time()
try:
return super()._run(*args, **kwargs)
finally:
self._add_runtime(time.time() - start)
asyncio.events.Handle = TaskTimingHandle
async def test():
time.sleep(0.2)
await asyncio.sleep(0.8)
async def main():
task = asyncio.create_task(test())
await task
print(task.runtime)
if __name__ == "__main__":
task_timing_patch()
asyncio.run(main(), debug=True)
@matianjun1
Copy link

That's so hackish. Hahaha~

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment