Skip to content

Commit a02f4db

Browse files
adeebshihadehadeebshihadeh
committed
msgq: configurable ringbuffer size
1 parent deb807e commit a02f4db

10 files changed

Lines changed: 32 additions & 31 deletions

File tree

‎msgq/__init__.py‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -26,16 +26,16 @@ def fake_event_handle(endpoint: str, identifier: Optional[str] = None, override:
2626

2727
return handle
2828

29-
def pub_sock(endpoint: str) -> PubSocket:
29+
def pub_sock(endpoint: str, segment_size: int = 0) -> PubSocket:
3030
sock = PubSocket()
31-
sock.connect(context, endpoint)
31+
sock.connect(context, endpoint, segment_size)
3232
return sock
3333

3434

3535
def sub_sock(endpoint: str, poller: Optional[Poller] = None, addr: str = "127.0.0.1",
36-
conflate: bool = False, timeout: Optional[int] = None) -> SubSocket:
36+
conflate: bool = False, timeout: Optional[int] = None, segment_size: int = 0) -> SubSocket:
3737
sock = SubSocket()
38-
sock.connect(context, endpoint, addr.encode('utf8'), conflate)
38+
sock.connect(context, endpoint, addr.encode('utf8'), conflate, segment_size)
3939

4040
if timeout is not None:
4141
sock.setTimeout(timeout)

‎msgq/impl_fake.h‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -31,7 +31,7 @@ class FakeSubSocket: public TSubSocket {
3131
}
3232
}
3333

