#ifdef PLATFORM_WIN32
#define WINVER 0x0500
#define _WIN32_WINNT 0x0500
-#define THREADS_PER_CPU 25
-#else
-#define THREADS_PER_CPU 50
#endif
+#define THREADS_PER_CPU 5 /* 20 + THREADS_PER_CPU * number of CPUs */
+
#include <mono/metadata/domain-internals.h>
#include <mono/metadata/tabledefs.h>
#include <mono/metadata/threads.h>
#undef EPOLL_DEBUG
/* maximum number of worker threads */
-static int mono_max_worker_threads = THREADS_PER_CPU;
-static int mono_min_worker_threads = 0;
-static int mono_io_max_worker_threads = THREADS_PER_CPU * 2;
+static int mono_max_worker_threads;
+static int mono_min_worker_threads;
+static int mono_io_max_worker_threads;
/* current number of worker threads */
static int mono_worker_threads = 0;
static HANDLE job_added;
static HANDLE io_job_added;
+/* Keep in sync with the System.MonoAsyncCall class which provides GC tracking */
typedef struct {
+ MonoObject object;
MonoMethodMessage *msg;
- HANDLE wait_event;
MonoMethod *cb_method;
MonoDelegate *cb_target;
MonoObject *state;
MonoObject *res;
MonoArray *out_args;
+ /* This is a HANDLE, we use guint64 so the managed object layout remains constant */
+ guint64 wait_event;
} ASyncCall;
static void async_invoke_thread (gpointer data);
static GList *async_call_queue = NULL;
static GList *async_io_queue = NULL;
+static MonoClass *async_call_klass;
static MonoClass *socket_async_call_klass;
#define INIT_POLLFD(a, b, c) {(a)->fd = b; (a)->events = c; (a)->revents = 0;}
MonoThread *thread;
thread = mono_thread_current ();
thread->threadpool_thread = TRUE;
- thread->state |= ThreadState_Background;
+ ves_icall_System_Threading_Thread_SetState (thread, ThreadState_Background);
for (;;) {
MonoSocketAsyncResult *state;
}
domain = ((MonoObject *)ar)->vtable->domain;
+ mono_thread_push_appdomain_ref (domain);
if (mono_domain_set (domain, FALSE)) {
ASyncCall *ac;
- mono_thread_push_appdomain_ref (domain);
mono_async_invoke (ar);
- ac = (ASyncCall *) ar->data;
+ ac = (ASyncCall *) ar->object_data;
+ /*
if (ac->msg->exc != NULL)
mono_unhandled_exception (ac->msg->exc);
- mono_thread_pop_appdomain_ref ();
+ */
+ mono_domain_set (mono_get_root_domain (), TRUE);
}
+ mono_thread_pop_appdomain_ref ();
InterlockedDecrement (&busy_io_worker_threads);
}
InterlockedIncrement (&busy_io_worker_threads);
InterlockedIncrement (&io_worker_threads);
domain = ((ares) ? ((MonoObject *) ares)->vtable->domain : mono_domain_get ());
- mono_thread_create (domain, async_invoke_io_thread, ares);
+ mono_thread_create (mono_get_root_domain (), async_invoke_io_thread, ares);
} else {
append_job (&io_queue_lock, &async_io_queue, ares);
ReleaseSemaphore (io_job_added, 1, NULL);
if (pfd->fd == -1)
continue;
- ret = mono_poll (pfds, 1, 0);
+ ret = mono_poll (pfd, 1, 0);
if (ret == -1 && errno == EBADF) {
pfd->revents |= MONO_POLLNVAL;
count++;
thread = mono_thread_current ();
thread->threadpool_thread = TRUE;
- thread->state |= ThreadState_Background;
+ ves_icall_System_Threading_Thread_SetState (thread, ThreadState_Background);
allocated = INITIAL_POLLFD_SIZE;
pfds = g_new0 (mono_pollfd, allocated);
epollfd = data->epollfd;
thread = mono_thread_current ();
thread->threadpool_thread = TRUE;
- thread->state |= ThreadState_Background;
+ ves_icall_System_Threading_Thread_SetState (thread, ThreadState_Background);
events = g_new0 (struct epoll_event, nevents);
while (1) {
g_assert (data->pipe [0] != INVALID_SOCKET);
closesocket (srv);
#endif
+ mono_io_max_worker_threads = mono_max_worker_threads / 2;
+ if (mono_io_max_worker_threads < 10)
+ mono_io_max_worker_threads = 10;
data->sock_to_state = g_hash_table_new (g_direct_hash, g_direct_equal);
GSList *list;
SocketIOData *data = &socket_io_data;
+#if defined(PLATFORM_MACOSX) || defined(PLATFORM_BSD6)
+ /* select() for connect() does not work well on the Mac. Bug #75436. */
+ /* Bug #77637 for the BSD 6 case */
+ if (state->operation == AIO_OP_CONNECT && state->blocking == TRUE) {
+ start_io_thread_or_queue (state);
+ return;
+ }
+#endif
WaitForSingleObject (data->new_sem, INFINITE);
if (data->newpfd == NULL)
data->newpfd = g_new0 (mono_pollfd, 1);
socket_io_add (MonoAsyncResult *ares, MonoSocketAsyncResult *state)
{
socket_io_init (&socket_io_data);
- state->ares = ares;
+ MONO_OBJECT_SETREF (state, ares, ares);
#ifdef HAVE_EPOLL
if (socket_io_data.epoll_disabled == FALSE) {
if (socket_io_add_epoll (state))
static void
mono_async_invoke (MonoAsyncResult *ares)
{
- ASyncCall *ac = (ASyncCall *)ares->data;
+ ASyncCall *ac = (ASyncCall *)ares->object_data;
+ MonoThread *thread = NULL;
+
+ if (ares->execution_context) {
+ /* use captured ExecutionContext (if available) */
+ thread = mono_thread_current ();
+ MONO_OBJECT_SETREF (ares, original_context, thread->execution_context);
+ MONO_OBJECT_SETREF (thread, execution_context, ares->execution_context);
+ } else {
+ ares->original_context = NULL;
+ }
ac->msg->exc = NULL;
ac->res = mono_message_invoke (ares->async_delegate, ac->msg,
MonoObject *exc = NULL;
void *pa = &ares;
mono_runtime_invoke (ac->cb_method, ac->cb_target, pa, &exc);
- if (!ac->msg->exc)
- ac->msg->exc = exc;
+ /* 'exc' will be the previous ac->msg->exc if not NULL and not
+ * catched. If catched, this will be set to NULL and the
+ * exception will not be printed. */
+ MONO_OBJECT_SETREF (ac->msg, exc, exc);
+ }
+
+ /* restore original thread execution context if flow isn't suppressed, i.e. non null */
+ if (ares->original_context) {
+ MONO_OBJECT_SETREF (thread, execution_context, ares->original_context);
+ ares->original_context = NULL;
}
/* notify listeners */
MONO_GC_REGISTER_ROOT (ares_htable);
InitializeCriticalSection (&socket_io_data.io_lock);
InitializeCriticalSection (&ares_lock);
- ares_htable = mono_g_hash_table_new (NULL, NULL);
+ ares_htable = mono_g_hash_table_new_type (NULL, NULL, MONO_HASH_KEY_VALUE_GC);
job_added = CreateSemaphore (NULL, 0, 0x7fffffff, NULL);
GetSystemInfo (&info);
- if (getenv ("MONO_THREADS_PER_CPU") != NULL) {
- threads_per_cpu = atoi (getenv ("MONO_THREADS_PER_CPU"));
+ if (g_getenv ("MONO_THREADS_PER_CPU") != NULL) {
+ threads_per_cpu = atoi (g_getenv ("MONO_THREADS_PER_CPU"));
if (threads_per_cpu <= 0)
threads_per_cpu = THREADS_PER_CPU;
}
- mono_max_worker_threads = threads_per_cpu * info.dwNumberOfProcessors;
+ mono_max_worker_threads = 20 + threads_per_cpu * info.dwNumberOfProcessors;
+
+ async_call_klass = mono_class_from_name (mono_defaults.corlib, "System", "MonoAsyncCall");
+ g_assert (async_call_klass);
}
MonoAsyncResult *
#ifdef HAVE_BOEHM_GC
ac = GC_MALLOC (sizeof (ASyncCall));
+#elif defined(HAVE_SGEN_GC)
+ ac = mono_object_new (mono_domain_get (), async_call_klass);
#else
/* We'll leak the event if creaated... */
ac = g_new0 (ASyncCall, 1);
#endif
ac->wait_event = NULL;
- ac->msg = msg;
- ac->state = state;
+ MONO_OBJECT_SETREF (ac, msg, msg);
+ MONO_OBJECT_SETREF (ac, state, state);
if (async_callback) {
ac->cb_method = mono_get_delegate_invoke (((MonoObject *)async_callback)->vtable->klass);
ac->cb_target = async_callback;
}
- ares = mono_async_result_new (domain, NULL, ac->state, ac);
- ares->async_delegate = target;
+ ares = mono_async_result_new (domain, NULL, ac->state, NULL, ac);
+ MONO_OBJECT_SETREF (ares, async_delegate, target);
EnterCriticalSection (&ares_lock);
mono_g_hash_table_insert (ares_htable, ares, ares);
InterlockedIncrement (&mono_worker_threads);
InterlockedIncrement (&busy_worker_threads);
domain = ((MonoObject *) ares)->vtable->domain;
- mono_thread_create (domain, async_invoke_thread, ares);
+ mono_thread_create (mono_get_root_domain (), async_invoke_thread, ares);
} else {
append_job (&mono_delegate_section, &async_call_queue, ares);
ReleaseSemaphore (job_added, 1, NULL);
}
ares->endinvoke_called = 1;
- ac = (ASyncCall *)ares->data;
+ ac = (ASyncCall *)ares->object_data;
g_assert (ac != NULL);
if (!ares->completed) {
if (ares->handle == NULL) {
ac->wait_event = CreateEvent (NULL, TRUE, FALSE, NULL);
- ares->handle = (MonoObject *) mono_wait_handle_new (mono_object_domain (ares), ac->wait_event);
+ MONO_OBJECT_SETREF (ares, handle, (MonoObject *) mono_wait_handle_new (mono_object_domain (ares), ac->wait_event));
}
mono_monitor_exit ((MonoObject *) ares);
WaitForSingleObjectEx (ac->wait_event, INFINITE, TRUE);
mono_monitor_exit ((MonoObject *) ares);
}
- *exc = ac->msg->exc;
+ *exc = ac->msg->exc; /* FIXME: GC add write barrier */
*out_args = ac->out_args;
return ac->res;
thread = mono_thread_current ();
thread->threadpool_thread = TRUE;
- thread->state |= ThreadState_Background;
+ ves_icall_System_Threading_Thread_SetState (thread, ThreadState_Background);
for (;;) {
MonoAsyncResult *ar;
/* worker threads invokes methods in different domains,
* so we need to set the right domain here */
domain = ((MonoObject *)ar)->vtable->domain;
+ mono_thread_push_appdomain_ref (domain);
if (mono_domain_set (domain, FALSE)) {
ASyncCall *ac;
- mono_thread_push_appdomain_ref (domain);
mono_async_invoke (ar);
- ac = (ASyncCall *) ar->data;
+ ac = (ASyncCall *) ar->object_data;
+ /*
if (ac->msg->exc != NULL)
mono_unhandled_exception (ac->msg->exc);
- mono_thread_pop_appdomain_ref ();
+ */
+ mono_domain_set (mono_get_root_domain (), TRUE);
}
+ mono_thread_pop_appdomain_ref ();
InterlockedDecrement (&busy_worker_threads);
}
data = dequeue_job (&mono_delegate_section, &async_call_queue);
-
+
if (!data) {
guint32 wr;
int timeout = 10000;