Improve VisionIPC error handling (commaai#184)

* add return values to visionbuf

* add mock logger and log prints

* add logging to server

* unused

* link in common

SConscript

@@ -1,4 +1,4 @@
-Import('env', 'envCython', 'arch')
+Import('env', 'envCython', 'arch', 'common')
 
 import shutil
 
@@ -40,10 +40,10 @@ messaging_objects = env.SharedObject([
 messaging_lib = env.Library('messaging', messaging_objects)
 Depends('messaging/impl_zmq.cc', services_h)
 
-env.Program('messaging/bridge', ['messaging/bridge.cc'], LIBS=[messaging_lib, 'zmq'])
+env.Program('messaging/bridge', ['messaging/bridge.cc'], LIBS=[messaging_lib, 'zmq', common])
 Depends('messaging/bridge.cc', services_h)
 
-envCython.Program('messaging/messaging_pyx.so', 'messaging/messaging_pyx.pyx', LIBS=envCython["LIBS"]+[messaging_lib, "zmq"])
+envCython.Program('messaging/messaging_pyx.so', 'messaging/messaging_pyx.pyx', LIBS=envCython["LIBS"]+[messaging_lib, "zmq", common])
 
 
 # Build Vision IPC
@@ -63,7 +63,7 @@ vipc_objects = env.SharedObject(vipc_sources)
 vipc = env.Library('visionipc', vipc_objects)
 
 
-libs = envCython["LIBS"]+["OpenCL", "zmq", vipc, messaging_lib]
+libs = envCython["LIBS"]+["OpenCL", "zmq", vipc, messaging_lib, common]
 if arch == "aarch64":
   libs += ["adreno_utils"]
 if arch == "Darwin":
@@ -73,5 +73,5 @@ envCython.Program('visionipc/visionipc_pyx.so', 'visionipc/visionipc_pyx.pyx', L
 
 
 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'])
+  env.Program('messaging/test_runner', ['messaging/test_runner.cc', 'messaging/msgq_tests.cc'], LIBS=[messaging_lib, common])
+  env.Program('visionipc/test_runner', ['visionipc/test_runner.cc', 'visionipc/visionipc_tests.cc'], LIBS=[vipc, messaging_lib, 'zmq', 'pthread', 'OpenCL', common])

SConstruct

@@ -9,6 +9,7 @@ if platform.system() == "Darwin":
 
 cereal_dir = Dir('.')
 messaging_dir = Dir('./messaging')
+common = ''
 
 cpppath = [
   cereal_dir,
@@ -55,7 +56,7 @@ env = Environment(
   tools=["default", "cython"]
 )
 
-Export('env', 'arch')
+Export('env', 'arch', 'common')
 
 envCython = env.Clone(LIBS=[])
 envCython["CCFLAGS"] += ["-Wno-#warnings", "-Wno-deprecated-declarations"]

logger/logger.h

@@ -0,0 +1,20 @@
+#pragma once
+
+#ifdef SWAGLOG
+#include "selfdrive/common/swaglog.h"
+#else
+
+#define CLOUDLOG_DEBUG 10
+#define CLOUDLOG_INFO 20
+#define CLOUDLOG_WARNING 30
+#define CLOUDLOG_ERROR 40
+#define CLOUDLOG_CRITICAL 50
+
+#define cloudlog(lvl, fmt, ...) printf(fmt "\n", ## __VA_ARGS__)
+
+#define LOGD(fmt, ...) cloudlog(CLOUDLOG_DEBUG, fmt, ## __VA_ARGS__)
+#define LOG(fmt, ...) cloudlog(CLOUDLOG_INFO, fmt, ## __VA_ARGS__)
+#define LOGW(fmt, ...) cloudlog(CLOUDLOG_WARNING, fmt, ## __VA_ARGS__)
+#define LOGE(fmt, ...) cloudlog(CLOUDLOG_ERROR, fmt, ## __VA_ARGS__)
+
+#endif

visionipc/visionbuf.h

@@ -55,8 +55,8 @@ class VisionBuf {
   void init_cl(cl_device_id device_id, cl_context ctx);
   void init_rgb(size_t width, size_t height, size_t stride);
   void init_yuv(size_t width, size_t height);
-  void sync(int dir);
-  void free();
+  int sync(int dir);
+  int free();
 };
 
 void visionbuf_compute_aligned_width_and_height(int width, int height, int *aligned_w, int *aligned_h);

visionipc/visionbuf_cl.cc

@@ -60,27 +60,36 @@ void VisionBuf::import(){
 }
 
 
-void VisionBuf::sync(int dir) {
+int VisionBuf::sync(int dir) {
   int err = 0;
-  if (!this->buf_cl) return;
+  if (!this->buf_cl) return 0;
 
   if (dir == VISIONBUF_SYNC_FROM_DEVICE) {
     err = clEnqueueReadBuffer(this->copy_q, this->buf_cl, CL_FALSE, 0, this->len, this->addr, 0, NULL, NULL);
   } else {
     err = clEnqueueWriteBuffer(this->copy_q, this->buf_cl, CL_FALSE, 0, this->len, this->addr, 0, NULL, NULL);
   }
-  assert(err == 0);
-  clFinish(this->copy_q);
+
+  if (err == 0){
+    err = clFinish(this->copy_q);
+  }
+
+  return err;
 }
 
-void VisionBuf::free() {
+int VisionBuf::free() {
+  int err = 0;
   if (this->buf_cl){
-    int err = clReleaseMemObject(this->buf_cl);
-    assert(err == 0);
+    err = clReleaseMemObject(this->buf_cl);
+    if (err != 0) return err;
 
-    clReleaseCommandQueue(this->copy_q);
+    err = clReleaseCommandQueue(this->copy_q);
+    if (err != 0) return err;
   }
 
-  munmap(this->addr, this->len);
-  close(this->fd);
+  err = munmap(this->addr, this->len);
+  if (err != 0) return err;
+
+  err = close(this->fd);
+  return err;
 }

visionipc/visionbuf_ion.cc

@@ -104,9 +104,7 @@ void VisionBuf::init_cl(cl_device_id device_id, cl_context ctx) {
 }
 
 
-void VisionBuf::sync(int dir) {
-  int err;
-
+int VisionBuf::sync(int dir) {
   struct ion_flush_data flush_data = {0};
   flush_data.handle = this->handle;
   flush_data.vaddr = this->addr;
@@ -124,19 +122,23 @@ void VisionBuf::sync(int dir) {
      ION_IOC_INV_CACHES : ION_IOC_CLEAN_CACHES;
 
   custom_data.arg = (unsigned long)&flush_data;
-  err = ioctl(ion_fd, ION_IOC_CUSTOM, &custom_data);
-  assert(err == 0);
+  return ioctl(ion_fd, ION_IOC_CUSTOM, &custom_data);
 }
 
-void VisionBuf::free() {
+int VisionBuf::free() {
+  int err = 0;
+
   if (this->buf_cl){
-    int err = clReleaseMemObject(this->buf_cl);
-    assert(err == 0);
+    err = clReleaseMemObject(this->buf_cl);
+    if (err != 0) return err;
   }
 
-  munmap(this->addr, this->mmap_len);
-  close(this->fd);
+  err = munmap(this->addr, this->mmap_len);
+  if (err != 0) return err;
+
+  err = close(this->fd);
+  if (err != 0) return err;
 
   struct ion_handle_data handle_data = {.handle = this->handle};
-  ioctl(ion_fd, ION_IOC_FREE, &handle_data);
+  return ioctl(ion_fd, ION_IOC_FREE, &handle_data);
 }

visionipc/visionipc_client.cc

@@ -3,9 +3,10 @@
 #include <iostream>
 #include <thread>
 
-#include "ipc.h"
-#include "visionipc_client.h"
-#include "visionipc_server.h"
+#include "visionipc/ipc.h"
+#include "visionipc/visionipc_client.h"
+#include "visionipc/visionipc_server.h"
+#include "logger/logger.h"
 
 VisionIpcClient::VisionIpcClient(std::string name, VisionStreamType type, bool conflate, cl_device_id device_id, cl_context ctx) : name(name), type(type), device_id(device_id), ctx(ctx) {
   msg_ctx = Context::create();
@@ -21,8 +22,11 @@ bool VisionIpcClient::connect(bool blocking){
 
   // Cleanup old buffers on reconnect
   for (size_t i = 0; i < num_buffers; i++){
-    buffers[i].free();
+    if (buffers[i].free() != 0) {
+      LOGE("Failed to free buffer %zu", i);
+    }
   }
+
   num_buffers = 0;
 
   // Connect to server socket and ask for all FDs of type
@@ -101,7 +105,10 @@ VisionBuf * VisionIpcClient::recv(VisionIpcBufExtra * extra, const int timeout_m
     *extra = packet->extra;
   }
 
-  buf->sync(VISIONBUF_SYNC_TO_DEVICE);
+  if (buf->sync(VISIONBUF_SYNC_TO_DEVICE) != 0) {
+    LOGE("Failed to sync buffer");
+  }
+
   delete r;
   return buf;
 }
@@ -110,7 +117,9 @@ VisionBuf * VisionIpcClient::recv(VisionIpcBufExtra * extra, const int timeout_m
 
 VisionIpcClient::~VisionIpcClient(){
   for (size_t i = 0; i < num_buffers; i++){
-    buffers[i].free();
+    if (buffers[i].free() != 0) {
+      LOGE("Failed to free buffer %zu", i);
+    }
   }
 
   delete sock;

visionipc/visionipc_server.cc

@@ -10,6 +10,7 @@
 #include "messaging/messaging.h"
 #include "visionipc/ipc.h"
 #include "visionipc/visionipc_server.h"
+#include "logger/logger.h"
 
 std::string get_endpoint_name(std::string name, VisionStreamType type){
   if (messaging_use_zmq()){
@@ -145,7 +146,11 @@ VisionBuf * VisionIpcServer::get_buffer(VisionStreamType type){
 }
 
 void VisionIpcServer::send(VisionBuf * buf, VisionIpcBufExtra * extra, bool sync){
-  if (sync) buf->sync(VISIONBUF_SYNC_FROM_DEVICE);
+  if (sync) {
+    if (buf->sync(VISIONBUF_SYNC_FROM_DEVICE) != 0) {
+      LOGE("Failed to sync buffer");
+    }
+  }
   assert(buffers.count(buf->type));
   assert(buf->idx < buffers[buf->type].size());
 
@@ -165,7 +170,9 @@ VisionIpcServer::~VisionIpcServer(){
   // VisionBuf cleanup
   for( auto const& [type, buf] : buffers ) {
     for (VisionBuf* b : buf){
-      b->free();
+      if (b->free() != 0) {
+        LOGE("Failed to free buffer");
+      }
       delete b;
     }
   }