Visionipc v2.0 (commaai#101)

* add visionbuf

make static

ignore that

* Needs decleration

* add test binary

* c++

* create some structure

* some impl

* socket stuff

* Accept socket connection

* Alloc some buffers

* Create pub sockets and send buffer id

* make listener private

* Implement receive

* use error check macros in cl_helpers

* constructors to pass in opencl context

* add some convenience values in struct

* refactor creating buffers

* rgb is not so simple

* add fake stride and expose buffers

* add comment

* add extra data struct

* support conflate

* init opencl on all buffers

* make ion compile

* fix qcom2

* correctly setup yuv pointers when importing buffer

* also included from c

* Remove send print statements

* send metadata

* reveive metadata

* also used in c code

* dont start listener automatically

* Was started in 2 places

* set 100ms timeout on socket

* verify server id to detect reconnects

* handle reconnect

* buffer cleanup

* let user handle opencl creation

* add default values

* Add support for aligned rgb buffers

* add align macro

* dont use namespace

* use poller

* apple ifdef in ipc.cc

* VisionBuf is C++ class

* Install opencl headers

* cppcheck c++

* remove c header guard

* fix mac build

* simplify constructors

* Update visionipc/visionipc.h

Co-authored-by: Adeeb Shihadeh <adeebshihadeh@gmail.com>

* Update visionipc/visionbuf_ion.cc

Co-authored-by: Adeeb Shihadeh <adeebshihadeh@gmail.com>

* add brackets

* s/VIPCBufExtra/VisionIpcBufExtra/g

* Add unittesting harness

* remove vipc demo

* very basic tests

* add conflate test

* Install opencl

* suppress msgq warnings

* Make it work using zmq

* cl in qcom replay

* run unittests in zmq mode as well

* non blocking connect

* always larger frame queues

Co-authored-by: Adeeb Shihadeh <adeebshihadeh@gmail.com>

.github/workflows/tests.yml

@@ -34,7 +34,18 @@ jobs:
       run: eval "$BUILD"
     - name: Unit Tests
       run: |
-        $RUN "scons --test --asan -j$(nproc) && messaging/test_runner"
+        $RUN "scons --test --asan -j$(nproc) && messaging/test_runner && visionipc/test_runner"
+
+  unit_tests_zmq:
+    name: unit tests zmq
+    runs-on: ubuntu-20.04
+    steps:
+    - uses: actions/checkout@v2
+    - name: Build docker image
+      run: eval "$BUILD"
+    - name: Unit Tests
+      run: |
+        $RUN "scons --test --asan -j$(nproc) && ZMQ=1 messaging/test_runner && ZMQ=1 visionipc/test_runner"
 
   test_zmq:
     name: zmq

.gitignore

@@ -6,10 +6,16 @@ package-lock.json
 __pycache__
 .*.swp
 .*.swo
-libcereal*.a
+*.os
+*.o
+*.a
+
+test_runner
+
 libmessaging.*
 libmessaging_shared.*
 services.h
 .sconsign.dblite
 libcereal_shared.*
 .mypy_cache/
+catch2/

.pre-commit-config.yaml

@@ -36,6 +36,7 @@ repos:
         exclude: '^(messaging/msgq_tests.cc|messaging/test_runner.cc|messaging/catch2/)'
         args:
         - --error-exitcode=1
+        - --language=c++
         - --force
         - --quiet
         - -j4

Dockerfile

@@ -6,27 +6,29 @@ RUN apt-get update && apt-get install -y --no-install-recommends \
     build-essential \
     ca-certificates \
     capnproto \
-    cppcheck \
     clang \
+    cppcheck \
     curl \
     git \
-    libzmq3-dev \
-    libcapnp-dev \
-    libtool \
-    libssl-dev \
     libbz2-dev \
-    libreadline-dev \
-    libsqlite3-dev \
-    libncurses5-dev \
-    libncursesw5-dev \
+    libcapnp-dev \
     libffi-dev \
     liblzma-dev \
+    libncurses5-dev \
+    libncursesw5-dev \
+    libreadline-dev \
+    libsqlite3-dev \
+    libssl-dev \
+    libtool \
+    libzmq3-dev \
     llvm \
     make \
+    ocl-icd-opencl-dev \
+    opencl-headers  \
     python-openssl \
     tk-dev \
-    xz-utils \
     wget \
+    xz-utils \
     zlib1g-dev \
   && rm -rf /var/lib/apt/lists/*
 
@@ -42,6 +44,7 @@ RUN git clone https://github.com/catchorg/Catch2.git && \
     cd Catch2 && \
     git checkout 229cc4823c8cbe67366da8179efc6089dd3893e9 && \
     mv single_include/catch2 ../catch2 && \
+    cd .. \
     rm -rf Catch2
 
 WORKDIR /project/cereal

SConscript

@@ -1,12 +1,13 @@
-Import('env', 'envCython', 'arch', 'zmq')
+Import('env', 'envCython', 'arch', 'zmq', 'QCOM_REPLAY')
 
 import shutil
 
 cereal_dir = Dir('.')
 gen_dir = Dir('gen')
 messaging_dir = Dir('messaging')
 
-# TODO: remove src-prefix and cereal from command string. can we set working directory?
+# Build cereal
+
 env.Command(["gen/c/include/c++.capnp.h", "gen/c/include/java.capnp.h"], [], "mkdir -p " + gen_dir.path + "/c/include && touch $TARGETS")
 env.Command(['gen/cpp/car.capnp.c++', 'gen/cpp/log.capnp.c++', 'gen/cpp/car.capnp.h', 'gen/cpp/log.capnp.h'],
             ['car.capnp', 'log.capnp'],
@@ -26,7 +27,8 @@ cereal_objects = env.SharedObject([
 env.Library('cereal', cereal_objects)
 env.SharedLibrary('cereal_shared', cereal_objects)
 
-cereal_dir = Dir('.')
+# Build messaging
+
 services_h = env.Command(['services.h'],
                           ['service_list.yaml', 'services.py'],
                           'python3 ' + cereal_dir.path + '/services.py > $TARGET')
@@ -52,10 +54,25 @@ if arch == "aarch64":
 env.Program('messaging/bridge', ['messaging/bridge.cc'], LIBS=[messaging_lib, 'zmq'])
 Depends('messaging/bridge.cc', services_h)
 
-# different target?
-#env.Program('messaging/demo', ['messaging/demo.cc'], LIBS=[messaging_lib, 'zmq'])
-
 envCython.Program('messaging/messaging_pyx.so', 'messaging/messaging_pyx.pyx', LIBS=envCython["LIBS"]+[messaging_lib, "zmq"])
 
+
+# Build Vision IPC
+vipc_sources = [
+  'visionipc/ipc.cc',
+  'visionipc/visionipc_server.cc',
+  'visionipc/visionipc_client.cc',
+  'visionipc/visionbuf.cc',
+]
+
+if arch in ["aarch64", "larch64"] and (not QCOM_REPLAY):
+  vipc_sources += ['visionipc/visionbuf_ion.cc']
+else:
+  vipc_sources += ['visionipc/visionbuf_cl.cc']
+
+vipc_objects = env.SharedObject(vipc_sources)
+vipc = env.Library('visionipc', vipc_objects)
+
 if GetOption('test'):
   env.Program('messaging/test_runner', ['messaging/test_runner.cc', 'messaging/msgq_tests.cc'], LIBS=[messaging_lib])
+  env.Program('visionipc/test_runner', ['visionipc/test_runner.cc', 'visionipc/visionipc_tests.cc'], LIBS=[vipc, messaging_lib, 'zmq', 'pthread', 'OpenCL'])

SConstruct

@@ -1,15 +1,20 @@
 import os
+import platform
 import subprocess
 import sysconfig
 
 zmq = 'zmq'
-arch = subprocess.check_output(["uname", "-m"], encoding='utf8').rstrip()
 
+arch = subprocess.check_output(["uname", "-m"], encoding='utf8').rstrip()
+if platform.system() == "Darwin":
+  arch = "Darwin"
 
 cereal_dir = Dir('.')
+messaging_dir = Dir('./messaging')
 
 cpppath = [
   cereal_dir,
+  messaging_dir,
   '/usr/lib/include',
   sysconfig.get_paths()['include'],
 ]
@@ -45,24 +50,21 @@ env = Environment(
   CYTHONCFILESUFFIX=".cpp",
   tools=["default", "cython"]
 )
-Export('env', 'zmq', 'arch')
 
+QCOM_REPLAY = False
+Export('env', 'zmq', 'arch', 'QCOM_REPLAY')
 
-envCython = env.Clone()
-envCython["CCFLAGS"] += ["-Wno-#warnings", "-Wno-deprecated-declarations"]
 
-python_libs = []
+envCython = env.Clone(LIBS=[])
+envCython["CCFLAGS"] += ["-Wno-#warnings", "-Wno-deprecated-declarations"]
 if arch == "Darwin":
   envCython["LINKFLAGS"] = ["-bundle", "-undefined", "dynamic_lookup"]
 elif arch == "aarch64":
   envCython["LINKFLAGS"] = ["-shared"]
-
-  python_libs.append(os.path.basename(python_path))
+  envCython["LIBS"] = [os.path.basename(python_path)]
 else:
   envCython["LINKFLAGS"] = ["-pthread", "-shared"]
 
-envCython["LIBS"] = python_libs
-
 Export('envCython')
 
 

messaging/impl_msgq.cc

@@ -5,24 +5,32 @@
 #include <csignal>
 #include <cerrno>
 
-
+#include "services.h"
 #include "impl_msgq.hpp"
 
+
 volatile sig_atomic_t msgq_do_exit = 0;
 
 void sig_handler(int signal) {
   assert(signal == SIGINT || signal == SIGTERM);
   msgq_do_exit = 1;
 }
 
+static bool service_exists(std::string path){
+  for (const auto& it : services) {
+    if (it.name == path) {
+      return true;
+    }
+  }
+  return false;
+}
+
 static size_t get_size(std::string endpoint){
   size_t sz = DEFAULT_SEGMENT_SIZE;
 
-#if !defined(QCOM) && !defined(QCOM2)
   if (endpoint == "frame" || endpoint == "frontFrame" || endpoint == "wideFrame"){
     sz *= 10;
   }
-#endif
 
   return sz;
 }
@@ -61,10 +69,14 @@ MSGQMessage::~MSGQMessage() {
   this->close();
 }
 
-int MSGQSubSocket::connect(Context *context, std::string endpoint, std::string address, bool conflate){
+int MSGQSubSocket::connect(Context *context, std::string endpoint, std::string address, bool conflate, bool check_endpoint){
   assert(context);
   assert(address == "127.0.0.1");
 
+  if (check_endpoint && !service_exists(std::string(endpoint))){
+    std::cout << "Warning, " << std::string(endpoint) << " is not in service list." << std::endl;
+  }
+
   q = new msgq_queue_t;
   int r = msgq_new_queue(q, endpoint.c_str(), get_size(endpoint));
   if (r != 0){
@@ -150,9 +162,13 @@ MSGQSubSocket::~MSGQSubSocket(){
   }
 }
 
-int MSGQPubSocket::connect(Context *context, std::string endpoint){
+int MSGQPubSocket::connect(Context *context, std::string endpoint, bool check_endpoint){
   assert(context);
 
+  if (check_endpoint && !service_exists(std::string(endpoint))){
+    std::cout << "Warning, " << std::string(endpoint) << " is not in service list." << std::endl;
+  }
+
   q = new msgq_queue_t;
   int r = msgq_new_queue(q, endpoint.c_str(), get_size(endpoint));
   if (r != 0){

messaging/impl_msgq.hpp

@@ -34,7 +34,7 @@ class MSGQSubSocket : public SubSocket {
   msgq_queue_t * q = NULL;
   int timeout;
 public:
-  int connect(Context *context, std::string endpoint, std::string address, bool conflate=false);
+  int connect(Context *context, std::string endpoint, std::string address, bool conflate=false, bool check_endpoint=true);
   void setTimeout(int timeout);
   void * getRawSocket() {return (void*)q;}
   Message *receive(bool non_blocking=false);
@@ -45,7 +45,7 @@ class MSGQPubSocket : public PubSocket {
 private:
   msgq_queue_t * q = NULL;
 public:
-  int connect(Context *context, std::string endpoint);
+  int connect(Context *context, std::string endpoint, bool check_endpoint=true);
   int sendMessage(Message *message);
   int send(char *data, size_t size);
   ~MSGQPubSocket();

messaging/impl_zmq.cc

@@ -54,7 +54,7 @@ ZMQMessage::~ZMQMessage() {
 }
 
 
-int ZMQSubSocket::connect(Context *context, std::string endpoint, std::string address, bool conflate){
+int ZMQSubSocket::connect(Context *context, std::string endpoint, std::string address, bool conflate, bool check_endpoint){
   sock = zmq_socket(context->getRawContext(), ZMQ_SUB);
   if (sock == NULL){
     return -1;
@@ -71,7 +71,11 @@ int ZMQSubSocket::connect(Context *context, std::string endpoint, std::string ad
   zmq_setsockopt(sock, ZMQ_RECONNECT_IVL_MAX, &reconnect_ivl, sizeof(reconnect_ivl));
 
   full_endpoint = "tcp://" + address + ":";
-  full_endpoint += std::to_string(get_port(endpoint));
+  if (check_endpoint){
+    full_endpoint += std::to_string(get_port(endpoint));
+  } else {
+    full_endpoint += endpoint;
+  }
 
   return zmq_connect(sock, full_endpoint.c_str());
 }
@@ -103,14 +107,18 @@ ZMQSubSocket::~ZMQSubSocket(){
   zmq_close(sock);
 }
 
-int ZMQPubSocket::connect(Context *context, std::string endpoint){
+int ZMQPubSocket::connect(Context *context, std::string endpoint, bool check_endpoint){
   sock = zmq_socket(context->getRawContext(), ZMQ_PUB);
   if (sock == NULL){
     return -1;
   }
 
   full_endpoint = "tcp://*:";
-  full_endpoint += std::to_string(get_port(endpoint));
+  if (check_endpoint){
+    full_endpoint += std::to_string(get_port(endpoint));
+  } else {
+    full_endpoint += endpoint;
+  }
 
   return zmq_bind(sock, full_endpoint.c_str());
 }

messaging/impl_zmq.hpp

@@ -32,7 +32,7 @@ class ZMQSubSocket : public SubSocket {
   void * sock;
   std::string full_endpoint;
 public:
-  int connect(Context *context, std::string endpoint, std::string address, bool conflate=false);
+  int connect(Context *context, std::string endpoint, std::string address, bool conflate=false, bool check_endpoint=true);
   void setTimeout(int timeout);
   void * getRawSocket() {return sock;}
   Message *receive(bool non_blocking=false);
@@ -44,7 +44,7 @@ class ZMQPubSocket : public PubSocket {
   void * sock;
   std::string full_endpoint;
 public:
-  int connect(Context *context, std::string endpoint);
+  int connect(Context *context, std::string endpoint, bool check_endpoint=true);
   int sendMessage(Message *message);
   int send(char *data, size_t size);
   ~ZMQPubSocket();