Problem: not all sockets support ZMQ_FD

[minrk]

Related to: https://github.com/zeromq/libzmq/issues/1434

To integrate well with existing eventloops, zmq sockets must be pollable via a system select/poll mechanism without using zmq_poll, because forcing the replacement the inner-loop implementation of an eventloop in order to use zmq sockets is generally not advisable, and not always available. ZMQ_FD is the mechanism for this, but it seems that some sockets, such as the new threadsafe variants and inproc connections, lack this feature (https://github.com/zeromq/pyzmq/issues/1139). This means that those sockets cannot be integrated into any application with an eventloop without resorting to periodic polling. At the most basic level, all sockets and all transports should support the existing ZMQ_FD.

An alternate solution that would solve longstanding issued with exposing the edge-triggered FD would be a new, traditional level-triggered FD API, with as many of the following features as achievable:

1. it's level-triggered
2. it works for all sockets and all transports
3. it's not an internal implementation detail that is exposed to users, but an explicit user API
4. either support read/write signaling on one FD or separate read-only FDs for read and write
5. closing the FD results in closing the socket, not crashing the process (I don't know how feasible that is)

Minimal test code / Steps to reproduce the issue

import zmq
s = zmq.Context().socket(zmq.DISH)
s.FD

What's the actual result? (include assertion message & call stack if applicable)

EINVAL

What's the expected result?

an FD I can use to integrate with select.

[bluca]

IIRC it was an explicit design decision to avoid exposing an FD for polling for the thread safe socket types, but can't remember much - @somdoron do you remember the story there?

[minrk]

I'm totally okay with not supporting the existing FD API if there's a problem with it. However, that doesn't make it okay for these sockets to lack the ability to integrate with eventloops. It just means that there should put even more pressure to create an alternative FD API for eventloop integration, because until then the draft sockets simply cannot be used in many applications.

[bluca]

That sounds like a reasonable request to me. AFAIK the intention was to use the new socket poller API, but I don't know if it can be used integrated in an event look - I'm afraid I haven't followed the development closely enough to be able to say.

[somdoron]

you can use them with zmq_poller.

Also you can use them with other event loops api, however it is a little more complicated. We might have to expose a nicer api for that.

[minrk]

zmq_poller can be used with an existing eventloop? Note that blocking polls other than the eventloop's own are not allowed, so I need to wake an existing system select (kqueue/epoll/uvloop/etc.) call when an event arrives on a zmq socket. How can I get an FD that can be used in this way from the DISH/RADIO sockets?

[somdoron]

Yes, so we need a function to get the zmq_poller FD. which is the FD of the zmq_poller signaler, from here:

https://github.com/zeromq/libzmq/blob/master/src/socket_poller.hpp#L106

We have an optimization now that only create the signaler if we have a thread safe socket in the poller (like DISH or RADIO).

Anyway, add that FD to your eventloop and when signaled call zmq_poller_wait with zero timeout.

[minrk]

ok, so at the pyzmq level I can mimic the existing FD behavior on sockets by:

1. instantiating a poller per socket
2. aliasing the FD on these sockets to to the signaler on the poller

right?

Is the signaler only for the draft sockets or will it be a polling API that works for all sockets?

[somdoron]

Yes.

The signaler is only for the draft (thread-safe) sockets.

You also need to make a PR to libzmq that expose the zmq_poller FD (zmq_poller_fd).

[garlick]

it seems that some sockets, such as the new threadsafe variants and inproc connections, lack this feature

inproc connections at least used to work with ZMQ_FD / ZMQ_EVENTS - see discussion in #1434. If that is changing, I will have to do something about it as my application depends heavily on inproc, ipc, and tcp sockets integrated with libev.

Ideally I'd like to support the new socket types too, so thank you for raising this issue @minrk.

[minrk]

@somdoron thanks for the info! I'll wait to support the draft sockets until there's an API that works for all zmq sockets.

@garlick my comment about inproc was based on my reading of that Issue that there have been problems with inproc + ZMQ_FD, I could be wrong! I don't think the situation has changed, so maybe #1434 was just another misunderstanding of edge-triggered FD behavior and it's only the draft sockets that don't support eventloops.

[garlick]

@minrk - thanks for that clarification. My experience is that inproc works fine with ZMQ_FD at least with libzmq-4.2.2 and earlier.

[NoAnyLove]

Yes, so we need a function to get the zmq_poller FD. which is the FD of the zmq_poller signaler, from here: https://github.com/zeromq/libzmq/blob/master/src/socket_poller.hpp#L106 We have an optimization now that only create the signaler if we have a thread safe socket in the poller (like DISH or RADIO). Anyway, add that FD to your eventloop and when signaled call zmq_poller_wait with zero timeout.

Associating a thread-safe socket with a zmq_poller is inconvenient and not straightforward, it might also degrade the performance according to #3487.

I have a quick scan of the code, please correct me if I'm wrong. zmq::socket_poller_t basically adds its internal signaler to the thread-safe socket when we register the socket to the zmq poller (zmq::socket_poller_t::add), and the thread-safe socket adds the signaler to its mailbox_safe_t (zmq::socket_base_t::add_signaler). In that case, why don't we just support the ZMQ_FD option in thread-safe socket by adding an internal signaler for thread-safe socket, for example,

int zmq::socket_base_t::getsockopt (int option_,
                                    void *optval_,
                                    size_t *optvallen_)
{  
    // omit some lines
    if (option_ == ZMQ_FD) {
        if (_thread_safe) {
            if (_signaler == NULL) {
                _signaler = new (std::nothrow) signaler_t ();
                // error handling
                add_signaler(_signaler);
            }
           return do_getsockopt<fd_t> (optval_, optvallen_, _signaler->get_fd ());
        }

        return do_getsockopt<fd_t> (
          optval_, optvallen_,
          (static_cast<mailbox_t *> (_mailbox))->get_fd ());
    }
    // omit some lines
}

It also allow us to support async interfaces for thread-safe sockets without too much efforts. e.g., it requires almost no code change in pyzmq to support https://github.com/zeromq/pyzmq/issues/1139.