Skip to content

Commit

Permalink
Problem: test_large_msg kills my system temporarily
Browse files Browse the repository at this point in the history 
And I'm on a reasonably sized laptop. I think allocating INT_MAX
memory is dangerous in a test case.

Solution: expose this as a context option. I've used ZMQ_MAX_MSGSZ
and documented it and implemented the API. However I don't know how
to get the parent context for a socket, so the code in zmq.cpp is
still unfinished.
  • Loading branch information
@hintjens
hintjens committed Feb 9, 2016
1 parent 7470c00 commit 62c66ae
Show file tree
Hide file tree
Showing 8 changed files with 121 additions and 79 deletions.
10 changes: 10 additions & 0 deletions doc/zmq_ctx_get.txt
View file
Expand Up @@ -26,20 +26,30 @@ ZMQ_IO_THREADS: Get number of I/O threads
The 'ZMQ_IO_THREADS' argument returns the size of the 0MQ thread pool
for this context.


ZMQ_MAX_SOCKETS: Get maximum number of sockets
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
The 'ZMQ_MAX_SOCKETS' argument returns the maximum number of sockets
allowed for this context.


ZMQ_MAX_MSGSZ: Get maximum message size
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
The 'ZMQ_MAX_MSGSZ' argument returns the maximum size of a message
allowed for this context. Default value is INT_MAX.


ZMQ_SOCKET_LIMIT: Get largest configurable number of sockets
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
The 'ZMQ_SOCKET_LIMIT' argument returns the largest number of sockets that
linkzmq:zmq_ctx_set[3] will accept.


ZMQ_IPV6: Set IPv6 option
~~~~~~~~~~~~~~~~~~~~~~~~~
The 'ZMQ_IPV6' argument returns the IPv6 option for the context.


ZMQ_BLOCKY: Get blocky setting
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
The 'ZMQ_BLOCKY' argument returns 1 if the context will block on terminate,
Expand Down
16 changes: 16 additions & 0 deletions doc/zmq_ctx_set.txt
View file
Expand Up @@ -47,6 +47,7 @@ context.
[horizontal]
Default value:: 1


ZMQ_THREAD_SCHED_POLICY: Set scheduling policy for I/O threads
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
The 'ZMQ_THREAD_SCHED_POLICY' argument sets the scheduling policy for
Expand All @@ -58,6 +59,7 @@ This option only applies before creating any sockets on the context.
[horizontal]
Default value:: -1


ZMQ_THREAD_PRIORITY: Set scheduling priority for I/O threads
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
The 'ZMQ_THREAD_PRIORITY' argument sets scheduling priority for
Expand All @@ -69,6 +71,19 @@ This option only applies before creating any sockets on the context.
[horizontal]
Default value:: -1


ZMQ_MAX_MSGSZ: Set maximum message size
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
The 'ZMQ_MAX_MSGSZ' argument sets the maximum allowed size
of a message sent in the context. You can query the maximal
allowed value with linkzmq:zmq_ctx_get[3] using the
'ZMQ_MAX_MSGSZ' option.

[horizontal]
Default value:: INT_MAX
Maximum value:: INT_MAX


ZMQ_MAX_SOCKETS: Set maximum number of sockets
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
The 'ZMQ_MAX_SOCKETS' argument sets the maximum number of sockets allowed
Expand All @@ -78,6 +93,7 @@ linkzmq:zmq_ctx_get[3] using the 'ZMQ_SOCKET_LIMIT' option.
[horizontal]
Default value:: 1024


ZMQ_IPV6: Set IPv6 option
~~~~~~~~~~~~~~~~~~~~~~~~~
The 'ZMQ_IPV6' argument sets the IPv6 value for all sockets created in
Expand Down
2 changes: 1 addition & 1 deletion include/zmq.h
View file
Expand Up @@ -185,13 +185,13 @@ ZMQ_EXPORT void zmq_version (int *major, int *minor, int *patch);
/* 0MQ infrastructure (a.k.a. context) initialisation & termination. */
/******************************************************************************/

