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
 692         inited = InterlockedCompareExchange (&data->inited, -1, -1);
 693         if (inited == 1)
 694                 return;
 695
 696         EnterCriticalSection (&data->io_lock);
 697         inited = InterlockedCompareExchange (&data->inited, -1, -1);
 698         if (inited == 1) {
 699                 LeaveCriticalSection (&data->io_lock);
 700                 return;
 701         }
 702
 703 #ifdef HAVE_EPOLL
 704         data->epoll_disabled = (g_getenv ("MONO_DISABLE_AIO") != NULL);
 705         if (FALSE == data->epoll_disabled) {
 706                 data->epollfd = epoll_create (256);
 707                 data->epoll_disabled = (data->epollfd == -1);
 708                 if (data->epoll_disabled && g_getenv ("MONO_DEBUG"))
 709                         g_message ("epoll_create() failed. Using plain poll().");
 710         } else {
 711                 data->epollfd = -1;
 712         }
 713 #else
 714         data->epoll_disabled = TRUE;
 715 #endif
 716
 717 #ifndef HOST_WIN32
 718         if (data->epoll_disabled) {
 719                 if (pipe (data->pipe) != 0) {
 720                         int err = errno;
 721                         perror ("mono");
 722                         g_assert (err);
 723                 }
 724         } else {
 725                 data->pipe [0] = -1;
 726                 data->pipe [1] = -1;
 727         }
 728 #else
 729         srv = socket (AF_INET, SOCK_STREAM, IPPROTO_TCP);
 730         g_assert (srv != INVALID_SOCKET);
 731         data->pipe [1] = socket (AF_INET, SOCK_STREAM, IPPROTO_TCP);
 732         g_assert (data->pipe [1] != INVALID_SOCKET);
 733
 734         server.sin_family = AF_INET;
 735         server.sin_addr.s_addr = inet_addr ("127.0.0.1");
 736         server.sin_port = 0;
 737         if (bind (srv, (SOCKADDR *) &server, sizeof (server))) {
 738                 g_print ("%d\n", WSAGetLastError ());
 739                 g_assert (1 != 0);
 740         }
 741
 742         len = sizeof (server);
 743         getsockname (srv, (SOCKADDR *) &server, &len);
 744         listen (srv, 1);
 745         mono_thread_create (mono_get_root_domain (), connect_hack, &server);
 746         len = sizeof (server);
 747         data->pipe [0] = accept (srv, (SOCKADDR *) &client, &len);
 748         g_assert (data->pipe [0] != INVALID_SOCKET);
 749         closesocket (srv);
 750 #endif
 751         data->sock_to_state = mono_g_hash_table_new_type (g_direct_hash, g_direct_equal, MONO_HASH_VALUE_GC);
 752
 753         if (data->epoll_disabled) {
 754                 data->new_sem = CreateSemaphore (NULL, 1, 1, NULL);
 755                 g_assert (data->new_sem != NULL);
 756         }
 757         if (data->epoll_disabled) {
 758                 mono_thread_create_internal (mono_get_root_domain (), socket_io_poll_main, data, TRUE);
 759         }
 760 #ifdef HAVE_EPOLL
 761         else {
 762                 mono_thread_create_internal (mono_get_root_domain (), socket_io_epoll_main, data, TRUE);
 763         }
 764 #endif
 765         InterlockedCompareExchange (&data->inited, 1, 0);
 766         LeaveCriticalSection (&data->io_lock);
 767 }
 768
 769 static void
 770 socket_io_add_poll (MonoSocketAsyncResult *state)
 771 {
 772         int events;
 773         char msg [1];
 774         MonoMList *list;
 775         SocketIOData *data = &socket_io_data;
 776         int w;
 777
 778 #if defined(PLATFORM_MACOSX) || defined(PLATFORM_BSD) || defined(HOST_WIN32) || defined(PLATFORM_SOLARIS)
 779         /* select() for connect() does not work well on the Mac. Bug #75436. */
 780         /* Bug #77637 for the BSD 6 case */
 781         /* Bug #78888 for the Windows case */
 782         if (state->operation == AIO_OP_CONNECT && state->blocking == TRUE) {
 783                 //FIXME: increment number of threads while this one is waiting?
 784                 threadpool_append_job (&async_io_tp, (MonoObject *) state);
 785                 return;
 786         }
 787 #endif
 788         WaitForSingleObject (data->new_sem, INFINITE);
 789         if (data->newpfd == NULL)
 790                 data->newpfd = g_new0 (mono_pollfd, 1);
 791
 792         EnterCriticalSection (&data->io_lock);
 793         /* FIXME: 64 bit issue: handle can be a pointer on windows? */
 794         list = mono_g_hash_table_lookup (data->sock_to_state, GINT_TO_POINTER (state->handle));
 795         if (list == NULL) {
 796                 list = mono_mlist_alloc ((MonoObject*)state);
 797         } else {
 798                 list = mono_mlist_append (list, (MonoObject*)state);
 799         }
 800
 801         events = get_events_from_list (list);
 802         INIT_POLLFD (data->newpfd, GPOINTER_TO_INT (state->handle), events);
 803         mono_g_hash_table_replace (data->sock_to_state, GINT_TO_POINTER (state->handle), list);
 804         LeaveCriticalSection (&data->io_lock);
 805         *msg = (char) state->operation;
 806 #ifndef HOST_WIN32
 807         w = write (data->pipe [1], msg, 1);
 808         w = w;
 809 #else
 810         send ((SOCKET) data->pipe [1], msg, 1, 0);
 811 #endif
 812 }
 813
 814 #ifdef HAVE_EPOLL
 815 static gboolean
 816 socket_io_add_epoll (MonoSocketAsyncResult *state)
 817 {
 818         MonoMList *list;
 819         SocketIOData *data = &socket_io_data;
 820         struct epoll_event event;
 821         int epoll_op, ievt;
 822         int fd;
 823
 824         memset (&event, 0, sizeof (struct epoll_event));
 825         fd = GPOINTER_TO_INT (state->handle);
 826         EnterCriticalSection (&data->io_lock);
 827         list = mono_g_hash_table_lookup (data->sock_to_state, GINT_TO_POINTER (fd));
 828         if (list == NULL) {
 829                 list = mono_mlist_alloc ((MonoObject*)state);
 830                 epoll_op = EPOLL_CTL_ADD;
 831         } else {
 832                 list = mono_mlist_append (list, (MonoObject*)state);
 833                 epoll_op = EPOLL_CTL_MOD;
 834         }
 835
 836         ievt = get_events_from_list (list);
 837         if ((ievt & MONO_POLLIN) != 0)
 838                 event.events |= EPOLLIN;
 839         if ((ievt & MONO_POLLOUT) != 0)
 840                 event.events |= EPOLLOUT;
 841
 842         mono_g_hash_table_replace (data->sock_to_state, state->handle, list);
 843         event.data.fd = fd;
 844         EPOLL_DEBUG ("%s %d with %d", epoll_op == EPOLL_CTL_ADD ? "ADD" : "MOD", fd, event.events);
 845         if (epoll_ctl (data->epollfd, epoll_op, fd, &event) == -1) {
 846                 int err = errno;
 847                 if (epoll_op == EPOLL_CTL_ADD && err == EEXIST) {
 848                         epoll_op = EPOLL_CTL_MOD;
 849                         if (epoll_ctl (data->epollfd, epoll_op, fd, &event) == -1) {
 850                                 g_message ("epoll_ctl(MOD): %d %s", err, g_strerror (err));
 851                         }
 852                 }
 853         }
 854         LeaveCriticalSection (&data->io_lock);
 855
 856         return TRUE;
 857 }
 858 #endif
 859
 860 static void
 861 socket_io_add (MonoAsyncResult *ares, MonoSocketAsyncResult *state)
 862 {
 863         socket_io_init (&socket_io_data);
 864         MONO_OBJECT_SETREF (state, ares, ares);
 865 #ifdef HAVE_EPOLL
 866         if (socket_io_data.epoll_disabled == FALSE) {
 867                 if (socket_io_add_epoll (state))
 868                         return;
 869         }
 870 #endif
 871         socket_io_add_poll (state);
 872 }
 873
 874 #ifndef DISABLE_SOCKETS
 875 static gboolean
 876 socket_io_filter (MonoObject *target, MonoObject *state)
 877 {
 878         gint op;
 879         MonoSocketAsyncResult *sock_res = (MonoSocketAsyncResult *) state;
 880         MonoClass *klass;
 881
 882         if (target == NULL || state == NULL)
 883                 return FALSE;
 884
 885         if (socket_async_call_klass == NULL) {
 886                 klass = target->vtable->klass;
 887                 /* Check if it's SocketAsyncCall in System.Net.Sockets
 888                  * FIXME: check the assembly is signed correctly for extra care
 889                  */
 890                 if (klass->name [0] == 'S' && strcmp (klass->name, "SocketAsyncCall") == 0
 891                                 && strcmp (mono_image_get_name (klass->image), "System") == 0
 892                                 && klass->nested_in && strcmp (klass->nested_in->name, "Socket") == 0)
 893                         socket_async_call_klass = klass;
 894         }
 895
 896         if (process_async_call_klass == NULL) {
 897                 klass = target->vtable->klass;
 898                 /* Check if it's AsyncReadHandler in System.Diagnostics.Process
 899                  * FIXME: check the assembly is signed correctly for extra care
 900                  */
 901                 if (klass->name [0] == 'A' && strcmp (klass->name, "AsyncReadHandler") == 0
 902                                 && strcmp (mono_image_get_name (klass->image), "System") == 0
 903                                 && klass->nested_in && strcmp (klass->nested_in->name, "Process") == 0)
 904                         process_async_call_klass = klass;
 905         }
 906         /* return both when socket_async_call_klass has not been seen yet and when
 907          * the object is not an instance of the class.
 908          */
 909         if (target->vtable->klass != socket_async_call_klass && target->vtable->klass != process_async_call_klass)
 910                 return FALSE;
 911
 912         op = sock_res->operation;
 913         if (op < AIO_OP_FIRST || op >= AIO_OP_LAST)
 914                 return FALSE;
 915
 916         return TRUE;
 917 }
 918 #endif /* !DISABLE_SOCKETS */
 919
 920 /* Returns the exception thrown when invoking, if any */
 921 static MonoObject *
 922 mono_async_invoke (ThreadPool *tp, MonoAsyncResult *ares)
 923 {
 924         ASyncCall *ac = (ASyncCall *)ares->object_data;
 925         MonoObject *res, *exc = NULL;
 926         MonoArray *out_args = NULL;
 927         HANDLE wait_event = NULL;
 928
 929         if (ares->execution_context) {
 930                 /* use captured ExecutionContext (if available) */
 931                 MONO_OBJECT_SETREF (ares, original_context, mono_thread_get_execution_context ());
 932                 mono_thread_set_execution_context (ares->execution_context);
 933         } else {
 934                 ares->original_context = NULL;
 935         }
 936
 937         if (ac == NULL) {
 938                 /* Fast path from ThreadPool.*QueueUserWorkItem */
 939                 void *pa = ares->async_state;
 940                 mono_runtime_delegate_invoke (ares->async_delegate, &pa, &exc);
 941         } else {
 942                 ac->msg->exc = NULL;
 943                 res = mono_message_invoke (ares->async_delegate, ac->msg, &exc, &out_args);
 944                 MONO_OBJECT_SETREF (ac, res, res);
 945                 MONO_OBJECT_SETREF (ac, msg->exc, exc);
 946                 MONO_OBJECT_SETREF (ac, out_args, out_args);
 947
 948                 mono_monitor_enter ((MonoObject *) ares);
 949                 ares->completed = 1;
 950                 if (ares->handle != NULL)
 951                         wait_event = mono_wait_handle_get_handle ((MonoWaitHandle *) ares->handle);
 952                 mono_monitor_exit ((MonoObject *) ares);
 953                 /* notify listeners */
 954                 if (wait_event != NULL)
 955                         SetEvent (wait_event);
 956
 957                 /* call async callback if cb_method != null*/
 958                 if (ac != NULL && ac->cb_method) {
 959                         MonoObject *exc = NULL;
 960                         void *pa = &ares;
 961                         mono_runtime_invoke (ac->cb_method, ac->cb_target, pa, &exc);
 962                         /* 'exc' will be the previous ac->msg->exc if not NULL and not
 963                          * catched. If catched, this will be set to NULL and the
 964                          * exception will not be printed. */
 965                         MONO_OBJECT_SETREF (ac->msg, exc, exc);
 966                 }
 967         }
 968
 969         /* restore original thread execution context if flow isn't suppressed, i.e. non null */
 970         if (ares->original_context) {
 971                 mono_thread_set_execution_context (ares->original_context);
 972                 ares->original_context = NULL;
 973         }
 974         return exc;
 975 }
 976
 977 static void
 978 threadpool_start_idle_threads (ThreadPool *tp)
 979 {
 980         int n;
 981
 982         do {
 983                 while (1) {
 984                         n = tp->nthreads;
 985                         if (n >= tp->min_threads)
 986                                 return;
 987                         if (InterlockedCompareExchange (&tp->nthreads, n + 1, n) == n)
 988                                 break;
 989                 }
 990                 mono_perfcounter_update_value (tp->pc_nthreads, TRUE, 1);
 991                 mono_thread_create_internal (mono_get_root_domain (), tp->async_invoke, tp, TRUE);
 992                 SleepEx (100, TRUE);
 993         } while (1);
 994 }
 995
 996 static void
 997 threadpool_init (ThreadPool *tp, int min_threads, int max_threads, void (*async_invoke) (gpointer))
 998 {
 999         memset (tp, 0, sizeof (ThreadPool));
1000         MONO_SEM_INIT (&tp->lock, 1);
1001         tp->min_threads = min_threads;
1002         tp->max_threads = max_threads;
1003         tp->async_invoke = async_invoke;
1004         MONO_SEM_INIT (&tp->new_job, 0);
1005 }
1006
1007 static void *
1008 init_perf_counter (const char *category, const char *counter)
1009 {
1010         MonoString *category_str;
1011         MonoString *counter_str;
1012         MonoString *machine;
1013         MonoDomain *root;
1014         MonoBoolean custom;
1015         int type;
1016
1017         if (category == NULL || counter == NULL)
1018                 return NULL;
1019         root = mono_get_root_domain ();
1020         category_str = mono_string_new (root, category);
1021         counter_str = mono_string_new (root, counter);
1022         machine = mono_string_new (root, ".");
1023         return mono_perfcounter_get_impl (category_str, counter_str, NULL, machine, &type, &custom);
1024 }
1025
1026 void
1027 mono_thread_pool_init ()
1028 {
1029         gint threads_per_cpu = 1;
1030         gint thread_count;
1031         gint cpu_count = mono_cpu_count ();
1032         int result;
1033
1034         if (tp_inited == 2)
1035                 return;
1036
1037         result = InterlockedCompareExchange (&tp_inited, 1, 0);
1038         if (result == 1) {
1039                 while (1) {
1040                         SleepEx (1, FALSE);
1041                         if (tp_inited == 2)
1042                                 return;
1043                 }
1044         }
1045
1046         MONO_GC_REGISTER_ROOT (async_tp.first);
1047         MONO_GC_REGISTER_ROOT (async_tp.last);
1048         MONO_GC_REGISTER_ROOT (async_tp.unused);
1049         MONO_GC_REGISTER_ROOT (async_io_tp.first);
1050         MONO_GC_REGISTER_ROOT (async_io_tp.unused);
1051         MONO_GC_REGISTER_ROOT (async_io_tp.last);
1052
1053         MONO_GC_REGISTER_ROOT (socket_io_data.sock_to_state);
1054         InitializeCriticalSection (&socket_io_data.io_lock);
1055         if (g_getenv ("MONO_THREADS_PER_CPU") != NULL) {
1056                 threads_per_cpu = atoi (g_getenv ("MONO_THREADS_PER_CPU"));
1057                 if (threads_per_cpu < 1)
1058                         threads_per_cpu = 1;
1059         }
1060
1061         thread_count = MIN (cpu_count * threads_per_cpu, MAX_POOL_THREADS);
1062         threadpool_init (&async_tp, thread_count, MAX_POOL_THREADS, async_invoke_thread);
1063         threadpool_init (&async_io_tp, cpu_count * 2, cpu_count * 4, async_invoke_thread);
1064         async_io_tp.is_io = TRUE;
1065
1066         async_call_klass = mono_class_from_name (mono_defaults.corlib, "System", "MonoAsyncCall");
1067         g_assert (async_call_klass);
1068
1069         InitializeCriticalSection (&wsqs_lock);
1070         wsqs = g_ptr_array_sized_new (MAX_POOL_THREADS);
1071         mono_wsq_init ();
1072
1073         async_tp.pc_nitems = init_perf_counter ("Mono Threadpool", "Work Items Added");
1074         g_assert (async_tp.pc_nitems);
1075
1076         async_io_tp.pc_nitems = init_perf_counter ("Mono Threadpool", "IO Work Items Added");
1077         g_assert (async_io_tp.pc_nitems);
1078
1079         async_tp.pc_nthreads = init_perf_counter ("Mono Threadpool", "# of Threads");
1080         g_assert (async_tp.pc_nthreads);
1081
1082         async_io_tp.pc_nthreads = init_perf_counter ("Mono Threadpool", "# of IO Threads");
1083         g_assert (async_io_tp.pc_nthreads);
1084         tp_inited = 2;
1085 }
1086
1087 MonoAsyncResult *
1088 mono_thread_pool_add (MonoObject *target, MonoMethodMessage *msg, MonoDelegate *async_callback,
1089                       MonoObject *state)
1090 {
1091         MonoDomain *domain = mono_domain_get ();
1092         MonoAsyncResult *ares;
1093         ASyncCall *ac;
1094
1095         ac = (ASyncCall*)mono_object_new (domain, async_call_klass);
1096         MONO_OBJECT_SETREF (ac, msg, msg);
1097         MONO_OBJECT_SETREF (ac, state, state);
1098
1099         if (async_callback) {
1100                 ac->cb_method = mono_get_delegate_invoke (((MonoObject *)async_callback)->vtable->klass);
1101                 MONO_OBJECT_SETREF (ac, cb_target, async_callback);
1102         }
1103
1104         ares = mono_async_result_new (domain, NULL, ac->state, NULL, (MonoObject*)ac);
1105         MONO_OBJECT_SETREF (ares, async_delegate, target);
1106
1107 #ifndef DISABLE_SOCKETS
1108         if (socket_io_filter (target, state)) {
1109                 socket_io_add (ares, (MonoSocketAsyncResult *) state);
1110                 return ares;
1111         }
1112 #endif
1113         threadpool_append_job (&async_tp, (MonoObject *) ares);
1114         return ares;
1115 }
1116
1117 MonoObject *
1118 mono_thread_pool_finish (MonoAsyncResult *ares, MonoArray **out_args, MonoObject **exc)
1119 {
1120         ASyncCall *ac;
1121         HANDLE wait_event;
1122
1123         *exc = NULL;
1124         *out_args = NULL;
1125
1126         /* check if already finished */
1127         mono_monitor_enter ((MonoObject *) ares);
1128
1129         if (ares->endinvoke_called) {
1130                 *exc = (MonoObject *) mono_get_exception_invalid_operation (NULL);
1131                 mono_monitor_exit ((MonoObject *) ares);
1132                 return NULL;
1133         }
1134
1135         ares->endinvoke_called = 1;
1136         /* wait until we are really finished */
1137         if (!ares->completed) {
1138                 if (ares->handle == NULL) {
1139                         wait_event = CreateEvent (NULL, TRUE, FALSE, NULL);
1140                         g_assert(wait_event != 0);
1141                         MONO_OBJECT_SETREF (ares, handle, (MonoObject *) mono_wait_handle_new (mono_object_domain (ares), wait_event));
1142                 } else {
1143                         wait_event = mono_wait_handle_get_handle ((MonoWaitHandle *) ares->handle);
1144                 }
1145                 mono_monitor_exit ((MonoObject *) ares);
1146                 WaitForSingleObjectEx (wait_event, INFINITE, TRUE);
1147         } else {
1148                 mono_monitor_exit ((MonoObject *) ares);
1149         }
1150
1151         ac = (ASyncCall *) ares->object_data;
1152         g_assert (ac != NULL);
1153         *exc = ac->msg->exc; /* FIXME: GC add write barrier */
1154         *out_args = ac->out_args;
1155
1156         return ac->res;
1157 }
1158
1159 static void
1160 threadpool_kill_idle_threads (ThreadPool *tp)
1161 {
1162         gint n;
1163
1164         n = (gint) InterlockedCompareExchange (&tp->max_threads, 0, -1);
1165         while (n) {
1166                 n--;
1167                 MONO_SEM_POST (&tp->new_job);
1168         }
1169 }
1170
1171 void
1172 mono_thread_pool_cleanup (void)
1173 {
1174         if (async_tp.pool_status == 1 && InterlockedCompareExchange (&async_tp.pool_status, 2, 1) == 2)
1175                 return;
1176
1177         InterlockedExchange (&async_io_tp.pool_status, 2);
1178         MONO_SEM_WAIT (&async_tp.lock);
1179         threadpool_free_queue (&async_tp);
1180         threadpool_kill_idle_threads (&async_tp);
1181         MONO_SEM_POST (&async_tp.lock);
1182
1183         socket_io_cleanup (&socket_io_data); /* Empty when DISABLE_SOCKETS is defined */
1184         MONO_SEM_WAIT (&async_io_tp.lock);
1185         threadpool_free_queue (&async_io_tp);
1186         threadpool_kill_idle_threads (&async_io_tp);
1187         MONO_SEM_POST (&async_io_tp.lock);
1188         MONO_SEM_DESTROY (&async_io_tp.new_job);
1189
1190         EnterCriticalSection (&wsqs_lock);
1191         mono_wsq_cleanup ();
1192         if (wsqs)
1193                 g_ptr_array_free (wsqs, TRUE);
1194         wsqs = NULL;
1195         LeaveCriticalSection (&wsqs_lock);
1196         MONO_SEM_DESTROY (&async_tp.new_job);
1197 }
1198
1199 /* Caller must enter &tp->lock */
1200 static MonoObject*
1201 dequeue_job_nolock (ThreadPool *tp)
1202 {
1203         MonoObject *ar;
1204         MonoArray *array;
1205         MonoMList *list;
1206
1207         list = tp->first;
1208         do {
1209                 if (mono_runtime_is_shutting_down ())
1210                         return NULL;
1211                 if (!list || tp->head == tp->tail)
1212                         return NULL;
1213
1214                 array = (MonoArray *) mono_mlist_get_data (list);
1215                 ar = mono_array_get (array, MonoObject *, tp->head % QUEUE_LENGTH);
1216                 mono_array_set (array, MonoObject *, tp->head % QUEUE_LENGTH, NULL);
1217                 tp->head++;
1218                 if ((tp->head % QUEUE_LENGTH) == 0) {
1219                         list = tp->first;
1220                         tp->first = mono_mlist_next (list);
1221                         if (tp->first == NULL)
1222                                 tp->last = NULL;
1223                         if (mono_mlist_length (tp->unused) < 20) {
1224                                 /* reuse this chunk */
1225                                 tp->unused = mono_mlist_set_next (list, tp->unused);
1226                         }
1227                         tp->head -= QUEUE_LENGTH;
1228                         tp->tail -= QUEUE_LENGTH;
1229                 }
1230                 list = tp->first;
1231         } while (ar == NULL);
1232         return ar;
1233 }
1234
1235 /* Call after entering &tp->lock */
1236 static gboolean
1237 threadpool_start_thread (ThreadPool *tp)
1238 {
1239         gint n;
1240
1241         while ((n = tp->nthreads) < tp->max_threads) {
1242                 if (InterlockedCompareExchange (&tp->nthreads, n + 1, n) == n) {
1243                         mono_perfcounter_update_value (tp->pc_nthreads, TRUE, 1);
1244                         mono_thread_create_internal (mono_get_root_domain (), tp->async_invoke, tp, TRUE);
1245                         return TRUE;
1246                 }
1247         }
1248
1249         return FALSE;
1250 }
1251
1252 static void
1253 pulse_on_new_job (ThreadPool *tp)
1254 {
1255         if (tp->waiting)
1256                 MONO_SEM_POST (&tp->new_job);
1257 }
1258
1259 void
1260 icall_append_job (MonoObject *ar)
1261 {
1262         threadpool_append_job (&async_tp, ar);
1263 }
1264
1265 static void
1266 threadpool_append_job (ThreadPool *tp, MonoObject *ar)
1267 {
1268         threadpool_append_jobs (tp, &ar, 1);
1269 }
1270
1271 static MonoMList *
1272 create_or_reuse_list (ThreadPool *tp)
1273 {
1274         MonoMList *list;
1275         MonoArray *array;
1276
1277         list = NULL;
1278         if (tp->unused) {
1279                 list = tp->unused;
1280                 tp->unused = mono_mlist_next (list);
1281                 mono_mlist_set_next (list, NULL);
1282                 //TP_DEBUG (tp->nodes_reused++);
1283         } else {
1284                 array = mono_array_new_cached (mono_get_root_domain (), mono_defaults.object_class, QUEUE_LENGTH);
1285                 list = mono_mlist_alloc ((MonoObject *) array);
1286                 //TP_DEBUG (tp->nodes_created++);
1287         }
1288         return list;
1289 }
1290
1291 static void
1292 threadpool_append_jobs (ThreadPool *tp, MonoObject **jobs, gint njobs)
1293 {
1294         MonoArray *array;
1295         MonoMList *list;
1296         MonoObject *ar;
1297         gint i;
1298         gboolean lock_taken = FALSE; /* We won't take the lock when the local queue is used */
1299
1300         if (mono_runtime_is_shutting_down ())
1301                 return;
1302
1303         if (tp->pool_status == 0 && InterlockedCompareExchange (&tp->pool_status, 1, 0) == 0)
1304                 mono_thread_create_internal (mono_get_root_domain (), threadpool_start_idle_threads, tp, TRUE);
1305
1306         for (i = 0; i < njobs; i++) {
1307                 ar = jobs [i];
1308                 if (ar == NULL)
1309                         continue; /* Might happen when cleaning domain jobs */
1310                 if (!tp->is_io) {
1311                         MonoAsyncResult *o = (MonoAsyncResult *) ar;
1312                         o->add_time = mono_100ns_ticks ();
1313                 }
1314                 threadpool_jobs_inc (ar);
1315                 mono_perfcounter_update_value (tp->pc_nitems, TRUE, 1);
1316                 if (!tp->is_io && mono_wsq_local_push (ar))
1317                         continue;
1318
1319                 if (!lock_taken) {
1320                         MONO_SEM_WAIT (&tp->lock);
1321                         lock_taken = TRUE;
1322                 }
1323                 if ((tp->tail % QUEUE_LENGTH) == 0) {
1324                         list = create_or_reuse_list (tp);
1325                         if (tp->last != NULL)
1326                                 mono_mlist_set_next (tp->last, list);
1327                         tp->last = list;
1328                         if (tp->first == NULL)
1329                                 tp->first = tp->last;
1330                 }
1331
1332                 array = (MonoArray *) mono_mlist_get_data (tp->last);
1333                 mono_array_setref (array, tp->tail % QUEUE_LENGTH, ar);
1334                 tp->tail++;
1335         }
1336         if (lock_taken)
1337                 MONO_SEM_POST (&tp->lock);
1338         for (i = 0; i < MIN(njobs, tp->max_threads); i++)
1339                 pulse_on_new_job (tp);
1340 }
1341
1342 static void
1343 threadpool_clear_queue (ThreadPool *tp, MonoDomain *domain)
1344 {
1345         MonoMList *current;
1346         MonoArray *array;
1347         MonoObject *obj;
1348         int domain_count;
1349         int i;
1350
1351         domain_count = 0;
1352         MONO_SEM_WAIT (&tp->lock);
1353         current = tp->first;
1354         while (current) {
1355                 array = (MonoArray *) mono_mlist_get_data (current);
1356                 for (i = 0; i < QUEUE_LENGTH; i++) {
1357                         obj = mono_array_get (array, MonoObject*, i);
1358                         if (obj != NULL && obj->vtable->domain == domain) {
1359                                 domain_count++;
1360                                 mono_array_setref (array, i, NULL);
1361                         }
1362                 }
1363                 current = mono_mlist_next (current);
1364         }
1365
1366         if (!domain_count) {
1367                 MONO_SEM_POST (&tp->lock);
1368                 return;
1369         }
1370
1371         current = tp->first;
1372         tp->first = NULL;
1373         tp->last = NULL;
1374         tp->head = 0;
1375         tp->tail = 0;
1376         MONO_SEM_POST (&tp->lock);
1377         /* Re-add everything but the nullified elements */
1378         while (current) {
1379                 array = (MonoArray *) mono_mlist_get_data (current);
1380                 threadpool_append_jobs (tp, mono_array_addr (array, MonoObject *, 0), QUEUE_LENGTH);
1381                 memset (mono_array_addr (array, MonoObject *, 0), 0, sizeof (MonoObject *) * QUEUE_LENGTH);
1382                 current = mono_mlist_next (current);
1383         }
1384 }
1385
1386 /*
1387  * Clean up the threadpool of all domain jobs.
1388  * Can only be called as part of the domain unloading process as
1389  * it will wait for all jobs to be visible to the interruption code.
1390  */
1391 gboolean
1392 mono_thread_pool_remove_domain_jobs (MonoDomain *domain, int timeout)
1393 {
1394         HANDLE sem_handle;
1395         int result = TRUE;
1396         guint32 start_time = 0;
1397
1398         g_assert (domain->state == MONO_APPDOMAIN_UNLOADING);
1399
1400         threadpool_clear_queue (&async_tp, domain);
1401         threadpool_clear_queue (&async_io_tp, domain);
1402
1403         /*
1404          * There might be some threads out that could be about to execute stuff from the given domain.
1405          * We avoid that by setting up a semaphore to be pulsed by the thread that reaches zero.
1406          */
1407         sem_handle = CreateSemaphore (NULL, 0, 1, NULL);
1408
1409         domain->cleanup_semaphore = sem_handle;
1410         /*
1411          * The memory barrier here is required to have global ordering between assigning to cleanup_semaphone
1412          * and reading threadpool_jobs.
1413          * Otherwise this thread could read a stale version of threadpool_jobs and wait forever.
1414          */
1415         mono_memory_write_barrier ();
1416
1417         if (domain->threadpool_jobs && timeout != -1)
1418                 start_time = mono_msec_ticks ();
1419         while (domain->threadpool_jobs) {
1420                 WaitForSingleObject (sem_handle, timeout);
1421                 if (timeout != -1 && (mono_msec_ticks () - start_time) > timeout) {
1422                         result = FALSE;
1423                         break;
1424                 }
1425         }
1426
1427         domain->cleanup_semaphore = NULL;
1428         CloseHandle (sem_handle);
1429         return result;
1430 }
1431
1432 static void
1433 threadpool_free_queue (ThreadPool *tp)
1434 {
1435         tp->head = tp->tail = 0;
1436         tp->first = NULL;
1437         tp->unused = NULL;
1438 }
1439
1440 gboolean
1441 mono_thread_pool_is_queue_array (MonoArray *o)
1442 {
1443         gpointer obj = o;
1444
1445         // FIXME: need some fix in sgen code.
1446         // There are roots at: async*tp.unused (MonoMList) and wsqs [n]->queue (MonoArray)
1447         return obj == async_tp.first || obj == async_io_tp.first;
1448 }
1449
1450 static void
1451 add_wsq (MonoWSQ *wsq)
1452 {
1453         int i;
1454
1455         if (wsq == NULL)
1456                 return;
1457
1458         EnterCriticalSection (&wsqs_lock);
1459         if (wsqs == NULL) {
1460                 LeaveCriticalSection (&wsqs_lock);
1461                 return;
1462         }
1463         for (i = 0; i < wsqs->len; i++) {
1464                 if (g_ptr_array_index (wsqs, i) == NULL) {
1465                         wsqs->pdata [i] = wsq;
1466                         LeaveCriticalSection (&wsqs_lock);
1467                         return;
1468                 }
1469         }
1470         g_ptr_array_add (wsqs, wsq);
1471         LeaveCriticalSection (&wsqs_lock);
1472 }
1473
1474 static void
1475 remove_wsq (MonoWSQ *wsq)
1476 {
1477         int i;
1478
1479         if (wsq == NULL)
1480                 return;
1481
1482         EnterCriticalSection (&wsqs_lock);
1483         if (wsqs == NULL) {
1484                 LeaveCriticalSection (&wsqs_lock);
1485                 return;
1486         }
1487         for (i = 0; i < wsqs->len; i++) {
1488                 if (g_ptr_array_index (wsqs, i) == wsq) {
1489                         wsqs->pdata [i] = NULL;
1490                         LeaveCriticalSection (&wsqs_lock);
1491                         return;
1492                 }
1493         }
1494         /* Should not happen */
1495         LeaveCriticalSection (&wsqs_lock);
1496 }
1497
1498 static void
1499 try_steal (gpointer *data, gboolean retry)
1500 {
1501         int i;
1502         int ms;
1503
1504         if (wsqs == NULL || data == NULL || *data != NULL)
1505                 return;
1506
1507         ms = 0;
1508         do {
1509                 if (mono_runtime_is_shutting_down ())
1510                         return;
1511                 for (i = 0; wsqs != NULL && i < wsqs->len; i++) {
1512                         if (mono_runtime_is_shutting_down ()) {
1513                                 return;
1514                         }
1515                         mono_wsq_try_steal (wsqs->pdata [i], data, ms);
1516                         if (*data != NULL) {
1517                                 return;
1518                         }
1519                 }
1520                 ms += 10;
1521         } while (retry && ms < 11);
1522 }
1523
1524 static gboolean
1525 dequeue_or_steal (ThreadPool *tp, gpointer *data)
1526 {
1527         if (mono_runtime_is_shutting_down ())
1528                 return FALSE;
1529         TP_DEBUG ("Dequeue");
1530         MONO_SEM_WAIT (&tp->lock);
1531         *data = dequeue_job_nolock (tp);
1532         MONO_SEM_POST (&tp->lock);
1533         if (!tp->is_io && !*data)
1534                 try_steal (data, FALSE);
1535         return (*data != NULL);
1536 }
1537
1538 static void
1539 process_idle_times (ThreadPool *tp, gint64 t)
1540 {
1541         gint64 ticks;
1542         gint64 avg;
1543         gboolean compute_avg;
1544         gint new_threads;
1545         gint64 per1;
1546
1547         if (tp->ignore_times)
1548                 return;
1549
1550         compute_avg = FALSE;
1551         ticks = mono_100ns_ticks ();
1552         t = ticks - t;
1553         SPIN_LOCK (tp->sp_lock);
1554         if (tp->ignore_times) {
1555                 SPIN_UNLOCK (tp->sp_lock);
1556                 return;
1557         }
1558         tp->time_sum += t;
1559         tp->n_sum++;
1560         if (tp->last_check == 0)
1561                 tp->last_check = ticks;
1562         else if ((ticks - tp->last_check) > 5000000) {
1563                 tp->ignore_times = 1;
1564                 compute_avg = TRUE;
1565         }
1566         SPIN_UNLOCK (tp->sp_lock);
1567
1568         if (!compute_avg)
1569                 return;
1570
1571         //printf ("Items: %d Time elapsed: %.3fs\n", tp->n_sum, (ticks - tp->last_check) / 10000.0);
1572         new_threads = 0;
1573         avg = tp->time_sum / tp->n_sum;
1574         if (tp->averages [1] == 0) {
1575                 tp->averages [1] = avg;
1576         } else {
1577                 per1 = ((100 * (ABS (avg - tp->averages [1]))) / tp->averages [1]);
1578                 if (per1 > 5) {
1579                         if (avg > tp->averages [1]) {
1580                                 if (tp->averages [1] < tp->averages [0]) {
1581                                         new_threads = -1;
1582                                 } else {
1583                                         new_threads = 1;
1584                                 }
1585                         } else if (avg < tp->averages [1] && tp->averages [1] < tp->averages [0]) {
1586                                 new_threads = 1;
1587                         }
1588                 } else {
1589                         int min, n;
1590                         min = tp->min_threads;
1591                         n = tp->nthreads;
1592                         if ((n - min) < min && tp->busy_threads == n)
1593                                 new_threads = 1;
1594                 }
1595                 /*
1596                 if (new_threads != 0) {
1597                         printf ("n: %d per1: %lld avg=%lld avg1=%lld avg0=%lld\n", new_threads, per1, avg, tp->averages [1], tp->averages [0]);
1598                 }
1599                 */
1600         }
1601
1602         tp->time_sum = 0;
1603         tp->n_sum = 0;
1604         tp->last_check = mono_100ns_ticks ();
1605
1606         tp->averages [0] = tp->averages [1];
1607         tp->averages [1] = avg;
1608         tp->ignore_times = 0;
1609
1610         if (tp->waiting == 0 && new_threads == 1) {
1611                 threadpool_start_thread (tp);
1612         } else if (new_threads == -1) {
1613                 if (InterlockedCompareExchange (&tp->destroy_thread, 1, 0) == 0)
1614                         pulse_on_new_job (tp);
1615         }
1616 }
1617
1618 static gboolean
1619 should_i_die (ThreadPool *tp)
1620 {
1621         gboolean result = FALSE;
1622         if (tp->destroy_thread == 1 && InterlockedCompareExchange (&tp->destroy_thread, 0, 1) == 1)
1623                 result = (tp->nthreads > tp->min_threads);
1624         return result;
1625 }
1626
1627 static void
1628 async_invoke_thread (gpointer data)
1629 {
1630         MonoDomain *domain;
1631         MonoInternalThread *thread;
1632         MonoWSQ *wsq;
1633         ThreadPool *tp;
1634         gboolean must_die;
1635
1636         tp = data;
1637         wsq = NULL;
1638         if (!tp->is_io) {
1639                 wsq = mono_wsq_create ();
1640                 add_wsq (wsq);
1641         }
1642
1643         thread = mono_thread_internal_current ();
1644         if (tp_start_func)
1645                 tp_start_func (tp_hooks_user_data);
1646         data = NULL;
1647         for (;;) {
1648                 MonoAsyncResult *ar;
1649                 gboolean is_io_task;
1650
1651                 is_io_task = FALSE;
1652                 ar = (MonoAsyncResult *) data;
1653                 if (ar) {
1654 #ifndef DISABLE_SOCKETS
1655                         is_io_task = (strcmp (((MonoObject *) data)->vtable->klass->name, "AsyncResult"));
1656                         if (is_io_task) {
1657                                 MonoSocketAsyncResult *state = (MonoSocketAsyncResult *) data;
1658                                 ar = state->ares;
1659                                 switch (state->operation) {
1660                                 case AIO_OP_RECEIVE:
1661                                         state->total = ICALL_RECV (state);
1662                                         break;
1663                                 case AIO_OP_SEND:
1664                                         state->total = ICALL_SEND (state);
1665                                         break;
1666                                 }
1667                         }
1668 #endif
1669
1670                         /* worker threads invokes methods in different domains,
1671                          * so we need to set the right domain here */
1672                         domain = ((MonoObject *)ar)->vtable->domain;
1673                         g_assert (domain);
1674
1675                         if (mono_domain_is_unloading (domain) || mono_runtime_is_shutting_down ()) {
1676                                 threadpool_jobs_dec ((MonoObject *)ar);
1677                                 data = NULL;
1678                                 ar = NULL;
1679                                 InterlockedDecrement (&tp->busy_threads);
1680                         } else {
1681                                 mono_thread_push_appdomain_ref (domain);
1682                                 if (threadpool_jobs_dec ((MonoObject *)ar)) {
1683                                         data = NULL;
1684                                         ar = NULL;
1685                                         mono_thread_pop_appdomain_ref ();
1686                                         continue;
1687                                 }
1688
1689                                 if (mono_domain_set (domain, FALSE)) {
1690                                         /* ASyncCall *ac; */
1691
1692                                         if (tp_item_begin_func)
1693                                                 tp_item_begin_func (tp_item_user_data);
1694
1695                                         if (!is_io_task)
1696                                                 process_idle_times (tp, ar->add_time);
1697                                         /*FIXME: Do something with the exception returned? */
1698                                         mono_async_invoke (tp, ar);
1699                                         if (tp_item_end_func)
1700                                                 tp_item_end_func (tp_item_user_data);
1701                                         /*
1702                                         ac = (ASyncCall *) ar->object_data;
1703                                         if (ac->msg->exc != NULL)
1704                                                 mono_unhandled_exception (ac->msg->exc);
1705                                         */
1706                                         mono_domain_set (mono_get_root_domain (), TRUE);
1707                                 }
1708                                 mono_thread_pop_appdomain_ref ();
1709                                 InterlockedDecrement (&tp->busy_threads);
1710                                 /* If the callee changes the background status, set it back to TRUE */
1711                                 if (!mono_thread_test_state (thread , ThreadState_Background))
1712                                         ves_icall_System_Threading_Thread_SetState (thread, ThreadState_Background);
1713                         }
1714                 }
1715
1716                 ar = NULL;
1717                 data = NULL;
1718                 must_die = should_i_die (tp);
1719                 TP_DEBUG ("Trying to get a job");
1720                 if (!must_die && (tp->is_io || !mono_wsq_local_pop (&data)))
1721                         dequeue_or_steal (tp, &data);
1722                 TP_DEBUG ("Done trying to get a job %p", data);
1723
1724                 while (!must_die && !data) {
1725                         gboolean res;
1726
1727                         TP_DEBUG ("Waiting");
1728                         InterlockedIncrement (&tp->waiting);
1729                         while ((res = MONO_SEM_WAIT (&tp->new_job)) && errno == EINTR) {
1730                                 if (mono_runtime_is_shutting_down ())
1731                                         break;
1732                                 if (THREAD_WANTS_A_BREAK (thread))
1733                                         mono_thread_interruption_checkpoint ();
1734                         }
1735                         TP_DEBUG ("Done waiting");
1736                         InterlockedDecrement (&tp->waiting);
1737                         //FIXME: res == 0 on windows when interrupted!
1738                         if (mono_runtime_is_shutting_down ())
1739                                 break;
1740                         must_die = should_i_die (tp);
1741                         dequeue_or_steal (tp, &data);
1742                 }
1743
1744                 if (!data) {
1745                         gint nt;
1746                         gboolean down;
1747                         while (1) {
1748                                 nt = tp->nthreads;
1749                                 down = mono_runtime_is_shutting_down ();
1750                                 if (!down && nt <= tp->min_threads)
1751                                         break;
1752                                 if (down || InterlockedCompareExchange (&tp->nthreads, nt - 1, nt) == nt) {
1753                                         mono_perfcounter_update_value (tp->pc_nthreads, TRUE, -1);
1754                                         TP_DEBUG ("DIE");
1755                                         if (!tp->is_io) {
1756                                                 remove_wsq (wsq);
1757                                                 mono_wsq_destroy (wsq);
1758                                         }
1759                                         if (tp_finish_func)
1760                                                 tp_finish_func (tp_hooks_user_data);
1761                                         return;
1762                                 }
1763                         }
1764                 }
1765
1766                 InterlockedIncrement (&tp->busy_threads);
1767         }
1768
1769         g_assert_not_reached ();
1770 }
1771
1772 void
1773 ves_icall_System_Threading_ThreadPool_GetAvailableThreads (gint *workerThreads, gint *completionPortThreads)
1774 {
1775         *workerThreads = 1024;//async_tp.max_threads - (gint) InterlockedCompareExchange (&async_tp.busy_threads, 0, -1);
1776         *completionPortThreads = async_io_tp.max_threads - (gint) InterlockedCompareExchange (&async_io_tp.busy_threads, 0, -1);
1777 }
1778
1779 void
1780 ves_icall_System_Threading_ThreadPool_GetMaxThreads (gint *workerThreads, gint *completionPortThreads)
1781 {
1782         *workerThreads = (gint) InterlockedCompareExchange (&async_tp.max_threads, 0, -1);
1783         *completionPortThreads = (gint) InterlockedCompareExchange (&async_io_tp.max_threads, 0, -1);
1784 }
1785
1786 void
1787 ves_icall_System_Threading_ThreadPool_GetMinThreads (gint *workerThreads, gint *completionPortThreads)
1788 {
1789         *workerThreads = (gint) InterlockedCompareExchange (&async_tp.min_threads, 0, -1);
1790         *completionPortThreads = (gint) InterlockedCompareExchange (&async_io_tp.min_threads, 0, -1);
1791 }
1792
1793 MonoBoolean
1794 ves_icall_System_Threading_ThreadPool_SetMinThreads (gint workerThreads, gint completionPortThreads)
1795 {
1796         gint max_threads;
1797         gint max_io_threads;
1798
1799         max_threads = InterlockedCompareExchange (&async_tp.max_threads, -1, -1);
1800         if (workerThreads <= 0 || workerThreads > max_threads)
1801                 return FALSE;
1802
1803         max_io_threads = InterlockedCompareExchange (&async_io_tp.max_threads, -1, -1);
1804         if (completionPortThreads <= 0 || completionPortThreads > max_io_threads)
1805                 return FALSE;
1806
1807         InterlockedExchange (&async_tp.min_threads, workerThreads);
1808         InterlockedExchange (&async_io_tp.min_threads, completionPortThreads);
1809         //FIXME: check the number of threads before starting this one
1810         mono_thread_create_internal (mono_get_root_domain (), threadpool_start_idle_threads, &async_tp, TRUE);
1811         return TRUE;
1812 }
1813
1814 MonoBoolean
1815 ves_icall_System_Threading_ThreadPool_SetMaxThreads (gint workerThreads, gint completionPortThreads)
1816 {
1817         gint min_threads;
1818         gint min_io_threads;
1819         gint cpu_count;
1820
1821         cpu_count = mono_cpu_count ();
1822         min_threads = InterlockedCompareExchange (&async_tp.min_threads, -1, -1);
1823         if (workerThreads < min_threads || workerThreads < cpu_count)
1824                 return FALSE;
1825
1826         /* We don't really have the concept of completion ports. Do we care here? */
1827         min_io_threads = InterlockedCompareExchange (&async_io_tp.min_threads, -1, -1);
1828         if (completionPortThreads < min_io_threads || completionPortThreads < cpu_count)
1829                 return FALSE;
1830
1831         InterlockedExchange (&async_tp.max_threads, workerThreads);
1832         InterlockedExchange (&async_io_tp.max_threads, completionPortThreads);
1833         return TRUE;
1834 }
1835
1836 /**
1837  * mono_install_threadpool_thread_hooks
1838  * @start_func: the function to be called right after a new threadpool thread is created. Can be NULL.
1839  * @finish_func: the function to be called right before a thredpool thread is exiting. Can be NULL.
1840  * @user_data: argument passed to @start_func and @finish_func.
1841  *
1842  * @start_fun will be called right after a threadpool thread is created and @finish_func right before a threadpool thread exits.
1843  * The calls will be made from the thread itself.
1844  */
1845 void
1846 mono_install_threadpool_thread_hooks (MonoThreadPoolFunc start_func, MonoThreadPoolFunc finish_func, gpointer user_data)
1847 {
1848         tp_start_func = start_func;
1849         tp_finish_func = finish_func;
1850         tp_hooks_user_data = user_data;
1851 }
1852
1853 /**
1854  * mono_install_threadpool_item_hooks
1855  * @begin_func: the function to be called before a threadpool work item processing starts.
1856  * @end_func: the function to be called after a threadpool work item is finished.
1857  * @user_data: argument passed to @begin_func and @end_func.
1858  *
1859  * The calls will be made from the thread itself and from the same AppDomain
1860  * where the work item was executed.
1861  *
1862  */
1863 void
1864 mono_install_threadpool_item_hooks (MonoThreadPoolItemFunc begin_func, MonoThreadPoolItemFunc end_func, gpointer user_data)
1865 {
1866         tp_item_begin_func = begin_func;
1867         tp_item_end_func = end_func;
1868         tp_item_user_data = user_data;
1869 }
1870