A simplified libfbp component and a general purpose work-stealing, back-off threadpool to get things going.

photopin cc

Sucked into a general purpose threadpool!

TL;DR To paraphrase Thoreau "Decouple, decouple, decouple!" the clumsy component that needed to take a dequeuer is simplified to a virtual interface PPM API facilitating generalisation for PXMs. There's also a threadpool PXM so that FBP can run out of the box.

As previously stated the primary strategic goal here is that the libfbp implementation of Flow-Based Programming (FBP) as a Parallel Programming Model (PPM), though the FBP paradigm is a lot more more than simply a PPM, should be agnostic to its underlying Program eXecution Model (PXM) but must, never-the-less, provide an Application Programming Interface (API) that remains versatile in its implementation. 

In an effort to achieve this goal the original component here was overly complex and took on too much of the scheduling burden the simplified component presents a single virtual operator()() member function that returns a component*. The API contract is simple, the returned component* points to the next component to be executed and a returned value of null_ptr removes the component from the execution pipeline.

Part (2/3): A general purpose thread-safe internally synchronized message queue.

When it's time to end the relationship.

So here is the code for the single lock general purpose message queue as the obviously correct base queue type, but with one important addition...

The ability to close() the message queue.

But why?

TL;DR github.com/ifknot/libfbp

Part (1/3): A general purpose thread-safe internally synchronized message queue.

"The English can form a queue of 1"

Flow-Based Programming (FBP) [1], Communicating Sequential Processes (CSP) [2] and Active Objects (AO) [3] rely on message passing between otherwise environmentally opaque conccurent components. 

There are lots of ways of providing this communication[4], be that synchronous or asynchronous, but one of the most frequently used is that of a message queue.

The principle reference for my development of the concurrent queues is the, totally excellent, The Art of Multiprocessor Programming (Herlihy & Shavit 2008)

Part (1/2): Active Objects for Asynchronous Logging.

Asynchronous physical humour 

In his 2010 Dr Dobbs article[1] Herb Sutter espouses

"Prefer Using Active Objects Instead of Naked Threads"

The laudability of this is derived from Sutter's 2009 and 2007 observations[2],[3] that

"Use Threads Correctly = Isolation + Asynchronous Messages" 

(N.B. It is fascinating to read Sutter's 2009[2] & 2007[3] similarities to Sir Tony Hoare[6] 1978 seminal work on Communicating Sequential Processes(CSP)[7] and J. Paul Morrison[8] 1970s original ideas on Flow-Based Programming(FBP)[9] but more of this later...)

TL;DR Think you can walk-the-walk? You don't know Jack about thread programming if you haven't read [3], [2] & [1]