mono/metadata/threadpool-io.c

   1 /**
   2  * \file
   3  * Microsoft IO threadpool runtime support
   4  *
   5  * Author:
   6  *      Ludovic Henry (ludovic.henry@xamarin.com)
   7  *
   8  * Copyright 2015 Xamarin, Inc (http://www.xamarin.com)
   9  * Licensed under the MIT license. See LICENSE file in the project root for full license information.
  10  */
  11
  12 #include <config.h>
  13 #include <glib.h>
  14 #include <mono/metadata/threadpool-io.h>
  15
  16 #ifndef DISABLE_SOCKETS
  17
  18 #if defined(HOST_WIN32)
  19 #include <windows.h>
  20 #else
  21 #include <errno.h>
  22 #include <fcntl.h>
  23 #endif
  24
  25 #include <mono/metadata/gc-internals.h>
  26 #include <mono/metadata/mono-hash-internals.h>
  27 #include <mono/metadata/mono-mlist.h>
  28 #include <mono/metadata/threadpool.h>
  29 #include <mono/utils/atomic.h>
  30 #include <mono/utils/mono-threads.h>
  31 #include <mono/utils/mono-lazy-init.h>
  32 #include <mono/utils/mono-logger-internals.h>
  33 #include <mono/utils/w32api.h>
  34
  35 typedef struct {
  36         gboolean (*init) (gint wakeup_pipe_fd);
  37         void     (*register_fd) (gint fd, gint events, gboolean is_new);
  38         void     (*remove_fd) (gint fd);
  39         gint     (*event_wait) (void (*callback) (gint fd, gint events, gpointer user_data), gpointer user_data);
  40 } ThreadPoolIOBackend;
  41
  42 /* Keep in sync with System.IOOperation in mcs/class/System/System/IOSelector.cs */
  43 enum MonoIOOperation {
  44         EVENT_IN   = 1 << 0,
  45         EVENT_OUT  = 1 << 1,
  46         EVENT_ERR  = 1 << 2, /* not in managed */
  47 };
  48
  49 #include "threadpool-io-epoll.c"
  50 #include "threadpool-io-kqueue.c"
  51 #include "threadpool-io-poll.c"
  52
  53 #define UPDATES_CAPACITY 128
  54
  55 /* Keep in sync with System.IOSelectorJob in mcs/class/System/System/IOSelector.cs */
  56 struct _MonoIOSelectorJob {
  57         MonoObject object;
  58         gint32 operation;
  59         MonoObject *callback;
  60         MonoObject *state;
  61 };
  62
  63 typedef enum {
  64         UPDATE_EMPTY = 0,
  65         UPDATE_ADD,
  66         UPDATE_REMOVE_SOCKET,
  67         UPDATE_REMOVE_DOMAIN,
  68 } ThreadPoolIOUpdateType;
  69
  70 typedef struct {
  71         gint fd;
  72         MonoIOSelectorJob *job;
  73 } ThreadPoolIOUpdate_Add;
  74
  75 typedef struct {
  76         gint fd;
  77 } ThreadPoolIOUpdate_RemoveSocket;
  78
  79 typedef struct {
  80         MonoDomain *domain;
  81 } ThreadPoolIOUpdate_RemoveDomain;
  82
  83 typedef struct {
  84         ThreadPoolIOUpdateType type;
  85         union {
  86                 ThreadPoolIOUpdate_Add add;
  87                 ThreadPoolIOUpdate_RemoveSocket remove_socket;
  88                 ThreadPoolIOUpdate_RemoveDomain remove_domain;
  89         } data;
  90 } ThreadPoolIOUpdate;
  91
  92 typedef struct {
  93         ThreadPoolIOBackend backend;
  94
  95         ThreadPoolIOUpdate updates [UPDATES_CAPACITY];
  96         gint updates_size;
  97         MonoCoopMutex updates_lock;
  98         MonoCoopCond updates_cond;
  99
 100 #if !defined(HOST_WIN32)
 101         gint wakeup_pipes [2];
 102 #else
 103         SOCKET wakeup_pipes [2];
 104 #endif
 105 } ThreadPoolIO;
 106
 107 static mono_lazy_init_t io_status = MONO_LAZY_INIT_STATUS_NOT_INITIALIZED;
 108
 109 static gboolean io_selector_running = FALSE;
 110
 111 static ThreadPoolIO* threadpool_io;
 112
 113 static MonoIOSelectorJob*
 114 get_job_for_event (MonoMList **list, gint32 event)
 115 {
 116         MonoMList *current;
 117
 118         g_assert (list);
 119
 120         for (current = *list; current; current = mono_mlist_next (current)) {
 121                 MonoIOSelectorJob *job = (MonoIOSelectorJob*) mono_mlist_get_data (current);
 122                 if (job->operation == event) {
 123                         *list = mono_mlist_remove_item (*list, current);
 124                         mono_mlist_set_data (current, NULL);
 125                         return job;
 126                 }
 127         }
 128
 129         return NULL;
 130 }
 131
 132 static gint
 133 get_operations_for_jobs (MonoMList *list)
 134 {
 135         MonoMList *current;
 136         gint operations = 0;
 137
 138         for (current = list; current; current = mono_mlist_next (current))
 139                 operations |= ((MonoIOSelectorJob*) mono_mlist_get_data (current))->operation;
 140
 141         return operations;
 142 }
 143
 144 static void
 145 selector_thread_wakeup (void)
 146 {
 147         gchar msg = 'c';
 148         gint written;
 149
 150         for (;;) {
 151 #if !defined(HOST_WIN32)
 152                 written = write (threadpool_io->wakeup_pipes [1], &msg, 1);
 153                 if (written == 1)
 154                         break;
 155                 if (written == -1) {
 156                         g_warning ("selector_thread_wakeup: write () failed, error (%d) %s\n", errno, g_strerror (errno));
 157                         break;
 158                 }
 159 #else
 160                 written = send (threadpool_io->wakeup_pipes [1], &msg, 1, 0);
 161                 if (written == 1)
 162                         break;
 163                 if (written == SOCKET_ERROR) {
 164                         g_warning ("selector_thread_wakeup: write () failed, error (%d)\n", WSAGetLastError ());
 165                         break;
 166                 }
 167 #endif
 168         }
 169 }
 170
 171 static void
 172 selector_thread_wakeup_drain_pipes (void)
 173 {
 174         gchar buffer [128];
 175         gint received;
 176
 177         for (;;) {
 178 #if !defined(HOST_WIN32)
 179                 received = read (threadpool_io->wakeup_pipes [0], buffer, sizeof (buffer));
 180                 if (received == 0)
 181                         break;
 182                 if (received == -1) {
 183 #ifdef ERESTART
 184                         /*
 185                          * some unices (like AIX) send ERESTART, which doesn't
 186                          * exist on some other OSes errno
 187                          */
 188                         if (errno != EINTR && errno != EAGAIN && errno != ERESTART)
 189 #else
 190                         if (errno != EINTR && errno != EAGAIN)
 191 #endif
 192                                 g_warning ("selector_thread_wakeup_drain_pipes: read () failed, error (%d) %s\n", errno, g_strerror (errno));
 193                         break;
 194                 }
 195 #else
 196                 received = recv (threadpool_io->wakeup_pipes [0], buffer, sizeof (buffer), 0);
 197                 if (received == 0)
 198                         break;
 199                 if (received == SOCKET_ERROR) {
 200                         if (WSAGetLastError () != WSAEINTR && WSAGetLastError () != WSAEWOULDBLOCK)
 201                                 g_warning ("selector_thread_wakeup_drain_pipes: recv () failed, error (%d)\n", WSAGetLastError ());
 202                         break;
 203                 }
 204 #endif
 205         }
 206 }
 207
 208 typedef struct {
 209         MonoDomain *domain;
 210         MonoGHashTable *states;
 211 } FilterSockaresForDomainData;
 212
 213 static void
 214 filter_jobs_for_domain (gpointer key, gpointer value, gpointer user_data)
 215 {
 216         FilterSockaresForDomainData *data;
 217         MonoMList *list = (MonoMList *)value, *element;
 218         MonoDomain *domain;
 219         MonoGHashTable *states;
 220
 221         g_assert (user_data);
 222         data = (FilterSockaresForDomainData *)user_data;
 223         domain = data->domain;
 224         states = data->states;
 225
 226         for (element = list; element; element = mono_mlist_next (element)) {
 227                 MonoIOSelectorJob *job = (MonoIOSelectorJob*) mono_mlist_get_data (element);
 228                 if (mono_object_domain (job) == domain)
 229                         mono_mlist_set_data (element, NULL);
 230         }
 231
 232         /* we skip all the first elements which are NULL */
 233         for (; list; list = mono_mlist_next (list)) {
 234                 if (mono_mlist_get_data (list))
 235                         break;
 236         }
 237
 238         if (list) {
 239                 g_assert (mono_mlist_get_data (list));
 240
 241                 /* we delete all the NULL elements after the first one */
 242                 for (element = list; element;) {
 243                         MonoMList *next;
 244                         if (!(next = mono_mlist_next (element)))
 245                                 break;
 246                         if (mono_mlist_get_data (next))
 247                                 element = next;
 248                         else
 249                                 mono_mlist_set_next (element, mono_mlist_next (next));
 250                 }
 251         }
 252
 253         mono_g_hash_table_replace (states, key, list);
 254 }
 255
 256 static void
 257 wait_callback (gint fd, gint events, gpointer user_data)
 258 {
 259         ERROR_DECL (error);
 260
 261         if (mono_runtime_is_shutting_down ())
 262                 return;
 263
 264         if (fd == threadpool_io->wakeup_pipes [0]) {
 265                 mono_trace (G_LOG_LEVEL_DEBUG, MONO_TRACE_IO_SELECTOR, "io threadpool: wke");
 266                 selector_thread_wakeup_drain_pipes ();
 267         } else {
 268                 MonoGHashTable *states;
 269                 MonoMList *list = NULL;
 270                 gpointer k;
 271                 gboolean remove_fd = FALSE;
 272                 gint operations;
 273
 274                 g_assert (user_data);
 275                 states = (MonoGHashTable *)user_data;
 276
 277                 mono_trace (G_LOG_LEVEL_DEBUG, MONO_TRACE_IO_SELECTOR, "io threadpool: cal fd %3d, events = %2s | %2s | %3s",
 278                         fd, (events & EVENT_IN) ? "RD" : "..", (events & EVENT_OUT) ? "WR" : "..", (events & EVENT_ERR) ? "ERR" : "...");
 279
 280                 if (!mono_g_hash_table_lookup_extended (states, GINT_TO_POINTER (fd), &k, (gpointer*) &list))
 281                         g_error ("wait_callback: fd %d not found in states table", fd);
 282
 283                 if (list && (events & EVENT_IN) != 0) {
 284                         MonoIOSelectorJob *job = get_job_for_event (&list, EVENT_IN);
 285                         if (job) {
 286                                 mono_threadpool_enqueue_work_item (((MonoObject*) job)->vtable->domain, (MonoObject*) job, error);
 287                                 mono_error_assert_ok (error);
 288                         }
 289
 290                 }
 291                 if (list && (events & EVENT_OUT) != 0) {
 292                         MonoIOSelectorJob *job = get_job_for_event (&list, EVENT_OUT);
 293                         if (job) {
 294                                 mono_threadpool_enqueue_work_item (((MonoObject*) job)->vtable->domain, (MonoObject*) job, error);
 295                                 mono_error_assert_ok (error);
 296                         }
 297                 }
 298
 299                 remove_fd = (events & EVENT_ERR) == EVENT_ERR;
 300                 if (!remove_fd) {
 301                         mono_g_hash_table_replace (states, GINT_TO_POINTER (fd), list);
 302
 303                         operations = get_operations_for_jobs (list);
 304
 305                         mono_trace (G_LOG_LEVEL_DEBUG, MONO_TRACE_IO_SELECTOR, "io threadpool: res fd %3d, events = %2s | %2s | %3s",
 306                                 fd, (operations & EVENT_IN) ? "RD" : "..", (operations & EVENT_OUT) ? "WR" : "..", (operations & EVENT_ERR) ? "ERR" : "...");
 307
 308                         threadpool_io->backend.register_fd (fd, operations, FALSE);
 309                 } else {
 310                         mono_trace (G_LOG_LEVEL_DEBUG, MONO_TRACE_IO_SELECTOR, "io threadpool: err fd %d", fd);
 311
 312                         mono_g_hash_table_remove (states, GINT_TO_POINTER (fd));
 313
 314                         threadpool_io->backend.remove_fd (fd);
 315                 }
 316         }
 317 }
 318
 319 static void
 320 selector_thread_interrupt (gpointer unused)
 321 {
 322         selector_thread_wakeup ();
 323 }
 324
 325 static gsize WINAPI
 326 selector_thread (gpointer data)
 327 {
 328         ERROR_DECL (error);
 329         MonoGHashTable *states;
 330
 331         MonoString *thread_name = mono_string_new_checked (mono_get_root_domain (), "Thread Pool I/O Selector", error);
 332         mono_error_assert_ok (error);
 333         mono_thread_set_name_internal (mono_thread_internal_current (), thread_name, FALSE, TRUE, error);
 334         mono_error_assert_ok (error);
 335
 336         if (mono_runtime_is_shutting_down ()) {
 337                 io_selector_running = FALSE;
 338                 return 0;
 339         }
 340
 341         states = mono_g_hash_table_new_type_internal (g_direct_hash, NULL, MONO_HASH_VALUE_GC, MONO_ROOT_SOURCE_THREAD_POOL, NULL, "Thread Pool I/O State Table");
 342
 343         while (!mono_runtime_is_shutting_down ()) {
 344                 gint i, j;
 345                 gint res;
 346                 gboolean interrupted = FALSE;
 347
 348                 if (mono_thread_interruption_checkpoint_bool ())
 349                         continue;
 350
 351                 mono_coop_mutex_lock (&threadpool_io->updates_lock);
 352
 353                 for (i = 0; i < threadpool_io->updates_size; ++i) {
 354                         ThreadPoolIOUpdate *update = &threadpool_io->updates [i];
 355
 356                         switch (update->type) {
 357                         case UPDATE_EMPTY:
 358                                 break;
 359                         case UPDATE_ADD: {
 360                                 gint fd;
 361                                 gint operations;
 362                                 gpointer k;
 363                                 gboolean exists;
 364                                 MonoMList *list = NULL;
 365                                 MonoIOSelectorJob *job;
 366
 367                                 fd = update->data.add.fd;
 368                                 g_assert (fd >= 0);
 369
 370                                 job = update->data.add.job;
 371                                 g_assert (job);
 372
 373                                 exists = mono_g_hash_table_lookup_extended (states, GINT_TO_POINTER (fd), &k, (gpointer*) &list);
 374                                 list = mono_mlist_append_checked (list, (MonoObject*) job, error);
 375                                 mono_error_assert_ok (error);
 376                                 mono_g_hash_table_replace (states, GINT_TO_POINTER (fd), list);
 377
 378                                 operations = get_operations_for_jobs (list);
 379
 380                                 mono_trace (G_LOG_LEVEL_DEBUG, MONO_TRACE_IO_SELECTOR, "io threadpool: %3s fd %3d, operations = %2s | %2s | %3s",
 381                                         exists ? "mod" : "add", fd, (operations & EVENT_IN) ? "RD" : "..", (operations & EVENT_OUT) ? "WR" : "..", (operations & EVENT_ERR) ? "ERR" : "...");
 382
 383                                 threadpool_io->backend.register_fd (fd, operations, !exists);
 384
 385                                 break;
 386                         }
 387                         case UPDATE_REMOVE_SOCKET: {
 388                                 gint fd;
 389                                 gpointer k;
 390                                 MonoMList *list = NULL;
 391
 392                                 fd = update->data.remove_socket.fd;
 393                                 g_assert (fd >= 0);
 394
 395                                 if (mono_g_hash_table_lookup_extended (states, GINT_TO_POINTER (fd), &k, (gpointer*) &list)) {
 396                                         mono_g_hash_table_remove (states, GINT_TO_POINTER (fd));
 397
 398                                         for (j = i + 1; j < threadpool_io->updates_size; ++j) {
 399                                                 ThreadPoolIOUpdate *update = &threadpool_io->updates [j];
 400                                                 if (update->type == UPDATE_ADD && update->data.add.fd == fd)
 401                                                         memset (update, 0, sizeof (ThreadPoolIOUpdate));
 402                                         }
 403
 404                                         for (; list; list = mono_mlist_remove_item (list, list)) {
 405                                                 mono_threadpool_enqueue_work_item (mono_object_domain (mono_mlist_get_data (list)), mono_mlist_get_data (list), error);
 406                                                 mono_mlist_set_data (list, NULL);
 407                                                 mono_error_assert_ok (error);
 408                                         }
 409
 410                                         mono_trace (G_LOG_LEVEL_DEBUG, MONO_TRACE_IO_SELECTOR, "io threadpool: del fd %3d", fd);
 411                                         threadpool_io->backend.remove_fd (fd);
 412                                 }
 413
 414                                 break;
 415                         }
 416                         case UPDATE_REMOVE_DOMAIN: {
 417                                 MonoDomain *domain;
 418
 419                                 domain = update->data.remove_domain.domain;
 420                                 g_assert (domain);
 421
 422                                 FilterSockaresForDomainData user_data;
 423                                 memset (&user_data, 0, sizeof (user_data));
 424                                 user_data.domain = domain;
 425                                 user_data.states = states;
 426                                 mono_g_hash_table_foreach (states, filter_jobs_for_domain, &user_data);
 427
 428                                 for (j = i + 1; j < threadpool_io->updates_size; ++j) {
 429                                         ThreadPoolIOUpdate *update = &threadpool_io->updates [j];
 430                                         if (update->type == UPDATE_ADD && mono_object_domain (update->data.add.job) == domain)
 431                                                 memset (update, 0, sizeof (ThreadPoolIOUpdate));
 432                                 }
 433
 434                                 break;
 435                         }
 436                         default:
 437                                 g_assert_not_reached ();
 438                         }
 439                 }
 440
 441                 mono_coop_cond_broadcast (&threadpool_io->updates_cond);
 442
 443                 if (threadpool_io->updates_size > 0) {
 444                         threadpool_io->updates_size = 0;
 445                         memset (&threadpool_io->updates, 0, UPDATES_CAPACITY * sizeof (ThreadPoolIOUpdate));
 446                 }
 447
 448                 mono_coop_mutex_unlock (&threadpool_io->updates_lock);
 449
 450                 mono_trace (G_LOG_LEVEL_DEBUG, MONO_TRACE_IO_SELECTOR, "io threadpool: wai");
 451
 452                 mono_thread_info_install_interrupt (selector_thread_interrupt, NULL, &interrupted);
 453                 if (interrupted)
 454                         continue;
 455
 456                 res = threadpool_io->backend.event_wait (wait_callback, states);
 457                 if (res == -1)
 458                         break;
 459
 460                 mono_thread_info_uninstall_interrupt (&interrupted);
 461         }
 462
 463         mono_g_hash_table_destroy (states);
 464
 465         mono_coop_mutex_lock (&threadpool_io->updates_lock);
 466
 467         io_selector_running = FALSE;
 468         mono_coop_cond_broadcast (&threadpool_io->updates_cond);
 469
 470         mono_coop_mutex_unlock (&threadpool_io->updates_lock);
 471
 472         return 0;
 473 }
 474
 475 /* Locking: threadpool_io->updates_lock must be held */
 476 static ThreadPoolIOUpdate*
 477 update_get_new (void)
 478 {
 479         ThreadPoolIOUpdate *update = NULL;
 480         g_assert (threadpool_io->updates_size <= UPDATES_CAPACITY);
 481
 482         while (threadpool_io->updates_size == UPDATES_CAPACITY) {
 483                 /* we wait for updates to be applied in the selector_thread and we loop
 484                  * as long as none are available. if it happends too much, then we need
 485                  * to increase UPDATES_CAPACITY */
 486                 mono_coop_cond_wait (&threadpool_io->updates_cond, &threadpool_io->updates_lock);
 487         }
 488
 489         g_assert (threadpool_io->updates_size < UPDATES_CAPACITY);
 490
 491         update = &threadpool_io->updates [threadpool_io->updates_size ++];
 492
 493         return update;
 494 }
 495
 496 static void
 497 wakeup_pipes_init (void)
 498 {
 499 #if !defined(HOST_WIN32)
 500         if (pipe (threadpool_io->wakeup_pipes) == -1)
 501                 g_error ("wakeup_pipes_init: pipe () failed, error (%d) %s\n", errno, g_strerror (errno));
 502         if (fcntl (threadpool_io->wakeup_pipes [0], F_SETFL, O_NONBLOCK) == -1)
 503                 g_error ("wakeup_pipes_init: fcntl () failed, error (%d) %s\n", errno, g_strerror (errno));
 504 #else
 505         struct sockaddr_in client;
 506         struct sockaddr_in server;
 507         SOCKET server_sock;
 508         gulong arg;
 509         gint size;
 510
 511         server_sock = socket (AF_INET, SOCK_STREAM, IPPROTO_TCP);
 512         g_assert (server_sock != INVALID_SOCKET);
 513         threadpool_io->wakeup_pipes [1] = socket (AF_INET, SOCK_STREAM, IPPROTO_TCP);
 514         g_assert (threadpool_io->wakeup_pipes [1] != INVALID_SOCKET);
 515
 516         server.sin_family = AF_INET;
 517         server.sin_addr.s_addr = inet_addr ("127.0.0.1");
 518         server.sin_port = 0;
 519         if (bind (server_sock, (SOCKADDR*) &server, sizeof (server)) == SOCKET_ERROR) {
 520                 closesocket (server_sock);
 521                 g_error ("wakeup_pipes_init: bind () failed, error (%d)\n", WSAGetLastError ());
 522         }
 523
 524         size = sizeof (server);
 525         if (getsockname (server_sock, (SOCKADDR*) &server, &size) == SOCKET_ERROR) {
 526                 closesocket (server_sock);
 527                 g_error ("wakeup_pipes_init: getsockname () failed, error (%d)\n", WSAGetLastError ());
 528         }
 529         if (listen (server_sock, 1024) == SOCKET_ERROR) {
 530                 closesocket (server_sock);
 531                 g_error ("wakeup_pipes_init: listen () failed, error (%d)\n", WSAGetLastError ());
 532         }
 533         if (connect ((SOCKET) threadpool_io->wakeup_pipes [1], (SOCKADDR*) &server, sizeof (server)) == SOCKET_ERROR) {
 534                 closesocket (server_sock);
 535                 g_error ("wakeup_pipes_init: connect () failed, error (%d)\n", WSAGetLastError ());
 536         }
 537
 538         size = sizeof (client);
 539         threadpool_io->wakeup_pipes [0] = accept (server_sock, (SOCKADDR *) &client, &size);
 540         g_assert (threadpool_io->wakeup_pipes [0] != INVALID_SOCKET);
 541
 542         arg = 1;
 543         if (ioctlsocket (threadpool_io->wakeup_pipes [0], FIONBIO, &arg) == SOCKET_ERROR) {
 544                 closesocket (threadpool_io->wakeup_pipes [0]);
 545                 closesocket (server_sock);
 546                 g_error ("wakeup_pipes_init: ioctlsocket () failed, error (%d)\n", WSAGetLastError ());
 547         }
 548
 549         closesocket (server_sock);
 550 #endif
 551 }
 552
 553 static void
 554 initialize (void)
 555 {
 556         g_assert (!threadpool_io);
 557         threadpool_io = g_new0 (ThreadPoolIO, 1);
 558         g_assert (threadpool_io);
 559
 560         mono_coop_mutex_init (&threadpool_io->updates_lock);
 561         mono_coop_cond_init (&threadpool_io->updates_cond);
 562         mono_gc_register_root ((char *)&threadpool_io->updates [0], sizeof (threadpool_io->updates), MONO_GC_DESCRIPTOR_NULL, MONO_ROOT_SOURCE_THREAD_POOL, NULL, "Thread Pool I/O Update List");
 563
 564         threadpool_io->updates_size = 0;
 565
 566         threadpool_io->backend = backend_poll;
 567         if (g_hasenv ("MONO_ENABLE_AIO")) {
 568 #if defined(HAVE_EPOLL)
 569                 threadpool_io->backend = backend_epoll;
 570 #elif defined(HAVE_KQUEUE)
 571                 threadpool_io->backend = backend_kqueue;
 572 #endif
 573         }
 574
 575         wakeup_pipes_init ();
 576
 577         if (!threadpool_io->backend.init (threadpool_io->wakeup_pipes [0]))
 578                 g_error ("initialize: backend->init () failed");
 579
 580         mono_coop_mutex_lock (&threadpool_io->updates_lock);
 581
 582         io_selector_running = TRUE;
 583
 584         ERROR_DECL (error);
 585         if (!mono_thread_create_internal (mono_get_root_domain (), (gpointer)selector_thread, NULL, (MonoThreadCreateFlags)(MONO_THREAD_CREATE_FLAGS_THREADPOOL | MONO_THREAD_CREATE_FLAGS_SMALL_STACK), error))
 586                 g_error ("initialize: mono_thread_create_internal () failed due to %s", mono_error_get_message (error));
 587
 588         mono_coop_mutex_unlock (&threadpool_io->updates_lock);
 589 }
 590
 591 static void
 592 cleanup (void)
 593 {
 594         // FIXME destroy everything
 595 }
 596
 597 void
 598 mono_threadpool_io_cleanup (void)
 599 {
 600         mono_lazy_cleanup (&io_status, cleanup);
 601 }
 602
 603 void
 604 ves_icall_System_IOSelector_Add (gpointer handle, MonoIOSelectorJob *job)
 605 {
 606         ThreadPoolIOUpdate *update;
 607
 608         g_assert (handle);
 609
 610         g_assert ((job->operation == EVENT_IN) ^ (job->operation == EVENT_OUT));
 611         g_assert (job->callback);
 612
 613         if (mono_runtime_is_shutting_down ())
 614                 return;
 615         if (mono_domain_is_unloading (mono_object_domain (job)))
 616                 return;
 617
 618         mono_lazy_initialize (&io_status, initialize);
 619
 620         mono_coop_mutex_lock (&threadpool_io->updates_lock);
 621
 622         if (!io_selector_running) {
 623                 mono_coop_mutex_unlock (&threadpool_io->updates_lock);
 624                 return;
 625         }
 626
 627         update = update_get_new ();
 628         update->type = UPDATE_ADD;
 629         update->data.add.fd = GPOINTER_TO_INT (handle);
 630         update->data.add.job = job;
 631         mono_memory_barrier (); /* Ensure this is safely published before we wake up the selector */
 632
 633         selector_thread_wakeup ();
 634
 635         mono_coop_mutex_unlock (&threadpool_io->updates_lock);
 636 }
 637
 638 void
 639 ves_icall_System_IOSelector_Remove (gpointer handle)
 640 {
 641         mono_threadpool_io_remove_socket (GPOINTER_TO_INT (handle));
 642 }
 643
 644 void
 645 mono_threadpool_io_remove_socket (int fd)
 646 {
 647         ThreadPoolIOUpdate *update;
 648
 649         if (!mono_lazy_is_initialized (&io_status))
 650                 return;
 651
 652         mono_coop_mutex_lock (&threadpool_io->updates_lock);
 653
 654         if (!io_selector_running) {
 655                 mono_coop_mutex_unlock (&threadpool_io->updates_lock);
 656                 return;
 657         }
 658
 659         update = update_get_new ();
 660         update->type = UPDATE_REMOVE_SOCKET;
 661         update->data.add.fd = fd;
 662         mono_memory_barrier (); /* Ensure this is safely published before we wake up the selector */
 663
 664         selector_thread_wakeup ();
 665
 666         mono_coop_cond_wait (&threadpool_io->updates_cond, &threadpool_io->updates_lock);
 667
 668         mono_coop_mutex_unlock (&threadpool_io->updates_lock);
 669 }
 670
 671 void
 672 mono_threadpool_io_remove_domain_jobs (MonoDomain *domain)
 673 {
 674         ThreadPoolIOUpdate *update;
 675
 676         if (!mono_lazy_is_initialized (&io_status))
 677                 return;
 678
 679         mono_coop_mutex_lock (&threadpool_io->updates_lock);
 680
 681         if (!io_selector_running) {
 682                 mono_coop_mutex_unlock (&threadpool_io->updates_lock);
 683                 return;
 684         }
 685
 686         update = update_get_new ();
 687         update->type = UPDATE_REMOVE_DOMAIN;
 688         update->data.remove_domain.domain = domain;
 689         mono_memory_barrier (); /* Ensure this is safely published before we wake up the selector */
 690
 691         selector_thread_wakeup ();
 692
 693         mono_coop_cond_wait (&threadpool_io->updates_cond, &threadpool_io->updates_lock);
 694
 695         mono_coop_mutex_unlock (&threadpool_io->updates_lock);
 696 }
 697
 698 #else
 699
 700 void
 701 ves_icall_System_IOSelector_Add (gpointer handle, MonoIOSelectorJob *job)
 702 {
 703         g_assert_not_reached ();
 704 }
 705
 706 void
 707 ves_icall_System_IOSelector_Remove (gpointer handle)
 708 {
 709         g_assert_not_reached ();
 710 }
 711
 712 void
 713 mono_threadpool_io_cleanup (void)
 714 {
 715         g_assert_not_reached ();
 716 }
 717
 718 void
 719 mono_threadpool_io_remove_socket (int fd)
 720 {
 721         g_assert_not_reached ();
 722 }
 723
 724 void
 725 mono_threadpool_io_remove_domain_jobs (MonoDomain *domain)
 726 {
 727         g_assert_not_reached ();
 728 }
 729
 730 #endif