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