Editor's note: The author describes how to combine the data-centric publish/subscribe-based Data Distribution Service model with the Leader/Followers thread-pool concurrency thread management pattern.
As recent technology trends usher us into the many-core era, we need novel techniques that allow high-performance embedded applications to exploit massive local concurrency. To position software applications to do more on machines with more cores—i.e., re-enabling the proverbial free lunch—requires substantial restructuring of the embedded software stacks, which include applications, middleware, and the operating system. New operating system and middleware mechanisms are required to handle multithreading, scheduling, resource-sharing, and communication in many-core systems.
As contemporary many-core processors have cores in high double digits, keeping all the cores busy is not simple. Existing software stacks are rarely designed to adapt and scale on machines with up to a dozen cores. multithreading is a popular choice for implementing concurrency in infrastructure software. It is, however, extremely hard to get right without the systematic use of the proven practices and patterns of managing concurrency.
Concurrency is that elephant in the story of blindfolded men who make wildly different perceptions about the elephant depending on where they touch it. That is, concurrency has many dimensions. Further, whether you are modernizing an existing code base or starting with a clean slate will determine your perception.
Nevertheless, the enduring solution is likely to use timeless techniques as the foundation, combined with modern technology. So Real-Time Innovations (RTI) and University of North Carolina (UNC) came up with just that when we set out on a path-finding mission to modernize an existing data distribution middleware.
We found four promising ways, which often intersect:
- Concurrency patterns for effective multithreading,
- Component-based software design for scalable many-core applications,
- Hardware-accelerated messaging transports for inter-core communication, and
- Smart scheduling for many-core processors.
Concurrency patterns are specific multithreading techniques that help improve responsiveness and the overall throughput. The component-based software design supports effective data and functional partitioning, which is the key to enabling shared-nothing parallel programming—an enduring principle. Further, it helps developers specify dataflow requirements between the components and manage their lifecycle. Modern messaging transports such as Tilera iLib  library and Multicore Communication API (MCAPI)  enable message-passing between cores. When high-throughput networks are available directly in the processor, the inter-core communication latency is reduced to just a few cycles.
Finally, smart scheduling algorithms use the dataflow requirements to assign components to clusters of cores to efficiently use the underlying processing capacity and minimize data movement.
Concurrency Patterns for Effective multithreading
Adopting multithreading best practices and patterns often results in a high ROI if your infrastructure software is already multithreaded. Concurrency patterns  help infrastructure software scale on multicore platforms. A large number of research papers, articles, and books have been written on improving concurrency using explicit multithreading. It is used widely and we expect its use to grow as the concurrency patterns are better understood.
The Leader/Followers  concurrency pattern is a proven thread-management technique that allows multiple threads to take turns and share a set of event sources in order to detect, dispatch, and process service requests. Little or no synchronization between threads is necessary when they execute logically independent service requests. This pattern also minimizes latency because of the multiple threads.
We applied the Leader/Followers thread-pool in RTI Connext DDS to support concurrent DataReaders. Figure 1 shows a simplified test scenario where process #1 is publishing various shapes objects to process #2. Process #2 was tested with up to 10 concurrent subscribers served by 10 threads organized in the Leader/Followers fashion. This initial thread-pool setup helped achieve a nearly 250% increase in the overall messaging throughput when tested on a CPU-bound, mostly data-parallel work-load.
Data Distribution Service (DDS)  is a standard managed by the Object Management Group (OMG). Several satellite standards have been defined around DDS, including C++ and Java APIs   and a wire-interoperability protocol . DDS defines a data-centric publish-subscribe architecture for connecting anonymous information providers with information consumers. Like SOA, DDS promotes loose coupling between system components. A distributed application is composed of data providers and consumers, each potentially running in a separate address space, possibly on different computers. A data provider publishes typed data-flows, identified by names called topics, to which consumers can subscribe. RTI Connext DDS is the industry-leading implementation of the DDS standard.
Component-based software design for shared-nothing parallel programming
Designing scalable, adaptable, and maintainable applications for many-core architectures requires three key constituents:
- Functional partitioning without shared state to enable Multiple Instruction Multiple Data (MIMD)-style parallelism
- High-throughput messaging infrastructure for inter-component communication (i.e., data-flow)
- Effective resource allocation and scheduling algorithms to exploit local massive concurrency within a reasonable power budget
Component-based design is well-aligned with functional partitioning because components are designed to be modular, cohesive, and independently deployable. Components facilitate the development of pipelined software architectures, which can exploit concurrency as much as the depth of the pipeline. For example, Figure 2 shows a pipeline for an image-processing application where every block represents a component that can execute concurrently with other components. Each arrow represents an explicit messaging channel more amenable to monitoring and control than implicit communication such as function invocation. The components communicate with each other exclusively using messages and do not share state in any other way.
This architecture has roots in the Actor model , which advocates lock-free, shared-nothing concurrency and asynchronous message-passing between actors. A number of commercial and open-source platforms are using multiple agents  to simplify programming of massively parallel architectures. The next section describes how components deployed on a single many-core host can communicate efficiently without tight coupling.
High-performance data-centric messaging for intra-node communication
Shared memory is still the preferred communication method on commodity hardware with a handful of cores. This trend, however, is reversing due to performance and correctness issues on many-core architectures. Shared-memory techniques have limited scalability on many-core processors due to the overhead of cache coherence protocols. Moreover, shared-memory systems resist composability—the ability to build complex applications by composing smaller modules. Judicious use of message-passing and shared memory is the preferred way of sharing data on many-core processors.
Message passing is well aligned with component-based programming. Components that use message passing for communication provide a powerful mechanism of isolation since sharing takes place via message exchanges. Validation is simplified because the programmer only needs to check the externally observable component interaction to derive correctness properties for the system as a whole.
Chip manufacturers have developed novel processor architectures with on-chip high-performance interconnects for efficient message-passing across cores (Figure 3 ). Vendor-specific message-passing libraries (e.g., RCCE  from Intel, iLib from Tilera) and messaging standards such as MCAPI are available to program this new crop of processors.
The message-passing libraries are specialized for closely distributed environments. They are often lightweight and offer higher performance due to on-chip hardware-level support for message-passing between cores. The programming model offered by these libraries is very similar to that of the Message Passing Interface (MPI). MPI is particularly suitable for fully synchronized (i.e., matched send/receive calls) communication, which is often preferred in low-level modules because the most basic building blocks can be implemented in hardware and higher-level abstractions can be built without much overhead.
The MPI-style programming model, however, is not suitable for many real-time distributed systems where dynamic workloads, changing topologies, and stringent Quality-of-Service (QoS) requirements must be supported. The DDS programming model is particularly suitable in such environments thanks to its built-in discovery protocol, nearly two dozen QoS policies, and a publish-subscribe communication model. Table 1 summarizes the key architectural differences between DDS and MPI-style programming. DDS provides the necessary anonymous publish-subscribe programming model to create large-scale real-time distributed systems.
A new technology is needed to combine the best of both worlds. When multiple communicating applications are deployed on a single many-core host, the infrastructure should use the most appropriate communication mechanism to share data. The details of low-level APIs should be hidden from the applications so that a different transport can be plugged-in without changing the application sources. For example, RTI Connext DDS provides a powerful data-centric publish-/subscribe programming model for real-time applications. When these applications run on a single many-core host, a variant of the pluggable transport  layer abstracts the details of the low-level transport. Through the pluggable transport, RTI Connext DDS uses the MPI-style messaging primitives internally. Applications use the standard DDS API without losing efficiency and/or loose coupling on many-core platforms.
Table 1: Key architectural differences between DDS and MPI-style programming APIs
RTI, in collaboration with UNC, implemented a prototype transport using OpenMCAPI—an implementation of the MCAPI standard. Using this transport plugin, we were able to execute existing applications and RTI infrastructure services without any change to the source. This novel solution allows applications to use the most suitable transport on a given platform without changing the programming model and/or the source code. As operating systems gain more capabilities for many-core systems, the middleware and applications can take advantage of these capabilities simply by developing new pluggable transports.
Many-Core Resource Allocation and Scheduling Algorithms
The infrastructure middleware must ensure an optimal allocation of concurrent components to clusters of cores and schedule their execution so that the instruction throughput can be maximized. RTI and UNC developed new techniques for efficient allocation of flow-based computations to enable scalable performance on a many-core platform (Intel Single-Chip Cloud Computer).
Processing Graph Method (PGM)  is an expressive formalism for representing flow-based computations. In PGM, a computation is represented as a directed acyclic graph (DAG), in which each vertex is a task (sequential program) and each edge is a typed, first-in first-out (FIFO) queue that connects a producing task with a consuming task; such queues abstract message communication. We are leveraging prior work on PGM graphs to determine how to best allocate flow-based computations on many-core platforms. Specifically, we specify application-specific component assembly as PGM graphs, to schedule them using deadline-based clustered scheduling algorithms, and to analyze them for schedulability assuming that deadlines are “soft” and can be missed by a bounded amount. Note that if bounded deadline tardiness is ensured, long-term processing rates will be as prescribed.
RTI and UNC are developing techniques for assigning PGM nodes to clusters of cores within a many-core platform. Such techniques are extensions of similar existing techniques  for clustered scheduling of PGM graphs in networked systems. These techniques will utilize the cores efficiently and will avoid excessive data movement across cores.
In summary, hundreds of cores will soon stop being a novelty. The onus is on the programmers to get the most out of the processor, which is essentially a supercomputer. Data-centric middleware offers a powerful programming model combined with scalability and flexibility necessary to cope with the ever-changing landscape of many-core processors.
This paper was presented as part of a class taught at the Spring 2012 ESC DESIGN West by Dr. Sumant Tambe and Dr James H. Anderson.
Dr. Sumant Tambe is Senior Software Research Engineer at Real-Time Innovations, specializing in standards-based data-centric and component-oriented middleware for distributed, real-time, and embedded (DRE) systems. His expertise includes static and adaptive fault-recovery protocols for component-based, multi-tier real-time systems, publish-subscribe middleware, real-time embedded systems, model-driven deployment and configuration of QoS-enabled middleware, and techniques for component behavior and dependability modeling. He has a Ph.D. in Computer Science from Vanderbilt University and an M.S. degree in Computer Science from New Mexico State University. He can be contacted at .
Prof. James H. Anderson is a professor in the Department of Computer Science at the University of North Carolina at Chapel Hill. He received a B.S. in Computer Science from Michigan State University in 1982, an M.S. in Computer Science from Purdue University in 1983, and a Ph.D. in Computer Sciences from the University of Texas at Austin in 1990. His main research interests are within the areas of concurrent and distributed computing and real-time systems. Before joining UNC-Chapel Hill in 1993, he was with the Computer Science Department at the University of Maryland between 1990 and 1993. In 1995, Dr. Anderson received the U.S. Army Research Office Young Investigator Award, and in 1996, he was named Alfred P. Sloan Research Fellow.
1. David Wentzlaff, Patrick Griffin, Henry Hoffmann, Liewei Bao, Bruce Edwards, Carl Ramey, Matthew Mattina, Chyi-Chang Miao, John F. Brown III, Anant Agarwal, “On-chip Interconnection Architecture of the Tile Processor”, IEEE Micro , 27(5):15–31, 2007
2. The Multicore Communications API (MCAPI). Accessed September 2012
3. Douglas C. Schmidt, Carlos O’Ryan, Michael Kircher, Irfan Pyarali, and Frank Buschmann; Leader/Followers . Retrieved September 2012
4. The Data Distribution Service specification , v1.2.
5. DDS-PSM-Cxx: ISO/IEC C++ 2003 Language DDS PSM
6. DDS-Java: Java 5 Language PSM for DDS
8. D. J. Kaplan, “An introduction to the processing graph method”. In the Proceedings of Engineering of Computer-based Systems , 1997
9. C. Liu and J. Anderson. “Supporting graph-based real-time applications in distributed systems”. In the Proceedings of the 17th IEEE International Conference on Embedded and Real-Time Computing Systems and Applications . IEEE Press, August 2011
10. Gul Agha, Actors: A Model of Concurrent Computation in Distributed Systems . Doctoral Dissertation. MIT Press, 1986
11. Michael Wooldridge, An Introduction to Multi-Agent Systems . Publisher John Wiley & Sons Ltd, 2009
12. RCCE: A Small Library for Many-Core Communication (RCCE Specification),
13. Douglas C. Schmidt, Carlos O’Ryan, Ossama Othman, Fred Kuhns, and Jeff Parsons, “Applying Patterns to Develop a Pluggable Protocols Framework for ORB Middleware”
14. Douglas Schmidt, Michael Stal, Hans Rohnert, Frank Buschmann, Pattern-Oriented Software Architecture Volume 2: Patterns for Concurrent and Networked Objects , Wiley 2000, ISBN-10: 0471606952