One producer thread and multiple non-blocking simultaneous consumer threads (MROW) python
Question: I have one producer thread reading some data and processing them and other
threads reading that that, I was using a condition variable in the producer in
order to avoid busy wait and notify when the new adquisition of data is ready,
but every consumer takes the lock when they wanna read that data so the other
consumer threads have to wait, what would be the better aproach to get all the
consumers read the data at once when the producer is done?,

Here is some of the code:

The produccer:

    
    
    condition = threading.Condition()
    
    
    def get_data():
    while(1):
        global actumulated_data, sdev1, mean1, number_of_adq, instant_data
        time.sleep(1.3660619/1000000.0)
        number_of_adq = number_of_adq + 1
        fpga.write_int('adc_data_ctrl',0)
        fpga.write_int('adc_data_ctrl',7)
        data = fpga.read('adc_data_bram',8192,0)
        data2 = map(hex, map(ord, data))
        for i in range(8192):
            data2[i] = fix8_7(twos_comp(int(data2[i],16),8))
        condition.acquire() # condition variable
        instant_data = numpy.array(data2)
        actumulated_data = numpy.concatenate((actumulated_data,instant_data),axis=0)    
        condition.notifyAll()
        condition.release()
    

One of the consumers:

    
    
    def plot_adc_out():
    while(1):
        condition.acquire()
        plt.clf()
        plt.plot(instant_data)
        plt.grid(True)
        plt.xlim(0, len(instant_data))
        plt.ylim(-1.5,1.5)
        plt.draw()
        condition.wait()
        condition.release() 
    

So there is a few consumers like this one, I was thinking on something like in
C++ with read and write locks in which if the lock is write lock is gonna be
blocking but if is a reader lock is non blocking so can be a lot of threads
reading that item at the same time.

[Readers–writer
lock](http://en.wikipedia.org/wiki/Readers%E2%80%93writer_lock)

    
    
    pthread_rwlock_wrlock
    pthread_rwlock_rdlock
    

But I belive that python dosen't have read and write locks.


Answer: The correct way for threads to communicate is by using locks, but the
queue.Queue data structure offers a much more convenient mechanism for inter-
thread communication, with all the ugly locking stuff comfortably hidden under
the covers. Here's an example adapted from from my O'Reilly School of
Technology classes that converts a string to upper case using by piping each
character to a separate thread. Note the code below is Python 3, but it is
easily adapted to Python 2.

Everything is started off by the control thread (which imports the output and
worker threads shown below from their respective modules). It first creates
the input and output queues. These are standard FIFOs, with a limit of 50%
more than the number of worker threads to avoid locking up too much memory in
buffered objects. Then it creates and starts the output thread, and finally
creates and starts as many worker threads as configured by the WORKERS
constant. Worker threads get from the input queue and put to the output queue.

    
    
    """
    control.py: Creates queues, starts output and worker threads,
                and pushes inputs into the input queue.
    """
    from queue import Queue
    from output import OutThread
    from worker import WorkerThread
    
    WORKERS = 10
    
    inq = Queue(maxsize=int(WORKERS*1.5))
    outq = Queue(maxsize=int(WORKERS*1.5))
    
    ot = OutThread(WORKERS, outq)
    ot.start()
    
    for i in range(WORKERS):
        w = WorkerThread(inq, outq)
        w.start()
    instring = input("Words of wisdom: ")
    for work in enumerate(instring):
        inq.put(work)
    for i in range(WORKERS):
        inq.put(None)
    inq.join()
    print("Control thread terminating")
    

The Worker threads have been cast so as to make interactions easy. The work
units received from the input queue are (index, character) pairs, and the
output units are also pairs. The processing is split out into a separate
method to make subclassing easier—simply override the process() method.

    
    
    """
    worker.py: a sample worker thread that receives input
               through one Queue and routes output through another.
    """
    from threading import Thread
    
    class WorkerThread(Thread):
        def __init__(self, iq, oq, *args, **kw):
            """Initialize thread and save Queue references."""
            Thread.__init__(self, *args, **kw)
            self.iq, self.oq = iq, oq
        def run(self):
            while True:
                work = self.iq.get()
                if work is None:
                    self.oq.put(None)
                    print("Worker", self.name, "done")
                    self.iq.task_done()
                    break
                i, c = work
                result = (i, self.process(c)) # this is the "work"
                self.oq.put(result)
                self.iq.task_done()
        def process(self, s):
            """This defines how the string is processed to produce a result"""
            return s.upper()
    

The output thread simply has to extract output packets from a queue where they
are placed by the worker threads. As each worker thread terminates, it posts a
None to the queue. When a None has been received from each thread, the output
thread terminates. The output thread is told on initialization how many worker
threads there are, and each time it receives another None it decrements the
worker count until eventually there are no workers left. At that point, the
output thread terminates. Since the worker threads aren't guaranteed to return
in any particular order the results can be sorted. Without sorting you can see
the order the results arrived in.

    
    
    """
    output.py: The output thread for the miniature framework.
    """
    identity = lambda x: x
    
    import threading
    class OutThread(threading.Thread):
        def __init__(self, N, q, sorting=True, *args, **kw):
            """Initialize thread and save queue reference."""
            threading.Thread.__init__(self, *args, **kw)
            self.queue = q
            self.workers = N
            self.sorting = sorting
            self.output = []
        def run(self):
            """Extract items from the output queue and print until all done."""
            while self.workers:
                p = self.queue.get()
                if p is None:
                    self.workers -= 1
                else:
                    # This is a real output packet
                    self.output.append(p)
            print("".join(c for (i, c) in (sorted if self.sorting else identity)(self.output)))
            print ("Output thread terminating"
    

