threadpool-ms-io.cpp 20 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739
  1. /*
  2. * threadpool-ms-io.c: Microsoft IO threadpool runtime support
  3. *
  4. * Author:
  5. * Ludovic Henry (ludovic.henry@xamarin.com)
  6. *
  7. * Copyright 2015 Xamarin, Inc (http://www.xamarin.com)
  8. * Licensed under the MIT license. See LICENSE file in the project root for full license information.
  9. */
  10. #include "il2cpp-config.h"
  11. #include "gc/WriteBarrier.h"
  12. #ifndef DISABLE_SOCKETS
  13. #ifndef IL2CPP_USE_PIPES_FOR_WAKEUP
  14. #define IL2CPP_USE_PIPES_FOR_WAKEUP !(IL2CPP_TARGET_WINDOWS || IL2CPP_TARGET_XBOXONE || IL2CPP_TARGET_PS4 || IL2CPP_TARGET_PSP2)
  15. #endif
  16. #ifndef IL2CPP_USE_EVENTFD_FOR_WAKEUP
  17. #define IL2CPP_USE_EVENTFD_FOR_WAKEUP (0)
  18. #endif
  19. #if !IL2CPP_USE_PIPES_FOR_WAKEUP && !IL2CPP_USE_EVENTFD_FOR_WAKEUP
  20. #include "os/Win32/WindowsHeaders.h"
  21. #else
  22. #include <errno.h>
  23. #include <fcntl.h>
  24. #include <unistd.h>
  25. #if IL2CPP_USE_EVENTFD_FOR_WAKEUP
  26. #include <sys/eventfd.h>
  27. #endif
  28. #endif
  29. #include <vector>
  30. #include "gc/Allocator.h"
  31. #include "mono/ThreadPool/threadpool-ms.h"
  32. #include "mono/ThreadPool/threadpool-ms-io.h"
  33. #include "mono/ThreadPool/threadpool-ms-io-poll.h"
  34. #include "il2cpp-object-internals.h"
  35. #include "os/ConditionVariable.h"
  36. #include "os/Mutex.h"
  37. #include "os/Socket.h"
  38. #include "utils/CallOnce.h"
  39. #include "utils/Il2CppHashMap.h"
  40. #include "vm/Domain.h"
  41. #include "vm/Runtime.h"
  42. #include "vm/Thread.h"
  43. #define UPDATES_CAPACITY 128
  44. typedef std::vector<Il2CppObject*, il2cpp::gc::Allocator<Il2CppObject*> > ManagedList;
  45. struct ThreadPoolStateHasher
  46. {
  47. size_t operator()(int thread) const
  48. {
  49. return thread;
  50. }
  51. };
  52. typedef Il2CppHashMap<int, ManagedList*, ThreadPoolStateHasher> ThreadPoolStateHash;
  53. typedef enum {
  54. UPDATE_EMPTY = 0,
  55. UPDATE_ADD,
  56. UPDATE_REMOVE_SOCKET,
  57. UPDATE_REMOVE_DOMAIN,
  58. } ThreadPoolIOUpdateType;
  59. typedef struct {
  60. int fd;
  61. Il2CppIOSelectorJob *job;
  62. } ThreadPoolIOUpdate_Add;
  63. typedef struct {
  64. int fd;
  65. } ThreadPoolIOUpdate_RemoveSocket;
  66. typedef struct {
  67. Il2CppDomain *domain;
  68. } ThreadPoolIOUpdate_RemoveDomain;
  69. typedef struct {
  70. ThreadPoolIOUpdateType type;
  71. union {
  72. ThreadPoolIOUpdate_Add add;
  73. ThreadPoolIOUpdate_RemoveSocket remove_socket;
  74. ThreadPoolIOUpdate_RemoveDomain remove_domain;
  75. } data;
  76. } ThreadPoolIOUpdate;
  77. typedef struct
  78. {
  79. bool(*init)(int wakeup_pipe_fd);
  80. void(*register_fd)(int fd, int events, bool is_new);
  81. void(*remove_fd)(int fd);
  82. int(*event_wait)(void(*callback)(int fd, int events, void* user_data), void* user_data);
  83. } ThreadPoolIOBackend;
  84. typedef struct {
  85. ThreadPoolIOBackend backend;
  86. ThreadPoolIOUpdate* updates;
  87. int updates_size;
  88. il2cpp::os::FastMutex updates_lock;
  89. il2cpp::os::ConditionVariable updates_cond;
  90. #if IL2CPP_USE_PIPES_FOR_WAKEUP || IL2CPP_USE_EVENTFD_FOR_WAKEUP
  91. int32_t wakeup_pipes [2];
  92. #else
  93. il2cpp::os::Socket* wakeup_pipes [2];
  94. #endif
  95. } ThreadPoolIO;
  96. static il2cpp::utils::OnceFlag lazy_init_io_status;
  97. static bool io_selector_running = false;
  98. static ThreadPoolIO* threadpool_io;
  99. static ThreadPoolIOBackend backend_poll = { poll_init, poll_register_fd, poll_remove_fd, poll_event_wait };
  100. static Il2CppIOSelectorJob* get_job_for_event (ManagedList *list, int32_t event)
  101. {
  102. IL2CPP_ASSERT(list);
  103. Il2CppIOSelectorJob* foundJob = NULL;
  104. int matchIndex = -1;
  105. for (size_t i = 0; i < list->size(); i++)
  106. {
  107. Il2CppIOSelectorJob *job = (Il2CppIOSelectorJob*)(*list)[i];
  108. if (job->operation == event)
  109. {
  110. foundJob = job;
  111. matchIndex = (int)i;
  112. break;
  113. }
  114. }
  115. if (foundJob == NULL)
  116. return NULL;
  117. list->erase(list->begin() + matchIndex);
  118. return foundJob;
  119. }
  120. static int get_operations_for_jobs (ManagedList *list)
  121. {
  122. int operations = 0;
  123. for (size_t i = 0; i < list->size(); i++)
  124. {
  125. operations |= ((Il2CppIOSelectorJob*)(*list)[i])->operation;
  126. }
  127. return operations;
  128. }
  129. static void selector_thread_wakeup (void)
  130. {
  131. const char msg = 'c';
  132. for (;;)
  133. {
  134. #if IL2CPP_USE_PIPES_FOR_WAKEUP
  135. int32_t written = write (threadpool_io->wakeup_pipes [1], &msg, 1);
  136. if (written == 1)
  137. break;
  138. if (written == -1)
  139. break;
  140. #elif IL2CPP_USE_EVENTFD_FOR_WAKEUP
  141. eventfd_t val = 1;
  142. int32_t written = eventfd_write(threadpool_io->wakeup_pipes[0], val);
  143. if (written == 0)
  144. break;
  145. if (written == -1)
  146. break;
  147. #else
  148. int32_t written = 0;
  149. const il2cpp::os::WaitStatus status = threadpool_io->wakeup_pipes[1]->Send((const uint8_t*)&msg, 1, il2cpp::os::kSocketFlagsNone, &written);
  150. if (written == 1)
  151. break;
  152. if (written == -1)
  153. {
  154. //g_warning ("selector_thread_wakeup: write () failed, error (%d)\n", WSAGetLastError ());
  155. break;
  156. }
  157. if (status == kWaitStatusFailure)
  158. break;
  159. #endif
  160. }
  161. }
  162. static void selector_thread_wakeup_drain_pipes (void)
  163. {
  164. uint8_t buffer [128];
  165. int32_t received;
  166. for (;;) {
  167. #if IL2CPP_USE_PIPES_FOR_WAKEUP
  168. received = read (threadpool_io->wakeup_pipes [0], buffer, sizeof (buffer));
  169. if (received == 0)
  170. break;
  171. if (received == -1) {
  172. if (errno != EINTR && errno != EAGAIN)
  173. IL2CPP_ASSERT(0 && "selector_thread_wakeup_drain_pipes: read () failed");
  174. break;
  175. }
  176. #elif IL2CPP_USE_EVENTFD_FOR_WAKEUP
  177. eventfd_t val;
  178. received = eventfd_read(threadpool_io->wakeup_pipes[0], &val);
  179. if (received == 0)
  180. break;
  181. if (received == -1) {
  182. if (errno != EINTR && errno != EAGAIN)
  183. IL2CPP_ASSERT(0 && "selector_thread_wakeup_drain_pipes: read () failed");
  184. break;
  185. }
  186. #else
  187. il2cpp::os::WaitStatus status = threadpool_io->wakeup_pipes[0]->Receive(buffer, 128, il2cpp::os::kSocketFlagsNone, &received);
  188. if (received == 0)
  189. break;
  190. if (status == kWaitStatusFailure)
  191. break;
  192. #endif
  193. }
  194. }
  195. typedef struct {
  196. Il2CppDomain *domain;
  197. ThreadPoolStateHash *states;
  198. } FilterSockaresForDomainData;
  199. static void filter_jobs_for_domain (void* key, void* value, void* user_data)
  200. {
  201. //FilterSockaresForDomainData *data;
  202. //MonoMList *list = (MonoMList *)value, *element;
  203. //MonoDomain *domain;
  204. //MonoGHashTable *states;
  205. //IL2CPP_ASSERT(user_data);
  206. //data = (FilterSockaresForDomainData *)user_data;
  207. //domain = data->domain;
  208. //states = data->states;
  209. //for (element = list; element; element = mono_mlist_next (element)) {
  210. // Il2CppIOSelectorJob *job = (Il2CppIOSelectorJob*) mono_mlist_get_data (element);
  211. // if (il2cpp::vm::Domain::GetCurrent() == domain)
  212. // mono_mlist_set_data (element, NULL);
  213. //}
  214. ///* we skip all the first elements which are NULL */
  215. //for (; list; list = mono_mlist_next (list)) {
  216. // if (mono_mlist_get_data (list))
  217. // break;
  218. //}
  219. //if (list) {
  220. // IL2CPP_ASSERT(mono_mlist_get_data (list));
  221. // /* we delete all the NULL elements after the first one */
  222. // for (element = list; element;) {
  223. // MonoMList *next;
  224. // if (!(next = mono_mlist_next (element)))
  225. // break;
  226. // if (mono_mlist_get_data (next))
  227. // element = next;
  228. // else
  229. // mono_mlist_set_next (element, mono_mlist_next (next));
  230. // }
  231. //}
  232. //mono_g_hash_table_replace (states, key, list);
  233. IL2CPP_NOT_IMPLEMENTED("TODO");
  234. }
  235. static void wait_callback (int fd, int events, void* user_data)
  236. {
  237. //Il2CppError error;
  238. if (il2cpp::vm::Runtime::IsShuttingDown ())
  239. return;
  240. #if IL2CPP_USE_PIPES_FOR_WAKEUP || IL2CPP_USE_EVENTFD_FOR_WAKEUP
  241. if (fd == threadpool_io->wakeup_pipes [0]) {
  242. #else
  243. if (fd == threadpool_io->wakeup_pipes [0]->GetDescriptor()) {
  244. #endif
  245. //mono_trace (G_LOG_LEVEL_DEBUG, MONO_TRACE_IO_THREADPOOL, "io threadpool: wke");
  246. selector_thread_wakeup_drain_pipes ();
  247. } else {
  248. ThreadPoolStateHash *states;
  249. ManagedList *list = NULL;
  250. //void* k;
  251. bool remove_fd = false;
  252. int operations;
  253. IL2CPP_ASSERT(user_data);
  254. states = (ThreadPoolStateHash *)user_data;
  255. /*mono_trace (G_LOG_LEVEL_DEBUG, MONO_TRACE_IO_THREADPOOL, "io threadpool: cal fd %3d, events = %2s | %2s | %3s",
  256. fd, (events & EVENT_IN) ? "RD" : "..", (events & EVENT_OUT) ? "WR" : "..", (events & EVENT_ERR) ? "ERR" : "...");*/
  257. ThreadPoolStateHash::iterator iter = states->find(fd);
  258. bool exists = iter != states->end();
  259. if (!exists)
  260. IL2CPP_ASSERT("wait_callback: fd not found in states table");
  261. else
  262. list = iter->second;
  263. if (list && (events & EVENT_IN) != 0) {
  264. Il2CppIOSelectorJob *job = get_job_for_event (list, EVENT_IN);
  265. if (job) {
  266. threadpool_ms_enqueue_work_item (il2cpp::vm::Domain::GetCurrent(), (Il2CppObject*) job);
  267. }
  268. }
  269. if (list && (events & EVENT_OUT) != 0) {
  270. Il2CppIOSelectorJob *job = get_job_for_event (list, EVENT_OUT);
  271. if (job) {
  272. threadpool_ms_enqueue_work_item (il2cpp::vm::Domain::GetCurrent(), (Il2CppObject*) job);
  273. }
  274. }
  275. remove_fd = (events & EVENT_ERR) == EVENT_ERR;
  276. if (!remove_fd) {
  277. //mono_g_hash_table_replace (states, int_TO_POINTER (fd), list);
  278. states->insert(ThreadPoolStateHash::value_type(fd, list));
  279. operations = get_operations_for_jobs (list);
  280. /*mono_trace (G_LOG_LEVEL_DEBUG, MONO_TRACE_IO_THREADPOOL, "io threadpool: res fd %3d, events = %2s | %2s | %3s",
  281. fd, (operations & EVENT_IN) ? "RD" : "..", (operations & EVENT_OUT) ? "WR" : "..", (operations & EVENT_ERR) ? "ERR" : "...");*/
  282. threadpool_io->backend.register_fd (fd, operations, false);
  283. } else {
  284. //mono_trace (G_LOG_LEVEL_DEBUG, MONO_TRACE_IO_THREADPOOL, "io threadpool: err fd %d", fd);
  285. states->erase(ThreadPoolStateHash::key_type(fd));
  286. //mono_g_hash_table_remove (states, int_TO_POINTER (fd));
  287. threadpool_io->backend.remove_fd (fd);
  288. }
  289. }
  290. }
  291. static void selector_thread (void* data)
  292. {
  293. //Il2CppError error;
  294. ThreadPoolStateHash *states;
  295. io_selector_running = true;
  296. if (il2cpp::vm::Runtime::IsShuttingDown ()) {
  297. io_selector_running = false;
  298. return;
  299. }
  300. states = new ThreadPoolStateHash();
  301. //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");
  302. for (;;) {
  303. int i, j;
  304. int res;
  305. threadpool_io->updates_lock.Lock();
  306. for (i = 0; i < threadpool_io->updates_size; ++i) {
  307. ThreadPoolIOUpdate *update = &threadpool_io->updates [i];
  308. switch (update->type) {
  309. case UPDATE_EMPTY:
  310. break;
  311. case UPDATE_ADD: {
  312. int fd;
  313. int operations;
  314. //void* k;
  315. bool exists;
  316. ManagedList *list = NULL;
  317. Il2CppIOSelectorJob *job;
  318. fd = update->data.add.fd;
  319. IL2CPP_ASSERT(fd >= 0);
  320. job = update->data.add.job;
  321. IL2CPP_ASSERT(job);
  322. ThreadPoolStateHash::iterator iter = states->find(fd);
  323. exists = iter != states->end();
  324. if (!exists)
  325. list = new ManagedList();
  326. else
  327. list = iter->second;
  328. //exists = mono_g_hash_table_lookup_extended (states, int_TO_POINTER (fd), &k, (void**) &list);
  329. list->push_back((Il2CppObject*)job);
  330. il2cpp::gc::GarbageCollector::SetWriteBarrier((void**)&(*list)[list->size()-1]);
  331. states->insert(ThreadPoolStateHash::value_type(fd, list));
  332. //mono_g_hash_table_replace (states, int_TO_POINTER (fd), list);
  333. operations = get_operations_for_jobs (list);
  334. /*mono_trace (G_LOG_LEVEL_DEBUG, MONO_TRACE_IO_THREADPOOL, "io threadpool: %3s fd %3d, operations = %2s | %2s | %3s",
  335. exists ? "mod" : "add", fd, (operations & EVENT_IN) ? "RD" : "..", (operations & EVENT_OUT) ? "WR" : "..", (operations & EVENT_ERR) ? "ERR" : "...");*/
  336. threadpool_io->backend.register_fd (fd, operations, !exists);
  337. break;
  338. }
  339. case UPDATE_REMOVE_SOCKET: {
  340. int fd;
  341. //void* k;
  342. ManagedList *list = NULL;
  343. fd = update->data.remove_socket.fd;
  344. IL2CPP_ASSERT(fd >= 0);
  345. ThreadPoolStateHash::iterator iter = states->find(fd);
  346. bool exists = iter != states->end();
  347. /*if (mono_g_hash_table_lookup_extended (states, int_TO_POINTER (fd), &k, (void**) &list))*/
  348. if (exists)
  349. {
  350. states->erase(ThreadPoolStateHash::key_type(fd));
  351. //mono_g_hash_table_remove (states, int_TO_POINTER (fd));
  352. for (j = i + 1; j < threadpool_io->updates_size; ++j) {
  353. ThreadPoolIOUpdate *update = &threadpool_io->updates [j];
  354. if (update->type == UPDATE_ADD && update->data.add.fd == fd)
  355. memset (update, 0, sizeof (ThreadPoolIOUpdate));
  356. }
  357. for (size_t i = 0; i < list->size(); i++)
  358. {
  359. threadpool_ms_enqueue_work_item(il2cpp::vm::Domain::GetCurrent(), (*list)[i]);
  360. }
  361. list->clear();
  362. //mono_trace (G_LOG_LEVEL_DEBUG, MONO_TRACE_IO_THREADPOOL, "io threadpool: del fd %3d", fd);
  363. threadpool_io->backend.remove_fd (fd);
  364. }
  365. break;
  366. }
  367. case UPDATE_REMOVE_DOMAIN: {
  368. Il2CppDomain *domain;
  369. domain = update->data.remove_domain.domain;
  370. IL2CPP_ASSERT(domain);
  371. FilterSockaresForDomainData user_data = { domain, states };
  372. //mono_g_hash_table_foreach (states, filter_jobs_for_domain, &user_data);
  373. for (j = i + 1; j < threadpool_io->updates_size; ++j) {
  374. ThreadPoolIOUpdate *update = &threadpool_io->updates [j];
  375. if (update->type == UPDATE_ADD && il2cpp::vm::Domain::GetCurrent() == domain)
  376. memset (update, 0, sizeof (ThreadPoolIOUpdate));
  377. }
  378. break;
  379. }
  380. default:
  381. IL2CPP_ASSERT(0 && "Should not be reached");
  382. }
  383. }
  384. threadpool_io->updates_cond.Broadcast();
  385. if (threadpool_io->updates_size > 0) {
  386. threadpool_io->updates_size = 0;
  387. memset (threadpool_io->updates, 0, UPDATES_CAPACITY * sizeof (ThreadPoolIOUpdate));
  388. }
  389. threadpool_io->updates_lock.Unlock();
  390. //mono_trace (G_LOG_LEVEL_DEBUG, MONO_TRACE_IO_THREADPOOL, "io threadpool: wai");
  391. res = threadpool_io->backend.event_wait (wait_callback, states);
  392. if (res == -1 || il2cpp::vm::Runtime::IsShuttingDown ())
  393. break;
  394. }
  395. delete states;
  396. io_selector_running = false;
  397. }
  398. /* Locking: threadpool_io->updates_lock must be held */
  399. static ThreadPoolIOUpdate* update_get_new (void)
  400. {
  401. ThreadPoolIOUpdate *update = NULL;
  402. IL2CPP_ASSERT(threadpool_io->updates_size <= UPDATES_CAPACITY);
  403. while (threadpool_io->updates_size == UPDATES_CAPACITY) {
  404. /* we wait for updates to be applied in the selector_thread and we loop
  405. * as long as none are available. if it happends too much, then we need
  406. * to increase UPDATES_CAPACITY */
  407. threadpool_io->updates_cond.Wait(&threadpool_io->updates_lock);
  408. }
  409. IL2CPP_ASSERT(threadpool_io->updates_size < UPDATES_CAPACITY);
  410. update = &threadpool_io->updates [threadpool_io->updates_size ++];
  411. return update;
  412. }
  413. static void wakeup_pipes_init(void)
  414. {
  415. #if IL2CPP_USE_PIPES_FOR_WAKEUP
  416. if (pipe (threadpool_io->wakeup_pipes) == -1)
  417. IL2CPP_ASSERT(0 && "wakeup_pipes_init: pipe () failed");
  418. if (fcntl (threadpool_io->wakeup_pipes [0], F_SETFL, O_NONBLOCK) == -1)
  419. IL2CPP_ASSERT(0 && "wakeup_pipes_init: fcntl () failed");
  420. #elif IL2CPP_USE_EVENTFD_FOR_WAKEUP
  421. threadpool_io->wakeup_pipes[0] = eventfd(0, EFD_NONBLOCK);
  422. threadpool_io->wakeup_pipes[1] = -1;
  423. #else
  424. il2cpp::os::Socket serverSock(NULL);
  425. serverSock.Create(il2cpp::os::kAddressFamilyInterNetwork, il2cpp::os::kSocketTypeStream, il2cpp::os::kProtocolTypeTcp);
  426. threadpool_io->wakeup_pipes[1] = new il2cpp::os::Socket(NULL);
  427. il2cpp::os::WaitStatus status = threadpool_io->wakeup_pipes[1]->Create(il2cpp::os::kAddressFamilyInterNetwork, il2cpp::os::kSocketTypeStream, il2cpp::os::kProtocolTypeTcp);
  428. IL2CPP_ASSERT(status != kWaitStatusFailure);
  429. if (serverSock.Bind("127.0.0.1", 0) == kWaitStatusFailure)
  430. {
  431. serverSock.Close();
  432. IL2CPP_ASSERT(0 && "wakeup_pipes_init: bind () failed");
  433. }
  434. il2cpp::os::EndPointInfo info;
  435. memset(&info, 0x00, sizeof(il2cpp::os::EndPointInfo));
  436. if (serverSock.GetLocalEndPointInfo(info) == kWaitStatusFailure)
  437. {
  438. serverSock.Close();
  439. IL2CPP_ASSERT(0 && "wakeup_pipes_init: getsockname () failed");
  440. }
  441. if (serverSock.Listen(1024) == kWaitStatusFailure)
  442. {
  443. serverSock.Close();
  444. IL2CPP_ASSERT(0 && "wakeup_pipes_init: listen () failed");
  445. }
  446. if (threadpool_io->wakeup_pipes[1]->Connect(info.data.inet.address, info.data.inet.port) == kWaitStatusFailure)
  447. {
  448. serverSock.Close();
  449. IL2CPP_ASSERT(0 && "wakeup_pipes_init: connect () failed");
  450. }
  451. status = serverSock.Accept(&threadpool_io->wakeup_pipes[0]);
  452. IL2CPP_ASSERT(status != kWaitStatusFailure);
  453. status = threadpool_io->wakeup_pipes[0]->SetBlocking(false);
  454. if (status == kWaitStatusFailure)
  455. {
  456. threadpool_io->wakeup_pipes[0]->Close();
  457. IL2CPP_ASSERT(0 && "wakeup_pipes_init: SetBlocking () failed");
  458. }
  459. status = threadpool_io->wakeup_pipes[0]->SetSocketOption(il2cpp::os::kSocketOptionLevelTcp, il2cpp::os::kSocketOptionNameNoDelay, 1);
  460. if (status == kWaitStatusFailure)
  461. {
  462. threadpool_io->wakeup_pipes[0]->Close();
  463. IL2CPP_ASSERT(0 && "wakeup_pipes_init: SetSocketOption () failed");
  464. }
  465. status = threadpool_io->wakeup_pipes[1]->SetSocketOption(il2cpp::os::kSocketOptionLevelTcp, il2cpp::os::kSocketOptionNameNoDelay, 1);
  466. if (status == kWaitStatusFailure)
  467. {
  468. threadpool_io->wakeup_pipes[1]->Close();
  469. IL2CPP_ASSERT(0 && "wakeup_pipes_init: SetSocketOption () failed");
  470. }
  471. serverSock.Close();
  472. #endif
  473. }
  474. static bool lazy_is_initialized()
  475. {
  476. return lazy_init_io_status.IsSet();
  477. }
  478. static void threadpool_ms_io_initialize(void* args)
  479. {
  480. IL2CPP_ASSERT(!threadpool_io);
  481. threadpool_io = new ThreadPoolIO();
  482. IL2CPP_ASSERT(threadpool_io);
  483. threadpool_io->updates = (ThreadPoolIOUpdate*)il2cpp::gc::GarbageCollector::AllocateFixed(sizeof(ThreadPoolIOUpdate) * UPDATES_CAPACITY, NULL);
  484. threadpool_io->updates_size = 0;
  485. threadpool_io->backend = backend_poll;
  486. // if (g_getenv ("MONO_ENABLE_AIO") != NULL) {
  487. //#if defined(HAVE_EPOLL)
  488. // threadpool_io->backend = backend_epoll;
  489. //#elif defined(HAVE_KQUEUE)
  490. // threadpool_io->backend = backend_kqueue;
  491. //#endif
  492. // }
  493. wakeup_pipes_init ();
  494. #if IL2CPP_USE_PIPES_FOR_WAKEUP || IL2CPP_USE_EVENTFD_FOR_WAKEUP
  495. if (!threadpool_io->backend.init ((int)threadpool_io->wakeup_pipes [0]))
  496. #else
  497. if (!threadpool_io->backend.init ((int)threadpool_io->wakeup_pipes [0]->GetDescriptor()))
  498. #endif
  499. IL2CPP_ASSERT(0 && "initialize: backend->init () failed");
  500. if (!il2cpp::vm::Thread::CreateInternal(selector_thread, NULL, true, SMALL_STACK))
  501. IL2CPP_ASSERT(0 && "initialize: vm::Thread::CreateInternal () failed ");
  502. }
  503. static void threadpool_ms_io_lazy_initialize()
  504. {
  505. il2cpp::utils::CallOnce(lazy_init_io_status, threadpool_ms_io_initialize, NULL);
  506. }
  507. static void cleanup_ms_io (void)
  508. {
  509. /* we make the assumption along the code that we are
  510. * cleaning up only if the runtime is shutting down */
  511. IL2CPP_ASSERT(il2cpp::vm::Runtime::IsShuttingDown ());
  512. selector_thread_wakeup ();
  513. while (io_selector_running)
  514. il2cpp::vm::Thread::Sleep(1000);
  515. }
  516. void threadpool_ms_io_cleanup (void)
  517. {
  518. if (lazy_init_io_status.IsSet())
  519. cleanup_ms_io();
  520. }
  521. void ves_icall_System_IOSelector_Add (intptr_t handle, Il2CppIOSelectorJob *job)
  522. {
  523. ThreadPoolIOUpdate *update;
  524. IL2CPP_ASSERT(handle != 0);
  525. IL2CPP_ASSERT((job->operation == EVENT_IN) ^ (job->operation == EVENT_OUT));
  526. IL2CPP_ASSERT(job->callback);
  527. if (il2cpp::vm::Runtime::IsShuttingDown ())
  528. return;
  529. /*if (mono_domain_is_unloading (mono_object_domain (job)))
  530. return;*/
  531. threadpool_ms_io_lazy_initialize ();
  532. threadpool_io->updates_lock.Lock();
  533. update = update_get_new ();
  534. il2cpp::os::SocketHandleWrapper socketHandle(il2cpp::os::PointerToSocketHandle(reinterpret_cast<void*>(handle)));
  535. // At least one user has seen an intermittent crash where the socket is null. We're unsure what conditions cause
  536. // this to happen, but checking for a value of NULL here seems to allow their project to continue without
  537. // problems. So let's do the same here. If the value is NULL, we set the update type to be "empty". That will
  538. // cause the selector thread to simply skip this update.
  539. il2cpp::os::Socket* socket = socketHandle.GetSocket();
  540. if (socket != NULL)
  541. {
  542. update->type = UPDATE_ADD;
  543. update->data.add.fd = (int)socket->GetDescriptor();
  544. }
  545. else
  546. {
  547. update->type = UPDATE_EMPTY;
  548. }
  549. il2cpp::gc::WriteBarrier::GenericStore(&update->data.add.job, job);
  550. il2cpp::os::Atomic::FullMemoryBarrier(); /* Ensure this is safely published before we wake up the selector */
  551. selector_thread_wakeup ();
  552. threadpool_io->updates_lock.Unlock();
  553. }
  554. void ves_icall_System_IOSelector_Remove (intptr_t handle)
  555. {
  556. il2cpp::os::SocketHandleWrapper socketHandle(il2cpp::os::PointerToSocketHandle(reinterpret_cast<void*>(handle)));
  557. threadpool_ms_io_remove_socket ((int)socketHandle.GetSocket()->GetDescriptor());
  558. }
  559. void threadpool_ms_io_remove_socket (int fd)
  560. {
  561. ThreadPoolIOUpdate *update;
  562. if (!lazy_is_initialized ())
  563. return;
  564. threadpool_io->updates_lock.Lock();
  565. update = update_get_new ();
  566. update->type = UPDATE_REMOVE_SOCKET;
  567. update->data.add.fd = fd;
  568. il2cpp::os::Atomic::FullMemoryBarrier(); /* Ensure this is safely published before we wake up the selector */
  569. selector_thread_wakeup ();
  570. threadpool_io->updates_cond.Wait(&threadpool_io->updates_lock);
  571. threadpool_io->updates_lock.Unlock();
  572. }
  573. #else
  574. void ves_icall_System_IOSelector_Add (intptr_t handle, Il2CppIOSelectorJob *job)
  575. {
  576. IL2CPP_ASSERT(0 && "Should not be called");
  577. }
  578. void ves_icall_System_IOSelector_Remove (intptr_t handle)
  579. {
  580. IL2CPP_ASSERT(0 && "Should not be called");
  581. }
  582. void threadpool_ms_io_cleanup (void)
  583. {
  584. IL2CPP_ASSERT(0 && "Should not be called");
  585. }
  586. void threadpool_ms_io_remove_socket (int fd)
  587. {
  588. IL2CPP_ASSERT(0 && "Should not be called");
  589. }
  590. #endif