mono/metadata/threadpool-ms-io.c

   1 /*
   2  * threadpool-ms-io.c: Microsoft IO threadpool runtime support
   3  *
   4  * Author:
   5  *      Ludovic Henry (ludovic.henry@xamarin.com)
   6  *
   7  * Copyright 2015 Xamarin, Inc (http://www.xamarin.com)
   8  */
   9
  10 #include <config.h>
  11
  12 #ifndef DISABLE_SOCKETS
  13
  14 #include <glib.h>
  15
  16 #if defined(HOST_WIN32)
  17 #include <windows.h>
  18 #else
  19 #include <errno.h>
  20 #include <fcntl.h>
  21 #endif
  22
  23 #if defined(HAVE_EPOLL)
  24 #include <sys/epoll.h>
  25 #elif defined(HAVE_KQUEUE)
  26 #include <sys/types.h>
  27 #include <sys/event.h>
  28 #include <sys/time.h>
  29 #endif
  30
  31 #include <mono/metadata/gc-internal.h>
  32 #include <mono/metadata/mono-mlist.h>
  33 #include <mono/metadata/threadpool-internals.h>
  34 #include <mono/metadata/threadpool-ms.h>
  35 #include <mono/metadata/threadpool-ms-io.h>
  36 #include <mono/utils/atomic.h>
  37 #include <mono/utils/mono-poll.h>
  38 #include <mono/utils/mono-threads.h>
  39
  40 /* Keep in sync with System.Net.Sockets.MonoSocketRuntimeWorkItem */
  41 struct _MonoSocketRuntimeWorkItem {
  42         MonoObject object;
  43         MonoSocketAsyncResult *socket_async_result;
  44 };
  45
  46 /* Keep in sync with System.Net.Sockets.Socket.SocketOperation */
  47 enum {
  48         AIO_OP_FIRST,
  49         AIO_OP_ACCEPT = 0,
  50         AIO_OP_CONNECT,
  51         AIO_OP_RECEIVE,
  52         AIO_OP_RECEIVEFROM,
  53         AIO_OP_SEND,
  54         AIO_OP_SENDTO,
  55         AIO_OP_RECV_JUST_CALLBACK,
  56         AIO_OP_SEND_JUST_CALLBACK,
  57         AIO_OP_READPIPE,
  58         AIO_OP_CONSOLE2,
  59         AIO_OP_DISCONNECT,
  60         AIO_OP_ACCEPTRECEIVE,
  61         AIO_OP_RECEIVE_BUFFERS,
  62         AIO_OP_SEND_BUFFERS,
  63         AIO_OP_LAST
  64 };
  65
  66 typedef enum {
  67         BACKEND_EPOLL,
  68         BACKEND_KQUEUE,
  69         BACKEND_POLL,
  70 } ThreadPoolIOBackend;
  71
  72 typedef struct {
  73         gint fd;
  74
  75         union {
  76 #if defined(HAVE_EPOLL)
  77                 struct {
  78                         struct epoll_event *event;
  79                         gint op;
  80                 } epoll;
  81 #elif defined(HAVE_KQUEUE)
  82                 struct {
  83                         struct kevent *event;
  84                 } kqueue;
  85 #endif
  86                 struct {
  87                         mono_pollfd fd;
  88                 } poll;
  89         };
  90 } ThreadPoolIOUpdate;
  91
  92 typedef struct {
  93         MonoGHashTable *states;
  94         mono_mutex_t states_lock;
  95
  96         ThreadPoolIOBackend backend;
  97
  98         ThreadPoolIOUpdate *updates;
  99         guint updates_size;
 100         mono_mutex_t updates_lock;
 101
 102 #if !defined(HOST_WIN32)
 103         gint wakeup_pipes [2];
 104 #else
 105         SOCKET wakeup_pipes [2];
 106 #endif
 107
 108         union {
 109 #if defined(HAVE_EPOLL)
 110                 struct {
 111                         gint fd;
 112                         struct epoll_event *events;
 113                 } epoll;
 114 #elif defined(HAVE_KQUEUE)
 115                 struct {
 116                         gint fd;
 117                         struct kevent *events;
 118                 } kqueue;
 119 #endif
 120                 struct {
 121                         mono_pollfd *fds;
 122                         guint fds_size;
 123                         guint fds_max;
 124                 } poll;
 125         };
 126 } ThreadPoolIO;
 127
 128 static gint32 io_status = STATUS_NOT_INITIALIZED;
 129 static gint32 io_thread_status = STATUS_NOT_INITIALIZED;
 130
 131 static ThreadPoolIO* threadpool_io;
 132
 133 static int
 134 get_events_from_state (MonoSocketAsyncResult *ares)
 135 {
 136         switch (ares->operation) {
 137         case AIO_OP_ACCEPT:
 138         case AIO_OP_RECEIVE:
 139         case AIO_OP_RECV_JUST_CALLBACK:
 140         case AIO_OP_RECEIVEFROM:
 141         case AIO_OP_READPIPE:
 142         case AIO_OP_ACCEPTRECEIVE:
 143         case AIO_OP_RECEIVE_BUFFERS:
 144                 return MONO_POLLIN;
 145         case AIO_OP_SEND:
 146         case AIO_OP_SEND_JUST_CALLBACK:
 147         case AIO_OP_SENDTO:
 148         case AIO_OP_CONNECT:
 149         case AIO_OP_SEND_BUFFERS:
 150         case AIO_OP_DISCONNECT:
 151                 return MONO_POLLOUT;
 152         default:
 153                 g_assert_not_reached ();
 154         }
 155 }
 156
 157 static MonoSocketAsyncResult*
 158 get_state (MonoMList **list, gint event)
 159 {
 160         MonoSocketAsyncResult *state = NULL;
 161         MonoMList *current;
 162
 163         g_assert (list);
 164
 165         for (current = *list; current; current = mono_mlist_next (current)) {
 166                 state = (MonoSocketAsyncResult*) mono_mlist_get_data (current);
 167                 if (get_events_from_state ((MonoSocketAsyncResult*) state) == event)
 168                         break;
 169                 state = NULL;
 170         }
 171
 172         if (current)
 173                 *list = mono_mlist_remove_item (*list, current);
 174
 175         return state;
 176 }
 177
 178 static gint
 179 get_events (MonoMList *list)
 180 {
 181         MonoSocketAsyncResult *ares;
 182         gint events = 0;
 183
 184         for (; list; list = mono_mlist_next (list))
 185                 if ((ares = (MonoSocketAsyncResult*) mono_mlist_get_data (list)))
 186                         events |= get_events_from_state (ares);
 187
 188         return events;
 189 }
 190
 191 static void
 192 polling_thread_wakeup (void)
 193 {
 194         gchar msg = 'c';
 195         gint written;
 196
 197         for (;;) {
 198 #if !defined(HOST_WIN32)
 199                 written = write (threadpool_io->wakeup_pipes [1], &msg, 1);
 200                 if (written == 1)
 201                         break;
 202                 if (written == -1) {
 203                         g_warning ("polling_thread_wakeup: write () failed, error (%d) %s\n", errno, g_strerror (errno));
 204                         break;
 205                 }
 206 #else
 207                 written = send (threadpool_io->wakeup_pipes [1], &msg, 1, 0);
 208                 if (written == 1)
 209                         break;
 210                 if (written == SOCKET_ERROR) {
 211                         g_warning ("polling_thread_wakeup: write () failed, error (%d)\n", WSAGetLastError ());
 212                         break;
 213                 }
 214 #endif
 215         }
 216 }
 217
 218 static void
 219 polling_thread_drain_wakeup_pipes (void)
 220 {
 221         gchar buffer [128];
 222         gint received;
 223
 224         for (;;) {
 225 #if !defined(HOST_WIN32)
 226                 received = read (threadpool_io->wakeup_pipes [0], buffer, sizeof (buffer));
 227                 if (received == 0)
 228                         break;
 229                 if (received == -1) {
 230                         if (errno != EINTR && errno != EAGAIN)
 231                                 g_warning ("poll_thread: read () failed, error (%d) %s\n", errno, g_strerror (errno));
 232                         break;
 233                 }
 234 #else
 235                 received = recv (threadpool_io->wakeup_pipes [0], buffer, sizeof (buffer), 0);
 236                 if (received == 0)
 237                         break;
 238                 if (received == SOCKET_ERROR) {
 239                         if (WSAGetLastError () != WSAEINTR && WSAGetLastError () != WSAEWOULDBLOCK)
 240                                 g_warning ("poll_thread: recv () failed, error (%d) %s\n", WSAGetLastError ());
 241                         break;
 242                 }
 243 #endif
 244         }
 245 }
 246
 247 #if defined(HAVE_EPOLL)
 248
 249 #if defined(HOST_WIN32)
 250 /* We assume that epoll is not available on windows */
 251 #error
 252 #endif
 253
 254 #define EPOLL_NEVENTS 128
 255
 256 static gboolean
 257 epoll_init (void)
 258 {
 259 #ifdef EPOOL_CLOEXEC
 260         threadpool_io->epoll.fd = epoll_create1 (EPOLL_CLOEXEC);
 261 #else
 262         threadpool_io->epoll.fd = epoll_create (256);
 263         fcntl (threadpool_io->epoll.fd, F_SETFD, FD_CLOEXEC);
 264 #endif
 265
 266         if (threadpool_io->epoll.fd == -1) {
 267 #ifdef EPOOL_CLOEXEC
 268                 g_warning ("epoll_init: epoll (EPOLL_CLOEXEC) failed, error (%d) %s\n", errno, g_strerror (errno));
 269 #else
 270                 g_warning ("epoll_init: epoll (256) failed, error (%d) %s\n", errno, g_strerror (errno));
 271 #endif
 272                 return FALSE;
 273         }
 274
 275         if (epoll_ctl (threadpool_io->epoll.fd, EPOLL_CTL_ADD, threadpool_io->wakeup_pipes [0], EPOLLIN) == -1) {
 276                 g_warning ("epoll_init: epoll_ctl () failed, error (%d) %s", errno, g_strerror (errno));
 277                 close (threadpool_io->epoll.fd);
 278                 return FALSE;
 279         }
 280
 281         threadpool_io->epoll.events = g_new0 (struct epoll_event, EPOLL_NEVENTS);
 282
 283         return TRUE;
 284 }
 285
 286 static void
 287 epoll_cleanup (void)
 288 {
 289         g_free (threadpool_io->epoll.events);
 290
 291         close (threadpool_io->epoll.fd);
 292 }
 293
 294 static void
 295 epoll_update (gint fd, gint events, gboolean is_new)
 296 {
 297         ThreadPoolIOUpdate *update;
 298         struct epoll_event *event;
 299         gchar msg = 'c';
 300
 301         event = g_new0 (struct epoll_event, 1);
 302         event->data.fd = fd;
 303         if ((events & MONO_POLLIN) != 0)
 304                 event->events |= EPOLLIN;
 305         if ((events & MONO_POLLOUT) != 0)
 306                 event->events |= EPOLLOUT;
 307
 308         mono_mutex_lock (&threadpool_io->updates_lock);
 309         threadpool_io->updates_size += 1;
 310         threadpool_io->updates = g_renew (ThreadPoolIOUpdate, threadpool_io->updates, threadpool_io->updates_size);
 311
 312         update = &threadpool_io->updates [threadpool_io->updates_size - 1];
 313         update->fd = fd;
 314         update->epoll.event = event;
 315         update->epoll.op = is_new ? EPOLL_CTL_ADD : EPOLL_CTL_MOD;
 316         mono_mutex_unlock (&threadpool_io->updates_lock);
 317
 318         polling_thread_wakeup ();
 319 }
 320
 321 static void
 322 epoll_thread_add_update (ThreadPoolIOUpdate *update)
 323 {
 324         if (epoll_ctl (threadpool_io->epoll.fd, update->epoll.op, update->fd, update->epoll.event) == -1)
 325                 g_warning ("epoll_thread_add_update: epoll_ctl(%s) failed, error (%d) %s", update->epoll.op == EPOLL_CTL_ADD ? "EPOLL_CTL_ADD" : "EPOLL_CTL_MOD", errno, g_strerror (errno));
 326         g_free (update->epoll.event);
 327 }
 328
 329 static gint
 330 epoll_thread_wait_for_event (void)
 331 {
 332         gint ready;
 333
 334         ready = epoll_wait (threadpool_io->epoll.fd, threadpool_io->epoll.events, EPOLL_NEVENTS, -1);
 335         if (ready == -1) {
 336                 switch (errno) {
 337                 case EINTR:
 338                         check_for_interruption_critical ();
 339                         ready = 0;
 340                         break;
 341                 default:
 342                         g_warning ("epoll_thread_wait_for_event: epoll_wait () failed, error (%d) %s", errno, g_strerror (errno));
 343                         break;
 344                 }
 345         }
 346
 347         return ready;
 348 }
 349
 350 static inline gint
 351 epoll_thread_get_fd_at (guint i)
 352 {
 353         return threadpool_io->epoll.events [i].data.fd;
 354 }
 355
 356 static gboolean
 357 epoll_thread_create_socket_async_results (gint fd, struct epoll_event *epoll_event, MonoMList **list)
 358 {
 359         g_assert (epoll_event);
 360         g_assert (list);
 361
 362         if (!*list) {
 363                 epoll_ctl (threadpool_io->epoll.fd, EPOLL_CTL_DEL, fd, epoll_event);
 364         } else {
 365                 gint events;
 366
 367                 if ((epoll_event->events & (EPOLLIN | EPOLLERR | EPOLLHUP)) != 0) {
 368                         MonoSocketAsyncResult *io_event = get_state (list, MONO_POLLIN);
 369                         if (io_event)
 370                                 mono_threadpool_io_enqueue_socket_async_result (((MonoObject*) io_event)->vtable->domain, io_event);
 371                 }
 372                 if ((epoll_event->events & (EPOLLOUT | EPOLLERR | EPOLLHUP)) != 0) {
 373                         MonoSocketAsyncResult *io_event = get_state (list, MONO_POLLOUT);
 374                         if (io_event)
 375                                 mono_threadpool_io_enqueue_socket_async_result (((MonoObject*) io_event)->vtable->domain, io_event);
 376                 }
 377
 378                 events = get_events (*list);
 379                 epoll_event->events = ((events & MONO_POLLOUT) ? EPOLLOUT : 0) | ((events & MONO_POLLIN) ? EPOLLIN : 0);
 380                 if (epoll_ctl (threadpool_io->epoll.fd, EPOLL_CTL_MOD, fd, epoll_event) == -1) {
 381                         if (epoll_ctl (threadpool_io->epoll.fd, EPOLL_CTL_ADD, fd, epoll_event) == -1)
 382                                 g_warning ("epoll_thread_create_socket_async_results: epoll_ctl () failed, error (%d) %s", errno, g_strerror (errno));
 383                 }
 384         }
 385
 386         return TRUE;
 387 }
 388
 389 #elif defined(HAVE_KQUEUE)
 390
 391 #if defined(HOST_WIN32)
 392 /* We assume that kqueue is not available on windows */
 393 #error
 394 #endif
 395
 396 #define KQUEUE_NEVENTS 128
 397
 398 static gboolean
 399 kqueue_init (void)
 400 {
 401         struct kevent event;
 402
 403         threadpool_io->kqueue.fd = kqueue ();
 404         if (threadpool_io->kqueue.fd == -1) {
 405                 g_warning ("kqueue_init: kqueue () failed, error (%d) %s", errno, g_strerror (errno));
 406                 return FALSE;
 407         }
 408
 409         EV_SET (&event, threadpool_io->wakeup_pipes [0], EVFILT_READ, EV_ADD | EV_ENABLE, 0, 0, 0);
 410         if (kevent (threadpool_io->kqueue.fd, &event, 1, NULL, 0, NULL) == -1) {
 411                 g_warning ("kqueue_init: kevent () failed, error (%d) %s", errno, g_strerror (errno));
 412                 close (threadpool_io->kqueue.fd);
 413                 return FALSE;
 414         }
 415
 416         threadpool_io->kqueue.events = g_new0 (struct kevent, KQUEUE_NEVENTS);
 417
 418         return TRUE;
 419 }
 420
 421 static void
 422 kqueue_cleanup (void)
 423 {
 424         g_free (threadpool_io->kqueue.events);
 425
 426         close (threadpool_io->kqueue.fd);
 427 }
 428
 429 static void
 430 kqueue_update (gint fd, gint events, gboolean is_new)
 431 {
 432         ThreadPoolIOUpdate *update;
 433         struct kevent *event;
 434
 435         event = g_new0 (struct kevent, 1);
 436         if ((events & MONO_POLLIN) != 0)
 437                 EV_SET (event, fd, EVFILT_READ, EV_ADD | EV_ENABLE | EV_ONESHOT, 0, 0, 0);
 438         if ((events & MONO_POLLOUT) != 0)
 439                 EV_SET (event, fd, EVFILT_WRITE, EV_ADD | EV_ENABLE | EV_ONESHOT, 0, 0, 0);
 440
 441         mono_mutex_lock (&threadpool_io->updates_lock);
 442         threadpool_io->updates_size += 1;
 443         threadpool_io->updates = g_renew (ThreadPoolIOUpdate, threadpool_io->updates, threadpool_io->updates_size);
 444
 445         update = &threadpool_io->updates [threadpool_io->updates_size - 1];
 446         update->fd = fd;
 447         update->kqueue.event = event;
 448         mono_mutex_unlock (&threadpool_io->updates_lock);
 449
 450         polling_thread_wakeup ();
 451 }
 452
 453 static void
 454 kqueue_thread_add_update (ThreadPoolIOUpdate *update)
 455 {
 456         if (kevent (threadpool_io->kqueue.fd, update->kqueue.event, 1, NULL, 0, NULL) == -1)
 457                 g_warning ("kqueue_thread_add_update: kevent(update) failed, error (%d) %s", errno, g_strerror (errno));
 458         g_free (update->kqueue.event);
 459 }
 460
 461 static gint
 462 kqueue_thread_wait_for_event (void)
 463 {
 464         gint ready;
 465
 466         ready = kevent (threadpool_io->kqueue.fd, NULL, 0, threadpool_io->kqueue.events, KQUEUE_NEVENTS, NULL);
 467         if (ready == -1) {
 468                 switch (errno) {
 469                 case EINTR:
 470                         check_for_interruption_critical ();
 471                         ready = 0;
 472                         break;
 473                 default:
 474                         g_warning ("kqueue_thread_wait_for_event: kevent () failed, error (%d) %s", errno, g_strerror (errno));
 475                         break;
 476                 }
 477         }
 478
 479         return ready;
 480 }
 481
 482 static inline gint
 483 kqueue_thread_get_fd_at (guint i)
 484 {
 485         return threadpool_io->kqueue.events [i].ident;
 486 }
 487
 488 static gboolean
 489 kqueue_thread_create_socket_async_results (gint fd, struct kevent *kqueue_event, MonoMList **list)
 490 {
 491         g_assert (kqueue_event);
 492         g_assert (list);
 493
 494         if (*list) {
 495                 gint events;
 496
 497                 if (kqueue_event->filter == EVFILT_READ || (kqueue_event->flags & EV_ERROR) != 0) {
 498                         MonoSocketAsyncResult *io_event = get_state (list, MONO_POLLIN);
 499                         if (io_event)
 500                                 mono_threadpool_io_enqueue_socket_async_result (((MonoObject*) io_event)->vtable->domain, io_event);
 501                 }
 502                 if (kqueue_event->filter == EVFILT_WRITE || (kqueue_event->flags & EV_ERROR) != 0) {
 503                         MonoSocketAsyncResult *io_event = get_state (list, MONO_POLLOUT);
 504                         if (io_event)
 505                                 mono_threadpool_io_enqueue_socket_async_result (((MonoObject*) io_event)->vtable->domain, io_event);
 506                 }
 507
 508                 events = get_events (*list);
 509                 if (kqueue_event->filter == EVFILT_READ && (events & MONO_POLLIN) != 0) {
 510                         EV_SET (kqueue_event, fd, EVFILT_READ, EV_ADD | EV_ENABLE | EV_ONESHOT, 0, 0, 0);
 511                         if (kevent (threadpool_io->kqueue.fd, kqueue_event, 1, NULL, 0, NULL) == -1)
 512                                 g_warning ("kqueue_thread_create_socket_async_results: kevent (read) failed, error (%d) %s", errno, g_strerror (errno));
 513                 }
 514                 if (kqueue_event->filter == EVFILT_WRITE && (events & MONO_POLLOUT) != 0) {
 515                         EV_SET (kqueue_event, fd, EVFILT_WRITE, EV_ADD | EV_ENABLE | EV_ONESHOT, 0, 0, 0);
 516                         if (kevent (threadpool_io->kqueue.fd, kqueue_event, 1, NULL, 0, NULL) == -1)
 517                                 g_warning ("kqueue_thread_create_socket_async_results: kevent (write) failed, error (%d) %s", errno, g_strerror (errno));
 518                 }
 519         }
 520
 521         return TRUE;
 522 }
 523
 524 #endif
 525
 526 #define POLL_NEVENTS 1024
 527
 528 static inline void
 529 POLL_INIT_FD (mono_pollfd *poll_fd, gint fd, gint events)
 530 {
 531         poll_fd->fd = fd;
 532         poll_fd->events = events;
 533         poll_fd->revents = 0;
 534 }
 535
 536 static gboolean
 537 poll_init (void)
 538 {
 539         guint i;
 540
 541         threadpool_io->poll.fds_max = 1;
 542         threadpool_io->poll.fds_size = POLL_NEVENTS;
 543         threadpool_io->poll.fds = g_new0 (mono_pollfd, threadpool_io->poll.fds_size);
 544
 545         POLL_INIT_FD (threadpool_io->poll.fds, threadpool_io->wakeup_pipes [0], MONO_POLLIN);
 546         for (i = threadpool_io->poll.fds_max; i < threadpool_io->poll.fds_size; ++i)
 547                 POLL_INIT_FD (threadpool_io->poll.fds + i, -1, 0);
 548
 549         return TRUE;
 550 }
 551
 552 static void
 553 poll_cleanup (void)
 554 {
 555         g_free (threadpool_io->poll.fds);
 556 }
 557
 558 static void
 559 poll_update (gint fd, gint events, gboolean is_new)
 560 {
 561         ThreadPoolIOUpdate *update;
 562
 563         mono_mutex_lock (&threadpool_io->updates_lock);
 564         threadpool_io->updates_size += 1;
 565         threadpool_io->updates = g_renew (ThreadPoolIOUpdate, threadpool_io->updates, threadpool_io->updates_size);
 566
 567         update = &threadpool_io->updates [threadpool_io->updates_size - 1];
 568         update->fd = fd;
 569         POLL_INIT_FD (&update->poll.fd, fd, events);
 570         mono_mutex_unlock (&threadpool_io->updates_lock);
 571
 572         polling_thread_wakeup ();
 573 }
 574
 575 static gint
 576 poll_mark_bad_fds (mono_pollfd *poll_fds, gint poll_fds_size)
 577 {
 578         gint i;
 579         gint ret;
 580         gint ready = 0;
 581         mono_pollfd *poll_fd;
 582
 583         for (i = 0; i < poll_fds_size; i++) {
 584                 poll_fd = poll_fds + i;
 585                 if (poll_fd->fd == -1)
 586                         continue;
 587
 588                 ret = mono_poll (poll_fd, 1, 0);
 589                 if (ret == 1)
 590                         ready++;
 591                 if (ret == -1) {
 592 #if !defined(HOST_WIN32)
 593                         if (errno == EBADF)
 594 #else
 595                         if (WSAGetLastError () == WSAEBADF)
 596 #endif
 597                         {
 598                                 poll_fd->revents |= MONO_POLLNVAL;
 599                                 ready++;
 600                         }
 601                 }
 602         }
 603
 604         return ready;
 605 }
 606
 607 static void
 608 poll_thread_add_update (ThreadPoolIOUpdate *update)
 609 {
 610         gboolean found = FALSE;
 611         gint j, k;
 612
 613         for (j = 1; j < threadpool_io->poll.fds_size; ++j) {
 614                 mono_pollfd *poll_fd = threadpool_io->poll.fds + j;
 615                 if (poll_fd->fd == update->poll.fd.fd) {
 616                         found = TRUE;
 617                         break;
 618                 }
 619         }
 620
 621         if (!found) {
 622                 for (j = 1; j < threadpool_io->poll.fds_size; ++j) {
 623                         mono_pollfd *poll_fd = threadpool_io->poll.fds + j;
 624                         if (poll_fd->fd == -1)
 625                                 break;
 626                 }
 627         }
 628
 629         if (j == threadpool_io->poll.fds_size) {
 630                 threadpool_io->poll.fds_size += POLL_NEVENTS;
 631                 threadpool_io->poll.fds = g_renew (mono_pollfd, threadpool_io->poll.fds, threadpool_io->poll.fds_size);
 632                 for (k = j; k < threadpool_io->poll.fds_size; ++k)
 633                         POLL_INIT_FD (threadpool_io->poll.fds + k, -1, 0);
 634         }
 635
 636         POLL_INIT_FD (threadpool_io->poll.fds + j, update->poll.fd.fd, update->poll.fd.events);
 637
 638         if (j >= threadpool_io->poll.fds_max)
 639                 threadpool_io->poll.fds_max = j + 1;
 640 }
 641
 642 static gint
 643 poll_thread_wait_for_event (void)
 644 {
 645         gint ready;
 646
 647         ready = mono_poll (threadpool_io->poll.fds, threadpool_io->poll.fds_max, -1);
 648         if (ready == -1) {
 649                 /*
 650                  * Apart from EINTR, we only check EBADF, for the rest:
 651                  *  EINVAL: mono_poll() 'protects' us from descriptor
 652                  *      numbers above the limit if using select() by marking
 653                  *      then as MONO_POLLERR.  If a system poll() is being
 654                  *      used, the number of descriptor we're passing will not
 655                  *      be over sysconf(_SC_OPEN_MAX), as the error would have
 656                  *      happened when opening.
 657                  *
 658                  *  EFAULT: we own the memory pointed by pfds.
 659                  *  ENOMEM: we're doomed anyway
 660                  *
 661                  */
 662 #if !defined(HOST_WIN32)
 663                 switch (errno)
 664 #else
 665                 switch (WSAGetLastError ())
 666 #endif
 667                 {
 668 #if !defined(HOST_WIN32)
 669                 case EINTR:
 670 #else
 671                 case WSAEINTR:
 672 #endif
 673                         check_for_interruption_critical ();
 674                         ready = 0;
 675                         break;
 676 #if !defined(HOST_WIN32)
 677                 case EBADF:
 678 #else
 679                 case WSAEBADF:
 680 #endif
 681                         ready = poll_mark_bad_fds (threadpool_io->poll.fds, threadpool_io->poll.fds_max);
 682                         break;
 683                 default:
 684 #if !defined(HOST_WIN32)
 685                         g_warning ("poll_thread_wait_for_event: mono_poll () failed, error (%d) %s", errno, g_strerror (errno));
 686 #else
 687                         g_warning ("poll_thread_wait_for_event: mono_poll () failed, error (%d)\n", WSAGetLastError ());
 688 #endif
 689                         break;
 690                 }
 691         }
 692
 693         return ready;
 694 }
 695
 696 static inline gint
 697 poll_thread_get_fd_at (guint i)
 698 {
 699         return threadpool_io->poll.fds [i].fd;
 700 }
 701
 702 static gboolean
 703 poll_thread_create_socket_async_results (gint fd, mono_pollfd *poll_fd, MonoMList **list)
 704 {
 705         g_assert (poll_fd);
 706         g_assert (list);
 707
 708         if (fd == -1 || poll_fd->revents == 0)
 709                 return FALSE;
 710
 711         if (!*list) {
 712                 POLL_INIT_FD (poll_fd, -1, 0);
 713         } else {
 714                 if ((poll_fd->revents & (MONO_POLLIN | MONO_POLLERR | MONO_POLLHUP | MONO_POLLNVAL)) != 0) {
 715                         MonoSocketAsyncResult *io_event = get_state (list, MONO_POLLIN);
 716                         if (io_event)
 717                                 mono_threadpool_io_enqueue_socket_async_result (((MonoObject*) io_event)->vtable->domain, io_event);
 718                 }
 719                 if ((poll_fd->revents & (MONO_POLLOUT | MONO_POLLERR | MONO_POLLHUP | MONO_POLLNVAL)) != 0) {
 720                         MonoSocketAsyncResult *io_event = get_state (list, MONO_POLLOUT);
 721                         if (io_event)
 722                                 mono_threadpool_io_enqueue_socket_async_result (((MonoObject*) io_event)->vtable->domain, io_event);
 723                 }
 724
 725                 poll_fd->events = get_events (*list);
 726         }
 727
 728         return TRUE;
 729 }
 730
 731 static void
 732 polling_thread (gpointer data)
 733 {
 734         io_thread_status = STATUS_INITIALIZED;
 735
 736         for (;;) {
 737                 guint i;
 738                 guint max;
 739                 gint ready = 0;
 740
 741                 mono_gc_set_skip_thread (TRUE);
 742
 743                 mono_mutex_lock (&threadpool_io->updates_lock);
 744                 for (i = 0; i < threadpool_io->updates_size; ++i) {
 745                         switch (threadpool_io->backend) {
 746 #if defined(HAVE_EPOLL)
 747                         case BACKEND_EPOLL:
 748                                 epoll_thread_add_update (&threadpool_io->updates [i]);
 749                                 break;
 750 #elif defined(HAVE_KQUEUE)
 751                         case BACKEND_KQUEUE:
 752                                 kqueue_thread_add_update (&threadpool_io->updates [i]);
 753                                 break;
 754 #endif
 755                         case BACKEND_POLL:
 756                                 poll_thread_add_update (&threadpool_io->updates [i]);
 757                                 break;
 758                         default:
 759                                 g_assert_not_reached ();
 760                         }
 761
 762                 }
 763                 if (threadpool_io->updates_size > 0) {
 764                         threadpool_io->updates_size = 0;
 765                         threadpool_io->updates = g_renew (ThreadPoolIOUpdate, threadpool_io->updates, threadpool_io->updates_size);
 766                 }
 767                 mono_mutex_unlock (&threadpool_io->updates_lock);
 768
 769                 switch (threadpool_io->backend) {
 770 #if defined(HAVE_EPOLL)
 771                 case BACKEND_EPOLL:
 772                         ready = epoll_thread_wait_for_event ();
 773                         break;
 774 #elif defined(HAVE_KQUEUE)
 775                 case BACKEND_KQUEUE:
 776                         ready = kqueue_thread_wait_for_event ();
 777                         break;
 778 #endif
 779                 case BACKEND_POLL:
 780                         ready = poll_thread_wait_for_event ();
 781                         break;
 782                 default:
 783                         g_assert_not_reached ();
 784                 }
 785
 786                 mono_gc_set_skip_thread (FALSE);
 787
 788                 if (ready == -1 || mono_runtime_is_shutting_down ())
 789                         break;
 790
 791                 switch (threadpool_io->backend) {
 792 #if defined(HAVE_EPOLL)
 793                 case BACKEND_EPOLL:
 794                         max = EPOLL_NEVENTS;
 795                         break;
 796 #elif defined(HAVE_KQUEUE)
 797                 case BACKEND_KQUEUE:
 798                         max = KQUEUE_NEVENTS;
 799                         break;
 800 #endif
 801                 case BACKEND_POLL:
 802                         max = threadpool_io->poll.fds_max;
 803                         break;
 804                 default:
 805                         g_assert_not_reached ();
 806                 }
 807
 808                 mono_mutex_lock (&threadpool_io->states_lock);
 809                 for (i = 0; i < max && ready > 0; ++i) {
 810                         MonoMList *list;
 811                         gboolean created;
 812                         gint fd;
 813
 814                         switch (threadpool_io->backend) {
 815 #if defined(HAVE_EPOLL)
 816                         case BACKEND_EPOLL:
 817                                 fd = epoll_thread_get_fd_at (i);
 818                                 break;
 819 #elif defined(HAVE_KQUEUE)
 820                         case BACKEND_KQUEUE:
 821                                 fd = kqueue_thread_get_fd_at (i);
 822                                 break;
 823 #endif
 824                         case BACKEND_POLL:
 825                                 fd = poll_thread_get_fd_at (i);
 826                                 break;
 827                         default:
 828                                 g_assert_not_reached ();
 829                         }
 830
 831                         if (fd == threadpool_io->wakeup_pipes [0]) {
 832                                 polling_thread_drain_wakeup_pipes ();
 833                                 ready -= 1;
 834                                 continue;
 835                         }
 836
 837                         list = mono_g_hash_table_lookup (threadpool_io->states, GINT_TO_POINTER (fd));
 838
 839                         switch (threadpool_io->backend) {
 840 #if defined(HAVE_EPOLL)
 841                         case BACKEND_EPOLL:
 842                                 created = epoll_thread_create_socket_async_results (fd, &threadpool_io->epoll.events [i], &list);
 843                                 break;
 844 #elif defined(HAVE_KQUEUE)
 845                         case BACKEND_KQUEUE:
 846                                 created = kqueue_thread_create_socket_async_results (fd, &threadpool_io->kqueue.events [i], &list);
 847                                 break;
 848 #endif
 849                         case BACKEND_POLL:
 850                                 created = poll_thread_create_socket_async_results (fd, &threadpool_io->poll.fds [i], &list);
 851                                 break;
 852                         default:
 853                                 g_assert_not_reached ();
 854                         }
 855
 856                         if (!created)
 857                                 continue;
 858
 859                         if (list)
 860                                 mono_g_hash_table_replace (threadpool_io->states, GINT_TO_POINTER (fd), list);
 861                         else
 862                                 mono_g_hash_table_remove (threadpool_io->states, GINT_TO_POINTER (fd));
 863
 864                         ready -= 1;
 865                 }
 866                 mono_mutex_unlock (&threadpool_io->states_lock);
 867         }
 868
 869         io_thread_status = STATUS_CLEANED_UP;
 870 }
 871
 872 static void
 873 wakeup_pipes_init (void)
 874 {
 875 #if !defined(HOST_WIN32)
 876         if (pipe (threadpool_io->wakeup_pipes) == -1)
 877                 g_error ("wakeup_pipes_init: pipe () failed, error (%d) %s\n", errno, g_strerror (errno));
 878         if (fcntl (threadpool_io->wakeup_pipes [0], F_SETFL, O_NONBLOCK) == -1)
 879                 g_error ("wakeup_pipes_init: fcntl () failed, error (%d) %s\n", errno, g_strerror (errno));
 880 #else
 881         struct sockaddr_in client;
 882         struct sockaddr_in server;
 883         SOCKET server_sock;
 884         gulong arg;
 885         gint size;
 886
 887         server_sock = socket (AF_INET, SOCK_STREAM, IPPROTO_TCP);
 888         g_assert (server_sock != INVALID_SOCKET);
 889         threadpool_io->wakeup_pipes [1] = socket (AF_INET, SOCK_STREAM, IPPROTO_TCP);
 890         g_assert (threadpool_io->wakeup_pipes [1] != INVALID_SOCKET);
 891
 892         server.sin_family = AF_INET;
 893         server.sin_addr.s_addr = inet_addr ("127.0.0.1");
 894         server.sin_port = 0;
 895         if (bind (server_sock, (SOCKADDR*) &server, sizeof (server)) == SOCKET_ERROR) {
 896                 closesocket (server_sock);
 897                 g_error ("wakeup_pipes_init: bind () failed, error (%d)\n", WSAGetLastError ());
 898         }
 899
 900         size = sizeof (server);
 901         if (getsockname (server_sock, (SOCKADDR*) &server, &size) == SOCKET_ERROR) {
 902                 closesocket (server_sock);
 903                 g_error ("wakeup_pipes_init: getsockname () failed, error (%d)\n", WSAGetLastError ());
 904         }
 905         if (listen (server_sock, 1024) == SOCKET_ERROR) {
 906                 closesocket (server_sock);
 907                 g_error ("wakeup_pipes_init: listen () failed, error (%d)\n", WSAGetLastError ());
 908         }
 909         if (connect ((SOCKET) threadpool_io->wakeup_pipes [1], (SOCKADDR*) &server, sizeof (server)) == SOCKET_ERROR) {
 910                 closesocket (server_sock);
 911                 g_error ("wakeup_pipes_init: connect () failed, error (%d)\n", WSAGetLastError ());
 912         }
 913
 914         size = sizeof (client);
 915         threadpool_io->wakeup_pipes [0] = accept (server_sock, (SOCKADDR *) &client, &size);
 916         g_assert (threadpool_io->wakeup_pipes [0] != INVALID_SOCKET);
 917
 918         arg = 1;
 919         if (ioctlsocket (threadpool_io->wakeup_pipes [0], FIONBIO, &arg) == SOCKET_ERROR) {
 920                 closesocket (threadpool_io->wakeup_pipes [0]);
 921                 closesocket (server_sock);
 922                 g_error ("wakeup_pipes_init: ioctlsocket () failed, error (%d)\n", WSAGetLastError ());
 923         }
 924
 925         closesocket (server_sock);
 926 #endif
 927 }
 928
 929 static void
 930 ensure_initialized (void)
 931 {
 932         if (io_status >= STATUS_INITIALIZED)
 933                 return;
 934         if (io_status == STATUS_INITIALIZING || InterlockedCompareExchange (&io_status, STATUS_INITIALIZING, STATUS_NOT_INITIALIZED) != STATUS_NOT_INITIALIZED) {
 935                 while (io_status == STATUS_INITIALIZING)
 936                         mono_thread_info_yield ();
 937                 g_assert (io_status >= STATUS_INITIALIZED);
 938                 return;
 939         }
 940
 941         g_assert (!threadpool_io);
 942         threadpool_io = g_new0 (ThreadPoolIO, 1);
 943         g_assert (threadpool_io);
 944
 945         threadpool_io->states = mono_g_hash_table_new_type (g_direct_hash, g_direct_equal, MONO_HASH_VALUE_GC);
 946         MONO_GC_REGISTER_ROOT_FIXED (threadpool_io->states);
 947         mono_mutex_init (&threadpool_io->states_lock);
 948
 949         threadpool_io->updates = NULL;
 950         threadpool_io->updates_size = 0;
 951         mono_mutex_init (&threadpool_io->updates_lock);
 952
 953 #if defined(HAVE_EPOLL)
 954         threadpool_io->backend = BACKEND_EPOLL;
 955 #elif defined(HAVE_KQUEUE)
 956         threadpool_io->backend = BACKEND_KQUEUE;
 957 #else
 958         threadpool_io->backend = BACKEND_POLL;
 959 #endif
 960         if (g_getenv ("MONO_DISABLE_AIO") != NULL)
 961                 threadpool_io->backend = BACKEND_POLL;
 962
 963         wakeup_pipes_init ();
 964
 965 retry_init_backend:
 966         switch (threadpool_io->backend) {
 967 #if defined(HAVE_EPOLL)
 968         case BACKEND_EPOLL:
 969                 if (!epoll_init ()) {
 970                         threadpool_io->backend = BACKEND_POLL;
 971                         goto retry_init_backend;
 972                 }
 973                 break;
 974 #elif defined(HAVE_KQUEUE)
 975         case BACKEND_KQUEUE:
 976                 if (!kqueue_init ()) {
 977                         threadpool_io->backend = BACKEND_POLL;
 978                         goto retry_init_backend;
 979                 }
 980                 break;
 981 #endif
 982         case BACKEND_POLL:
 983                 if (!poll_init ())
 984                         g_error ("ensure_initialized: poll_init () failed");
 985                 break;
 986         default:
 987                 g_assert_not_reached ();
 988         }
 989
 990         if (!mono_thread_create_internal (mono_get_root_domain (), polling_thread, NULL, TRUE, SMALL_STACK))
 991                 g_error ("ensure_initialized: mono_thread_create_internal () failed");
 992
 993         io_thread_status = STATUS_INITIALIZING;
 994         mono_memory_write_barrier ();
 995
 996         io_status = STATUS_INITIALIZED;
 997 }
 998
 999 static void
