mono/metadata/threadpool.c

   1 /*
   2  * threadpool.c: global thread pool
   3  *
   4  * Authors:
   5  *   Dietmar Maurer (dietmar@ximian.com)
   6  *   Gonzalo Paniagua Javier (gonzalo@ximian.com)
   7  *
   8  * Copyright 2001-2003 Ximian, Inc (http://www.ximian.com)
   9  * Copyright 2004-2010 Novell, Inc (http://www.novell.com)
  10  * Copyright 2001 Xamarin Inc (http://www.xamarin.com)
  11  */
  12
  13 #include <config.h>
  14 #include <glib.h>
  15
  16 #include <mono/metadata/profiler-private.h>
  17 #include <mono/metadata/threads.h>
  18 #include <mono/metadata/threads-types.h>
  19 #include <mono/metadata/threadpool-internals.h>
  20 #include <mono/metadata/exception.h>
  21 #include <mono/metadata/environment.h>
  22 #include <mono/metadata/mono-mlist.h>
  23 #include <mono/metadata/mono-perfcounters.h>
  24 #include <mono/metadata/socket-io.h>
  25 #include <mono/metadata/mono-cq.h>
  26 #include <mono/metadata/mono-wsq.h>
  27 #include <mono/metadata/mono-ptr-array.h>
  28 #include <mono/io-layer/io-layer.h>
  29 #include <mono/utils/mono-time.h>
  30 #include <mono/utils/mono-proclib.h>
  31 #include <mono/utils/mono-semaphore.h>
  32 #include <mono/utils/atomic.h>
  33 #include <errno.h>
  34 #ifdef HAVE_SYS_TIME_H
  35 #include <sys/time.h>
  36 #endif
  37 #include <sys/types.h>
  38 #include <fcntl.h>
  39 #ifdef HAVE_UNISTD_H
  40 #include <unistd.h>
  41 #endif
  42 #include <string.h>
  43 #ifdef HAVE_SYS_SOCKET_H
  44 #include <sys/socket.h>
  45 #endif
  46 #include <mono/utils/mono-poll.h>
  47 #ifdef HAVE_EPOLL
  48 #include <sys/epoll.h>
  49 #endif
  50 #ifdef HAVE_KQUEUE
  51 #include <sys/event.h>
  52 #endif
  53
  54
  55 #ifndef DISABLE_SOCKETS
  56 #include "mono/io-layer/socket-wrappers.h"
  57 #endif
  58
  59 #include "threadpool.h"
  60
  61 #define THREAD_WANTS_A_BREAK(t) ((t->state & (ThreadState_StopRequested | \
  62                                                 ThreadState_SuspendRequested)) != 0)
  63
  64 #define SPIN_TRYLOCK(i) (InterlockedCompareExchange (&(i), 1, 0) == 0)
  65 #define SPIN_LOCK(i) do { \
  66                                 if (SPIN_TRYLOCK (i)) \
  67                                         break; \
  68                         } while (1)
  69
  70 #define SPIN_UNLOCK(i) i = 0
  71 #define SMALL_STACK (128 * (sizeof (gpointer) / 4) * 1024)
  72
  73 /* DEBUG: prints tp data every 2s */
  74 #undef DEBUG
  75
  76 /* mono_thread_pool_init called */
  77 static volatile int tp_inited;
  78
  79 enum {
  80         POLL_BACKEND,
  81         EPOLL_BACKEND,
  82         KQUEUE_BACKEND
  83 };
  84
  85 typedef struct {
  86         CRITICAL_SECTION io_lock; /* access to sock_to_state */
  87         int inited; // 0 -> not initialized , 1->initializing, 2->initialized, 3->cleaned up
  88         MonoGHashTable *sock_to_state;
  89
  90         gint event_system;
  91         gpointer event_data;
  92         void (*modify) (gpointer p, int fd, int operation, int events, gboolean is_new);
  93         void (*wait) (gpointer sock_data);
  94         void (*shutdown) (gpointer event_data);
  95 } SocketIOData;
  96
  97 static SocketIOData socket_io_data;
  98
  99 /* Keep in sync with the System.MonoAsyncCall class which provides GC tracking */
 100 typedef struct {
 101         MonoObject         object;
 102         MonoMethodMessage *msg;
 103         MonoMethod        *cb_method;
 104         MonoDelegate      *cb_target;
 105         MonoObject        *state;
 106         MonoObject        *res;
 107         MonoArray         *out_args;
 108 } ASyncCall;
 109
 110 typedef struct {
 111         MonoSemType lock;
 112         MonoCQ *queue; /* GC root */
 113         MonoSemType new_job;
 114         volatile gint waiting; /* threads waiting for a work item */
 115
 116         /**/
 117         volatile gint pool_status; /* 0 -> not initialized, 1 -> initialized, 2 -> cleaning up */
 118         /* min, max, n and busy -> Interlocked */
 119         volatile gint min_threads;
 120         volatile gint max_threads;
 121         volatile gint nthreads;
 122         volatile gint busy_threads;
 123
 124         void (*async_invoke) (gpointer data);
 125         void *pc_nitems; /* Performance counter for total number of items in added */
 126         void *pc_nthreads; /* Performance counter for total number of active threads */
 127         /**/
 128         volatile gint destroy_thread;
 129         volatile gint ignore_times; /* Used when there's a thread being created or destroyed */
 130         volatile gint sp_lock; /* spin lock used to protect ignore_times */
 131         volatile gint64 last_check;
 132         volatile gint64 time_sum;
 133         volatile gint n_sum;
 134         gint64 averages [2];
 135         gboolean is_io;
 136 } ThreadPool;
 137
 138 static ThreadPool async_tp;
 139 static ThreadPool async_io_tp;
 140
 141 static void async_invoke_thread (gpointer data);
 142 static MonoObject *mono_async_invoke (ThreadPool *tp, MonoAsyncResult *ares);
 143 static void threadpool_free_queue (ThreadPool *tp);
 144 static void threadpool_append_job (ThreadPool *tp, MonoObject *ar);
 145 static void threadpool_append_jobs (ThreadPool *tp, MonoObject **jobs, gint njobs);
 146 static void threadpool_init (ThreadPool *tp, int min_threads, int max_threads, void (*async_invoke) (gpointer));
 147 static void threadpool_start_idle_threads (ThreadPool *tp);
 148 static void threadpool_kill_idle_threads (ThreadPool *tp);
 149 static gboolean threadpool_start_thread (ThreadPool *tp);
 150 static void monitor_thread (gpointer data);
 151 static void socket_io_cleanup (SocketIOData *data);
 152 static MonoObject *get_io_event (MonoMList **list, gint event);
 153 static int get_events_from_list (MonoMList *list);
 154 static int get_event_from_state (MonoSocketAsyncResult *state);
 155 static void check_for_interruption_critical (void);
 156
 157 static MonoClass *async_call_klass;
 158 static MonoClass *socket_async_call_klass;
 159 static MonoClass *process_async_call_klass;
 160
 161 static GPtrArray *wsqs;
 162 CRITICAL_SECTION wsqs_lock;
 163 static gboolean suspended;
 164
 165 /* Hooks */
 166 static MonoThreadPoolFunc tp_start_func;
 167 static MonoThreadPoolFunc tp_finish_func;
 168 static gpointer tp_hooks_user_data;
 169 static MonoThreadPoolItemFunc tp_item_begin_func;
 170 static MonoThreadPoolItemFunc tp_item_end_func;
 171 static gpointer tp_item_user_data;
 172
 173 enum {
 174         AIO_OP_FIRST,
 175         AIO_OP_ACCEPT = 0,
 176         AIO_OP_CONNECT,
 177         AIO_OP_RECEIVE,
 178         AIO_OP_RECEIVEFROM,
 179         AIO_OP_SEND,
 180         AIO_OP_SENDTO,
 181         AIO_OP_RECV_JUST_CALLBACK,
 182         AIO_OP_SEND_JUST_CALLBACK,
 183         AIO_OP_READPIPE,
 184         AIO_OP_CONSOLE2,
 185         AIO_OP_DISCONNECT,
 186         AIO_OP_ACCEPTRECEIVE,
 187         AIO_OP_RECEIVE_BUFFERS,
 188         AIO_OP_SEND_BUFFERS,
 189         AIO_OP_LAST
 190 };
 191
 192 #include <mono/metadata/tpool-poll.c>
 193 #ifdef HAVE_EPOLL
 194 #include <mono/metadata/tpool-epoll.c>
 195 #elif defined(USE_KQUEUE_FOR_THREADPOOL)
 196 #include <mono/metadata/tpool-kqueue.c>
 197 #endif
 198 /*
 199  * Functions to check whenever a class is given system class. We need to cache things in MonoDomain since some of the
 200  * assemblies can be unloaded.
 201  */
 202
 203 static gboolean
 204 is_system_type (MonoDomain *domain, MonoClass *klass)
 205 {
 206         if (domain->system_image == NULL)
 207                 domain->system_image = mono_image_loaded ("System");
 208
 209         return klass->image == domain->system_image;
 210 }
 211
 212 static gboolean
 213 is_corlib_type (MonoDomain *domain, MonoClass *klass)
 214 {
 215         return klass->image == mono_defaults.corlib;
 216 }
 217
 218 /*
 219  * Note that we call it is_socket_type() where 'socket' refers to the image
 220  * that contains the System.Net.Sockets.Socket type.
 221 */
 222 static gboolean
 223 is_socket_type (MonoDomain *domain, MonoClass *klass)
 224 {
 225         return is_system_type (domain, klass);
 226 }
 227
 228 #define check_type_cached(domain, ASSEMBLY, _class, _namespace, _name, loc) do { \
 229         if (*loc) \
 230                 return *loc == _class; \
 231         if (is_##ASSEMBLY##_type (domain, _class) && !strcmp (_name, _class->name) && !strcmp (_namespace, _class->name_space)) { \
 232                 *loc = _class; \
 233                 return TRUE; \
 234         } \
 235         return FALSE; \
 236 } while (0) \
 237
 238 #define check_corlib_type_cached(domain, _class, _namespace, _name, loc) check_type_cached (domain, corlib, _class, _namespace, _name, loc)
 239
 240 #define check_socket_type_cached(domain, _class, _namespace, _name, loc) check_type_cached (domain, socket, _class, _namespace, _name, loc)
 241
 242 #define check_system_type_cached(domain, _class, _namespace, _name, loc) check_type_cached (domain, system, _class, _namespace, _name, loc)
 243
 244 static gboolean
 245 is_corlib_asyncresult (MonoDomain *domain, MonoClass *klass)
 246 {
 247         check_corlib_type_cached (domain, klass, "System.Runtime.Remoting.Messaging", "AsyncResult", &domain->corlib_asyncresult_class);
 248 }
 249
 250 static gboolean
 251 is_socket (MonoDomain *domain, MonoClass *klass)
 252 {
 253         check_socket_type_cached (domain, klass, "System.Net.Sockets", "Socket", &domain->socket_class);
 254 }
 255
 256 static gboolean
 257 is_socketasyncresult (MonoDomain *domain, MonoClass *klass)
 258 {
 259         return (klass->nested_in &&
 260                         is_socket (domain, klass->nested_in) &&
 261                         !strcmp (klass->name, "SocketAsyncResult"));
 262 }
 263
 264 static gboolean
 265 is_socketasynccall (MonoDomain *domain, MonoClass *klass)
 266 {
 267         return (klass->nested_in &&
 268                         is_socket (domain, klass->nested_in) &&
 269                         !strcmp (klass->name, "SocketAsyncCall"));
 270 }
 271
 272 static gboolean
 273 is_appdomainunloaded_exception (MonoDomain *domain, MonoClass *klass)
 274 {
 275         check_corlib_type_cached (domain, klass, "System", "AppDomainUnloadedException", &domain->ad_unloaded_ex_class);
 276 }
 277
 278 static gboolean
 279 is_sd_process (MonoDomain *domain, MonoClass *klass)
 280 {
 281         check_system_type_cached (domain, klass, "System.Diagnostics", "Process", &domain->process_class);
 282 }
 283
 284 static gboolean
 285 is_sdp_asyncreadhandler (MonoDomain *domain, MonoClass *klass)
 286 {
 287
 288         return (klass->nested_in &&
 289                         is_sd_process (domain, klass->nested_in) &&
 290                 !strcmp (klass->name, "AsyncReadHandler"));
 291 }
 292
 293
 294 #ifdef DISABLE_SOCKETS
 295
 296 #define socket_io_cleanup(x)
 297
 298 static int
 299 get_event_from_state (MonoSocketAsyncResult *state)
 300 {
 301         g_assert_not_reached ();
 302         return -1;
 303 }
 304
 305 static int
 306 get_events_from_list (MonoMList *list)
 307 {
 308         return 0;
 309 }
 310
 311 #else
 312
 313 static void
 314 socket_io_cleanup (SocketIOData *data)
 315 {
 316         EnterCriticalSection (&data->io_lock);
 317         if (data->inited != 2) {
 318                 LeaveCriticalSection (&data->io_lock);
 319                 return;
 320         }
 321         data->inited = 3;
 322         data->shutdown (data->event_data);
 323         LeaveCriticalSection (&data->io_lock);
 324 }
 325
 326 static int
 327 get_event_from_state (MonoSocketAsyncResult *state)
 328 {
 329         switch (state->operation) {
 330         case AIO_OP_ACCEPT:
 331         case AIO_OP_RECEIVE:
 332         case AIO_OP_RECV_JUST_CALLBACK:
 333         case AIO_OP_RECEIVEFROM:
 334         case AIO_OP_READPIPE:
 335         case AIO_OP_ACCEPTRECEIVE:
 336         case AIO_OP_RECEIVE_BUFFERS:
 337                 return MONO_POLLIN;
 338         case AIO_OP_SEND:
 339         case AIO_OP_SEND_JUST_CALLBACK:
 340         case AIO_OP_SENDTO:
 341         case AIO_OP_CONNECT:
 342         case AIO_OP_SEND_BUFFERS:
 343         case AIO_OP_DISCONNECT:
 344                 return MONO_POLLOUT;
 345         default: /* Should never happen */
 346                 g_message ("get_event_from_state: unknown value in switch!!!");
 347                 return 0;
 348         }
 349 }
 350
 351 static int
 352 get_events_from_list (MonoMList *list)
 353 {
 354         MonoSocketAsyncResult *state;
 355         int events = 0;
 356
 357         while (list && (state = (MonoSocketAsyncResult *)mono_mlist_get_data (list))) {
 358                 events |= get_event_from_state (state);
 359                 list = mono_mlist_next (list);
 360         }
 361
 362         return events;
 363 }
 364
 365 #define ICALL_RECV(x)   ves_icall_System_Net_Sockets_Socket_Receive_internal (\
 366                                 (SOCKET)(gssize)x->handle, x->buffer, x->offset, x->size,\
 367                                  x->socket_flags, &x->error);
 368
 369 #define ICALL_SEND(x)   ves_icall_System_Net_Sockets_Socket_Send_internal (\
 370                                 (SOCKET)(gssize)x->handle, x->buffer, x->offset, x->size,\
 371                                  x->socket_flags, &x->error);
 372
 373 #endif /* !DISABLE_SOCKETS */
 374
 375 static void
 376 threadpool_jobs_inc (MonoObject *obj)
 377 {
 378         if (obj)
 379                 InterlockedIncrement (&obj->vtable->domain->threadpool_jobs);
 380 }
 381
 382 static gboolean
 383 threadpool_jobs_dec (MonoObject *obj)
 384 {
 385         MonoDomain *domain;
 386         int remaining_jobs;
 387
 388         if (obj == NULL)
 389                 return FALSE;
 390
 391         domain = obj->vtable->domain;
 392         remaining_jobs = InterlockedDecrement (&domain->threadpool_jobs);
 393         if (remaining_jobs == 0 && domain->cleanup_semaphore) {
 394                 ReleaseSemaphore (domain->cleanup_semaphore, 1, NULL);
 395                 return TRUE;
 396         }
 397         return FALSE;
 398 }
 399
 400 static MonoObject *
 401 get_io_event (MonoMList **list, gint event)
 402 {
 403         MonoObject *state;
 404         MonoMList *current;
 405         MonoMList *prev;
 406
 407         current = *list;
 408         prev = NULL;
 409         state = NULL;
 410         while (current) {
 411                 state = mono_mlist_get_data (current);
 412                 if (get_event_from_state ((MonoSocketAsyncResult *) state) == event)
 413                         break;
 414
 415                 state = NULL;
 416                 prev = current;
 417                 current = mono_mlist_next (current);
 418         }
 419
 420         if (current) {
 421                 if (prev) {
 422                         mono_mlist_set_next (prev, mono_mlist_next (current));
 423                 } else {
 424                         *list = mono_mlist_next (*list);
 425                 }
 426         }
 427
 428         return state;
 429 }
 430
 431 /*
 432  * select/poll wake up when a socket is closed, but epoll just removes
 433  * the socket from its internal list without notification.
 434  */
 435 void
 436 mono_thread_pool_remove_socket (int sock)
 437 {
 438         MonoMList *list;
 439         MonoSocketAsyncResult *state;
 440         MonoObject *ares;
 441
 442         if (socket_io_data.inited == 0)
 443                 return;
 444
 445         EnterCriticalSection (&socket_io_data.io_lock);
 446         if (socket_io_data.sock_to_state == NULL) {
 447                 LeaveCriticalSection (&socket_io_data.io_lock);
 448                 return;
 449         }
 450         list = mono_g_hash_table_lookup (socket_io_data.sock_to_state, GINT_TO_POINTER (sock));
 451         if (list)
 452                 mono_g_hash_table_remove (socket_io_data.sock_to_state, GINT_TO_POINTER (sock));
 453         LeaveCriticalSection (&socket_io_data.io_lock);
 454
 455         while (list) {
 456                 state = (MonoSocketAsyncResult *) mono_mlist_get_data (list);
 457                 if (state->operation == AIO_OP_RECEIVE)
 458                         state->operation = AIO_OP_RECV_JUST_CALLBACK;
 459                 else if (state->operation == AIO_OP_SEND)
 460                         state->operation = AIO_OP_SEND_JUST_CALLBACK;
 461
 462                 ares = get_io_event (&list, MONO_POLLIN);
 463                 threadpool_append_job (&async_io_tp, ares);
 464                 if (list) {
 465                         ares = get_io_event (&list, MONO_POLLOUT);
 466                         threadpool_append_job (&async_io_tp, ares);
 467                 }
 468         }
 469 }
 470
 471 static void
 472 init_event_system (SocketIOData *data)
 473 {
 474 #ifdef HAVE_EPOLL
 475         if (data->event_system == EPOLL_BACKEND) {
 476                 data->event_data = tp_epoll_init (data);
 477                 if (data->event_data == NULL) {
 478                         if (g_getenv ("MONO_DEBUG"))
 479                                 g_message ("Falling back to poll()");
 480                         data->event_system = POLL_BACKEND;
 481                 }
 482         }
 483 #elif defined(USE_KQUEUE_FOR_THREADPOOL)
 484         if (data->event_system == KQUEUE_BACKEND)
 485                 data->event_data = tp_kqueue_init (data);
 486 #endif
 487         if (data->event_system == POLL_BACKEND)
 488                 data->event_data = tp_poll_init (data);
 489 }
 490
 491 static void
 492 socket_io_init (SocketIOData *data)
 493 {
 494         int inited;
 495
 496         if (data->inited >= 2) // 2 -> initialized, 3-> cleaned up
 497                 return;
 498
 499         inited = InterlockedCompareExchange (&data->inited, 1, 0);
 500         if (inited >= 1) {
 501                 while (TRUE) {
 502                         if (data->inited >= 2)
 503                                 return;
 504                         SleepEx (1, FALSE);
 505                 }
 506         }
 507
 508         EnterCriticalSection (&data->io_lock);
 509         data->sock_to_state = mono_g_hash_table_new_type (g_direct_hash, g_direct_equal, MONO_HASH_VALUE_GC);
 510 #ifdef HAVE_EPOLL
 511         data->event_system = EPOLL_BACKEND;
 512 #elif defined(USE_KQUEUE_FOR_THREADPOOL)
 513         data->event_system = KQUEUE_BACKEND;
 514 #else
 515         data->event_system = POLL_BACKEND;
 516 #endif
 517         if (g_getenv ("MONO_DISABLE_AIO") != NULL)
 518                 data->event_system = POLL_BACKEND;
 519
 520         init_event_system (data);
 521         mono_thread_create_internal (mono_get_root_domain (), data->wait, data, TRUE, FALSE, SMALL_STACK);
 522         LeaveCriticalSection (&data->io_lock);
 523         data->inited = 2;
 524         threadpool_start_thread (&async_io_tp);
 525 }
 526
 527 static void
 528 socket_io_add (MonoAsyncResult *ares, MonoSocketAsyncResult *state)
 529 {
 530         MonoMList *list;
 531         SocketIOData *data = &socket_io_data;
 532         int fd;
 533         gboolean is_new;
 534         int ievt;
 535
 536         socket_io_init (&socket_io_data);
 537         if (mono_runtime_is_shutting_down () || data->inited == 3 || data->sock_to_state == NULL)
 538                 return;
 539         if (async_tp.pool_status == 2)
 540                 return;
 541
 542         MONO_OBJECT_SETREF (state, ares, ares);
 543
 544         fd = GPOINTER_TO_INT (state->handle);
 545         EnterCriticalSection (&data->io_lock);
 546         if (data->sock_to_state == NULL) {
 547                 LeaveCriticalSection (&data->io_lock);
 548                 return;
 549         }
 550         list = mono_g_hash_table_lookup (data->sock_to_state, GINT_TO_POINTER (fd));
 551         if (list == NULL) {
 552                 list = mono_mlist_alloc ((MonoObject*)state);
 553                 is_new = TRUE;
 554         } else {
 555                 list = mono_mlist_append (list, (MonoObject*)state);
 556                 is_new = FALSE;
 557         }
 558
 559         mono_g_hash_table_replace (data->sock_to_state, state->handle, list);
 560         ievt = get_events_from_list (list);
 561         /* The modify function leaves the io_lock critical section. */
 562         data->modify (data, fd, state->operation, ievt, is_new);
 563 }
 564
 565 #ifndef DISABLE_SOCKETS
 566 static gboolean
 567 socket_io_filter (MonoObject *target, MonoObject *state)
 568 {
 569         gint op;
 570         MonoSocketAsyncResult *sock_res;
 571         MonoClass *klass;
 572         MonoDomain *domain;
 573
 574         if (target == NULL || state == NULL)
 575                 return FALSE;
 576
 577         domain = target->vtable->domain;
 578         klass = target->vtable->klass;
 579         if (socket_async_call_klass == NULL && is_socketasynccall (domain, klass))
 580                 socket_async_call_klass = klass;
 581
 582         if (process_async_call_klass == NULL && is_sdp_asyncreadhandler (domain, klass))
 583                 process_async_call_klass = klass;
 584
 585         if (klass != socket_async_call_klass && klass != process_async_call_klass)
 586                 return FALSE;
 587
 588         sock_res = (MonoSocketAsyncResult *) state;
 589         op = sock_res->operation;
 590         if (op < AIO_OP_FIRST || op >= AIO_OP_LAST)
 591                 return FALSE;
 592
 593         return TRUE;
 594 }
 595 #endif /* !DISABLE_SOCKETS */
 596
 597 /* Returns the exception thrown when invoking, if any */
 598 static MonoObject *
 599 mono_async_invoke (ThreadPool *tp, MonoAsyncResult *ares)
 600 {
 601         ASyncCall *ac = (ASyncCall *)ares->object_data;
 602         MonoObject *res, *exc = NULL;
 603         MonoArray *out_args = NULL;
 604         HANDLE wait_event = NULL;
 605         MonoInternalThread *thread = mono_thread_internal_current ();
 606
 607         if (ares->execution_context) {
 608                 /* use captured ExecutionContext (if available) */
 609                 MONO_OBJECT_SETREF (ares, original_context, mono_thread_get_execution_context ());
 610                 mono_thread_set_execution_context (ares->execution_context);
 611         } else {
 612                 ares->original_context = NULL;
 613         }
 614
 615         if (ac == NULL) {
 616                 /* Fast path from ThreadPool.*QueueUserWorkItem */
 617                 void *pa = ares->async_state;
 618                 /* The debugger needs this */
 619                 thread->async_invoke_method = ((MonoDelegate*)ares->async_delegate)->method;
 620                 res = mono_runtime_delegate_invoke (ares->async_delegate, &pa, &exc);
 621                 thread->async_invoke_method = NULL;
 622         } else {
 623                 MonoObject *cb_exc = NULL;
 624
 625                 ac->msg->exc = NULL;
 626                 res = mono_message_invoke (ares->async_delegate, ac->msg, &exc, &out_args);
 627                 MONO_OBJECT_SETREF (ac, res, res);
 628                 MONO_OBJECT_SETREF (ac, msg->exc, exc);
 629                 MONO_OBJECT_SETREF (ac, out_args, out_args);
 630
 631                 mono_monitor_enter ((MonoObject *) ares);
 632                 ares->completed = 1;
 633                 if (ares->handle != NULL)
 634                         wait_event = mono_wait_handle_get_handle ((MonoWaitHandle *) ares->handle);
 635                 mono_monitor_exit ((MonoObject *) ares);
 636                 /* notify listeners */
 637                 if (wait_event != NULL)
 638                         SetEvent (wait_event);
 639
 640                 /* call async callback if cb_method != null*/
 641                 if (ac != NULL && ac->cb_method) {
 642                         void *pa = &ares;
 643                         cb_exc = NULL;
 644                         thread->async_invoke_method = ac->cb_method;
 645                         mono_runtime_invoke (ac->cb_method, ac->cb_target, pa, &cb_exc);
 646                         thread->async_invoke_method = NULL;
 647                         exc = cb_exc;
 648                 } else {
 649                         exc = NULL;
 650                 }
 651         }
 652
 653         /* restore original thread execution context if flow isn't suppressed, i.e. non null */
 654         if (ares->original_context) {
 655                 mono_thread_set_execution_context (ares->original_context);
 656                 ares->original_context = NULL;
 657         }
 658         return exc;
 659 }
 660
 661 static void
 662 threadpool_start_idle_threads (ThreadPool *tp)
 663 {
 664         int n;
 665         guint32 stack_size;
 666
 667         stack_size = (!tp->is_io) ? 0 : SMALL_STACK;
 668         do {
 669                 while (1) {
 670                         n = tp->nthreads;
 671                         if (n >= tp->min_threads)
 672                                 return;
 673                         if (InterlockedCompareExchange (&tp->nthreads, n + 1, n) == n)
 674                                 break;
 675                 }
 676 #ifndef DISABLE_PERFCOUNTERS
 677                 mono_perfcounter_update_value (tp->pc_nthreads, TRUE, 1);
 678 #endif
 679                 mono_thread_create_internal (mono_get_root_domain (), tp->async_invoke, tp, TRUE, FALSE, stack_size);
 680                 SleepEx (100, TRUE);
 681         } while (1);
 682 }
 683
 684 static void
 685 threadpool_init (ThreadPool *tp, int min_threads, int max_threads, void (*async_invoke) (gpointer))
 686 {
 687         memset (tp, 0, sizeof (ThreadPool));
 688         tp->min_threads = min_threads;
 689         tp->max_threads = max_threads;
 690         tp->async_invoke = async_invoke;
 691         tp->queue = mono_cq_create ();
 692         MONO_SEM_INIT (&tp->new_job, 0);
 693 }
 694
 695 #ifndef DISABLE_PERFCOUNTERS
 696 static void *
 697 init_perf_counter (const char *category, const char *counter)
 698 {
 699         MonoString *category_str;
 700         MonoString *counter_str;
 701         MonoString *machine;
 702         MonoDomain *root;
 703         MonoBoolean custom;
 704         int type;
 705
 706         if (category == NULL || counter == NULL)
 707                 return NULL;
 708         root = mono_get_root_domain ();
 709         category_str = mono_string_new (root, category);
 710         counter_str = mono_string_new (root, counter);
 711         machine = mono_string_new (root, ".");
 712         return mono_perfcounter_get_impl (category_str, counter_str, NULL, machine, &type, &custom);
 713 }
 714 #endif
 715
 716 #ifdef DEBUG
 717 static void
 718 print_pool_info (ThreadPool *tp)
 719 {
 720
 721 //      if (tp->tail - tp->head == 0)
 722 //              return;
 723
 724         g_print ("Pool status? %d\n", InterlockedCompareExchange (&tp->pool_status, 0, 0));
 725         g_print ("Min. threads: %d\n", InterlockedCompareExchange (&tp->min_threads, 0, 0));
 726         g_print ("Max. threads: %d\n", InterlockedCompareExchange (&tp->max_threads, 0, 0));
 727         g_print ("nthreads: %d\n", InterlockedCompareExchange (&tp->nthreads, 0, 0));
 728         g_print ("busy threads: %d\n", InterlockedCompareExchange (&tp->busy_threads, 0, 0));
 729         g_print ("Waiting: %d\n", InterlockedCompareExchange (&tp->waiting, 0, 0));
 730         g_print ("Queued: %d\n", (tp->tail - tp->head));
 731         if (tp == &async_tp) {
 732                 int i;
 733                 EnterCriticalSection (&wsqs_lock);
 734                 for (i = 0; i < wsqs->len; i++) {
 735                         g_print ("\tWSQ %d: %d\n", i, mono_wsq_count (g_ptr_array_index (wsqs, i)));
 736                 }
 737                 LeaveCriticalSection (&wsqs_lock);
 738         } else {
 739                 g_print ("\tSockets: %d\n", mono_g_hash_table_size (socket_io_data.sock_to_state));
 740         }
 741         g_print ("-------------\n");
 742 }
 743
 744 static void
 745 signal_handler (int signo)
 746 {
 747         ThreadPool *tp;
 748
 749         tp = &async_tp;
 750         g_print ("\n-----Non-IO-----\n");
 751         print_pool_info (tp);
 752         tp = &async_io_tp;
 753         g_print ("\n-----IO-----\n");
 754         print_pool_info (tp);
 755         alarm (2);
 756 }
 757 #endif
 758
 759 static void
 760 monitor_thread (gpointer unused)
 761 {
 762         ThreadPool *pools [2];
 763         MonoInternalThread *thread;
 764         guint32 ms;
 765         gboolean need_one;
 766         int i;
 767
 768         pools [0] = &async_tp;
 769         pools [1] = &async_io_tp;
 770         thread = mono_thread_internal_current ();
 771         ves_icall_System_Threading_Thread_SetName_internal (thread, mono_string_new (mono_domain_get (), "Threadpool monitor"));
 772         while (1) {
 773                 ms = 500;
 774                 i = 10; //number of spurious awakes we tolerate before doing a round of rebalancing.
 775                 do {
 776                         guint32 ts;
 777                         ts = mono_msec_ticks ();
 778                         if (SleepEx (ms, TRUE) == 0)
 779                                 break;
 780                         ms -= (mono_msec_ticks () - ts);
 781                         if (mono_runtime_is_shutting_down ())
 782                                 break;
 783                         if (THREAD_WANTS_A_BREAK (thread))
 784                                 mono_thread_interruption_checkpoint ();
 785                 } while (ms > 0 && i--);
 786
 787                 if (mono_runtime_is_shutting_down ())
 788                         break;
 789
 790                 if (suspended)
 791                         continue;
 792
 793                 for (i = 0; i < 2; i++) {
 794                         ThreadPool *tp;
 795                         tp = pools [i];
 796                         if (tp->waiting > 0)
 797                                 continue;
 798                         need_one = (mono_cq_count (tp->queue) > 0);
 799                         if (!need_one && !tp->is_io) {
 800                                 EnterCriticalSection (&wsqs_lock);
 801                                 for (i = 0; wsqs != NULL && i < wsqs->len; i++) {
 802                                         MonoWSQ *wsq;
 803                                         wsq = g_ptr_array_index (wsqs, i);
 804                                         if (mono_wsq_count (wsq) != 0) {
 805                                                 need_one = TRUE;
 806                                                 break;
 807                                         }
 808                                 }
 809                                 LeaveCriticalSection (&wsqs_lock);
 810                         }
 811                         if (need_one)
 812                                 threadpool_start_thread (tp);
 813                 }
 814         }
 815 }
 816
 817 void
 818 mono_thread_pool_init_tls (void)
 819 {
 820         mono_wsq_init ();
 821 }
 822
 823 void
 824 mono_thread_pool_init (void)
 825 {
 826         gint threads_per_cpu = 1;
 827         gint thread_count;
 828         gint cpu_count = mono_cpu_count ();
 829         int result;
 830
 831         if (tp_inited == 2)
 832                 return;
 833
 834         result = InterlockedCompareExchange (&tp_inited, 1, 0);
 835         if (result == 1) {
 836                 while (1) {
 837                         SleepEx (1, FALSE);
 838                         if (tp_inited == 2)
 839                                 return;
 840                 }
 841         }
 842
 843         MONO_GC_REGISTER_ROOT_FIXED (socket_io_data.sock_to_state);
 844         InitializeCriticalSection (&socket_io_data.io_lock);
 845         if (g_getenv ("MONO_THREADS_PER_CPU") != NULL) {
 846                 threads_per_cpu = atoi (g_getenv ("MONO_THREADS_PER_CPU"));
 847                 if (threads_per_cpu < 1)
 848                         threads_per_cpu = 1;
 849         }
 850
 851         thread_count = MIN (cpu_count * threads_per_cpu, 100 * cpu_count);
 852         threadpool_init (&async_tp, thread_count, MAX (100 * cpu_count, thread_count), async_invoke_thread);
 853         threadpool_init (&async_io_tp, cpu_count * 2, cpu_count * 4, async_invoke_thread);
 854         async_io_tp.is_io = TRUE;
 855
 856         async_call_klass = mono_class_from_name (mono_defaults.corlib, "System", "MonoAsyncCall");
 857         g_assert (async_call_klass);
 858
 859         InitializeCriticalSection (&wsqs_lock);
 860         wsqs = g_ptr_array_sized_new (MAX (100 * cpu_count, thread_count));
 861
 862 #ifndef DISABLE_PERFCOUNTERS
 863         async_tp.pc_nitems = init_perf_counter ("Mono Threadpool", "Work Items Added");
 864         g_assert (async_tp.pc_nitems);
 865
 866         async_io_tp.pc_nitems = init_perf_counter ("Mono Threadpool", "IO Work Items Added");
 867         g_assert (async_io_tp.pc_nitems);
 868
 869         async_tp.pc_nthreads = init_perf_counter ("Mono Threadpool", "# of Threads");
 870         g_assert (async_tp.pc_nthreads);
 871
 872         async_io_tp.pc_nthreads = init_perf_counter ("Mono Threadpool", "# of IO Threads");
 873         g_assert (async_io_tp.pc_nthreads);
 874 #endif
 875         tp_inited = 2;
 876 #ifdef DEBUG
 877         signal (SIGALRM, signal_handler);
 878         alarm (2);
 879 #endif
 880 }
 881
 882 static MonoAsyncResult *
 883 create_simple_asyncresult (MonoObject *target, MonoObject *state)
 884 {
 885         MonoDomain *domain = mono_domain_get ();
 886         MonoAsyncResult *ares;
 887
 888         /* Don't call mono_async_result_new() to avoid capturing the context */
 889         ares = (MonoAsyncResult *) mono_object_new (domain, mono_defaults.asyncresult_class);
 890         MONO_OBJECT_SETREF (ares, async_delegate, target);
 891         MONO_OBJECT_SETREF (ares, async_state, state);
 892         return ares;
 893 }
 894
 895 void
 896 icall_append_io_job (MonoObject *target, MonoSocketAsyncResult *state)
 897 {
 898         MonoAsyncResult *ares;
 899
 900         ares = create_simple_asyncresult (target, (MonoObject *) state);
 901         socket_io_add (ares, state);
 902 }
 903
 904 MonoAsyncResult *
 905 mono_thread_pool_add (MonoObject *target, MonoMethodMessage *msg, MonoDelegate *async_callback,
 906                       MonoObject *state)
 907 {
 908         MonoDomain *domain = mono_domain_get ();
 909         MonoAsyncResult *ares;
 910         ASyncCall *ac;
 911
 912         ac = (ASyncCall*)mono_object_new (domain, async_call_klass);
 913         MONO_OBJECT_SETREF (ac, msg, msg);
 914         MONO_OBJECT_SETREF (ac, state, state);
 915
 916         if (async_callback) {
 917                 ac->cb_method = mono_get_delegate_invoke (((MonoObject *)async_callback)->vtable->klass);
 918                 MONO_OBJECT_SETREF (ac, cb_target, async_callback);
 919         }
 920
 921         ares = mono_async_result_new (domain, NULL, ac->state, NULL, (MonoObject*)ac);
 922         MONO_OBJECT_SETREF (ares, async_delegate, target);
 923
 924 #ifndef DISABLE_SOCKETS
 925         if (socket_io_filter (target, state)) {
 926                 socket_io_add (ares, (MonoSocketAsyncResult *) state);
 927                 return ares;
 928         }
 929 #endif
 930         threadpool_append_job (&async_tp, (MonoObject *) ares);
 931         return ares;
 932 }
 933
 934 MonoObject *
 935 mono_thread_pool_finish (MonoAsyncResult *ares, MonoArray **out_args, MonoObject **exc)
 936 {
 937         ASyncCall *ac;
 938         HANDLE wait_event;
 939
 940         *exc = NULL;
 941         *out_args = NULL;
 942
 943         /* check if already finished */
 944         mono_monitor_enter ((MonoObject *) ares);
 945
 946         if (ares->endinvoke_called) {
 947                 *exc = (MonoObject *) mono_get_exception_invalid_operation (NULL);
 948                 mono_monitor_exit ((MonoObject *) ares);
 949                 return NULL;
 950         }
 951
 952         ares->endinvoke_called = 1;
 953         /* wait until we are really finished */
 954         if (!ares->completed) {
 955                 if (ares->handle == NULL) {
 956                         wait_event = CreateEvent (NULL, TRUE, FALSE, NULL);
 957                         g_assert(wait_event != 0);
 958                         MONO_OBJECT_SETREF (ares, handle, (MonoObject *) mono_wait_handle_new (mono_object_domain (ares), wait_event));
 959                 } else {
 960                         wait_event = mono_wait_handle_get_handle ((MonoWaitHandle *) ares->handle);
 961                 }
 962                 mono_monitor_exit ((MonoObject *) ares);
 963                 WaitForSingleObjectEx (wait_event, INFINITE, TRUE);
 964         } else {
 965                 mono_monitor_exit ((MonoObject *) ares);
 966         }
 967
 968         ac = (ASyncCall *) ares->object_data;
 969         g_assert (ac != NULL);
 970         *exc = ac->msg->exc; /* FIXME: GC add write barrier */
 971         *out_args = ac->out_args;
 972
 973         return ac->res;
 974 }
 975
 976 static void
 977 threadpool_kill_idle_threads (ThreadPool *tp)
 978 {
 979         gint n;
 980
 981         n = (gint) InterlockedCompareExchange (&tp->max_threads, 0, -1);
 982         while (n) {
 983                 n--;
 984                 MONO_SEM_POST (&tp->new_job);
 985         }
 986 }
 987
 988 void
 989 mono_thread_pool_cleanup (void)
 990 {
 991         if (InterlockedExchange (&async_io_tp.pool_status, 2) == 1) {
 992                 socket_io_cleanup (&socket_io_data); /* Empty when DISABLE_SOCKETS is defined */
 993                 threadpool_kill_idle_threads (&async_io_tp);
 994         }
 995
 996         if (async_io_tp.queue != NULL) {
 997                 MONO_SEM_DESTROY (&async_io_tp.new_job);
 998                 threadpool_free_queue (&async_io_tp);
 999         }
1000
1001
1002         if (InterlockedExchange (&async_tp.pool_status, 2) == 1) {
1003                 threadpool_kill_idle_threads (&async_tp);
1004                 threadpool_free_queue (&async_tp);
1005         }
1006
1007         if (wsqs) {
1008                 EnterCriticalSection (&wsqs_lock);
1009                 mono_wsq_cleanup ();
1010                 if (wsqs)
1011                         g_ptr_array_free (wsqs, TRUE);
1012                 wsqs = NULL;
1013                 LeaveCriticalSection (&wsqs_lock);
1014                 MONO_SEM_DESTROY (&async_tp.new_job);
1015         }
1016 }
1017
1018 static gboolean
1019 threadpool_start_thread (ThreadPool *tp)
1020 {
1021         gint n;
1022         guint32 stack_size;
1023
1024         stack_size = (!tp->is_io) ? 0 : SMALL_STACK;
1025         while (!mono_runtime_is_shutting_down () && (n = tp->nthreads) < tp->max_threads) {
1026                 if (InterlockedCompareExchange (&tp->nthreads, n + 1, n) == n) {
1027 #ifndef DISABLE_PERFCOUNTERS
1028                         mono_perfcounter_update_value (tp->pc_nthreads, TRUE, 1);
1029 #endif
1030                         mono_thread_create_internal (mono_get_root_domain (), tp->async_invoke, tp, TRUE, FALSE, stack_size);
1031                         return TRUE;
1032                 }
1033         }
1034
1035         return FALSE;
1036 }
1037
1038 static void
1039 pulse_on_new_job (ThreadPool *tp)
1040 {
1041         if (tp->waiting)
1042                 MONO_SEM_POST (&tp->new_job);
1043 }
1044
1045 void
1046 icall_append_job (MonoObject *ar)
1047 {
1048         threadpool_append_jobs (&async_tp, &ar, 1);
1049 }
1050
1051 static void
1052 threadpool_append_job (ThreadPool *tp, MonoObject *ar)
1053 {
1054         threadpool_append_jobs (tp, &ar, 1);
1055 }
1056
1057 static void
1058 threadpool_append_jobs (ThreadPool *tp, MonoObject **jobs, gint njobs)
1059 {
1060         static int job_counter;
1061         MonoObject *ar;
1062         gint i;
1063
1064         if (mono_runtime_is_shutting_down ())
1065                 return;
1066
1067         if (tp->pool_status == 0 && InterlockedCompareExchange (&tp->pool_status, 1, 0) == 0) {
1068                 if (!tp->is_io) {
1069                         mono_thread_create_internal (mono_get_root_domain (), monitor_thread, NULL, TRUE, FALSE, SMALL_STACK);
1070                         threadpool_start_thread (tp);
1071                 }
1072                 /* Create on demand up to min_threads to avoid startup penalty for apps that don't use
1073                  * the threadpool that much
1074                 * mono_thread_create_internal (mono_get_root_domain (), threadpool_start_idle_threads, tp, TRUE, FALSE, SMALL_STACK);
1075                 */
1076         }
1077
1078         for (i = 0; i < njobs; i++) {
1079                 ar = jobs [i];
1080                 if (ar == NULL || mono_domain_is_unloading (ar->vtable->domain))
1081                         continue; /* Might happen when cleaning domain jobs */
1082                 if (!tp->is_io && (InterlockedIncrement (&job_counter) % 10) == 0) {
1083                         MonoAsyncResult *o = (MonoAsyncResult *) ar;
1084                         o->add_time = mono_100ns_ticks ();
1085                 }
1086                 threadpool_jobs_inc (ar);
1087 #ifndef DISABLE_PERFCOUNTERS
1088                 mono_perfcounter_update_value (tp->pc_nitems, TRUE, 1);
1089 #endif
1090                 if (!tp->is_io && mono_wsq_local_push (ar))
1091                         continue;
1092
1093                 mono_cq_enqueue (tp->queue, ar);
1094         }
1095
1096         for (i = 0; tp->waiting > 0 && i < MIN(njobs, tp->max_threads); i++)
1097                 pulse_on_new_job (tp);
1098 }
1099
1100 static void
1101 threadpool_clear_queue (ThreadPool *tp, MonoDomain *domain)
1102 {
1103         MonoObject *obj;
1104         MonoMList *other = NULL;
1105         MonoCQ *queue = tp->queue;
1106
1107         if (!queue)
1108                 return;
1109
1110         while (mono_cq_dequeue (queue, &obj)) {
1111                 if (obj == NULL)
1112                         continue;
1113                 if (obj->vtable->domain != domain)
1114                         other = mono_mlist_prepend (other, obj);
1115                 threadpool_jobs_dec (obj);
1116         }
1117
1118         if (mono_runtime_is_shutting_down ())
1119                 return;
1120
1121         while (other) {
1122                 threadpool_append_job (tp, (MonoObject *) mono_mlist_get_data (other));
1123                 other = mono_mlist_next (other);
1124         }
1125 }
1126
1127 static gboolean
1128 remove_sockstate_for_domain (gpointer key, gpointer value, gpointer user_data)
1129 {
1130         MonoMList *list = value;
1131         gboolean remove = FALSE;
1132         while (list) {
1133                 MonoObject *data = mono_mlist_get_data (list);
1134                 if (mono_object_domain (data) == user_data) {
1135                         remove = TRUE;
1136                         mono_mlist_set_data (list, NULL);
1137                 }
1138                 list = mono_mlist_next (list);
1139         }
1140         //FIXME is there some sort of additional unregistration we need to perform here?
1141         return remove;
1142 }
1143
1144 /*
1145  * Clean up the threadpool of all domain jobs.
1146  * Can only be called as part of the domain unloading process as
1147  * it will wait for all jobs to be visible to the interruption code.
1148  */
1149 gboolean
1150 mono_thread_pool_remove_domain_jobs (MonoDomain *domain, int timeout)
1151 {
1152         HANDLE sem_handle;
1153         int result = TRUE;
1154         guint32 start_time = 0;
1155
1156         g_assert (domain->state == MONO_APPDOMAIN_UNLOADING);
1157
1158         threadpool_clear_queue (&async_tp, domain);
1159         threadpool_clear_queue (&async_io_tp, domain);
1160
1161         EnterCriticalSection (&socket_io_data.io_lock);
1162         if (socket_io_data.sock_to_state)
1163                 mono_g_hash_table_foreach_remove (socket_io_data.sock_to_state, remove_sockstate_for_domain, domain);
1164
1165         LeaveCriticalSection (&socket_io_data.io_lock);
1166
1167         /*
1168          * There might be some threads out that could be about to execute stuff from the given domain.
1169          * We avoid that by setting up a semaphore to be pulsed by the thread that reaches zero.
1170          */
1171         sem_handle = CreateSemaphore (NULL, 0, 1, NULL);
1172
1173         domain->cleanup_semaphore = sem_handle;
1174         /*
1175          * The memory barrier here is required to have global ordering between assigning to cleanup_semaphone
1176          * and reading threadpool_jobs.
1177          * Otherwise this thread could read a stale version of threadpool_jobs and wait forever.
1178          */
1179         mono_memory_write_barrier ();
1180
1181         if (domain->threadpool_jobs && timeout != -1)
1182                 start_time = mono_msec_ticks ();
1183         while (domain->threadpool_jobs) {
1184                 WaitForSingleObject (sem_handle, timeout);
1185                 if (timeout != -1 && (mono_msec_ticks () - start_time) > timeout) {
1186                         result = FALSE;
1187                         break;
1188                 }
1189         }
1190
1191         domain->cleanup_semaphore = NULL;
1192         CloseHandle (sem_handle);
1193         return result;
1194 }
1195
1196 static void
1197 threadpool_free_queue (ThreadPool *tp)
1198 {
1199         mono_cq_destroy (tp->queue);
1200         tp->queue = NULL;
1201 }
1202
1203 gboolean
1204 mono_thread_pool_is_queue_array (MonoArray *o)
1205 {
1206         // gpointer obj = o;
1207
1208         // FIXME: need some fix in sgen code.
1209         return FALSE;
1210 }
1211
1212 static MonoWSQ *
1213 add_wsq (void)
1214 {
1215         int i;
1216         MonoWSQ *wsq;
1217
1218         EnterCriticalSection (&wsqs_lock);
1219         wsq = mono_wsq_create ();
1220         if (wsqs == NULL) {
1221                 LeaveCriticalSection (&wsqs_lock);
1222                 return NULL;
1223         }
1224         for (i = 0; i < wsqs->len; i++) {
1225                 if (g_ptr_array_index (wsqs, i) == NULL) {
1226                         wsqs->pdata [i] = wsq;
1227                         LeaveCriticalSection (&wsqs_lock);
1228                         return wsq;
1229                 }
1230         }
1231         g_ptr_array_add (wsqs, wsq);
1232         LeaveCriticalSection (&wsqs_lock);
1233         return wsq;
1234 }
1235
1236 static void
1237 remove_wsq (MonoWSQ *wsq)
1238 {
1239         gpointer data;
1240
1241         if (wsq == NULL)
1242                 return;
1243
1244         EnterCriticalSection (&wsqs_lock);
1245         if (wsqs == NULL) {
1246                 LeaveCriticalSection (&wsqs_lock);
1247                 return;
1248         }
1249         g_ptr_array_remove_fast (wsqs, wsq);
1250         data = NULL;
1251         /*
1252          * Only clean this up when shutting down, any other case will error out
1253          * if we're removing a queue that still has work items.
1254          */
1255         if (mono_runtime_is_shutting_down ()) {
1256                 while (mono_wsq_local_pop (&data)) {
1257                         threadpool_jobs_dec (data);
1258                         data = NULL;
1259                 }
1260         }
1261         mono_wsq_destroy (wsq);
1262         LeaveCriticalSection (&wsqs_lock);
1263 }
1264
1265 static void
1266 try_steal (MonoWSQ *local_wsq, gpointer *data, gboolean retry)
1267 {
1268         int i;
1269         int ms;
1270
1271         if (wsqs == NULL || data == NULL || *data != NULL)
1272                 return;
1273
1274         ms = 0;
1275         do {
1276                 if (mono_runtime_is_shutting_down ())
1277                         return;
1278
1279                 EnterCriticalSection (&wsqs_lock);
1280                 for (i = 0; wsqs != NULL && i < wsqs->len; i++) {
1281                         MonoWSQ *wsq;
1282
1283                         wsq = wsqs->pdata [i];
1284                         if (wsq == local_wsq || mono_wsq_count (wsq) == 0)
1285                                 continue;
1286                         mono_wsq_try_steal (wsqs->pdata [i], data, ms);
1287                         if (*data != NULL) {
1288                                 LeaveCriticalSection (&wsqs_lock);
1289                                 return;
1290                         }
1291                 }
1292                 LeaveCriticalSection (&wsqs_lock);
1293                 ms += 10;
1294         } while (retry && ms < 11);
1295 }
1296
1297 static gboolean
1298 dequeue_or_steal (ThreadPool *tp, gpointer *data, MonoWSQ *local_wsq)
1299 {
1300         if (mono_runtime_is_shutting_down ())
1301                 return FALSE;
1302         mono_cq_dequeue (tp->queue, (MonoObject **) data);
1303         if (!tp->is_io && !*data)
1304                 try_steal (local_wsq, data, FALSE);
1305         return (*data != NULL);
1306 }
1307
1308 static void
1309 process_idle_times (ThreadPool *tp, gint64 t)
1310 {
1311         gint64 ticks;
1312         gint64 avg;
1313         gboolean compute_avg;
1314         gint new_threads;
1315         gint64 per1;
1316
1317         if (tp->ignore_times || t <= 0)
1318                 return;
1319
1320         compute_avg = FALSE;
1321         ticks = mono_100ns_ticks ();
1322         t = ticks - t;
1323         SPIN_LOCK (tp->sp_lock);
1324         if (tp->ignore_times) {
1325                 SPIN_UNLOCK (tp->sp_lock);
1326                 return;
1327         }
1328         tp->time_sum += t;
1329         tp->n_sum++;
1330         if (tp->last_check == 0)
1331                 tp->last_check = ticks;
1332         else if (tp->last_check > 0 && (ticks - tp->last_check) > 5000000) {
1333                 tp->ignore_times = 1;
1334                 compute_avg = TRUE;
1335         }
1336         SPIN_UNLOCK (tp->sp_lock);
1337
1338         if (!compute_avg)
1339                 return;
1340
1341         //printf ("Items: %d Time elapsed: %.3fs\n", tp->n_sum, (ticks - tp->last_check) / 10000.0);
1342         tp->last_check = ticks;
1343         new_threads = 0;
1344         avg = tp->time_sum / tp->n_sum;
1345         if (tp->averages [1] == 0) {
1346                 tp->averages [1] = avg;
1347         } else {
1348                 per1 = ((100 * (ABS (avg - tp->averages [1]))) / tp->averages [1]);
1349                 if (per1 > 5) {
1350                         if (avg > tp->averages [1]) {
1351                                 if (tp->averages [1] < tp->averages [0]) {
1352                                         new_threads = -1;
1353                                 } else {
1354                                         new_threads = 1;
1355                                 }
1356                         } else if (avg < tp->averages [1] && tp->averages [1] < tp->averages [0]) {
1357                                 new_threads = 1;
1358                         }
1359                 } else {
1360                         int min, n;
1361                         min = tp->min_threads;
1362                         n = tp->nthreads;
1363                         if ((n - min) < min && tp->busy_threads == n)
1364                                 new_threads = 1;
1365                 }
1366                 /*
1367                 if (new_threads != 0) {
1368                         printf ("n: %d per1: %lld avg=%lld avg1=%lld avg0=%lld\n", new_threads, per1, avg, tp->averages [1], tp->averages [0]);
1369                 }
1370                 */
1371         }
1372
1373         tp->time_sum = 0;
1374         tp->n_sum = 0;
1375
1376         tp->averages [0] = tp->averages [1];
1377         tp->averages [1] = avg;
1378         tp->ignore_times = 0;
1379
1380         if (new_threads == -1) {
1381                 if (tp->destroy_thread == 0 && InterlockedCompareExchange (&tp->destroy_thread, 1, 0) == 0)
1382                         pulse_on_new_job (tp);
1383         }
1384 }
1385
1386 static gboolean
1387 should_i_die (ThreadPool *tp)
1388 {
1389         gboolean result = FALSE;
1390         if (tp->destroy_thread == 1 && InterlockedCompareExchange (&tp->destroy_thread, 0, 1) == 1)
1391                 result = (tp->nthreads > tp->min_threads);
1392         return result;
1393 }
1394
1395 static void
1396 set_tp_thread_info (ThreadPool *tp)
1397 {
1398         const gchar *name;
1399         MonoInternalThread *thread = mono_thread_internal_current ();
1400
1401         mono_profiler_thread_start (thread->tid);
1402         name = (tp->is_io) ? "IO Threadpool worker" : "Threadpool worker";
1403         mono_thread_set_name_internal (thread, mono_string_new (mono_domain_get (), name), FALSE);
1404 }
1405
1406 static void
1407 clear_thread_state (void)
1408 {
1409         MonoInternalThread *thread = mono_thread_internal_current ();
1410         /* If the callee changes the background status, set it back to TRUE */
1411         mono_thread_clr_state (thread , ~ThreadState_Background);
1412         if (!mono_thread_test_state (thread , ThreadState_Background))
1413                 ves_icall_System_Threading_Thread_SetState (thread, ThreadState_Background);
1414 }
1415
1416 static void
1417 check_for_interruption_critical (void)
1418 {
1419         MonoInternalThread *thread;
1420         /*RULE NUMBER ONE OF SKIP_THREAD: NEVER POKE MANAGED STATE.*/
1421         mono_gc_set_skip_thread (FALSE);
1422
1423         thread = mono_thread_internal_current ();
1424         if (THREAD_WANTS_A_BREAK (thread))
1425                 mono_thread_interruption_checkpoint ();
1426
1427         /*RULE NUMBER TWO OF SKIP_THREAD: READ RULE NUMBER ONE.*/
1428         mono_gc_set_skip_thread (TRUE);
1429 }
1430
1431 static void
1432 fire_profiler_thread_end (void)
1433 {
1434         MonoInternalThread *thread = mono_thread_internal_current ();
1435         mono_profiler_thread_end (thread->tid);
1436 }
1437
1438 static void
1439 async_invoke_thread (gpointer data)
1440 {
1441         MonoDomain *domain;
1442         MonoWSQ *wsq;
1443         ThreadPool *tp;
1444         gboolean must_die;
1445
1446         tp = data;
1447         wsq = NULL;
1448         if (!tp->is_io)
1449                 wsq = add_wsq ();
1450
1451         set_tp_thread_info (tp);
1452
1453         if (tp_start_func)
1454                 tp_start_func (tp_hooks_user_data);
1455
1456         data = NULL;
1457         for (;;) {
1458                 MonoAsyncResult *ar;
1459                 MonoClass *klass;
1460                 gboolean is_io_task;
1461                 gboolean is_socket;
1462                 int n_naps = 0;
1463
1464                 is_io_task = FALSE;
1465                 ar = (MonoAsyncResult *) data;
1466                 if (ar) {
1467                         InterlockedIncrement (&tp->busy_threads);
1468                         domain = ((MonoObject *)ar)->vtable->domain;
1469 #ifndef DISABLE_SOCKETS
1470                         klass = ((MonoObject *) data)->vtable->klass;
1471                         is_io_task = !is_corlib_asyncresult (domain, klass);
1472                         is_socket = FALSE;
1473                         if (is_io_task) {
1474                                 MonoSocketAsyncResult *state = (MonoSocketAsyncResult *) data;
1475                                 is_socket = is_socketasyncresult (domain, klass);
1476                                 ar = state->ares;
1477                                 switch (state->operation) {
1478                                 case AIO_OP_RECEIVE:
1479                                         state->total = ICALL_RECV (state);
1480                                         break;
1481                                 case AIO_OP_SEND:
1482                                         state->total = ICALL_SEND (state);
1483                                         break;
1484                                 }
1485                         }
1486 #endif
1487                         /* worker threads invokes methods in different domains,
1488                          * so we need to set the right domain here */
1489                         g_assert (domain);
1490
1491                         if (mono_domain_is_unloading (domain) || mono_runtime_is_shutting_down ()) {
1492                                 threadpool_jobs_dec ((MonoObject *)ar);
1493                                 data = NULL;
1494                                 ar = NULL;
1495                                 InterlockedDecrement (&tp->busy_threads);
1496                         } else {
1497                                 mono_thread_push_appdomain_ref (domain);
1498                                 if (threadpool_jobs_dec ((MonoObject *)ar)) {
1499                                         data = NULL;
1500                                         ar = NULL;
1501                                         mono_thread_pop_appdomain_ref ();
1502                                         InterlockedDecrement (&tp->busy_threads);
1503                                         continue;
1504                                 }
1505
1506                                 if (mono_domain_set (domain, FALSE)) {
1507                                         MonoObject *exc;
1508
1509                                         if (tp_item_begin_func)
1510                                                 tp_item_begin_func (tp_item_user_data);
1511
1512                                         if (!is_io_task && ar->add_time > 0)
1513                                                 process_idle_times (tp, ar->add_time);
1514                                         exc = mono_async_invoke (tp, ar);
1515                                         if (tp_item_end_func)
1516                                                 tp_item_end_func (tp_item_user_data);
1517                                         if (exc)
1518                                                 mono_internal_thread_unhandled_exception (exc);
1519                                         if (is_socket && tp->is_io) {
1520                                                 MonoSocketAsyncResult *state = (MonoSocketAsyncResult *) data;
1521
1522                                                 if (state->completed && state->callback) {
1523                                                         MonoAsyncResult *cb_ares;
1524                                                         cb_ares = create_simple_asyncresult ((MonoObject *) state->callback,
1525                                                                                                 (MonoObject *) state);
1526                                                         icall_append_job ((MonoObject *) cb_ares);
1527                                                 }
1528                                         }
1529                                         mono_domain_set (mono_get_root_domain (), TRUE);
1530                                 }
1531                                 mono_thread_pop_appdomain_ref ();
1532                                 InterlockedDecrement (&tp->busy_threads);
1533                                 clear_thread_state ();
1534                         }
1535                 }
1536
1537                 ar = NULL;
1538                 data = NULL;
1539                 must_die = should_i_die (tp);
1540                 if (!must_die && (tp->is_io || !mono_wsq_local_pop (&data)))
1541                         dequeue_or_steal (tp, &data, wsq);
1542
1543                 n_naps = 0;
1544                 while (!must_die && !data && n_naps < 4) {
1545                         gboolean res;
1546
1547                         InterlockedIncrement (&tp->waiting);
1548
1549                         // Another thread may have added a job into its wsq since the last call to dequeue_or_steal
1550                         // Check all the queues again before entering the wait loop
1551                         dequeue_or_steal (tp, &data, wsq);
1552                         if (data) {
1553                                 InterlockedDecrement (&tp->waiting);
1554                                 break;
1555                         }
1556
1557                         mono_gc_set_skip_thread (TRUE);
1558
1559 #if defined(__OpenBSD__)
1560                         while (mono_cq_count (tp->queue) == 0 && (res = mono_sem_wait (&tp->new_job, TRUE)) == -1) {// && errno == EINTR) {
1561 #else
1562                         while (mono_cq_count (tp->queue) == 0 && (res = mono_sem_timedwait (&tp->new_job, 2000, TRUE)) == -1) {// && errno == EINTR) {
1563 #endif
1564                                 if (mono_runtime_is_shutting_down ())
1565                                         break;
1566                                 check_for_interruption_critical ();
1567                         }
1568                         InterlockedDecrement (&tp->waiting);
1569
1570                         mono_gc_set_skip_thread (FALSE);
1571
1572                         if (mono_runtime_is_shutting_down ())
1573                                 break;
1574                         must_die = should_i_die (tp);
1575                         dequeue_or_steal (tp, &data, wsq);
1576                         n_naps++;
1577                 }
1578
1579                 if (!data && !tp->is_io && !mono_runtime_is_shutting_down ()) {
1580                         mono_wsq_local_pop (&data);
1581                         if (data && must_die) {
1582                                 InterlockedCompareExchange (&tp->destroy_thread, 1, 0);
1583                                 pulse_on_new_job (tp);
1584                         }
1585                 }
1586
1587                 if (!data) {
1588                         gint nt;
1589                         gboolean down;
1590                         while (1) {
1591                                 nt = tp->nthreads;
1592                                 down = mono_runtime_is_shutting_down ();
1593                                 if (!down && nt <= tp->min_threads)
1594                                         break;
1595                                 if (down || InterlockedCompareExchange (&tp->nthreads, nt - 1, nt) == nt) {
1596 #ifndef DISABLE_PERFCOUNTERS
1597                                         mono_perfcounter_update_value (tp->pc_nthreads, TRUE, -1);
1598 #endif
1599                                         if (!tp->is_io) {
1600                                                 remove_wsq (wsq);
1601                                         }
1602
1603                                         fire_profiler_thread_end ();
1604
1605                                         if (tp_finish_func)
1606                                                 tp_finish_func (tp_hooks_user_data);
1607                                         return;
1608                                 }
1609                         }
1610                 }
1611         }
1612
1613         g_assert_not_reached ();
1614 }
1615
1616 void
1617 ves_icall_System_Threading_ThreadPool_GetAvailableThreads (gint *workerThreads, gint *completionPortThreads)
1618 {
1619         *workerThreads = async_tp.max_threads - async_tp.busy_threads;
1620         *completionPortThreads = async_io_tp.max_threads - async_io_tp.busy_threads;
1621 }
1622
1623 void
1624 ves_icall_System_Threading_ThreadPool_GetMaxThreads (gint *workerThreads, gint *completionPortThreads)
1625 {
1626         *workerThreads = async_tp.max_threads;
1627         *completionPortThreads = async_io_tp.max_threads;
1628 }
1629
1630 void
1631 ves_icall_System_Threading_ThreadPool_GetMinThreads (gint *workerThreads, gint *completionPortThreads)
1632 {
1633         *workerThreads = async_tp.min_threads;
1634         *completionPortThreads = async_io_tp.min_threads;
1635 }
1636
1637 MonoBoolean
1638 ves_icall_System_Threading_ThreadPool_SetMinThreads (gint workerThreads, gint completionPortThreads)
1639 {
1640         gint max_threads;
1641         gint max_io_threads;
1642
1643         max_threads = async_tp.max_threads;
1644         if (workerThreads <= 0 || workerThreads > max_threads)
1645                 return FALSE;
1646
1647         max_io_threads = async_io_tp.max_threads;
1648         if (completionPortThreads <= 0 || completionPortThreads > max_io_threads)
1649                 return FALSE;
1650
1651         InterlockedExchange (&async_tp.min_threads, workerThreads);
1652         InterlockedExchange (&async_io_tp.min_threads, completionPortThreads);
1653         if (workerThreads > async_tp.nthreads)
1654                 mono_thread_create_internal (mono_get_root_domain (), threadpool_start_idle_threads, &async_tp, TRUE, FALSE, SMALL_STACK);
1655         if (completionPortThreads > async_io_tp.nthreads)
1656                 mono_thread_create_internal (mono_get_root_domain (), threadpool_start_idle_threads, &async_io_tp, TRUE, FALSE, SMALL_STACK);
1657         return TRUE;
1658 }
1659
1660 MonoBoolean
1661 ves_icall_System_Threading_ThreadPool_SetMaxThreads (gint workerThreads, gint completionPortThreads)
1662 {
1663         gint min_threads;
1664         gint min_io_threads;
1665         gint cpu_count;
1666
1667         cpu_count = mono_cpu_count ();
1668         min_threads = async_tp.min_threads;
1669         if (workerThreads < min_threads || workerThreads < cpu_count)
1670                 return FALSE;
1671
1672         /* We don't really have the concept of completion ports. Do we care here? */
1673         min_io_threads = async_io_tp.min_threads;
1674         if (completionPortThreads < min_io_threads || completionPortThreads < cpu_count)
1675                 return FALSE;
1676
1677         InterlockedExchange (&async_tp.max_threads, workerThreads);
1678         InterlockedExchange (&async_io_tp.max_threads, completionPortThreads);
1679         return TRUE;
1680 }
1681
1682 /**
1683  * mono_install_threadpool_thread_hooks
1684  * @start_func: the function to be called right after a new threadpool thread is created. Can be NULL.
1685  * @finish_func: the function to be called right before a thredpool thread is exiting. Can be NULL.
1686  * @user_data: argument passed to @start_func and @finish_func.
1687  *
1688  * @start_fun will be called right after a threadpool thread is created and @finish_func right before a threadpool thread exits.
1689  * The calls will be made from the thread itself.
1690  */
1691 void
1692 mono_install_threadpool_thread_hooks (MonoThreadPoolFunc start_func, MonoThreadPoolFunc finish_func, gpointer user_data)
1693 {
1694         tp_start_func = start_func;
1695         tp_finish_func = finish_func;
1696         tp_hooks_user_data = user_data;
1697 }
1698
1699 /**
1700  * mono_install_threadpool_item_hooks
1701  * @begin_func: the function to be called before a threadpool work item processing starts.
1702  * @end_func: the function to be called after a threadpool work item is finished.
1703  * @user_data: argument passed to @begin_func and @end_func.
1704  *
1705  * The calls will be made from the thread itself and from the same AppDomain
1706  * where the work item was executed.
1707  *
1708  */
1709 void
1710 mono_install_threadpool_item_hooks (MonoThreadPoolItemFunc begin_func, MonoThreadPoolItemFunc end_func, gpointer user_data)
1711 {
1712         tp_item_begin_func = begin_func;
1713         tp_item_end_func = end_func;
1714         tp_item_user_data = user_data;
1715 }
1716
1717 void
1718 mono_internal_thread_unhandled_exception (MonoObject* exc)
1719 {
1720         if (mono_runtime_unhandled_exception_policy_get () == MONO_UNHANDLED_POLICY_CURRENT) {
1721                 gboolean unloaded;
1722                 MonoClass *klass;
1723
1724                 klass = exc->vtable->klass;
1725                 unloaded = is_appdomainunloaded_exception (exc->vtable->domain, klass);
1726                 if (!unloaded && klass != mono_defaults.threadabortexception_class) {
1727                         mono_unhandled_exception (exc);
1728                         if (mono_environment_exitcode_get () == 1)
1729                                 exit (255);
1730                 }
1731                 if (klass == mono_defaults.threadabortexception_class)
1732                  mono_thread_internal_reset_abort (mono_thread_internal_current ());
1733         }
1734 }
1735
1736 /*
1737  * Suspend creation of new threads.
1738  */
1739 void
1740 mono_thread_pool_suspend (void)
1741 {
1742         suspended = TRUE;
1743 }
1744
1745 /*
1746  * Resume creation of new threads.
1747  */
1748 void
1749 mono_thread_pool_resume (void)
1750 {
1751         suspended = FALSE;
1752 }