[Python] Problems while using python queue.Queue() with multiple threads. Kindly explain

The objective of my python program is to take-in microphone input and convert it to frequencies and the amplitude of the frequencies.

This is the basic flow of the program:

1. Take-in input from microphone
2. Convert the microphone input (which is in bytes) to int values
3. Take the converted values and perform an FFT/DFT
4. Take the FFT values (these contain frequencies and their amplitudes) and write them to a file

Understanding the program in more details:

The main thread is taking the input from microphone and writing it to a queue.Queue() (mic_input_queue variable in the code). mic_input_queue is read by a separate thread (byte_to_amplitude_converter_thread) that runs byte_to_amplitude_converter function. byte_to_amplitude_converter function converts the values in the mic_input_queue (which are python byte values like b'\xff\xa0') to int values. These integer values are then written to another queue.Queue() that is named converted_to_amplitudes_queue. Each member of converted_to_amplitudes_queue is a list (amplitudes_chunk in the code). The byte_to_amplitude_converter function also checks and starts the gpu_handler_thread which runs the gpu_handler_function function. The gpu_handler_function function sends the values from converted_to_amplitudes_queue which is the amplitudes_queue in the gpu_handler_function. Each element in amplitudes_queue is saved in current_set which is then sent to run be stored and run in the gpu. The GPU function/kernel fft_computations (with the @cuda.jit decorator) runs the FFT on the gpu and the result is then stored in the variable (another queue.Queue())converted_to_frequencies_queue. the gpu_handler_function then runs file_writer_thread and keeps checking the status of the thread file_writer_thread to run it. The information from converted_to_frequencies_queue is written to a file by file_writer function run by the file_writer_thread. (Sorry for avoiding pronouns, I tried to be as clear as I can be).

Problems

Where I encounter problems is when trying to get data from the queues (mic_input_queue, converted_to_amplitudes_queue, amplitudes_queue, converted_to_frequencies_queue). I don't get any data even though it has been inserted into the queues. Secondly, I would also like to know if I have to explicitly use Queue.task_done() method? Where and when am I supposed to use it? I tried to use after each get() but I encounter errors (ValueErrors as given in screenshots below) which I think I should not. Finally, I am also very dissatisfied by the explanations on the official python documentation for the queue.Queue() class. official python queue.Queue() documentation. queues have internal variables like mutex, unfinished_tasks. Am I supposed to manually modify these variables when I am using the queue? Say after a get() or a put() call.

Other comments

This is the code below run on windows 11 using powershell. I am using python 3.13 along with libraries numpy, numba (for cuda, parallelization), pyaudio (for getting data from microphone), msvcrt(available on windows only) among other python standard libs. The abundant print statements are there to help understand where the program has reached.

Additional details

# this is the main program------------------------------------------------------------------
import pyaudio as pya
import numpy as np
from time import sleep
import CustomThread
from msvcrt import kbhit
import queue as q
from numba import cuda
import math
from ThreadTerminationInformation import ThreadTerminationInformation

COUNT_OF_NVIDIA_RTX_2060_STREAMING_MULTIPROCESSORS = 30 # Q1
COUNT_OF_NVIDIA_RTX_2060_CUDA_CORES_PER_SM = 64 # Q2
# Q1 * Q2 = 1920. The amount of 'cores' nvidia lists for RTX 2060
THREADS_PER_BLOCK = COUNT_OF_NVIDIA_RTX_2060_CUDA_CORES_PER_SM
FORMAT = pya.paInt16
CHANNELS = 1
RATE = 48000
SAMPLES_WITHIN_A_SECOND_WHILE_PRONOUNCING_AN_ALPHABET = math.ceil(RATE * .2)
BLOCKS_PER_GRID = math.ceil(SAMPLES_WITHIN_A_SECOND_WHILE_PRONOUNCING_AN_ALPHABET/THREADS_PER_BLOCK)
# 48000 samples per second multiplied by .2 (200 milliseconds)
# I take around 200 milliseconds to pronounce and alphabet
CHUNK_SIZE = SAMPLES_WITHIN_A_SECOND_WHILE_PRONOUNCING_AN_ALPHABET # THREADS_PER_BLOCK * COUNT_OF_NVIDIA_RTX_2060_STREAMING_MULTIPROCESSORS

