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