diff --git a/SConstruct b/SConstruct index c2719fc32e..957632f597 100644 --- a/SConstruct +++ b/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) diff --git a/openpilot/cereal/SConscript b/openpilot/cereal/SConscript index bee7620b1e..c4d1e5a4c0 100644 --- a/openpilot/cereal/SConscript +++ b/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']) diff --git a/openpilot/common/SConscript b/openpilot/common/SConscript index 847d16446a..26b9972939 100644 --- a/openpilot/common/SConscript +++ b/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']) diff --git a/openpilot/common/hardware/hw.h b/openpilot/common/hardware/hw.h index 83dc452da8..5c94c52fe1 100644 --- a/openpilot/common/hardware/hw.h +++ b/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() { diff --git a/openpilot/common/hardware/hw.py b/openpilot/common/hardware/hw.py index 1041a17c1c..c5d9440731 100644 --- a/openpilot/common/hardware/hw.py +++ b/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: diff --git a/openpilot/common/swaglog.cc b/openpilot/common/swaglog.cc index 74c617fa61..347ba99831 100644 --- a/openpilot/common/swaglog.cc +++ b/openpilot/common/swaglog.cc @@ -9,8 +9,10 @@ #include #include -#include #include +#include +#include +#include #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; }; diff --git a/openpilot/common/swaglog.h b/openpilot/common/swaglog.h index 06d45b1d98..74d13b96e6 100644 --- a/openpilot/common/swaglog.h +++ b/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))) diff --git a/openpilot/common/swaglog.py b/openpilot/common/swaglog.py index 5af8fd3e69..fb8038eb06 100644 --- a/openpilot/common/swaglog.py +++ 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.*") 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 diff --git a/openpilot/common/tests/test_swaglog.cc b/openpilot/common/tests/test_swaglog.cc index 07e2b16bc1..283bc2ff11 100644 --- a/openpilot/common/tests/test_swaglog.cc +++ b/openpilot/common/tests/test_swaglog.cc @@ -1,7 +1,11 @@ #include +#include #include -#include +#include +#include +#include +#include #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() { diff --git a/openpilot/system/logmessaged.py b/openpilot/system/logmessaged.py index 937a0741d0..1ae81e5166 100755 --- a/openpilot/system/logmessaged.py +++ b/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: diff --git a/openpilot/system/tests/test_logmessaged.py b/openpilot/system/tests/test_logmessaged.py index 247bfd8a42..0438314e32 100644 --- a/openpilot/system/tests/test_logmessaged.py +++ b/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