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