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 #ifdef MONO_SMALL_CONFIG
  16 #define QUEUE_LENGTH 16 /* Must be 2^N */
  17 #else
  18 #define QUEUE_LENGTH 64 /* Must be 2^N */
  19 #endif
  20
  21 #include <mono/metadata/domain-internals.h>
  22 #include <mono/metadata/tabledefs.h>
  23 #include <mono/metadata/threads.h>
  24 #include <mono/metadata/threads-types.h>
  25 #include <mono/metadata/threadpool-internals.h>
  26 #include <mono/metadata/exception.h>
  27 #include <mono/metadata/file-io.h>
  28 #include <mono/metadata/monitor.h>
  29 #include <mono/metadata/mono-mlist.h>
  30 #include <mono/metadata/marshal.h>
  31 #include <mono/metadata/mono-perfcounters.h>
  32 #include <mono/metadata/socket-io.h>
  33 #include <mono/metadata/mono-wsq.h>
  34 #include <mono/io-layer/io-layer.h>
  35 #include <mono/metadata/gc-internal.h>
  36 #include <mono/utils/mono-time.h>
  37 #include <mono/utils/mono-proclib.h>
  38 #include <mono/utils/mono-semaphore.h>
  39 #include <errno.h>
  40 #ifdef HAVE_SYS_TIME_H
  41 #include <sys/time.h>
  42 #endif
  43 #include <sys/types.h>
  44 #include <fcntl.h>
  45 #ifdef HAVE_UNISTD_H
  46 #include <unistd.h>
  47 #endif
  48 #include <string.h>
  49 #ifdef HAVE_SYS_SOCKET_H
  50 #include <sys/socket.h>
  51 #endif
  52 #include <mono/utils/mono-poll.h>
  53 #ifdef HAVE_EPOLL
  54 #include <sys/epoll.h>
  55 #endif
  56
  57 #ifndef DISABLE_SOCKETS
  58 #include "mono/io-layer/socket-wrappers.h"
  59 #endif
  60
  61 #include "threadpool.h"
  62
  63 #define THREAD_WANTS_A_BREAK(t) ((t->state & (ThreadState_StopRequested | \
  64                                                 ThreadState_SuspendRequested)) != 0)
  65
  66 #define SPIN_TRYLOCK(i) (InterlockedCompareExchange (&(i), 1, 0) == 0)
  67 #define SPIN_LOCK(i) do { \
  68                                 if (SPIN_TRYLOCK (i)) \
  69                                         break; \
  70                         } while (1)
  71
  72 #define SPIN_UNLOCK(i) i = 0
  73
  74 #define EPOLL_DEBUG(...)
  75 #define EPOLL_DEBUG_STMT(...)
  76 #define TP_DEBUG(...)
  77 #define TP_DEBUG_STMT(...)
  78
  79 /* DEBUG: prints tp data every 2s */
  80 #undef DEBUG
  81
  82 /*
  83 #define EPOLL_DEBUG(...) g_message(__VA_ARGS__)
  84 #define EPOLL_DEBUG_STMT(...) do { __VA_ARGS__ } while (0)
  85 #define TP_DEBUG(...) g_message(__VA_ARGS__)
  86 #define TP_DEBUG_STMT(...) do { __VA_ARGS__ } while (0)
  87 */
  88
  89 /* map of CounterSample.cs */
  90 struct _MonoCounterSample {
  91         gint64 rawValue;
  92         gint64 baseValue;
  93         gint64 counterFrequency;
  94         gint64 systemFrequency;
  95         gint64 timeStamp;
  96         gint64 timeStamp100nSec;
  97         gint64 counterTimeStamp;
  98         int counterType;
  99 };
 100
 101 /* mono_thread_pool_init called */
 102 static volatile int tp_inited;
 103
 104 typedef struct {
 105         CRITICAL_SECTION io_lock; /* access to sock_to_state */
 106         int inited; // 0 -> not initialized , 1->initializing, 2->initialized, 3->cleaned up
 107         int pipe [2];
 108         MonoGHashTable *sock_to_state;
 109
 110         HANDLE new_sem; /* access to newpfd and write side of the pipe */
 111         mono_pollfd *newpfd;
 112         gboolean epoll_disabled;
 113 #ifdef HAVE_EPOLL
 114         int epollfd;
 115 #endif
 116 } SocketIOData;
 117
 118 static SocketIOData socket_io_data;
 119
 120 /* Keep in sync with the System.MonoAsyncCall class which provides GC tracking */
 121 typedef struct {
 122         MonoObject         object;
 123         MonoMethodMessage *msg;
 124         MonoMethod        *cb_method;
 125         MonoDelegate      *cb_target;
 126         MonoObject        *state;
 127         MonoObject        *res;
 128         MonoArray         *out_args;
 129 } ASyncCall;
 130
 131 typedef struct {
 132         MonoSemType lock;
 133         MonoMList *first; /* GC root */
 134         MonoMList *last;
 135         MonoMList *unused; /* Up to 20 chunks. GC root */
 136         gint head;
 137         gint tail;
 138         MonoSemType new_job;
 139         volatile gint waiting; /* threads waiting for a work item */
 140
 141         /**/
 142         volatile gint pool_status; /* 0 -> not initialized, 1 -> initialized, 2 -> cleaning up */
 143         /* min, max, n and busy -> Interlocked */
 144         volatile gint min_threads;
 145         volatile gint max_threads;
 146         volatile gint nthreads;
 147         volatile gint busy_threads;
 148
 149         void (*async_invoke) (gpointer data);
 150         void *pc_nitems; /* Performance counter for total number of items in added */
 151         void *pc_nthreads; /* Performance counter for total number of active threads */
 152         /**/
 153         volatile gint destroy_thread;
 154         volatile gint ignore_times; /* Used when there's a thread being created or destroyed */
 155         volatile gint sp_lock; /* spin lock used to protect ignore_times */
 156         volatile gint64 last_check;
 157         volatile gint64 time_sum;
 158         volatile gint n_sum;
 159         gint64 averages [2];
 160         /**/
 161         //TP_DEBUG_ONLY (gint nodes_created);
 162         //TP_DEBUG_ONLY (gint nodes_reused);
 163         gboolean is_io;
 164 } ThreadPool;
 165
 166 static ThreadPool async_tp;
 167 static ThreadPool async_io_tp;
 168
 169 static void async_invoke_thread (gpointer data);
 170 static MonoObject *mono_async_invoke (ThreadPool *tp, MonoAsyncResult *ares);
 171 static void threadpool_free_queue (ThreadPool *tp);
 172 static void threadpool_append_job (ThreadPool *tp, MonoObject *ar);
 173 static void threadpool_append_jobs (ThreadPool *tp, MonoObject **jobs, gint njobs);
 174 static void threadpool_init (ThreadPool *tp, int min_threads, int max_threads, void (*async_invoke) (gpointer));
 175 static void threadpool_start_idle_threads (ThreadPool *tp);
 176 static void threadpool_kill_idle_threads (ThreadPool *tp);
 177
 178 static MonoClass *async_call_klass;
 179 static MonoClass *socket_async_call_klass;
 180 static MonoClass *process_async_call_klass;
 181
 182 static GPtrArray *wsqs;
 183 CRITICAL_SECTION wsqs_lock;
 184
 185 /* Hooks */
 186 static MonoThreadPoolFunc tp_start_func;
 187 static MonoThreadPoolFunc tp_finish_func;
 188 static gpointer tp_hooks_user_data;
 189 static MonoThreadPoolItemFunc tp_item_begin_func;
 190 static MonoThreadPoolItemFunc tp_item_end_func;
 191 static gpointer tp_item_user_data;
 192
 193 #define INIT_POLLFD(a, b, c) {(a)->fd = b; (a)->events = c; (a)->revents = 0;}
 194 enum {
 195         AIO_OP_FIRST,
 196         AIO_OP_ACCEPT = 0,
 197         AIO_OP_CONNECT,
 198         AIO_OP_RECEIVE,
 199         AIO_OP_RECEIVEFROM,
 200         AIO_OP_SEND,
 201         AIO_OP_SENDTO,
 202         AIO_OP_RECV_JUST_CALLBACK,
 203         AIO_OP_SEND_JUST_CALLBACK,
 204         AIO_OP_READPIPE,
 205         AIO_OP_LAST
 206 };
 207
 208 #ifdef DISABLE_SOCKETS
 209 #define socket_io_cleanup(x)
 210 #else
 211 static void
 212 socket_io_cleanup (SocketIOData *data)
 213 {
 214         EnterCriticalSection (&data->io_lock);
 215         if (data->inited != 2) {
 216                 LeaveCriticalSection (&data->io_lock);
 217                 return;
 218         }
 219         data->inited = 3;
 220
 221 #ifdef HOST_WIN32
 222         closesocket (data->pipe [0]);
 223         closesocket (data->pipe [1]);
 224 #else
 225         if (data->pipe [0] > -1)
 226                 close (data->pipe [0]);
 227         if (data->pipe [1] > -1)
 228                 close (data->pipe [1]);
 229 #endif
 230         data->pipe [0] = -1;
 231         data->pipe [1] = -1;
 232         if (data->new_sem)
 233                 CloseHandle (data->new_sem);
 234         data->new_sem = NULL;
 235         mono_g_hash_table_destroy (data->sock_to_state);
 236         data->sock_to_state = NULL;
 237         g_free (data->newpfd);
 238         data->newpfd = NULL;
 239 #ifdef HAVE_EPOLL
 240         if (FALSE == data->epoll_disabled)
 241                 close (data->epollfd);
 242 #endif
 243         LeaveCriticalSection (&data->io_lock);
 244 }
 245
 246 static int
 247 get_event_from_state (MonoSocketAsyncResult *state)
 248 {
 249         switch (state->operation) {
 250         case AIO_OP_ACCEPT:
 251         case AIO_OP_RECEIVE:
 252         case AIO_OP_RECV_JUST_CALLBACK:
 253         case AIO_OP_RECEIVEFROM:
 254         case AIO_OP_READPIPE:
 255                 return MONO_POLLIN;
 256         case AIO_OP_SEND:
 257         case AIO_OP_SEND_JUST_CALLBACK:
 258         case AIO_OP_SENDTO:
 259         case AIO_OP_CONNECT:
 260                 return MONO_POLLOUT;
 261         default: /* Should never happen */
 262                 g_message ("get_event_from_state: unknown value in switch!!!");
 263                 return 0;
 264         }
 265 }
 266
 267 static int
 268 get_events_from_list (MonoMList *list)
 269 {
 270         MonoSocketAsyncResult *state;
 271         int events = 0;
 272
 273         while (list && (state = (MonoSocketAsyncResult *)mono_mlist_get_data (list))) {
 274                 events |= get_event_from_state (state);
 275                 list = mono_mlist_next (list);
 276         }
 277
 278         return events;
 279 }
 280
 281 #define ICALL_RECV(x)   ves_icall_System_Net_Sockets_Socket_Receive_internal (\
 282                                 (SOCKET)(gssize)x->handle, x->buffer, x->offset, x->size,\
 283                                  x->socket_flags, &x->error);
 284
 285 #define ICALL_SEND(x)   ves_icall_System_Net_Sockets_Socket_Send_internal (\
 286                                 (SOCKET)(gssize)x->handle, x->buffer, x->offset, x->size,\
 287                                  x->socket_flags, &x->error);
 288
 289 #endif /* !DISABLE_SOCKETS */
 290
 291 static void
 292 threadpool_jobs_inc (MonoObject *obj)
 293 {
 294         if (obj)
 295                 InterlockedIncrement (&obj->vtable->domain->threadpool_jobs);
 296 }
 297
 298 static gboolean
 299 threadpool_jobs_dec (MonoObject *obj)
 300 {
 301         MonoDomain *domain = obj->vtable->domain;
 302         int remaining_jobs = InterlockedDecrement (&domain->threadpool_jobs);
 303         if (remaining_jobs == 0 && domain->cleanup_semaphore) {
 304                 ReleaseSemaphore (domain->cleanup_semaphore, 1, NULL);
 305                 return TRUE;
 306         }
 307         return FALSE;
 308 }
 309
 310 #ifdef HAVE_EPOLL
 311 static MonoObject *
 312 get_io_event (MonoMList **list, gint event)
 313 {
 314         MonoObject *state;
 315         MonoMList *current;
 316         MonoMList *prev;
 317
 318         current = *list;
 319         prev = NULL;
 320         state = NULL;
 321         while (current) {
 322                 state = mono_mlist_get_data (current);
 323                 if (get_event_from_state ((MonoSocketAsyncResult *) state) == event)
 324                         break;
 325
 326                 state = NULL;
 327                 prev = current;
 328                 current = mono_mlist_next (current);
 329         }
 330
 331         if (current) {
 332                 if (prev) {
 333                         mono_mlist_set_next (prev, mono_mlist_next (current));
 334                 } else {
 335                         *list = mono_mlist_next (*list);
 336                 }
 337         }
 338
 339         return state;
 340 }
 341 #endif
 342
 343 static MonoMList *
 344 process_io_event (MonoMList *list, int event)
 345 {
 346         MonoSocketAsyncResult *state;
 347         MonoMList *oldlist;
 348
 349         oldlist = list;
 350         state = NULL;
 351         while (list) {
 352                 state = (MonoSocketAsyncResult *) mono_mlist_get_data (list);
 353                 if (get_event_from_state (state) == event)
 354                         break;
 355
 356                 list = mono_mlist_next (list);
 357         }
 358
 359         if (list != NULL) {
 360                 oldlist = mono_mlist_remove_item (oldlist, list);
 361                 EPOLL_DEBUG ("Dispatching event %d on socket %p", event, state->handle);
 362                 threadpool_append_job (&async_io_tp, (MonoObject *) state);
 363         }
 364
 365         return oldlist;
 366 }
 367
 368 static int
 369 mark_bad_fds (mono_pollfd *pfds, int nfds)
 370 {
 371         int i, ret;
 372         mono_pollfd *pfd;
 373         int count = 0;
 374
 375         for (i = 0; i < nfds; i++) {
 376                 pfd = &pfds [i];
 377                 if (pfd->fd == -1)
 378                         continue;
 379
 380                 ret = mono_poll (pfd, 1, 0);
 381                 if (ret == -1 && errno == EBADF) {
 382                         pfd->revents |= MONO_POLLNVAL;
 383                         count++;
 384                 } else if (ret == 1) {
 385                         count++;
 386                 }
 387         }
 388
 389         return count;
 390 }
 391
 392 static void
 393 socket_io_poll_main (gpointer p)
 394 {
 395 #if MONO_SMALL_CONFIG
 396 #define INITIAL_POLLFD_SIZE     128
 397 #else
 398 #define INITIAL_POLLFD_SIZE     1024
 399 #endif
 400 #define POLL_ERRORS (MONO_POLLERR | MONO_POLLHUP | MONO_POLLNVAL)
 401         SocketIOData *data = p;
 402         mono_pollfd *pfds;
 403         gint maxfd = 1;
 404         gint allocated;
 405         gint i;
 406         MonoInternalThread *thread;
 407
 408         thread = mono_thread_internal_current ();
 409
 410         allocated = INITIAL_POLLFD_SIZE;
 411         pfds = g_new0 (mono_pollfd, allocated);
 412         INIT_POLLFD (pfds, data->pipe [0], MONO_POLLIN);
 413         for (i = 1; i < allocated; i++)
 414                 INIT_POLLFD (&pfds [i], -1, 0);
 415
 416         while (1) {
 417                 int nsock = 0;
 418                 mono_pollfd *pfd;
 419                 char one [1];
 420                 MonoMList *list;
 421
 422                 do {
 423                         if (nsock == -1) {
 424                                 if (THREAD_WANTS_A_BREAK (thread))
 425                                         mono_thread_interruption_checkpoint ();
 426                         }
 427
 428                         nsock = mono_poll (pfds, maxfd, -1);
 429                 } while (nsock == -1 && errno == EINTR);
 430
 431                 /*
 432                  * Apart from EINTR, we only check EBADF, for the rest:
 433                  *  EINVAL: mono_poll() 'protects' us from descriptor
 434                  *      numbers above the limit if using select() by marking
 435                  *      then as MONO_POLLERR.  If a system poll() is being
 436                  *      used, the number of descriptor we're passing will not
 437                  *      be over sysconf(_SC_OPEN_MAX), as the error would have
 438                  *      happened when opening.
 439                  *
 440                  *  EFAULT: we own the memory pointed by pfds.
 441                  *  ENOMEM: we're doomed anyway
 442                  *
 443                  */
 444
 445                 if (nsock == -1 && errno == EBADF) {
 446                         pfds->revents = 0; /* Just in case... */
 447                         nsock = mark_bad_fds (pfds, maxfd);
 448                 }
 449
 450                 if ((pfds->revents & POLL_ERRORS) != 0) {
 451                         /* We're supposed to die now, as the pipe has been closed */
 452                         g_free (pfds);
 453                         socket_io_cleanup (data);
 454                         return;
 455                 }
 456
 457                 /* Got a new socket */
 458                 if ((pfds->revents & MONO_POLLIN) != 0) {
 459                         int nread;
 460
 461                         for (i = 1; i < allocated; i++) {
 462                                 pfd = &pfds [i];
 463                                 if (pfd->fd == -1 || data->newpfd == NULL ||
 464                                         pfd->fd == data->newpfd->fd)
 465                                         break;
 466                         }
 467
 468                         if (i == allocated) {
 469                                 mono_pollfd *oldfd;
 470
 471                                 oldfd = pfds;
 472                                 i = allocated;
 473                                 allocated = allocated * 2;
 474                                 pfds = g_renew (mono_pollfd, oldfd, allocated);
 475                                 g_free (oldfd);
 476                                 for (; i < allocated; i++)
 477                                         INIT_POLLFD (&pfds [i], -1, 0);
 478                         }
 479 #ifndef HOST_WIN32
 480                         nread = read (data->pipe [0], one, 1);
 481 #else
 482                         nread = recv ((SOCKET) data->pipe [0], one, 1, 0);
 483 #endif
 484                         if (nread <= 0) {
 485                                 g_free (pfds);
 486                                 return; /* we're closed */
 487                         }
 488
 489                         INIT_POLLFD (&pfds [i], data->newpfd->fd, data->newpfd->events);
 490                         ReleaseSemaphore (data->new_sem, 1, NULL);
 491                         if (i >= maxfd)
 492                                 maxfd = i + 1;
 493                         nsock--;
 494                 }
 495
 496                 if (nsock == 0)
 497                         continue;
 498
 499                 EnterCriticalSection (&data->io_lock);
 500                 if (data->inited == 3) {
 501                         g_free (pfds);
 502                         LeaveCriticalSection (&data->io_lock);
 503                         return; /* cleanup called */
 504                 }
 505
 506                 for (i = 1; i < maxfd && nsock > 0; i++) {
 507                         pfd = &pfds [i];
 508                         if (pfd->fd == -1 || pfd->revents == 0)
 509                                 continue;
 510
 511                         nsock--;
 512                         list = mono_g_hash_table_lookup (data->sock_to_state, GINT_TO_POINTER (pfd->fd));
 513                         if (list != NULL && (pfd->revents & (MONO_POLLIN | POLL_ERRORS)) != 0) {
 514                                 list = process_io_event (list, MONO_POLLIN);
 515                         }
 516
 517                         if (list != NULL && (pfd->revents & (MONO_POLLOUT | POLL_ERRORS)) != 0) {
 518                                 list = process_io_event (list, MONO_POLLOUT);
 519                         }
 520
 521                         if (list != NULL) {
 522                                 mono_g_hash_table_replace (data->sock_to_state, GINT_TO_POINTER (pfd->fd), list);
 523                                 pfd->events = get_events_from_list (list);
 524                         } else {
 525                                 mono_g_hash_table_remove (data->sock_to_state, GINT_TO_POINTER (pfd->fd));
 526                                 pfd->fd = -1;
 527                                 if (i == maxfd - 1)
 528                                         maxfd--;
 529                         }
 530                 }
 531                 LeaveCriticalSection (&data->io_lock);
 532         }
 533 }
 534
 535 #ifdef HAVE_EPOLL
 536 #define EPOLL_ERRORS (EPOLLERR | EPOLLHUP)
 537 #define EPOLL_NEVENTS   128
 538 static void
 539 socket_io_epoll_main (gpointer p)
 540 {
 541         SocketIOData *data;
 542         int epollfd;
 543         MonoInternalThread *thread;
 544         struct epoll_event *events, *evt;
 545         int ready = 0, i;
 546         gpointer async_results [EPOLL_NEVENTS * 2]; // * 2 because each loop can add up to 2 results here
 547         gint nresults;
 548
 549         data = p;
 550         epollfd = data->epollfd;
 551         thread = mono_thread_internal_current ();
 552         events = g_new0 (struct epoll_event, EPOLL_NEVENTS);
 553
 554         while (1) {
 555                 do {
 556                         if (ready == -1) {
 557                                 if (THREAD_WANTS_A_BREAK (thread))
 558                                         mono_thread_interruption_checkpoint ();
 559                         }
 560                         EPOLL_DEBUG ("epoll_wait init");
 561                         ready = epoll_wait (epollfd, events, EPOLL_NEVENTS, -1);
 562                         EPOLL_DEBUG_STMT(
 563                                 int err = errno;
 564                                 EPOLL_DEBUG ("epoll_wait end with %d ready sockets (%d %s).", ready, err, (ready == -1) ? g_strerror (err) : "");
 565                                 errno = err;
 566                         );
 567                 } while (ready == -1 && errno == EINTR);
 568
 569                 if (ready == -1) {
 570                         int err = errno;
 571                         g_free (events);
 572                         if (err != EBADF)
 573                                 g_warning ("epoll_wait: %d %s", err, g_strerror (err));
 574
 575                         close (epollfd);
 576                         return;
 577                 }
 578
 579                 EnterCriticalSection (&data->io_lock);
 580                 if (data->inited == 3) {
 581                         g_free (events);
 582                         close (epollfd);
 583                         LeaveCriticalSection (&data->io_lock);
 584                         return; /* cleanup called */
 585                 }
 586
 587                 nresults = 0;
 588                 for (i = 0; i < ready; i++) {
 589                         int fd;
 590                         MonoMList *list;
 591                         MonoObject *ares;
 592
 593                         evt = &events [i];
 594                         fd = evt->data.fd;
 595                         list = mono_g_hash_table_lookup (data->sock_to_state, GINT_TO_POINTER (fd));
 596                         EPOLL_DEBUG ("Event %d on %d list length: %d", evt->events, fd, mono_mlist_length (list));
 597                         if (list != NULL && (evt->events & (EPOLLIN | EPOLL_ERRORS)) != 0) {
 598                                 ares = get_io_event (&list, MONO_POLLIN);
 599                                 if (ares != NULL)
 600                                         async_results [nresults++] = ares;
 601                         }
 602
 603                         if (list != NULL && (evt->events & (EPOLLOUT | EPOLL_ERRORS)) != 0) {
 604                                 ares = get_io_event (&list, MONO_POLLOUT);
 605                                 if (ares != NULL)
 606                                         async_results [nresults++] = ares;
 607                         }
 608
 609                         if (list != NULL) {
 610                                 mono_g_hash_table_replace (data->sock_to_state, GINT_TO_POINTER (fd), list);
 611                                 evt->events = get_events_from_list (list);
 612                                 EPOLL_DEBUG ("MOD %d to %d", fd, evt->events);
 613                                 if (epoll_ctl (epollfd, EPOLL_CTL_MOD, fd, evt)) {
 614                                         if (epoll_ctl (epollfd, EPOLL_CTL_ADD, fd, evt) == -1) {
 615                                                 EPOLL_DEBUG_STMT (
 616                                                         int err = errno;
 617                                                         EPOLL_DEBUG ("epoll_ctl(MOD): %d %s fd: %d events: %d", err, g_strerror (err), fd, evt->events);
 618                                                         errno = err;
 619                                                 );
 620                                         }
 621                                 }
 622                         } else {
 623                                 mono_g_hash_table_remove (data->sock_to_state, GINT_TO_POINTER (fd));
 624                                 EPOLL_DEBUG ("DEL %d", fd);
 625                                 epoll_ctl (epollfd, EPOLL_CTL_DEL, fd, evt);
 626                         }
 627                 }
 628                 LeaveCriticalSection (&data->io_lock);
 629                 threadpool_append_jobs (&async_io_tp, (MonoObject **) async_results, nresults);
 630                 memset (async_results, 0, sizeof (gpointer) * nresults);
 631         }
 632 }
 633 #undef EPOLL_NEVENTS
 634 #endif
 635
 636 /*
 637  * select/poll wake up when a socket is closed, but epoll just removes
 638  * the socket from its internal list without notification.
 639  */
 640 void
 641 mono_thread_pool_remove_socket (int sock)
 642 {
 643         MonoMList *list, *next;
 644         MonoSocketAsyncResult *state;
 645
 646         if (socket_io_data.inited == 0)
 647                 return;
 648
 649         EnterCriticalSection (&socket_io_data.io_lock);
 650         if (socket_io_data.sock_to_state == NULL) {
 651                 LeaveCriticalSection (&socket_io_data.io_lock);
 652                 return;
 653         }
 654         list = mono_g_hash_table_lookup (socket_io_data.sock_to_state, GINT_TO_POINTER (sock));
 655         if (list)
 656                 mono_g_hash_table_remove (socket_io_data.sock_to_state, GINT_TO_POINTER (sock));
 657         LeaveCriticalSection (&socket_io_data.io_lock);
 658
 659         while (list) {
 660                 state = (MonoSocketAsyncResult *) mono_mlist_get_data (list);
 661                 if (state->operation == AIO_OP_RECEIVE)
 662                         state->operation = AIO_OP_RECV_JUST_CALLBACK;
 663                 else if (state->operation == AIO_OP_SEND)
 664                         state->operation = AIO_OP_SEND_JUST_CALLBACK;
 665
 666                 next = mono_mlist_remove_item (list, list);
 667                 list = process_io_event (list, MONO_POLLIN);
 668                 if (list)
 669                         process_io_event (list, MONO_POLLOUT);
 670
 671                 list = next;
 672         }
 673 }
 674
 675 #ifdef HOST_WIN32
 676 static void
 677 connect_hack (gpointer x)
 678 {
 679         struct sockaddr_in *addr = (struct sockaddr_in *) x;
 680         int count = 0;
 681
 682         while (connect ((SOCKET) socket_io_data.pipe [1], (SOCKADDR *) addr, sizeof (struct sockaddr_in))) {
 683                 Sleep (500);
 684                 if (++count > 3) {
 685                         g_warning ("Error initializing async. sockets %d.", WSAGetLastError ());
 686                         g_assert (WSAGetLastError ());
 687                 }
 688         }
 689 }
 690 #endif
 691
 692 static void
 693 socket_io_init (SocketIOData *data)
 694 {
 695 #ifdef HOST_WIN32
 696         struct sockaddr_in server;
 697         struct sockaddr_in client;
 698         SOCKET srv;
 699         int len;
 700 #endif
 701         int inited;
 702         guint32 stack_size;
 703
 704         if (data->inited >= 2) // 2 -> initialized, 3-> cleaned up
 705                 return;
 706
 707         inited = InterlockedCompareExchange (&data->inited, 1, 0);
 708         if (inited >= 1) {
 709                 while (TRUE) {
 710                         if (data->inited >= 2)
 711                                 return;
 712                         SleepEx (1, FALSE);
 713                 }
 714         }
 715
 716         EnterCriticalSection (&data->io_lock);
 717
 718 #ifdef HAVE_EPOLL
 719         data->epoll_disabled = (g_getenv ("MONO_DISABLE_AIO") != NULL);
 720         if (FALSE == data->epoll_disabled) {
 721                 data->epollfd = epoll_create (256);
 722                 data->epoll_disabled = (data->epollfd == -1);
 723                 if (data->epoll_disabled && g_getenv ("MONO_DEBUG"))
 724                         g_message ("epoll_create() failed. Using plain poll().");
 725         } else {
 726                 data->epollfd = -1;
 727         }
 728 #else
 729         data->epoll_disabled = TRUE;
 730 #endif
 731
 732 #ifndef HOST_WIN32
 733         if (data->epoll_disabled) {
 734                 if (pipe (data->pipe) != 0) {
 735                         int err = errno;
 736                         perror ("mono");
 737                         g_assert (err);
 738                 }
 739         } else {
 740                 data->pipe [0] = -1;
 741                 data->pipe [1] = -1;
 742         }
 743 #else
 744         srv = socket (AF_INET, SOCK_STREAM, IPPROTO_TCP);
 745         g_assert (srv != INVALID_SOCKET);
 746         data->pipe [1] = socket (AF_INET, SOCK_STREAM, IPPROTO_TCP);
 747         g_assert (data->pipe [1] != INVALID_SOCKET);
 748
 749         server.sin_family = AF_INET;
 750         server.sin_addr.s_addr = inet_addr ("127.0.0.1");
 751         server.sin_port = 0;
 752         if (bind (srv, (SOCKADDR *) &server, sizeof (server))) {
 753                 g_print ("%d\n", WSAGetLastError ());
 754                 g_assert (1 != 0);
 755         }
 756
 757         len = sizeof (server);
 758         getsockname (srv, (SOCKADDR *) &server, &len);
 759         listen (srv, 1);
 760         mono_thread_create (mono_get_root_domain (), connect_hack, &server);
 761         len = sizeof (server);
 762         data->pipe [0] = accept (srv, (SOCKADDR *) &client, &len);
 763         g_assert (data->pipe [0] != INVALID_SOCKET);
 764         closesocket (srv);
 765 #endif
 766         data->sock_to_state = mono_g_hash_table_new_type (g_direct_hash, g_direct_equal, MONO_HASH_VALUE_GC);
 767
 768         if (data->epoll_disabled) {
 769                 data->new_sem = CreateSemaphore (NULL, 1, 1, NULL);
 770                 g_assert (data->new_sem != NULL);
 771         }
 772
 773         stack_size = mono_threads_get_default_stacksize ();
 774         mono_threads_set_default_stacksize (128 * 1024);
 775         if (data->epoll_disabled) {
 776                 mono_thread_create_internal (mono_get_root_domain (), socket_io_poll_main, data, TRUE);
 777         }
 778 #ifdef HAVE_EPOLL
 779         else {
 780                 mono_thread_create_internal (mono_get_root_domain (), socket_io_epoll_main, data, TRUE);
 781         }
 782 #endif
 783         mono_threads_set_default_stacksize (stack_size);
 784         LeaveCriticalSection (&data->io_lock);
 785         data->inited = 2;
 786 }
 787
 788 static void
 789 socket_io_add_poll (MonoSocketAsyncResult *state)
 790 {
 791         int events;
 792         char msg [1];
 793         MonoMList *list;
 794         SocketIOData *data = &socket_io_data;
 795         int w;
 796
 797         if (mono_runtime_is_shutting_down () || data->inited == 3 || data->sock_to_state == NULL)
 798                 return;
 799
 800 #if defined(PLATFORM_MACOSX) || defined(PLATFORM_BSD) || defined(HOST_WIN32) || defined(PLATFORM_SOLARIS)
 801         /* select() for connect() does not work well on the Mac. Bug #75436. */
 802         /* Bug #77637 for the BSD 6 case */
 803         /* Bug #78888 for the Windows case */
 804         if (state->operation == AIO_OP_CONNECT && state->blocking == TRUE) {
 805                 //FIXME: increment number of threads while this one is waiting?
 806                 threadpool_append_job (&async_io_tp, (MonoObject *) state);
 807                 return;
 808         }
 809 #endif
 810         WaitForSingleObject (data->new_sem, INFINITE);
 811         if (data->newpfd == NULL)
 812                 data->newpfd = g_new0 (mono_pollfd, 1);
 813
 814         EnterCriticalSection (&data->io_lock);
 815         if (data->sock_to_state == NULL) {
 816                 LeaveCriticalSection (&data->io_lock);
 817                 return;
 818         }
 819
 820         /* FIXME: 64 bit issue: handle can be a pointer on windows? */
 821         list = mono_g_hash_table_lookup (data->sock_to_state, GINT_TO_POINTER (state->handle));
 822         if (list == NULL) {
 823                 list = mono_mlist_alloc ((MonoObject*)state);
 824         } else {
 825                 list = mono_mlist_append (list, (MonoObject*)state);
 826         }
 827
 828         events = get_events_from_list (list);
 829         INIT_POLLFD (data->newpfd, GPOINTER_TO_INT (state->handle), events);
 830         mono_g_hash_table_replace (data->sock_to_state, GINT_TO_POINTER (state->handle), list);
 831         LeaveCriticalSection (&data->io_lock);
 832         *msg = (char) state->operation;
 833 #ifndef HOST_WIN32
 834         w = write (data->pipe [1], msg, 1);
 835         w = w;
 836 #else
 837         send ((SOCKET) data->pipe [1], msg, 1, 0);
 838 #endif
 839 }
 840
 841 #ifdef HAVE_EPOLL
 842 static gboolean
 843 socket_io_add_epoll (MonoSocketAsyncResult *state)
 844 {
 845         MonoMList *list;
 846         SocketIOData *data = &socket_io_data;
 847         struct epoll_event event;
 848         int epoll_op, ievt;
 849         int fd;
 850
 851         if (mono_runtime_is_shutting_down () || data->inited == 3 || data->sock_to_state == NULL)
 852                 return TRUE;
 853
 854         memset (&event, 0, sizeof (struct epoll_event));
 855         fd = GPOINTER_TO_INT (state->handle);
 856         EnterCriticalSection (&data->io_lock);
 857         if (data->sock_to_state == NULL) {
 858                 LeaveCriticalSection (&data->io_lock);
 859                 return TRUE;
 860         }
 861         list = mono_g_hash_table_lookup (data->sock_to_state, GINT_TO_POINTER (fd));
 862         if (list == NULL) {
 863                 list = mono_mlist_alloc ((MonoObject*)state);
 864                 epoll_op = EPOLL_CTL_ADD;
 865         } else {
 866                 list = mono_mlist_append (list, (MonoObject*)state);
 867                 epoll_op = EPOLL_CTL_MOD;
 868         }
 869
 870         ievt = get_events_from_list (list);
 871         if ((ievt & MONO_POLLIN) != 0)
 872                 event.events |= EPOLLIN;
 873         if ((ievt & MONO_POLLOUT) != 0)
 874                 event.events |= EPOLLOUT;
 875
 876         mono_g_hash_table_replace (data->sock_to_state, state->handle, list);
 877         event.data.fd = fd;
 878         EPOLL_DEBUG ("%s %d with %d", epoll_op == EPOLL_CTL_ADD ? "ADD" : "MOD", fd, event.events);
 879         if (epoll_ctl (data->epollfd, epoll_op, fd, &event) == -1) {
 880                 int err = errno;
 881                 if (epoll_op == EPOLL_CTL_ADD && err == EEXIST) {
 882                         epoll_op = EPOLL_CTL_MOD;
 883                         if (epoll_ctl (data->epollfd, epoll_op, fd, &event) == -1) {
 884                                 g_message ("epoll_ctl(MOD): %d %s", err, g_strerror (err));
 885                         }
 886                 }
 887         }
 888         LeaveCriticalSection (&data->io_lock);
 889
 890         return TRUE;
 891 }
 892 #endif
 893
 894 static void
 895 socket_io_add (MonoAsyncResult *ares, MonoSocketAsyncResult *state)
 896 {
 897         if (async_tp.pool_status == 2 || mono_runtime_is_shutting_down ())
 898                 return;
 899
 900         socket_io_init (&socket_io_data);
 901
 902         MONO_OBJECT_SETREF (state, ares, ares);
 903 #ifdef HAVE_EPOLL
 904         if (socket_io_data.epoll_disabled == FALSE) {
 905                 if (socket_io_add_epoll (state))
 906                         return;
 907         }
 908 #endif
 909         socket_io_add_poll (state);
 910 }
 911
 912 #ifndef DISABLE_SOCKETS
 913 static gboolean
 914 socket_io_filter (MonoObject *target, MonoObject *state)
 915 {
 916         gint op;
 917         MonoSocketAsyncResult *sock_res = (MonoSocketAsyncResult *) state;
 918         MonoClass *klass;
 919
 920         if (target == NULL || state == NULL)
 921                 return FALSE;
 922
 923         if (socket_async_call_klass == NULL) {
 924                 klass = target->vtable->klass;
 925                 /* Check if it's SocketAsyncCall in System.Net.Sockets
 926                  * FIXME: check the assembly is signed correctly for extra care
 927                  */
 928                 if (klass->name [0] == 'S' && strcmp (klass->name, "SocketAsyncCall") == 0
 929                                 && strcmp (mono_image_get_name (klass->image), "System") == 0
 930                                 && klass->nested_in && strcmp (klass->nested_in->name, "Socket") == 0)
 931                         socket_async_call_klass = klass;
 932         }
 933
 934         if (process_async_call_klass == NULL) {
 935                 klass = target->vtable->klass;
 936                 /* Check if it's AsyncReadHandler in System.Diagnostics.Process
 937                  * FIXME: check the assembly is signed correctly for extra care
 938                  */
 939                 if (klass->name [0] == 'A' && strcmp (klass->name, "AsyncReadHandler") == 0
 940                                 && strcmp (mono_image_get_name (klass->image), "System") == 0
 941                                 && klass->nested_in && strcmp (klass->nested_in->name, "Process") == 0)
 942                         process_async_call_klass = klass;
 943         }
 944         /* return both when socket_async_call_klass has not been seen yet and when
 945          * the object is not an instance of the class.
 946          */
 947         if (target->vtable->klass != socket_async_call_klass && target->vtable->klass != process_async_call_klass)
 948                 return FALSE;
 949
 950         op = sock_res->operation;
 951         if (op < AIO_OP_FIRST || op >= AIO_OP_LAST)
 952                 return FALSE;
 953
 954         return TRUE;
 955 }
 956 #endif /* !DISABLE_SOCKETS */
 957
 958 /* Returns the exception thrown when invoking, if any */
 959 static MonoObject *
 960 mono_async_invoke (ThreadPool *tp, MonoAsyncResult *ares)
 961 {
 962         ASyncCall *ac = (ASyncCall *)ares->object_data;
 963         MonoObject *res, *exc = NULL;
 964         MonoArray *out_args = NULL;
 965         HANDLE wait_event = NULL;
 966
 967         if (ares->execution_context) {
 968                 /* use captured ExecutionContext (if available) */
 969                 MONO_OBJECT_SETREF (ares, original_context, mono_thread_get_execution_context ());
 970                 mono_thread_set_execution_context (ares->execution_context);
 971         } else {
 972                 ares->original_context = NULL;
 973         }
 974
 975         if (ac == NULL) {
 976                 /* Fast path from ThreadPool.*QueueUserWorkItem */
 977                 void *pa = ares->async_state;
 978                 mono_runtime_delegate_invoke (ares->async_delegate, &pa, &exc);
 979         } else {
 980                 ac->msg->exc = NULL;
 981                 res = mono_message_invoke (ares->async_delegate, ac->msg, &exc, &out_args);
 982                 MONO_OBJECT_SETREF (ac, res, res);
 983                 MONO_OBJECT_SETREF (ac, msg->exc, exc);
 984                 MONO_OBJECT_SETREF (ac, out_args, out_args);
 985
 986                 mono_monitor_enter ((MonoObject *) ares);
 987                 ares->completed = 1;
 988                 if (ares->handle != NULL)
 989                         wait_event = mono_wait_handle_get_handle ((MonoWaitHandle *) ares->handle);
 990                 mono_monitor_exit ((MonoObject *) ares);
 991                 /* notify listeners */
 992                 if (wait_event != NULL)
 993                         SetEvent (wait_event);
 994
 995                 /* call async callback if cb_method != null*/
 996                 if (ac != NULL && ac->cb_method) {
 997                         MonoObject *exc = NULL;
 998                         void *pa = &ares;
 999                         mono_runtime_invoke (ac->cb_method, ac->cb_target, pa, &exc);
1000                         /* 'exc' will be the previous ac->msg->exc if not NULL and not
1001                          * catched. If catched, this will be set to NULL and the
1002                          * exception will not be printed. */
1003                         MONO_OBJECT_SETREF (ac->msg, exc, exc);
1004                 }
1005         }
1006
1007         /* restore original thread execution context if flow isn't suppressed, i.e. non null */
1008         if (ares->original_context) {
1009                 mono_thread_set_execution_context (ares->original_context);
1010                 ares->original_context = NULL;
1011         }
1012         return exc;
1013 }
1014
1015 static void
1016 threadpool_start_idle_threads (ThreadPool *tp)
1017 {
1018         int n;
1019
1020         do {
1021                 while (1) {
1022                         n = tp->nthreads;
1023                         if (n >= tp->min_threads)
1024                                 return;
1025                         if (InterlockedCompareExchange (&tp->nthreads, n + 1, n) == n)
1026                                 break;
1027                 }
1028                 mono_perfcounter_update_value (tp->pc_nthreads, TRUE, 1);
1029                 mono_thread_create_internal (mono_get_root_domain (), tp->async_invoke, tp, TRUE);
1030                 SleepEx (100, TRUE);
1031         } while (1);
1032 }
1033
1034 static void
1035 threadpool_init (ThreadPool *tp, int min_threads, int max_threads, void (*async_invoke) (gpointer))
1036 {
1037         memset (tp, 0, sizeof (ThreadPool));
1038         MONO_SEM_INIT (&tp->lock, 1);
1039         tp->min_threads = min_threads;
1040         tp->max_threads = max_threads;
1041         tp->async_invoke = async_invoke;
1042         MONO_SEM_INIT (&tp->new_job, 0);
1043 }
1044
1045 static void *
1046 init_perf_counter (const char *category, const char *counter)
1047 {
1048         MonoString *category_str;
1049         MonoString *counter_str;
1050         MonoString *machine;
1051         MonoDomain *root;
1052         MonoBoolean custom;
1053         int type;
1054
1055         if (category == NULL || counter == NULL)
1056                 return NULL;
1057         root = mono_get_root_domain ();
1058         category_str = mono_string_new (root, category);
1059         counter_str = mono_string_new (root, counter);
1060         machine = mono_string_new (root, ".");
1061         return mono_perfcounter_get_impl (category_str, counter_str, NULL, machine, &type, &custom);
1062 }
1063
1064 #ifdef DEBUG
1065 static void
1066 print_pool_info (ThreadPool *tp)
1067 {
1068
1069 //      if (tp->tail - tp->head == 0)
1070 //              return;
1071
1072         g_print ("Pool status? %d\n", InterlockedCompareExchange (&tp->pool_status, 0, 0));
1073         g_print ("Min. threads: %d\n", InterlockedCompareExchange (&tp->min_threads, 0, 0));
1074         g_print ("Max. threads: %d\n", InterlockedCompareExchange (&tp->max_threads, 0, 0));
1075         g_print ("nthreads: %d\n", InterlockedCompareExchange (&tp->nthreads, 0, 0));
1076         g_print ("busy threads: %d\n", InterlockedCompareExchange (&tp->busy_threads, 0, 0));
1077         g_print ("Waiting: %d\n", InterlockedCompareExchange (&tp->waiting, 0, 0));
1078         g_print ("Queued: %d\n", (tp->tail - tp->head));
1079         if (tp == &async_tp) {
1080                 int i;
1081                 EnterCriticalSection (&wsqs_lock);
1082                 for (i = 0; i < wsqs->len; i++) {
1083                         g_print ("\tWSQ %d: %d\n", i, mono_wsq_count (g_ptr_array_index (wsqs, i)));
1084                 }
1085                 LeaveCriticalSection (&wsqs_lock);
1086         } else {
1087                 g_print ("\tSockets: %d\n", mono_g_hash_table_size (socket_io_data.sock_to_state));
1088         }
1089         g_print ("-------------\n");
1090 }
1091
1092 static void
1093 signal_handler (int signo)
1094 {
1095         ThreadPool *tp;
1096
1097         tp = &async_tp;
1098         MONO_SEM_WAIT (&tp->lock);
1099         g_print ("\n-----Non-IO-----\n");
1100         print_pool_info (tp);
1101         MONO_SEM_POST (&tp->lock);
1102         tp = &async_io_tp;
1103         MONO_SEM_WAIT (&tp->lock);
1104         g_print ("\n-----IO-----\n");
1105         print_pool_info (tp);
1106         MONO_SEM_POST (&tp->lock);
1107         alarm (2);
1108 }
1109 #endif
1110
1111 void
1112 mono_thread_pool_init ()
1113 {
1114         gint threads_per_cpu = 1;
1115         gint thread_count;
1116         gint cpu_count = mono_cpu_count ();
1117         int result;
1118
1119         if (tp_inited == 2)
1120                 return;
1121
1122         result = InterlockedCompareExchange (&tp_inited, 1, 0);
1123         if (result == 1) {
1124                 while (1) {
1125                         SleepEx (1, FALSE);
1126                         if (tp_inited == 2)
1127                                 return;
1128                 }
1129         }
1130
1131         MONO_GC_REGISTER_ROOT (async_tp.first);
1132         MONO_GC_REGISTER_ROOT (async_tp.last);
1133         MONO_GC_REGISTER_ROOT (async_tp.unused);
1134         MONO_GC_REGISTER_ROOT (async_io_tp.first);
1135         MONO_GC_REGISTER_ROOT (async_io_tp.unused);
1136         MONO_GC_REGISTER_ROOT (async_io_tp.last);
1137
1138         MONO_GC_REGISTER_ROOT (socket_io_data.sock_to_state);
1139         InitializeCriticalSection (&socket_io_data.io_lock);
1140         if (g_getenv ("MONO_THREADS_PER_CPU") != NULL) {
1141                 threads_per_cpu = atoi (g_getenv ("MONO_THREADS_PER_CPU"));
1142                 if (threads_per_cpu < 1)
1143                         threads_per_cpu = 1;
1144         }
1145
1146         thread_count = MIN (cpu_count * threads_per_cpu, 100 * cpu_count);
1147         threadpool_init (&async_tp, thread_count, MAX (100 * cpu_count, thread_count), async_invoke_thread);
1148         threadpool_init (&async_io_tp, cpu_count * 2, cpu_count * 4, async_invoke_thread);
1149         async_io_tp.is_io = TRUE;
1150
1151         async_call_klass = mono_class_from_name (mono_defaults.corlib, "System", "MonoAsyncCall");
1152         g_assert (async_call_klass);
1153
1154         InitializeCriticalSection (&wsqs_lock);
1155         wsqs = g_ptr_array_sized_new (MAX (100 * cpu_count, thread_count));
1156         mono_wsq_init ();
1157
1158         async_tp.pc_nitems = init_perf_counter ("Mono Threadpool", "Work Items Added");
1159         g_assert (async_tp.pc_nitems);
1160
1161         async_io_tp.pc_nitems = init_perf_counter ("Mono Threadpool", "IO Work Items Added");
1162         g_assert (async_io_tp.pc_nitems);
1163
1164         async_tp.pc_nthreads = init_perf_counter ("Mono Threadpool", "# of Threads");
1165         g_assert (async_tp.pc_nthreads);
1166
1167         async_io_tp.pc_nthreads = init_perf_counter ("Mono Threadpool", "# of IO Threads");
1168         g_assert (async_io_tp.pc_nthreads);
1169         tp_inited = 2;
1170 #ifdef DEBUG
1171         signal (SIGALRM, signal_handler);
1172         alarm (2);
1173 #endif
1174 }
1175
1176 MonoAsyncResult *
1177 mono_thread_pool_add (MonoObject *target, MonoMethodMessage *msg, MonoDelegate *async_callback,
1178                       MonoObject *state)
1179 {
1180         MonoDomain *domain = mono_domain_get ();
1181         MonoAsyncResult *ares;
1182         ASyncCall *ac;
1183
1184         ac = (ASyncCall*)mono_object_new (domain, async_call_klass);
1185         MONO_OBJECT_SETREF (ac, msg, msg);
1186         MONO_OBJECT_SETREF (ac, state, state);
1187
1188         if (async_callback) {
1189                 ac->cb_method = mono_get_delegate_invoke (((MonoObject *)async_callback)->vtable->klass);
1190                 MONO_OBJECT_SETREF (ac, cb_target, async_callback);
1191         }
1192
1193         ares = mono_async_result_new (domain, NULL, ac->state, NULL, (MonoObject*)ac);
1194         MONO_OBJECT_SETREF (ares, async_delegate, target);
1195
1196 #ifndef DISABLE_SOCKETS
1197         if (socket_io_filter (target, state)) {
1198                 socket_io_add (ares, (MonoSocketAsyncResult *) state);
1199                 return ares;
1200         }
1201 #endif
1202         threadpool_append_job (&async_tp, (MonoObject *) ares);
1203         return ares;
1204 }
1205
1206 MonoObject *
1207 mono_thread_pool_finish (MonoAsyncResult *ares, MonoArray **out_args, MonoObject **exc)
1208 {
1209         ASyncCall *ac;
1210         HANDLE wait_event;
1211
1212         *exc = NULL;
1213         *out_args = NULL;
1214
1215         /* check if already finished */
1216         mono_monitor_enter ((MonoObject *) ares);
1217
1218         if (ares->endinvoke_called) {
1219                 *exc = (MonoObject *) mono_get_exception_invalid_operation (NULL);
1220                 mono_monitor_exit ((MonoObject *) ares);
1221                 return NULL;
1222         }
1223
1224         ares->endinvoke_called = 1;
1225         /* wait until we are really finished */
1226         if (!ares->completed) {
1227                 if (ares->handle == NULL) {
1228                         wait_event = CreateEvent (NULL, TRUE, FALSE, NULL);
1229                         g_assert(wait_event != 0);
1230                         MONO_OBJECT_SETREF (ares, handle, (MonoObject *) mono_wait_handle_new (mono_object_domain (ares), wait_event));
1231                 } else {
1232                         wait_event = mono_wait_handle_get_handle ((MonoWaitHandle *) ares->handle);
1233                 }
1234                 mono_monitor_exit ((MonoObject *) ares);
1235                 WaitForSingleObjectEx (wait_event, INFINITE, TRUE);
1236         } else {
1237                 mono_monitor_exit ((MonoObject *) ares);
1238         }
1239
1240         ac = (ASyncCall *) ares->object_data;
1241         g_assert (ac != NULL);
1242         *exc = ac->msg->exc; /* FIXME: GC add write barrier */
1243         *out_args = ac->out_args;
1244
1245         return ac->res;
1246 }
1247
1248 static void
1249 threadpool_kill_idle_threads (ThreadPool *tp)
1250 {
1251         gint n;
1252
1253         n = (gint) InterlockedCompareExchange (&tp->max_threads, 0, -1);
1254         while (n) {
1255                 n--;
1256                 MONO_SEM_POST (&tp->new_job);
1257         }
1258 }
1259
1260 void
1261 mono_thread_pool_cleanup (void)
1262 {
1263         if (async_tp.pool_status == 0 || async_tp.pool_status == 2)
1264                 return;
1265
1266         if (async_tp.pool_status == 1 && InterlockedCompareExchange (&async_tp.pool_status, 2, 1) == 2)
1267                 return;
1268
1269         InterlockedExchange (&async_io_tp.pool_status, 2);
1270         MONO_SEM_WAIT (&async_tp.lock);
1271         threadpool_free_queue (&async_tp);
1272         threadpool_kill_idle_threads (&async_tp);
1273         MONO_SEM_POST (&async_tp.lock);
1274
1275         socket_io_cleanup (&socket_io_data); /* Empty when DISABLE_SOCKETS is defined */
1276         MONO_SEM_WAIT (&async_io_tp.lock);
1277         threadpool_free_queue (&async_io_tp);
1278         threadpool_kill_idle_threads (&async_io_tp);
1279         MONO_SEM_POST (&async_io_tp.lock);
1280         MONO_SEM_DESTROY (&async_io_tp.new_job);
1281
1282         EnterCriticalSection (&wsqs_lock);
1283         mono_wsq_cleanup ();
1284         if (wsqs)
1285                 g_ptr_array_free (wsqs, TRUE);
1286         wsqs = NULL;
1287         LeaveCriticalSection (&wsqs_lock);
1288         MONO_SEM_DESTROY (&async_tp.new_job);
1289 }
1290
1291 /* Caller must enter &tp->lock */
1292 static MonoObject*
1293 dequeue_job_nolock (ThreadPool *tp)
1294 {
1295         MonoObject *ar;
1296         MonoArray *array;
1297         MonoMList *list;
1298
1299         list = tp->first;
1300         do {
1301                 if (mono_runtime_is_shutting_down ())
1302                         return NULL;
1303                 if (!list || tp->head == tp->tail)
1304                         return NULL;
1305
1306                 array = (MonoArray *) mono_mlist_get_data (list);
1307                 ar = mono_array_get (array, MonoObject *, tp->head % QUEUE_LENGTH);
1308                 mono_array_set (array, MonoObject *, tp->head % QUEUE_LENGTH, NULL);
1309                 tp->head++;
1310                 if ((tp->head % QUEUE_LENGTH) == 0) {
1311                         list = tp->first;
1312                         tp->first = mono_mlist_next (list);
1313                         if (tp->first == NULL)
1314                                 tp->last = NULL;
1315                         if (mono_mlist_length (tp->unused) < 20) {
1316                                 /* reuse this chunk */
1317                                 tp->unused = mono_mlist_set_next (list, tp->unused);
1318                         }
1319                         tp->head -= QUEUE_LENGTH;
1320                         tp->tail -= QUEUE_LENGTH;
1321                 }
1322                 list = tp->first;
1323         } while (ar == NULL);
1324         return ar;
1325 }
1326
1327 static gboolean
1328 threadpool_start_thread (ThreadPool *tp)
1329 {
1330         gint n;
1331
1332         while (!mono_runtime_is_shutting_down () && (n = tp->nthreads) < tp->max_threads) {
1333                 if (InterlockedCompareExchange (&tp->nthreads, n + 1, n) == n) {
1334                         mono_perfcounter_update_value (tp->pc_nthreads, TRUE, 1);
1335                         mono_thread_create_internal (mono_get_root_domain (), tp->async_invoke, tp, TRUE);
1336                         return TRUE;
1337                 }
1338         }
1339
1340         return FALSE;
1341 }
1342
1343 static void
1344 pulse_on_new_job (ThreadPool *tp)
1345 {
1346         if (tp->waiting)
1347                 MONO_SEM_POST (&tp->new_job);
1348 }
1349
1350 void
1351 icall_append_job (MonoObject *ar)
1352 {
1353         threadpool_append_job (&async_tp, ar);
1354 }
1355
1356 static void
1357 threadpool_append_job (ThreadPool *tp, MonoObject *ar)
1358 {
1359         threadpool_append_jobs (tp, &ar, 1);
1360 }
1361
1362 static MonoMList *
1363 create_or_reuse_list (ThreadPool *tp)
1364 {
1365         MonoMList *list;
1366         MonoArray *array;
1367
1368         list = NULL;
1369         if (tp->unused) {
1370                 list = tp->unused;
1371                 tp->unused = mono_mlist_next (list);
1372                 mono_mlist_set_next (list, NULL);
1373                 //TP_DEBUG (tp->nodes_reused++);
1374         } else {
1375                 array = mono_array_new_cached (mono_get_root_domain (), mono_defaults.object_class, QUEUE_LENGTH);
1376                 list = mono_mlist_alloc ((MonoObject *) array);
1377                 //TP_DEBUG (tp->nodes_created++);
1378         }
1379         return list;
1380 }
1381
1382 static void
1383 threadpool_append_jobs (ThreadPool *tp, MonoObject **jobs, gint njobs)
1384 {
1385         static int job_counter;
1386         MonoArray *array;
1387         MonoMList *list;
1388         MonoObject *ar;
1389         gint i;
1390         gboolean lock_taken = FALSE; /* We won't take the lock when the local queue is used */
1391
1392         if (mono_runtime_is_shutting_down ())
1393                 return;
1394
1395         if (tp->pool_status == 0 && InterlockedCompareExchange (&tp->pool_status, 1, 0) == 0)
1396                 mono_thread_create_internal (mono_get_root_domain (), threadpool_start_idle_threads, tp, TRUE);
1397
1398         for (i = 0; i < njobs; i++) {
1399                 ar = jobs [i];
1400                 if (ar == NULL || mono_domain_is_unloading (ar->vtable->domain))
1401                         continue; /* Might happen when cleaning domain jobs */
1402                 if (!tp->is_io && (InterlockedIncrement (&job_counter) % 10) == 0) {
1403                         MonoAsyncResult *o = (MonoAsyncResult *) ar;
1404                         o->add_time = mono_100ns_ticks ();
1405                 }
1406                 threadpool_jobs_inc (ar);
1407                 mono_perfcounter_update_value (tp->pc_nitems, TRUE, 1);
1408                 if (!tp->is_io && mono_wsq_local_push (ar))
1409                         continue;
1410
1411                 if (!lock_taken) {
1412                         MONO_SEM_WAIT (&tp->lock);
1413                         lock_taken = TRUE;
1414                 }
1415                 if ((tp->tail % QUEUE_LENGTH) == 0) {
1416                         list = create_or_reuse_list (tp);
1417                         if (tp->last != NULL)
1418                                 mono_mlist_set_next (tp->last, list);
1419                         tp->last = list;
1420                         if (tp->first == NULL)
1421                                 tp->first = tp->last;
1422                 }
1423
1424                 array = (MonoArray *) mono_mlist_get_data (tp->last);
1425                 mono_array_setref (array, tp->tail % QUEUE_LENGTH, ar);
1426                 tp->tail++;
1427         }
1428         if (lock_taken)
1429                 MONO_SEM_POST (&tp->lock);
1430
1431         if (!tp->is_io && tp->waiting == 0) {
1432                 gint64 ticks = mono_100ns_ticks ();
1433
1434                 if (tp->last_check == 0 || (ticks - tp->last_check) > 5000000) {
1435                         SPIN_LOCK (tp->sp_lock);
1436                         tp->last_check = ticks;
1437                         SPIN_UNLOCK (tp->sp_lock);
1438                         threadpool_start_thread (tp);
1439                 }
1440         }
1441         for (i = 0; i < MIN(njobs, tp->max_threads); i++)
1442                 pulse_on_new_job (tp);
1443 }
1444
1445 static void
1446 threadpool_clear_queue (ThreadPool *tp, MonoDomain *domain)
1447 {
1448         MonoMList *current;
1449         MonoArray *array;
1450         MonoObject *obj;
1451         int domain_count;
1452         int i;
1453
1454         domain_count = 0;
1455         MONO_SEM_WAIT (&tp->lock);
1456         current = tp->first;
1457         while (current) {
1458                 array = (MonoArray *) mono_mlist_get_data (current);
1459                 for (i = 0; i < QUEUE_LENGTH; i++) {
1460                         obj = mono_array_get (array, MonoObject*, i);
1461                         if (obj != NULL && obj->vtable->domain == domain) {
1462                                 domain_count++;
1463                                 mono_array_setref (array, i, NULL);
1464                                 threadpool_jobs_dec (obj);
1465                         }
1466                 }
1467                 current = mono_mlist_next (current);
1468         }
1469
1470         if (!domain_count) {
1471                 MONO_SEM_POST (&tp->lock);
1472                 return;
1473         }
1474
1475         current = tp->first;
1476         tp->first = NULL;
1477         tp->last = NULL;
1478         tp->head = 0;
1479         tp->tail = 0;
1480         MONO_SEM_POST (&tp->lock);
1481         /* Re-add everything but the nullified elements */
1482         while (current) {
1483                 array = (MonoArray *) mono_mlist_get_data (current);
1484                 threadpool_append_jobs (tp, mono_array_addr (array, MonoObject *, 0), QUEUE_LENGTH);
1485                 memset (mono_array_addr (array, MonoObject *, 0), 0, sizeof (MonoObject *) * QUEUE_LENGTH);
1486                 current = mono_mlist_next (current);
1487         }
1488 }
1489
1490 /*
1491  * Clean up the threadpool of all domain jobs.
1492  * Can only be called as part of the domain unloading process as
1493  * it will wait for all jobs to be visible to the interruption code.
1494  */
1495 gboolean
1496 mono_thread_pool_remove_domain_jobs (MonoDomain *domain, int timeout)
1497 {
1498         HANDLE sem_handle;
1499         int result = TRUE;
1500         guint32 start_time = 0;
1501
1502         g_assert (domain->state == MONO_APPDOMAIN_UNLOADING);
1503
1504         threadpool_clear_queue (&async_tp, domain);
1505         threadpool_clear_queue (&async_io_tp, domain);
1506
1507         /*
1508          * There might be some threads out that could be about to execute stuff from the given domain.
1509          * We avoid that by setting up a semaphore to be pulsed by the thread that reaches zero.
1510          */
1511         sem_handle = CreateSemaphore (NULL, 0, 1, NULL);
1512
1513         domain->cleanup_semaphore = sem_handle;
1514         /*
1515          * The memory barrier here is required to have global ordering between assigning to cleanup_semaphone
1516          * and reading threadpool_jobs.
1517          * Otherwise this thread could read a stale version of threadpool_jobs and wait forever.
1518          */
1519         mono_memory_write_barrier ();
1520
1521         if (domain->threadpool_jobs && timeout != -1)
1522                 start_time = mono_msec_ticks ();
1523         while (domain->threadpool_jobs) {
1524                 WaitForSingleObject (sem_handle, timeout);
1525                 if (timeout != -1 && (mono_msec_ticks () - start_time) > timeout) {
1526                         result = FALSE;
1527                         break;
1528                 }
1529         }
1530
1531         domain->cleanup_semaphore = NULL;
1532         CloseHandle (sem_handle);
1533         return result;
1534 }
1535
1536 static void
1537 threadpool_free_queue (ThreadPool *tp)
1538 {
1539         tp->head = tp->tail = 0;
1540         tp->first = NULL;
1541         tp->unused = NULL;
1542 }
1543
1544 gboolean
1545 mono_thread_pool_is_queue_array (MonoArray *o)
1546 {
1547         gpointer obj = o;
1548
1549         // FIXME: need some fix in sgen code.
1550         // There are roots at: async*tp.unused (MonoMList) and wsqs [n]->queue (MonoArray)
1551         return obj == async_tp.first || obj == async_io_tp.first;
1552 }
1553
1554 static void
1555 add_wsq (MonoWSQ *wsq)
1556 {
1557         int i;
1558
1559         if (wsq == NULL)
1560                 return;
1561
1562         EnterCriticalSection (&wsqs_lock);
1563         if (wsqs == NULL) {
1564                 LeaveCriticalSection (&wsqs_lock);
1565                 return;
1566         }
1567         for (i = 0; i < wsqs->len; i++) {
1568                 if (g_ptr_array_index (wsqs, i) == NULL) {
1569                         wsqs->pdata [i] = wsq;
1570                         LeaveCriticalSection (&wsqs_lock);
1571                         return;
1572                 }
1573         }
1574         g_ptr_array_add (wsqs, wsq);
1575         LeaveCriticalSection (&wsqs_lock);
1576 }
1577
1578 static void
1579 remove_wsq (MonoWSQ *wsq)
1580 {
1581         if (wsq == NULL)
1582                 return;
1583
1584         EnterCriticalSection (&wsqs_lock);
1585         if (wsqs == NULL) {
1586                 LeaveCriticalSection (&wsqs_lock);
1587                 return;
1588         }
1589         g_ptr_array_remove_fast (wsqs, wsq);
1590         LeaveCriticalSection (&wsqs_lock);
1591 }
1592
1593 static void
1594 try_steal (gpointer *data, gboolean retry)
1595 {
1596         int i;
1597         int ms;
1598
1599         if (wsqs == NULL || data == NULL || *data != NULL)
1600                 return;
1601
1602         ms = 0;
1603         do {
1604                 if (mono_runtime_is_shutting_down ())
1605                         return;
1606                 for (i = 0; wsqs != NULL && i < wsqs->len; i++) {
1607                         if (mono_runtime_is_shutting_down ()) {
1608                                 return;
1609                         }
1610                         mono_wsq_try_steal (wsqs->pdata [i], data, ms);
1611                         if (*data != NULL) {
1612                                 return;
1613                         }
1614                 }
1615                 ms += 10;
1616         } while (retry && ms < 11);
1617 }
1618
1619 static gboolean
1620 dequeue_or_steal (ThreadPool *tp, gpointer *data)
1621 {
1622         if (mono_runtime_is_shutting_down ())
1623                 return FALSE;
1624         TP_DEBUG ("Dequeue");
1625         MONO_SEM_WAIT (&tp->lock);
1626         *data = dequeue_job_nolock (tp);
1627         MONO_SEM_POST (&tp->lock);
1628         if (!tp->is_io && !*data)
1629                 try_steal (data, FALSE);
1630         return (*data != NULL);
1631 }
1632
1633 static void
1634 process_idle_times (ThreadPool *tp, gint64 t)
1635 {
1636         gint64 ticks;
1637         gint64 avg;
1638         gboolean compute_avg;
1639         gint new_threads;
1640         gint64 per1;
1641
1642         if (tp->ignore_times || t <= 0)
1643                 return;
1644
1645         compute_avg = FALSE;
1646         ticks = mono_100ns_ticks ();
1647         t = ticks - t;
1648         SPIN_LOCK (tp->sp_lock);
1649         if (tp->ignore_times) {
1650                 SPIN_UNLOCK (tp->sp_lock);
1651                 return;
1652         }
1653         tp->time_sum += t;
1654         tp->n_sum++;
1655         if (tp->last_check == 0)
1656                 tp->last_check = ticks;
1657         else if (tp->last_check > 0 && (ticks - tp->last_check) > 5000000) {
1658                 tp->ignore_times = 1;
1659                 compute_avg = TRUE;
1660         }
1661         SPIN_UNLOCK (tp->sp_lock);
1662
1663         if (!compute_avg)
1664                 return;
1665
1666         //printf ("Items: %d Time elapsed: %.3fs\n", tp->n_sum, (ticks - tp->last_check) / 10000.0);
1667         tp->last_check = ticks;
1668         new_threads = 0;
1669         avg = tp->time_sum / tp->n_sum;
1670         if (tp->averages [1] == 0) {
1671                 tp->averages [1] = avg;
1672         } else {
1673                 per1 = ((100 * (ABS (avg - tp->averages [1]))) / tp->averages [1]);
1674                 if (per1 > 5) {
1675                         if (avg > tp->averages [1]) {
1676                                 if (tp->averages [1] < tp->averages [0]) {
1677                                         new_threads = -1;
1678                                 } else {
1679                                         new_threads = 1;
1680                                 }
1681                         } else if (avg < tp->averages [1] && tp->averages [1] < tp->averages [0]) {
1682                                 new_threads = 1;
1683                         }
1684                 } else {
1685                         int min, n;
1686                         min = tp->min_threads;
1687                         n = tp->nthreads;
1688                         if ((n - min) < min && tp->busy_threads == n)
1689                                 new_threads = 1;
1690                 }
1691                 /*
1692                 if (new_threads != 0) {
1693                         printf ("n: %d per1: %lld avg=%lld avg1=%lld avg0=%lld\n", new_threads, per1, avg, tp->averages [1], tp->averages [0]);
1694                 }
1695                 */
1696         }
1697
1698         tp->time_sum = 0;
1699         tp->n_sum = 0;
1700
1701         tp->averages [0] = tp->averages [1];
1702         tp->averages [1] = avg;
1703         tp->ignore_times = 0;
1704
1705         if (tp->waiting == 0 && new_threads == 1) {
1706                 threadpool_start_thread (tp);
1707         } else if (new_threads == -1) {
1708                 if (tp->destroy_thread == 0 && InterlockedCompareExchange (&tp->destroy_thread, 1, 0) == 0)
1709                         pulse_on_new_job (tp);
1710         }
1711 }
1712
1713 static gboolean
1714 should_i_die (ThreadPool *tp)
1715 {
1716         gboolean result = FALSE;
1717         if (tp->destroy_thread == 1 && InterlockedCompareExchange (&tp->destroy_thread, 0, 1) == 1)
1718                 result = (tp->nthreads > tp->min_threads);
1719         return result;
1720 }
1721
1722 static void
1723 async_invoke_thread (gpointer data)
1724 {
1725         MonoDomain *domain;
1726         MonoInternalThread *thread;
1727         MonoWSQ *wsq;
1728         ThreadPool *tp;
1729         gboolean must_die;
1730
1731         tp = data;
1732         wsq = NULL;
1733         if (!tp->is_io) {
1734                 wsq = mono_wsq_create ();
1735                 add_wsq (wsq);
1736         }
1737
1738         thread = mono_thread_internal_current ();
1739         if (tp_start_func)
1740                 tp_start_func (tp_hooks_user_data);
1741         data = NULL;
1742         for (;;) {
1743                 MonoAsyncResult *ar;
1744                 gboolean is_io_task;
1745                 int n_naps = 0;
1746
1747                 is_io_task = FALSE;
1748                 ar = (MonoAsyncResult *) data;
1749                 if (ar) {
1750                         InterlockedIncrement (&tp->busy_threads);
1751 #ifndef DISABLE_SOCKETS
1752                         is_io_task = (strcmp (((MonoObject *) data)->vtable->klass->name, "AsyncResult"));
1753                         if (is_io_task) {
1754                                 MonoSocketAsyncResult *state = (MonoSocketAsyncResult *) data;
1755                                 ar = state->ares;
1756                                 switch (state->operation) {
1757                                 case AIO_OP_RECEIVE:
1758                                         state->total = ICALL_RECV (state);
1759                                         break;
1760                                 case AIO_OP_SEND:
1761                                         state->total = ICALL_SEND (state);
1762                                         break;
1763                                 }
1764                         }
1765 #endif
1766                         /* worker threads invokes methods in different domains,
1767                          * so we need to set the right domain here */
1768                         domain = ((MonoObject *)ar)->vtable->domain;
1769                         g_assert (domain);
1770
1771                         if (mono_domain_is_unloading (domain) || mono_runtime_is_shutting_down ()) {
1772                                 threadpool_jobs_dec ((MonoObject *)ar);
1773                                 data = NULL;
1774                                 ar = NULL;
1775                                 InterlockedDecrement (&tp->busy_threads);
1776                         } else {
1777                                 mono_thread_push_appdomain_ref (domain);
1778                                 if (threadpool_jobs_dec ((MonoObject *)ar)) {
1779                                         data = NULL;
1780                                         ar = NULL;
1781                                         mono_thread_pop_appdomain_ref ();
1782                                         InterlockedDecrement (&tp->busy_threads);
1783                                         continue;
1784                                 }
1785
1786                                 if (mono_domain_set (domain, FALSE)) {
1787                                         /* ASyncCall *ac; */
1788
1789                                         if (tp_item_begin_func)
1790                                                 tp_item_begin_func (tp_item_user_data);
1791
1792                                         if (!is_io_task && ar->add_time > 0)
1793                                                 process_idle_times (tp, ar->add_time);
1794                                         /*FIXME: Do something with the exception returned? */
1795                                         mono_async_invoke (tp, ar);
1796                                         if (tp_item_end_func)
1797                                                 tp_item_end_func (tp_item_user_data);
1798                                         /*
1799                                         ac = (ASyncCall *) ar->object_data;
1800                                         if (ac->msg->exc != NULL)
1801                                                 mono_unhandled_exception (ac->msg->exc);
1802                                         */
1803                                         mono_domain_set (mono_get_root_domain (), TRUE);
1804                                 }
1805                                 mono_thread_pop_appdomain_ref ();
1806                                 InterlockedDecrement (&tp->busy_threads);
1807                                 /* If the callee changes the background status, set it back to TRUE */
1808                                 if (!mono_thread_test_state (thread , ThreadState_Background))
1809                                         ves_icall_System_Threading_Thread_SetState (thread, ThreadState_Background);
1810                         }
1811                 }
1812
1813                 ar = NULL;
1814                 data = NULL;
1815                 must_die = should_i_die (tp);
1816                 TP_DEBUG ("Trying to get a job");
1817                 if (!must_die && (tp->is_io || !mono_wsq_local_pop (&data)))
1818                         dequeue_or_steal (tp, &data);
1819                 TP_DEBUG ("Done trying to get a job %p", data);
1820
1821                 n_naps = 0;
1822                 while (!must_die && !data && n_naps < 4) {
1823                         gboolean res;
1824
1825                         TP_DEBUG ("Waiting");
1826                         InterlockedIncrement (&tp->waiting);
1827 #if defined(__OpenBSD__)
1828                         while ((res = mono_sem_wait (&tp->new_job, TRUE)) == -1) {// && errno == EINTR) {
1829 #else
1830                         while ((res = mono_sem_timedwait (&tp->new_job, 2000, TRUE)) == -1) {// && errno == EINTR) {
1831 #endif
1832                                 if (mono_runtime_is_shutting_down ())
1833                                         break;
1834                                 if (THREAD_WANTS_A_BREAK (thread))
1835                                         mono_thread_interruption_checkpoint ();
1836                         }
1837                         TP_DEBUG ("Done waiting");
1838                         InterlockedDecrement (&tp->waiting);
1839                         if (mono_runtime_is_shutting_down ())
1840                                 break;
1841                         must_die = should_i_die (tp);
1842                         dequeue_or_steal (tp, &data);
1843                         n_naps++;
1844                 }
1845
1846                 if (!data && tp->is_io && !mono_runtime_is_shutting_down ()) {
1847                         mono_wsq_local_pop (&data);
1848                         if (data && must_die) {
1849                                 InterlockedCompareExchange (&tp->destroy_thread, 1, 0);
1850                                 pulse_on_new_job (tp);
1851                         }
1852                 }
1853
1854                 if (!data) {
1855                         gint nt;
1856                         gboolean down;
1857                         while (1) {
1858                                 nt = tp->nthreads;
1859                                 down = mono_runtime_is_shutting_down ();
1860                                 if (!down && nt <= tp->min_threads)
1861                                         break;
1862                                 if (down || InterlockedCompareExchange (&tp->nthreads, nt - 1, nt) == nt) {
1863                                         mono_perfcounter_update_value (tp->pc_nthreads, TRUE, -1);
1864                                         TP_DEBUG ("DIE");
1865                                         if (!tp->is_io) {
1866                                                 remove_wsq (wsq);
1867                                                 while (mono_wsq_local_pop (&data)) {
1868                                                         threadpool_jobs_dec (data);
1869                                                         data = NULL;
1870                                                 }
1871                                                 mono_wsq_destroy (wsq);
1872                                         }
1873                                         if (tp_finish_func)
1874                                                 tp_finish_func (tp_hooks_user_data);
1875                                         return;
1876                                 }
1877                         }
1878                 }
1879         }
1880
1881         g_assert_not_reached ();
1882 }
1883
1884 void
1885 ves_icall_System_Threading_ThreadPool_GetAvailableThreads (gint *workerThreads, gint *completionPortThreads)
1886 {
1887         *workerThreads = async_tp.max_threads - async_tp.busy_threads;
1888         *completionPortThreads = async_io_tp.max_threads - async_io_tp.busy_threads;
1889 }
1890
1891 void
1892 ves_icall_System_Threading_ThreadPool_GetMaxThreads (gint *workerThreads, gint *completionPortThreads)
1893 {
1894         *workerThreads = async_tp.max_threads;
1895         *completionPortThreads = async_io_tp.max_threads;
1896 }
1897
1898 void
1899 ves_icall_System_Threading_ThreadPool_GetMinThreads (gint *workerThreads, gint *completionPortThreads)
1900 {
1901         *workerThreads = async_tp.min_threads;
1902         *completionPortThreads = async_io_tp.min_threads;
1903 }
1904
1905 MonoBoolean
1906 ves_icall_System_Threading_ThreadPool_SetMinThreads (gint workerThreads, gint completionPortThreads)
1907 {
1908         gint max_threads;
1909         gint max_io_threads;
1910
1911         max_threads = async_tp.max_threads;
1912         if (workerThreads <= 0 || workerThreads > max_threads)
1913                 return FALSE;
1914
1915         max_io_threads = async_io_tp.max_threads;
1916         if (completionPortThreads <= 0 || completionPortThreads > max_io_threads)
1917                 return FALSE;
1918
1919         InterlockedExchange (&async_tp.min_threads, workerThreads);
1920         InterlockedExchange (&async_io_tp.min_threads, completionPortThreads);
1921         if (workerThreads > async_tp.nthreads)
1922                 mono_thread_create_internal (mono_get_root_domain (), threadpool_start_idle_threads, &async_tp, TRUE);
1923         if (completionPortThreads > async_io_tp.nthreads)
1924                 mono_thread_create_internal (mono_get_root_domain (), threadpool_start_idle_threads, &async_io_tp, TRUE);
1925         return TRUE;
1926 }
1927
1928 MonoBoolean
1929 ves_icall_System_Threading_ThreadPool_SetMaxThreads (gint workerThreads, gint completionPortThreads)
1930 {
1931         gint min_threads;
1932         gint min_io_threads;
1933         gint cpu_count;
1934
1935         cpu_count = mono_cpu_count ();
1936         min_threads = async_tp.min_threads;
1937         if (workerThreads < min_threads || workerThreads < cpu_count)
1938                 return FALSE;
1939
1940         /* We don't really have the concept of completion ports. Do we care here? */
1941         min_io_threads = async_io_tp.min_threads;
1942         if (completionPortThreads < min_io_threads || completionPortThreads < cpu_count)
1943                 return FALSE;
1944
1945         InterlockedExchange (&async_tp.max_threads, workerThreads);
1946         InterlockedExchange (&async_io_tp.max_threads, completionPortThreads);
1947         return TRUE;
1948 }
1949
1950 /**
1951  * mono_install_threadpool_thread_hooks
1952  * @start_func: the function to be called right after a new threadpool thread is created. Can be NULL.
1953  * @finish_func: the function to be called right before a thredpool thread is exiting. Can be NULL.
1954  * @user_data: argument passed to @start_func and @finish_func.
1955  *
1956  * @start_fun will be called right after a threadpool thread is created and @finish_func right before a threadpool thread exits.
1957  * The calls will be made from the thread itself.
1958  */
1959 void
1960 mono_install_threadpool_thread_hooks (MonoThreadPoolFunc start_func, MonoThreadPoolFunc finish_func, gpointer user_data)
1961 {
1962         tp_start_func = start_func;
1963         tp_finish_func = finish_func;
1964         tp_hooks_user_data = user_data;
1965 }
1966
1967 /**
1968  * mono_install_threadpool_item_hooks
1969  * @begin_func: the function to be called before a threadpool work item processing starts.
1970  * @end_func: the function to be called after a threadpool work item is finished.
1971  * @user_data: argument passed to @begin_func and @end_func.
1972  *
1973  * The calls will be made from the thread itself and from the same AppDomain
1974  * where the work item was executed.
1975  *
1976  */
1977 void
1978 mono_install_threadpool_item_hooks (MonoThreadPoolItemFunc begin_func, MonoThreadPoolItemFunc end_func, gpointer user_data)
1979 {
1980         tp_item_begin_func = begin_func;
1981         tp_item_end_func = end_func;
1982         tp_item_user_data = user_data;
1983 }
1984