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