Multithreaded single-reader single-writer fifo que

问题:

I need a queue for passing messages from one thread (A) to another (B), however ive not been able to find one that really does what I want, since they generally allow adding an item to fail, a case which in my situation is pretty much fatal since the message needs to be processed, and the thread really cant stop and wait for spare room.

• Only thread A adds items, and only thread B reads them
• Thread A must never block, however thread B is not performance critical, so it can
• Adding items must always succeed, so the queue cant have an upper size limit (short of running out of memory on the system)
• If the queue is empty, thread B should wait until there is an item to process

回答1:

Here's how to write a lock-free queue in C++:

http://www.ddj.com/hpc-high-performance-computing/210604448

But when you say "thread A must not block", are you sure that's the requirement? Windows is not a real-time operating system (and neither is linux, in normal use). If you want Thread A to be able to use all available system memory, then it needs to allocate memory (or wait while someone else does). The OS itself cannot provide timing guarantees any better than those you'd have if both reader and writer took an in-process lock (i.e. a non-shared mutex) in order to manipulate the list. And the worst-case of adding a message is going to have to go to the OS to get memory.

In short, there's a reason those queues you don't like have a fixed capacity - it's so that they don't have to allocate memory in the supposedly low-latency thread.

So the lock-free code will generally be less block-y, but due to the memory allocation it isn't guaranteed to be, and performance with a mutex shouldn't be all that shabby unless you have a truly huge stream of events to process (like, you're writing a network driver and the messages are incoming ethernet packets).

So, in pseudo-code, the first thing I'd try would be:

Writer:
    allocate message and fill it in
    acquire lock
        append node to intrusive list
        signal condition variable
    release lock

Reader:
    for(;;)
        acquire lock
            for(;;)
                if there's a node
                    remove it
                    break
                else
                   wait on condition variable
                endif
            endfor
        release lock
        process message
        free message
    endfor

Only if this proves to introduce unacceptable delays in the writer thread would I go to lock-free code, (unless I happened to have a suitable queue already lying around).

回答2:

Visual Studio 2010 is adding 2 new libraries which support this scenario very well, the Asynchronous Agents Library and Parallel Pattern Library.

The agents library has support or asynchronous message passing and contains message blocks for sending messages to 'targets' and for receiving messages from 'sources'

An unbounded_buffer is a template class which offers what I believe you are looking for:

#include <agents.h>
#include <ppl.h>
#include <iostream>

using namespace ::Concurrency;
using namespace ::std;

int main()
{
   //to hold our messages, the buffer is unbounded...
   unbounded_buffer<int> buf1;
   task_group tasks;

   //thread 1 sends messages to the unbounded_buffer
   //without blocking
   tasks.run([&buf1](){
      for(int i = 0 ; i < 10000; ++i)
         send(&buf1,i)
     //signal exit 
     send(&buf1,-1);
   });

   //thread 2 receives messages and blocks if there are none

   tasks.run([&buf1](){
      int result;
      while(result = receive(&buf1)!=-1)
      {
           cout << "I got a " << result << endl;
      }
   });

   //wait for the threads to end
   tasks.wait();
}

回答3:

• Why not use STL <list> or <deque> with a mutex around add/remove? Is the thread-safety of STL insufficient?

• Why not create your own (singly/doubly) linked-list-node class that contains a pointer, and have the items to be added/removed inherit from that? Thus making additional allocation unnecessary. You just frob a few pointers in threadA::add() and threadB::remove() and you are done. (While you'd want to do that under a mutex, the blocking effect on threadA would be negligible unless you did something really wrong...)

• If you're using pthreads, check out sem_post() and sem_wait(). The idea is that threadB can block indefinitely via sem_wait() until threadA puts something on the queue. Then threadA invokes sem_post(). Which wakes up threadB to do it's work. After which threadB can go back to sleep. It's an efficient way of handling asynchronous signaling, supporting things like multiple threadA::add()'s before threadB::remove() completes.

回答4:

You might want to consider your requirements - is it truly the case that A can't discard any queue items whatsoever? Or is it that you don't want B to pull two consecutive elements out of the queue that weren't consecutive items going in because that would somehow misrepresent a sequence of events?

For example, if this is some kind of data logging system, you (understandably) wouldn't want gaps in the record -- but without an unlimited memory, the reality is that in some corner case somewhere you probably could overrun your queue capacity..

In which case one solution is to have some kind of special element that can be put in the queue, which represents the case of A discovering that it had to drop items. Basically you keep one extra element around, which is null most of the time. Every time A goes to add elements to the queue, if this extra element is not null, that goes in. If A discovers there is no room in the queue, then it configures this extra element to say 'hey, the queue was full'.

This way, A never blocks, you can drop elements when the system is Very Busy, but you don't lose sight of the fact that elements were dropped, because as soon as queue space becomes available, this mark goes in to indicate where data drop occurred. Process B then does whatever it needs to do when it discovers it has pulled this overrun mark element out of the queue.