How can I create just three threads or a fixed number of threads and distribute X of lists to them

Question

Suppose that there is a method receiving data from somewhere. In this method we should create three threads or a fixed number of threads. Every piece of received data will be assigned to a thread.

In other words, if you will receive 10 lists of data (Note that the number of data items not fixed and can't be known), the first list will be assigned to the first thread, the second list to the second thread, third list to the third thread and the fourth list will start over and assign to the first thread or to any available thread and like so.

So the only number we know is the number of the threads that they will running in that method.

Note that the three threads should be running in the same time. Once a thread becomes available or finishes its task it will get the next data item and process it.

This is what I am doing now but if I have 30 lists of data, 30 threads will be created which is terrible.

threads = []
for ip in ip_list:
    for cmd in commands:
        th = threading.Thread(target=test_ssh_conn(ip,cmd), args=(ip,))  # args is a tuple with a single element
        th.start()
        threads.append(th)

for th in threads:
    th.join()

Answer 1

You can create a fixed number of threads and use a thread-safe global work queue to store tasks. While there are tasks, worker threads poll one and work on it. Once the work queue is empty, threads can re-join main.

Since Python's interpreter is single-threaded, consider using multiprocessing. The API is the same, so it's easy to switch between the two according to need.

Here's a basic example using some mock data and a function stub to simulate :

from multiprocessing import Process, Queue
from queue import Empty
from random import uniform
from time import sleep

def work():
    while 1:
        try:
            test_ssh_conn(*tasks.get(timeout=0.5))
        except Empty:
            break
            
    print("thread exiting")

def test_ssh_conn(ip, cmd):
    print("working on %s %d" % (ip, cmd))
    sleep(uniform(1.0, 2.0)) # pretend to do work
    print("done working on %s %d" % (ip, cmd))

if __name__ == '__main__':  
    thread_count = 3
    threads = []
    tasks = Queue()
    ip_list = ["172.16.0.0", "172.31.255.255", "192.168.0.0"]
    cmds = list(range(5))

    for ip in ip_list:
        for cmd in cmds:
            tasks.put((ip, cmd))
    
    for _ in range(thread_count):
        th = Process(target=work)
        threads.append(th)
        th.start()

    for th in threads:
        th.join()