kqueue.cpp 6.12 KB
Newer Older
Martin Sustrik's avatar
Martin Sustrik committed
1
/*
2
    Copyright (c) 2007-2016 Contributors as noted in the AUTHORS file
Martin Sustrik's avatar
Martin Sustrik committed
3

4
    This file is part of libzmq, the ZeroMQ core engine in C++.
Martin Sustrik's avatar
Martin Sustrik committed
5

6 7 8
    libzmq is free software; you can redistribute it and/or modify it under
    the terms of the GNU Lesser General Public License (LGPL) as published
    by the Free Software Foundation; either version 3 of the License, or
Martin Sustrik's avatar
Martin Sustrik committed
9 10
    (at your option) any later version.

11 12 13 14 15 16 17 18 19 20 21 22 23 24
    As a special exception, the Contributors give you permission to link
    this library with independent modules to produce an executable,
    regardless of the license terms of these independent modules, and to
    copy and distribute the resulting executable under terms of your choice,
    provided that you also meet, for each linked independent module, the
    terms and conditions of the license of that module. An independent
    module is a module which is not derived from or based on this library.
    If you modify this library, you must extend this exception to your
    version of the library.

    libzmq is distributed in the hope that it will be useful, but WITHOUT
    ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
    FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public
    License for more details.
Martin Sustrik's avatar
Martin Sustrik committed
25

26
    You should have received a copy of the GNU Lesser General Public License
Martin Sustrik's avatar
Martin Sustrik committed
27 28 29
    along with this program.  If not, see <http://www.gnu.org/licenses/>.
*/

30
#include "precompiled.hpp"
31
#include "kqueue.hpp"
32
#if defined ZMQ_IOTHREAD_POLLER_USE_KQUEUE
Martin Sustrik's avatar
Martin Sustrik committed
33 34 35 36 37 38 39

#include <sys/time.h>
#include <sys/types.h>
#include <sys/event.h>
#include <stdlib.h>
#include <unistd.h>
#include <algorithm>
40
#include <new>
Martin Sustrik's avatar
Martin Sustrik committed
41

42
#include "macros.hpp"
Martin Sustrik's avatar
Martin Sustrik committed
43 44 45 46
#include "kqueue.hpp"
#include "err.hpp"
#include "config.hpp"
#include "i_poll_events.hpp"
47
#include "likely.hpp"
Martin Sustrik's avatar
Martin Sustrik committed
48

Martin Lucina's avatar
Martin Lucina committed
49 50 51 52 53 54 55 56
//  NetBSD defines (struct kevent).udata as intptr_t, everyone else
//  as void *.
#if defined ZMQ_HAVE_NETBSD
#define kevent_udata_t intptr_t
#else
#define kevent_udata_t void *
#endif

57
zmq::kqueue_t::kqueue_t (const zmq::thread_ctx_t &ctx_) :
58
    worker_poller_base_t (ctx_)
Martin Sustrik's avatar
Martin Sustrik committed
59 60 61 62
{
    //  Create event queue
    kqueue_fd = kqueue ();
    errno_assert (kqueue_fd != -1);
63
#ifdef HAVE_FORK
64
    pid = getpid ();
65
#endif
Martin Sustrik's avatar
Martin Sustrik committed
66 67
}

Martin Sustrik's avatar
Martin Sustrik committed
68
zmq::kqueue_t::~kqueue_t ()
Martin Sustrik's avatar
Martin Sustrik committed
69
{
70
    stop_worker ();
Martin Sustrik's avatar
Martin Sustrik committed
71 72 73
    close (kqueue_fd);
}

