swaglog: replace zmq with unix datagram socket (#39157)

SConstruct

@@ -292,7 +292,7 @@ SConscript(['opendbc_repo/opendbc/can/SConscript'])
 # Build common module
 SConscript(['openpilot/common/SConscript'])
 Import('_common')
-common = [_common, 'json11', 'zmq']
+common = [_common, 'json11']
 Export('common')
 
 # Build messaging (cereal + msgq + socketmaster + their dependencies)

openpilot/cereal/SConscript

@@ -15,7 +15,7 @@ cereal = env.Library('cereal', [f'gen/cpp/{s}.c++' for s in all_output])
 
 # Build messaging
 services_h = env.Command(['services.h'], ['services.py'], 'python3 ' + cereal_dir.path + '/services.py > $TARGET')
-env.Program('messaging/bridge', ['messaging/bridge.cc', 'messaging/msgq_to_zmq.cc', 'messaging/bridge_zmq.cc'], LIBS=[msgq, common, 'pthread'])
+env.Program('messaging/bridge', ['messaging/bridge.cc', 'messaging/msgq_to_zmq.cc', 'messaging/bridge_zmq.cc'], LIBS=[msgq, common, 'zmq', 'pthread'])
 
 socketmaster = env.Library('socketmaster', ['messaging/socketmaster.cc'])
 

openpilot/common/SConscript

@@ -11,12 +11,12 @@ common_libs = [
 _common = env.Library('common', common_libs, LIBS="json11")
 Export('_common')
 
-params_python = env.SharedLibrary('params_c', 'params_c.cc', LIBS=[_common, 'zmq', 'json11'])
+params_python = env.SharedLibrary('params_c', 'params_c.cc', LIBS=[_common, 'json11'])
 common_python = [params_python]
 Export('common_python')
 
 if GetOption('extras'):
-  env.Program('tests/test_swaglog', 'tests/test_swaglog.cc', LIBS=[_common, 'json11', 'zmq', 'pthread'])
+  env.Program('tests/test_swaglog', 'tests/test_swaglog.cc', LIBS=[_common, 'json11', 'pthread'])
 
   env.Program('tests/bin/test_params_fork_lock', 'tests/test_params_fork_lock.cc',
               LIBS=[_common, 'json11', 'zmq', 'capnp', 'kj', 'pthread'])

openpilot/common/hardware/hw.h

@@ -38,7 +38,7 @@ namespace Path {
   }
 
   inline std::string swaglog_ipc() {
-    return "ipc:///tmp/logmessage" + Path::openpilot_prefix();
+    return "/tmp/logmessage" + Path::openpilot_prefix();
   }
 
   inline std::string download_cache_root() {

openpilot/common/hardware/hw.py

@@ -29,7 +29,7 @@ class Paths:
 
   @staticmethod
   def swaglog_ipc() -> str:
-    return "ipc:///tmp/logmessage" + os.environ.get("OPENPILOT_PREFIX", "")
+    return "/tmp/logmessage" + os.environ.get("OPENPILOT_PREFIX", "")
 
   @staticmethod
   def download_cache_root() -> str:

openpilot/common/swaglog.cc

@@ -9,8 +9,10 @@
 #include <mutex>
 #include <string>
 
-#include <zmq.h>
 #include <stdarg.h>
+#include <sys/socket.h>
+#include <sys/un.h>
+#include <unistd.h>
 #include "json11/json11.hpp"
 #include "common/version.h"
 #include "common/hardware/hw.h"
@@ -18,13 +20,11 @@
 class SwaglogState {
 public:
   SwaglogState() {
-    zctx = zmq_ctx_new();
-    sock = zmq_socket(zctx, ZMQ_PUSH);
-
-    // Timeout on shutdown for messages to be received by the logging process
-    int timeout = 100;
-    zmq_setsockopt(sock, ZMQ_LINGER, &timeout, sizeof(timeout));
-    zmq_connect(sock, Path::swaglog_ipc().c_str());
+    sock = socket(AF_UNIX, SOCK_DGRAM, 0);
+    int sndbuf = SWAGLOG_MAX_SIZE;
+    setsockopt(sock, SOL_SOCKET, SO_SNDBUF, &sndbuf, sizeof(sndbuf));
+    addr.sun_family = AF_UNIX;
+    strncpy(addr.sun_path, Path::swaglog_ipc().c_str(), sizeof(addr.sun_path) - 1);
 
     // workaround for https://github.com/dropbox/json11/issues/38
     setlocale(LC_NUMERIC, "C");
@@ -62,8 +62,7 @@ public:
   }
 
   ~SwaglogState() {
-    zmq_close(sock);
-    zmq_ctx_destroy(zctx);
+    close(sock);
   }
 
   void log(int levelnum, const char* filename, int lineno, const char* func, const char* msg, const std::string& log_s) {
@@ -71,12 +70,13 @@ public:
     if (levelnum >= print_level) {
       printf("%s: %s\n", filename, msg);
     }
-    zmq_send(sock, log_s.data(), log_s.length(), ZMQ_NOBLOCK);
+    // drop if logmessaged is busy, not running, or the message is too big
+    sendto(sock, log_s.data(), log_s.length(), MSG_DONTWAIT, (struct sockaddr*)&addr, sizeof(addr));
   }
 
   std::mutex lock;
-  void* zctx = nullptr;
-  void* sock = nullptr;
+  int sock = -1;
+  struct sockaddr_un addr = {};
   int print_level;
   json11::Json::object ctx_j;
 };

openpilot/common/swaglog.h

@@ -8,6 +8,7 @@
 #define CLOUDLOG_ERROR 40
 #define CLOUDLOG_CRITICAL 50
 
+#define SWAGLOG_MAX_SIZE (4*1024*1024)
 
 #ifdef __GNUC__
 #define SWAG_LOG_CHECK_FMT(a, b) __attribute__ ((format (printf, a, b)))

openpilot/common/swaglog.py

@@ -1,16 +1,16 @@
 import logging
 import os
+import socket
 import time
-import warnings
 from pathlib import Path
 from logging.handlers import BaseRotatingHandler
 
-import zmq
-
 from openpilot.common.logging_extra import SwagLogger, SwagFormatter, SwagLogFileFormatter
 from openpilot.common.hardware.hw import Paths
 
 
+SWAGLOG_MAX_SIZE = 4*1024*1024
+
 def get_file_handler():
   from openpilot.starpilot.connect.provider import cloudlog_root
   directory = cloudlog_root()
@@ -71,8 +71,6 @@ class UnixDomainSocketHandler(logging.Handler):
     logging.Handler.__init__(self)
     self.setFormatter(formatter)
     self.pid = None
-
-    self.zctx = None
     self.sock = None
 
   def __del__(self):
@@ -81,29 +79,24 @@ class UnixDomainSocketHandler(logging.Handler):
   def close(self):
     if self.sock is not None:
       self.sock.close()
-    if self.zctx is not None:
-      self.zctx.term()
+      self.sock = None
 
   def connect(self):
-    self.zctx = zmq.Context()
-    self.sock = self.zctx.socket(zmq.PUSH)
-    self.sock.setsockopt(zmq.LINGER, 10)
-    self.sock.connect(Paths.swaglog_ipc())
+    self.close()
+    self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
+    self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, SWAGLOG_MAX_SIZE)
+    self.sock.setblocking(False)
     self.pid = os.getpid()
 
   def emit(self, record):
     if os.getpid() != self.pid:
-      # TODO suppresses warning about forking proc with zmq socket, fix root cause
-      warnings.filterwarnings("ignore", category=ResourceWarning, message="unclosed.*<zmq.*>")
       self.connect()
 
     msg = self.format(record).rstrip('\n')
-    # print("SEND".format(repr(msg)))
     try:
-      s = chr(record.levelno)+msg
-      self.sock.send(s.encode('utf8'), zmq.NOBLOCK)
-    except zmq.error.Again:
-      # drop :/
+      self.sock.sendto((chr(record.levelno)+msg).encode('utf8'), Paths.swaglog_ipc())
+    except OSError:
+      # logmessaged is busy, not running, or the message is too big. drop :/
       pass
 
 

openpilot/common/tests/test_swaglog.cc

@@ -1,7 +1,11 @@
 #include <cstdlib>
+#include <cstring>
 #include <string>
 
-#include <zmq.h>
+#include <sys/socket.h>
+#include <sys/time.h>
+#include <sys/un.h>
+#include <unistd.h>
 
 #include "common/hardware/hw.h"
 #include "common/swaglog.h"
@@ -13,18 +17,20 @@ void test_swaglog() {
   setenv("DONGLE_ID", "test_dongle_id", 1);
   setenv("CLEAN", "1", 1);
 
-  void *context = zmq_ctx_new();
-  CHECK(context != nullptr);
-  void *socket = zmq_socket(context, ZMQ_PULL);
-  CHECK(socket != nullptr);
-  int timeout = 5000;
-  CHECK(zmq_setsockopt(socket, ZMQ_RCVTIMEO, &timeout, sizeof(timeout)) == 0);
-  CHECK(zmq_bind(socket, Path::swaglog_ipc().c_str()) == 0);
+  const std::string path = Path::swaglog_ipc();
+  unlink(path.c_str());
+  int sock = socket(AF_UNIX, SOCK_DGRAM, 0);
+  CHECK(sock >= 0);
+  struct timeval timeout = {.tv_sec = 5, .tv_usec = 0};
+  CHECK(setsockopt(sock, SOL_SOCKET, SO_RCVTIMEO, &timeout, sizeof(timeout)) == 0);
+  struct sockaddr_un addr = {.sun_family = AF_UNIX};
+  strncpy(addr.sun_path, path.c_str(), sizeof(addr.sun_path) - 1);
+  CHECK(bind(sock, (struct sockaddr*)&addr, sizeof(addr)) == 0);
 
   LOGD("native-cpp-log");
 
   char buffer[4096] = {};
-  const int size = zmq_recv(socket, buffer, sizeof(buffer), 0);
+  const int size = recv(sock, buffer, sizeof(buffer), 0);
   CHECK(size > 1);
   CHECK(buffer[0] == CLOUDLOG_DEBUG);
   std::string error;
@@ -38,8 +44,8 @@ void test_swaglog() {
   CHECK(message["ctx"]["dongle_id"].string_value() == "test_dongle_id");
   CHECK(message["ctx"]["dirty"].bool_value() == false);
 
-  CHECK(zmq_close(socket) == 0);
-  CHECK(zmq_ctx_destroy(context) == 0);
+  CHECK(close(sock) == 0);
+  CHECK(unlink(path.c_str()) == 0);
 }
 
 int main() {

openpilot/system/logmessaged.py

@@ -1,11 +1,12 @@
 #!/usr/bin/env python3
-import zmq
+import os
+import socket
 from typing import NoReturn
 
 import openpilot.cereal.messaging as messaging
 from openpilot.common.logging_extra import SwagLogFileFormatter
 from openpilot.common.hardware.hw import Paths
-from openpilot.common.swaglog import get_file_handler
+from openpilot.common.swaglog import SWAGLOG_MAX_SIZE, cloudlog, get_file_handler
 
 
 def main() -> NoReturn:
@@ -13,9 +14,19 @@ def main() -> NoReturn:
   log_handler.setFormatter(SwagLogFileFormatter(None))
   log_level = 20  # logging.INFO
 
-  ctx = zmq.Context.instance()
-  sock = ctx.socket(zmq.PULL)
-  sock.bind(Paths.swaglog_ipc())
+  path = Paths.swaglog_ipc()
+  if os.path.exists(path):
+    os.unlink(path)
+  sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
+  sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, SWAGLOG_MAX_SIZE)
+  sock.bind(path)
+  buf = bytearray(SWAGLOG_MAX_SIZE)
+
+  # senders' SO_SNDBUF caps the log size, and the kernel may clamp it (net.core.wmem_max on Linux)
+  sock.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, SWAGLOG_MAX_SIZE)
+  max_size = sock.getsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF)
+  if max_size < 256*1024:
+    cloudlog.warning(f"logs bigger than {max_size} bytes will be dropped")
 
   # and we publish them
   log_message_sock = messaging.pub_sock('logMessage')
@@ -23,9 +34,11 @@ def main() -> NoReturn:
 
   try:
     while True:
-      dat = b''.join(sock.recv_multipart())
-      level = dat[0]
-      record = dat[1:].decode("utf-8")
+      n = sock.recv_into(buf)
+      if n == 0:
+        continue
+      level = buf[0]
+      record = buf[1:n].decode("utf-8")
       if level >= log_level:
         log_handler.emit(record)
 
@@ -43,7 +56,6 @@ def main() -> NoReturn:
         error_log_message_sock.send(msg.to_bytes())
   finally:
     sock.close()
-    ctx.term()
 
     # can hit this if interrupted during a rollover
     try:

openpilot/system/tests/test_logmessaged.py

@@ -6,16 +6,14 @@ from openpilot.common.test import OpenpilotTestCase
 import openpilot.cereal.messaging as messaging
 from openpilot.system.manager.process_config import managed_processes
 from openpilot.common.hardware.hw import Paths
-from openpilot.common.swaglog import cloudlog, ipchandler
+from openpilot.common.swaglog import SWAGLOG_MAX_SIZE, cloudlog
 
 
 class TestLogmessaged(OpenpilotTestCase):
   def setup_method(self):
-    # clear the IPC buffer in case some other tests used cloudlog and filled it
-    ipchandler.close()
-    ipchandler.connect()
-
     managed_processes['logmessaged'].start()
+    while not os.path.exists(Paths.swaglog_ipc()):
+      time.sleep(0.01)
     self.sock = messaging.sub_sock("logMessage", timeout=1000, conflate=False)
     self.error_sock = messaging.sub_sock("logMessage", timeout=1000, conflate=False)
 
@@ -43,13 +41,22 @@ class TestLogmessaged(OpenpilotTestCase):
 
   def test_big_log(self):
     n = 10
-    msg = "a"*3*1024*1024
+    msg = "a"*300*1024
     for _ in range(n):
       cloudlog.info(msg)
+      time.sleep(0.05)
     time.sleep(0.5)
 
     msgs = messaging.drain_sock(self.sock)
-    assert len(msgs) == 0
+    assert len(msgs) == n
 
     logsize = sum([os.path.getsize(f) for f in self._get_log_files()])
     assert (n*len(msg)) < logsize < (n*(len(msg)+1024))
+
+  def test_too_big_log(self):
+    cloudlog.info("a"*SWAGLOG_MAX_SIZE)
+    cloudlog.info("small")
+    time.sleep(0.5)
+
+    msgs = messaging.drain_sock(self.sock)
+    assert len(msgs) == 1