Merge remote-tracking branch 'upstream/master' into cereal_fix_signal

Author: sshane

msgq/impl_zmq.cc

@@ -7,10 +7,19 @@
 
 #include "msgq/impl_zmq.h"
 
+static size_t fnv1a_hash(const std::string &str) {
+    const size_t fnv_prime = 0x100000001b3;
+    size_t hash_value = 0xcbf29ce484222325;
+    for (char c : str) {
+        hash_value ^= (unsigned char)c;
+        hash_value *= fnv_prime;
+    }
+    return hash_value;
+}
+
 //FIXME: This is a hack to get the port number from the socket name, might have collisions
 static int get_port(std::string endpoint) {
-    std::hash<std::string> hasher;
-    size_t hash_value = hasher(endpoint);
+    size_t hash_value = fnv1a_hash(endpoint);
     int start_port = 8023;
     int max_port = 65535;
     int port = start_port + (hash_value % (max_port - start_port));

msgq/visionipc/tests/test_visionipc.py

@@ -11,10 +11,10 @@ def zmq_sleep(t=1):
 
 class TestVisionIpc:
 
-  def setup_vipc(self, name, *stream_types, num_buffers=1, rgb=False, width=100, height=100, conflate=False):
+  def setup_vipc(self, name, *stream_types, num_buffers=1, width=100, height=100, conflate=False):
     self.server = VisionIpcServer(name)
     for stream_type in stream_types:
-      self.server.create_buffers(stream_type, num_buffers, rgb, width, height)
+      self.server.create_buffers(stream_type, num_buffers, width, height)
     self.server.start_listener()
 
     if len(stream_types):
@@ -51,15 +51,6 @@ def test_buffers(self):
     del self.client
     del self.server
 
-  def test_yuv_rgb(self):
-    _, client_yuv = self.setup_vipc("camerad", VisionStreamType.VISION_STREAM_ROAD, rgb=False)
-    _, client_rgb = self.setup_vipc("navd", VisionStreamType.VISION_STREAM_MAP, rgb=True)
-    assert client_rgb.rgb
-    assert not client_yuv.rgb
-    del client_yuv
-    del client_rgb
-    del self.server
-
   def test_send_single_buffer(self):
     self.setup_vipc("camerad", VisionStreamType.VISION_STREAM_ROAD)
 

msgq/visionipc/visionbuf.cc

@@ -1,21 +1,6 @@
 #include "msgq/visionipc/visionbuf.h"
 
-#define ALIGN(x, align) (((x) + (align)-1) & ~((align)-1))
-
-void visionbuf_compute_aligned_width_and_height(int width, int height, int *aligned_w, int *aligned_h) {
-  *aligned_w = width;
-  *aligned_h = height;
-}
-
-void VisionBuf::init_rgb(size_t init_width, size_t init_height, size_t init_stride) {
-  this->rgb = true;
-  this->width = init_width;
-  this->height = init_height;
-  this->stride = init_stride;
-}
-
 void VisionBuf::init_yuv(size_t init_width, size_t init_height, size_t init_stride, size_t init_uv_offset){
-  this->rgb = false;
   this->width = init_width;
   this->height = init_height;
   this->stride = init_stride;

msgq/visionipc/visionbuf.h

@@ -29,7 +29,6 @@ class VisionBuf {
   uint64_t *frame_id;
   int fd = 0;
 
-  bool rgb = false;
   size_t width = 0;
   size_t height = 0;
   size_t stride = 0;
@@ -54,13 +53,10 @@ class VisionBuf {
   void allocate(size_t len);
   void import();
   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, size_t stride, size_t uv_offset);
   int sync(int dir);
   int free();
 
   void set_frame_id(uint64_t id);
   uint64_t get_frame_id();
 };
-
-void visionbuf_compute_aligned_width_and_height(int width, int height, int *aligned_w, int *aligned_h);

msgq/visionipc/visionipc.pxd

@@ -21,7 +21,6 @@ cdef extern from "msgq/visionipc/visionbuf.h":
 
   cdef cppclass VisionBuf:
     void * addr
-    bool rgb
     size_t len
     size_t width
     size_t height
@@ -42,8 +41,8 @@ cdef extern from "msgq/visionipc/visionipc_server.h":
 
   cdef cppclass VisionIpcServer:
     VisionIpcServer(string, void*, void*)
-    void create_buffers(VisionStreamType, size_t, bool, size_t, size_t)
-    void create_buffers_with_sizes(VisionStreamType, size_t, bool, size_t, size_t, size_t, size_t, size_t)
+    void create_buffers(VisionStreamType, size_t, size_t, size_t)
+    void create_buffers_with_sizes(VisionStreamType, size_t, size_t, size_t, size_t, size_t, size_t)
     VisionBuf * get_buffer(VisionStreamType)
     void send(VisionBuf *, VisionIpcBufExtra *, bool)
     void start_listener()

msgq/visionipc/visionipc_client.cc

@@ -7,8 +7,8 @@
 #include "msgq/visionipc/visionipc.h"
 #include "msgq/visionipc/visionipc_client.h"
 #include "msgq/visionipc/visionipc_server.h"
-#include "logger/logger.h"
-#include "logger/logger.h"
+#include "msgq/logger/logger.h"
+#include "msgq/logger/logger.h"
 
 static int connect_to_vipc_server(const std::string &name, bool blocking) {
   const std::string ipc_path = get_ipc_path(name);
@@ -54,6 +54,12 @@ bool VisionIpcClient::connect(bool blocking){
   int fds[VISIONIPC_MAX_FDS];
   VisionBuf bufs[VISIONIPC_MAX_FDS];
   r = ipc_sendrecv_with_fds(false, socket_fd, &bufs, sizeof(bufs), fds, VISIONIPC_MAX_FDS, &num_buffers);
+  if (r < 0) {
+    // only expected error is server shutting down
+    assert(errno == ECONNRESET);
+    close(socket_fd);
+    return false;
+  }
 
   assert(num_buffers >= 0);
   assert(r == sizeof(VisionBuf) * num_buffers);
@@ -63,11 +69,7 @@ bool VisionIpcClient::connect(bool blocking){
     buffers[i] = bufs[i];
     buffers[i].fd = fds[i];
     buffers[i].import();
-    if (buffers[i].rgb) {
-      buffers[i].init_rgb(buffers[i].width, buffers[i].height, buffers[i].stride);
-    } else {
-      buffers[i].init_yuv(buffers[i].width, buffers[i].height, buffers[i].stride, buffers[i].uv_offset);
-    }
+    buffers[i].init_yuv(buffers[i].width, buffers[i].height, buffers[i].stride, buffers[i].uv_offset);
 
     if (device_id) buffers[i].init_cl(device_id, ctx);
   }
@@ -126,7 +128,14 @@ std::set<VisionStreamType> VisionIpcClient::getAvailableStreams(const std::strin
 
   VisionStreamType available_streams[VISION_STREAM_MAX] = {};
   r = ipc_sendrecv_with_fds(false, socket_fd, &available_streams, sizeof(available_streams), nullptr, 0, nullptr);
-  assert((r >= 0) && (r % sizeof(VisionStreamType) == 0));
+  if (r < 0) {
+    // only expected error is server shutting down
+    assert(errno == ECONNRESET);
+    close(socket_fd);
+    return {};
+  }
+
+  assert(r % sizeof(VisionStreamType) == 0);
   close(socket_fd);
   return std::set<VisionStreamType>(available_streams, available_streams + r / sizeof(VisionStreamType));
 }

msgq/visionipc/visionipc_pyx.pyx

@@ -55,22 +55,18 @@ cdef class VisionBuf:
   def uv_offset(self):
     return self.buf.uv_offset
 
-  @property
-  def rgb(self):
-    return self.buf.rgb
-
 
 cdef class VisionIpcServer:
   cdef cppVisionIpcServer * server
 
   def __init__(self, string name):
     self.server = new cppVisionIpcServer(name, NULL, NULL)
 
-  def create_buffers(self, VisionStreamType tp, size_t num_buffers, bool rgb, size_t width, size_t height):
-    self.server.create_buffers(tp, num_buffers, rgb, width, height)
+  def create_buffers(self, VisionStreamType tp, size_t num_buffers, size_t width, size_t height):
+    self.server.create_buffers(tp, num_buffers, width, height)
 
-  def create_buffers_with_sizes(self, VisionStreamType tp, size_t num_buffers, bool rgb, size_t width, size_t height, size_t size, size_t stride, size_t uv_offset):
-    self.server.create_buffers_with_sizes(tp, num_buffers, rgb, width, height, size, stride, uv_offset)
+  def create_buffers_with_sizes(self, VisionStreamType tp, size_t num_buffers, size_t width, size_t height, size_t size, size_t stride, size_t uv_offset):
+    self.server.create_buffers_with_sizes(tp, num_buffers, width, height, size, stride, uv_offset)
 
   def send(self, VisionStreamType tp, const unsigned char[:] data, uint32_t frame_id=0, uint64_t timestamp_sof=0, uint64_t timestamp_eof=0):
     cdef cppVisionBuf * buf = self.server.get_buffer(tp)
@@ -123,10 +119,6 @@ cdef class VisionIpcClient:
   def uv_offset(self):
     return self.client.buffers[0].uv_offset if self.client.num_buffers else None
 
-  @property
-  def rgb(self):
-    return self.client.buffers[0].rgb if self.client.num_buffers else None
-
   @property
   def buffer_len(self):
     return self.client.buffers[0].len if self.client.num_buffers else None

msgq/visionipc/visionipc_server.cc

@@ -11,7 +11,7 @@
 #include "msgq/ipc.h"
 #include "msgq/visionipc/visionipc.h"
 #include "msgq/visionipc/visionipc_server.h"
-#include "logger/logger.h"
+#include "msgq/logger/logger.h"
 
 std::string get_endpoint_name(std::string name, VisionStreamType type){
   if (messaging_use_zmq()){
@@ -38,29 +38,22 @@ VisionIpcServer::VisionIpcServer(std::string name, cl_device_id device_id, cl_co
   server_id = distribution(rd);
 }
 
-void VisionIpcServer::create_buffers(VisionStreamType type, size_t num_buffers, bool rgb, size_t width, size_t height){
+void VisionIpcServer::create_buffers(VisionStreamType type, size_t num_buffers, size_t width, size_t height){
   // TODO: assert that this type is not created yet
   assert(num_buffers < VISIONIPC_MAX_FDS);
-  int aligned_w = 0, aligned_h = 0;
 
   size_t size = 0;
   size_t stride = 0;
   size_t uv_offset = 0;
 
-  if (rgb) {
-    visionbuf_compute_aligned_width_and_height(width, height, &aligned_w, &aligned_h);
-    size = (size_t)aligned_w * (size_t)aligned_h * 3;
-    stride = aligned_w * 3;
-  } else {
-    size = width * height * 3 / 2;
-    stride = width;
-    uv_offset = width * height;
-  }
+  size = width * height * 3 / 2;
+  stride = width;
+  uv_offset = width * height;
 
-  create_buffers_with_sizes(type, num_buffers, rgb, width, height, size, stride, uv_offset);
+  create_buffers_with_sizes(type, num_buffers, width, height, size, stride, uv_offset);
 }
 
-void VisionIpcServer::create_buffers_with_sizes(VisionStreamType type, size_t num_buffers, bool rgb, size_t width, size_t height, size_t size, size_t stride, size_t uv_offset) {
+void VisionIpcServer::create_buffers_with_sizes(VisionStreamType type, size_t num_buffers, size_t width, size_t height, size_t size, size_t stride, size_t uv_offset) {
   // Create map + alloc requested buffers
   for (size_t i = 0; i < num_buffers; i++){
     VisionBuf* buf = new VisionBuf();
@@ -70,7 +63,7 @@ void VisionIpcServer::create_buffers_with_sizes(VisionStreamType type, size_t nu
 
     if (device_id) buf->init_cl(device_id, ctx);
 
-    rgb ? buf->init_rgb(width, height, stride) : buf->init_yuv(width, height, stride, uv_offset);
+    buf->init_yuv(width, height, stride, uv_offset);
 
     buffers[type].push_back(buf);
   }
@@ -167,11 +160,17 @@ void VisionIpcServer::listener(){
 
 
 
-VisionBuf * VisionIpcServer::get_buffer(VisionStreamType type){
+VisionBuf * VisionIpcServer::get_buffer(VisionStreamType type, int idx){
   // Do we want to keep track if the buffer has been sent out yet and warn user?
   assert(buffers.count(type));
   auto b = buffers[type];
-  return b[cur_idx[type]++ % b.size()];
+  if (idx < 0) {
+    idx = cur_idx[type]++ % b.size();
+  } else {
+    assert(idx < b.size() && idx >= 0);
+    cur_idx[type] = idx;
+  }
+  return b[idx];
 }
 
 void VisionIpcServer::send(VisionBuf * buf, VisionIpcBufExtra * extra, bool sync){

msgq/visionipc/visionipc_server.h

@@ -33,10 +33,10 @@ class VisionIpcServer {
   VisionIpcServer(std::string name, cl_device_id device_id=nullptr, cl_context ctx=nullptr);
   ~VisionIpcServer();
 
-  VisionBuf * get_buffer(VisionStreamType type);
+  VisionBuf * get_buffer(VisionStreamType type, int idx = -1);
 
-  void create_buffers(VisionStreamType type, size_t num_buffers, bool rgb, size_t width, size_t height);
-  void create_buffers_with_sizes(VisionStreamType type, size_t num_buffers, bool rgb, size_t width, size_t height, size_t size, size_t stride, size_t uv_offset);
+  void create_buffers(VisionStreamType type, size_t num_buffers, size_t width, size_t height);
+  void create_buffers_with_sizes(VisionStreamType type, size_t num_buffers, size_t width, size_t height, size_t size, size_t stride, size_t uv_offset);
   void send(VisionBuf * buf, VisionIpcBufExtra * extra, bool sync=true);
   void start_listener();
 };

msgq/visionipc/visionipc_tests.cc

@@ -15,7 +15,7 @@ static void zmq_sleep(int milliseconds=1000){
 
 TEST_CASE("Connecting"){
   VisionIpcServer server("camerad");
-  server.create_buffers(VISION_STREAM_ROAD, 1, false, 100, 100);
+  server.create_buffers(VISION_STREAM_ROAD, 1, 100, 100);
   server.start_listener();
 
   VisionIpcClient client = VisionIpcClient("camerad", VISION_STREAM_ROAD, false);
@@ -26,8 +26,8 @@ TEST_CASE("Connecting"){
 
 TEST_CASE("getAvailableStreams"){
   VisionIpcServer server("camerad");
-  server.create_buffers(VISION_STREAM_ROAD, 1, false, 100, 100);
-  server.create_buffers(VISION_STREAM_WIDE_ROAD, 1, false, 100, 100);
+  server.create_buffers(VISION_STREAM_ROAD, 1, 100, 100);
+  server.create_buffers(VISION_STREAM_WIDE_ROAD, 1, 100, 100);
   server.start_listener();
   auto available_streams = VisionIpcClient::getAvailableStreams("camerad");
   REQUIRE(available_streams.size() == 2);
@@ -38,7 +38,7 @@ TEST_CASE("getAvailableStreams"){
 TEST_CASE("Check buffers"){
   size_t width = 100, height = 200, num_buffers = 5;
   VisionIpcServer server("camerad");
-  server.create_buffers(VISION_STREAM_ROAD, num_buffers, false, width, height);
+  server.create_buffers(VISION_STREAM_ROAD, num_buffers, width, height);
   server.start_listener();
 
   VisionIpcClient client = VisionIpcClient("camerad", VISION_STREAM_ROAD, false);
@@ -50,24 +50,9 @@ TEST_CASE("Check buffers"){
   REQUIRE(client.num_buffers == num_buffers);
 }
 
-TEST_CASE("Check yuv/rgb"){
-  VisionIpcServer server("camerad");
-  server.create_buffers(VISION_STREAM_ROAD, 1, false, 100, 100);
-  server.create_buffers(VISION_STREAM_MAP, 1, true, 100, 100);
-  server.start_listener();
-
-  VisionIpcClient client_yuv = VisionIpcClient("camerad", VISION_STREAM_ROAD, false);
-  VisionIpcClient client_rgb = VisionIpcClient("camerad", VISION_STREAM_MAP, false);
-  client_yuv.connect();
-  client_rgb.connect();
-
-  REQUIRE(client_rgb.buffers[0].rgb == true);
-  REQUIRE(client_yuv.buffers[0].rgb == false);
-}
-
 TEST_CASE("Send single buffer"){
   VisionIpcServer server("camerad");
-  server.create_buffers(VISION_STREAM_ROAD, 1, true, 100, 100);
+  server.create_buffers(VISION_STREAM_ROAD, 1, 100, 100);
   server.start_listener();
 
   VisionIpcClient client = VisionIpcClient("camerad", VISION_STREAM_ROAD, false);
@@ -96,7 +81,7 @@ TEST_CASE("Send single buffer"){
 
 TEST_CASE("Test no conflate"){
   VisionIpcServer server("camerad");
-  server.create_buffers(VISION_STREAM_ROAD, 1, true, 100, 100);
+  server.create_buffers(VISION_STREAM_ROAD, 1, 100, 100);
   server.start_listener();
 
   VisionIpcClient client = VisionIpcClient("camerad", VISION_STREAM_ROAD, false);
@@ -124,7 +109,7 @@ TEST_CASE("Test no conflate"){
 
 TEST_CASE("Test conflate"){
   VisionIpcServer server("camerad");
-  server.create_buffers(VISION_STREAM_ROAD, 1, true, 100, 100);
+  server.create_buffers(VISION_STREAM_ROAD, 1, 100, 100);
   server.start_listener();
 
   VisionIpcClient client = VisionIpcClient("camerad", VISION_STREAM_ROAD, true);