forked from commaai/msgq
-
Notifications
You must be signed in to change notification settings - Fork 1
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge branch 'master' into fix_munmap
- Loading branch information
Showing
74 changed files
with
509 additions
and
5,621 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -5,6 +5,7 @@ __pycache__ | |
| .*.swp | ||
| .*.swo | ||
| *.os | ||
| *.so | ||
| *.o | ||
| *.a | ||
|
|
||
|
|
||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,60 +1,54 @@ | ||
| # What is cereal? [](https://github.com/commaai/cereal/actions) [](https://codecov.io/gh/commaai/cereal) | ||
| # MSGQ: A lock free single producer multi consumer message queue | ||
|
|
||
| cereal is both a messaging spec for robotics systems as well as generic high performance IPC pub sub messaging with a single publisher and multiple subscribers. | ||
| ## What is this library? | ||
| MSGQ is a generic high performance IPC pub sub system with a single publisher and multiple subscribers. MSGQ is designed to be a high performance replacement for ZMQ-like SUB/PUB patterns. It uses a ring buffer in shared memory to efficiently read and write data. Each read requires a copy. Writing can be done without a copy, as long as the size of the data is known in advance. While MSGQ is the core of this library, this library also allows replacing the MSGQ backend with ZMQ or a spoofed implementation that can be used for deterministic testing. This library also contains visionipc, an IPC system specifically for large contiguous buffers (like images/video). | ||
|
|
||
| Imagine this use case: | ||
| * A sensor process reads gyro measurements directly from an IMU and publishes a `sensorEvents` packet | ||
| * A calibration process subscribes to the `sensorEvents` packet to use the IMU | ||
| * A localization process subscribes to the `sensorEvents` packet to use the IMU also | ||
| ## Storage | ||
| The storage for the queue consists of an area of metadata, and the actual buffer. The metadata contains: | ||
|
|
||
| 1. A counter to the number of readers that are active | ||
| 2. A pointer to the head of the queue for writing. From now on referred to as *write pointer* | ||
| 3. A cycle counter for the writer. This counter is incremented when the writer wraps around | ||
| 4. N pointers, pointing to the current read position for all the readers. From now on referred to as *read pointer* | ||
| 5. N counters, counting the number of cycles for all the readers | ||
| 6. N booleans, indicating validity for all the readers. From now on referred to as *validity flag* | ||
|
|
||
| ## Messaging Spec | ||
| The counter and the pointer are both 32 bit values, packed into 64 bit so they can be read and written atomically. | ||
|
|
||
| You'll find the message types in [log.capnp](log.capnp). It uses [Cap'n proto](https://capnproto.org/capnp-tool.html) and defines one struct called `Event`. | ||
| The data buffer is a ring buffer. All messages are prefixed by an 8 byte size field, followed by the data. A size of -1 indicates a wrap-around, and means the next message is stored at the beginning of the buffer. | ||
|
|
||
| All `Events` have a `logMonoTime` and a `valid`. Then a big union defines the packet type. | ||
|
|
||
| ### Best Practices | ||
| ## Writing | ||
| Writing involves the following steps: | ||
|
|
||
| - **All fields must describe quantities in SI units**, unless otherwise specified in the field name. | ||
| - In the context of the message they are in, field names should be completely unambiguous. | ||
| - All values should be easy to plot and be human-readable with minimal parsing. | ||
| 1. Check if the area that is to be written overlaps with any of the read pointers, mark those readers as invalid by clearing the validity flag. | ||
| 2. Write the message | ||
| 3. Increase the write pointer by the size of the message | ||
|
|
||
| ### Maintaining backwards-compatibility | ||
| In case there is not enough space at the end of the buffer, a special empty message with a prefix of -1 is written. The cycle counter is incremented by one. In this case step 1 will check there are no read pointers pointing to the remainder of the buffer. Then another write cycle will start with the actual message. | ||
|
|
||
| When making changes to the messaging spec you want to maintain backwards-compatibility, such that old logs can | ||
| be parsed with a new version of cereal. Adding structs and adding members to structs is generally safe, most other | ||
| things are not. Read more details [here](https://capnproto.org/language.html). | ||
| There always needs to be 8 bytes of empty space at the end of the buffer. By doing this there is always space to write the -1. | ||
|
|
||
| ### Custom forks | ||
| ## Reset reader | ||
| When the reader is lagging too much behind the read pointer becomes invalid and no longer points to the beginning of a valid message. To reset a reader to the current write pointer, the following steps are performed: | ||
|
|
||
| Forks of [openpilot](https://github.com/commaai/openpilot) might want to add things to the messaging | ||
| spec, however this could conflict with future changes made in mainline cereal/openpilot. Rebasing against mainline openpilot | ||
| then means breaking backwards-compatibility with all old logs of your fork. So we added reserved events in | ||
| [custom.capnp](custom.capnp) that we will leave empty in mainline cereal/openpilot. **If you only modify those, you can ensure your | ||
| fork will remain backwards-compatible with all versions of mainline cereal/openpilot and your fork.** | ||
| 1. Set valid flag | ||
| 2. Set read cycle counter to that of the writer | ||
| 3. Set read pointer to write pointer | ||
|
|
||
| ## Pub Sub Backends | ||
| ## Reading | ||
| Reading involves the following steps: | ||
|
|
||
| cereal supports two backends, one based on [zmq](https://zeromq.org/) and another called [msgq](messaging/msgq.cc), a custom pub sub based on shared memory that doesn't require the bytes to pass through the kernel. | ||
| 1. Read the size field at the current read pointer | ||
| 2. Read the validity flag | ||
| 3. Copy the data out of the buffer | ||
| 4. Increase the read pointer by the size of the message | ||
| 5. Check the validity flag again | ||
|
|
||
| Example | ||
| --- | ||
| ```python | ||
| import cereal.messaging as messaging | ||
| Before starting the copy, the valid flag is checked. This is to prevent a race condition where the size prefix was invalid, and the read could read outside of the buffer. Make sure that step 1 and 2 are not reordered by your compiler or CPU. | ||
|
|
||
| # in subscriber | ||
| sm = messaging.SubMaster(['sensorEvents']) | ||
| while 1: | ||
| sm.update() | ||
| print(sm['sensorEvents']) | ||
| If a writer overwrites the data while it's being copied out, the data will be invalid. Therefore the validity flag is also checked after reading it. The order of step 4 and 5 does not matter. | ||
|
|
||
| ``` | ||
| If at steps 2 or 5 the validity flag is not set, the reader is reset. Any data that was already read is discarded. After the reader is reset, the reading starts from the beginning. | ||
|
|
||
| ```python | ||
| # in publisher | ||
| pm = messaging.PubMaster(['sensorEvents']) | ||
| dat = messaging.new_message('sensorEvents', size=1) | ||
| dat.sensorEvents[0] = {"gyro": {"v": [0.1, -0.1, 0.1]}} | ||
| pm.send('sensorEvents', dat) | ||
| ``` | ||
| If a message with size -1 is encountered, step 3 and 4 are replaced by increasing the cycle counter and setting the read pointer to the beginning of the buffer. After that another read is performed. |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,78 +1,48 @@ | ||
| Import('env', 'envCython', 'arch', 'common') | ||
|
|
||
| import shutil | ||
|
|
||
| cereal_dir = Dir('.') | ||
| visionipc_dir = Dir('msgq/visionipc') | ||
| gen_dir = Dir('gen') | ||
| messaging_dir = Dir('messaging') | ||
|
|
||
| # Build cereal | ||
|
|
||
| schema_files = ['log.capnp', 'car.capnp', 'legacy.capnp', 'custom.capnp'] | ||
| env.Command(["gen/c/include/c++.capnp.h"], [], "mkdir -p " + gen_dir.path + "/c/include && touch $TARGETS") | ||
| env.Command([f'gen/cpp/{s}.c++' for s in schema_files] + [f'gen/cpp/{s}.h' for s in schema_files], | ||
| schema_files, | ||
| f"capnpc --src-prefix={cereal_dir.path} $SOURCES -o c++:{gen_dir.path}/cpp/") | ||
|
|
||
| # TODO: remove non shared cereal and messaging | ||
| cereal_objects = env.SharedObject([f'gen/cpp/{s}.c++' for s in schema_files]) | ||
|
|
||
| cereal = env.Library('cereal', cereal_objects) | ||
| env.SharedLibrary('cereal_shared', cereal_objects) | ||
|
|
||
| # Build messaging | ||
|
|
||
| services_h = env.Command(['services.h'], ['services.py'], 'python3 ' + cereal_dir.path + '/services.py > $TARGET') | ||
|
|
||
| messaging_objects = env.SharedObject([ | ||
| 'messaging/messaging.cc', | ||
| 'messaging/event.cc', | ||
| 'messaging/impl_zmq.cc', | ||
| 'messaging/impl_msgq.cc', | ||
| 'messaging/impl_fake.cc', | ||
| 'messaging/msgq.cc', | ||
| 'messaging/socketmaster.cc', | ||
| # Build msgq | ||
| msgq_objects = env.SharedObject([ | ||
| 'msgq/ipc.cc', | ||
| 'msgq/event.cc', | ||
| 'msgq/impl_zmq.cc', | ||
| 'msgq/impl_msgq.cc', | ||
| 'msgq/impl_fake.cc', | ||
| 'msgq/msgq.cc', | ||
| ]) | ||
|
|
||
| messaging = env.Library('messaging', messaging_objects) | ||
| Depends('messaging/impl_zmq.cc', services_h) | ||
|
|
||
| env.Program('messaging/bridge', ['messaging/bridge.cc'], LIBS=[messaging, 'zmq', common]) | ||
| Depends('messaging/bridge.cc', services_h) | ||
|
|
||
| messaging_python = envCython.Program('messaging/messaging_pyx.so', 'messaging/messaging_pyx.pyx', LIBS=envCython["LIBS"]+[messaging, "zmq", common]) | ||
| msgq = env.Library('msgq', msgq_objects) | ||
| msgq_python = envCython.Program('msgq/ipc_pyx.so', 'msgq/ipc_pyx.pyx', LIBS=envCython["LIBS"]+[msgq, "zmq", common]) | ||
|
|
||
| # Build Vision IPC | ||
| vipc_sources = [ | ||
| 'visionipc/ipc.cc', | ||
| 'visionipc/visionipc_server.cc', | ||
| 'visionipc/visionipc_client.cc', | ||
| 'visionipc/visionbuf.cc', | ||
| ] | ||
| vipc_files = ['visionipc.cc', 'visionipc_server.cc', 'visionipc_client.cc', 'visionbuf.cc'] | ||
| vipc_sources = [f'{visionipc_dir.abspath}/{f}' for f in vipc_files] | ||
|
|
||
| if arch == "larch64": | ||
| vipc_sources += ['visionipc/visionbuf_ion.cc'] | ||
| vipc_sources += [f'{visionipc_dir.abspath}/visionbuf_ion.cc'] | ||
| else: | ||
| vipc_sources += ['visionipc/visionbuf_cl.cc'] | ||
| vipc_sources += [f'{visionipc_dir.abspath}/visionbuf_cl.cc'] | ||
|
|
||
| vipc_objects = env.SharedObject(vipc_sources) | ||
| visionipc = env.Library('visionipc', vipc_objects) | ||
|
|
||
|
|
||
| vipc_frameworks = [] | ||
| vipc_libs = envCython["LIBS"] + [visionipc, messaging, common, "zmq"] | ||
| vipc_libs = envCython["LIBS"] + [visionipc, msgq, common, "zmq"] | ||
| if arch == "Darwin": | ||
| vipc_frameworks.append('OpenCL') | ||
| else: | ||
| vipc_libs.append('OpenCL') | ||
| envCython.Program('visionipc/visionipc_pyx.so', 'visionipc/visionipc_pyx.pyx', | ||
| envCython.Program(f'{visionipc_dir.abspath}/visionipc_pyx.so', f'{visionipc_dir.abspath}/visionipc_pyx.pyx', | ||
| LIBS=vipc_libs, FRAMEWORKS=vipc_frameworks) | ||
|
|
||
| if GetOption('extras'): | ||
| env.Program('messaging/test_runner', ['messaging/test_runner.cc', 'messaging/msgq_tests.cc'], LIBS=[messaging, common]) | ||
|
|
||
| env.Program('visionipc/test_runner', ['visionipc/test_runner.cc', 'visionipc/visionipc_tests.cc'], | ||
| env.Program('msgq/test_runner', ['msgq/test_runner.cc', 'msgq/msgq_tests.cc'], LIBS=[msgq, common]) | ||
| env.Program(f'{visionipc_dir.abspath}/test_runner', | ||
| [f'{visionipc_dir.abspath}/test_runner.cc', f'{visionipc_dir.abspath}/visionipc_tests.cc'], | ||
| LIBS=['pthread'] + vipc_libs, FRAMEWORKS=vipc_frameworks) | ||
|
|
||
|
|
||
| Export('cereal', 'messaging', 'messaging_python', 'visionipc') | ||
| Export('visionipc', 'msgq', 'msgq_python') |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file was deleted.
Oops, something went wrong.
Oops, something went wrong.