Timeouts for multiprocessing?

Question

I've searched StackOverflow and although I've found many questions on this, I haven't found an answer that fits for my situation/not a strong python programmer to adapt their answer to fit my need.

I've looked here to no avail:

kill a function after a certain time in windows

Python: kill or terminate subprocess when timeout

signal.alarm replacement in Windows [Python]

I am using multiprocessing to run multiple SAP windows at once to pull reports. The is set up to run on a schedule every 5 minutes. Every once in a while, one of the reports gets stalled due to the GUI interface and never ends. I don't get an error or exception, it just stalls forever. What I would like is to have a timeout function that during this part of the code that is executed in SAP, if it takes longer than 4 minutes, it times out, closes SAP, skips the rest of the code, and waits for next scheduled report time.

I am using Windows Python 2.7

import multiprocessing
from multiprocessing import Manager, Process
import time
import datetime

### OPEN SAP ###
def start_SAP():
   print 'opening SAP program'

### REPORTS IN SAP ###
def report_1(q, lock):
   while True:  # logic to get shared queue
       if not q.empty():
           lock.acquire()
           k = q.get()
           time.sleep(1)
           lock.release()
           break
       else:
           time.sleep(1)
   print 'running report 1'

def report_2(q, lock):
   while True:  # logic to get shared queue
       if not q.empty():
           lock.acquire()
           k = q.get()
           time.sleep(1)
           lock.release()
           break
       else:
           time.sleep(1)
   print 'running report 2'

def report_3(q, lock):
   while True:  # logic to get shared queue
       if not q.empty():
           lock.acquire()
           k = q.get()
           time.sleep(1)
           lock.release()
           break
       else:
           time.sleep(1)

   time.sleep(60000) #mimicking the stall for report 3 that takes longer than allotted time
   print 'running report 3'

def report_N(q, lock):
   while True:  # logic to get shared queue
       if not q.empty():
           lock.acquire()
           k = q.get()
           time.sleep(1)
           lock.release()
           break
       else:
           time.sleep(1)
   print 'running report N'

### CLOSES SAP ###
def close_SAP():
   print 'closes SAP'

def format_file():
   print 'formatting files'

def multi_daily_pull():

    lock = multiprocessing.Lock()  # creating a lock in multiprocessing

    shared_list = range(6)  # creating a shared list for all functions to use
    q = multiprocessing.Queue()  # creating an empty queue in mulitprocessing
    for n in shared_list:  # putting list into the queue
        q.put(n)
    print 'Starting process at ', time.strftime('%m/%d/%Y %H:%M:%S')

    print 'Starting SAP Pulls at ', time.strftime('%m/%d/%Y %H:%M:%S')

    StartSAP = Process(target=start_SAP)
    StartSAP.start()
    StartSAP.join()

    report1= Process(target=report_1, args=(q, lock))
    report2= Process(target=report_2, args=(q, lock))
    report3= Process(target=report_3, args=(q, lock))
    reportN= Process(target=report_N, args=(q, lock))

    report1.start()
    report2.start()
    report3.start()
    reportN.start()

    report1.join()
    report2.join()
    report3.join()
    reportN.join()

    EndSAP = Process(target=close_SAP)
    EndSAP.start()
    EndSAP.join()

    formatfile = Process(target=format_file)
    formatfile .start()
    formatfile .join()

if __name__ == '__main__':
    multi_daily_pull()

