#include <mono/metadata/threadpool-ms.h>
#include <mono/metadata/threadpool-ms-io.h>
#include <mono/utils/atomic.h>
-#include <mono/utils/mono-poll.h>
#include <mono/utils/mono-threads.h>
+#include <mono/utils/mono-lazy-init.h>
+#include <mono/utils/mono-logger-internal.h>
typedef struct {
gboolean (*init) (gint wakeup_pipe_fd);
void (*cleanup) (void);
- void (*update_add) (gint fd, gint events, gboolean is_new);
- gint (*event_wait) (void);
- gint (*event_get_fd_max) (void);
- gint (*event_get_fd_at) (gint i, gint *events);
- void (*event_reset_fd_at) (gint i, gint events);
+ void (*register_fd) (gint fd, gint events, gboolean is_new);
+ void (*remove_fd) (gint fd);
+ gint (*event_wait) (void (*callback) (gint fd, gint events, gpointer user_data), gpointer user_data);
} ThreadPoolIOBackend;
+enum {
+ EVENT_IN = 1 << 0,
+ EVENT_OUT = 1 << 1,
+} ThreadPoolIOEvent;
+
#include "threadpool-ms-io-epoll.c"
#include "threadpool-ms-io-kqueue.c"
#include "threadpool-ms-io-poll.c"
+#define UPDATES_CAPACITY 128
+
/* Keep in sync with System.Net.Sockets.Socket.SocketOperation */
enum {
AIO_OP_FIRST,
AIO_OP_LAST
};
+typedef enum {
+ UPDATE_EMPTY = 0,
+ UPDATE_ADD,
+ UPDATE_REMOVE_SOCKET,
+ UPDATE_REMOVE_DOMAIN,
+} ThreadPoolIOUpdateType;
+
typedef struct {
+ gint fd;
MonoSocketAsyncResult *sockares;
-} ThreadPoolIOUpdate;
+} ThreadPoolIOUpdate_Add;
typedef struct {
- MonoGHashTable *states;
- mono_mutex_t states_lock;
+ gint fd;
+} ThreadPoolIOUpdate_RemoveSocket;
+
+typedef struct {
+ MonoDomain *domain;
+} ThreadPoolIOUpdate_RemoveDomain;
+typedef struct {
+ ThreadPoolIOUpdateType type;
+ union {
+ ThreadPoolIOUpdate_Add add;
+ ThreadPoolIOUpdate_RemoveSocket remove_socket;
+ ThreadPoolIOUpdate_RemoveDomain remove_domain;
+ } data;
+} ThreadPoolIOUpdate;
+
+typedef struct {
ThreadPoolIOBackend backend;
- ThreadPoolIOUpdate *updates;
- guint updates_size;
- guint updates_capacity;
+ ThreadPoolIOUpdate updates [UPDATES_CAPACITY];
+ gint updates_size;
mono_mutex_t updates_lock;
+ mono_cond_t updates_cond;
#if !defined(HOST_WIN32)
gint wakeup_pipes [2];
#endif
} ThreadPoolIO;
-static gint32 io_status = STATUS_NOT_INITIALIZED;
-static gint32 io_thread_status = STATUS_NOT_INITIALIZED;
+static mono_lazy_init_t io_status = MONO_LAZY_INIT_STATUS_NOT_INITIALIZED;
+
+static gboolean io_selector_running = FALSE;
static ThreadPoolIO* threadpool_io;
case AIO_OP_READPIPE:
case AIO_OP_ACCEPTRECEIVE:
case AIO_OP_RECEIVE_BUFFERS:
- return MONO_POLLIN;
+ return EVENT_IN;
case AIO_OP_SEND:
case AIO_OP_SEND_JUST_CALLBACK:
case AIO_OP_SENDTO:
case AIO_OP_CONNECT:
case AIO_OP_SEND_BUFFERS:
case AIO_OP_DISCONNECT:
- return MONO_POLLOUT;
+ return EVENT_OUT;
default:
g_assert_not_reached ();
}
static MonoSocketAsyncResult*
get_sockares_for_event (MonoMList **list, gint event)
{
- MonoSocketAsyncResult *state = NULL;
MonoMList *current;
g_assert (list);
for (current = *list; current; current = mono_mlist_next (current)) {
- state = (MonoSocketAsyncResult*) mono_mlist_get_data (current);
- if (get_events_from_sockares ((MonoSocketAsyncResult*) state) == event)
- break;
- state = NULL;
+ MonoSocketAsyncResult *ares = (MonoSocketAsyncResult*) mono_mlist_get_data (current);
+ if (get_events_from_sockares (ares) == event) {
+ *list = mono_mlist_remove_item (*list, current);
+ return ares;
+ }
}
- if (current)
- *list = mono_mlist_remove_item (*list, current);
-
- return state;
+ return NULL;
}
static gint
get_events (MonoMList *list)
{
- MonoSocketAsyncResult *ares;
+ MonoMList *current;
gint events = 0;
- for (; list; list = mono_mlist_next (list))
- if ((ares = (MonoSocketAsyncResult*) mono_mlist_get_data (list)))
- events |= get_events_from_sockares (ares);
+ for (current = list; current; current = mono_mlist_next (current))
+ events |= get_events_from_sockares ((MonoSocketAsyncResult*) mono_mlist_get_data (current));
return events;
}
}
}
+typedef struct {
+ MonoDomain *domain;
+ MonoGHashTable *states;
+} FilterSockaresForDomainData;
+
static void
-selector_thread (gpointer data)
+filter_sockares_for_domain (gpointer key, gpointer value, gpointer user_data)
{
- io_thread_status = STATUS_INITIALIZED;
+ FilterSockaresForDomainData *data;
+ MonoMList *list = value, *element;
+ MonoDomain *domain;
+ MonoGHashTable *states;
- for (;;) {
- guint i;
- guint max;
- gint ready = 0;
+ g_assert (user_data);
+ data = user_data;
+ domain = data->domain;
+ states = data->states;
- mono_mutex_lock (&threadpool_io->states_lock);
- mono_mutex_lock (&threadpool_io->updates_lock);
+ for (element = list; element; element = mono_mlist_next (element)) {
+ MonoSocketAsyncResult *sockares = (MonoSocketAsyncResult*) mono_mlist_get_data (element);
+ if (mono_object_domain (sockares) == domain)
+ mono_mlist_set_data (element, NULL);
+ }
- for (i = 0; i < threadpool_io->updates_size; ++i) {
- ThreadPoolIOUpdate *update;
- MonoMList *list;
+ /* we skip all the first elements which are NULL */
+ for (; list; list = mono_mlist_next (list)) {
+ if (mono_mlist_get_data (list))
+ break;
+ }
- update = &threadpool_io->updates [i];
+ if (list) {
+ g_assert (mono_mlist_get_data (list));
+
+ /* we delete all the NULL elements after the first one */
+ for (element = list; element;) {
+ MonoMList *next;
+ if (!(next = mono_mlist_next (element)))
+ break;
+ if (mono_mlist_get_data (next))
+ element = next;
+ else
+ mono_mlist_set_next (element, mono_mlist_next (next));
+ }
+ }
- g_assert (update->sockares);
+ mono_g_hash_table_replace (states, key, list);
+}
- list = mono_g_hash_table_lookup (threadpool_io->states, update->sockares->handle);
- list = mono_mlist_append (list, (MonoObject*) update->sockares);
- mono_g_hash_table_replace (threadpool_io->states, update->sockares->handle, list);
+static void
+wait_callback (gint fd, gint events, gpointer user_data)
+{
+ if (mono_runtime_is_shutting_down ())
+ return;
- threadpool_io->backend.update_add (GPOINTER_TO_INT (update->sockares->handle), get_events (list), mono_mlist_next (list) == NULL);
- }
- if (threadpool_io->updates_size > 0) {
- ThreadPoolIOUpdate *updates_old;
+ if (fd == threadpool_io->wakeup_pipes [0]) {
+ mono_trace (G_LOG_LEVEL_DEBUG, MONO_TRACE_IO_THREADPOOL, "io threadpool: wke");
+ selector_thread_wakeup_drain_pipes ();
+ } else {
+ MonoGHashTable *states;
+ MonoMList *list = NULL;
+ gpointer k;
- threadpool_io->updates_size = 0;
- threadpool_io->updates_capacity = 128;
+ g_assert (user_data);
+ states = user_data;
- updates_old = threadpool_io->updates;
+ mono_trace (G_LOG_LEVEL_DEBUG, MONO_TRACE_IO_THREADPOOL, "io threadpool: cal fd %3d, events = %2s | %2s",
+ fd, (events & EVENT_IN) ? "RD" : "..", (events & EVENT_OUT) ? "WR" : "..");
- threadpool_io->updates = mono_gc_alloc_fixed (sizeof (ThreadPoolIOUpdate) * threadpool_io->updates_capacity, MONO_GC_DESCRIPTOR_NULL);
- g_assert (threadpool_io->updates);
+ if (!mono_g_hash_table_lookup_extended (states, GINT_TO_POINTER (fd), &k, (gpointer*) &list))
+ g_error ("wait_callback: fd %d not found in states table", fd);
- mono_gc_free_fixed (updates_old);
+ if (list && (events & EVENT_IN) != 0) {
+ MonoSocketAsyncResult *sockares = get_sockares_for_event (&list, EVENT_IN);
+ if (sockares)
+ mono_threadpool_ms_enqueue_work_item (((MonoObject*) sockares)->vtable->domain, (MonoObject*) sockares);
+ }
+ if (list && (events & EVENT_OUT) != 0) {
+ MonoSocketAsyncResult *sockares = get_sockares_for_event (&list, EVENT_OUT);
+ if (sockares)
+ mono_threadpool_ms_enqueue_work_item (((MonoObject*) sockares)->vtable->domain, (MonoObject*) sockares);
}
- mono_mutex_unlock (&threadpool_io->updates_lock);
- mono_mutex_unlock (&threadpool_io->states_lock);
+ mono_g_hash_table_replace (states, GINT_TO_POINTER (fd), list);
- mono_gc_set_skip_thread (TRUE);
+ events = get_events (list);
- ready = threadpool_io->backend.event_wait ();
+ mono_trace (G_LOG_LEVEL_DEBUG, MONO_TRACE_IO_THREADPOOL, "io threadpool: res fd %3d, events = %2s | %2s",
+ fd, (events & EVENT_IN) ? "RD" : "..", (events & EVENT_OUT) ? "WR" : "..");
- mono_gc_set_skip_thread (FALSE);
+ threadpool_io->backend.register_fd (fd, events, FALSE);
+ }
+}
- if (ready == -1 || mono_runtime_is_shutting_down ())
- break;
+static void
+selector_thread (gpointer data)
+{
+ MonoGHashTable *states;
- max = threadpool_io->backend.event_get_fd_max ();
+ io_selector_running = TRUE;
- mono_mutex_lock (&threadpool_io->states_lock);
+ if (mono_runtime_is_shutting_down ()) {
+ io_selector_running = FALSE;
+ return;
+ }
- for (i = 0; i < max && ready > 0; ++i) {
- gint events;
- gint fd = threadpool_io->backend.event_get_fd_at (i, &events);
+ states = mono_g_hash_table_new_type (g_direct_hash, g_direct_equal, MONO_HASH_VALUE_GC);
- if (fd == -1)
- continue;
+ for (;;) {
+ gint i, j;
+ gint res;
- if (fd == threadpool_io->wakeup_pipes [0]) {
- selector_thread_wakeup_drain_pipes ();
- } else {
- MonoMList *list = mono_g_hash_table_lookup (threadpool_io->states, GINT_TO_POINTER (fd));
+ mono_mutex_lock (&threadpool_io->updates_lock);
- if (list && (events & MONO_POLLIN) != 0) {
- MonoSocketAsyncResult *sockares = get_sockares_for_event (&list, MONO_POLLIN);
- if (sockares)
- mono_threadpool_ms_enqueue_work_item (((MonoObject*) sockares)->vtable->domain, (MonoObject*) sockares);
- }
- if (list && (events & MONO_POLLOUT) != 0) {
- MonoSocketAsyncResult *sockares = get_sockares_for_event (&list, MONO_POLLOUT);
- if (sockares)
- mono_threadpool_ms_enqueue_work_item (((MonoObject*) sockares)->vtable->domain, (MonoObject*) sockares);
+ for (i = 0; i < threadpool_io->updates_size; ++i) {
+ ThreadPoolIOUpdate *update = &threadpool_io->updates [i];
+
+ switch (update->type) {
+ case UPDATE_EMPTY:
+ break;
+ case UPDATE_ADD: {
+ gint fd;
+ gint events;
+ gpointer k;
+ gboolean exists;
+ MonoMList *list = NULL;
+ MonoSocketAsyncResult *sockares;
+
+ fd = update->data.add.fd;
+ g_assert (fd >= 0);
+
+ sockares = update->data.add.sockares;
+ g_assert (sockares);
+
+ exists = mono_g_hash_table_lookup_extended (states, GINT_TO_POINTER (fd), &k, (gpointer*) &list);
+ list = mono_mlist_append (list, (MonoObject*) sockares);
+ mono_g_hash_table_replace (states, sockares->handle, list);
+
+ events = get_events (list);
+
+ mono_trace (G_LOG_LEVEL_DEBUG, MONO_TRACE_IO_THREADPOOL, "io threadpool: %3s fd %3d, events = %2s | %2s",
+ exists ? "mod" : "add", fd, (events & EVENT_IN) ? "RD" : "..", (events & EVENT_OUT) ? "WR" : "..");
+
+ threadpool_io->backend.register_fd (fd, events, !exists);
+
+ break;
+ }
+ case UPDATE_REMOVE_SOCKET: {
+ gint fd;
+ gpointer k;
+ MonoMList *list = NULL;
+
+ fd = update->data.remove_socket.fd;
+ g_assert (fd >= 0);
+
+ if (mono_g_hash_table_lookup_extended (states, GINT_TO_POINTER (fd), &k, (gpointer*) &list)) {
+ mono_g_hash_table_remove (states, GINT_TO_POINTER (fd));
+
+ for (j = i + 1; j < threadpool_io->updates_size; ++j) {
+ ThreadPoolIOUpdate *update = &threadpool_io->updates [j];
+ if (update->type == UPDATE_ADD && update->data.add.fd == fd)
+ memset (update, 0, sizeof (ThreadPoolIOUpdate));
+ }
+
+ for (; list; list = mono_mlist_remove_item (list, list))
+ mono_threadpool_ms_enqueue_work_item (mono_object_domain (mono_mlist_get_data (list)), mono_mlist_get_data (list));
+
+ mono_trace (G_LOG_LEVEL_DEBUG, MONO_TRACE_IO_THREADPOOL, "io threadpool: del fd %3d", fd);
+ threadpool_io->backend.remove_fd (fd);
}
- if (!list)
- mono_g_hash_table_remove (threadpool_io->states, GINT_TO_POINTER (fd));
- else
- mono_g_hash_table_replace (threadpool_io->states, GINT_TO_POINTER (fd), list);
+ break;
+ }
+ case UPDATE_REMOVE_DOMAIN: {
+ MonoDomain *domain;
+
+ domain = update->data.remove_domain.domain;
+ g_assert (domain);
+
+ FilterSockaresForDomainData user_data = { .domain = domain, .states = states };
+ mono_g_hash_table_foreach (states, filter_sockares_for_domain, &user_data);
- threadpool_io->backend.event_reset_fd_at (i, get_events (list));
+ for (j = i + 1; j < threadpool_io->updates_size; ++j) {
+ ThreadPoolIOUpdate *update = &threadpool_io->updates [j];
+ if (update->type == UPDATE_ADD && mono_object_domain (update->data.add.sockares) == domain)
+ memset (update, 0, sizeof (ThreadPoolIOUpdate));
+ }
+
+ break;
}
+ default:
+ g_assert_not_reached ();
+ }
+ }
- ready -= 1;
+ mono_cond_broadcast (&threadpool_io->updates_cond);
+
+ if (threadpool_io->updates_size > 0) {
+ threadpool_io->updates_size = 0;
+ memset (&threadpool_io->updates, 0, UPDATES_CAPACITY * sizeof (ThreadPoolIOUpdate));
}
- mono_mutex_unlock (&threadpool_io->states_lock);
+ mono_mutex_unlock (&threadpool_io->updates_lock);
+
+ mono_trace (G_LOG_LEVEL_DEBUG, MONO_TRACE_IO_THREADPOOL, "io threadpool: wai");
+
+ res = threadpool_io->backend.event_wait (wait_callback, states);
+
+ if (res == -1 || mono_runtime_is_shutting_down ())
+ break;
}
- io_thread_status = STATUS_CLEANED_UP;
+ mono_g_hash_table_destroy (states);
+
+ io_selector_running = FALSE;
+}
+
+/* Locking: threadpool_io->updates_lock must be held */
+static ThreadPoolIOUpdate*
+update_get_new (void)
+{
+ ThreadPoolIOUpdate *update = NULL;
+ g_assert (threadpool_io->updates_size <= UPDATES_CAPACITY);
+
+ while (threadpool_io->updates_size == UPDATES_CAPACITY) {
+ /* we wait for updates to be applied in the selector_thread and we loop
+ * as long as none are available. if it happends too much, then we need
+ * to increase UPDATES_CAPACITY */
+ mono_cond_wait (&threadpool_io->updates_cond, &threadpool_io->updates_lock);
+ }
+
+ g_assert (threadpool_io->updates_size < UPDATES_CAPACITY);
+
+ update = &threadpool_io->updates [threadpool_io->updates_size ++];
+
+ return update;
}
static void
}
static void
-ensure_initialized (void)
+initialize (void)
{
- if (io_status >= STATUS_INITIALIZED)
- return;
- if (io_status == STATUS_INITIALIZING || InterlockedCompareExchange (&io_status, STATUS_INITIALIZING, STATUS_NOT_INITIALIZED) != STATUS_NOT_INITIALIZED) {
- while (io_status == STATUS_INITIALIZING)
- mono_thread_info_yield ();
- g_assert (io_status >= STATUS_INITIALIZED);
- return;
- }
-
g_assert (!threadpool_io);
threadpool_io = g_new0 (ThreadPoolIO, 1);
g_assert (threadpool_io);
- threadpool_io->states = mono_g_hash_table_new_type (g_direct_hash, g_direct_equal, MONO_HASH_VALUE_GC);
- MONO_GC_REGISTER_ROOT_FIXED (threadpool_io->states);
- mono_mutex_init (&threadpool_io->states_lock);
+ mono_mutex_init_recursive (&threadpool_io->updates_lock);
+ mono_cond_init (&threadpool_io->updates_cond, NULL);
+ mono_gc_register_root ((void*)&threadpool_io->updates [0], sizeof (threadpool_io->updates), MONO_GC_DESCRIPTOR_NULL);
- threadpool_io->updates = NULL;
threadpool_io->updates_size = 0;
- threadpool_io->updates_capacity = 0;
- mono_mutex_init (&threadpool_io->updates_lock);
+ threadpool_io->backend = backend_poll;
+ if (g_getenv ("MONO_ENABLE_AIO") != NULL) {
#if defined(HAVE_EPOLL)
- threadpool_io->backend = backend_epoll;
+ threadpool_io->backend = backend_epoll;
#elif defined(HAVE_KQUEUE)
- threadpool_io->backend = backend_kqueue;
-#else
- threadpool_io->backend = backend_poll;
+ threadpool_io->backend = backend_kqueue;
#endif
- if (g_getenv ("MONO_DISABLE_AIO") != NULL)
- threadpool_io->backend = backend_poll;
+ }
wakeup_pipes_init ();
if (!threadpool_io->backend.init (threadpool_io->wakeup_pipes [0]))
- g_error ("ensure_initialized: backend->init () failed");
+ g_error ("initialize: backend->init () failed");
if (!mono_thread_create_internal (mono_get_root_domain (), selector_thread, NULL, TRUE, SMALL_STACK))
- g_error ("ensure_initialized: mono_thread_create_internal () failed");
-
- io_thread_status = STATUS_INITIALIZING;
- mono_memory_write_barrier ();
-
- io_status = STATUS_INITIALIZED;
+ g_error ("initialize: mono_thread_create_internal () failed");
}
static void
-ensure_cleanedup (void)
+cleanup (void)
{
- if (io_status == STATUS_NOT_INITIALIZED && InterlockedCompareExchange (&io_status, STATUS_CLEANED_UP, STATUS_NOT_INITIALIZED) == STATUS_NOT_INITIALIZED)
- return;
- if (io_status == STATUS_INITIALIZING) {
- while (io_status == STATUS_INITIALIZING)
- mono_thread_info_yield ();
- }
- if (io_status == STATUS_CLEANED_UP)
- return;
- if (io_status == STATUS_CLEANING_UP || InterlockedCompareExchange (&io_status, STATUS_CLEANING_UP, STATUS_INITIALIZED) != STATUS_INITIALIZED) {
- while (io_status == STATUS_CLEANING_UP)
- mono_thread_info_yield ();
- g_assert (io_status == STATUS_CLEANED_UP);
- return;
- }
-
/* we make the assumption along the code that we are
* cleaning up only if the runtime is shutting down */
g_assert (mono_runtime_is_shutting_down ());
selector_thread_wakeup ();
- while (io_thread_status != STATUS_CLEANED_UP)
+ while (io_selector_running)
g_usleep (1000);
- MONO_GC_UNREGISTER_ROOT (threadpool_io->states);
- mono_g_hash_table_destroy (threadpool_io->states);
- mono_mutex_destroy (&threadpool_io->states_lock);
-
- if (threadpool_io->updates)
- mono_gc_free_fixed (threadpool_io->updates);
mono_mutex_destroy (&threadpool_io->updates_lock);
+ mono_cond_destroy (&threadpool_io->updates_cond);
threadpool_io->backend.cleanup ();
g_free (threadpool_io);
threadpool_io = NULL;
g_assert (!threadpool_io);
-
- io_status = STATUS_CLEANED_UP;
}
static gboolean
MonoClass *socket_async_callback_class = NULL;
socket_async_callback_class = mono_class_from_name (system_image, "System.Net.Sockets", "SocketAsyncCallback");
- g_assert (socket_async_callback_class);
return class == socket_async_callback_class;
}
MonoClass *async_read_handler_class = NULL;
async_read_handler_class = mono_class_from_name (system_image, "System.Diagnostics", "Process/AsyncReadHandler");
- g_assert (async_read_handler_class);
return class == async_read_handler_class;
}
void
mono_threadpool_ms_io_cleanup (void)
{
- ensure_cleanedup ();
+ mono_lazy_cleanup (&io_status, cleanup);
}
MonoAsyncResult *
if (mono_runtime_is_shutting_down ())
return NULL;
+ if (mono_domain_is_unloading (mono_object_domain (sockares)))
+ return NULL;
- ensure_initialized ();
+ mono_lazy_initialize (&io_status, initialize);
MONO_OBJECT_SETREF (sockares, ares, ares);
mono_mutex_lock (&threadpool_io->updates_lock);
- threadpool_io->updates_size += 1;
- if (threadpool_io->updates_size > threadpool_io->updates_capacity) {
- ThreadPoolIOUpdate *updates_new, *updates_old;
- gint updates_new_capacity, updates_old_capacity;
-
- updates_old_capacity = threadpool_io->updates_capacity;
- updates_new_capacity = updates_old_capacity + 128;
-
- updates_old = threadpool_io->updates;
- updates_new = mono_gc_alloc_fixed (sizeof (ThreadPoolIOUpdate) * updates_new_capacity, MONO_GC_DESCRIPTOR_NULL);
- g_assert (updates_new);
+ update = update_get_new ();
+ update->type = UPDATE_ADD;
+ update->data.add.fd = GPOINTER_TO_INT (sockares->handle);
+ update->data.add.sockares = sockares;
+ mono_memory_barrier (); /* Ensure this is safely published before we wake up the selector */
- if (updates_old)
- memcpy (updates_new, updates_old, sizeof (ThreadPoolIOUpdate) * updates_old_capacity);
-
- threadpool_io->updates = updates_new;
- threadpool_io->updates_capacity = updates_new_capacity;
-
- if (updates_old)
- mono_gc_free_fixed (updates_old);
- }
-
- update = &threadpool_io->updates [threadpool_io->updates_size - 1];
- update->sockares = sockares;
+ selector_thread_wakeup ();
mono_mutex_unlock (&threadpool_io->updates_lock);
- selector_thread_wakeup ();
-
return ares;
}
void
mono_threadpool_ms_io_remove_socket (int fd)
{
- MonoMList *list;
- gint i;
+ ThreadPoolIOUpdate *update;
- if (io_status != STATUS_INITIALIZED)
+ if (!mono_lazy_is_initialized (&io_status))
return;
- mono_mutex_lock (&threadpool_io->states_lock);
mono_mutex_lock (&threadpool_io->updates_lock);
- g_assert (threadpool_io->states);
-
- list = mono_g_hash_table_lookup (threadpool_io->states, GINT_TO_POINTER (fd));
- if (list)
- mono_g_hash_table_remove (threadpool_io->states, GINT_TO_POINTER (fd));
+ update = update_get_new ();
+ update->type = UPDATE_REMOVE_SOCKET;
+ update->data.add.fd = fd;
+ mono_memory_barrier (); /* Ensure this is safely published before we wake up the selector */
- for (i = 0; i < threadpool_io->updates_size; ++i) {
- ThreadPoolIOUpdate *update = &threadpool_io->updates [i];
-
- g_assert (update->sockares);
-
- if (GPOINTER_TO_INT (update->sockares->handle) == fd) {
- if (i < threadpool_io->updates_size - 1)
- memmove (threadpool_io->updates + i, threadpool_io->updates + i + 1, sizeof (ThreadPoolIOUpdate) * threadpool_io->updates_size - i - 1);
- memset (threadpool_io->updates + threadpool_io->updates_size - 1, 0, sizeof (ThreadPoolIOUpdate));
+ selector_thread_wakeup ();
- threadpool_io->updates_size -= 1;
- }
- }
+ mono_cond_wait (&threadpool_io->updates_cond, &threadpool_io->updates_lock);
mono_mutex_unlock (&threadpool_io->updates_lock);
- mono_mutex_unlock (&threadpool_io->states_lock);
-
- for (; list; list = mono_mlist_remove_item (list, list)) {
- MonoSocketAsyncResult *sockares = (MonoSocketAsyncResult*) mono_mlist_get_data (list);
-
- if (!sockares)
- continue;
-
- switch (sockares->operation) {
- case AIO_OP_RECEIVE:
- sockares->operation = AIO_OP_RECV_JUST_CALLBACK;
- break;
- case AIO_OP_SEND:
- sockares->operation = AIO_OP_SEND_JUST_CALLBACK;
- break;
- }
-
- mono_threadpool_ms_enqueue_work_item (((MonoObject*) sockares)->vtable->domain, (MonoObject*) sockares);
- }
-}
-
-static gboolean
-remove_sockstate_for_domain (gpointer key, gpointer value, gpointer user_data)
-{
- MonoMList *list;
- gboolean remove = FALSE;
-
- for (list = value; list; list = mono_mlist_next (list)) {
- MonoObject *data = mono_mlist_get_data (list);
- if (mono_object_domain (data) == user_data) {
- remove = TRUE;
- mono_mlist_set_data (list, NULL);
- }
- }
-
- //FIXME is there some sort of additional unregistration we need to perform here?
- return remove;
}
void
mono_threadpool_ms_io_remove_domain_jobs (MonoDomain *domain)
{
- gint i;
+ ThreadPoolIOUpdate *update;
- if (io_status != STATUS_INITIALIZED)
+ if (!mono_lazy_is_initialized (&io_status))
return;
- mono_mutex_lock (&threadpool_io->states_lock);
mono_mutex_lock (&threadpool_io->updates_lock);
- mono_g_hash_table_foreach_remove (threadpool_io->states, remove_sockstate_for_domain, domain);
-
- for (i = 0; i < threadpool_io->updates_size; ++i) {
- ThreadPoolIOUpdate *update = &threadpool_io->updates [i];
-
- g_assert (update->sockares);
+ update = update_get_new ();
+ update->type = UPDATE_REMOVE_DOMAIN;
+ update->data.remove_domain.domain = domain;
+ mono_memory_barrier (); /* Ensure this is safely published before we wake up the selector */
- if (mono_object_domain (update->sockares) == domain) {
- if (i < threadpool_io->updates_size - 1)
- memmove (threadpool_io->updates + i, threadpool_io->updates + i + 1, sizeof (ThreadPoolIOUpdate) * threadpool_io->updates_size - i - 1);
- memset (threadpool_io->updates + threadpool_io->updates_size - 1, 0, sizeof (ThreadPoolIOUpdate));
+ selector_thread_wakeup ();
- threadpool_io->updates_size -= 1;
- }
- }
+ mono_cond_wait (&threadpool_io->updates_cond, &threadpool_io->updates_lock);
mono_mutex_unlock (&threadpool_io->updates_lock);
- mono_mutex_unlock (&threadpool_io->states_lock);
}
void
g_assert_not_reached ();
}
-#endif
\ No newline at end of file
+#endif
projects / mono.git / blobdiff