utils.py 594 B

1234567891011121314151617181920212223242526272829303132333435
  1. import os
  2. from pathlib import Path
  3. from typing import Union
  4. def create_workers(workers: int) -> Union[set[int], None]:
  5. pids = set()
  6. for i in range(workers):
  7. print(f'Forking worker #{i + 1}')
  8. pid = os.fork()
  9. print(f'Forked worker #{i + 1}')
  10. if pid > 0:
  11. # On master
  12. pids.add(pid)
  13. elif pid == 0:
  14. # In worker
  15. return None
  16. return pids
  17. def set_socket_permissions(path: str) -> None:
  18. if path.startswith('unix:'):
  19. path = path[5:]
  20. spath = Path(path)
  21. spath.chmod(0o777)