Skip to content

Instantly share code, notes, and snippets.

@aholmes
Created January 26, 2022 21:48
Show Gist options
  • Save aholmes/5595b7d80fe83475668a74a84d0e1e76 to your computer and use it in GitHub Desktop.
Save aholmes/5595b7d80fe83475668a74a84d0e1e76 to your computer and use it in GitHub Desktop.
Behavior of @register in Python
import os
import threading
from multiprocessing import Pool
from atexit import register
from sys import exit
from typing import Union
@register
def term():
print("TERM!")
@register
def term2():
print("TERM2!")
@register
def term3():
print("TERM3!")
def f(x: int, source: str):
@register
def _():
print(f"TERM! f(x={x}, source='{source}')")
print(f"PID: {os.getpid()}")
return x * x
class FooThread(threading.Thread):
_return: int
def __init__(self, x):
threading.Thread.__init__(self)
self.x = x
def run(self):
self._return = f(self.x, "thread")
def join(self, timeout: Union[float, None] = None):
threading.Thread.join(self, timeout)
return self._return
if __name__ != "__main__":
print("-not __main__-")
exit(0)
print("-using Pool-")
with Pool(5) as p:
print(p.starmap(f, [(i, "pool") for i in range(1, 4)]))
print("-using FooThread-")
threads = [FooThread(i) for i in range(1, 4)]
[thread.start() for thread in threads]
print([thread.join() for thread in threads])
print("-done-")
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment