dispatch-1.0/src/queue_kevent.c

   1 /*
   2  * Copyright (c) 2008-2009 Apple Inc. All rights reserved.
   3  *
   4  * @APPLE_APACHE_LICENSE_HEADER_START@
   5  *
   6  * Licensed under the Apache License, Version 2.0 (the "License");
   7  * you may not use this file except in compliance with the License.
   8  * You may obtain a copy of the License at
   9  *
  10  *     http://www.apache.org/licenses/LICENSE-2.0
  11  *
  12  * Unless required by applicable law or agreed to in writing, software
  13  * distributed under the License is distributed on an "AS IS" BASIS,
  14  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15  * See the License for the specific language governing permissions and
  16  * limitations under the License.
  17  *
  18  * @APPLE_APACHE_LICENSE_HEADER_END@
  19  */
  20
  21 #include "internal.h"
  22 #include "kevent_internal.h"
  23
  24
  25 static bool _dispatch_select_workaround;
  26 static fd_set _dispatch_rfds;
  27 static fd_set _dispatch_wfds;
  28 static void *_dispatch_rfd_ptrs[FD_SETSIZE];
  29 static void *_dispatch_wfd_ptrs[FD_SETSIZE];
  30
  31 static int _dispatch_kq;
  32
  33 static void
  34 _dispatch_get_kq_init(void *context __attribute__((unused)))
  35 {
  36         static const struct kevent kev = {
  37                 .ident = 1,
  38                 .filter = EVFILT_USER,
  39                 .flags = EV_ADD|EV_CLEAR,
  40         };
  41
  42         _dispatch_kq = kqueue();
  43         _dispatch_safe_fork = false;
  44         // in case we fall back to select()
  45         FD_SET(_dispatch_kq, &_dispatch_rfds);
  46
  47         if (_dispatch_kq == -1) {
  48                 dispatch_assert_zero(errno);
  49         }
  50
  51         (void)dispatch_assume_zero(kevent(_dispatch_kq, &kev, 1, NULL, 0,
  52             NULL));
  53
  54         _dispatch_queue_push(_dispatch_mgr_q.do_targetq, &_dispatch_mgr_q);
  55 }
  56
  57 static int
  58 _dispatch_get_kq(void)
  59 {
  60         static dispatch_once_t pred;
  61
  62         dispatch_once_f(&pred, NULL, _dispatch_get_kq_init);
  63
  64         return _dispatch_kq;
  65 }
  66
  67 static void
  68 _dispatch_mgr_thread2(struct kevent *kev, size_t cnt)
  69 {
  70         size_t i;
  71
  72         for (i = 0; i < cnt; i++) {
  73                 // EVFILT_USER isn't used by sources
  74                 if (kev[i].filter == EVFILT_USER) {
  75                         // If _dispatch_mgr_thread2() ever is changed to return to the
  76                         // caller, then this should become _dispatch_queue_drain()
  77                         _dispatch_queue_serial_drain_till_empty(&_dispatch_mgr_q);
  78                 } else {
  79                         _dispatch_source_drain_kevent(&kev[i]);
  80                 }
  81     }
  82 }
  83
  84 static dispatch_queue_t
  85 _dispatch_mgr_invoke(dispatch_queue_t dq)
  86 {
  87         static const struct timespec timeout_immediately = { 0, 0 };
  88         struct timespec timeout;
  89         const struct timespec *timeoutp;
  90         struct timeval sel_timeout, *sel_timeoutp;
  91         fd_set tmp_rfds, tmp_wfds;
  92         struct kevent kev[1];
  93         int k_cnt, k_err, i, r;
  94
  95         _dispatch_thread_setspecific(dispatch_queue_key, dq);
  96
  97         for (;;) {
  98                 _dispatch_run_timers();
  99
 100                 timeoutp = _dispatch_get_next_timer_fire(&timeout);
 101
 102                 if (_dispatch_select_workaround) {
 103                         FD_COPY(&_dispatch_rfds, &tmp_rfds);
 104                         FD_COPY(&_dispatch_wfds, &tmp_wfds);
 105                         if (timeoutp) {
 106                                 sel_timeout.tv_sec = timeoutp->tv_sec;
 107                                 sel_timeout.tv_usec = (typeof(sel_timeout.tv_usec))(timeoutp->tv_nsec / 1000u);
 108                                 sel_timeoutp = &sel_timeout;
 109                         } else {
 110                                 sel_timeoutp = NULL;
 111                         }
 112
 113                         r = select(FD_SETSIZE, &tmp_rfds, &tmp_wfds, NULL, sel_timeoutp);
 114                         if (r == -1) {
 115                                 if (errno != EBADF) {
 116                                         (void)dispatch_assume_zero(errno);
 117                                         continue;
 118                                 }
 119                                 for (i = 0; i < FD_SETSIZE; i++) {
 120                                         if (i == _dispatch_kq) {
 121                                                 continue;
 122                                         }
 123                                         if (!FD_ISSET(i, &_dispatch_rfds) && !FD_ISSET(i, &_dispatch_wfds)) {
 124                                                 continue;
 125                                         }
 126                                         r = dup(i);
 127                                         if (r != -1) {
 128                                                 close(r);
 129                                         } else {
 130                                                 FD_CLR(i, &_dispatch_rfds);
 131                                                 FD_CLR(i, &_dispatch_wfds);
 132                                                 _dispatch_rfd_ptrs[i] = 0;
 133                                                 _dispatch_wfd_ptrs[i] = 0;
 134                                         }
 135                                 }
 136                                 continue;
 137                         }
 138
 139                         if (r > 0) {
 140                                 for (i = 0; i < FD_SETSIZE; i++) {
 141                                         if (i == _dispatch_kq) {
 142                                                 continue;
 143                                         }
 144                                         if (FD_ISSET(i, &tmp_rfds)) {
 145                                                 FD_CLR(i, &_dispatch_rfds);     // emulate EV_DISABLE
 146                                                 EV_SET(&kev[0], i, EVFILT_READ, EV_ADD|EV_ENABLE|EV_DISPATCH, 0, 1, _dispatch_rfd_ptrs[i]);
 147                                                 _dispatch_rfd_ptrs[i] = 0;
 148                                                 _dispatch_mgr_thread2(kev, 1);
 149                                         }
 150                                         if (FD_ISSET(i, &tmp_wfds)) {
 151                                                 FD_CLR(i, &_dispatch_wfds);     // emulate EV_DISABLE
 152                                                 EV_SET(&kev[0], i, EVFILT_WRITE, EV_ADD|EV_ENABLE|EV_DISPATCH, 0, 1, _dispatch_wfd_ptrs[i]);
 153                                                 _dispatch_wfd_ptrs[i] = 0;
 154                                                 _dispatch_mgr_thread2(kev, 1);
 155                                         }
 156                                 }
 157                         }
 158
 159                         timeoutp = &timeout_immediately;
 160                 }
 161
 162                 k_cnt = kevent(_dispatch_kq, NULL, 0, kev, sizeof(kev) / sizeof(kev[0]), timeoutp);
 163                 k_err = errno;
 164
 165                 switch (k_cnt) {
 166                 case -1:
 167                         if (k_err == EBADF) {
 168                                 DISPATCH_CLIENT_CRASH("Do not close random Unix descriptors");
 169                         }
 170                         (void)dispatch_assume_zero(k_err);
 171                         continue;
 172                 default:
 173                         _dispatch_mgr_thread2(kev, (size_t)k_cnt);
 174                         // fall through
 175                 case 0:
 176                         _dispatch_force_cache_cleanup();
 177                         continue;
 178                 }
 179         }
 180
 181         return NULL;
 182 }
 183
 184 static bool
 185 _dispatch_mgr_wakeup(dispatch_queue_t dq)
 186 {
 187         static const struct kevent kev = {
 188                 .ident = 1,
 189                 .filter = EVFILT_USER,
 190                 .fflags = NOTE_TRIGGER,
 191         };
 192
 193         _dispatch_debug("waking up the _dispatch_mgr_q: %p", dq);
 194
 195         _dispatch_update_kq(&kev);
 196
 197         return false;
 198 }
 199
 200 void
 201 _dispatch_update_kq(const struct kevent *kev)
 202 {
 203         struct kevent kev_copy = *kev;
 204         kev_copy.flags |= EV_RECEIPT;
 205
 206         if (kev_copy.flags & EV_DELETE) {
 207                 switch (kev_copy.filter) {
 208                 case EVFILT_READ:
 209                         if (FD_ISSET((int)kev_copy.ident, &_dispatch_rfds)) {
 210                                 FD_CLR((int)kev_copy.ident, &_dispatch_rfds);
 211                                 _dispatch_rfd_ptrs[kev_copy.ident] = 0;
 212                                 return;
 213                         }
 214                 case EVFILT_WRITE:
 215                         if (FD_ISSET((int)kev_copy.ident, &_dispatch_wfds)) {
 216                                 FD_CLR((int)kev_copy.ident, &_dispatch_wfds);
 217                                 _dispatch_wfd_ptrs[kev_copy.ident] = 0;
 218                                 return;
 219                         }
 220                 default:
 221                         break;
 222                 }
 223         }
 224
 225         int rval = kevent(_dispatch_get_kq(), &kev_copy, 1, &kev_copy, 1, NULL);
 226         if (rval == -1) {
 227                 // If we fail to register with kevents, for other reasons aside from
 228                 // changelist elements.
 229                 (void)dispatch_assume_zero(errno);
 230                 //kev_copy.flags |= EV_ERROR;
 231                 //kev_copy.data = error;
 232                 return;
 233         }
 234
 235         // The following select workaround only applies to adding kevents
 236         if (!(kev->flags & EV_ADD)) {
 237                 return;
 238         }
 239
 240         switch (kev_copy.data) {
 241         case 0:
 242                 return;
 243         case EBADF:
 244                 break;
 245         default:
 246                 // If an error occurred while registering with kevent, and it was
 247                 // because of a kevent changelist processing && the kevent involved
 248                 // either doing a read or write, it would indicate we were trying
 249                 // to register a /dev/* port; fall back to select
 250                 switch (kev_copy.filter) {
 251                 case EVFILT_READ:
 252                         _dispatch_select_workaround = true;
 253                         FD_SET((int)kev_copy.ident, &_dispatch_rfds);
 254                         _dispatch_rfd_ptrs[kev_copy.ident] = kev_copy.udata;
 255                         break;
 256                 case EVFILT_WRITE:
 257                         _dispatch_select_workaround = true;
 258                         FD_SET((int)kev_copy.ident, &_dispatch_wfds);
 259                         _dispatch_wfd_ptrs[kev_copy.ident] = kev_copy.udata;
 260                         break;
 261                 default:
 262                         _dispatch_source_drain_kevent(&kev_copy);
 263                         break;
 264                 }
 265                 break;
 266         }
 267 }
 268
 269 static const struct dispatch_queue_vtable_s _dispatch_queue_mgr_vtable = {
 270         .do_type = DISPATCH_QUEUE_MGR_TYPE,
 271         .do_kind = "mgr-queue",
 272         .do_invoke = _dispatch_mgr_invoke,
 273         .do_debug = dispatch_queue_debug,
 274         .do_probe = _dispatch_mgr_wakeup,
 275 };
 276
 277 // 6618342 Contact the team that owns the Instrument DTrace probe before renaming this symbol
 278 struct dispatch_queue_s _dispatch_mgr_q = {
 279         .do_vtable = &_dispatch_queue_mgr_vtable,
 280         .do_ref_cnt = DISPATCH_OBJECT_GLOBAL_REFCNT,
 281         .do_xref_cnt = DISPATCH_OBJECT_GLOBAL_REFCNT,
 282         .do_suspend_cnt = DISPATCH_OBJECT_SUSPEND_LOCK,
 283         .do_targetq = &_dispatch_root_queues[DISPATCH_ROOT_QUEUE_COUNT - 1],
 284
 285         .dq_label = "com.apple.libdispatch-manager",
 286         .dq_width = 1,
 287         .dq_serialnum = 2,
 288 };
 289