1
\$\begingroup\$

I'm new to multi-threading and concurrency. I've got 2 lists that I want to process concurrently. Processing both lists requires a lot of time.sleep() and waiting for other servers to finish, so I figured rather than do one at a time I could process them simultaneously.

The first list is un-validated items and I try to validate them and if successful add them to the second list for processing. I want to start validating items in the first list and then while that is going start on the processing of the 2nd list. I've since found out I wanted a Queue for the 2nd list and that has helped but I can't help but feel that even though this works I'm missing something.

def process_sr(jira, tuple, args, log):
    ... # does some pretty straight forward stuff but lots of waiting
    return "done"

def waiting_for_build(tuple, validqueue, log):
    sr, pr = tuple
    iter = 0
    log.info("Waiting for {} to re-build".format(pr))
    while iter < 30:  # don't add. It's been more than 10 minutes and we need to finish
        log.debug("%s still waiting for Jenkins", pr)
        sleep(20)
        status = pr.build_status()
        if status is None:
            break  # don't add. Build finished "FAILURE" or didn't build in 10 min
        elif status == "Wait":
            iter += 1
        elif status == "Done":
            validqueue.put((sr,pr))

def build_lists(jira, log):
    validqueue = Queue()
    waiting_for_jenkins = []
    ...  # this builds the two lists
    return validqueue, waiting_for_jenkins

def main():
    args, log = cli_args()
    jira = JiraWrapper()

    validqueue, waiting_for_jenkins = build_lists(jira, log)

    workers = len(waiting_for_jenkins) + validqueue.qsize()

    threads = []
    for item in waiting_for_jenkins:
        name = str(item[1])
        threads.append(threading.Thread(target=waiting_for_build, name=name, args=(item, validqueue, log)))
        threads[-1].start()

    def worker(i):
        log.debug("started")
        results = []
        while True:
            log.debug("blocking on queue")
            item = validqueue.get()
            if item is None:
                log.debug("done")
                return results
            log.debug("got item %s", item)
            result = process_sr(jira, item, args, log)
            results.append({result:item})
            log.debug("task %s done", item)
            validqueue.task_done()

    log.debug("%d workers will be spawned", workers)
    if not workers:
        log.info("Nothing to do here")
    else:
        with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as executor:  # can't use thread_name_prefix until python 3.6
            futures = executor.map(worker, range(workers))

            for thread in threads:
                thread.join()
            validqueue.join()

            for i in range(workers):
                validqueue.put(None)
            for future in futures:
                log.debug(future)

    exit(0)


if __name__ == '__main__':
    main()

I've included as much as I think is needed. The main() method and specifically the two multi-threaded bits are where I'm looking for feedback.

\$\endgroup\$