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