This fixes a hang in the epoll code that triggers on heavy, bursty
connections (such as the live SPNS APNs notifier).
It turns out that side-effects of processing our sockets could leave
other sockets (that we processed earlier in the loop) in a
needs-attention state which we might not notice if we go back to
epoll_wait right away. zmq::poll apparently takes care of this (and so
is safe to re-poll even in this state), but when we are using epoll we
need to worry about it by always checking for zmq events (which itself
has side effects) and, if we get any, re-enter the loop body immediately
*without* polling to deal with them.
Each call to zmq::poll is painfully slow when we have many open zmq
sockets, such as when we have 1800 outbound connections (i.e. connected
to every other service node, as services nodes might have sometimes and
the Session push notification server *always* has).
In testing on my local Ryzen 5950 system each time we go back to
zmq::poll incurs about 1.5ms of (mostly system) CPU time with 2000 open
outbound sockets, and so if we're being pelted with a nearly constant
stream of requests (such as happens with the Session push notification
server) we incur massive CPU costs every time we finish processing
messages and go back to wait (via zmq::poll) for more.
In testing a simple ZMQ (no OxenMQ) client/server that establishes 2000
connections to a server, and then has the server send a message back on
a random connection every 1ms, we get atrocious CPU usage: the proxy
thread spends a constant 100% CPU time. Virtually all of this is in the
poll call itself, though, so we aren't really bottlenecked by how much
can go through the proxy thread: in such a scenario the poll call uses
its CPU then returns right away, we process the queue of messages, and
return to another poll call. If we have lots of messages received in
that time, though (because messages are coming fast and the poll was
slow) then we process a lot all at once before going back to the poll,
so the main consequences here are that:
1) We use a huge amount of CPU
2) We introduce latency in a busy situation because the CPU has to make
the poll call (e.g. 1.5ms) before the next message can be processed.
3) If traffic is very bursty then the latency can manifest another
problem: in the time it takes to poll we could accumulate enough
incoming messages to overfill our internal per-category job queue,
which was happening in the SPNS.
(I also tested with 20k connections, and the poll time scaling was
linear: we still processed everything, but in larger chunks because
every poll call took about 15ms, and so we'd have about 15 messages at a
time to process with added latency of up to 15ms).
Switching to epoll *drastically* reduces the CPU usage in two ways:
1) It's massively faster by design: there's a single setup and
communication of all the polling details to the kernel which we only
have to do when our set of zmq sockets changes (which is relatively
rare).
2) We can further reduce CPU time because epoll tells us *which* sockets
need attention, and so if only 1 connection out of the 2000 sent us
something we can only bother checking that single socket for
messages. (In theory we can do the same with zmq::poll by querying
for events available on the socket, but in practice it doesn't
improve anything over just trying to read from them all).
In my straight zmq test script, using epoll instead reduced CPU usage in
the sends-every-1ms scenario from a constant pegged 100% of a core to an
average of 2-3% of a single core. (Moreover this CPU usage level didn't
noticeably change when using 20k connections instead of 2k).
This is probably slightly more efficient (as it avoids going through
uniform_int_distribution), but more importantly, won't trigger some of
Apple's new xcode buggy crap.
There's a very rare race condition where a tagged thread doesn't seem to
exist when the proxy tries syncing startup with them, and so the proxy
thread hangs in startup.
This addresses it by avoiding looking at the `proxy_thread` variable
(which probably isn't thread safe) in the worker's startup, and
signalling the you-need-to-shutdown condition via a third option for the
(formerly boolean) `tagged_go`.
MAX_SOCKETS wasn't working properly because ZMQ uses it when the context
is initialized, which happens when the first socket is constructed on
that context.
For OxenMQ, we had several sockets constructed on the context during
OxenMQ construction, which meant the context_t was being initialized
during OxenMQ construction, rather than during start(), and so setting
MAX_SOCKETS would have no effect and you'd always get the default.
This fixes it by making all the member variable zmq::socket_t's
default-constructed, then replacing them with proper zmq::socket_t's
during startup() so that we also defer zmq::context_t initialization to
the right place.
A second issue found during testing (also fixed here) is that the socket
worker threads use to communicate to the proxy could fail if the worker
socket creation would violate the zmq max sockets limit, which wound up
throwing an uncaught exception and aborting. This pre-initializes (but
doesn't connect) all potential worker threads sockets during start() so
that the lazily-initialized worker thread will have one already set up
rather than having to create a new one (which could fail).
We can just leave the dangling pointer value in the `run` object: even
though we just deleted it, there's no need to reset this value because
it will never be used again. (And even if we did, we don't check
against nullptr anyway so having a nullptr here doesn't make anything
safter than a dangling pointer).
The assignment (into the variant) uses a small amount of CPU (via
std::variant), so better for performance to just leave it dangling.
Internal messages (control messages, worker messages) are always 3 parts
or less, so we can optimize by using a stack allocated std::array for
those cases rather than needing to continually clear and expand a heap
allocated vector.
Change the internal worker routing id to be "w" followed by the raw
integer bytes, so that we can just memcpy them into a uint32_t rather
than needing to do str -> integer conversion on each received worker
message.
(This also eliminates a vestigal call into oxenc internals).
I don't know what this set was originally meant to be doing, but it
currently does nothing (except adding overhead).
The comment says it "owns" the instances but that isn't really true; the
instances effectively already manage themselves as they pass the pointer
through the communications between proxy and workers.
This adds a much simpler `Job` implementation of `Batch` that is used
for simple no-return, no-completion jobs (as are initiated via
`omq.job(...)`).
This reduces the overhead involved in constructing/destroying the Batch
instance for these common jobs.
bt_*, hex, base32z, base64 all moved to oxen-encoding a while ago; this
finishes the move by removing them from oxenmq and instead making oxenmq
depend on oxen-encoding.
libzmq's IPv6 support is buggy when also using DNS hostname: in
particular, if you try to connect to a DNS name that has an IPv6
address, then zmq will *only* try an IPv6 connection, even if the local
client has no IPv6 connectivity, and even if the remote is only
listening on its IPv4 address.
This is much too unreliable to enable by default.
Currently if the proxy thread fails to start (typically because a bind
fails) the exception happens in the proxy thread which is uncatchable by
the caller (and aborts the program).
This makes it nicer by transporting startup exceptions back to the
start() call.
Currently if you pass a nullptr for Logger you get a random
std::bad_function_call called from some random thread the first time a
log message goes out.
This fixes it allow a nullptr that logs nothing.
Makes some send/connection options more robust to "do nothing" runtime
value, which the Python wrapper needs.
Also found a bunch of doc typos and fixes.
Bump version to 1.2.8 so that new pyoxenmq can build-depend on it.
This allows for on-the-fly encoding/decoding, and also allows for
on-the-fly transcoding between types without needing intermediate string
allocations (see added test cases for examples).
- Add {to,from}_{base64,base32z,hex}_size functions to calculate the
resulting output size from a given input size.
- Use it internally
- Make b32z and b64 validity checking slightly stricter: currently we
"accept" some b32z and b64 strings that contain an extra character
that leave us with 5-7 trailing bits (base32z) or 6 trailing bits
(base64). We simply ignore the extra one if decoding, but we
shouldn't accept it in the "is valid" calls.
Changes the 3-iterator versions of to_hex, from_b32z, etc. to return the
final output iterator, which allows for much easier in-place "from"
conversion without needing a new string by doing something like:
std::string data = /* some hex */;
auto end = oxenmq::from_hex(data.begin(), data.end(), data.begin();
data.erase(end, data.end());
Returning from the "to" converters is a bit less useful but doing it
anyway for consistency (and because it could still have some use, e.g.
if output is into some fixed buffer it lets you determine how much was
written).
inproc support is special in zmq: in particular it completely bypasses
the auth layer, which causes problems in OxenMQ because we assume that a
message will always have auth information (set during initial connection
handshake).
This adds an "always-on" inproc listener and adds a new `connect_inproc`
method for a caller to establish a connection to it.
It also throws exceptions if you try to `listen_plain` or `listen_curve`
on an inproc address, because that won't work for the reasons detailed
above.
The recent PR that revamped the connection IDs missed a case when
connecting to service nodes where we store the SN pubkey in peers, but
then fail to find the peer when we look it up by connection id.
This adds the required tracking to fix that case (and adds a test that
fails without the fix here).
Jason Rhinelander
Thomas Winget
Sean Darcy
Jeff Becker