mirror of
https://github.com/FairRootGroup/FairMQ.git
synced 2025-10-16 10:01:47 +00:00
Update README
This commit is contained in:
Dennis Klein
@@ -1,5 +0,0 @@
|
||||
Al-Turany, Mohammad
|
||||
Klein, Dennis
|
||||
Kollegger, Thorsten
|
||||
Rybalchenko, Alexey
|
||||
Winckler, Nicolas
|
||||
@@ -1,8 +0,0 @@
|
||||
Binet, Sebastien
|
||||
Eulisse, Giulio
|
||||
Karabowicz, Radoslaw
|
||||
Krzewicki, Mikolaj
|
||||
Neskovic, Gvozden
|
||||
Richter, Matthias
|
||||
Uhlig, Florian
|
||||
Wenzel, Sandro
|
||||
@@ -1,29 +0,0 @@
|
||||
# FairMQ
|
||||
|
||||
Standard FairRoot is running all the different analysis tasks within one process. FairMQ ([Message Queue](http://en.wikipedia.org/wiki/Message_queue)) allows starting tasks on different processes and provides the communication layer between these processes.
|
||||
|
||||
1. [Device](docs/Device.md#1-device)
|
||||
1. [Topology](docs/Device.md#11-topology)
|
||||
2. [Communication Patterns](docs/Device.md#12-communication-patterns)
|
||||
3. [State Machine](docs/Device.md#13-state-machine)
|
||||
4. [Multiple devices in the same process](docs/Device.md#15-multiple-devices-in-the-same-process)
|
||||
2. [Transport Interface](docs/Transport.md#2-transport-interface)
|
||||
1. [Message](docs/Transport.md#21-message)
|
||||
1. [Ownership](docs/Transport.md#211-ownership)
|
||||
2. [Channel](docs/Transport.md#22-channel)
|
||||
3. [Poller](docs/Transport.md#23-poller)
|
||||
3. [Configuration](docs/Configuration.md#3-configuration)
|
||||
1. [Device Configuration](docs/Configuration.md#31-device-configuration)
|
||||
2. [Communication Channels Configuration](docs/Configuration.md#32-communication-channels-configuration)
|
||||
1. [JSON Parser](docs/Configuration.md#321-json-parser)
|
||||
2. [SuboptParser](docs/Configuration.md#322-suboptparser)
|
||||
3. [Introspection](docs/Configuration.md#33-introspection)
|
||||
4. [Development](docs/Development.md#4-development)
|
||||
1. [Testing](docs/Development.md#41-testing)
|
||||
5. [Logging](docs/Logging.md#5-logging)
|
||||
1. [Log severity](docs/Logging.md#51-log-severity)
|
||||
2. [Log verbosity](docs/Logging.md#52-log-verbosity)
|
||||
3. [Color for console output](docs/Logging.md#53-color)
|
||||
4. [File output](docs/Logging.md#54-file-output)
|
||||
5. [Custom sinks](docs/Logging.md#55-custom-sinks)
|
||||
6. [Examples](docs/Examples.md#6-examples)
|
||||
@@ -1,113 +0,0 @@
|
||||
← [Back](../README.md)
|
||||
|
||||
# 3. Configuration
|
||||
|
||||
## 3.1 Device Configuration
|
||||
|
||||
Devices receive configuration primarily via provided command line options (that can be extended per device).
|
||||
|
||||
## 3.2 Communication Channels Configuration
|
||||
|
||||
The communication channels can be configured via configuration parsers. The parser system is extendable, so if provided parsers do not suit your style, you can write your own and plug them in the configuration system.
|
||||
|
||||
The provided parsers are:
|
||||
|
||||
### 3.2.1 JSON Parser
|
||||
|
||||
This parser reads channel configuration from a JSON file. Example:
|
||||
|
||||
```JSON
|
||||
{
|
||||
"fairMQOptions": {
|
||||
"devices": [
|
||||
{
|
||||
"id": "sampler1",
|
||||
"channels": [
|
||||
{
|
||||
"name": "data",
|
||||
"sockets": [
|
||||
{
|
||||
"type": "push",
|
||||
"method": "bind",
|
||||
"address": "tcp://*:5555",
|
||||
"sndBufSize": 1000,
|
||||
"rcvBufSize": 1000,
|
||||
"rateLogging": 1
|
||||
}
|
||||
]
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"id": "sink1",
|
||||
"channels": [
|
||||
{
|
||||
"name": "data",
|
||||
"sockets": [
|
||||
{
|
||||
"type": "pull",
|
||||
"method": "connect",
|
||||
"address": "tcp://localhost:5555",
|
||||
"sndBufSize": 1000,
|
||||
"rcvBufSize": 1000,
|
||||
"rateLogging": 1
|
||||
}
|
||||
]
|
||||
}
|
||||
]
|
||||
}
|
||||
]
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
The JSON file can contain configuration for multiple devices.
|
||||
|
||||
- The mapping between device and configuration happens via the device ID (the `--id` parameter of the launched device and the `"id"` entry in the config).
|
||||
- Instead of `"id"`, JSON file may contain device configurations under `"key"`, which allows launched devices to share configuration, e.g.: `my-device-executable --id <device-id> --config-key <config-key>`.
|
||||
- Socket options must contain at least *type*, *method* and *address*, the rest of the values are optional and will get default values of the channel.
|
||||
- If a channel has multiple sub-channels, common properties can be defined under channel directly, and will be shared by all sub-channels, e.g.:
|
||||
|
||||
```JSON
|
||||
"channels": [{
|
||||
"name": "data",
|
||||
"type": "push",
|
||||
"method": "bind",
|
||||
"sockets": [{
|
||||
"address": "tcp://*:5555",
|
||||
"address": "tcp://*:5556",
|
||||
"address": "tcp://*:5557"
|
||||
}]
|
||||
}]
|
||||
```
|
||||
|
||||
### 3.2.2 SuboptParser
|
||||
|
||||
This parser configures channels directly from the command line.
|
||||
The parser handles a comma separated key=value list format by using the getsubopt function of the standard library.
|
||||
The option key `--channel-config` can be used with the list of key/value pairs, e.g.:
|
||||
|
||||
```
|
||||
--channel-config name=output,type=push,method=bind,address=tcp://127.0.0.1:5555
|
||||
```
|
||||
|
||||
## 3.3 Introspection
|
||||
|
||||
A compiled device executable repots its available configuration. Run the device with one of the following options to see the corresponding help:
|
||||
|
||||
- `-h [ --help ]`: All available command line options with their descriptions and default values.
|
||||
|
||||
- `--print-options`: All available command line options in a machine-readable format: `<option>:<computed-value>:<<type>>:<description>`.
|
||||
|
||||
- `--print-channels`: Prints registered channels in a machine-readable format: `<channel name>:<minimum sub-channels>:<maximum sub-channels>`. There are devices where channels names are not known in advance before the configuration takes place (e.g. FairMQMultiplier has configurable channel names at runtime). This options will only print channels that have been registered in the device by implementing the following method:
|
||||
|
||||
```C++
|
||||
void YourDevice::RegisterChannelEndpoints()
|
||||
{
|
||||
// provide channel name, minimum and maximum number of subchannels
|
||||
RegisterChannelEndpoint("channelA", 1, 10000);
|
||||
RegisterChannelEndpoint("channelB", 1, 1);
|
||||
}
|
||||
```
|
||||
|
||||
← [Back](../README.md)
|
||||
@@ -1,13 +0,0 @@
|
||||
← [Back](../README.md)
|
||||
|
||||
# 4. Development
|
||||
|
||||
## 4.1 Testing
|
||||
|
||||
For unit testing it is often not feasible to boot up a full-blown distributed system with dozens of processes.
|
||||
|
||||
In some scenarios it is useful to not even instantiate a `FairMQDevice` at all. Please see [this example](../test/protocols/_push_pull_multipart.cxx) for single and multi threaded unit test without a device instance. If you store your transport factories and channels on the heap, pls make sure, you destroy the channels before you destroy the related transport factory for proper shutdown. Channels provide all the `Send/Receive` and `New*Message/New*Poller` APIs provided by the device too.
|
||||
|
||||
TODO Multiple devices in one process.
|
||||
|
||||
← [Back](../README.md)
|
||||
@@ -1,52 +0,0 @@
|
||||
← [Back](../README.md)
|
||||
|
||||
# 1. Device
|
||||
|
||||
The components encapsulating the tasks are called **devices** and derive from the common base class `FairMQDevice`. FairMQ provides ready to use devices to organize the dataflow between the components (without touching the contents of a message), providing functionality like merging and splitting of the data stream (see subdirectory `devices`).
|
||||
|
||||
## 1.1 Topology
|
||||
|
||||
Devices are arranged into **topologies** where each device has a defined number of data inputs and outputs.
|
||||
|
||||
Example of a simple FairMQ topology:
|
||||
|
||||
|
||||
|
||||
Within a topology each device needs a unique id (given to it via required command line option `--id`).
|
||||
|
||||
Topology configuration is currently happening via setup scripts. This is very rudimentary and a much more flexible system is now in development. For now, example setup scripts can be found in directory `FairRoot/example/Tutorial3/` along with some additional documentation.
|
||||
|
||||
## 1.2 Communication Patterns
|
||||
|
||||
FairMQ devices communicate via the communication patterns offered by ZeroMQ (or nanomsg): PUSH-PULL, PUB-SUB, REQ-REP, PAIR, [more info here](http://api.zeromq.org/4-0:zmq-socket). Each transport may provide further patterns.
|
||||
|
||||
## 1.3 State Machine
|
||||
|
||||
Each FairMQ device has an internal state machine:
|
||||
|
||||
|
||||
|
||||
The state machine can be querried and controlled via `GetCurrentStateName()` and `ChangeState("<state name>")` methods. Only legal state transitions are allowed (see image above). Illegal transitions will fail with an error.
|
||||
|
||||
If the device is running in interactive mode (default), states can be changed via keyboard input:
|
||||
|
||||
- `'h'` - help
|
||||
- `'p'` - pause
|
||||
- `'r'` - run
|
||||
- `'s'` - stop
|
||||
- `'t'` - reset task
|
||||
- `'d'` - reset device
|
||||
- `'q'` - end
|
||||
- `'j'` - init task
|
||||
- `'i'` - init device
|
||||
|
||||
Without the interactive mode, for example for a run in background, two other control mechanisms are available:
|
||||
|
||||
- static (`--control static`) - device goes through a simple init -> run -> reset -> exit chain.
|
||||
- dds (`--control dds`) - device is controled by external command, in this case using dds commands (fairmq-dds-command-ui).
|
||||
|
||||
## 1.4 Multiple devices in the same process
|
||||
|
||||
Technically one can create two or more devices within the same process without any conflicts. However the configuration (FairMQProgOptions) currently assumes the supplied configuration values are for one device/process.
|
||||
|
||||
← [Back](../README.md)
|
||||
@@ -1,9 +0,0 @@
|
||||
← [Back](../README.md)
|
||||
|
||||
# 6. Examples
|
||||
|
||||
A collection of simple examples in `FairRoot/examples/MQ` directory demonstrates some common usage patterns of FairMQ.
|
||||
|
||||
A number of devices to handle the data from the Tutorial3 FairTestDetector of FairRoot are provided as an example and can be found in `FairRoot/base/MQ` directory. The implementation of the tasks run by these devices can be found `FairRoot/examples/advanced/Tutorial3`. The implementation includes sending raw binary data as well as serializing the data with either [Boost Serialization](http://www.boost.org/doc/libs/release/libs/serialization/), [Google Protocol Buffers](https://developers.google.com/protocol-buffers/) or [Root TMessage](http://root.cern.ch/root/html/TMessage.html). Following the examples you can implement your own devices to transport arbitrary data.
|
||||
|
||||
← [Back](../README.md)
|
||||
@@ -1,103 +0,0 @@
|
||||
← [Back](../README.md)
|
||||
|
||||
# 5. Logging
|
||||
|
||||
The FairMQLogger header uses fair::Logger library for logging.
|
||||
|
||||
All log calls go through the provided LOG(severity) macro. Output through this macro is thread-safe. Logging is done to cout, file output and/or custom sinks.
|
||||
|
||||
## 5.1 Log severity
|
||||
|
||||
The log severity is controlled via:
|
||||
```C++
|
||||
fair::Logger::SetConsoleSeverity("<severity level>");
|
||||
// and/or
|
||||
fair::Logger::SetFileSeverity("<severity level>");
|
||||
// and/or
|
||||
fair::Logger::SetCustomSeverity("<customSinkName>", "<severity level>");
|
||||
```
|
||||
|
||||
where severity level is one of the following:
|
||||
|
||||
```C++
|
||||
"nolog",
|
||||
"fatal",
|
||||
"error",
|
||||
"warn",
|
||||
"state",
|
||||
"info",
|
||||
"debug",
|
||||
"debug1",
|
||||
"debug2",
|
||||
"debug3",
|
||||
"debug4",
|
||||
"trace",
|
||||
```
|
||||
|
||||
Logger will log the chosen severity and all above it (except "nolog", which deactivates logging for that sink completely). Fatal severity is always logged.
|
||||
|
||||
When running a FairMQ device, the log severity can be simply provided via `--severity <level>` cmd option.
|
||||
|
||||
## 5.2 Log verbosity
|
||||
|
||||
The log verbosity is controlled via:
|
||||
```C++
|
||||
fair::Logget::SetVerbosity("<verbosity level>");
|
||||
```
|
||||
|
||||
it is same for all sinks, and is one of the following values: `low`, `medium`, `high`, `veryhigh`, which translates to following output:
|
||||
|
||||
```
|
||||
low: [severity] message
|
||||
medium: [HH:MM:SS][severity] message
|
||||
high: [process name][HH:MM:SS:µS][severity] message
|
||||
veryhigh: [process name][HH:MM:SS:µS][severity][file:line:function] message
|
||||
```
|
||||
|
||||
When running a FairMQ device, the log severity can be simply provided via `--verbosity <level>` cmd option.
|
||||
|
||||
## 5.3 Color
|
||||
|
||||
Colored output on console can be activated with:
|
||||
```C++
|
||||
Logger::SetConsoleColor(true);
|
||||
```
|
||||
|
||||
When running a FairMQ device, the log color (console) can be simply provided via `--color <true/false>` cmd option (default is true).
|
||||
|
||||
## 5.4 File output
|
||||
|
||||
Output to file can be enabled via:
|
||||
```C++
|
||||
Logger::InitFileSink("<severity level>", "test_log", true);
|
||||
```
|
||||
which will add output to "test_log" filename (if third parameter is `true` it will add timestamp to the file name) with `<severity level>` severity.
|
||||
|
||||
When running a FairMQ device, the log file can be simply provided via `--log-to-file <filename_prefix>` cmd option (this will also turn off console output).
|
||||
|
||||
## 5.5 Custom sinks
|
||||
|
||||
Custom sinks can be added via `Logger::AddCustomSink("sink name", "<severity>", callback)` method.
|
||||
|
||||
Here is an example adding a custom sink for all severities ("trace" and above). It has access to the log content and meta data. Custom log calls are also thread-safe.
|
||||
|
||||
```C++
|
||||
Logger::AddCustomSink("MyCustomSink", "trace", [](const string& content, const LogMetaData& metadata)
|
||||
{
|
||||
cout << "content: " << content << endl;
|
||||
|
||||
cout << "available metadata: " << endl;
|
||||
cout << "std::time_t timestamp: " << metadata.timestamp << endl;
|
||||
cout << "std::chrono::microseconds us: " << metadata.us.count() << endl;
|
||||
cout << "std::string process_name: " << metadata.process_name << endl;
|
||||
cout << "std::string file: " << metadata.file << endl;
|
||||
cout << "std::string line: " << metadata.line << endl;
|
||||
cout << "std::string func: " << metadata.func << endl;
|
||||
cout << "std::string severity_name: " << metadata.severity_name << endl;
|
||||
cout << "fair::Severity severity: " << static_cast<int>(metadata.severity) << endl;
|
||||
});
|
||||
```
|
||||
|
||||
If only output from custom sinks is desirable, console/file sinks must be deactivated by setting their severity to `"nolog"`.
|
||||
|
||||
← [Back](../README.md)
|
||||
@@ -1,94 +0,0 @@
|
||||
← [Back](../README.md)
|
||||
|
||||
# 2. Transport Interface
|
||||
|
||||
The communication layer is available through the transport interface. Three interface implementations are currently available. Main implementation uses the [ZeroMQ](http://zeromq.org) library. Alternative implementation relies on the [nanomsg](http://nanomsg.org) library. Third transport implementation is using shared memory via boost::interprocess & ZeroMQ combination.
|
||||
|
||||
Here is an overview to give an idea how the interface is implemented:
|
||||
|
||||
|
||||
|
||||
Currently, the transports have been tested to work with these communication patterns:
|
||||
|
||||
| | zeromq | nanomsg | shmem |
|
||||
| ------------- |--------| ------- | ----- |
|
||||
| PAIR | yes | yes | yes |
|
||||
| PUSH/PULL | yes | yes | yes |
|
||||
| PUB/SUB | yes | yes | no |
|
||||
| REQ/REP | yes | yes | yes |
|
||||
|
||||
The next table shows the supported address types for each transport implementation:
|
||||
|
||||
| | zeromq | nanomsg | shmem | comment |
|
||||
| ----------- | ------ | ------- | ----- | --------------------------------------------- |
|
||||
| `inproc://` | yes | yes | yes | in process: useful for unit testing |
|
||||
| `ipc://` | yes | yes | yes | inter process comm: useful on single machine |
|
||||
| `tcp://` | yes | yes | yes | useful for any communication, local or remote |
|
||||
|
||||
## 2.1 Message
|
||||
|
||||
Devices transport data between each other in form of `FairMQMessage`s. These can be filled with arbitrary content. Message can be initialized in three different ways by calling `NewMessage()`:
|
||||
|
||||
```cpp
|
||||
FairMQMessagePtr NewMessage() const;
|
||||
```
|
||||
**with no parameters**: Initializes an empty message (typically used for receiving).
|
||||
|
||||
```cpp
|
||||
FairMQMessagePtr NewMessage(const size_t size) const;
|
||||
```
|
||||
**given message size**: Initializes message body with a given size. Fill the created contents via buffer pointer.
|
||||
|
||||
```cpp
|
||||
using fairmq_free_fn = void(void* data, void* hint);
|
||||
FairMQMessagePtr NewMessage(void* data, const size_t size, fairmq_free_fn* ffn, void* hint = nullptr) const;
|
||||
```
|
||||
**given existing buffer and a size**: Initialize the message from an existing buffer. In case of ZeroMQ this is a zero-copy operation.
|
||||
|
||||
Additionally, FairMQ provides two more message factories for convenience:
|
||||
```cpp
|
||||
template<typename T>
|
||||
FairMQMessagePtr NewSimpleMessage(const T& data) const
|
||||
```
|
||||
**copy and own**: Copy the `data` argument into the returned message and take ownership (free memory after message is sent). This interface is useful for small, [trivially copyable](http://en.cppreference.com/w/cpp/concept/TriviallyCopyable) data.
|
||||
|
||||
```cpp
|
||||
template<typename T>
|
||||
FairMQMessagePtr NewStaticMessage(const T& data) const
|
||||
```
|
||||
**point to existing memory**: The returned message will point to the `data` argument, but not take ownership (someone else must destruct this variable). Make sure that `data` lives long enough to be successfully sent. This interface is most useful for third party managed, contiguous memory (Be aware of shallow types with internal pointer references! These will not be sent.)
|
||||
|
||||
## 2.1.1 Ownership
|
||||
|
||||
The component of a program, that is reponsible for the allocation or destruction of data in memory, is taking ownership over this data. Ownership may be passed along to another component. It is also possible that multiple components share ownership of data. In this case, some strategy must be in place to determine the last user of the data and assign her the responsibility of destruction.
|
||||
|
||||
After queuing a message for sending in FairMQ, the transport takes ownership over the message body and will free it with `free()` after it is no longer used. A callback can be passed to the message object, to be called instead of the destruction with `free()` (for initialization via buffer+size).
|
||||
|
||||
```cpp
|
||||
static void FairMQNoCleanup(void* /*data*/, void* /*obj*/) {}
|
||||
|
||||
template<typename T>
|
||||
static void FairMQSimpleMsgCleanup(void* /*data*/, void* obj) { delete static_cast<T*>(obj); }
|
||||
```
|
||||
For convenience, two common deleter callbacks are already defined in the `FairMQTransportFactory` class to aid the user in controlling ownership of the data.
|
||||
|
||||
## 2.2 Channel
|
||||
|
||||
A channel represents a communication endpoint in FairMQ. Usage is similar to a traditional Unix network socket. A device usually contains a number of channels that can either listen for incoming connections from channels of other devices or they can connect to other listening channels. Channels are organized by a channel name and a subchannel index.
|
||||
|
||||
```cpp
|
||||
const FairMQChannel& GetChannel(const std::string& channelName, const int index = 0) const;
|
||||
```
|
||||
|
||||
All subchannels with a common channel name need to be of the same transport type.
|
||||
|
||||
## 2.3 Poller
|
||||
|
||||
A poller allows to wait on multiple channels either to receive or send a message.
|
||||
|
||||
```cpp
|
||||
FairMQPollerPtr NewPoller(const std::vector<const FairMQChannel*>& channels)
|
||||
```
|
||||
**list channels**: This poller waits on all supplied channels. Currently, it is limited to channels of the same transport type only.
|
||||
|
||||
← [Back](../README.md)
|
||||
Binary file not shown.
|
Before Width: | Height: | Size: 104 KiB |
Binary file not shown.
|
Before Width: | Height: | Size: 11 KiB |
Binary file not shown.
|
Before Width: | Height: | Size: 27 KiB |
Binary file not shown.
|
Before Width: | Height: | Size: 30 KiB |
Reference in New Issue
Block a user