YMQ I/O abstraction layer

[rafa-be]

I spent the past few days trying to find a common abstraction API for system calls (sockets, pipes, files ...) that works with Windows and Unixes non-blocking sockets (Epoll, kQueue, Windows's IOCP ...).

Reactive vs proactive events

The main issue, as pointed by @gxuu, is that Epoll, kQueue and select() are reactive, while IOCP is proactive:

// Epoll/kQueue/Select
//
// The user subscribes on FD events (read/write/close/accept/...), and the OS wakes up when the event occurs.

sock = socket();
poll_add(sock, READ);

auto event = poll_wait({sock});

if (event == READ) {
    buffer = read(sock);
}

// Windows's IOCP
//
// The user starts the async syscall, and the OS then notifies the user on completion

sock = socket();
sock.prepare_read(on_read_handle);

iocp_wait({on_read_handle});

buffer = sock.read();

That makes it hard to provide a common Socket and EventPool API, as the calling code would have to reverse the calling order of the respective calls:

// Epoll/kQueue/Select
pool.wait(sock);
sock.read();

// Proactive:
sock.prepare_read();
pool.wait(sock);
sock.read();

Current implementation

@gxuu solved this problem by requiring the user code to always call socket.prepare_read() before waiting on the event loop, with the call being a no-op on UNIXes.

While that works, it complicates the socket logic as, contrary to the simplified example, the call to prepare_read() might happen in a different iteration of the event loop (as we are handling multiple sockets).

The second issue is that the current event loop cannot notify which actual syscall completed (i.e. you don't know which prepare_read() event is associated with the event loop returned event). This creates weird code paths.

Compare this with Python event loop API (Boost's asio behaves similarly):

reader, writer = await asyncio.open_connection("127.0.0.1", 9000)
buffer = await reader.read(100)

Proposed changes

I'm proposing:

• to add an associated Transport class, provided by the EventLoop implementation and wrapping lower level sockets, pipe, file APIs. This class "wraps" basic I/O calls (read(), write() ...) inside the event loop. Python asyncio uses a similar class internally;
• use callbacks instead of return values for these I/O calls. For example void read(size_t n, std::function<void(ReadResult)>). That removes the need of a call to prepare_read(), and create a 1:1 matching for I/O calls and their respective results. Callbacks are functionally equivalent to asyncio/asio.

For example, usage would be:

EventLoop loop;

RawConnectionTCPFD sock;
EventLoop::Transport<RawConnectionTCPFD> asyncSock = loop.getTransport(sock); 
// (we might want to provide simpler API, e.g. `loop.getSocket()`, like Python asyncio does)

auto onReadCompleted = [](const std::span<bytes> buffer) {
    std::cout << "Read completed " << buffer << '\n'; 
};

asyncSock.read(1024, onReadCompleted);

loop.run();

[gxuu]

Is there a way to make the changes without changes to user interface? I don't like to introduce this loop.run now. Here's the reason:

• We are to have C++ implementation of scheduler which requires some form of async execution. So the scheduler launches a coroutine task loop. If we have this loop.run here, which is blocking, I am not sure how the timeout manager will interact with the scheduler.

I have been thinking about this scheduler async model for a while. The idea is that the scheduler will have something like this:

class AsyncContext {
  void spawn(); // that spawns task, similar to what you have in create_async_loop for scaler
  void run(); // that executes loop.wait() in python
  ...// of course, there will be other ops.
}

Meanwhile, calls to ymq will be transformed to coroutine. This way, the structure will look just like what we currently have in Python.

With the loop.run being a blocking call I am not sure whether it will integrate well with this idea. I specifically did not go with this loop.run approach when designing ymq in the user interface even though that what asio does because of this reason.

Though we provide executeAt in YMQ, I don't want to expose this to the user because of potential lifetime issues.

[rafa-be]

Is there a way to make the changes without changes to user interface? I don't like to introduce this loop.run now. Here's the reason:

* We are to have C++ implementation of `scheduler` which requires some form of async execution. So the `scheduler` launches a coroutine task loop. If we have this `loop.run` here, which is blocking, I am not sure how the timeout manager will interact with the scheduler.

Oh I'm not planning to change the current event loop semantic much. The examples above were simplified a lot, and the call to EventLoop::run() will still occur inside the event loop thread.

[gxuu]

Oh. I thought this is an API change as I interpreted "usage" as the user(e.g. scaler)'s usage. Then that part looks fine to me.

[rafa-be]

As a side note, libuv provides a very similar loop + callback API, although it's C based:
https://docs.libuv.org/en/v1.x/guide/introduction.html

That makes me wondering if we shouldn't just use libuv, as:

• it's fully cross-platform. All our platform specific code (#ifdef, CMake conditions ...) will be gone;
• it provides cross-platform TCP, Pipe/local sockets and file abstractions;
• it allows the event loop to be interrupted from another thread (required by PyYMQ);
• it supports scheduled tasks (required for reconnect);
• it supports synchronization primitives. That might help us solving the issue in removeIOSocket().
• it's does not seem to add much overhead, as the library is fairly simple.

The only drawback I see is that it is written in C. There are C++ wrappers, but I couldn't asses the quality, overhead and documentation of these.

@gxuu @magniloquency @sharpener6 Thoughts?

[gxuu]

We technically can. However, I don't think the benefit would pay off the cost we make the changes.

There are some downsides making this change:

• We have no primary control over libuv. That's actually the same problem we have for zmq. We could have maintain a fork ourselves but we chose not to.
• Extra development time as we need to adopt the library.
• No control to changes from upstream.
• Perhaps harder integration to coroutine because I don't have experience with C++ coroutine integration with C code.

it's fully cross-platform. All our platform specific code (#ifdef, CMake conditions ...) will be gone;
it provides cross-platform TCP, Pipe/local sockets and file abstractions;

These points above are not my concerns. As:

• Cross platform specific code doesn't go away. It's just not maintained by us. I'm okay with moving the whole YMQ out to be a separate project and I will still maintain it should we find this to be a better approach.
• We have these abstractions as well. Currently we don't have file abstraction but it's relatively easy to add into YMQ.

it allows the event loop to be interrupted from another thread (required by PyYMQ);

The point is also do-able in ymq as it's not real interrupt.

• one can do executeNow(...) to wake up the event loop and let it execute a function. This change can be done within 30 minutes.

it supports scheduled tasks (required for reconnect);

The fourth point is already addressed in ymq:

• as we already have the reconnect utility.
• The hard problem of reconnect is for one end to identify the reason of a connection being broken. I don't know if they addresses this problem.

P.S. There's an issue about removeIOSocket. When writing the issue, I discovered that a timeout mechanism is needed anyway and this "waiting remote end to close connection" approach is not really feasible due to the fact that network can go down any point of execution.

[rafa-be]

Sorry, only noticed your answer now :-(.

• We have no primary control over libuv. That's actually the same problem we have for zmq. We could have maintain a fork ourselves but we chose not to.
• Extra development time as we need to adopt the library.
• No control to changes from upstream.

True. Although libuv is way simpler than ZMQ.

The point is also do-able in ymq as it's not real interrupt.

• one can do executeNow(...) to wake up the event loop and let it execute a function. This change can be done within 30 minutes.
  The fourth point is already addressed in ymq:
• as we already have the reconnect utility.

True, but I was pointing out that libuv supports all these things that YMQ requires. My point was that YMQ could rely libuv.

• The hard problem of reconnect is for one end to identify the reason of a connection being broken. I don't know if they addresses this problem.

Yes. Their socket abstraction can raise errors like UV_ECONNREFUSED, UV_ECONNRESET, UV_ETIMEDOUT... These match POSIX errors more or less 1:1.

[gxuu]

Another problem to achieve a common abstraction for all the sockets is that on Windows server and client FD (the ones fed into acceptEx and connectEx) requires extra memory that are not needed for connections. I don't want to have these extra bytes for connections, that's why there are three handles instead of one. I am not sure how this abstraction is going to resolve that.

[magniloquency]

I think this is going in the right direction. I agree that the code path for receiving an event on Windows is strange, it feels a bit like putting a round peg in a round hole.

I'm not sure that I totally understand the purpose / shape of this Transport class vs. polymorphism or a private internal impl pointer, but it does make sense that each event loop type would need custom logic around these ops, so it makes sense that something is required.

I'm not too familiar with libuv but I have heard a lot about it and after a brief look, I think I like it. I do feel like we are solving a bit of an already solved problem, and using a library such as libuv or similar could help us a lot. The usage seems quite simple, you init a loop, then you make requests with a callback. It feels similar to the api we have now. I am in favour of investigating integrating this.

[rafa-be]

putting a round peg in a round hole.

More like the opposite 😂

I'm not sure that I totally understand the purpose / shape of this Transport class vs. polymorphism or a private internal impl pointer, but it does make sense that each event loop type would need custom logic around these ops, so it makes sense that something is required.

These are equivalent.

The idea (snatched from Python's asyncio) is that the Transport class abstracts how I/O are interacting with the event loop, while the "raw" socket, file, pipe classes do the actual I/O.

That's not easy to explain, but think like this:

• RawConnectionTCPFD objects provide basic cross-platform syscalls.
  For example:

  std::optional<ErrorCode> RawConnectionTCPFD::write(Buffer)

• Transport<RawConnectionTCPFD> add the "reactive" API (that uses the event loop) on top of this class.
  For example:

  void Transport<RawConnectionTCPFD>::write(Buffer, OnWriteCallback)

  The transport objects are the ones responsible for registering the FDs to the event loop, and reacting to events.

It's slighly more complex in practice, mostly because of server sockets. But that's the general idea.