pre_calculated_exponents = np.zeros(shape=(1, SAMPLES_WITHIN_A_SECOND_WHILE_PRONOUNCING_AN_ALPHABET),
dtype=np.float16)
pre_calculated_exponents_on_gpu = ''
converted_to_amplitudes_queue = q.Queue()
converted_to_frequencies_queue = q.Queue()
threads_in_this_program = dict()
threads_in_this_program['byte_to_amplitude_converter_thread'] = CustomThread.CustomThread()
threads_in_this_program['gpu_handler_thread'] = CustomThread.CustomThread()
threads_in_this_program['file_writer_thread'] = CustomThread.CustomThread()
terminate_threads = ThreadTerminationInformation()
terminate_threads.acquire(blocking=True, timeout=-1)
terminate_threads.terminate = False
terminate_threads.release()

@cuda.jit
def initialize_fft_computation_assists(a_pre_calculated_exponents_on_gpu,
a_SAMPLES_WITHIN_A_SECOND_WHILE_PRONOUNCING_AN_ALPHABET):
"""
This function calculates some arguments for cosine and sine mathematical functions.
This calculation is done on the GPU and stored on the GPU.
"""
base_computation = ((cuda.threadIdx.x) + (cuda.blockIdx.x * THREADS_PER_BLOCK))
y_index = base_computation
if y_index >=\
a_SAMPLES_WITHIN_A_SECOND_WHILE_PRONOUNCING_AN_ALPHABET:
a_pre_calculated_exponents_on_gpu[0][y_index] =\
2 * math.pi * (y_index/a_SAMPLES_WITHIN_A_SECOND_WHILE_PRONOUNCING_AN_ALPHABET)


def file_writer(a_converted_to_frequencies_queue):
"""
This function just writes the parameter0 to a file.
The function will be run on a separate thread.
"""
global terminate_threads
try:
audio_frequencies_file = open('audio_frequencies.txt', 'a')
except FileNotFoundError:
audio_frequencies_file = open('audio_frequencies.txt', 'w')
audio_frequencies_file.close()
audio_frequencies_file = open('audio_frequencies.txt', 'r+')
line = ''
while True:
print('fw')
try: first_element = a_converted_to_frequencies_queue.get(block=False)
except: pass
a_converted_to_frequencies_queue.task_done()
if len(first_element) > 0:
audio_frequencies_file.write('<af>\n')
# the <af> tags are: 'audio/<wave properties> frequencies'
line = ''
for index0 in range(0,first_element.shape[0]):
line = line + str(first_element[index0]) + ','
if line[len(line)-1] == ',': line = line[0:len(line)-1]
audio_frequencies_file.write(line + '\n')
audio_frequencies_file.write('<\\af>\n')
# the <af> tags are: 'audio/<wave properties> frequencies'
sleep(0.05)
print('checking fw termination')
if terminate_threads.terminate == True and a_converted_to_frequencies_queue.qsize() > 0: break
print('continuing fw')
sleep(0.05)
audio_frequencies_file.close()
print('exiting file writer')

