What is an efficient way that I can trigger some function once the length of a list changes by a certain amount?
I have a nested list to which I add data 100 times per second, and I want to trigger a function once the length of the list increased by some value. I tried doing this with an if statement inside a while loop (see my_loop() below). This works, but this seemingly simple operation takes up 100% of one of my CPU cores. It seems to me that constantly querying the size of the list is the limiting factor of the script (adding data to the list in the while loop is not resource-intensive).
Here is what I have tried so far:
from threading import Event, Threadimport timedef add_indefinitely(list_, kill_signal): """ list_ : list List to which data is added. kill_signal : threading.Event """ while not kill_signal.is_set(): list_.append([1] * 32) time.sleep(0.01) # Equivalent to 100 Hz.def my_loop(buffer_len, kill_signal): """ buffer_len: int, float Size of the data buffer in seconds. Gets converted to n_samples by multiplying by the sampling frequency (i.e., 100). kill_signal : threading.Event """ buffer_len *= 100 b0 = len(list_) while not kill_signal.is_set(): if len(list_) - b0 > buffer_len: b0 = len(list_) print("Len of list_ is {}".format(b0))list_ = []kill_signal = Event()buffer_len = 2 # Print something every 2 seconds.data_thread = Thread(target=add_indefinitely, args=(list_, kill_signal))data_thread.start()loop_thread = Thread(target=my_loop, args=(buffer_len, kill_signal))loop_thread.start()def stop_all(): """Stop appending to and querying the list. SO users, call this function to clean up! """ kill_signal.set() data_thread.join() loop_thread.join()Example output:
Len of list_ is 202Len of list_ is 403Len of list_ is 604Len of list_ is 805Len of list_ is 1006Advertisement
Answer
It’s not very safe to access a list from two threads, so I’ll suggest a safer way to communicate between threads. In CPython, your code won’t corrupt the contents of the list, but you might not get exactly 200 items each time you process a batch. If you started removing items from the list in my_loop(), you could run into trouble. If you use other versions of Python without the GIL, you could have more trouble.
Before that, though, here’s the smallest change I can think of to solve the problem you asked about: CPU usage. I just added a sleep to my_loop() and cleaned up the buffer calculations so it now reports a mostly steady 201, 401, 601. Occasionally, I see a 1002.
from threading import Event, Threadimport timedef add_indefinitely(list_, kill_signal): """ list_ : list List to which data is added. kill_signal : threading.Event """ while not kill_signal.is_set(): list_.append([1] * 32) time.sleep(0.01) # Equivalent to 100 Hz.def my_loop(buffer_len, kill_signal): """ buffer_len: int, float Size of the data buffer in seconds. Gets converted to n_samples by multiplying by the sampling frequency (i.e., 100). kill_signal : threading.Event """ buffer_len *= 100 b0 = len(list_) while not kill_signal.is_set(): time.sleep(0.01) if len(list_) - b0 >= buffer_len: b0 += buffer_len print("Len of list_ is {}".format(len(list_)))list_ = []kill_signal = Event()buffer_len = 2 # Print something every 2 seconds.data_thread = Thread(target=add_indefinitely, args=(list_, kill_signal))data_thread.start()loop_thread = Thread(target=my_loop, args=(buffer_len, kill_signal))loop_thread.start()def stop_all(): """Stop appending to and querying the list. SO users, call this function to clean up! """ kill_signal.set() data_thread.join() loop_thread.join()time.sleep(30)stop_all()Now, to do this safely, I suggest you use a queue. That will allow many threads to read or write to the queue, and it will handle the communication. If a thread tries to read from an empty queue, it just blocks until some other thread adds an item to the queue.
I wasn’t sure exactly what you wanted to do with the items, so I just put them in a list and left them there. However, now that the list is only being accessed by one thread, it’s safe to clear it out after each batch of 100 items are processed.
Because the my_loop() is now blocking, it won’t necessarily notice when you set the kill signal. Instead, I used a sentry value of None in the request queue to tell it to shut down. If that doesn’t work for you, you can use a timeout when getting an item from the queue, check the kill signal, and then try getting an item again.
from threading import Event, Threadfrom queue import Queueimport timedef add_indefinitely(request_queue, kill_signal): """ list_ : list List to which data is added. kill_signal : threading.Event """ while not kill_signal.is_set(): request_queue.put([1] * 32) time.sleep(0.01) # Equivalent to 100 Hz. request_queue.put(None) # Signal to shut downdef my_loop(buffer_len, kill_signal): """ buffer_len: int, float Size of the data buffer in seconds. Gets converted to n_samples by multiplying by the sampling frequency (i.e., 100). kill_signal : threading.Event """ received_items = [] # replaces list_ buffer_len *= 100 while True: item = request_queue.get() if item is None: break received_items.append(item) if len(received_items) % buffer_len == 0: print("Len of received_items is {}".format(len(received_items)))request_queue = Queue()kill_signal = Event()buffer_len = 2 # Print something every 2 seconds.data_thread = Thread(target=add_indefinitely, args=(request_queue, kill_signal))data_thread.start()loop_thread = Thread(target=my_loop, args=(buffer_len, kill_signal))loop_thread.start()def stop_all(): """Stop appending to and querying the list. SO users, call this function to clean up! """ kill_signal.set() data_thread.join() loop_thread.join()time.sleep(30)stop_all()