In real world usage, there have been reported signaler failures where the
eventfd read() or socket recv() system call in signaler::recv() fails,
despite having made a prior successful signaler::wait() call.
this patch creates a signaler::recv_failable() method that allows
unreadable eventfd / socket to return an error without asserting.
These tests connected CLIENT and SERVER to DEALER... this isn't
allowed. I changed to CLIENT-to-SERVER in both cases. The result
was aborts in client.cpp and server.cpp which cannot handle
invalid multipart data.
I removed the asserts in each of these in xsend.
Solution: fix the test cases and remove the (unwanted?) asserts
in client.cpp:xsend and server.cpp:xsend.
Tests were failing, because some deque calls were causing undefined
behavior: calling front() or pop_front() on an empty deque. Such
calls are now safeguarded.
This patch adds a new member of type deque to the
xpub class that contains pointers to metadata_t.
This deque is then used (alongside pending_data
and pending_flags) to preserve metadata when
copying messages.
The routing id (identity) frame return when reading from
a router doesn't have the same metadata as the "real"
message that follows.
For example, The ZAP "User-Id" property is missing.
This patch attach the "data message"'s metadata
to the "identity message" when it is read from the router.
While sending very large messages (far beyond what fits in a the tcp
buffer, so it takes multiple sendto system calls for it to finish),
zmq_close will close the connection regardless of ZMQ_LINGER.
In case no engine is attached, a pipe->check_read() is needed to look
for the delimiter in the pipe and ultimately trigger the pipe
termination.
However, if there *is* an engine attached, the check_read() looks ahead
and finds the delimiter and terminates the connection even though the
engine might actually still be in the middle of sending a message.
This happens because while the io_thread is still busy sending the data,
the pipe can get terminated and the io thread ends up being terminated.
Despite the old comments, re-initing the msg_t leaks a refcount to
metadata in some situations.
v1_decoder looks like it isn't tested any more, but it seems like a good
idea to fix it because it has the exact same piece of buggy code
v2_decoder does.
Fixes not receiving unsubscription messages in XPUB socket with
ZMQ_XPUB_VERBOSE and using a XSUB-XPUB proxy in front.
This adds two modifications:
- It adds a new flag, ZMQ_XPUB_VERBOSE_UNSUBSCRIBE, to enable verbose
unsubscription messages, necessary when using a XSUB/XPUB proxy.
- It adds a boolean switch to zmq::mtrie_t::rm () to control if the
callback is invoked every time or only in the last removal. Necessary
when a pipe is terminated and the verbose mode for unsubscriptions is
enabled.
A memcpy is eliminated when receiving data on a ZMQ_STREAM socket. Instead
of receiving into a static buffer and then copying the data into the
buffer malloced in msg_t::init_size, the raw_decoder allocates the memory
for together with the reference-counter and creates a msg_t object
on top of that memory. This saves the memcpy operation.
For small messages, data is still copied and the receive buffer is reused.
Set the ZMQ_HEARTBEAT_TIMEOUT to default to the value of
ZMQ_HEARTBEAT_IVL if it's not explicitly set.
Change the units of ZMQ_HEARTBEAT_TTL to milliseconds in the API
and round down to the nearest decisecond so that all the options
are using the same units.
Make the maximum heartbeat TTL match the spec (6553 seconds)
The shared reference count was not shared but copied. msg_t cannot
store the refcnt itsef but has to store a pointer to an externally
allocated (shared) refcnter. The changes to lmsg are reverted to
use content_t again. Howver, this introduces an allocation in v2_decoder
when creating the message which can be avoided. When allocating the reception
buffer, space is allocated for the maximum number of reference counts
(8192 / max_vsm_size = 8192/64 = 128 zmq:atomic_counter objects). This
increases the buffer by 128*sizeof(atomic_counter) = 128*4 = 512 bytes only.
When creating a message, the refcnt member is set to the address of one of the
pre-allocated atomic_counter_t objects. To do so, a new msg_t type zcmsg
is introduced because msg::copy must discriminate between the message types
when releasing memory.
zero-copy msg_t::init cannot be used when the message exceeds either
the buffer end or the last received byte. To detect this, the buffer
is now resized to the numnber of received bytes.