def byte_to_amplitude_converter(amplitudes_as_bytes):
"""
The PyAudio library only outputs in byte values.
So in python you get something like: b'\xff\x00'.
That value is converted to numbers here
"""
starting_index = 0
amplitudes_chunk = list()
global threads_in_this_program
global converted_to_amplitudes_queue
while True:
#print('b2ac')
try: one_chunk = amplitudes_as_bytes.get(block=False)
except: pass
amplitudes_as_bytes.task_done()
if len(one_chunk) > 0:
starting_index = 0
amplitudes_chunk.clear()
while starting_index <= len(one_chunk)-2:
one_sample_as_bytes = one_chunk[starting_index:starting_index+2]
one_sample_as_amplitude = int.from_bytes(one_sample_as_bytes, byteorder='big',signed=True)
amplitudes_chunk.append(one_sample_as_amplitude)
starting_index = starting_index + 2
if len(amplitudes_chunk) == SAMPLES_WITHIN_A_SECOND_WHILE_PRONOUNCING_AN_ALPHABET:
converted_to_amplitudes_queue.put(amplitudes_chunk)
converted_to_amplitudes_queue.task_done()
amplitudes_chunk.clear()
if converted_to_amplitudes_queue.qsize() > 0 and \
not threads_in_this_program['gpu_handler_thread'].is_alive():
print('starting gpu handler thread')
threads_in_this_program['gpu_handler_thread'] = CustomThread.CustomThread(target=gpu_handler_function)
threads_in_this_program['gpu_handler_thread'].prerun(
args=(converted_to_amplitudes_queue)
)
threads_in_this_program['gpu_handler_thread'].start()
#amplitudes_as_bytes.task_done()
sleep(0.05)
#print('checking b2ac termination')
if terminate_threads.terminate == True and amplitudes_as_bytes.qsize() == 0: break
#print('continuing b2ac')
sleep(0.05)
print('exiting byte to amplitude converter')



def gpu_handler_function(a_amplitudes_queue):
"""
This is a function that runs on a separate thread.
Responsible for taking in amplitudes from the amplitude queue
and converting them to frequencies (frequencies and the coefficient for each frequency).
"""
amplitudes_queue = a_amplitudes_queue
global pre_calculated_exponents_on_gpu
global threads_in_this_program
global converted_to_frequencies_queue
global terminate_threads
cuda_streams = [cuda.stream(), cuda.stream(), cuda.stream()]
target_stream = 0
print('running gpu handler')
while True:
print('ghf')
try: current_set = amplitudes_queue.get(block=False)
except: pass
amplitudes_queue.task_done()
if len(current_set) > 0:
print('---------------------------------------------------------')
print(len(current_set))
current_set = np.asarray(a=current_set)
print(current_set.shape)
sample_amplitudes_chunk_array_on_gpu = cuda.to_device(current_set,
stream=cuda_streams[target_stream],
copy=True)
sample_fft_coefficients_chunk_array_on_gpu = cuda.device_array(dtype=np.float32, shape=(2, sample_amplitudes_chunk_array_on_gpu.shape[0]))
# fft computation section starts
print(math.ceil(SAMPLES_WITHIN_A_SECOND_WHILE_PRONOUNCING_AN_ALPHABET/THREADS_PER_BLOCK))
print(THREADS_PER_BLOCK)
#print(math.ceil(SAMPLES_WITHIN_A_SECOND_WHILE_PRONOUNCING_AN_ALPHABET/THREADS_PER_BLOCK) * THREADS_PER_BLOCK)
print(sample_amplitudes_chunk_array_on_gpu.shape)
print(sample_fft_coefficients_chunk_array_on_gpu.shape)
print('---------------------------------------------------------')
fft_computations[BLOCKS_PER_GRID,THREADS_PER_BLOCK
,cuda_streams[target_stream]
](sample_amplitudes_chunk_array_on_gpu,
sample_fft_coefficients_chunk_array_on_gpu,
pre_calculated_exponents_on_gpu,
SAMPLES_WITHIN_A_SECOND_WHILE_PRONOUNCING_AN_ALPHABET)
sample_fft_coefficients_on_cpu = sample_fft_coefficients_chunk_array_on_gpu.copy_to_host(
stream=cuda_streams[target_stream]
)
target_stream = (target_stream + 1) % 3
# fft computation section ends
converted_to_frequencies_queue.put(sample_fft_coefficients_on_cpu)
converted_to_frequencies_queue.task_done()
if converted_to_frequencies_queue.qsize() > 0 and\
not threads_in_this_program['file_writer_thread'].is_alive():
print('starting file writer thread')
threads_in_this_program['file_writer_thread'] = CustomThread.CustomThread(target=file_writer)
threads_in_this_program['file_writer_thread'].prerun((converted_to_frequencies_queue))
threads_in_this_program['file_writer_thread'].start()
#print(amplitudes_queue.unfinished_tasks)
sleep(0.05)
print('checking ghf termination')
if terminate_threads.terminate == True and amplitudes_queue.qsize() == 0: break
print('continuing ghf')
sleep(0.05)
print('exiting gpu handler')