Martin Sustrik's avatar
Martin Sustrik committed
74
void zmq::kqueue_t::kevent_add (fd_t fd_, short filter_, void *udata_)
Martin Sustrik's avatar
Martin Sustrik committed
75
{
76
    check_thread ();
Martin Sustrik's avatar
Martin Sustrik committed
77 78
    struct kevent ev;

79
    EV_SET (&ev, fd_, filter_, EV_ADD, 0, 0, (kevent_udata_t) udata_);
Martin Sustrik's avatar
Martin Sustrik committed
80 81 82 83
    int rc = kevent (kqueue_fd, &ev, 1, NULL, 0, NULL);
    errno_assert (rc != -1);
}

Martin Sustrik's avatar
Martin Sustrik committed
84
void zmq::kqueue_t::kevent_delete (fd_t fd_, short filter_)
Martin Sustrik's avatar
Martin Sustrik committed
85 86 87
{
    struct kevent ev;

88
    EV_SET (&ev, fd_, filter_, EV_DELETE, 0, 0, 0);
Martin Sustrik's avatar
Martin Sustrik committed
89 90 91 92
    int rc = kevent (kqueue_fd, &ev, 1, NULL, 0, NULL);
    errno_assert (rc != -1);
}

93
zmq::kqueue_t::handle_t zmq::kqueue_t::add_fd (fd_t fd_,
94
                                               i_poll_events *reactor_)
Martin Sustrik's avatar
Martin Sustrik committed
95
{
96
    check_thread ();
97
    poll_entry_t *pe = new (std::nothrow) poll_entry_t;
98
    alloc_assert (pe);
Martin Sustrik's avatar
Martin Sustrik committed
99 100 101 102 103 104

    pe->fd = fd_;
    pe->flag_pollin = 0;
    pe->flag_pollout = 0;
    pe->reactor = reactor_;

105 106
    adjust_load (1);

107
    return pe;
Martin Sustrik's avatar
Martin Sustrik committed
108 109
}

Martin Sustrik's avatar
Martin Sustrik committed
110
void zmq::kqueue_t::rm_fd (handle_t handle_)
Martin Sustrik's avatar
Martin Sustrik committed
111
{
112
    check_thread ();
113
    poll_entry_t *pe = (poll_entry_t *) handle_;
Martin Sustrik's avatar
Martin Sustrik committed
114 115 116 117 118 119
    if (pe->flag_pollin)
        kevent_delete (pe->fd, EVFILT_READ);
    if (pe->flag_pollout)
        kevent_delete (pe->fd, EVFILT_WRITE);
    pe->fd = retired_fd;
    retired.push_back (pe);
120 121

    adjust_load (-1);
Martin Sustrik's avatar
Martin Sustrik committed
122 123
}

Martin Sustrik's avatar
Martin Sustrik committed
124
void zmq::kqueue_t::set_pollin (handle_t handle_)
Martin Sustrik's avatar
Martin Sustrik committed
125
{
126
    check_thread ();
127
    poll_entry_t *pe = (poll_entry_t *) handle_;
128 129 130 131
    if (likely (!pe->flag_pollin)) {
        pe->flag_pollin = true;
        kevent_add (pe->fd, EVFILT_READ, pe);
    }
Martin Sustrik's avatar
Martin Sustrik committed
132 133
}

Martin Sustrik's avatar
Martin Sustrik committed
134
void zmq::kqueue_t::reset_pollin (handle_t handle_)
Martin Sustrik's avatar
Martin Sustrik committed
135
{
136
    check_thread ();
137
    poll_entry_t *pe = (poll_entry_t *) handle_;
138 139 140 141
    if (likely (pe->flag_pollin)) {
        pe->flag_pollin = false;
        kevent_delete (pe->fd, EVFILT_READ);
    }
Martin Sustrik's avatar
Martin Sustrik committed
142 143
}

Martin Sustrik's avatar
Martin Sustrik committed
144
void zmq::kqueue_t::set_pollout (handle_t handle_)
Martin Sustrik's avatar
Martin Sustrik committed
145
{
146
    check_thread ();
147
    poll_entry_t *pe = (poll_entry_t *) handle_;
148 149 150 151
    if (likely (!pe->flag_pollout)) {
        pe->flag_pollout = true;
        kevent_add (pe->fd, EVFILT_WRITE, pe);
    }
Martin Sustrik's avatar
Martin Sustrik committed
152 153
}

