Skip to content

Instantly share code, notes, and snippets.

@rootux
Last active January 7, 2019 19:08
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 rootux/ffdf9e874a7f4d25792efbf25b878429 to your computer and use it in GitHub Desktop.
Save rootux/ffdf9e874a7f4d25792efbf25b878429 to your computer and use it in GitHub Desktop.
Playing with processes (Helping a brother out)
import os
import sys
import random
import time
import tempfile
from datetime import datetime
from multiprocessing import Process
from utils import is_number
MAX_TIME_TO_SLEEP = 20
def sub_program():
pid = str(os.getpid())
sleep_time = random.randrange(MAX_TIME_TO_SLEEP)
print("Hello from pid {}\r\nTime {}".format(pid, datetime.now()))
print("Sleeping for {} seconds".format(sleep_time))
time.sleep(sleep_time)
if __name__ == '__main__':
parent_id = os.getpid()
pidfile = os.path.join(tempfile.gettempdir(),"my_unique_process.pid")
if(len(sys.argv) <= 1):
print("Error - Please supply number of processes as an argument. For example {} 3".format(sys.argv[0]))
sys.exit()
number_of_processes = sys.argv[1]
if not is_number(number_of_processes):
print("Error - Please supply number as the number of processes")
sys.exit()
number_of_processes = int(number_of_processes)
print("Started at time {}".format(datetime.now()))
# Check if already running
if os.path.isfile(pidfile):
print("{} already exists, exiting".format(pidfile))
sys.exit()
open(pidfile, "w").write("{}".format(parent_id))
try:
processes = []
for i in range(number_of_processes):
p = Process(target=sub_program)
processes.append(p)
for p in processes:
p.start()
success = 0
failure = 0
for p in processes:
p.join()
if p.exitcode == 0:
success+=1
else:
failure+=1
print("Success process {}, Failed process {}".format(success,failure))
finally:
print("Finished running.\r\nTime {}".format(datetime.now()))
os.unlink(pidfile)
@rootux
Copy link
Author

rootux commented Jan 7, 2019

And here is the utils function that checks for a number:

def is_number(s):
  try:
    float(s)
    return True
  except ValueError:
    pass

  try:
    import unicodedata
    unicodedata.numeric(s)
    return True
  except (TypeError, ValueError):
    pass

  return False

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