@cuda.jit
def fft_computations(amplitudes, frequencies, pre_calculated_exponents, samples_count):
"""
This is where the actual dft computation happens.
X_k = sum(n=0->n=N| x_n * (euler'sconstant)^(k * n / DFT_SIZE))
DFT_SIZE is SAMPLES_WITHIN_A_SECOND_WHILE_PRONOUNCING_AN_ALPHABET value
where k varies from 0 to N.
"""
y_index = cuda.threadIdx.x + THREADS_PER_BLOCK * cuda.blockIdx.x
if y_index >= samples_count: return
for index0 in range(0, samples_count):
argument = pre_calculated_exponents[0][y_index] * (index0)
frequencies[0][index0] = amplitudes[index0] * math.sqrt((math.cos(argument) ** 2) + (math.sin(argument) ** 2))
frequencies[1][index0] = index0 * (RATE / samples_count)

def convert_audio_to_frequencies():
"""
This is where the core calculation happens.
Takes input amplitude from microphone(s).
Does FFT using numpy.
Dump the data into frequency format for the model.
FORMAT: This is the resolution of data collected values can one of take 2^16 values (from 0 to (2^16)-1)
CHANNELS: In this case how many input channels to take data from. Depends on system hardware
RATE: Sampling rate of data (how many times data collected per second)
CHUNK: group of samples
"""
global pre_calculated_exponents
global pre_calculated_exponents_on_gpu
pre_calculated_exponents_on_gpu = cuda.to_device(pre_calculated_exponents)
initialize_fft_computation_assists[math.ceil((SAMPLES_WITHIN_A_SECOND_WHILE_PRONOUNCING_AN_ALPHABET)/64), 64](pre_calculated_exponents_on_gpu, SAMPLES_WITHIN_A_SECOND_WHILE_PRONOUNCING_AN_ALPHABET)
audio_facilities = pya.PyAudio()
microphone_input = audio_facilities.open(format=FORMAT,
channels=CHANNELS,
rate=RATE,
input=True,
frames_per_buffer=CHUNK_SIZE)
print('The recording has begun')
mic_input_queue = q.Queue()
while True:
mic_input_queue.put(microphone_input.read(CHUNK_SIZE))
#mic_input_queue.task_done()
if mic_input_queue.qsize() > 0 and\
not threads_in_this_program['byte_to_amplitude_converter_thread'].is_alive():
print('starting byte converter thread')
threads_in_this_program['byte_to_amplitude_converter_thread'] = CustomThread.CustomThread(target=byte_to_amplitude_converter)
threads_in_this_program['byte_to_amplitude_converter_thread'].prerun(mic_input_queue)
threads_in_this_program['byte_to_amplitude_converter_thread'].start()
print('Main thread running')
sleep(0.004)
break
if kbhit(): break
#print('The recording has ended')
microphone_input.stop_stream()
microphone_input.close()
audio_facilities.terminate()
#print('setting termination')
terminate_threads.acquire(blocking=True, timeout=-1)
terminate_threads.terminate = True
terminate_threads.release()
#print('set termination')
print(terminate_threads.terminate)
while True:
print('waiting on other threads to finish: ')
print('bacthread: ' + str(threads_in_this_program['byte_to_amplitude_converter_thread'].is_alive()))
print('ghthread: ' + str(threads_in_this_program['gpu_handler_thread'].is_alive()))
print('fwthread: ' + str(threads_in_this_program['file_writer_thread'].is_alive()))
print('aqtr: ' + str(converted_to_amplitudes_queue.unfinished_tasks))
print('fqtr: ' + str(converted_to_frequencies_queue.unfinished_tasks))
sleep(5)
print('exiting')

