Restructure msgq (#32652)

* Update ref

* Compiles

* compiles

* Refactor rest of libs

* import all

* small fiex

* cleanup import

* Need msgq simlink too

* Add to openpilot docker too

* try repo

* Updates

* Fix lint

* fix docs

* Try blank slate

* Revert "Try blank slate"

This reverts commit f078ce0.

* Maybe scons needs this to clear cache

* fix tests

* Disable test for now

* Update SConstruct

Co-authored-by: Adeeb Shihadeh <[email protected]>

* Fix whitespace

* Write skip normal

* small fixes

* add test path

* Revert repo

* linting

* whitespace

* Bump msgq

---------

Co-authored-by: Adeeb Shihadeh <[email protected]>

.gitmodules

@@ -5,7 +5,7 @@
   path = opendbc
   url = ../../commaai/opendbc.git
 [submodule "msgq"]
-  path = msgq
+  path = msgq_repo
   url = ../../commaai/msgq.git
 [submodule "rednose_repo"]
   path = rednose_repo

Dockerfile.openpilot

@@ -20,6 +20,7 @@ COPY ./release ${OPENPILOT_PATH}/release
 COPY ./common ${OPENPILOT_PATH}/common
 COPY ./opendbc ${OPENPILOT_PATH}/opendbc
 COPY ./cereal ${OPENPILOT_PATH}/cereal
+COPY ./msgq_repo ${OPENPILOT_PATH}/msgq_repo
 COPY ./msgq ${OPENPILOT_PATH}/msgq
 COPY ./panda ${OPENPILOT_PATH}/panda
 COPY ./selfdrive ${OPENPILOT_PATH}/selfdrive

SConstruct

@@ -358,9 +358,13 @@ gpucommon = [_gpucommon]
 
 Export('common', 'gpucommon')
 
-# Build cereal and messaging
-SConscript(['msgq/SConscript'])
+# Build messaging (cereal + msgq + socketmaster + their dependencies)
+SConscript(['msgq_repo/SConscript'])
 SConscript(['cereal/SConscript'])
+Import('socketmaster', 'msgq')
+messaging = [socketmaster, msgq, 'zmq', 'capnp', 'kj',]
+Export('messaging')
+
 
 # Build other submodules
 SConscript([

cereal/SConscript

@@ -1,10 +1,10 @@
-Import('env', 'envCython', 'arch', 'common', 'messaging')
+Import('env', 'envCython', 'arch', 'common', 'msgq')
 
 import shutil
 
 cereal_dir = Dir('.')
 gen_dir = Dir('gen')
-other_dir = Dir('#msgq/messaging')
+other_dir = Dir('#msgq')
 
 # Build cereal
 schema_files = ['log.capnp', 'car.capnp', 'legacy.capnp', 'custom.capnp']
@@ -22,7 +22,7 @@ env.SharedLibrary('cereal_shared', cereal_objects)
 # Build messaging
 
 services_h = env.Command(['services.h'], ['services.py'], 'python3 ' + cereal_dir.path + '/services.py > $TARGET')
-env.Program('messaging/bridge', ['messaging/bridge.cc'], LIBS=[messaging, 'zmq', common])
+env.Program('messaging/bridge', ['messaging/bridge.cc'], LIBS=[msgq, 'zmq', common])
 
 
 socketmaster = env.SharedObject(['messaging/socketmaster.cc'])

cereal/messaging/__init__.py

@@ -1,7 +1,8 @@
 # must be built with scons
-from msgq.messaging.messaging_pyx import Context, Poller, SubSocket, PubSocket, SocketEventHandle, toggle_fake_events, \
+from msgq.ipc_pyx import Context, Poller, SubSocket, PubSocket, SocketEventHandle, toggle_fake_events, \
                                 set_fake_prefix, get_fake_prefix, delete_fake_prefix, wait_for_one_event
-from msgq.messaging.messaging_pyx import MultiplePublishersError, MessagingError
+from msgq.ipc_pyx import MultiplePublishersError, IpcError
+from msgq import fake_event_handle, pub_sock, sub_sock, drain_sock_raw, context
 
 import os
 import capnp
@@ -13,27 +14,8 @@
 from cereal import log
 from cereal.services import SERVICE_LIST
 
-assert MultiplePublishersError
-assert MessagingError
-assert toggle_fake_events
-assert set_fake_prefix
-assert get_fake_prefix
-assert delete_fake_prefix
-assert wait_for_one_event
-
 NO_TRAVERSAL_LIMIT = 2**64-1
 
-context = Context()
-
-
-def fake_event_handle(endpoint: str, identifier: Optional[str] = None, override: bool = True, enable: bool = False) -> SocketEventHandle:
-  identifier = identifier or get_fake_prefix()
-  handle = SocketEventHandle(endpoint, identifier, override)
-  if override:
-    handle.enabled = enable
-
-  return handle
-
 
 def log_from_bytes(dat: bytes) -> capnp.lib.capnp._DynamicStructReader:
   with log.Event.from_bytes(dat, traversal_limit_in_words=NO_TRAVERSAL_LIMIT) as msg:
@@ -55,42 +37,6 @@ def new_message(service: Optional[str], size: Optional[int] = None, **kwargs) ->
   return dat
 
 
-def pub_sock(endpoint: str) -> PubSocket:
-  sock = PubSocket()
-  sock.connect(context, endpoint)
-  return sock
-
-
-def sub_sock(endpoint: str, poller: Optional[Poller] = None, addr: str = "127.0.0.1",
-             conflate: bool = False, timeout: Optional[int] = None) -> SubSocket:
-  sock = SubSocket()
-  sock.connect(context, endpoint, addr.encode('utf8'), conflate)
-
-  if timeout is not None:
-    sock.setTimeout(timeout)
-
-  if poller is not None:
-    poller.registerSocket(sock)
-  return sock
-
-
-def drain_sock_raw(sock: SubSocket, wait_for_one: bool = False) -> List[bytes]:
-  """Receive all message currently available on the queue"""
-  ret: List[bytes] = []
-  while 1:
-    if wait_for_one and len(ret) == 0:
-      dat = sock.receive()
-    else:
-      dat = sock.receive(non_blocking=True)
-
-    if dat is None:
-      break
-
-    ret.append(dat)
-
-  return ret
-
-
 def drain_sock(sock: SubSocket, wait_for_one: bool = False) -> List[capnp.lib.capnp._DynamicStructReader]:
   """Receive all message currently available on the queue"""
   msgs = drain_sock_raw(sock, wait_for_one=wait_for_one)

cereal/messaging/bridge.cc

@@ -8,8 +8,8 @@
 typedef void (*sighandler_t)(int sig);
 
 #include "cereal/services.h"
-#include "msgq/messaging/impl_msgq.h"
-#include "msgq/messaging/impl_zmq.h"
+#include "msgq/impl_msgq.h"
+#include "msgq/impl_zmq.h"
 
 std::atomic<bool> do_exit = false;
 static void set_do_exit(int sig) {

cereal/messaging/messaging.h

@@ -10,7 +10,7 @@
 #include <capnp/serialize.h>
 
 #include "cereal/gen/cpp/log.capnp.h"
-#include "msgq/messaging/messaging.h"
+#include "msgq/ipc.h"
 
 #ifdef __APPLE__
 #define CLOCK_BOOTTIME CLOCK_MONOTONIC