1
\$\begingroup\$

I am currently working on a problem that involves multiple producers and a single consumer through a queue object. The following list is about a few things that's special about this problem:

  1. One producer is conditional, meaning that it only produces data to the queue under certain condition.
  2. I'd like the problem to be as robust as possible, including graceful termination on ctrl-c.
  3. A particular simplification I did is to run one of the producer in the main thread, just so that I don't have too much code here.
  4. There are two type of data(job) that's got put into the queue, integer 1 and integer 2. Integer 1 is generated by the main thread(one producer) and integer 2 is generated less frequent(which I call it snapshot because that's what it is in my real-life project).

I pasted the full toy program below. I could have used logging package but this is what I did so I left it unchanged.

import threading
import time
import queue
import datetime

def process_job(evt, q, cond):
    while not evt.is_set():
        try:
            print(f'{datetime.now()} worker: getting a job from the queue')
            job = q.get(timeout=2)
        except:
            print(f'{datetime.now()} worker: q.get timed-out. continue...')
            continue

        if job == 1:
            print(f'{datetime.now()} worker: processing job == 1')
            q.task_done()
        elif job == 2: # job 2 is snapshot
            print(f'{datetime.now()} worker: processing job == 2')
            q.task_done()
        else:
            print(f'{datetime.now()} worker: bad job {job}')
            q.task_done()


def get_snapshot_long():
    print(f'{datetime.now()} get_snapshot: start...')
    time.sleep(10)
    print(f'{datetime.now()} get_snapshot: returning snapshot')
    return 2


def fetch_snapshot(evt, q, cond, flag):
    while not evt.is_set():
        with cond:
            print(f'{datetime.now()} fetcher: waiting for cond')
            ret = cond.wait_for(lambda : flag[0], timeout=5)
            if not ret: # timed-out
                print(f'{datetime.now()} fetcher: wait_for timed-out. continue...')
                continue

            ss = get_snapshot_long()
            q.put(ss)
            flag[0] = False


if __name__ == '__main__':
    stop_worker_thread_evt = threading.Event()

    stop_snapshot_thread_evt = threading.Event()
    flag_lock = threading.Lock()
    cond = threading.Condition(flag_lock)
    run_snapshot = [False] # just so I can modify in other threads

    q = queue.Queue()

    # -------------------------------------------------------
    work_thread = threading.Thread(
        target=process_job,
        args=(stop_worker_thread_evt, q, cond))

    snapshot_thread = threading.Thread(
        target=fetch_snapshot,
        args=(stop_snapshot_thread_evt, q, cond, run_snapshot))

    work_thread.start()
    snapshot_thread.start()

    # --------------------------------------------------------
    count = 0 # every 5 count, put in a request for snapshot
    print(f'{datetime.now()} main: start...')
    while True:
        try:
            time.sleep(5)
            q.put(1)
            count += 1
            if count % 5 == 0:
                print(f'{datetime.now()} main: trying to notify snapshot thread...')
                with cond:
                    run_snapshot[0] = True
                    cond.notify()
                print(f'{datetime.now()} main: snapshot notified')
        except KeyboardInterrupt:
            print(f'{datetime.now()} main: user terminating...')
            break

    # -------------------------------------------------------
    # terminate snapshot thread before terminating the consumer thread
    stop_snapshot_thread_evt.set()
    print(f'{datetime.now()} main: marked events set')

    snapshot_thread.join()
    print(f'{datetime.now()} main: snapshot thread joined')

    stop_worker_thread_evt.set()
    work_thread.join()
    print(f'{datetime.now()} main: work thread joined')

    print(f'{datetime.now()} main: q size={q.qsize()}')
    #q.join() I don't think we can join the q here.
    #because the stop of threads are controlled by event variables
    #instead of sentinel job in the queue. You just can't guarantee
    #an empty queue at this point
    print(f'{datetime.now()} main: q joined. DONE')

Please review and provide critical suggestions. In addition, I have a few specific questions about this program:

  1. I used threading.Event object to control the termination of non-main threads. Is that the best way to go about it? One issue I had with that is, we could potentially mark the consumer thread done where there are still jobs in it. I used to use a sentinel job in the queue to control that to ensure an empty queue.
  2. Is it a good idea to have a snapshot thread? Right now I am using an extra threading.Condition object to communicate to this thread. Is that a good solution?
\$\endgroup\$
6
  • 1
    \$\begingroup\$ The current question title, which states your concerns about the code, is too general to be useful here. Please edit to the site standard, which is for the title to simply state the task accomplished by the code. Please see How to get the best value out of Code Review: Asking Questions for guidance on writing good question titles. \$\endgroup\$ Nov 14 at 9:46
  • \$\begingroup\$ What exact version of Python are you targeting? \$\endgroup\$
    – Mast
    Nov 14 at 11:33
  • \$\begingroup\$ @Mast, I am targeting python 3.6+. \$\endgroup\$
    – dhu
    Nov 14 at 21:43
  • \$\begingroup\$ @TobySpeight I updated the title. I hope it is more descriptive about the issue I am trying to solve. \$\endgroup\$
    – dhu
    Nov 14 at 21:48
  • \$\begingroup\$ I've further edited to simplify to just what the code does, rather than your concerns about it. \$\endgroup\$ Nov 15 at 9:27

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.