if __name__ == '__main__':
convert_audio_to_frequencies()
# end of main program------------------------------------------------------------------
# These are supporting classes that I am using. They are in their own separate files
import threading as ts

class CustomThread(ts.Thread):
"""
Custom implementation of the original python class 'Thread'.
This implementation allows return of value from the function executed
by the thread.
"""
def __init__(self, target=None, args=()):
super().__init__(group=None, target=target, args=args)
self.result = None
self.args = args

def run(self) -> None: self.result = self._target(self.args)

def prerun(self, args): self.args = args
# ---------------------------------------------------------
import threading

class ThreadTerminationInformation(threading._RLock):
def __init__(self):
super().__init__()
self.terminate = False


This is the error from the latest run. I have split the powershell text into two screenshots.



EDIT: here is the error message as text

(newpro) PS C:\Users\vishw\Desktop\newpro> python .\convert_audio_to_frequencies.py
The recording has begun
starting byte converter thread
starting gpu handler thread
Main thread running
running gpu handler
ghf
Exception in thread Thread-5 (gpu_handler_function):
Traceback (most recent call last):
File "C:\Users\vishw\AppData\Local\Programs\Python\Python313\Lib\threading.py", line 1041, in _bootstrap_inner
self.run()
~~~~~~~~^^
File "C:\Users\vishw\Desktop\newpro\CustomThread.py", line 14, in run
def run(self) -> None: self.result = self._target(self.args)
~~~~~~~~~~~~^^^^^^^^^^^
File "C:\Users\vishw\Desktop\newpro\convert_audio_to_frequencies.py", line 156, in gpu_handler_function
amplitudes_queue.task_done()
~~~~~~~~~~~~~~~~~~~~~~~~~~^^
File "C:\Users\vishw\AppData\Local\Programs\Python\Python313\Lib\queue.py", line 93, in task_done
raise ValueError('task_done() called too many times')
ValueError: task_done() called too many times
Exception in thread Thread-4 (byte_to_amplitude_converter):
Traceback (most recent call last):
File "C:\Users\vishw\AppData\Local\Programs\Python\Python313\Lib\threading.py", line 1041, in _bootstrap_inner
self.run()
~~~~~~~~^^
File "C:\Users\vishw\Desktop\newpro\CustomThread.py", line 14, in run
def run(self) -> None: self.result = self._target(self.args)
~~~~~~~~~~~~^^^^^^^^^^^
File "C:\Users\vishw\Desktop\newpro\convert_audio_to_frequencies.py", line 107, in byte_to_amplitude_converter
amplitudes_as_bytes.task_done()
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~^^
File "C:\Users\vishw\AppData\Local\Programs\Python\Python313\Lib\queue.py", line 93, in task_done
raise ValueError('task_done() called too many times')
ValueError: task_done() called too many times
/* #####################################
The ValueError above is one of the problems.
*/
True
waiting on other threads to finish:
bacthread: False
ghthread: False
fwthread: False
aqtr: 0
fqtr: 0
waiting on other threads to finish:
bacthread: False
ghthread: False
fwthread: False
aqtr: 0
fqtr: 0
waiting on other threads to finish:
bacthread: False
ghthread: False
fwthread: False
aqtr: 0
fqtr: 0
Traceback (most recent call last):
File "C:\Users\vishw\Desktop\newpro\convert_audio_to_frequencies.py", line 272, in <module>
convert_audio_to_frequencies()
~~~~~~~~~~~~~~~~~~~~~~~~~~~~^^
File "C:\Users\vishw\Desktop\newpro\convert_audio_to_frequencies.py", line 268, in convert_audio_to_frequencies
sleep(5)
~~~~~^^^
KeyboardInterrupt
(newpro) PS C:\Users\vishw\Desktop\newpro>

Continue reading...