Open Common Middle-Ware

... 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.

General Schematic

OpenCMW combines ZeroMQ's Majordomo with LMAX's disruptor 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:

Glossary

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 pattern, the EventStore provides datastructures and setup methods to define processing pipelines based on incoming data. (N.B. Java specific: a special worker allows reclaiming memory from expired events using a special SharedPointer implementation.)

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.

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):

@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 basic HTML rendering is based on Apache's velocity template engine and can be customised. Alternatively, the ClipboardWorker can be used in combination with other UI technologies (e.g. chartfx) to create more complex monitoring or fixed-displays with only a few hundred lines of code. For more efficient, complex and cross-platform UI designs it is planned to allow embedding of WebAssembly-based (WASM) applications.

Performance

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) with 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. While -- of course -- always subject to improvements, this initial not yet optimised performance is acceptable for our initial purposes of interfacing with Java-based (often RMI-only) systems.

Documentation

.... more to follow.

Don't like Java?

For faster applications and interaction hardware-based systems (e.g. DAQ controller/digitizer, etc.), a C++-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. Stay tuned...

Acknowledgements

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, GNU-Radio real-time signal processing framework, [Javalin] for the RESTful interface, as well as previous implementations and experiences gained at GSI, FAIR and CERN.