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