/* New API */
/* Context options */
#define ZMQ_IO_THREADS 1
#define ZMQ_MAX_SOCKETS 2
#define ZMQ_SOCKET_LIMIT 3
#define ZMQ_THREAD_PRIORITY 3
#define ZMQ_THREAD_SCHED_POLICY 4
#define ZMQ_MAX_MSGSZ 5

/* Default for new contexts */
#define ZMQ_IO_THREADS_DFLT 1
Expand Down
15 changes: 13 additions & 2 deletions src/ctx.cpp
View file
Expand Up @@ -36,6 +36,7 @@
#endif

#include <limits>
#include <climits>
#include <new>
#include <string.h>

Expand Down Expand Up @@ -79,6 +80,7 @@ zmq::ctx_t::ctx_t () :
slot_count (0),
slots (NULL),
max_sockets (clipped_maxsocket (ZMQ_MAX_SOCKETS_DFLT)),
max_msgsz (INT_MAX),
io_thread_count (ZMQ_IO_THREADS_DFLT),
blocky (true),
ipv6 (false),
Expand Down Expand Up @@ -251,20 +253,26 @@ int zmq::ctx_t::set (int option_, int optval_)
if (option_ == ZMQ_THREAD_PRIORITY && optval_ >= 0) {
opt_sync.lock();
thread_priority = optval_;
opt_sync.unlock();
opt_sync.unlock ();
}
else
if (option_ == ZMQ_THREAD_SCHED_POLICY && optval_ >= 0) {
opt_sync.lock();
thread_sched_policy = optval_;
opt_sync.unlock();
opt_sync.unlock ();
}
else
if (option_ == ZMQ_BLOCKY && optval_ >= 0) {
opt_sync.lock ();
blocky = (optval_ != 0);
opt_sync.unlock ();
}
else
if (option_ == ZMQ_MAX_MSGSZ && optval_ >= 0) {
opt_sync.lock ();
max_msgsz = optval_ < INT_MAX? optval_: INT_MAX;
opt_sync.unlock ();
}
else {
errno = EINVAL;
rc = -1;
Expand All @@ -289,6 +297,9 @@ int zmq::ctx_t::get (int option_)
else
if (option_ == ZMQ_BLOCKY)
rc = blocky;
else
if (option_ == ZMQ_MAX_MSGSZ)
rc = max_msgsz;
else {
errno = EINVAL;
rc = -1;
Expand Down
3 changes: 3 additions & 0 deletions src/ctx.hpp
View file
Expand Up @@ -199,6 +199,9 @@ namespace zmq
// Maximum number of sockets that can be opened at the same time.
int max_sockets;

// Maximum allowed message size
int max_msgsz;

// Number of I/O threads to launch.
int io_thread_count;

Expand Down
2 changes: 1 addition & 1 deletion src/socket_base.hpp
View file
Expand Up @@ -245,7 +245,7 @@ namespace zmq
void update_pipe_options(int option_);

// Socket's mailbox object.
i_mailbox* mailbox;
i_mailbox *mailbox;

// List of attached pipes.
typedef array_t <pipe_t, 3> pipes_t;
Expand Down
41 changes: 22 additions & 19 deletions src/zmq.cpp
View file
Expand Up @@ -164,12 +164,12 @@ void *zmq_ctx_new (void)

int zmq_ctx_term (void *ctx_)
{
if (!ctx_ || !((zmq::ctx_t*) ctx_)->check_tag ()) {
if (!ctx_ || !((zmq::ctx_t *) ctx_)->check_tag ()) {
errno = EFAULT;
return -1;
}

int rc = ((zmq::ctx_t*) ctx_)->terminate ();
int rc = ((zmq::ctx_t *) ctx_)->terminate ();
int en = errno;

// Shut down only if termination was not interrupted by a signal.
Expand All @@ -193,30 +193,29 @@ int zmq_ctx_term (void *ctx_)

int zmq_ctx_shutdown (void *ctx_)
{
if (!ctx_ || !((zmq::ctx_t*) ctx_)->check_tag ()) {
if (!ctx_ || !((zmq::ctx_t *) ctx_)->check_tag ()) {
errno = EFAULT;
return -1;
}

return ((zmq::ctx_t*) ctx_)->shutdown ();
return ((zmq::ctx_t *) ctx_)->shutdown ();
}

int zmq_ctx_set (void *ctx_, int option_, int optval_)
{
if (!ctx_ || !((zmq::ctx_t*) ctx_)->check_tag ()) {
if (!ctx_ || !((zmq::ctx_t *) ctx_)->check_tag ()) {
errno = EFAULT;
return -1;
}
return ((zmq::ctx_t*) ctx_)->set (option_, optval_);
return ((zmq::ctx_t *) ctx_)->set (option_, optval_);
}

int zmq_ctx_get (void *ctx_, int option_)
{
if (!ctx_ || !((zmq::ctx_t*) ctx_)->check_tag ()) {
if (!ctx_ || !((zmq::ctx_t *) ctx_)->check_tag ()) {
errno = EFAULT;
return -1;
}
return ((zmq::ctx_t*) ctx_)->get (option_);
return ((zmq::ctx_t *) ctx_)->get (option_);
}

// Stable/legacy context API
Expand Down Expand Up @@ -247,11 +246,11 @@ int zmq_ctx_destroy (void *ctx_)

void *zmq_socket (void *ctx_, int type_)
{
if (!ctx_ || !((zmq::ctx_t*) ctx_)->check_tag ()) {
if (!ctx_ || !((zmq::ctx_t *) ctx_)->check_tag ()) {
errno = EFAULT;
return NULL;
}
zmq::ctx_t *ctx = (zmq::ctx_t*) ctx_;
zmq::ctx_t *ctx = (zmq::ctx_t *) ctx_;
zmq::socket_base_t *s = ctx->create_socket (type_);
return (void *) s;
}
Expand Down Expand Up @@ -366,15 +365,20 @@ int zmq_disconnect (void *s_, const char *addr_)

// Sending functions.

static int
static inline int
s_sendmsg (zmq::socket_base_t *s_, zmq_msg_t *msg_, int flags_)
{
size_t sz = zmq_msg_size (msg_);
int rc = s_->send ((zmq::msg_t *) msg_, flags_);
if (unlikely (rc < 0))
return -1;
// truncate returned size to INT_MAX to avoid overflow to negative values
return (int) (sz < INT_MAX ? sz : INT_MAX);

// This is what I'd like to do, my C++ fu is too weak -- PH 2016/02/09
// int max_msgsz = s_->parent->get (ZMQ_MAX_MSGSZ);
size_t max_msgsz = INT_MAX;

// Truncate returned size to INT_MAX to avoid overflow to negative values
return (int) (sz < max_msgsz? sz: max_msgsz);
}

/* To be deprecated once zmq_msg_send() is stable */
Expand Down Expand Up @@ -407,7 +411,6 @@ int zmq_send (void *s_, const void *buf_, size_t len_, int flags_)
errno = err;
return -1;
}

// Note the optimisation here. We don't close the msg object as it is
// empty anyway. This may change when implementation of zmq_msg_t changes.
return rc;
Expand All @@ -433,7 +436,6 @@ int zmq_send_const (void *s_, const void *buf_, size_t len_, int flags_)
errno = err;
return -1;
}

// Note the optimisation here. We don't close the msg object as it is
// empty anyway. This may change when implementation of zmq_msg_t changes.
return rc;
Expand Down Expand Up @@ -484,12 +486,13 @@ int zmq_sendiov (void *s_, iovec *a_, size_t count_, int flags_)
static int
s_recvmsg (zmq::socket_base_t *s_, zmq_msg_t *msg_, int flags_)
{
int rc = s_->recv ((zmq::msg_t*) msg_, flags_);
int rc = s_->recv ((zmq::msg_t *) msg_, flags_);
if (unlikely (rc < 0))
return -1;
// truncate returned size to INT_MAX to avoid overflow to negative values

// Truncate returned size to INT_MAX to avoid overflow to negative values
size_t sz = zmq_msg_size (msg_);
return (int) (sz < INT_MAX ? sz : INT_MAX);
return (int) (sz < INT_MAX? sz: INT_MAX);
}

/* To be deprecated once zmq_msg_recv() is stable */
Expand Down

0 comments on commit 62c66ae

Please sign in to comment.