Passing updated args to multiple threads periodically in python

Question

I have three base stations, they have to work in parallel, and they will receive a list every 10 seconds which contain information about their cluster, and I want to run this code for about 10 minutes. So, every 10 seconds my three threads have to call the target method with new arguments, and this process should last long for 10 minutes. I don't know how to do this, but I came up with the below idea which seems to be not quite a good one! Thus I appreciate any help.

I have a list named base_centroid_assign that I want to pass each item of it to a distinct thread. The list content will be updated frequently (supposed for instance 10 seconds), I so wish to recall my previous threads and give the update items to them.

In the below code, the list contains three items which have multiple items in them (it's nested). I want to have three threads stop after executing the quite simple target function, and then recall the threads with update item; however, when I run the below code, I ended up with 30 threads! (the run_time variable is 10 and list's length is 3).

How can I implement idea as mentioned above?

run_time = 10

def cluster_status_broadcasting(info_base_cent_avr):
    print(threading.current_thread().name)
    info_base_cent_avr.sort(key=lambda item: item[2], reverse=True)


start = time.time()

while(run_time > 0):
    for item in base_centroid_assign:
        t = threading.Thread(target=cluster_status_broadcasting, args=(item,))
        t.daemon = True
        t.start()

    print('Entire job took:', time.time() - start)
    run_time -= 1

Answer 1

Welcome to Stackoverflow.

Problems with thread synchronisation can be so tricky to handle that Python already has some very useful libraries specifically to handle such tasks. The primary such library is queue.Queue in Python 3. The idea is to have a queue for each "worker" thread. The main thread collect and put new data onto a queue, and have the subsidiary threads get the data from that queue.

When you call a Queue's get method its normal action is to block the thread until something is available, but presumably you want the threads to continue working on the current inputs until new ones are available, in which case it would make more sense to poll the queue and continue with the current data if there is nothing from the main thread.

I outline such an approach in my answer to this question, though in that case the worker threads are actually sending return values back on another queue.

The structure of your worker threads' run method would then need to be something like the following pseudo-code:

def run(self):
    request_data = self.inq.get()  # Wait for first item
    while True:
        process_with(request_data)
        try:
            request_data = self.inq.get(block=False)
        except queue.Empty:
            continue

You might like to add logic to terminate the thread cleanly when a sentinel value such as None is received.