mirror of
https://github.com/zeromq/libzmq.git
synced 2025-03-09 23:36:04 +00:00
Whitespace and style fixes
This commit is contained in:
parent
066606322c
commit
4ba34c9d70
@ -43,7 +43,8 @@ zmq::address_t::~address_t ()
|
||||
}
|
||||
}
|
||||
#if !defined ZMQ_HAVE_WINDOWS && !defined ZMQ_HAVE_OPENVMS
|
||||
else if (protocol == "ipc") {
|
||||
else
|
||||
if (protocol == "ipc") {
|
||||
if (resolved.ipc_addr) {
|
||||
delete resolved.ipc_addr;
|
||||
resolved.ipc_addr = 0;
|
||||
@ -55,15 +56,14 @@ zmq::address_t::~address_t ()
|
||||
int zmq::address_t::to_string (std::string &addr_) const
|
||||
{
|
||||
if (protocol == "tcp") {
|
||||
if (resolved.tcp_addr) {
|
||||
if (resolved.tcp_addr)
|
||||
return resolved.tcp_addr->to_string(addr_);
|
||||
}
|
||||
}
|
||||
#if !defined ZMQ_HAVE_WINDOWS && !defined ZMQ_HAVE_OPENVMS
|
||||
else if (protocol == "ipc") {
|
||||
if (resolved.ipc_addr) {
|
||||
else
|
||||
if (protocol == "ipc") {
|
||||
if (resolved.ipc_addr)
|
||||
return resolved.ipc_addr->to_string(addr_);
|
||||
}
|
||||
}
|
||||
#endif
|
||||
|
||||
|
@ -145,7 +145,8 @@ void zmq::ipc_connecter_t::start_connecting ()
|
||||
}
|
||||
|
||||
// Connection establishment may be delayed. Poll for its completion.
|
||||
else if (rc == -1 && errno == EINPROGRESS) {
|
||||
else
|
||||
if (rc == -1 && errno == EINPROGRESS) {
|
||||
handle = add_fd (s);
|
||||
handle_valid = true;
|
||||
set_pollout (handle);
|
||||
|
@ -53,7 +53,8 @@ zmq::mtrie_t::~mtrie_t ()
|
||||
delete next.node;
|
||||
next.node = 0;
|
||||
}
|
||||
else if (count > 1) {
|
||||
else
|
||||
if (count > 1) {
|
||||
for (unsigned short i = 0; i != count; ++i)
|
||||
if (next.table [i])
|
||||
delete next.table [i];
|
||||
@ -90,7 +91,8 @@ bool zmq::mtrie_t::add_helper (unsigned char *prefix_, size_t size_,
|
||||
count = 1;
|
||||
next.node = NULL;
|
||||
}
|
||||
else if (count == 1) {
|
||||
else
|
||||
if (count == 1) {
|
||||
unsigned char oldc = min;
|
||||
mtrie_t *oldp = next.node;
|
||||
count = (min < c ? c - min : min - c) + 1;
|
||||
@ -102,8 +104,8 @@ bool zmq::mtrie_t::add_helper (unsigned char *prefix_, size_t size_,
|
||||
min = std::min (min, c);
|
||||
next.table [oldc - min] = oldp;
|
||||
}
|
||||
else if (min < c) {
|
||||
|
||||
else
|
||||
if (min < c) {
|
||||
// The new character is above the current character range.
|
||||
unsigned short old_count = count;
|
||||
count = c - min + 1;
|
||||
@ -114,7 +116,6 @@ bool zmq::mtrie_t::add_helper (unsigned char *prefix_, size_t size_,
|
||||
next.table [i] = NULL;
|
||||
}
|
||||
else {
|
||||
|
||||
// The new character is below the current character range.
|
||||
unsigned short old_count = count;
|
||||
count = (min + old_count) - c;
|
||||
@ -244,7 +245,8 @@ void zmq::mtrie_t::rm_helper (pipe_t *pipe_, unsigned char **buff_,
|
||||
count = 0;
|
||||
}
|
||||
// Compact the node table if possible
|
||||
else if (live_nodes == 1) {
|
||||
else
|
||||
if (live_nodes == 1) {
|
||||
// If there's only one live node in the table we can
|
||||
// switch to using the more compact single-node
|
||||
// representation
|
||||
@ -257,7 +259,8 @@ void zmq::mtrie_t::rm_helper (pipe_t *pipe_, unsigned char **buff_,
|
||||
count = 1;
|
||||
min = new_min;
|
||||
}
|
||||
else if (new_min > min || new_max < min + count - 1) {
|
||||
else
|
||||
if (new_min > min || new_max < min + count - 1) {
|
||||
zmq_assert (new_max - new_min + 1 > 1);
|
||||
|
||||
mtrie_t **old_table = next.table;
|
||||
@ -342,7 +345,8 @@ bool zmq::mtrie_t::rm_helper (unsigned char *prefix_, size_t size_,
|
||||
free (next.table);
|
||||
next.node = oldp;
|
||||
}
|
||||
else if (c == min) {
|
||||
else
|
||||
if (c == min) {
|
||||
// We can compact the table "from the left"
|
||||
unsigned short i;
|
||||
for (i = 1; i < count; ++i)
|
||||
@ -358,7 +362,8 @@ bool zmq::mtrie_t::rm_helper (unsigned char *prefix_, size_t size_,
|
||||
memmove (next.table, old_table + i, sizeof (mtrie_t*) * count);
|
||||
free (old_table);
|
||||
}
|
||||
else if (c == min + count - 1) {
|
||||
else
|
||||
if (c == min + count - 1) {
|
||||
// We can compact the table "from the right"
|
||||
unsigned short i;
|
||||
for (i = 1; i < count; ++i)
|
||||
|
@ -187,16 +187,17 @@ void zmq::pgm_sender_t::out_event ()
|
||||
size_t nbytes = pgm_socket.send (out_buffer, write_size);
|
||||
|
||||
// We can write either all data or 0 which means rate limit reached.
|
||||
if (nbytes == write_size) {
|
||||
if (nbytes == write_size)
|
||||
write_size = 0;
|
||||
} else {
|
||||
else {
|
||||
zmq_assert (nbytes == 0);
|
||||
|
||||
if (errno == ENOMEM) {
|
||||
const long timeout = pgm_socket.get_tx_timeout ();
|
||||
add_timer (timeout, tx_timer_id);
|
||||
has_tx_timer = true;
|
||||
} else
|
||||
}
|
||||
else
|
||||
errno_assert (errno == EBUSY);
|
||||
}
|
||||
}
|
||||
@ -207,10 +208,13 @@ void zmq::pgm_sender_t::timer_event (int token)
|
||||
if (token == rx_timer_id) {
|
||||
has_rx_timer = false;
|
||||
in_event ();
|
||||
} else if (token == tx_timer_id) {
|
||||
}
|
||||
else
|
||||
if (token == tx_timer_id) {
|
||||
has_tx_timer = false;
|
||||
out_event ();
|
||||
} else
|
||||
}
|
||||
else
|
||||
zmq_assert (false);
|
||||
}
|
||||
|
||||
|
@ -239,7 +239,8 @@ int zmq::pgm_socket_t::init (bool udp_encapsulation_, const char *network_)
|
||||
!pgm_setsockopt (sock, IPPROTO_PGM, PGM_NAK_NCF_RETRIES,
|
||||
&nak_ncf_retries, sizeof (nak_ncf_retries)))
|
||||
goto err_abort;
|
||||
} else {
|
||||
}
|
||||
else {
|
||||
const int send_only = 1,
|
||||
max_rte = (int) ((options.rate * 1000) / 8),
|
||||
txw_max_tpdu = (int) pgm_max_tpdu,
|
||||
@ -473,7 +474,8 @@ size_t zmq::pgm_socket_t::send (unsigned char *data_, size_t data_len_)
|
||||
if (nbytes > 0) {
|
||||
zmq_assert (status == PGM_IO_STATUS_NORMAL);
|
||||
zmq_assert (nbytes == data_len_);
|
||||
} else {
|
||||
}
|
||||
else {
|
||||
zmq_assert (status == PGM_IO_STATUS_RATE_LIMITED ||
|
||||
status == PGM_IO_STATUS_WOULD_BLOCK);
|
||||
|
||||
@ -677,7 +679,8 @@ void zmq::pgm_socket_t::process_upstream ()
|
||||
|
||||
if (status == PGM_IO_STATUS_TIMER_PENDING)
|
||||
errno = EBUSY;
|
||||
else if (status == PGM_IO_STATUS_RATE_LIMITED)
|
||||
else
|
||||
if (status == PGM_IO_STATUS_RATE_LIMITED)
|
||||
errno = ENOMEM;
|
||||
else
|
||||
errno = EAGAIN;
|
||||
|
41
src/pipe.cpp
41
src/pipe.cpp
@ -179,11 +179,11 @@ void zmq::pipe_t::rollback ()
|
||||
// Remove incomplete message from the outbound pipe.
|
||||
msg_t msg;
|
||||
if (outpipe) {
|
||||
while (outpipe->unwrite (&msg)) {
|
||||
zmq_assert (msg.flags () & msg_t::more);
|
||||
int rc = msg.close ();
|
||||
errno_assert (rc == 0);
|
||||
}
|
||||
while (outpipe->unwrite (&msg)) {
|
||||
zmq_assert (msg.flags () & msg_t::more);
|
||||
int rc = msg.close ();
|
||||
errno_assert (rc == 0);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@ -324,32 +324,37 @@ void zmq::pipe_t::terminate (bool delay_)
|
||||
|
||||
// If the pipe is in the final phase of async termination, it's going to
|
||||
// closed anyway. No need to do anything special here.
|
||||
else if (state == terminating)
|
||||
else
|
||||
if (state == terminating)
|
||||
return;
|
||||
|
||||
// The simple sync termination case. Ask the peer to terminate and wait
|
||||
// for the ack.
|
||||
else if (state == active) {
|
||||
else
|
||||
if (state == active) {
|
||||
send_pipe_term (peer);
|
||||
state = terminated;
|
||||
}
|
||||
|
||||
// There are still pending messages available, but the user calls
|
||||
// 'terminate'. We can act as if all the pending messages were read.
|
||||
else if (state == pending && !delay) {
|
||||
else
|
||||
if (state == pending && !delay) {
|
||||
outpipe = NULL;
|
||||
send_pipe_term_ack (peer);
|
||||
state = terminating;
|
||||
}
|
||||
|
||||
// If there are pending messages still availabe, do nothing.
|
||||
else if (state == pending) {
|
||||
else
|
||||
if (state == pending) {
|
||||
}
|
||||
|
||||
// We've already got delimiter, but not term command yet. We can ignore
|
||||
// the delimiter and ack synchronously terminate as if we were in
|
||||
// active state.
|
||||
else if (state == delimited) {
|
||||
else
|
||||
if (state == delimited) {
|
||||
send_pipe_term (peer);
|
||||
state = terminated;
|
||||
}
|
||||
@ -363,15 +368,15 @@ void zmq::pipe_t::terminate (bool delay_)
|
||||
|
||||
if (outpipe) {
|
||||
|
||||
// Drop any unfinished outbound messages.
|
||||
rollback ();
|
||||
// Drop any unfinished outbound messages.
|
||||
rollback ();
|
||||
|
||||
// Write the delimiter into the pipe. Note that watermarks are not
|
||||
// checked; thus the delimiter can be written even when the pipe is full.
|
||||
msg_t msg;
|
||||
msg.init_delimiter ();
|
||||
outpipe->write (msg, false);
|
||||
flush ();
|
||||
// Write the delimiter into the pipe. Note that watermarks are not
|
||||
// checked; thus the delimiter can be written even when the pipe is full.
|
||||
msg_t msg;
|
||||
msg.init_delimiter ();
|
||||
outpipe->write (msg, false);
|
||||
flush ();
|
||||
}
|
||||
}
|
||||
|
||||
|
@ -41,7 +41,8 @@ void zmq::poller_base_t::adjust_load (int amount_)
|
||||
{
|
||||
if (amount_ > 0)
|
||||
load.add (amount_);
|
||||
else if (amount_ < 0)
|
||||
else
|
||||
if (amount_ < 0)
|
||||
load.sub (-amount_);
|
||||
}
|
||||
|
||||
|
@ -182,7 +182,8 @@ int zmq::router_t::xsend (msg_t *msg_, int flags_)
|
||||
bool ok = current_out->write (msg_);
|
||||
if (unlikely (!ok))
|
||||
current_out = NULL;
|
||||
else if (!more_out) {
|
||||
else
|
||||
if (!more_out) {
|
||||
current_out->flush ();
|
||||
current_out = NULL;
|
||||
}
|
||||
|
@ -479,7 +479,8 @@ void zmq::session_base_t::start_connecting (bool wait_)
|
||||
|
||||
send_attach (this, pgm_sender);
|
||||
}
|
||||
else if (options.type == ZMQ_SUB || options.type == ZMQ_XSUB) {
|
||||
else
|
||||
if (options.type == ZMQ_SUB || options.type == ZMQ_XSUB) {
|
||||
|
||||
// PGM receiver.
|
||||
pgm_receiver_t *pgm_receiver = new (std::nothrow) pgm_receiver_t (
|
||||
|
@ -147,7 +147,8 @@ int zmq::signaler_t::wait (int timeout_)
|
||||
errno_assert (errno == EINTR);
|
||||
return -1;
|
||||
}
|
||||
else if (unlikely (rc == 0)) {
|
||||
else
|
||||
if (unlikely (rc == 0)) {
|
||||
errno = EAGAIN;
|
||||
return -1;
|
||||
}
|
||||
|
@ -507,7 +507,8 @@ int zmq::socket_base_t::connect (const char *addr_)
|
||||
}
|
||||
}
|
||||
#if !defined ZMQ_HAVE_WINDOWS && !defined ZMQ_HAVE_OPENVMS
|
||||
else if(protocol == "ipc") {
|
||||
else
|
||||
if (protocol == "ipc") {
|
||||
paddr->resolved.ipc_addr = new (std::nothrow) ipc_address_t ();
|
||||
alloc_assert (paddr->resolved.ipc_addr);
|
||||
int rc = paddr->resolved.ipc_addr->resolve (address.c_str ());
|
||||
@ -1033,7 +1034,6 @@ int zmq::socket_base_t::monitor (const char *addr_, int events_)
|
||||
|
||||
// Register events to monitor
|
||||
monitor_events = events_;
|
||||
|
||||
monitor_socket = zmq_socket( get_ctx (), ZMQ_PAIR);
|
||||
if (monitor_socket == NULL)
|
||||
return -1;
|
||||
@ -1053,112 +1053,123 @@ int zmq::socket_base_t::monitor (const char *addr_, int events_)
|
||||
|
||||
void zmq::socket_base_t::event_connected (const char *addr_, int fd_)
|
||||
{
|
||||
zmq_event_t event;
|
||||
if (!(monitor_events & ZMQ_EVENT_CONNECTED)) return;
|
||||
event.event = ZMQ_EVENT_CONNECTED;
|
||||
event.data.connected.addr = (char *)addr_;
|
||||
event.data.connected.fd = fd_;
|
||||
monitor_event (event);
|
||||
if (monitor_events & ZMQ_EVENT_CONNECTED) {
|
||||
zmq_event_t event;
|
||||
event.event = ZMQ_EVENT_CONNECTED;
|
||||
event.data.connected.addr = (char *) addr_;
|
||||
event.data.connected.fd = fd_;
|
||||
monitor_event (event);
|
||||
}
|
||||
}
|
||||
|
||||
void zmq::socket_base_t::event_connect_delayed (const char *addr_, int err_)
|
||||
{
|
||||
zmq_event_t event;
|
||||
if (!(monitor_events & ZMQ_EVENT_CONNECT_DELAYED)) return;
|
||||
event.event = ZMQ_EVENT_CONNECT_DELAYED;
|
||||
event.data.connected.addr = (char *)addr_;
|
||||
event.data.connect_delayed.err = err_;
|
||||
monitor_event (event);
|
||||
if (monitor_events & ZMQ_EVENT_CONNECT_DELAYED) {
|
||||
zmq_event_t event;
|
||||
event.event = ZMQ_EVENT_CONNECT_DELAYED;
|
||||
event.data.connected.addr = (char *) addr_;
|
||||
event.data.connect_delayed.err = err_;
|
||||
monitor_event (event);
|
||||
}
|
||||
}
|
||||
|
||||
void zmq::socket_base_t::event_connect_retried (const char *addr_, int interval_)
|
||||
{
|
||||
zmq_event_t event;
|
||||
if (!(monitor_events & ZMQ_EVENT_CONNECT_RETRIED)) return;
|
||||
event.event = ZMQ_EVENT_CONNECT_RETRIED;
|
||||
event.data.connected.addr = (char *)addr_;
|
||||
event.data.connect_retried.interval = interval_;
|
||||
monitor_event (event);
|
||||
if (monitor_events & ZMQ_EVENT_CONNECT_RETRIED) {
|
||||
zmq_event_t event;
|
||||
event.event = ZMQ_EVENT_CONNECT_RETRIED;
|
||||
event.data.connected.addr = (char *) addr_;
|
||||
event.data.connect_retried.interval = interval_;
|
||||
monitor_event (event);
|
||||
}
|
||||
}
|
||||
|
||||
void zmq::socket_base_t::event_listening (const char *addr_, int fd_)
|
||||
{
|
||||
zmq_event_t event;
|
||||
if (!(monitor_events & ZMQ_EVENT_LISTENING)) return;
|
||||
event.event = ZMQ_EVENT_LISTENING;
|
||||
event.data.connected.addr = (char *)addr_;
|
||||
event.data.listening.fd = fd_;
|
||||
monitor_event (event);
|
||||
if (monitor_events & ZMQ_EVENT_LISTENING) {
|
||||
zmq_event_t event;
|
||||
event.event = ZMQ_EVENT_LISTENING;
|
||||
event.data.connected.addr = (char *) addr_;
|
||||
event.data.listening.fd = fd_;
|
||||
monitor_event (event);
|
||||
}
|
||||
}
|
||||
|
||||
void zmq::socket_base_t::event_bind_failed (const char *addr_, int err_)
|
||||
{
|
||||
zmq_event_t event;
|
||||
if (!(monitor_events & ZMQ_EVENT_BIND_FAILED)) return;
|
||||
event.event = ZMQ_EVENT_BIND_FAILED;
|
||||
event.data.connected.addr = (char *)addr_;
|
||||
event.data.bind_failed.err = err_;
|
||||
monitor_event (event);
|
||||
if (monitor_events & ZMQ_EVENT_BIND_FAILED) {
|
||||
zmq_event_t event;
|
||||
event.event = ZMQ_EVENT_BIND_FAILED;
|
||||
event.data.connected.addr = (char *) addr_;
|
||||
event.data.bind_failed.err = err_;
|
||||
monitor_event (event);
|
||||
}
|
||||
}
|
||||
|
||||
void zmq::socket_base_t::event_accepted (const char *addr_, int fd_)
|
||||
{
|
||||
zmq_event_t event;
|
||||
if (!(monitor_events & ZMQ_EVENT_ACCEPTED)) return;
|
||||
event.event = ZMQ_EVENT_ACCEPTED;
|
||||
event.data.connected.addr = (char *)addr_;
|
||||
event.data.accepted.fd = fd_;
|
||||
monitor_event (event);
|
||||
if (monitor_events & ZMQ_EVENT_ACCEPTED) {
|
||||
zmq_event_t event;
|
||||
event.event = ZMQ_EVENT_ACCEPTED;
|
||||
event.data.connected.addr = (char *) addr_;
|
||||
event.data.accepted.fd = fd_;
|
||||
monitor_event (event);
|
||||
}
|
||||
}
|
||||
|
||||
void zmq::socket_base_t::event_accept_failed (const char *addr_, int err_)
|
||||
{
|
||||
zmq_event_t event;
|
||||
if (!(monitor_events & ZMQ_EVENT_ACCEPT_FAILED)) return;
|
||||
event.event = ZMQ_EVENT_ACCEPT_FAILED;
|
||||
event.data.connected.addr = (char *)addr_;
|
||||
event.data.accept_failed.err= err_;
|
||||
monitor_event (event);
|
||||
if (monitor_events & ZMQ_EVENT_ACCEPT_FAILED) {
|
||||
zmq_event_t event;
|
||||
event.event = ZMQ_EVENT_ACCEPT_FAILED;
|
||||
event.data.connected.addr = (char *) addr_;
|
||||
event.data.accept_failed.err= err_;
|
||||
monitor_event (event);
|
||||
}
|
||||
}
|
||||
|
||||
void zmq::socket_base_t::event_closed (const char *addr_, int fd_)
|
||||
{
|
||||
zmq_event_t event;
|
||||
if (!(monitor_events & ZMQ_EVENT_CLOSED)) return;
|
||||
event.event = ZMQ_EVENT_CLOSED;
|
||||
event.data.connected.addr = (char *)addr_;
|
||||
event.data.closed.fd = fd_;
|
||||
monitor_event (event);
|
||||
if (monitor_events & ZMQ_EVENT_CLOSED) {
|
||||
zmq_event_t event;
|
||||
event.event = ZMQ_EVENT_CLOSED;
|
||||
event.data.connected.addr = (char *) addr_;
|
||||
event.data.closed.fd = fd_;
|
||||
monitor_event (event);
|
||||
}
|
||||
}
|
||||
|
||||
void zmq::socket_base_t::event_close_failed (const char *addr_, int err_)
|
||||
{
|
||||
zmq_event_t event;
|
||||
if (!(monitor_events & ZMQ_EVENT_CLOSE_FAILED)) return;
|
||||
event.event = ZMQ_EVENT_CLOSE_FAILED;
|
||||
event.data.connected.addr = (char *)addr_;
|
||||
event.data.close_failed.err = err_;
|
||||
monitor_event (event);
|
||||
if (monitor_events & ZMQ_EVENT_CLOSE_FAILED) {
|
||||
zmq_event_t event;
|
||||
event.event = ZMQ_EVENT_CLOSE_FAILED;
|
||||
event.data.connected.addr = (char *) addr_;
|
||||
event.data.close_failed.err = err_;
|
||||
monitor_event (event);
|
||||
}
|
||||
}
|
||||
|
||||
void zmq::socket_base_t::event_disconnected (const char *addr_, int fd_)
|
||||
{
|
||||
zmq_event_t event;
|
||||
if (!(monitor_events & ZMQ_EVENT_DISCONNECTED)) return;
|
||||
event.event = ZMQ_EVENT_DISCONNECTED;
|
||||
event.data.connected.addr = (char *)addr_;
|
||||
event.data.disconnected.fd = fd_;
|
||||
monitor_event (event);
|
||||
if (monitor_events & ZMQ_EVENT_DISCONNECTED) {
|
||||
zmq_event_t event;
|
||||
event.event = ZMQ_EVENT_DISCONNECTED;
|
||||
event.data.connected.addr = (char *) addr_;
|
||||
event.data.disconnected.fd = fd_;
|
||||
monitor_event (event);
|
||||
}
|
||||
}
|
||||
|
||||
void zmq::socket_base_t::monitor_event (zmq_event_t event_)
|
||||
{
|
||||
zmq_msg_t msg;
|
||||
if (!monitor_socket) return;
|
||||
zmq_msg_init_size (&msg, sizeof (event_));
|
||||
memcpy (zmq_msg_data (&msg), &event_, sizeof (event_));
|
||||
zmq_sendmsg (monitor_socket, &msg, 0);
|
||||
zmq_msg_close (&msg);
|
||||
if (monitor_socket) {
|
||||
zmq_msg_t msg;
|
||||
zmq_msg_init_size (&msg, sizeof (event_));
|
||||
memcpy (zmq_msg_data (&msg), &event_, sizeof (event_));
|
||||
zmq_sendmsg (monitor_socket, &msg, 0);
|
||||
zmq_msg_close (&msg);
|
||||
}
|
||||
}
|
||||
|
||||
void zmq::socket_base_t::stop_monitor()
|
||||
@ -1169,4 +1180,3 @@ void zmq::socket_base_t::stop_monitor()
|
||||
monitor_events = 0;
|
||||
}
|
||||
}
|
||||
|
||||
|
@ -51,7 +51,8 @@ int zmq::sub_t::xsetsockopt (int option_, const void *optval_,
|
||||
unsigned char *data = (unsigned char*) msg.data ();
|
||||
if (option_ == ZMQ_SUBSCRIBE)
|
||||
*data = 1;
|
||||
else if (option_ == ZMQ_UNSUBSCRIBE)
|
||||
else
|
||||
if (option_ == ZMQ_UNSUBSCRIBE)
|
||||
*data = 0;
|
||||
memcpy (data + 1, optval_, optvallen_);
|
||||
|
||||
|
@ -159,7 +159,8 @@ void zmq::tcp_connecter_t::start_connecting ()
|
||||
}
|
||||
|
||||
// Connection establishment may be delayed. Poll for its completion.
|
||||
else if (rc == -1 && errno == EINPROGRESS) {
|
||||
else
|
||||
if (rc == -1 && errno == EINPROGRESS) {
|
||||
handle = add_fd (s);
|
||||
handle_valid = true;
|
||||
set_pollout (handle);
|
||||
|
215
src/trie.cpp
215
src/trie.cpp
@ -48,7 +48,8 @@ zmq::trie_t::~trie_t ()
|
||||
delete next.node;
|
||||
next.node = 0;
|
||||
}
|
||||
else if (count > 1) {
|
||||
else
|
||||
if (count > 1) {
|
||||
for (unsigned short i = 0; i != count; ++i)
|
||||
if (next.table [i])
|
||||
delete next.table [i];
|
||||
@ -74,7 +75,8 @@ bool zmq::trie_t::add (unsigned char *prefix_, size_t size_)
|
||||
count = 1;
|
||||
next.node = NULL;
|
||||
}
|
||||
else if (count == 1) {
|
||||
else
|
||||
if (count == 1) {
|
||||
unsigned char oldc = min;
|
||||
trie_t *oldp = next.node;
|
||||
count = (min < c ? c - min : min - c) + 1;
|
||||
@ -86,8 +88,8 @@ bool zmq::trie_t::add (unsigned char *prefix_, size_t size_)
|
||||
min = std::min (min, c);
|
||||
next.table [oldc - min] = oldp;
|
||||
}
|
||||
else if (min < c) {
|
||||
|
||||
else
|
||||
if (min < c) {
|
||||
// The new character is above the current character range.
|
||||
unsigned short old_count = count;
|
||||
count = c - min + 1;
|
||||
@ -136,121 +138,120 @@ bool zmq::trie_t::add (unsigned char *prefix_, size_t size_)
|
||||
|
||||
bool zmq::trie_t::rm (unsigned char *prefix_, size_t size_)
|
||||
{
|
||||
// TODO: Shouldn't an error be reported if the key does not exist?
|
||||
// TODO: Shouldn't an error be reported if the key does not exist?
|
||||
if (!size_) {
|
||||
if (!refcnt)
|
||||
return false;
|
||||
refcnt--;
|
||||
return refcnt == 0;
|
||||
}
|
||||
unsigned char c = *prefix_;
|
||||
if (!count || c < min || c >= min + count)
|
||||
return false;
|
||||
|
||||
if (!size_) {
|
||||
if (!refcnt)
|
||||
return false;
|
||||
refcnt--;
|
||||
return refcnt == 0;
|
||||
}
|
||||
trie_t *next_node =
|
||||
count == 1 ? next.node : next.table [c - min];
|
||||
|
||||
unsigned char c = *prefix_;
|
||||
if (!count || c < min || c >= min + count)
|
||||
return false;
|
||||
if (!next_node)
|
||||
return false;
|
||||
|
||||
trie_t *next_node =
|
||||
count == 1 ? next.node : next.table [c - min];
|
||||
bool ret = next_node->rm (prefix_ + 1, size_ - 1);
|
||||
|
||||
if (!next_node)
|
||||
return false;
|
||||
// Prune redundant nodes
|
||||
if (next_node->is_redundant ()) {
|
||||
delete next_node;
|
||||
zmq_assert (count > 0);
|
||||
|
||||
bool ret = next_node->rm (prefix_ + 1, size_ - 1);
|
||||
if (count == 1) {
|
||||
// The just pruned node is was the only live node
|
||||
next.node = 0;
|
||||
count = 0;
|
||||
--live_nodes;
|
||||
zmq_assert (live_nodes == 0);
|
||||
}
|
||||
else {
|
||||
next.table [c - min] = 0;
|
||||
zmq_assert (live_nodes > 1);
|
||||
--live_nodes;
|
||||
|
||||
// Prune redundant nodes
|
||||
if (next_node->is_redundant ()) {
|
||||
delete next_node;
|
||||
zmq_assert (count > 0);
|
||||
// Compact the table if possible
|
||||
if (live_nodes == 1) {
|
||||
// We can switch to using the more compact single-node
|
||||
// representation since the table only contains one live node
|
||||
trie_t *node = 0;
|
||||
// Since we always compact the table the pruned node must
|
||||
// either be the left-most or right-most ptr in the node
|
||||
// table
|
||||
if (c == min) {
|
||||
// The pruned node is the left-most node ptr in the
|
||||
// node table => keep the right-most node
|
||||
node = next.table [count - 1];
|
||||
min += count - 1;
|
||||
}
|
||||
else
|
||||
if (c == min + count - 1) {
|
||||
// The pruned node is the right-most node ptr in the
|
||||
// node table => keep the left-most node
|
||||
node = next.table [0];
|
||||
}
|
||||
zmq_assert (node);
|
||||
free (next.table);
|
||||
next.node = node;
|
||||
count = 1;
|
||||
}
|
||||
else
|
||||
if (c == min) {
|
||||
// We can compact the table "from the left".
|
||||
// Find the left-most non-null node ptr, which we'll use as
|
||||
// our new min
|
||||
unsigned char new_min = min;
|
||||
for (unsigned short i = 1; i < count; ++i) {
|
||||
if (next.table [i]) {
|
||||
new_min = i + min;
|
||||
break;
|
||||
}
|
||||
}
|
||||
zmq_assert (new_min != min);
|
||||
|
||||
if (count == 1) {
|
||||
// The just pruned node is was the only live node
|
||||
next.node = 0;
|
||||
count = 0;
|
||||
--live_nodes;
|
||||
zmq_assert (live_nodes == 0);
|
||||
}
|
||||
else {
|
||||
next.table [c - min] = 0;
|
||||
zmq_assert (live_nodes > 1);
|
||||
--live_nodes;
|
||||
trie_t **old_table = next.table;
|
||||
zmq_assert (new_min > min);
|
||||
zmq_assert (count > new_min - min);
|
||||
|
||||
// Compact the table if possible
|
||||
if (live_nodes == 1) {
|
||||
// We can switch to using the more compact single-node
|
||||
// representation since the table only contains one live node
|
||||
trie_t *node = 0;
|
||||
// Since we always compact the table the pruned node must
|
||||
// either be the left-most or right-most ptr in the node
|
||||
// table
|
||||
if (c == min) {
|
||||
// The pruned node is the left-most node ptr in the
|
||||
// node table => keep the right-most node
|
||||
node = next.table [count - 1];
|
||||
min += count - 1;
|
||||
}
|
||||
else if (c == min + count - 1) {
|
||||
// The pruned node is the right-most node ptr in the
|
||||
// node table => keep the left-most node
|
||||
node = next.table [0];
|
||||
}
|
||||
count = count - (new_min - min);
|
||||
next.table = (trie_t**) malloc (sizeof (trie_t*) * count);
|
||||
alloc_assert (next.table);
|
||||
|
||||
zmq_assert (node);
|
||||
free (next.table);
|
||||
next.node = node;
|
||||
count = 1;
|
||||
}
|
||||
else if (c == min) {
|
||||
// We can compact the table "from the left".
|
||||
// Find the left-most non-null node ptr, which we'll use as
|
||||
// our new min
|
||||
unsigned char new_min = min;
|
||||
for (unsigned short i = 1; i < count; ++i) {
|
||||
if (next.table [i]) {
|
||||
new_min = i + min;
|
||||
break;
|
||||
}
|
||||
}
|
||||
zmq_assert (new_min != min);
|
||||
memmove (next.table, old_table + (new_min - min),
|
||||
sizeof (trie_t*) * count);
|
||||
free (old_table);
|
||||
|
||||
trie_t **old_table = next.table;
|
||||
zmq_assert (new_min > min);
|
||||
zmq_assert (count > new_min - min);
|
||||
min = new_min;
|
||||
}
|
||||
else
|
||||
if (c == min + count - 1) {
|
||||
// We can compact the table "from the right".
|
||||
// Find the right-most non-null node ptr, which we'll use to
|
||||
// determine the new table size
|
||||
unsigned short new_count = count;
|
||||
for (unsigned short i = 1; i < count; ++i) {
|
||||
if (next.table [count - 1 - i]) {
|
||||
new_count = count - i;
|
||||
break;
|
||||
}
|
||||
}
|
||||
zmq_assert (new_count != count);
|
||||
count = new_count;
|
||||
|
||||
count = count - (new_min - min);
|
||||
next.table = (trie_t**) malloc (sizeof (trie_t*) * count);
|
||||
alloc_assert (next.table);
|
||||
trie_t **old_table = next.table;
|
||||
next.table = (trie_t**) malloc (sizeof (trie_t*) * count);
|
||||
alloc_assert (next.table);
|
||||
|
||||
memmove (next.table, old_table + (new_min - min),
|
||||
sizeof (trie_t*) * count);
|
||||
free (old_table);
|
||||
|
||||
min = new_min;
|
||||
}
|
||||
else if (c == min + count - 1) {
|
||||
// We can compact the table "from the right".
|
||||
// Find the right-most non-null node ptr, which we'll use to
|
||||
// determine the new table size
|
||||
unsigned short new_count = count;
|
||||
for (unsigned short i = 1; i < count; ++i) {
|
||||
if (next.table [count - 1 - i]) {
|
||||
new_count = count - i;
|
||||
break;
|
||||
}
|
||||
}
|
||||
zmq_assert (new_count != count);
|
||||
count = new_count;
|
||||
|
||||
trie_t **old_table = next.table;
|
||||
next.table = (trie_t**) malloc (sizeof (trie_t*) * count);
|
||||
alloc_assert (next.table);
|
||||
|
||||
memmove (next.table, old_table, sizeof (trie_t*) * count);
|
||||
free (old_table);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return ret;
|
||||
memmove (next.table, old_table, sizeof (trie_t*) * count);
|
||||
free (old_table);
|
||||
}
|
||||
}
|
||||
}
|
||||
return ret;
|
||||
}
|
||||
|
||||
bool zmq::trie_t::check (unsigned char *data_, size_t size_)
|
||||
|
22
src/zmq.cpp
22
src/zmq.cpp
@ -674,15 +674,15 @@ int zmq_poll (zmq_pollitem_t *items_, int nitems_, long timeout_)
|
||||
int nevents = 0;
|
||||
|
||||
while (true) {
|
||||
|
||||
// Compute the timeout for the subsequent poll.
|
||||
int timeout;
|
||||
if (first_pass)
|
||||
timeout = 0;
|
||||
else if (timeout_ < 0)
|
||||
timeout = -1;
|
||||
else
|
||||
timeout = end - now;
|
||||
// Compute the timeout for the subsequent poll.
|
||||
int timeout;
|
||||
if (first_pass)
|
||||
timeout = 0;
|
||||
else
|
||||
if (timeout_ < 0)
|
||||
timeout = -1;
|
||||
else
|
||||
timeout = end - now;
|
||||
|
||||
// Wait for events.
|
||||
while (true) {
|
||||
@ -694,7 +694,6 @@ int zmq_poll (zmq_pollitem_t *items_, int nitems_, long timeout_)
|
||||
errno_assert (rc >= 0);
|
||||
break;
|
||||
}
|
||||
|
||||
// Check for the events.
|
||||
for (int i = 0; i != nitems_; i++) {
|
||||
|
||||
@ -848,7 +847,8 @@ int zmq_poll (zmq_pollitem_t *items_, int nitems_, long timeout_)
|
||||
timeout.tv_usec = 0;
|
||||
ptimeout = &timeout;
|
||||
}
|
||||
else if (timeout_ < 0)
|
||||
else
|
||||
if (timeout_ < 0)
|
||||
ptimeout = NULL;
|
||||
else {
|
||||
timeout.tv_sec = (long) ((end - now) / 1000);
|
||||
|
Loading…
x
Reference in New Issue
Block a user