... is a modular event-driven micro- and middle-ware library for equipment- and beam-based monitoring as well as feedback control systems for the FAIR Accelerator Facility (video) or any other project that may find this useful.
In a nut-shell: it provides common communication protocols, interfaces to numerical data visualisation and processing tools that shall aid accelerator engineers and physicist to write functional high-level monitoring and (semi-)automated feedback applications that interact, simplify and improve our accelerator operation or its individual sub-systems. Most notably, the focus is put on minimising the amount of boiler-plate code, programming expertise, and to significantly lower the entry-threshold that is required to perform simple to advanced monitoring or 'measure-and-correct' applications where the frame-work takes care of most of the communication, data-serialisation, data-aggregation and buffering, settings management, Role-Based-Access-Control (RBAC), and other boring but necessary control system integrations while still being open to expert-level modifications, extensions or improvements.
OpenCMW combines ZeroMQ's Majordomo with LMAX's disruptor (C++ port) design pattern that both provide a very efficient lock-free mechanisms for distributing, streaming and processing of data objects. A schematic outline of the internal architecture (local copy) is shown below:
Majordomo Broker or 'Broker': is the central authority where multiple workers can register their services, allowing clients to perform get, set or subscriptions requests. There can be multiple brokers for subset of services.
Worker: functional unit which provides one or more services that are registered with a broker. OpenCMW provides base implementations at different abstraction levels (BasicMdpWorker (low-level) and
MajordomoWorker) as well as different internal and external service workers, e.g. MajordomoRestPlugin or the broker's mmi services. Workers communicate with the broker using the OpenCMW worker protocol internally or externally via ZeroMQ sockets via inproc, tcp, udp or another suitable low-level network protocol scheme that is supported by ZeroMQ.
Endpoint: address for a service following the standardised URI convention of scheme:[//authority]path[?query][#fragment]. Services usually omit the authority part and provide only relative paths as this information is managed and added by their broker.
Each broker acts individually as a DNS for its own services as well as can forward this information to another (for the time being) central DNS Broker.
internal/mmi workers: each broker by default starts some lightweight management services as specified by the Majodomo mmi extension:
<optional broker name>/mmi.service: endpoints of all services registered at this broker<optional broker name>/mmi.openapi: openapi descriptions for the services<optional broker name>/mmi.dns: service lookup
Context: information that (if applicable) is matched to the URI's query parameter and required for every request and reply, specified by a domain object. They (partially) map to the filters used in the EventStore and the query parameters of the communication library. The context is used for (partial/wildcard) matching and can be used in the EventStore's filter config.
EventStore: based on LMAX's disruptor (C++ port) pattern, the EventStore provides datastructures and setup methods to define processing pipelines based on incoming data.
EventHandler: used to define specific internal processing steps based on EventStore events. The last EventHandler is usually also a Majordomo worker to export the processed information via the network.
Publisher: the DataSourcePublisher provides an interface to populate the EventStore
ring-buffer with events from OpenCMW, REST services or other sources.
While using disruptor ring-buffers is the preferred and most performing options, the client also supports classic patterns of registering call-back functions or returning Future<reyly objects> objects.
The OpenCMW Majordomo protocol is based on the ZeroMQ Majordomo protocol, both extending and slightly modifying it (see the comparison).
Service names must always start with /. For consistency, this also applies to the built-in MDP broker services like /mmi.service (instead of mmi.service without leading slash as in ZeroMQ Majordomo). A service name is a non-empty alphanumerical string (also allowing ., _), that must start with / but not end with /. It contain additional / to denote a hierarchical structure.
Examples:
/dashboards- valid/DeviceName/Acquisition- valid/mmi.service- validDeviceName/Acquisition- invalid (no leading slash)/DeviceName/Acquisition/- invalid (trailing slash)/a-service/- invalid (-not allowed)
The "topic" field (frame 5 in the OpenCMW MDP protocol) specifies the topic for subscriptions and GET/SET requests. It contains a URI with the service name as path and optional query parameters to specify further requests parameters and filter criteria.
Examples:
/DeviceName/Acquisition- service name only, no params/DeviceName/Acquisition?signal=test- service name (/DeviceName/Acquisition) and query/dashboards- service name (/dashboards)/dashboards/dashboard1?what=header- service name (/dashboards/dashboard1) and query
Note that the whole path is considered the service name, and that there's no additional path component denoting different entities, objects etc. neither for subscriptions nor GET/SET requests. Requesting specific objects like in the /dashboards/dashboard1 example are handled via the service-matching for requests in the broker, where a service name /dashboard/dashboard1 would match a worker /dashboard, which then can extract the dashboard1 component from the topic frame (see below).
See also the documentation for mdp::Topic.
With both the MDS/MDP-based ZeroMQ clients as well as the REST interface, a common scheme is used to map from mdp/hds/http(s) URLs used for subscriptions and requests to the OpenCMW service name and topic fields.
See, for example:
https://example.com:8080/serviceName?signal=test
\___/ \______________/\__________/ \_________/
| | | |
scheme authority path query
Here the URL is mapped by the REST interface reachable via example.com:8080 (the authority part), to the service name /serviceName (path) and the topic /serviceName?signal=test (path and query).
Other examples are:
https://example.com:8080/DeviceName/Acquisition?signal=test=> service name/DeviceName/Acquisition, topic/DeviceName/Acquisition?signal=test(REST).mds://example.com:8080/DeviceName/Acquisition?signal=test=> service name/DeviceName/Acquisition, topic/DeviceName/Acquisition?signal=test(subscription via mds).mdp://example.com:8080/dashboards/dashboard1?what=header=> service name/dashboards/dashboard1, topic/dashboards/dashboard1?what=header(Request via mdp).
The serialisers are based on a compile-time-reflection that efficiently transform domain-objects to and from the given wire-format (binary, JSON, ...).
Compile-time reflection will become part of C++23 as described by David Sankel et al. , “C++ Extensions for Reflection”, ISO/IEC CD TS 23619, N4856.
Until then, the compile-time reflection is emulated by the refl-cpp (header-only) library through a constexpr visitor-pattern and — for the use in opencmw —
simplified/feature-reduced ENABLE_REFLECTION_FOR(...) macro:
struct className {
int field1;
// ...
float field2;
std::string field3;
};
ENABLE_REFLECTION_FOR(className, field1, field2, field3)Beside common primitive and STL container types, it is possible to extend and to provide custom serialisation schemes for any other arbitrary type or struct/class constructs. Further, the interface
provides also an optional light-weight constexpr annotation template wrapper Annotated<type, unit, description> ... that can be used to provide some extra meta-information (e.g. unit, descriptions, etc.)
that in turn can be used to (re-)generate and document the class definition (e.g. for other programming languages or projects that do not have the primary domain-object definition at hand)
or to generate a generic OpenAPI definition. More details can be found here.
Note that building from source is only required if you want to modify opencmw-cpp itself.
If you only want to make use of opencmw-cpp in your project to implement a service, it is not required to clone/build opencmw-cpp. In that case, rather take a look at the project opencmw-cpp-example, which demonstrates how a simple, first service can be implemented.
For concrete build instructions, please check the build instructions page.
For an example on how to implement a simple, first service using opencmw-cpp, please take a look at the project opencmw-cpp-example.
The following provides some flavour of how a simple service can be implemented using OpenCMW with only a few lines of custom user-code (full sample):
// TODO: Java Concept to be ported/implemented in C++
@MetaInfo(description = "My first 'Hello World!' Service")
public static class HelloWorldWorker extends MajordomoWorker<BasicRequestCtx, NoData, ReplyData> {
public HelloWorldWorker(final ZContext ctx, final String serviceName, final RbacRole<?>... rbacRoles) {
super(ctx, serviceName, BasicRequestCtx.class, NoData.class, ReplyData.class, rbacRoles);
// the custom used code:
this.setHandler((rawCtx, requestContext, requestData, replyContext, replyData) -> {
final String name = Objects.requireNonNullElse(requestContext.name, "");
LOGGER.atInfo().addArgument(rawCtx.req.command).addArgument(rawCtx.req.topic)
.log("{} request for worker - requested topic '{}'");
replyData.returnValue = name.isBlank() ? "Hello World" : "Hello, " + name + "!";
replyContext.name = name.isBlank() ? "At" : (name + ", at") + " your service!";
});
// simple asynchronous notify example - (real-world use-cases would use another updater than Timer)
new Timer(true).scheduleAtFixedRate(new TimerTask() {
private final BasicRequestCtx notifyContext = new BasicRequestCtx(); // re-use to avoid gc
private final ReplyData notifyData = new ReplyData(); // re-use to avoid gc
private int i;
@Override
public void run() {
notifyContext.name = "update context #" + i;
notifyData.returnValue = "arbitrary data - update iteration #" + i++;
try {
HelloWorldWorker.this.notify(notifyContext, notifyData);
} catch (Exception e) {
LOGGER.atError().setCause(e).log("could not notify update");
// further handle exception if necessary
}
}
}, TimeUnit.SECONDS.toMillis(1), TimeUnit.SECONDS.toMillis(2));
}
}
@MetaInfo(description = "arbitrary request domain context object", direction = "IN")
public static class BasicRequestCtx {
@MetaInfo(description = " optional 'name' OpenAPI documentation")
public String name;
}
@MetaInfo(description = "arbitrary reply domain object", direction = "OUT")
public static class ReplyData {
@MetaInfo(description = " optional 'returnValue' OpenAPI documentation", unit = "a string")
public String returnValue;
}These services can be accessed using OpenCMW's own DataSourcePublisher client that queries or subscribes using one of the highly-optimised binary, JSON or other wire-formats and ZeroMQ- or RESTful (HTTP)-based high-level protocols, or through a simple RESTful web-interface that also provides simple 'get', 'set' and 'subscribe' functionalities while developing, for testing, or debugging:
The end-to-end transmission achieving roughly 10k messages per second for synchronous communications and about 140k messages per second for asynchronous and or publish-subscribe style data acquisition (TCP link via locahost). The domain-object abstraction and serialiser taking typically only 5% of the overall performance w.r.t. bare-metal transmissions (i.e. raw byte buffer transmission performance via ZeroMQ):
CPU:AMD Ryzen 9 5900X 12-Core Processor
description; n_exec; n_workers #0; #1; #2; #3; #4; avg
get, sync, future, domain-object ; 10000; 1; 7124.48; 10166.67; 10651.01; 10846.83; 10968.31; 10658.21
get, sync, eventStore, domain-object ; 10000; 1; 9842.93; 9789.00; 9777.39; 9270.77; 9805.15; 9660.58
get, sync, eventStore, raw-byte[] ; 10000; 1; 12237.14; 12256.34; 12259.75; 13151.36; 13171.80; 12709.81
get, async, eventStore, domain-object ; 10000; 1; 46134.05; 50850.82; 48108.52; 54487.72; 46171.67; 49904.68
get, async, eventStore, raw-byte[] ; 10000; 1; 48972.78; 53278.84; 52600.98; 54832.65; 53027.51; 53435.00
sub, async, eventStore, domain-object ; 10000; 1; 70222.57; 115074.45; 161601.24; 132852.50; 164151.85; 143420.01
sub, async, eventStore, raw-byte[] ; 10000; 1; 121308.73; 123829.95; 124283.37; 166348.23; 128094.40; 135638.99
sub, async, callback, domain-object ; 10000; 1; 111274.04; 118184.64; 123098.70; 116418.52; 107858.25; 116390.03
Your mileage may vary depending on the specific domain-object, processing logic, and choice of hardware (CPU/RAM), but you can check and compare the results for your platform using the RoundTripAndNotifyEvaluation and/or MdpImplementationBenchmark benchmarks.
.... more to follow.
For prototyping applications or services that do not interact with hardware-based systems, a Java-based OpenCMW twin-project is being developed which follows the same functional style but takes advantage of more concise implementation and C++-based type safety.
The implementation heavily relies upon and re-uses time-tried and well-established concepts from ZeroMQ (notably the Majordomo communication pattern, see Z-Guide for details), LMAX's lock-free ring-buffer disruptor (C++ port), GNU-Radio real-time signal processing framework, as well as previous implementations and experiences gained at GSI, FAIR and CERN.