msgq: configurable ringbuffer size

msgq/__init__.py

@@ -26,8 +26,8 @@ def fake_event_handle(endpoint: str, identifier: Optional[str] = None, override:
 
   return handle
 
-def pub_sock(endpoint: str) -> PubSocket:
-  sock = PubSocket()
+def pub_sock(endpoint: str, segment_size: int = 0) -> PubSocket:
+  sock = PubSocket(segment_size)
   sock.connect(context, endpoint)
   return sock
 

msgq/impl_msgq.cc

@@ -46,7 +46,7 @@ int MSGQSubSocket::connect(Context *context, std::string endpoint, std::string a
   assert(address == "127.0.0.1");
 
   q = new msgq_queue_t;
-  int r = msgq_new_queue(q, endpoint.c_str(), DEFAULT_SEGMENT_SIZE);
+  int r = msgq_new_queue(q, endpoint.c_str(), segment_size);
   if (r != 0){
     return r;
   }
@@ -118,7 +118,7 @@ int MSGQPubSocket::connect(Context *context, std::string endpoint, bool check_en
   //}
 
   q = new msgq_queue_t;
-  int r = msgq_new_queue(q, endpoint.c_str(), DEFAULT_SEGMENT_SIZE);
+  int r = msgq_new_queue(q, endpoint.c_str(), segment_size);
   if (r != 0){
     return r;
   }

msgq/impl_msgq.h

@@ -35,9 +35,11 @@ class MSGQSubSocket : public SubSocket {
 private:
   msgq_queue_t * q = NULL;
   int timeout;
+  size_t segment_size = DEFAULT_SEGMENT_SIZE;
 public:
   int connect(Context *context, std::string endpoint, std::string address, bool conflate=false, bool check_endpoint=true);
   void setTimeout(int timeout);
+  void setSegmentSize(size_t size) { segment_size = size; }
   void * getRawSocket() {return (void*)q;}
   Message *receive(bool non_blocking=false);
   ~MSGQSubSocket();
@@ -46,7 +48,9 @@ class MSGQSubSocket : public SubSocket {
 class MSGQPubSocket : public PubSocket {
 private:
   msgq_queue_t * q = NULL;
+  size_t segment_size;
 public:
+  MSGQPubSocket(size_t size = 0) : segment_size(size ? size : DEFAULT_SEGMENT_SIZE) {}
   int connect(Context *context, std::string endpoint, bool check_endpoint=true);
   int sendMessage(Message *message);
   int send(char *data, size_t size);

msgq/impl_zmq.h

@@ -48,6 +48,7 @@ class ZMQPubSocket : public PubSocket {
   std::string full_endpoint;
   int pid = -1;
 public:
+  ZMQPubSocket(size_t size = 0) { (void)size; }  // size ignored for ZMQ
   int connect(Context *context, std::string endpoint, bool check_endpoint=true);
   int sendMessage(Message *message);
   int send(char *data, size_t size);

msgq/ipc.cc

@@ -72,19 +72,19 @@ SubSocket * SubSocket::create(Context * context, std::string endpoint, std::stri
   }
 }
 
-PubSocket * PubSocket::create(){
+PubSocket * PubSocket::create(size_t segment_size){
   PubSocket * s;
   if (messaging_use_zmq()){
-    s = new ZMQPubSocket();
+    s = new ZMQPubSocket(segment_size);
   } else {
-    s = new MSGQPubSocket();
+    s = new MSGQPubSocket(segment_size);
   }
 
   return s;
 }
 
-PubSocket * PubSocket::create(Context * context, std::string endpoint, bool check_endpoint){
-  PubSocket *s = PubSocket::create();
+PubSocket * PubSocket::create(Context * context, std::string endpoint, bool check_endpoint, size_t segment_size){
+  PubSocket *s = PubSocket::create(segment_size);
   int r = s->connect(context, endpoint, check_endpoint);
 
   if (r == 0) {

msgq/ipc.h

@@ -52,9 +52,8 @@ class PubSocket {
   virtual int sendMessage(Message *message) = 0;
   virtual int send(char *data, size_t size) = 0;
   virtual bool all_readers_updated() = 0;
-  static PubSocket * create();
-  static PubSocket * create(Context * context, std::string endpoint, bool check_endpoint=true);
-  static PubSocket * create(Context * context, std::string endpoint, int port, bool check_endpoint=true);
+  static PubSocket * create(size_t segment_size = 0);
+  static PubSocket * create(Context * context, std::string endpoint, bool check_endpoint=true, size_t segment_size=0);
   virtual ~PubSocket(){}
 };
 

msgq/ipc.pxd

@@ -55,7 +55,7 @@ cdef extern from "msgq/ipc.h":
 
   cdef cppclass PubSocket:
     @staticmethod
-    PubSocket * create()
+    PubSocket * create(size_t)
     int connect(Context *, string)
     int sendMessage(Message *)
     int send(char *, size_t)

msgq/ipc_pyx.pyx

@@ -220,8 +220,8 @@ cdef class SubSocket:
 cdef class PubSocket:
   cdef cppPubSocket * socket
 
-  def __cinit__(self):
-    self.socket = cppPubSocket.create()
+  def __cinit__(self, size_t segment_size=0):
+    self.socket = cppPubSocket.create(segment_size)
     if self.socket == NULL:
       raise IpcError