diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/config.hpp | 3 | ||||
| -rw-r--r-- | src/ctx.cpp | 210 | ||||
| -rw-r--r-- | src/ctx.hpp | 16 | ||||
| -rw-r--r-- | src/xs.cpp | 11 |
4 files changed, 149 insertions, 91 deletions
diff --git a/src/config.hpp b/src/config.hpp index e4d948e..8107e17 100644 --- a/src/config.hpp +++ b/src/config.hpp @@ -29,9 +29,6 @@ namespace xs enum { - // Maximum number of sockets that can be opened at the same time. - max_sockets = 512, - // Number of new messages in message pipe needed to trigger new memory // allocation. Setting this parameter to 256 decreases the impact of // memory allocation by approximately 99.6% diff --git a/src/ctx.cpp b/src/ctx.cpp index 97a3e62..b6298e2 100644 --- a/src/ctx.cpp +++ b/src/ctx.cpp @@ -40,59 +40,11 @@ xs::ctx_t::ctx_t (uint32_t io_threads_) : tag (0xbadcafe0), - terminating (false) + starting (true), + terminating (false), + max_sockets (512), + io_thread_count (io_threads_) { - // Initialise the array of mailboxes. Additional three slots are for - // xs_term thread and reaper thread. - slot_count = max_sockets + io_threads_ + 3; - slots = (mailbox_t**) malloc (sizeof (mailbox_t*) * slot_count); - alloc_assert (slots); - - // Initialise the infrastructure for xs_term thread. - slots [term_tid] = &term_mailbox; - - // Create the reaper thread. - reaper = new (std::nothrow) reaper_t (this, reaper_tid); - alloc_assert (reaper); - slots [reaper_tid] = reaper->get_mailbox (); - reaper->start (); - - // Create I/O thread objects and launch them. - for (uint32_t i = 2; i != io_threads_ + 2; i++) { - io_thread_t *io_thread = new (std::nothrow) io_thread_t (this, i); - alloc_assert (io_thread); - io_threads.push_back (io_thread); - slots [i] = io_thread->get_mailbox (); - io_thread->start (); - } - - // In the unused part of the slot array, create a list of empty slots. - for (int32_t i = (int32_t) slot_count - 1; - i >= (int32_t) io_threads_ + 2; i--) { - empty_slots.push_back (i); - slots [i] = NULL; - } - - // Create the socket to send logs to. - log_socket = create_socket (XS_PUB); - xs_assert (log_socket); - int linger = 0; - int rc = log_socket->setsockopt (XS_LINGER, &linger, sizeof (linger)); - errno_assert (rc == 0); - int hwm = 1; - rc = log_socket->setsockopt (XS_SNDHWM, &hwm, sizeof (hwm)); - errno_assert (rc == 0); -#if !defined XS_HAVE_WINDOWS - rc = log_socket->connect ("ipc:///tmp/xslogs.ipc"); - errno_assert (rc == 0); -#endif - - // Create the monitor object. - io_thread_t *io_thread = choose_io_thread (0); - xs_assert (io_thread); - monitor = new (std::nothrow) monitor_t (io_thread); - alloc_assert (monitor); - monitor->start (); } bool xs::ctx_t::check_tag () @@ -128,60 +80,142 @@ xs::ctx_t::~ctx_t () int xs::ctx_t::terminate () { - // Check whether termination was already underway, but interrupted and now - // restarted. - slot_sync.lock (); - bool restarted = terminating; - terminating = true; - slot_sync.unlock (); + if (!starting) { - // First attempt to terminate the context. - if (!restarted) { + // Check whether termination was already underway, but interrupted and now + // restarted. + slot_sync.lock (); + bool restarted = terminating; + terminating = true; + slot_sync.unlock (); - // Close the monitor object. Wait for done command from the monitor. - monitor->stop (); + // First attempt to terminate the context. + if (!restarted) { + + // Close the monitor object. Wait for done command from the monitor. + monitor->stop (); + command_t cmd; + int rc = term_mailbox.recv (&cmd, -1); + xs_assert (rc == 0); + xs_assert (cmd.type == command_t::done); + + // Close the logging socket. + log_sync.lock (); + rc = log_socket->close (); + xs_assert (rc == 0); + log_socket = NULL; + log_sync.unlock (); + + // First send stop command to sockets so that any blocking calls + // can be interrupted. If there are no sockets we can ask reaper + // thread to stop. + slot_sync.lock (); + for (sockets_t::size_type i = 0; i != sockets.size (); i++) + sockets [i]->stop (); + if (sockets.empty ()) + reaper->stop (); + slot_sync.unlock (); + } + + // Wait till reaper thread closes all the sockets. command_t cmd; int rc = term_mailbox.recv (&cmd, -1); + if (rc == -1 && errno == EINTR) + return -1; xs_assert (rc == 0); xs_assert (cmd.type == command_t::done); - - // Close the logging socket. - log_sync.lock (); - rc = log_socket->close (); - xs_assert (rc == 0); - log_socket = NULL; - log_sync.unlock (); - - // First send stop command to sockets so that any blocking calls - // can be interrupted. If there are no sockets we can ask reaper - // thread to stop. slot_sync.lock (); - for (sockets_t::size_type i = 0; i != sockets.size (); i++) - sockets [i]->stop (); - if (sockets.empty ()) - reaper->stop (); + xs_assert (sockets.empty ()); slot_sync.unlock (); } - // Wait till reaper thread closes all the sockets. - command_t cmd; - int rc = term_mailbox.recv (&cmd, -1); - if (rc == -1 && errno == EINTR) - return -1; - xs_assert (rc == 0); - xs_assert (cmd.type == command_t::done); - slot_sync.lock (); - xs_assert (sockets.empty ()); - slot_sync.unlock (); - // Deallocate the resources. delete this; return 0; } +int xs::ctx_t::setctxopt (int option_, const void *optval_, size_t optvallen_) +{ + switch (option_) { + case XS_CTX_MAX_SOCKETS: + if (optvallen_ != sizeof (int) || *((int*) optval_) < 0) { + errno = EINVAL; + return -1; + } + opt_sync.lock (); + max_sockets = *((int*) optval_); + opt_sync.unlock (); + break; + default: + errno = EINVAL; + return -1; + } + return 0; +} + xs::socket_base_t *xs::ctx_t::create_socket (int type_) { + if (unlikely (starting)) { + + starting = false; + + // Initialise the array of mailboxes. Additional three slots are for + // xs_term thread and reaper thread. + opt_sync.lock (); + int maxs = max_sockets; + opt_sync.unlock (); + slot_count = maxs + io_thread_count + 3; + slots = (mailbox_t**) malloc (sizeof (mailbox_t*) * slot_count); + alloc_assert (slots); + + // Initialise the infrastructure for xs_term thread. + slots [term_tid] = &term_mailbox; + + // Create the reaper thread. + reaper = new (std::nothrow) reaper_t (this, reaper_tid); + alloc_assert (reaper); + slots [reaper_tid] = reaper->get_mailbox (); + reaper->start (); + + // Create I/O thread objects and launch them. + for (uint32_t i = 2; i != io_thread_count + 2; i++) { + io_thread_t *io_thread = new (std::nothrow) io_thread_t (this, i); + alloc_assert (io_thread); + io_threads.push_back (io_thread); + slots [i] = io_thread->get_mailbox (); + io_thread->start (); + } + + // In the unused part of the slot array, create a list of empty slots. + for (int32_t i = (int32_t) slot_count - 1; + i >= (int32_t) io_thread_count + 2; i--) { + empty_slots.push_back (i); + slots [i] = NULL; + } + + // Create the socket to send logs to. + log_socket = create_socket (XS_PUB); + xs_assert (log_socket); + int linger = 0; + int rc = log_socket->setsockopt (XS_LINGER, &linger, sizeof (linger)); + errno_assert (rc == 0); + int hwm = 1; + rc = log_socket->setsockopt (XS_SNDHWM, &hwm, sizeof (hwm)); + errno_assert (rc == 0); + #if !defined XS_HAVE_WINDOWS + rc = log_socket->connect ("ipc:///tmp/xslogs.ipc"); + errno_assert (rc == 0); + #endif + + // Create the monitor object. + io_thread_t *io_thread = choose_io_thread (0); + xs_assert (io_thread); + monitor = new (std::nothrow) monitor_t (io_thread); + alloc_assert (monitor); + monitor->start (); + } + slot_sync.lock (); // Once xs_term() was called, we can't create new sockets. diff --git a/src/ctx.hpp b/src/ctx.hpp index 8c3941f..56b5d4c 100644 --- a/src/ctx.hpp +++ b/src/ctx.hpp @@ -73,6 +73,9 @@ namespace xs // after the last one is closed. int terminate (); + // Set context option. + int setctxopt (int option_, const void *optval_, size_t optvallen_); + // Create and destroy a socket. xs::socket_base_t *create_socket (int type_); void destroy_socket (xs::socket_base_t *socket_); @@ -119,6 +122,10 @@ namespace xs typedef std::vector <uint32_t> emtpy_slots_t; emtpy_slots_t empty_slots; + // If true, xs_init has been called but no socket have been created + // yes. Launching of I/O threads is delayed. + bool starting; + // If true, xs_term was already called. bool terminating; @@ -160,6 +167,15 @@ namespace xs xs::socket_base_t *log_socket; mutex_t log_sync; + // Maximum number of sockets that can be opened at the same time. + int max_sockets; + + // Number of I/O threads to launch. + uint32_t io_thread_count; + + // Synchronisation of access to context options. + mutex_t opt_sync; + ctx_t (const ctx_t&); const ctx_t &operator = (const ctx_t&); }; @@ -169,6 +169,17 @@ int xs_term (void *ctx_) return rc; } +int xs_setctxopt (void *ctx_, int option_, const void *optval_, + size_t optvallen_) +{ + if (!ctx_ || !((xs::ctx_t*) ctx_)->check_tag ()) { + errno = EFAULT; + return -1; + } + + return ((xs::ctx_t*) ctx_)->setctxopt (option_, optval_, optvallen_); +} + void *xs_socket (void *ctx_, int type_) { if (!ctx_ || !((xs::ctx_t*) ctx_)->check_tag ()) { |