impl robust mutex & condition (using alephzero's mtx implementation)

Author: mutouyun

src/CMakeLists.txt

@@ -1,7 +1,8 @@
 project(ipc)
 
 if(UNIX)
-  file(GLOB SRC_FILES ${LIBIPC_PROJECT_DIR}/src/libipc/platform/linux/*.cpp)
+  file(GLOB SRC_FILES ${LIBIPC_PROJECT_DIR}/src/libipc/platform/linux/*.cpp
+                      ${LIBIPC_PROJECT_DIR}/src/libipc/platform/linux/a0/*.c)
 else()
   file(GLOB SRC_FILES ${LIBIPC_PROJECT_DIR}/src/libipc/platform/win/*.cpp)
 endif()
@@ -33,18 +34,18 @@ set_target_properties(${PROJECT_NAME}
 	PROPERTIES
 	ARCHIVE_OUTPUT_DIRECTORY "${CMAKE_BINARY_DIR}/lib"
 	LIBRARY_OUTPUT_DIRECTORY "${CMAKE_BINARY_DIR}/lib"
-	RUNTIME_OUTPUT_DIRECTORY "${CMAKE_BINARY_DIR}/bin" )
+	RUNTIME_OUTPUT_DIRECTORY "${CMAKE_BINARY_DIR}/bin")
 
 # set version
 set_target_properties(${PROJECT_NAME}
     PROPERTIES
-    VERSION 1.1.0
-    SOVERSION 2)
+    VERSION 1.2.0
+    SOVERSION 3)
 
 target_include_directories(${PROJECT_NAME}
   PUBLIC  ${LIBIPC_PROJECT_DIR}/include
   PRIVATE ${LIBIPC_PROJECT_DIR}/src
-)
+          $<$<BOOL:UNIX>:${LIBIPC_PROJECT_DIR}/src/libipc/platform/linux>)
 
 if(NOT MSVC)
   target_link_libraries(${PROJECT_NAME} PUBLIC
@@ -56,5 +57,4 @@ install(
   TARGETS ${PROJECT_NAME}
   RUNTIME DESTINATION bin
   LIBRARY DESTINATION lib
-  ARCHIVE DESTINATION lib
-)
+  ARCHIVE DESTINATION lib)

src/libipc/platform/linux/a0/LICENSE

@@ -0,0 +1,24 @@
+This is free and unencumbered software released into the public domain.
+
+Anyone is free to copy, modify, publish, use, compile, sell, or
+distribute this software, either in source code form or as a compiled
+binary, for any purpose, commercial or non-commercial, and by any
+means.
+
+In jurisdictions that recognize copyright laws, the author or authors
+of this software dedicate any and all copyright interest in the
+software to the public domain. We make this dedication for the benefit
+of the public at large and to the detriment of our heirs and
+successors. We intend this dedication to be an overt act of
+relinquishment in perpetuity of all present and future rights to this
+software under copyright law.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
+EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
+IN NO EVENT SHALL THE AUTHORS BE LIABLE FOR ANY CLAIM, DAMAGES OR
+OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE,
+ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
+OTHER DEALINGS IN THE SOFTWARE.
+
+For more information, please refer to <http://unlicense.org>

src/libipc/platform/linux/a0/README.md

@@ -0,0 +1,213 @@
+<h1 align="center">
+  <br>
+  <img src="https://raw.githubusercontent.com/alephzero/logo/master/rendered/alephzero.svg" width="256px">
+  <br>
+  AlephZero
+</h1>
+
+<h3 align="center">Simple, Robust, Fast IPC.</h3>
+
+<p align="center">
+  <a href="https://github.com/alephzero/alephzero/actions?query=workflow%3ACI"><img src="https://github.com/alephzero/alephzero/workflows/CI/badge.svg"></a>
+  <a href="https://codecov.io/gh/alephzero/alephzero"><img src="https://codecov.io/gh/alephzero/alephzero/branch/master/graph/badge.svg"></a>
+  <a href="https://alephzero.readthedocs.io/en/latest/?badge=latest"><img src="https://readthedocs.org/projects/alephzero/badge/?version=latest"></a>
+  <a href="http://unlicense.org"><img src="https://img.shields.io/badge/license-Unlicense-blue.svg"></a>
+</p>
+
+<p align="center">
+  <a href="#overview">Overview</a> •
+  <a href="#transport">Transport</a> •
+  <a href="#protocol">Protocol</a> •
+  <a href="#examples">Examples</a> •
+  <a href="#installation">Installation</a> •
+  <a href="#across-dockers">Across Dockers</a>
+</p>
+
+# Overview
+
+[Presentation from March 25, 2020](https://docs.google.com/presentation/d/12KE9UucjZPtpVnM1NljxOqBolBBKECWJdrCoE2yJaBw/edit#slide=id.p)
+
+AlephZero is a library for message based communication between programs running on the same machine.
+
+## Simple
+
+AlephZero's main goal is to be simple to use. Nothing is higher priority.
+
+There is no "master" process in between your nodes that is needed to do handshakes or exchanges of any kind. All you need is the topic name.
+
+See the <a href="#examples">Examples</a>.
+
+## Robust
+
+This is probably the main value of AlephZero, above similar libraries.
+
+AlephZero uses a lot of tricks to ensure the state of all channels is consistent, even when programs die. This includes double-buffering the state of the communication channel and [robustifying](https://man7.org/linux/man-pages/man3/pthread_mutexattr_setrobust.3.html) the locks and notification channels.
+
+## Fast
+
+AlephZero uses shared memory across multiple processes to read and write messages, minimizing the involvement of the kernel. The kernel only really gets involved in notifying a process that a new message exists, and for that we use futex (fast user-space mutex).
+
+TODO: Benchmarks
+
+# Transport
+
+AlephZero, at its core, is a simple allocator on top of a contiguous region of memory. Usually, shared-memory. The allocator of choice is a circular-linked-list, which is fast, simple, and sufficient for the protocol listed below. It also plays well with the robustness requirement.
+
+This has a number of implications. For one, this means that old messages are kept around until the space is needed. The oldest messages are always discarded before any more recent messages.
+
+# Protocol
+
+Rather than exposing the low-level transport directly, AlephZero provides a few higher level protocol:
+
+* <b>PubSub</b>: Broadcast published messages. Subscribers get notified.
+* <b>RPC</b>: Request-response.
+* <b>PRPC (Progressive RPC)</b>: Request-streaming response.
+* <b>Sessions</b>: Bi-directional channel of communication. Not yet implemented. Let me know if you want this.
+
+# Examples
+
+Many more example and an interactive experience can be found at: https://github.com/alephzero/playground
+
+For the curious, here are some simple snippets to get you started:
+
+To begin with, we need to include AlephZero:
+```cc
+#include <a0.h>
+```
+
+## PubSub
+
+You can have as many publisher and subscribers on the same topic as you wish. They just need to agree on the filename.
+
+```cc
+a0::Publisher p("my_pubsub_topic");
+p.pub("foo");
+```
+
+You just published `"foo"` to the `"my_pubsub_topic"`.
+
+To read those message, you can create a subscriber on the same topic:
+```cc
+a0::Subscriber sub(
+    "my_pubsub_topic",
+    A0_INIT_AWAIT_NEW,  // or MOST_RECENT or OLDEST
+    A0_ITER_NEWEST,     // or NEXT
+    [](a0::PacketView pkt_view) {
+      std::cout << "Got: " << pkt_view.payload() << std::endl;
+    });
+```
+The callback will trigger whenever a message is published.
+
+The `Subscriber` object spawns a thread that will read the topic and call the callback.
+
+The `A0_INIT` tells the subscriber where to start reading. 
+* `A0_INIT_AWAIT_NEW`: Start with messages published after the creation of the subscriber.
+* `A0_INIT_MOST_RECENT`: Start with the most recently published message. Useful for state and configuration. But be careful, this can be quite old!
+* `A0_INIT_OLDEST`: Topics keep a history of 16MB (unless configures otherwise). Start with the oldest thing still in there.
+
+The `A0_ITER` tells the subscriber how to continue reading messages. After each callback:
+* `A0_ITER_NEXT`: grab the sequentially next message. When you don't want to miss a thing.
+* `A0_ITER_NEWEST`: grab the newest available unread message. When you want to keep up with the firehose.
+
+```cc
+a0::SubscriberSync sub_sync(
+    "my_pubsub_topic",
+    A0_INIT_OLDEST, A0_ITER_NEXT);
+while (sub_sync.has_next()) {
+  auto pkt = sub_sync.next();
+  std::cout << "Got: " << pkt.payload() << std::endl;
+}
+```
+
+## RPC
+
+Create an `RpcServer`:
+
+```cc
+a0::RpcServer server(
+    "my_rpc_topic",
+    /* onrequest = */ [](a0::RpcRequest req) {
+        std::cout << "Got: " << req.pkt().payload() << std::endl;
+        req.reply("echo " + std::string(req.pkt().payload()));
+    },
+    /* oncancel = */ nullptr);
+```
+
+Create an `RpcClient`:
+
+```cc
+a0::RpcClient client("my_rpc_topic");
+client.send("client msg", [](a0::PacketView reply) {
+  std::cout << "Got: " << reply.payload() << std::endl;
+});
+```
+
+# Installation
+
+## Install From Source
+
+### Ubuntu Dependencies
+
+```sh
+apt install g++ make
+```
+
+### Alpine Dependencies
+
+```sh
+apk add g++ linux-headers make
+```
+
+### Download And Install
+
+```sh
+git clone https://github.com/alephzero/alephzero.git
+cd alephzero
+make install -j
+```
+
+## Install From Package
+
+Coming soon-ish. Let me know if you want this and I'll prioritize it. External support is much appreciated.
+
+## Integration
+
+### Command Line
+
+Add the following to g++ / clang commands.
+```sh
+-L${libdir} -lalephzero -lpthread
+```
+
+### Package-cfg
+
+```sh
+pkg-config --cflags --libs alephzero
+```
+
+### CMake
+
+Coming soon-ish. Let me know if you want this and I'll prioritize it. External support is much appreciated.
+
+### Bazel
+
+Coming soon-ish. Let me know if you want this and I'll prioritize it.
+
+# Across Dockers
+
+For programs running across different dockers to be able to communicate, we need to have them match up on two flags: `--ipc` and `--pid`.
+
+* `--ipc` shares the `/dev/shm` filesystem. This is necessary to open the same file topics.
+* `--pid` shares the process id namespace. This is necessary for the locking and notification systems.
+
+In the simplest case, you can set them both to `host` and talk through the system's global `/dev/shm` and process id namespace.
+```sh
+docker run --ipc=host --pid=host --name=foo foo_image
+docker run --ipc=host --pid=host --name=bar bar_image
+```
+
+Or, you can mark one as `shareable` and have the others connect to it:
+```sh
+docker run --ipc=shareable --pid=shareable --name=foo foo_image
+docker run --ipc=container:foo --pid=container:foo --name=bar bar_image
+```

src/libipc/platform/linux/a0/atomic.h

@@ -0,0 +1,36 @@
+#ifndef A0_SRC_ATOMIC_H
+#define A0_SRC_ATOMIC_H
+
+#include "a0/inline.h"
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+A0_STATIC_INLINE
+void a0_barrier() {
+  // 'atomic_thread_fence' is not supported with ‘-fsanitize=thread’
+  __sync_synchronize();
+}
+
+#define a0_atomic_fetch_add(P, V) __atomic_fetch_add((P), (V), __ATOMIC_RELAXED)
+#define a0_atomic_add_fetch(P, V) __atomic_add_fetch((P), (V), __ATOMIC_RELAXED)
+
+#define a0_atomic_fetch_and(P, V) __atomic_fetch_and((P), (V), __ATOMIC_RELAXED)
+#define a0_atomic_and_fetch(P, V) __atomic_and_fetch((P), (V), __ATOMIC_RELAXED)
+
+#define a0_atomic_fetch_or(P, V) __atomic_fetch_or((P), (V), __ATOMIC_RELAXED)
+#define a0_atomic_or_fetch(P, V) __atomic_or_fetch((P), (V), __ATOMIC_RELAXED)
+
+#define a0_atomic_load(P) __atomic_load_n((P), __ATOMIC_RELAXED)
+#define a0_atomic_store(P, V) __atomic_store_n((P), (V), __ATOMIC_RELAXED)
+
+// TODO(lshamis): Switch from __sync to __atomic.
+#define a0_cas_val(P, OV, NV) __sync_val_compare_and_swap((P), (OV), (NV))
+#define a0_cas(P, OV, NV) __sync_bool_compare_and_swap((P), (OV), (NV))
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif  // A0_SRC_ATOMIC_H

src/libipc/platform/linux/a0/clock.h

@@ -0,0 +1,60 @@
+#ifndef A0_SRC_CLOCK_H
+#define A0_SRC_CLOCK_H
+
+#include "a0/err.h"
+#include "a0/inline.h"
+
+#include <stdint.h>
+#include <time.h>
+
+#include "err_macro.h"
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+static const int64_t NS_PER_SEC = 1e9;
+
+typedef struct timespec timespec_t;
+
+A0_STATIC_INLINE
+a0_err_t a0_clock_now(clockid_t clk, timespec_t* out) {
+  A0_RETURN_SYSERR_ON_MINUS_ONE(clock_gettime(clk, out));
+  return A0_OK;
+}
+
+A0_STATIC_INLINE
+a0_err_t a0_clock_add(timespec_t ts, int64_t add_nsec, timespec_t* out) {
+  out->tv_sec = ts.tv_sec + add_nsec / NS_PER_SEC;
+  out->tv_nsec = ts.tv_nsec + add_nsec % NS_PER_SEC;
+  if (out->tv_nsec >= NS_PER_SEC) {
+    out->tv_sec++;
+    out->tv_nsec -= NS_PER_SEC;
+  } else if (out->tv_nsec < 0) {
+    out->tv_sec--;
+    out->tv_nsec += NS_PER_SEC;
+  }
+
+  return A0_OK;
+}
+
+A0_STATIC_INLINE
+a0_err_t a0_clock_convert(
+    clockid_t orig_clk,
+    timespec_t orig_ts,
+    clockid_t target_clk,
+    timespec_t* target_ts) {
+  timespec_t orig_now;
+  A0_RETURN_ERR_ON_ERR(a0_clock_now(orig_clk, &orig_now));
+  timespec_t target_now;
+  A0_RETURN_ERR_ON_ERR(a0_clock_now(target_clk, &target_now));
+
+  int64_t add_nsec = (orig_ts.tv_sec - orig_now.tv_sec) * NS_PER_SEC + (orig_ts.tv_nsec - orig_now.tv_nsec);
+  return a0_clock_add(target_now, add_nsec, target_ts);
+}
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif  // A0_SRC_CLOCK_H

src/libipc/platform/linux/a0/empty.h

@@ -0,0 +1,13 @@
+#ifndef A0_EMPTY_H
+#define A0_EMPTY_H
+
+// Bah. Why is there no consistent way to zero initialize a struct?
+#ifdef __cplusplus
+#define A0_EMPTY \
+  {}
+#else
+#define A0_EMPTY \
+  { 0 }
+#endif
+
+#endif  // A0_EMPTY_H