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