Please [edit] the code in your question and [create a minimal, complete, and verifiable example](https://stackoverflow.com/help/mcve). — martineau, Feb 19 '18 at 17:24
Hi @martineau, I have updated the code. Please let me know if you need any further clarifications! — Kevin, Feb 19 '18 at 17:49
Not even close...the you code have is still very far from minimal. For one thing there are more unused `import`s in it than actually are needed. For another, what's the `schedule` module? You need to post enough of the code needed to be able to run and reproduce the problem (but no more). — martineau, Feb 19 '18 at 18:22
@martineau sorry, I was just copying my code, I have removed the unused `imports` and the unnecessary class file. When running the code, you will see that on the 5th minute interval (starting at 0:00), the code will run. It will run report 1,2, & N but will not finish running report 3. It gets stalled (I mimicked this by using sleep for a long time). I need to find a way to identify that after X minutes, to end the run of code, run the close_SAP function, and wait for the next scheduled run. — Kevin, Feb 19 '18 at 20:03
Possible duplicate of [How to add a timeout to a function in Python](https://stackoverflow.com/questions/2196999/how-to-add-a-timeout-to-a-function-in-python) — noxdafox, Feb 19 '18 at 20:24
@noxdafox, that is only for Python 3, not 2.7. It also received a lot of comments on how it was incorrect. — Kevin, Feb 19 '18 at 20:26
Kevin: While that's a vast improvement, you've still got an `import schedule` in the code which isn't a standard Python module (and having/using it prevents others from running your code). Maybe what it does is part of the problem... — martineau, Feb 19 '18 at 20:36
Hi @martineau it allows my code function to be run on a specified interval of time but I understand your point that it is not standard and makes simulation harder for others. I have removed it as it isn't part of the problem! Thanks for the tips! — Kevin, Feb 19 '18 at 20:40
Kevin: Yes, that's more like it. Glad you're taking my advice in the spirit it's offered—to enable getting you one or more answers. — martineau, Feb 19 '18 at 21:07

martineau · Accepted Answer · 2019-04-08T18:47:43.743

One way to do what you want would be to use the optional timeout argument that the Process.join() method accepts. This will make it only block the calling thread at most that length of time.

I also set the daemon attribute of each Process instance so your main thread will be able to terminate even if one of the processes it started is still "running" (or has hung up).

One final point, you don't need a multiprocessing.Lock to control access a multiprocessing.Queue, because they handle that aspect of things automatically, so I removed it. You may still want to have one for some other reason, such as controlling access to stdout so printing to it from the various processes doesn't overlap and mess up what is output to the screen.

import multiprocessing
from multiprocessing import Process
import time
import datetime

def start_SAP():
    print 'opening SAP program'

### REPORTS IN SAP ###
def report_1(q):
    while True:  # logic to get shared queue
        if q.empty():
            time.sleep(1)
        else:
            k = q.get()
            time.sleep(1)
            break

    print 'report 1 finished'

def report_2(q):
    while True:  # logic to get shared queue
        if q.empty():
            time.sleep(1)
        else:
            k = q.get()
            time.sleep(1)
            break

    print 'report 2 finished'

def report_3(q):
    while True:  # logic to get shared queue
        if q.empty():
            time.sleep(1)
        else:
            k = q.get()
            time.sleep(60000) # Take longer than allotted time
            break

    print 'report 3 finished'


def report_N(q):
    while True:  # logic to get shared queue
        if q.empty():
            time.sleep(1)
        else:
            k = q.get()
            time.sleep(1)
            break

    print 'report N finished'

def close_SAP():
    print 'closing SAP'

def format_file():
    print 'formatting files'

def multi_daily_pull():
    shared_list = range(6)  # creating a shared list for all functions to use
    q = multiprocessing.Queue()  # creating an empty queue in mulitprocessing
    for n in shared_list:  # putting list into the queue
        q.put(n)
    print 'Starting process at ', time.strftime('%m/%d/%Y %H:%M:%S')

    print 'Starting SAP Pulls at ', time.strftime('%m/%d/%Y %H:%M:%S')

    StartSAP = Process(target=start_SAP)
    StartSAP.start()
    StartSAP.join()

    report1 = Process(target=report_1, args=(q,))
    report1.daemon = True
    report2 = Process(target=report_2, args=(q,))
    report2.daemon = True
    report3 = Process(target=report_3, args=(q,))
    report3.daemon = True
    reportN = Process(target=report_N, args=(q,))
    reportN.daemon = True

    report1.start()
    report2.start()
    report3.start()
    reportN.start()

    report1.join(30)
    report2.join(30)
    report3.join(30)
    reportN.join(30)

    EndSAP = Process(target=close_SAP)
    EndSAP.start()
    EndSAP.join()

    formatfile = Process(target=format_file)
    formatfile .start()
    formatfile .join()

if __name__ == '__main__':
    multi_daily_pull()

Thank you so much @martineau and for your patience with me. This solution works perfectly! Simple yet so powerful. All the best! — Kevin, Feb 20 '18 at 13:17
Kevin: You welcome. Note also that you might want to add code to check whether each `join()` timed-out or not, and if it did, then call `Process.terminate()` on it so no "zombie" processes are left running when the main script exits. Since doing that would be more-or-less the same for each report process, you could put all that logic into a function that handles the details for any given one that's passed to it as an argument. — martineau, Feb 20 '18 at 17:42

Timeouts for multiprocessing?

1 Answers1