Python Daemon Thread Clean Up Logic on Abrupt sys.exit()

Using Linux and Python 2.7.6, I have a script that uploads lots of files at one time. I am using multi-threading with the Queue and Threading modules.

I implemented a handler for SIGINT to stop the script if the user hits ctrl-C. I prefer to use daemon threads so I don't have to clear the queue, which would require alot of re-writing code to make the SIGINT handler have access to the Queue object since the handlers don't take parameters.

To make sure the daemon threads finish and clean up before sys.exit(), I am using threading.Event() and threading.clear() to make threads wait. This code seems to work as print threading.enumerate() only shows the main thread before the script terminates when I did debugging. Just to make sure, I was wondering if there is any kind of insight to this clean up implementation that I might be missing even though it seems to be working for me:

def signal_handler(signal, frame):
    global kill_received
    kill_received = True
    msg = (
         "\n\nYou pressed Ctrl+C!"
         "\nYour logs and their locations are:"
         "\n{}\n{}\n{}\n\n".format(debug, error, info))
    logger.info(msg)
    threads = threading.Event()
    threads.clear()

    while True:
        time.sleep(3)
        threads_remaining = len(threading.enumerate())
        print threads_remaining
        if threads_remaining == 1:
            sys.exit()

def do_the_uploads(file_list, file_quantity,
        retry_list, authenticate):
    """The uploading engine"""
    value = raw_input(
        "\nPlease enter how many concurent "
        "uploads you want at one time(example: 200)> ")
    value = int(value)
    logger.info('{} concurent uploads will be used.'.format(value))

    confirm = raw_input(
        "\nProceed to upload files? Enter [Y/y] for yes: ").upper()
    if confirm == "Y":
        kill_received = False
        sys.stdout.write("\x1b[2J\x1b[H")
        q = CustomQueue()

        def worker():
            global kill_received
            while not kill_received:
                item = q.get()
                upload_file(item, file_quantity, retry_list, authenticate, q)
                q.task_done()

        for i in range(value):
            t = Thread(target=worker)
            t.setDaemon(True)
            t.start()

        for item in file_list:
            q.put(item)

        q.join()

        print "Finished. Cleaning up processes...",
        #Allowing the threads to cleanup
        time.sleep(4)



def upload_file(file_obj, file_quantity, retry_list, authenticate, q):
    """Uploads a file. One file per it's own thread. No batch style. This way if one upload
       fails no others are effected."""
    absolute_path_filename, filename, dir_name, token, url = file_obj
    url = url + dir_name + '/' + filename
    try:
        with open(absolute_path_filename) as f:  
           r = requests.put(url, data=f, headers=header_collection, timeout=20)
    except requests.exceptions.ConnectionError as e:
        pass
    if src_md5 == r.headers['etag']:
       file_quantity.deduct()

Asked by: Alberta339 | Posted: 27-01-2022

Answer 1

If you want to handle Ctrl+C; it is enough to handle KeyboardInterrupt exception in the main thread. Don't use global X in a function unless you do X = some_value in it. Using time.sleep(4) to allow the threads to cleanup is a code smell. You don't need it.

I am using threading.Event() and threading.clear() to make threads wait.

This code has no effect on your threads:

# create local variable
threads = threading.Event()
# clear internal flag in it (that is returned by .is_set/.wait methods)
threads.clear()

Don't call logger.info() from a signal handler in a multithreaded program. It might deadlock your program. Only a limited set of functions can be called from a signal handler. The safe option is to set a global flag in it and exit:

def signal_handler(signal, frame):
    global kill_received
    kill_received = True
    # return (no more code)

The signal might be delayed until q.join() returns. Even if the signal were delivered immediately; q.get() blocks your child threads. They hang until the main thread exits. To fix both issues, you could use a sentinel to signal child processes that there are no more work, drop the signal handler completely in this case:

def worker(stopped, queue, *args):
    for item in iter(queue.get, None): # iterate until queue.get() returns None
        if not stopped.is_set(): # a simple global flag would also work here
           upload_file(item, *args)
        else:
           break # exit prematurely
    # do child specific clean up here

# start threads
q = Queue.Queue()
stopped = threading.Event() # set when threads should exit prematurely
threads = set()
for _ in range(number_of_threads):
    t = Thread(target=worker, args=(stopped, q)+other_args)
    threads.add(t)
    t.daemon = True
    t.start()

# provide work
for item in file_list:
    q.put(item)
for _ in threads:
    q.put(None) # put sentinel to signal the end

while threads: # until there are alive child threads
    try:
        for t in threads: 
            t.join(.3) # use a timeout to get KeyboardInterrupt sooner
            if not t.is_alive():
               threads.remove(t) # remove dead
               break
    except (KeyboardInterrupt, SystemExit):
        print("got Ctrl+C (SIGINT) or exit() is called")
        stopped.set() # signal threads to exit gracefully

I've renamed value to number_of_threads. I've used explicit threads set

If an individual upload_file() blocks then the program won't exit on Ctrl-C.

Your case seems to be simple enough for multiprocessing.Pool interface:

from multiprocessing.pool import ThreadPool
from functools import partial

def do_uploads(number_of_threads, file_list, **kwargs_for_upload_file):
    process_file = partial(upload_file, **kwargs_for_upload_file)
    pool = ThreadPool(number_of_threads) # number of concurrent uploads
    try:
        for _ in pool.imap_unordered(process_file, file_list):
            pass # you could report progress here
    finally:
        pool.close() # no more additional work
        pool.join() # wait until current work is done

It should gracefully exit on Ctrl-C i.e., uploads that are in progress are allowed to finish but new uploads are not started.

Answered by: Brooke864 | Posted: 28-02-2022

Still can't find your answer? Check out these communities...

top

Python Daemon Thread Clean Up Logic on Abrupt sys.exit()

Answer 1

Similar questions

Still can't find your answer? Check out these communities...

© Patterns & Practices 2022 Rights Reserved | Contact us