Remove ZMQ support (#674)

* remove zmq

* less abstraction

Author: adeebshihadeh

README.md

@@ -1,7 +1,7 @@
 # MSGQ: A lock free single producer multi consumer message queue
 
 ## What is this library?
-MSGQ is a generic high performance IPC pub sub system with a single publisher and multiple subscribers. MSGQ is designed to be a high performance replacement for ZMQ-like SUB/PUB patterns. It uses a ring buffer in shared memory to efficiently read and write data. Each read requires a copy. Writing can be done without a copy, as long as the size of the data is known in advance. While MSGQ is the core of this library, this library also allows replacing the MSGQ backend with ZMQ or a spoofed implementation that can be used for deterministic testing. This library also contains visionipc, an IPC system specifically for large contiguous buffers (like images/video).
+MSGQ is a generic high performance IPC pub sub system with a single publisher and multiple subscribers. It uses a ring buffer in shared memory to efficiently read and write data. Each read requires a copy. Writing can be done without a copy, as long as the size of the data is known in advance. This library also provides a spoofed implementation that can be used for deterministic testing, and visionipc, an IPC system specifically for large contiguous buffers (like images/video).
 
 ## Storage
 The storage for the queue consists of an area of metadata, and the actual buffer. The metadata contains:

SConscript

@@ -9,13 +9,12 @@ gen_dir = Dir('gen')
 msgq_objects = env.SharedObject([
   'msgq/ipc.cc',
   'msgq/event.cc',
-  'msgq/impl_zmq.cc',
   'msgq/impl_msgq.cc',
   'msgq/impl_fake.cc',
   'msgq/msgq.cc',
 ])
 msgq = env.Library('msgq', msgq_objects)
-msgq_python = envCython.Program('msgq/ipc_pyx.so', 'msgq/ipc_pyx.pyx', LIBS=envCython["LIBS"]+[msgq, "zmq", common])
+msgq_python = envCython.Program('msgq/ipc_pyx.so', 'msgq/ipc_pyx.pyx', LIBS=envCython["LIBS"]+[msgq, common])
 
 # Build Vision IPC
 vipc_files = ['visionipc.cc', 'visionipc_server.cc', 'visionipc_client.cc']
@@ -29,7 +28,7 @@ vipc_objects = env.SharedObject(vipc_sources)
 visionipc = env.Library('visionipc', vipc_objects)
 
 
-vipc_libs = envCython["LIBS"] + [visionipc, msgq, common, "zmq"]
+vipc_libs = envCython["LIBS"] + [visionipc, msgq, common]
 envCython.Program(f'{visionipc_dir.abspath}/visionipc_pyx.so', f'{visionipc_dir.abspath}/visionipc_pyx.pyx',
                   LIBS=vipc_libs)
 

msgq/__init__.py

@@ -1,7 +1,6 @@
 # must be built with scons
 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, \
-                                context_is_zmq
+                                set_fake_prefix, get_fake_prefix, delete_fake_prefix, wait_for_one_event
 from msgq.ipc_pyx import MultiplePublishersError, IpcError
 
 from typing import Optional, List, Union
@@ -13,7 +12,6 @@
 assert get_fake_prefix
 assert delete_fake_prefix
 assert wait_for_one_event
-assert context_is_zmq
 
 NO_TRAVERSAL_LIMIT = 2**64-1
 

msgq/conftest.py

@@ -1,14 +1,6 @@
-import os
 import pytest
 import msgq
 
-@pytest.fixture(params=[False, True], ids=["msgq", "zmq"], autouse=True)
-def zmq_mode(request):
-  if request.param:
-    os.environ["ZMQ"] = "1"
-  else:
-    os.environ.pop("ZMQ", None)
+@pytest.fixture(autouse=True)
+def msgq_context():
   msgq.context = msgq.Context()
-  assert msgq.context_is_zmq() == request.param
-  yield request.param
-  os.environ.pop("ZMQ", None)

msgq/impl_msgq.cc

@@ -8,36 +8,30 @@
 
 #include "msgq/impl_msgq.h"
 
-MSGQContext::MSGQContext() {
-}
-
-MSGQContext::~MSGQContext() {
-}
-
-void MSGQMessage::init(size_t sz) {
+void Message::init(size_t sz) {
   size = sz;
   data = new char[size];
 }
 
-void MSGQMessage::init(char * d, size_t sz) {
+void Message::init(char * d, size_t sz) {
   size = sz;
   data = new char[size];
   memcpy(data, d, size);
 }
 
-void MSGQMessage::takeOwnership(char * d, size_t sz) {
+void Message::takeOwnership(char * d, size_t sz) {
   size = sz;
   data = d;
 }
 
-void MSGQMessage::close() {
+void Message::close() {
   if (size > 0){
     delete[] data;
   }
   size = 0;
 }
 
-MSGQMessage::~MSGQMessage() {
+Message::~Message() {
   this->close();
 }
 
@@ -67,7 +61,7 @@ int MSGQSubSocket::connect(Context *context, std::string endpoint, std::string a
 Message * MSGQSubSocket::receive(bool non_blocking){
   msgq_msg_t msg;
 
-  MSGQMessage *r = NULL;
+  Message *r = NULL;
 
   int rc = msgq_msg_recv(&msg, q);
 
@@ -92,11 +86,11 @@ Message * MSGQSubSocket::receive(bool non_blocking){
   }
 
   if (rc > 0){
-    r = new MSGQMessage;
+    r = new Message;
     r->takeOwnership(msg.data, msg.size);
   }
 
-  return (Message*)r;
+  return r;
 }
 
 void MSGQSubSocket::setTimeout(int t){
@@ -110,14 +104,9 @@ MSGQSubSocket::~MSGQSubSocket(){
   }
 }
 
-int MSGQPubSocket::connect(Context *context, std::string endpoint, bool check_endpoint, size_t segment_size){
+int PubSocket::connect(Context *context, std::string endpoint, bool check_endpoint, size_t segment_size){
   assert(context);
 
-  // TODO
-  //if (check_endpoint && !service_exists(std::string(endpoint))){
-  //  std::cout << "Warning, " << std::string(endpoint) << " is not in service list." << std::endl;
-  //}
-
   q = new msgq_queue_t;
   size_t size = segment_size > 0 ? segment_size : DEFAULT_SEGMENT_SIZE;
   int r = msgq_new_queue(q, endpoint.c_str(), size);
@@ -130,27 +119,27 @@ int MSGQPubSocket::connect(Context *context, std::string endpoint, bool check_en
   return 0;
 }
 
-int MSGQPubSocket::sendMessage(Message *message){
+int PubSocket::sendMessage(Message *message){
   msgq_msg_t msg;
   msg.data = message->getData();
   msg.size = message->getSize();
 
   return msgq_msg_send(&msg, q);
 }
 
-int MSGQPubSocket::send(char *data, size_t size){
+int PubSocket::send(char *data, size_t size){
   msgq_msg_t msg;
   msg.data = data;
   msg.size = size;
 
   return msgq_msg_send(&msg, q);
 }
 
-bool MSGQPubSocket::all_readers_updated() {
+bool PubSocket::all_readers_updated() {
   return msgq_all_readers_updated(q);
 }
 
-MSGQPubSocket::~MSGQPubSocket(){
+PubSocket::~PubSocket(){
   if (q != NULL){
     msgq_close_queue(q);
     delete q;
@@ -160,7 +149,7 @@ MSGQPubSocket::~MSGQPubSocket(){
 
 void MSGQPoller::registerSocket(SubSocket * socket){
   assert(num_polls + 1 < MAX_POLLERS);
-  polls[num_polls].q = (msgq_queue_t*)socket->getRawSocket();
+  polls[num_polls].q = static_cast<MSGQSubSocket*>(socket)->getQueue();
 
   sockets.push_back(socket);
   num_polls++;

msgq/impl_msgq.h

@@ -8,52 +8,18 @@
 
 #define MAX_POLLERS 128
 
-class MSGQContext : public Context {
-private:
-  void * context = NULL;
-public:
-  MSGQContext();
-  void * getRawContext() {return context;}
-  ~MSGQContext();
-};
-
-class MSGQMessage : public Message {
-private:
-  char * data;
-  size_t size;
-public:
-  void init(size_t size);
-  void init(char *data, size_t size);
-  void takeOwnership(char *data, size_t size);
-  size_t getSize(){return size;}
-  char * getData(){return data;}
-  void close();
-  ~MSGQMessage();
-};
-
 class MSGQSubSocket : public SubSocket {
 private:
   msgq_queue_t * q = NULL;
   int timeout;
 public:
   int connect(Context *context, std::string endpoint, std::string address, bool conflate=false, bool check_endpoint=true, size_t segment_size=0);
   void setTimeout(int timeout);
-  void * getRawSocket() {return (void*)q;}
+  msgq_queue_t * getQueue() {return q;}
   Message *receive(bool non_blocking=false);
   ~MSGQSubSocket();
 };
 
-class MSGQPubSocket : public PubSocket {
-private:
-  msgq_queue_t * q = NULL;
-public:
-  int connect(Context *context, std::string endpoint, bool check_endpoint=true, size_t segment_size=0);
-  int sendMessage(Message *message);
-  int send(char *data, size_t size);
-  bool all_readers_updated();
-  ~MSGQPubSocket();
-};
-
 class MSGQPoller : public Poller {
 private:
   std::vector<SubSocket*> sockets;