34-
int connect(Context *context, std::string endpoint, std::string address, bool conflate=false, bool check_endpoint=true) override {
34+
int connect(Context *context, std::string endpoint, std::string address, bool conflate=false, bool check_endpoint=true, size_t segment_size=0) override {
3535
const char* cereal_prefix = std::getenv("CEREAL_FAKE_PREFIX");
3636

3737
char* mem;
@@ -42,7 +42,7 @@ class FakeSubSocket: public TSubSocket {
4242
this->recv_called = new Event(state->fds[EventPurpose::RECV_CALLED]);
4343
this->recv_ready = new Event(state->fds[EventPurpose::RECV_READY]);
4444

45-
return TSubSocket::connect(context, endpoint, address, conflate, check_endpoint);
45+
return TSubSocket::connect(context, endpoint, address, conflate, check_endpoint, segment_size);
4646
}
4747

4848
Message *receive(bool non_blocking=false) override {

‎msgq/impl_msgq.cc‎

Lines changed: 6 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -41,12 +41,13 @@ MSGQMessage::~MSGQMessage() {
4141
this->close();
4242
}
4343

44-
int MSGQSubSocket::connect(Context *context, std::string endpoint, std::string address, bool conflate, bool check_endpoint){
44+
int MSGQSubSocket::connect(Context *context, std::string endpoint, std::string address, bool conflate, bool check_endpoint, size_t segment_size){
4545
assert(context);
4646
assert(address == "127.0.0.1");
4747

4848
q = new msgq_queue_t;
49-
int r = msgq_new_queue(q, endpoint.c_str(), DEFAULT_SEGMENT_SIZE);
49+
size_t size = segment_size > 0 ? segment_size : DEFAULT_SEGMENT_SIZE;
50+
int r = msgq_new_queue(q, endpoint.c_str(), size);
5051
if (r != 0){
5152
return r;
5253
}
@@ -109,7 +110,7 @@ MSGQSubSocket::~MSGQSubSocket(){
109110
}
110111
}
111112

112-
int MSGQPubSocket::connect(Context *context, std::string endpoint, bool check_endpoint){
113+
int MSGQPubSocket::connect(Context *context, std::string endpoint, bool check_endpoint, size_t segment_size){
113114
assert(context);
114115

115116
// TODO
@@ -118,7 +119,8 @@ int MSGQPubSocket::connect(Context *context, std::string endpoint, bool check_en
118119
//}
119120

120121
q = new msgq_queue_t;
121-
int r = msgq_new_queue(q, endpoint.c_str(), DEFAULT_SEGMENT_SIZE);
122+
size_t size = segment_size > 0 ? segment_size : DEFAULT_SEGMENT_SIZE;
123+
int r = msgq_new_queue(q, endpoint.c_str(), size);
122124
if (r != 0){
123125
return r;
124126
}

‎msgq/impl_msgq.h‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -36,7 +36,7 @@ class MSGQSubSocket : public SubSocket {
3636
msgq_queue_t * q = NULL;
3737
int timeout;
3838
public:
39-
int connect(Context *context, std::string endpoint, std::string address, bool conflate=false, bool check_endpoint=true);
39+
int connect(Context *context, std::string endpoint, std::string address, bool conflate=false, bool check_endpoint=true, size_t segment_size=0);
4040
void setTimeout(int timeout);
4141
void * getRawSocket() {return (void*)q;}
4242
Message *receive(bool non_blocking=false);
@@ -47,7 +47,7 @@ class MSGQPubSocket : public PubSocket {
4747
private:
4848
msgq_queue_t * q = NULL;
4949
public:
50-
int connect(Context *context, std::string endpoint, bool check_endpoint=true);
50+
int connect(Context *context, std::string endpoint, bool check_endpoint=true, size_t segment_size=0);
5151
int sendMessage(Message *message);
5252
int send(char *data, size_t size);
5353
bool all_readers_updated();

‎msgq/impl_zmq.cc‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -59,7 +59,7 @@ ZMQMessage::~ZMQMessage() {
5959
}
6060

6161

62-
int ZMQSubSocket::connect(Context *context, std::string endpoint, std::string address, bool conflate, bool check_endpoint){
62+
int ZMQSubSocket::connect(Context *context, std::string endpoint, std::string address, bool conflate, bool check_endpoint, size_t segment_size){
6363
sock = zmq_socket(context->getRawContext(), ZMQ_SUB);
6464
if (sock == NULL){
6565
return -1;
@@ -113,7 +113,7 @@ ZMQSubSocket::~ZMQSubSocket(){
113113
zmq_close(sock);
114114
}
115115

116-
int ZMQPubSocket::connect(Context *context, std::string endpoint, bool check_endpoint){
116+
int ZMQPubSocket::connect(Context *context, std::string endpoint, bool check_endpoint, size_t segment_size){
117117
sock = zmq_socket(context->getRawContext(), ZMQ_PUB);
118118
if (sock == NULL){
119119
return -1;

‎msgq/impl_zmq.h‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -35,7 +35,7 @@ class ZMQSubSocket : public SubSocket {
3535
void * sock;
3636
std::string full_endpoint;
3737
public:
38-
int connect(Context *context, std::string endpoint, std::string address, bool conflate=false, bool check_endpoint=true);
38+
int connect(Context *context, std::string endpoint, std::string address, bool conflate=false, bool check_endpoint=true, size_t segment_size=0);
3939
void setTimeout(int timeout);
4040
void * getRawSocket() {return sock;}
4141
Message *receive(bool non_blocking=false);
@@ -48,7 +48,7 @@ class ZMQPubSocket : public PubSocket {
4848
std::string full_endpoint;
4949
int pid = -1;
5050
public:
51-
int connect(Context *context, std::string endpoint, bool check_endpoint=true);
51+
int connect(Context *context, std::string endpoint, bool check_endpoint=true, size_t segment_size=0);
5252
int sendMessage(Message *message);
5353
int send(char *data, size_t size);
5454
bool all_readers_updated();

‎msgq/ipc.cc‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -58,9 +58,9 @@ SubSocket * SubSocket::create(){
5858
return s;
5959
}
6060

61-
SubSocket * SubSocket::create(Context * context, std::string endpoint, std::string address, bool conflate, bool check_endpoint){
61+
SubSocket * SubSocket::create(Context * context, std::string endpoint, std::string address, bool conflate, bool check_endpoint, size_t segment_size){
6262
SubSocket *s = SubSocket::create();
63-
int r = s->connect(context, endpoint, address, conflate, check_endpoint);
63+
int r = s->connect(context, endpoint, address, conflate, check_endpoint, segment_size);
6464

6565
if (r == 0) {
6666
return s;
@@ -83,9 +83,9 @@ PubSocket * PubSocket::create(){
8383
return s;
8484
}
8585

86-
PubSocket * PubSocket::create(Context * context, std::string endpoint, bool check_endpoint){
86+
PubSocket * PubSocket::create(Context * context, std::string endpoint, bool check_endpoint, size_t segment_size){
8787
PubSocket *s = PubSocket::create();
88-
int r = s->connect(context, endpoint, check_endpoint);
88+
int r = s->connect(context, endpoint, check_endpoint, segment_size);
8989

9090
if (r == 0) {
9191
return s;

‎msgq/ipc.h‎

Lines changed: 4 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -37,24 +37,23 @@ class Message {
3737

3838
class SubSocket {
3939
public:
40-
virtual int connect(Context *context, std::string endpoint, std::string address, bool conflate=false, bool check_endpoint=true) = 0;
40+
virtual int connect(Context *context, std::string endpoint, std::string address, bool conflate=false, bool check_endpoint=true, size_t segment_size=0) = 0;
4141
virtual void setTimeout(int timeout) = 0;
4242
virtual Message *receive(bool non_blocking=false) = 0;
4343
virtual void * getRawSocket() = 0;
4444
static SubSocket * create();
45-
static SubSocket * create(Context * context, std::string endpoint, std::string address="127.0.0.1", bool conflate=false, bool check_endpoint=true);
45+
static SubSocket * create(Context * context, std::string endpoint, std::string address="127.0.0.1", bool conflate=false, bool check_endpoint=true, size_t segment_size=0);
4646
virtual ~SubSocket(){}
4747
};
4848

4949
class PubSocket {
5050
public:
51-
virtual int connect(Context *context, std::string endpoint, bool check_endpoint=true) = 0;
51+
virtual int connect(Context *context, std::string endpoint, bool check_endpoint=true, size_t segment_size=0) = 0;
5252
virtual int sendMessage(Message *message) = 0;
5353
virtual int send(char *data, size_t size) = 0;
5454
virtual bool all_readers_updated() = 0;
5555
static PubSocket * create();
56-
static PubSocket * create(Context * context, std::string endpoint, bool check_endpoint=true);
57-
static PubSocket * create(Context * context, std::string endpoint, int port, bool check_endpoint=true);
56+
static PubSocket * create(Context * context, std::string endpoint, bool check_endpoint=true, size_t segment_size=0);
5857
virtual ~PubSocket(){}
5958
};
6059

‎msgq/ipc.pxd‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -49,14 +49,14 @@ cdef extern from "msgq/ipc.h":
4949
cdef cppclass SubSocket:
5050
@staticmethod
5151
SubSocket * create() nogil
52-
int connect(Context *, string, string, bool) nogil
52+
int connect(Context *, string, string, bool, bool, size_t) nogil
5353
Message * receive(bool) nogil
5454
void setTimeout(int) nogil
5555

5656
cdef cppclass PubSocket:
5757
@staticmethod
5858
PubSocket * create()
59-
int connect(Context *, string)
59+
int connect(Context *, string, bool, size_t)
6060
int sendMessage(Message *)
6161
int send(char *, size_t)
6262
bool all_readers_updated()

‎msgq/ipc_pyx.pyx‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -186,10 +186,10 @@ cdef class SubSocket:
186186
self.is_owner = False
187187
self.socket = ptr
188188

189-
def connect(self, Context context, string endpoint, string address=b"127.0.0.1", bool conflate=False):
189+
def connect(self, Context context, string endpoint, string address=b"127.0.0.1", bool conflate=False, size_t segment_size=0):
190190
cdef int r
191191
with nogil:
192-
r = self.socket.connect(context.context, endpoint, address, conflate)
192+
r = self.socket.connect(context.context, endpoint, address, conflate, True, segment_size)
193193

194194
if r != 0:
195195
if errno.errno == errno.EADDRINUSE:
@@ -228,8 +228,8 @@ cdef class PubSocket:
228228
def __dealloc__(self):
229229
del self.socket
230230

231-
def connect(self, Context context, string endpoint):
232-
r = self.socket.connect(context.context, endpoint)
231+
def connect(self, Context context, string endpoint, size_t segment_size=0):
232+
r = self.socket.connect(context.context, endpoint, True, segment_size)
233233

234234
if r != 0:
235235
if errno.errno == errno.EADDRINUSE:

0 commit comments

Comments
 (0)