Martin Sustrik's avatar
Martin Sustrik committed
154
void zmq::kqueue_t::reset_pollout (handle_t handle_)
Martin Sustrik's avatar
Martin Sustrik committed
155
{
156
    check_thread ();
157
    poll_entry_t *pe = (poll_entry_t *) handle_;
158 159 160
    if (likely (pe->flag_pollout)) {
        pe->flag_pollout = false;
        kevent_delete (pe->fd, EVFILT_WRITE);
161
    }
Martin Sustrik's avatar
Martin Sustrik committed
162 163
}

Martin Sustrik's avatar
Martin Sustrik committed
164
void zmq::kqueue_t::stop ()
Martin Sustrik's avatar
Martin Sustrik committed
165 166 167
{
}

Richard Newton's avatar
Richard Newton committed
168
int zmq::kqueue_t::max_fds ()
169 170 171 172
{
    return -1;
}

Martin Sustrik's avatar
Martin Sustrik committed
173
void zmq::kqueue_t::loop ()
Martin Sustrik's avatar
Martin Sustrik committed
174
{
175
    while (true) {
176
        //  Execute any due timers.
177
        int timeout = (int) execute_timers ();
Martin Sustrik's avatar
Martin Sustrik committed
178

179 180 181 182 183 184 185 186
        if (get_load () == 0) {
            if (timeout == 0)
                break;

            // TODO sleep for timeout
            continue;
        }

Martin Sustrik's avatar
Martin Sustrik committed
187
        //  Wait for events.
188
        struct kevent ev_buf[max_io_events];
189
        timespec ts = {timeout / 1000, (timeout % 1000) * 1000000};
190 191
        int n = kevent (kqueue_fd, NULL, 0, &ev_buf[0], max_io_events,
                        timeout ? &ts : NULL);
192
#ifdef HAVE_FORK
193
        if (unlikely (pid != getpid ())) {
194 195 196 197 198
            //printf("zmq::kqueue_t::loop aborting on forked child %d\n", (int)getpid());
            // simply exit the loop in a forked process.
            return;
        }
#endif
Martin Hurton's avatar
Martin Hurton committed
199 200
        if (n == -1) {
            errno_assert (errno == EINTR);
Martin Sustrik's avatar
Martin Sustrik committed
201
            continue;
Martin Hurton's avatar
Martin Hurton committed
202
        }
Martin Sustrik's avatar
Martin Sustrik committed
203

204 205
        for (int i = 0; i < n; i++) {
            poll_entry_t *pe = (poll_entry_t *) ev_buf[i].udata;
Martin Sustrik's avatar
Martin Sustrik committed
206 207 208

            if (pe->fd == retired_fd)
                continue;
209
            if (ev_buf[i].flags & EV_EOF)
Martin Sustrik's avatar
Martin Sustrik committed
210 211 212
                pe->reactor->in_event ();
            if (pe->fd == retired_fd)
                continue;
213
            if (ev_buf[i].filter == EVFILT_WRITE)
Martin Sustrik's avatar
Martin Sustrik committed
214 215 216
                pe->reactor->out_event ();
            if (pe->fd == retired_fd)
                continue;
217
            if (ev_buf[i].filter == EVFILT_READ)
Martin Sustrik's avatar
Martin Sustrik committed
218 219 220 221
                pe->reactor->in_event ();
        }

        //  Destroy retired event sources.
222 223 224
        for (retired_t::iterator it = retired.begin (); it != retired.end ();
             ++it) {
            LIBZMQ_DELETE (*it);
225
        }
Martin Sustrik's avatar
Martin Sustrik committed
226 227 228 229 230
        retired.clear ();
    }
}

#endif