1000 ensure_cleanedup (void)
1001 {
1002         if (io_status == STATUS_NOT_INITIALIZED && InterlockedCompareExchange (&io_status, STATUS_CLEANED_UP, STATUS_NOT_INITIALIZED) == STATUS_NOT_INITIALIZED)
1003                 return;
1004         if (io_status == STATUS_INITIALIZING) {
1005                 while (io_status == STATUS_INITIALIZING)
1006                         mono_thread_info_yield ();
1007         }
1008         if (io_status == STATUS_CLEANED_UP)
1009                 return;
1010         if (io_status == STATUS_CLEANING_UP || InterlockedCompareExchange (&io_status, STATUS_CLEANING_UP, STATUS_INITIALIZED) != STATUS_INITIALIZED) {
1011                 while (io_status == STATUS_CLEANING_UP)
1012                         mono_thread_info_yield ();
1013                 g_assert (io_status == STATUS_CLEANED_UP);
1014                 return;
1015         }
1016
1017         /* we make the assumption along the code that we are
1018          * cleaning up only if the runtime is shutting down */
1019         g_assert (mono_runtime_is_shutting_down ());
1020
1021         polling_thread_wakeup ();
1022         while (io_thread_status != STATUS_CLEANED_UP)
1023                 usleep (1000);
1024
1025         MONO_GC_UNREGISTER_ROOT (threadpool_io->states);
1026         mono_g_hash_table_destroy (threadpool_io->states);
1027         mono_mutex_destroy (&threadpool_io->states_lock);
1028
1029         g_free (threadpool_io->updates);
1030         mono_mutex_destroy (&threadpool_io->updates_lock);
1031
1032         switch (threadpool_io->backend) {
1033 #if defined(HAVE_EPOLL)
1034         case BACKEND_EPOLL:
1035                 epoll_cleanup ();
1036                 break;
1037 #elif defined(HAVE_KQUEUE)
1038         case BACKEND_KQUEUE:
1039                 kqueue_cleanup ();
1040                 break;
1041 #endif
1042         case BACKEND_POLL:
1043                 poll_cleanup ();
1044                 break;
1045         default:
1046                 g_assert_not_reached ();
1047         }
1048
1049 #if !defined(HOST_WIN32)
1050         close (threadpool_io->wakeup_pipes [0]);
1051         close (threadpool_io->wakeup_pipes [1]);
1052 #else
1053         closesocket (threadpool_io->wakeup_pipes [0]);
1054         closesocket (threadpool_io->wakeup_pipes [1]);
1055 #endif
1056
1057         g_assert (threadpool_io);
1058         g_free (threadpool_io);
1059         threadpool_io = NULL;
1060         g_assert (!threadpool_io);
1061
1062         io_status = STATUS_CLEANED_UP;
1063 }
1064
1065 static gboolean
1066 is_socket_async_callback (MonoImage *system_image, MonoClass *class)
1067 {
1068         MonoClass *socket_async_callback_class = NULL;
1069
1070         socket_async_callback_class = mono_class_from_name (system_image, "System.Net.Sockets", "SocketAsyncCallback");
1071         g_assert (socket_async_callback_class);
1072
1073         return class == socket_async_callback_class;
1074 }
1075
1076 static gboolean
1077 is_async_read_handler (MonoImage *system_image, MonoClass *class)
1078 {
1079         MonoClass *process_class = NULL;
1080
1081         process_class = mono_class_from_name (system_image, "System.Diagnostics", "Process");
1082         g_assert (process_class);
1083
1084         return class->nested_in && class->nested_in == process_class && strcmp (class->name, "AsyncReadHandler") == 0;
1085 }
1086
1087 gboolean
1088 mono_threadpool_ms_is_io (MonoObject *target, MonoObject *state)
1089 {
1090         MonoImage *system_image;
1091         MonoSocketAsyncResult *sockares;
1092
1093         system_image = mono_image_loaded ("System");
1094         if (!system_image)
1095                 return FALSE;
1096
1097         if (!is_socket_async_callback (system_image, target->vtable->klass) && !is_async_read_handler (system_image, target->vtable->klass))
1098                 return FALSE;
1099
1100         sockares = (MonoSocketAsyncResult*) state;
1101         if (sockares->operation < AIO_OP_FIRST || sockares->operation >= AIO_OP_LAST)
1102                 return FALSE;
1103
1104         return TRUE;
1105 }
1106
1107 void
1108 mono_threadpool_ms_io_cleanup (void)
1109 {
1110         ensure_cleanedup ();
1111 }
1112
1113 MonoAsyncResult *
1114 mono_threadpool_ms_io_add (MonoAsyncResult *ares, MonoSocketAsyncResult *sockares)
1115 {
1116         MonoMList *list;
1117         gboolean is_new;
1118         gint events;
1119         gint fd;
1120
1121         g_assert (ares);
1122         g_assert (sockares);
1123
1124         if (mono_runtime_is_shutting_down ())
1125                 return NULL;
1126
1127         ensure_initialized ();
1128
1129         MONO_OBJECT_SETREF (sockares, ares, ares);
1130
1131         fd = GPOINTER_TO_INT (sockares->handle);
1132
1133         mono_mutex_lock (&threadpool_io->states_lock);
1134         g_assert (threadpool_io->states);
1135
1136         list = mono_g_hash_table_lookup (threadpool_io->states, GINT_TO_POINTER (fd));
1137         is_new = list == NULL;
1138         list = mono_mlist_append (list, (MonoObject*) sockares);
1139         mono_g_hash_table_replace (threadpool_io->states, sockares->handle, list);
1140
1141         events = get_events (list);
1142
1143         switch (threadpool_io->backend) {
1144 #if defined(HAVE_EPOLL)
1145         case BACKEND_EPOLL: {
1146                 epoll_update (fd, events, is_new);
1147                 break;
1148         }
1149 #elif defined(HAVE_KQUEUE)
1150         case BACKEND_KQUEUE: {
1151                 kqueue_update (fd, events, is_new);
1152                 break;
1153         }
1154 #endif
1155         case BACKEND_POLL: {
1156                 poll_update (fd, events, is_new);
1157                 break;
1158         }
1159         default:
1160                 g_assert_not_reached ();
1161         }
1162
1163         mono_mutex_unlock (&threadpool_io->states_lock);
1164
1165         return ares;
1166 }
1167
1168 void
1169 mono_threadpool_ms_io_remove_socket (int fd)
1170 {
1171         MonoMList *list;
1172
1173         if (io_status != STATUS_INITIALIZED)
1174                 return;
1175
1176         mono_mutex_lock (&threadpool_io->states_lock);
1177         g_assert (threadpool_io->states);
1178         list = mono_g_hash_table_lookup (threadpool_io->states, GINT_TO_POINTER (fd));
1179         if (list)
1180                 mono_g_hash_table_remove (threadpool_io->states, GINT_TO_POINTER (fd));
1181         mono_mutex_unlock (&threadpool_io->states_lock);
1182
1183         while (list) {
1184                 MonoSocketAsyncResult *sockares, *sockares2;
1185
1186                 sockares = (MonoSocketAsyncResult*) mono_mlist_get_data (list);
1187                 if (sockares->operation == AIO_OP_RECEIVE)
1188                         sockares->operation = AIO_OP_RECV_JUST_CALLBACK;
1189                 else if (sockares->operation == AIO_OP_SEND)
1190                         sockares->operation = AIO_OP_SEND_JUST_CALLBACK;
1191
1192                 sockares2 = get_state (&list, MONO_POLLIN);
1193                 if (sockares2)
1194                         mono_threadpool_io_enqueue_socket_async_result (((MonoObject*) sockares2)->vtable->domain, sockares2);
1195
1196                 if (!list)
1197                         break;
1198
1199                 sockares2 = get_state (&list, MONO_POLLOUT);
1200                 if (sockares2)
1201                         mono_threadpool_io_enqueue_socket_async_result (((MonoObject*) sockares2)->vtable->domain, sockares2);
1202         }
1203 }
1204
1205 static gboolean
1206 remove_sockstate_for_domain (gpointer key, gpointer value, gpointer user_data)
1207 {
1208         MonoMList *list;
1209         gboolean remove = FALSE;
1210
1211         for (list = value; list; list = mono_mlist_next (list)) {
1212                 MonoObject *data = mono_mlist_get_data (list);
1213                 if (mono_object_domain (data) == user_data) {
1214                         remove = TRUE;
1215                         mono_mlist_set_data (list, NULL);
1216                 }
1217         }
1218
1219         //FIXME is there some sort of additional unregistration we need to perform here?
1220         return remove;
1221 }
1222
1223 void
1224 mono_threadpool_ms_io_remove_domain_jobs (MonoDomain *domain)
1225 {
1226         if (io_status == STATUS_INITIALIZED) {
1227                 mono_mutex_lock (&threadpool_io->states_lock);
1228                 mono_g_hash_table_foreach_remove (threadpool_io->states, remove_sockstate_for_domain, domain);
1229                 mono_mutex_unlock (&threadpool_io->states_lock);
1230         }
1231 }
1232
1233 void
1234 mono_threadpool_io_enqueue_socket_async_result (MonoDomain *domain, MonoSocketAsyncResult *sockares)
1235 {
1236         MonoImage *system_image;
1237         MonoClass *socket_runtime_work_item_class;
1238         MonoSocketRuntimeWorkItem *srwi;
1239
1240         g_assert (sockares);
1241
1242         system_image = mono_image_loaded ("System");
1243         g_assert (system_image);
1244
1245         socket_runtime_work_item_class = mono_class_from_name (system_image, "System.Net.Sockets", "MonoSocketRuntimeWorkItem");
1246         g_assert (socket_runtime_work_item_class);
1247
1248         srwi = (MonoSocketRuntimeWorkItem*) mono_object_new (domain, socket_runtime_work_item_class);
1249         MONO_OBJECT_SETREF (srwi, socket_async_result, sockares);
1250
1251         mono_threadpool_ms_enqueue_work_item (domain, (MonoObject*) srwi);
1252 }
1253
1254 void
1255 ves_icall_System_Net_Sockets_MonoSocketRuntimeWorkItem_ExecuteWorkItem (MonoSocketRuntimeWorkItem *rwi)
1256 {
1257         MonoSocketAsyncResult *sockares;
1258         MonoAsyncResult *ares;
1259         MonoObject *exc = NULL;
1260
1261         g_assert (rwi);
1262
1263         sockares = rwi->socket_async_result;
1264         g_assert (sockares);
1265         g_assert (sockares->ares);
1266
1267         switch (sockares->operation) {
1268         case AIO_OP_RECEIVE:
1269                 sockares->total = ves_icall_System_Net_Sockets_Socket_Receive_internal ((SOCKET) (gssize) sockares->handle, sockares->buffer, sockares->offset,
1270                                                                                             sockares->size, sockares->socket_flags, &sockares->error);
1271                 break;
1272         case AIO_OP_SEND:
1273                 sockares->total = ves_icall_System_Net_Sockets_Socket_Send_internal ((SOCKET) (gssize) sockares->handle, sockares->buffer, sockares->offset,
1274                                                                                             sockares->size, sockares->socket_flags, &sockares->error);
1275                 break;
1276         }
1277
1278         ares = sockares->ares;
1279         g_assert (ares);
1280
1281         mono_async_result_invoke (ares, &exc);
1282
1283         if (sockares->completed && sockares->callback) {
1284                 MonoAsyncResult *cb_ares;
1285
1286                 /* Don't call mono_async_result_new() to avoid capturing the context */
1287                 cb_ares = (MonoAsyncResult*) mono_object_new (mono_domain_get (), mono_defaults.asyncresult_class);
1288                 MONO_OBJECT_SETREF (cb_ares, async_delegate, sockares->callback);
1289                 MONO_OBJECT_SETREF (cb_ares, async_state, (MonoObject*) sockares);
1290
1291                 mono_threadpool_ms_enqueue_async_result (mono_domain_get (), cb_ares);
1292         }
1293
1294         if (exc)
1295                 mono_raise_exception ((MonoException*) exc);
1296 }
1297
1298 #else
1299
1300 gboolean
1301 mono_threadpool_ms_is_io (MonoObject *target, MonoObject *state)
1302 {
1303         return FALSE;
1304 }
1305
1306 void
1307 mono_threadpool_ms_io_cleanup (void)
1308 {
1309         g_assert_not_reached ();
1310 }
1311
1312 MonoAsyncResult *
1313 mono_threadpool_ms_io_add (MonoAsyncResult *ares, MonoSocketAsyncResult *sockares)
1314 {
1315         g_assert_not_reached ();
1316 }
1317
1318 void
1319 mono_threadpool_ms_io_remove_socket (int fd)
1320 {
1321         g_assert_not_reached ();
1322 }
1323
1324 void
1325 mono_threadpool_ms_io_remove_domain_jobs (MonoDomain *domain)
1326 {
1327         g_assert_not_reached ();
1328 }
1329
1330 void
1331 mono_threadpool_io_enqueue_socket_async_result (MonoDomain *domain, MonoSocketAsyncResult *sockares)
1332 {
1333         g_assert_not_reached ();
1334 }
1335
1336 void
1337 ves_icall_System_Net_Sockets_MonoSocketRuntimeWorkItem_ExecuteWorkItem (MonoSocketRuntimeWorkItem *rwi)
1338 {
1339         g_assert_not_reached ();
1340 }
1341
1342 #endif