123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739 |
- /*
- * threadpool-ms-io.c: Microsoft IO threadpool runtime support
- *
- * Author:
- * Ludovic Henry (ludovic.henry@xamarin.com)
- *
- * Copyright 2015 Xamarin, Inc (http://www.xamarin.com)
- * Licensed under the MIT license. See LICENSE file in the project root for full license information.
- */
- #include "il2cpp-config.h"
- #include "gc/WriteBarrier.h"
- #ifndef DISABLE_SOCKETS
- #ifndef IL2CPP_USE_PIPES_FOR_WAKEUP
- #define IL2CPP_USE_PIPES_FOR_WAKEUP !(IL2CPP_TARGET_WINDOWS || IL2CPP_TARGET_XBOXONE || IL2CPP_TARGET_PS4 || IL2CPP_TARGET_PSP2)
- #endif
- #ifndef IL2CPP_USE_EVENTFD_FOR_WAKEUP
- #define IL2CPP_USE_EVENTFD_FOR_WAKEUP (0)
- #endif
- #if !IL2CPP_USE_PIPES_FOR_WAKEUP && !IL2CPP_USE_EVENTFD_FOR_WAKEUP
- #include "os/Win32/WindowsHeaders.h"
- #else
- #include <errno.h>
- #include <fcntl.h>
- #include <unistd.h>
- #if IL2CPP_USE_EVENTFD_FOR_WAKEUP
- #include <sys/eventfd.h>
- #endif
- #endif
- #include <vector>
- #include "gc/Allocator.h"
- #include "mono/ThreadPool/threadpool-ms.h"
- #include "mono/ThreadPool/threadpool-ms-io.h"
- #include "mono/ThreadPool/threadpool-ms-io-poll.h"
- #include "il2cpp-object-internals.h"
- #include "os/ConditionVariable.h"
- #include "os/Mutex.h"
- #include "os/Socket.h"
- #include "utils/CallOnce.h"
- #include "utils/Il2CppHashMap.h"
- #include "vm/Domain.h"
- #include "vm/Runtime.h"
- #include "vm/Thread.h"
- #define UPDATES_CAPACITY 128
- typedef std::vector<Il2CppObject*, il2cpp::gc::Allocator<Il2CppObject*> > ManagedList;
- struct ThreadPoolStateHasher
- {
- size_t operator()(int thread) const
- {
- return thread;
- }
- };
- typedef Il2CppHashMap<int, ManagedList*, ThreadPoolStateHasher> ThreadPoolStateHash;
- typedef enum {
- UPDATE_EMPTY = 0,
- UPDATE_ADD,
- UPDATE_REMOVE_SOCKET,
- UPDATE_REMOVE_DOMAIN,
- } ThreadPoolIOUpdateType;
- typedef struct {
- int fd;
- Il2CppIOSelectorJob *job;
- } ThreadPoolIOUpdate_Add;
- typedef struct {
- int fd;
- } ThreadPoolIOUpdate_RemoveSocket;
- typedef struct {
- Il2CppDomain *domain;
- } ThreadPoolIOUpdate_RemoveDomain;
- typedef struct {
- ThreadPoolIOUpdateType type;
- union {
- ThreadPoolIOUpdate_Add add;
- ThreadPoolIOUpdate_RemoveSocket remove_socket;
- ThreadPoolIOUpdate_RemoveDomain remove_domain;
- } data;
- } ThreadPoolIOUpdate;
- typedef struct
- {
- bool(*init)(int wakeup_pipe_fd);
- void(*register_fd)(int fd, int events, bool is_new);
- void(*remove_fd)(int fd);
- int(*event_wait)(void(*callback)(int fd, int events, void* user_data), void* user_data);
- } ThreadPoolIOBackend;
- typedef struct {
- ThreadPoolIOBackend backend;
- ThreadPoolIOUpdate* updates;
- int updates_size;
- il2cpp::os::FastMutex updates_lock;
- il2cpp::os::ConditionVariable updates_cond;
- #if IL2CPP_USE_PIPES_FOR_WAKEUP || IL2CPP_USE_EVENTFD_FOR_WAKEUP
- int32_t wakeup_pipes [2];
- #else
- il2cpp::os::Socket* wakeup_pipes [2];
- #endif
- } ThreadPoolIO;
- static il2cpp::utils::OnceFlag lazy_init_io_status;
- static bool io_selector_running = false;
- static ThreadPoolIO* threadpool_io;
- static ThreadPoolIOBackend backend_poll = { poll_init, poll_register_fd, poll_remove_fd, poll_event_wait };
- static Il2CppIOSelectorJob* get_job_for_event (ManagedList *list, int32_t event)
- {
- IL2CPP_ASSERT(list);
- Il2CppIOSelectorJob* foundJob = NULL;
- int matchIndex = -1;
- for (size_t i = 0; i < list->size(); i++)
- {
- Il2CppIOSelectorJob *job = (Il2CppIOSelectorJob*)(*list)[i];
- if (job->operation == event)
- {
- foundJob = job;
- matchIndex = (int)i;
- break;
- }
- }
- if (foundJob == NULL)
- return NULL;
- list->erase(list->begin() + matchIndex);
- return foundJob;
- }
- static int get_operations_for_jobs (ManagedList *list)
- {
- int operations = 0;
- for (size_t i = 0; i < list->size(); i++)
- {
- operations |= ((Il2CppIOSelectorJob*)(*list)[i])->operation;
- }
- return operations;
- }
- static void selector_thread_wakeup (void)
- {
- const char msg = 'c';
- for (;;)
- {
- #if IL2CPP_USE_PIPES_FOR_WAKEUP
- int32_t written = write (threadpool_io->wakeup_pipes [1], &msg, 1);
- if (written == 1)
- break;
- if (written == -1)
- break;
- #elif IL2CPP_USE_EVENTFD_FOR_WAKEUP
- eventfd_t val = 1;
- int32_t written = eventfd_write(threadpool_io->wakeup_pipes[0], val);
- if (written == 0)
- break;
- if (written == -1)
- break;
- #else
- int32_t written = 0;
- const il2cpp::os::WaitStatus status = threadpool_io->wakeup_pipes[1]->Send((const uint8_t*)&msg, 1, il2cpp::os::kSocketFlagsNone, &written);
- if (written == 1)
- break;
- if (written == -1)
- {
- //g_warning ("selector_thread_wakeup: write () failed, error (%d)\n", WSAGetLastError ());
- break;
- }
- if (status == kWaitStatusFailure)
- break;
- #endif
- }
- }
- static void selector_thread_wakeup_drain_pipes (void)
- {
- uint8_t buffer [128];
- int32_t received;
- for (;;) {
- #if IL2CPP_USE_PIPES_FOR_WAKEUP
- received = read (threadpool_io->wakeup_pipes [0], buffer, sizeof (buffer));
- if (received == 0)
- break;
- if (received == -1) {
- if (errno != EINTR && errno != EAGAIN)
- IL2CPP_ASSERT(0 && "selector_thread_wakeup_drain_pipes: read () failed");
- break;
- }
- #elif IL2CPP_USE_EVENTFD_FOR_WAKEUP
- eventfd_t val;
- received = eventfd_read(threadpool_io->wakeup_pipes[0], &val);
- if (received == 0)
- break;
- if (received == -1) {
- if (errno != EINTR && errno != EAGAIN)
- IL2CPP_ASSERT(0 && "selector_thread_wakeup_drain_pipes: read () failed");
- break;
- }
- #else
- il2cpp::os::WaitStatus status = threadpool_io->wakeup_pipes[0]->Receive(buffer, 128, il2cpp::os::kSocketFlagsNone, &received);
- if (received == 0)
- break;
- if (status == kWaitStatusFailure)
- break;
- #endif
- }
- }
- typedef struct {
- Il2CppDomain *domain;
- ThreadPoolStateHash *states;
- } FilterSockaresForDomainData;
- static void filter_jobs_for_domain (void* key, void* value, void* user_data)
- {
- //FilterSockaresForDomainData *data;
- //MonoMList *list = (MonoMList *)value, *element;
- //MonoDomain *domain;
- //MonoGHashTable *states;
- //IL2CPP_ASSERT(user_data);
- //data = (FilterSockaresForDomainData *)user_data;
- //domain = data->domain;
- //states = data->states;
- //for (element = list; element; element = mono_mlist_next (element)) {
- // Il2CppIOSelectorJob *job = (Il2CppIOSelectorJob*) mono_mlist_get_data (element);
- // if (il2cpp::vm::Domain::GetCurrent() == domain)
- // mono_mlist_set_data (element, NULL);
- //}
- ///* we skip all the first elements which are NULL */
- //for (; list; list = mono_mlist_next (list)) {
- // if (mono_mlist_get_data (list))
- // break;
- //}
- //if (list) {
- // IL2CPP_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));
- // }
- //}
- //mono_g_hash_table_replace (states, key, list);
- IL2CPP_NOT_IMPLEMENTED("TODO");
- }
- static void wait_callback (int fd, int events, void* user_data)
- {
- //Il2CppError error;
- if (il2cpp::vm::Runtime::IsShuttingDown ())
- return;
- #if IL2CPP_USE_PIPES_FOR_WAKEUP || IL2CPP_USE_EVENTFD_FOR_WAKEUP
- if (fd == threadpool_io->wakeup_pipes [0]) {
- #else
- if (fd == threadpool_io->wakeup_pipes [0]->GetDescriptor()) {
- #endif
- //mono_trace (G_LOG_LEVEL_DEBUG, MONO_TRACE_IO_THREADPOOL, "io threadpool: wke");
- selector_thread_wakeup_drain_pipes ();
- } else {
- ThreadPoolStateHash *states;
- ManagedList *list = NULL;
- //void* k;
- bool remove_fd = false;
- int operations;
- IL2CPP_ASSERT(user_data);
- states = (ThreadPoolStateHash *)user_data;
- /*mono_trace (G_LOG_LEVEL_DEBUG, MONO_TRACE_IO_THREADPOOL, "io threadpool: cal fd %3d, events = %2s | %2s | %3s",
- fd, (events & EVENT_IN) ? "RD" : "..", (events & EVENT_OUT) ? "WR" : "..", (events & EVENT_ERR) ? "ERR" : "...");*/
- ThreadPoolStateHash::iterator iter = states->find(fd);
- bool exists = iter != states->end();
- if (!exists)
- IL2CPP_ASSERT("wait_callback: fd not found in states table");
- else
- list = iter->second;
- if (list && (events & EVENT_IN) != 0) {
- Il2CppIOSelectorJob *job = get_job_for_event (list, EVENT_IN);
- if (job) {
- threadpool_ms_enqueue_work_item (il2cpp::vm::Domain::GetCurrent(), (Il2CppObject*) job);
- }
- }
- if (list && (events & EVENT_OUT) != 0) {
- Il2CppIOSelectorJob *job = get_job_for_event (list, EVENT_OUT);
- if (job) {
- threadpool_ms_enqueue_work_item (il2cpp::vm::Domain::GetCurrent(), (Il2CppObject*) job);
- }
- }
- remove_fd = (events & EVENT_ERR) == EVENT_ERR;
- if (!remove_fd) {
- //mono_g_hash_table_replace (states, int_TO_POINTER (fd), list);
- states->insert(ThreadPoolStateHash::value_type(fd, list));
- operations = get_operations_for_jobs (list);
- /*mono_trace (G_LOG_LEVEL_DEBUG, MONO_TRACE_IO_THREADPOOL, "io threadpool: res fd %3d, events = %2s | %2s | %3s",
- fd, (operations & EVENT_IN) ? "RD" : "..", (operations & EVENT_OUT) ? "WR" : "..", (operations & EVENT_ERR) ? "ERR" : "...");*/
- threadpool_io->backend.register_fd (fd, operations, false);
- } else {
- //mono_trace (G_LOG_LEVEL_DEBUG, MONO_TRACE_IO_THREADPOOL, "io threadpool: err fd %d", fd);
- states->erase(ThreadPoolStateHash::key_type(fd));
- //mono_g_hash_table_remove (states, int_TO_POINTER (fd));
- threadpool_io->backend.remove_fd (fd);
- }
- }
- }
- static void selector_thread (void* data)
- {
- //Il2CppError error;
- ThreadPoolStateHash *states;
- io_selector_running = true;
- if (il2cpp::vm::Runtime::IsShuttingDown ()) {
- io_selector_running = false;
- return;
- }
- states = new ThreadPoolStateHash();
- //states = mono_g_hash_table_new_type (g_direct_hash, g_direct_equal, MONO_HASH_VALUE_GC, MONO_ROOT_SOURCE_THREAD_POOL, "i/o thread pool states table");
- for (;;) {
- int i, j;
- int res;
- threadpool_io->updates_lock.Lock();
- 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: {
- int fd;
- int operations;
- //void* k;
- bool exists;
- ManagedList *list = NULL;
- Il2CppIOSelectorJob *job;
- fd = update->data.add.fd;
- IL2CPP_ASSERT(fd >= 0);
- job = update->data.add.job;
- IL2CPP_ASSERT(job);
- ThreadPoolStateHash::iterator iter = states->find(fd);
- exists = iter != states->end();
- if (!exists)
- list = new ManagedList();
- else
- list = iter->second;
- //exists = mono_g_hash_table_lookup_extended (states, int_TO_POINTER (fd), &k, (void**) &list);
- list->push_back((Il2CppObject*)job);
- il2cpp::gc::GarbageCollector::SetWriteBarrier((void**)&(*list)[list->size()-1]);
- states->insert(ThreadPoolStateHash::value_type(fd, list));
- //mono_g_hash_table_replace (states, int_TO_POINTER (fd), list);
- operations = get_operations_for_jobs (list);
- /*mono_trace (G_LOG_LEVEL_DEBUG, MONO_TRACE_IO_THREADPOOL, "io threadpool: %3s fd %3d, operations = %2s | %2s | %3s",
- exists ? "mod" : "add", fd, (operations & EVENT_IN) ? "RD" : "..", (operations & EVENT_OUT) ? "WR" : "..", (operations & EVENT_ERR) ? "ERR" : "...");*/
- threadpool_io->backend.register_fd (fd, operations, !exists);
- break;
- }
- case UPDATE_REMOVE_SOCKET: {
- int fd;
- //void* k;
- ManagedList *list = NULL;
- fd = update->data.remove_socket.fd;
- IL2CPP_ASSERT(fd >= 0);
- ThreadPoolStateHash::iterator iter = states->find(fd);
- bool exists = iter != states->end();
- /*if (mono_g_hash_table_lookup_extended (states, int_TO_POINTER (fd), &k, (void**) &list))*/
- if (exists)
- {
- states->erase(ThreadPoolStateHash::key_type(fd));
- //mono_g_hash_table_remove (states, int_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 (size_t i = 0; i < list->size(); i++)
- {
- threadpool_ms_enqueue_work_item(il2cpp::vm::Domain::GetCurrent(), (*list)[i]);
- }
- list->clear();
- //mono_trace (G_LOG_LEVEL_DEBUG, MONO_TRACE_IO_THREADPOOL, "io threadpool: del fd %3d", fd);
- threadpool_io->backend.remove_fd (fd);
- }
- break;
- }
- case UPDATE_REMOVE_DOMAIN: {
- Il2CppDomain *domain;
- domain = update->data.remove_domain.domain;
- IL2CPP_ASSERT(domain);
- FilterSockaresForDomainData user_data = { domain, states };
- //mono_g_hash_table_foreach (states, filter_jobs_for_domain, &user_data);
- for (j = i + 1; j < threadpool_io->updates_size; ++j) {
- ThreadPoolIOUpdate *update = &threadpool_io->updates [j];
- if (update->type == UPDATE_ADD && il2cpp::vm::Domain::GetCurrent() == domain)
- memset (update, 0, sizeof (ThreadPoolIOUpdate));
- }
- break;
- }
- default:
- IL2CPP_ASSERT(0 && "Should not be reached");
- }
- }
- threadpool_io->updates_cond.Broadcast();
- if (threadpool_io->updates_size > 0) {
- threadpool_io->updates_size = 0;
- memset (threadpool_io->updates, 0, UPDATES_CAPACITY * sizeof (ThreadPoolIOUpdate));
- }
- threadpool_io->updates_lock.Unlock();
- //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 || il2cpp::vm::Runtime::IsShuttingDown ())
- break;
- }
- delete states;
- io_selector_running = false;
- }
- /* Locking: threadpool_io->updates_lock must be held */
- static ThreadPoolIOUpdate* update_get_new (void)
- {
- ThreadPoolIOUpdate *update = NULL;
- IL2CPP_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 */
- threadpool_io->updates_cond.Wait(&threadpool_io->updates_lock);
- }
- IL2CPP_ASSERT(threadpool_io->updates_size < UPDATES_CAPACITY);
- update = &threadpool_io->updates [threadpool_io->updates_size ++];
- return update;
- }
- static void wakeup_pipes_init(void)
- {
- #if IL2CPP_USE_PIPES_FOR_WAKEUP
- if (pipe (threadpool_io->wakeup_pipes) == -1)
- IL2CPP_ASSERT(0 && "wakeup_pipes_init: pipe () failed");
- if (fcntl (threadpool_io->wakeup_pipes [0], F_SETFL, O_NONBLOCK) == -1)
- IL2CPP_ASSERT(0 && "wakeup_pipes_init: fcntl () failed");
- #elif IL2CPP_USE_EVENTFD_FOR_WAKEUP
- threadpool_io->wakeup_pipes[0] = eventfd(0, EFD_NONBLOCK);
- threadpool_io->wakeup_pipes[1] = -1;
- #else
- il2cpp::os::Socket serverSock(NULL);
- serverSock.Create(il2cpp::os::kAddressFamilyInterNetwork, il2cpp::os::kSocketTypeStream, il2cpp::os::kProtocolTypeTcp);
- threadpool_io->wakeup_pipes[1] = new il2cpp::os::Socket(NULL);
- il2cpp::os::WaitStatus status = threadpool_io->wakeup_pipes[1]->Create(il2cpp::os::kAddressFamilyInterNetwork, il2cpp::os::kSocketTypeStream, il2cpp::os::kProtocolTypeTcp);
- IL2CPP_ASSERT(status != kWaitStatusFailure);
- if (serverSock.Bind("127.0.0.1", 0) == kWaitStatusFailure)
- {
- serverSock.Close();
- IL2CPP_ASSERT(0 && "wakeup_pipes_init: bind () failed");
- }
- il2cpp::os::EndPointInfo info;
- memset(&info, 0x00, sizeof(il2cpp::os::EndPointInfo));
- if (serverSock.GetLocalEndPointInfo(info) == kWaitStatusFailure)
- {
- serverSock.Close();
- IL2CPP_ASSERT(0 && "wakeup_pipes_init: getsockname () failed");
- }
- if (serverSock.Listen(1024) == kWaitStatusFailure)
- {
- serverSock.Close();
- IL2CPP_ASSERT(0 && "wakeup_pipes_init: listen () failed");
- }
- if (threadpool_io->wakeup_pipes[1]->Connect(info.data.inet.address, info.data.inet.port) == kWaitStatusFailure)
- {
- serverSock.Close();
- IL2CPP_ASSERT(0 && "wakeup_pipes_init: connect () failed");
- }
- status = serverSock.Accept(&threadpool_io->wakeup_pipes[0]);
- IL2CPP_ASSERT(status != kWaitStatusFailure);
- status = threadpool_io->wakeup_pipes[0]->SetBlocking(false);
- if (status == kWaitStatusFailure)
- {
- threadpool_io->wakeup_pipes[0]->Close();
- IL2CPP_ASSERT(0 && "wakeup_pipes_init: SetBlocking () failed");
- }
- status = threadpool_io->wakeup_pipes[0]->SetSocketOption(il2cpp::os::kSocketOptionLevelTcp, il2cpp::os::kSocketOptionNameNoDelay, 1);
- if (status == kWaitStatusFailure)
- {
- threadpool_io->wakeup_pipes[0]->Close();
- IL2CPP_ASSERT(0 && "wakeup_pipes_init: SetSocketOption () failed");
- }
- status = threadpool_io->wakeup_pipes[1]->SetSocketOption(il2cpp::os::kSocketOptionLevelTcp, il2cpp::os::kSocketOptionNameNoDelay, 1);
- if (status == kWaitStatusFailure)
- {
- threadpool_io->wakeup_pipes[1]->Close();
- IL2CPP_ASSERT(0 && "wakeup_pipes_init: SetSocketOption () failed");
- }
- serverSock.Close();
- #endif
- }
- static bool lazy_is_initialized()
- {
- return lazy_init_io_status.IsSet();
- }
- static void threadpool_ms_io_initialize(void* args)
- {
- IL2CPP_ASSERT(!threadpool_io);
- threadpool_io = new ThreadPoolIO();
- IL2CPP_ASSERT(threadpool_io);
- threadpool_io->updates = (ThreadPoolIOUpdate*)il2cpp::gc::GarbageCollector::AllocateFixed(sizeof(ThreadPoolIOUpdate) * UPDATES_CAPACITY, NULL);
- threadpool_io->updates_size = 0;
- threadpool_io->backend = backend_poll;
- // if (g_getenv ("MONO_ENABLE_AIO") != NULL) {
- //#if defined(HAVE_EPOLL)
- // threadpool_io->backend = backend_epoll;
- //#elif defined(HAVE_KQUEUE)
- // threadpool_io->backend = backend_kqueue;
- //#endif
- // }
- wakeup_pipes_init ();
- #if IL2CPP_USE_PIPES_FOR_WAKEUP || IL2CPP_USE_EVENTFD_FOR_WAKEUP
- if (!threadpool_io->backend.init ((int)threadpool_io->wakeup_pipes [0]))
- #else
- if (!threadpool_io->backend.init ((int)threadpool_io->wakeup_pipes [0]->GetDescriptor()))
- #endif
- IL2CPP_ASSERT(0 && "initialize: backend->init () failed");
- if (!il2cpp::vm::Thread::CreateInternal(selector_thread, NULL, true, SMALL_STACK))
- IL2CPP_ASSERT(0 && "initialize: vm::Thread::CreateInternal () failed ");
- }
- static void threadpool_ms_io_lazy_initialize()
- {
- il2cpp::utils::CallOnce(lazy_init_io_status, threadpool_ms_io_initialize, NULL);
- }
- static void cleanup_ms_io (void)
- {
- /* we make the assumption along the code that we are
- * cleaning up only if the runtime is shutting down */
- IL2CPP_ASSERT(il2cpp::vm::Runtime::IsShuttingDown ());
- selector_thread_wakeup ();
- while (io_selector_running)
- il2cpp::vm::Thread::Sleep(1000);
- }
- void threadpool_ms_io_cleanup (void)
- {
- if (lazy_init_io_status.IsSet())
- cleanup_ms_io();
- }
- void ves_icall_System_IOSelector_Add (intptr_t handle, Il2CppIOSelectorJob *job)
- {
- ThreadPoolIOUpdate *update;
- IL2CPP_ASSERT(handle != 0);
- IL2CPP_ASSERT((job->operation == EVENT_IN) ^ (job->operation == EVENT_OUT));
- IL2CPP_ASSERT(job->callback);
- if (il2cpp::vm::Runtime::IsShuttingDown ())
- return;
- /*if (mono_domain_is_unloading (mono_object_domain (job)))
- return;*/
- threadpool_ms_io_lazy_initialize ();
- threadpool_io->updates_lock.Lock();
- update = update_get_new ();
- il2cpp::os::SocketHandleWrapper socketHandle(il2cpp::os::PointerToSocketHandle(reinterpret_cast<void*>(handle)));
- // At least one user has seen an intermittent crash where the socket is null. We're unsure what conditions cause
- // this to happen, but checking for a value of NULL here seems to allow their project to continue without
- // problems. So let's do the same here. If the value is NULL, we set the update type to be "empty". That will
- // cause the selector thread to simply skip this update.
- il2cpp::os::Socket* socket = socketHandle.GetSocket();
- if (socket != NULL)
- {
- update->type = UPDATE_ADD;
- update->data.add.fd = (int)socket->GetDescriptor();
- }
- else
- {
- update->type = UPDATE_EMPTY;
- }
- il2cpp::gc::WriteBarrier::GenericStore(&update->data.add.job, job);
- il2cpp::os::Atomic::FullMemoryBarrier(); /* Ensure this is safely published before we wake up the selector */
- selector_thread_wakeup ();
- threadpool_io->updates_lock.Unlock();
- }
- void ves_icall_System_IOSelector_Remove (intptr_t handle)
- {
- il2cpp::os::SocketHandleWrapper socketHandle(il2cpp::os::PointerToSocketHandle(reinterpret_cast<void*>(handle)));
- threadpool_ms_io_remove_socket ((int)socketHandle.GetSocket()->GetDescriptor());
- }
- void threadpool_ms_io_remove_socket (int fd)
- {
- ThreadPoolIOUpdate *update;
- if (!lazy_is_initialized ())
- return;
- threadpool_io->updates_lock.Lock();
- update = update_get_new ();
- update->type = UPDATE_REMOVE_SOCKET;
- update->data.add.fd = fd;
- il2cpp::os::Atomic::FullMemoryBarrier(); /* Ensure this is safely published before we wake up the selector */
- selector_thread_wakeup ();
- threadpool_io->updates_cond.Wait(&threadpool_io->updates_lock);
- threadpool_io->updates_lock.Unlock();
- }
- #else
- void ves_icall_System_IOSelector_Add (intptr_t handle, Il2CppIOSelectorJob *job)
- {
- IL2CPP_ASSERT(0 && "Should not be called");
- }
- void ves_icall_System_IOSelector_Remove (intptr_t handle)
- {
- IL2CPP_ASSERT(0 && "Should not be called");
- }
- void threadpool_ms_io_cleanup (void)
- {
- IL2CPP_ASSERT(0 && "Should not be called");
- }
- void threadpool_ms_io_remove_socket (int fd)
- {
- IL2CPP_ASSERT(0 && "Should not be called");
- }
- #endif
|