Correct usage of multithreading and Queue module in data collection application written in Python

Question

I am working on collecting data from several devices, and since the tests are long duration, I want to use Python's threading and Queue modules. I've written a short script to figure out how to use these, and it is very evident I don't understand the nuances of getting this to work.

Here is my script:

import ue9
import LJ_Util
import DAQ_Util
import threading
import Queue

from datetime import datetime
from time import sleep

queue = Queue.Queue()
now = datetime.now().isoformat()

def DAQThread(ue9ipAddr):
    print '\nExecuting in DAQThread at IP Address: %s' % ue9ipAddr
    a = ue9.UE9(ethernet=True, ipAddress=ue9ipAddr)
    SN = (a.commConfig()).get('SerialNumber')
    count = 5
    while count > 0: 
        reading = a.feedback()
        dataReturn = (SN, now, reading)
        queue.put(dataReturn)
        count -= 1
        print count
        sleep(5)


def listenThread(counter): 
    while queue.empty() != True:
        try: 
            outcome = queue.get() 
            print outcome 
            counter -=1 
            print counter 
        except: 
            return 'queue.get() command loop failing.' 


print "\nOpening device..."
ipAdd = '192.168.1.127'
feedbackThread = threading.Thread(target=DAQThread, args=(ipAdd,))
feedbackThread.start()

print "\nListening for data..."
queryThread = threading.Thread(target=listenThread, args = (10,))
queryThread.start()

print queue.get()
print(threading.activeCount())
print "\nDone"

Here is the output results from executing this script:

$ python threading-and-queue-test.py

Opening device...

Executing in DAQThread at IP Address: 192.168.1.127
Listening for data...

4
 (278956853, '2012-09-03T20:02:47.656024', {'AIN4': -0.012, 'AIN5': -0.012, 'CIODir': 0, 'AIN7': -0.012, 'EIODir': 0, 'AIN1': -0.012, 'AIN2': -0.012, 'AIN3': -0.012, 'MIOState': 7, 'AIN8': -0.012, 'AIN6': -0.012, 'AIN9': -0.012, 'CIOState': 15, 'AIN0': -0.012, 'Counter0': 0, 'Counter1': 0, 'EIOState': 255, 'TimerC': 0, 'TimerB': 0, 'TimerA': 0, 'MIODir': 0, 'FIODir': 32, 'AIN14': -0.012, 'AIN15': -0.012, 'AIN12': -0.012, 'AIN13': -0.012, 'AIN10': -0.012, 'AIN11': -0.012, 'FIOState': 255})
2

Done
3
2
1
0
$

It's clear the timing of the thread activities are 'off', but I'm not sure how to fix it, as I've never programmed with these modules before, nor used threads in general. Any comments or suggestions would be welcome. Thanks in advance!

Answer 1

As mentioned in the comment, one problem is in your listening thread. Once you "grab" an entry from queue, there are no longer any entries in queue and since you are only adding an entry every 5 seconds, your listening thread will empty the queue thus

while queue.empty() != True will evaluate False, exiting the loop

For example:

>>> import Queue
>>> q=Queue.Queue()
>>> q.put(1)
>>> q.empty()
False
>>> q.get()
1
>>> q.empty()
True
>>> q.empty()!=True
False

One way to get around this is to use another queue as a stop or cancel queue so modifying you listening thread you could do something like this:

stopQue=Queue.Queue()

def listenThread(counter): 
    while True:
        if queue.empty()!=True: 
            outcome = queue.get() 
            print outcome 
            counter -=1 
            print counter 
        if stopQue.empty()!=True:
            break
    print 'Exiting Listening Thread'

That way if you put anything in the stopQue, i.e. stopQue.put(1), it should exit.

Complete example based on your initial code. I cut out the code that does not have to do with the Queues and threading:

import threading
import Queue
from time import sleep

dataQue = Queue.Queue()
stopQue = Queue.Queue()

def DAQThread(ue9ipAddr):
    print 'Executing in DAQThread\n'
    count = 5
    while count > 0: 
        dataQue.put('data: %s' % count)
        count -= 1
        sleep(5)
    stopQue.put(1)
    print 'Exiting DAQThread\n'


def listenThread(counter): 
    while True:
        if dataQue.empty() != True:
            outcome = dataQue.get() 
            print outcome 
            counter -=1 
        if stopQue.empty() != True:
            break
    print 'Exiting Listening Thread'


print "Opening device..."
ipAdd = '192.168.1.127'
feedbackThread = threading.Thread(target=DAQThread, args=(ipAdd,))
feedbackThread.setDaemon(True)
feedbackThread.start()

print "Listening for data..."
queryThread = threading.Thread(target=listenThread, args = (10,))
queryThread.setDaemon(True)
queryThread.start()

print "Done"

Produces the output:

>>> 
Opening device...
Executing in DAQThread
Listening for data...

Donedata: 5        #Notice how the script is done however the threads are still running

>>> data: 4
data: 3
data: 2
data: 1
Exiting DAQThread
Exiting Listening Thread