ua_server_worker.c 18 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490
  1. /* This Source Code Form is subject to the terms of the Mozilla Public
  2. * License, v. 2.0. If a copy of the MPL was not distributed with this
  3. * file, You can obtain one at http://mozilla.org/MPL/2.0/. */
  4. #include "ua_util.h"
  5. #include "ua_server_internal.h"
  6. /**
  7. * There are four types of job execution:
  8. *
  9. * 1. Normal jobs (dispatched to worker threads if multithreading is activated)
  10. *
  11. * 2. Repeated jobs with a repetition interval (dispatched to worker threads)
  12. *
  13. * 3. Mainloop jobs are executed (once) from the mainloop and not in the worker threads. The server
  14. * contains a stack structure where all threads can add mainloop jobs for the next mainloop
  15. * iteration. This is used e.g. to trigger adding and removing repeated jobs without blocking the
  16. * mainloop.
  17. *
  18. * 4. Delayed jobs are executed once in a worker thread. But only when all normal jobs that were
  19. * dispatched earlier have been executed. This is achieved by a counter in the worker threads. We
  20. * compute from the counter if all previous jobs have finished. The delay can be very long, since we
  21. * try to not interfere too much with normal execution. A use case is to eventually free obsolete
  22. * structures that _could_ still be accessed from concurrent threads.
  23. *
  24. * - Remove the entry from the list
  25. * - mark it as "dead" with an atomic operation
  26. * - add a delayed job that frees the memory when all concurrent operations have completed
  27. *
  28. * This approach to concurrently accessible memory is known as epoch based reclamation [1]. According to
  29. * [2], it performs competitively well on many-core systems. Our version of EBR does however not require
  30. * a global epoch. Instead, every worker thread has its own epoch counter that we observe for changes.
  31. *
  32. * [1] Fraser, K. 2003. Practical lock freedom. Ph.D. thesis. Computer Laboratory, University of Cambridge.
  33. * [2] Hart, T. E., McKenney, P. E., Brown, A. D., & Walpole, J. (2007). Performance of memory reclamation
  34. * for lockless synchronization. Journal of Parallel and Distributed Computing, 67(12), 1270-1285.
  35. *
  36. * Future Plans: Use work-stealing to load-balance between cores.
  37. * [3] Le, Nhat Minh, et al. "Correct and efficient work-stealing for weak
  38. * memory models." ACM SIGPLAN Notices. Vol. 48. No. 8. ACM, 2013.
  39. */
  40. void
  41. UA_Server_processJob(UA_Server *server, UA_Job *job) {
  42. UA_ASSERT_RCU_UNLOCKED();
  43. UA_RCU_LOCK();
  44. switch(job->type) {
  45. case UA_JOBTYPE_NOTHING:
  46. break;
  47. case UA_JOBTYPE_DETACHCONNECTION:
  48. UA_Connection_detachSecureChannel(job->job.closeConnection);
  49. break;
  50. case UA_JOBTYPE_BINARYMESSAGE_NETWORKLAYER:
  51. {
  52. UA_Server_processBinaryMessage(server, job->job.binaryMessage.connection,
  53. &job->job.binaryMessage.message);
  54. UA_Connection *connection = job->job.binaryMessage.connection;
  55. connection->releaseRecvBuffer(connection, &job->job.binaryMessage.message);
  56. }
  57. break;
  58. case UA_JOBTYPE_BINARYMESSAGE_ALLOCATED:
  59. UA_Server_processBinaryMessage(server, job->job.binaryMessage.connection,
  60. &job->job.binaryMessage.message);
  61. UA_ByteString_deleteMembers(&job->job.binaryMessage.message);
  62. break;
  63. case UA_JOBTYPE_METHODCALL:
  64. case UA_JOBTYPE_METHODCALL_DELAYED:
  65. job->job.methodCall.method(server, job->job.methodCall.data);
  66. break;
  67. default:
  68. UA_LOG_WARNING(server->config.logger, UA_LOGCATEGORY_SERVER,
  69. "Trying to execute a job of unknown type");
  70. break;
  71. }
  72. UA_RCU_UNLOCK();
  73. }
  74. /*******************************/
  75. /* Worker Threads and Dispatch */
  76. /*******************************/
  77. #ifdef UA_ENABLE_MULTITHREADING
  78. struct DispatchJob {
  79. struct cds_wfcq_node node; // node for the queue
  80. UA_Job job;
  81. };
  82. static void *
  83. workerLoop(UA_Worker *worker) {
  84. UA_Server *server = worker->server;
  85. UA_UInt32 *counter = &worker->counter;
  86. volatile UA_Boolean *running = &worker->running;
  87. /* Initialize the (thread local) random seed with the ram address of worker */
  88. UA_random_seed((uintptr_t)worker);
  89. rcu_register_thread();
  90. while(*running) {
  91. struct DispatchJob *dj = (struct DispatchJob*)
  92. cds_wfcq_dequeue_blocking(&server->dispatchQueue_head, &server->dispatchQueue_tail);
  93. if(dj) {
  94. UA_Server_processJob(server, &dj->job);
  95. UA_free(dj);
  96. } else {
  97. /* nothing to do. sleep until a job is dispatched (and wakes up all worker threads) */
  98. pthread_mutex_lock(&server->dispatchQueue_mutex);
  99. pthread_cond_wait(&server->dispatchQueue_condition, &server->dispatchQueue_mutex);
  100. pthread_mutex_unlock(&server->dispatchQueue_mutex);
  101. }
  102. UA_atomic_add(counter, 1);
  103. }
  104. UA_ASSERT_RCU_UNLOCKED();
  105. rcu_barrier(); // wait for all scheduled call_rcu work to complete
  106. rcu_unregister_thread();
  107. UA_LOG_DEBUG(server->config.logger, UA_LOGCATEGORY_SERVER, "Worker shut down");
  108. return NULL;
  109. }
  110. void
  111. UA_Server_dispatchJob(UA_Server *server, const UA_Job *job) {
  112. struct DispatchJob *dj = UA_malloc(sizeof(struct DispatchJob));
  113. // todo: check malloc
  114. dj->job = *job;
  115. cds_wfcq_node_init(&dj->node);
  116. cds_wfcq_enqueue(&server->dispatchQueue_head, &server->dispatchQueue_tail, &dj->node);
  117. }
  118. static void
  119. emptyDispatchQueue(UA_Server *server) {
  120. while(!cds_wfcq_empty(&server->dispatchQueue_head, &server->dispatchQueue_tail)) {
  121. struct DispatchJob *dj = (struct DispatchJob*)
  122. cds_wfcq_dequeue_blocking(&server->dispatchQueue_head, &server->dispatchQueue_tail);
  123. UA_Server_processJob(server, &dj->job);
  124. UA_free(dj);
  125. }
  126. }
  127. #endif
  128. /****************/
  129. /* Delayed Jobs */
  130. /****************/
  131. #ifndef UA_ENABLE_MULTITHREADING
  132. typedef struct UA_DelayedJob {
  133. SLIST_ENTRY(UA_DelayedJob) next;
  134. UA_Job job;
  135. } UA_DelayedJob;
  136. UA_StatusCode
  137. UA_Server_delayedCallback(UA_Server *server, UA_ServerCallback callback, void *data) {
  138. UA_DelayedJob *dj = (UA_DelayedJob *)UA_malloc(sizeof(UA_DelayedJob));
  139. if(!dj)
  140. return UA_STATUSCODE_BADOUTOFMEMORY;
  141. dj->job.type = UA_JOBTYPE_METHODCALL;
  142. dj->job.job.methodCall.data = data;
  143. dj->job.job.methodCall.method = callback;
  144. SLIST_INSERT_HEAD(&server->delayedCallbacks, dj, next);
  145. return UA_STATUSCODE_GOOD;
  146. }
  147. static void
  148. processDelayedCallbacks(UA_Server *server) {
  149. UA_DelayedJob *dj, *dj_tmp;
  150. SLIST_FOREACH_SAFE(dj, &server->delayedCallbacks, next, dj_tmp) {
  151. SLIST_REMOVE(&server->delayedCallbacks, dj, UA_DelayedJob, next);
  152. UA_Server_processJob(server, &dj->job);
  153. UA_free(dj);
  154. }
  155. }
  156. #else
  157. #define DELAYEDJOBSSIZE 100 // Collect delayed jobs until we have DELAYEDWORKSIZE items
  158. struct DelayedJobs {
  159. struct DelayedJobs *next;
  160. UA_UInt32 *workerCounters; // initially NULL until the counter are set
  161. UA_UInt32 jobsCount; // the size of the array is DELAYEDJOBSSIZE, the count may be less
  162. UA_Job jobs[DELAYEDJOBSSIZE]; // when it runs full, a new delayedJobs entry is created
  163. };
  164. /* Dispatched as an ordinary job when the DelayedJobs list is full */
  165. static void getCounters(UA_Server *server, struct DelayedJobs *delayed) {
  166. UA_UInt32 *counters = UA_malloc(server->config.nThreads * sizeof(UA_UInt32));
  167. for(UA_UInt16 i = 0; i < server->config.nThreads; ++i)
  168. counters[i] = server->workers[i].counter;
  169. delayed->workerCounters = counters;
  170. }
  171. /* Call from the main thread only. This is the only function that modifies */
  172. /* server->delayedWork. processDelayedWorkQueue modifies the "next" (after the */
  173. /* head). */
  174. static void addDelayedJob(UA_Server *server, UA_Job *job) {
  175. struct DelayedJobs *dj = server->delayedJobs;
  176. if(!dj || dj->jobsCount >= DELAYEDJOBSSIZE) {
  177. /* create a new DelayedJobs and add it to the linked list */
  178. dj = UA_malloc(sizeof(struct DelayedJobs));
  179. if(!dj) {
  180. UA_LOG_ERROR(server->config.logger, UA_LOGCATEGORY_SERVER,
  181. "Not enough memory to add a delayed job");
  182. return;
  183. }
  184. dj->jobsCount = 0;
  185. dj->workerCounters = NULL;
  186. dj->next = server->delayedJobs;
  187. server->delayedJobs = dj;
  188. /* dispatch a method that sets the counter for the full list that comes afterwards */
  189. if(dj->next) {
  190. UA_Job setCounter = (UA_Job){
  191. .type = UA_JOBTYPE_METHODCALL, .job.methodCall =
  192. {.method = (void (*)(UA_Server*, void*))getCounters, .data = dj->next}};
  193. UA_Server_dispatchJob(server, &setCounter);
  194. }
  195. }
  196. dj->jobs[dj->jobsCount] = *job;
  197. ++dj->jobsCount;
  198. }
  199. static void
  200. delayed_free(UA_Server *server, void *data) {
  201. UA_free(data);
  202. }
  203. UA_StatusCode UA_Server_delayedFree(UA_Server *server, void *data) {
  204. return UA_Server_delayedCallback(server, delayed_free, data);
  205. }
  206. static void
  207. addDelayedJobAsync(UA_Server *server, UA_Job *job) {
  208. addDelayedJob(server, job);
  209. UA_free(job);
  210. }
  211. UA_StatusCode
  212. UA_Server_delayedCallback(UA_Server *server, UA_ServerCallback callback, void *data) {
  213. UA_Job *j = UA_malloc(sizeof(UA_Job));
  214. if(!j)
  215. return UA_STATUSCODE_BADOUTOFMEMORY;
  216. j->type = UA_JOBTYPE_METHODCALL;
  217. j->job.methodCall.data = data;
  218. j->job.methodCall.method = callback;
  219. struct MainLoopJob *mlw = UA_malloc(sizeof(struct MainLoopJob));
  220. mlw->job = (UA_Job) {.type = UA_JOBTYPE_METHODCALL, .job.methodCall =
  221. {.data = j, .method = (UA_ServerCallback)addDelayedJobAsync}};
  222. cds_lfs_push(&server->mainLoopJobs, &mlw->node);
  223. return UA_STATUSCODE_GOOD;
  224. }
  225. /* Find out which delayed jobs can be executed now */
  226. static void
  227. dispatchDelayedJobs(UA_Server *server, void *_) {
  228. /* start at the second */
  229. struct DelayedJobs *dw = server->delayedJobs, *beforedw = dw;
  230. if(dw)
  231. dw = dw->next;
  232. /* find the first delayedwork where the counters have been set and have moved */
  233. while(dw) {
  234. if(!dw->workerCounters) {
  235. beforedw = dw;
  236. dw = dw->next;
  237. continue;
  238. }
  239. UA_Boolean allMoved = true;
  240. for(size_t i = 0; i < server->config.nThreads; ++i) {
  241. if(dw->workerCounters[i] == server->workers[i].counter) {
  242. allMoved = false;
  243. break;
  244. }
  245. }
  246. if(allMoved)
  247. break;
  248. beforedw = dw;
  249. dw = dw->next;
  250. }
  251. /* process and free all delayed jobs from here on */
  252. while(dw) {
  253. for(size_t i = 0; i < dw->jobsCount; ++i)
  254. UA_Server_processJob(server, &dw->jobs[i]);
  255. struct DelayedJobs *next = UA_atomic_xchg((void**)&beforedw->next, NULL);
  256. UA_free(dw->workerCounters);
  257. UA_free(dw);
  258. dw = next;
  259. }
  260. }
  261. #endif
  262. /********************/
  263. /* Main Server Loop */
  264. /********************/
  265. #ifdef UA_ENABLE_MULTITHREADING
  266. static void processMainLoopJobs(UA_Server *server) {
  267. /* no synchronization required if we only use push and pop_all */
  268. struct cds_lfs_head *head = __cds_lfs_pop_all(&server->mainLoopJobs);
  269. if(!head)
  270. return;
  271. struct MainLoopJob *mlw = (struct MainLoopJob*)&head->node;
  272. struct MainLoopJob *next;
  273. do {
  274. UA_Server_processJob(server, &mlw->job);
  275. next = (struct MainLoopJob*)mlw->node.next;
  276. UA_free(mlw);
  277. //cppcheck-suppress unreadVariable
  278. } while((mlw = next));
  279. }
  280. #endif
  281. UA_StatusCode UA_Server_run_startup(UA_Server *server) {
  282. #ifdef UA_ENABLE_MULTITHREADING
  283. /* Spin up the worker threads */
  284. UA_LOG_INFO(server->config.logger, UA_LOGCATEGORY_SERVER,
  285. "Spinning up %u worker thread(s)", server->config.nThreads);
  286. pthread_cond_init(&server->dispatchQueue_condition, 0);
  287. pthread_mutex_init(&server->dispatchQueue_mutex, 0);
  288. server->workers = UA_malloc(server->config.nThreads * sizeof(UA_Worker));
  289. if(!server->workers)
  290. return UA_STATUSCODE_BADOUTOFMEMORY;
  291. for(size_t i = 0; i < server->config.nThreads; ++i) {
  292. UA_Worker *worker = &server->workers[i];
  293. worker->server = server;
  294. worker->counter = 0;
  295. worker->running = true;
  296. pthread_create(&worker->thr, NULL, (void* (*)(void*))workerLoop, worker);
  297. }
  298. /* Try to execute delayed callbacks every 10 sec */
  299. UA_Job processDelayed = {.type = UA_JOBTYPE_METHODCALL,
  300. .job.methodCall = {.method = dispatchDelayedJobs, .data = NULL} };
  301. UA_Server_addRepeatedJob(server, processDelayed, 10000, NULL);
  302. #endif
  303. /* Start the networklayers */
  304. UA_StatusCode result = UA_STATUSCODE_GOOD;
  305. for(size_t i = 0; i < server->config.networkLayersSize; ++i) {
  306. UA_ServerNetworkLayer *nl = &server->config.networkLayers[i];
  307. result |= nl->start(nl, server->config.logger);
  308. }
  309. return result;
  310. }
  311. /* completeMessages is run synchronous on the jobs returned from the network
  312. layer, so that the order for processing TCP packets is never mixed up. */
  313. static void
  314. completeMessages(UA_Server *server, UA_Job *job) {
  315. UA_Boolean realloced = UA_FALSE;
  316. UA_StatusCode retval = UA_Connection_completeMessages(job->job.binaryMessage.connection,
  317. &job->job.binaryMessage.message, &realloced);
  318. if(retval != UA_STATUSCODE_GOOD) {
  319. if(retval == UA_STATUSCODE_BADOUTOFMEMORY)
  320. UA_LOG_WARNING(server->config.logger, UA_LOGCATEGORY_NETWORK,
  321. "Lost message(s) from Connection %i as memory could not be allocated",
  322. job->job.binaryMessage.connection->sockfd);
  323. else if(retval != UA_STATUSCODE_GOOD)
  324. UA_LOG_INFO(server->config.logger, UA_LOGCATEGORY_NETWORK,
  325. "Could not merge half-received messages on Connection %i with error 0x%08x",
  326. job->job.binaryMessage.connection->sockfd, retval);
  327. job->type = UA_JOBTYPE_NOTHING;
  328. return;
  329. }
  330. if(realloced)
  331. job->type = UA_JOBTYPE_BINARYMESSAGE_ALLOCATED;
  332. /* discard the job if message is empty - also no leak is possible here */
  333. if(job->job.binaryMessage.message.length == 0)
  334. job->type = UA_JOBTYPE_NOTHING;
  335. }
  336. UA_UInt16 UA_Server_run_iterate(UA_Server *server, UA_Boolean waitInternal) {
  337. #ifdef UA_ENABLE_MULTITHREADING
  338. /* Run work assigned for the main thread */
  339. processMainLoopJobs(server);
  340. #endif
  341. /* Process repeated work */
  342. UA_DateTime now = UA_DateTime_nowMonotonic();
  343. UA_Boolean dispatched = false; /* to wake up worker threads */
  344. UA_DateTime nextRepeated = UA_Server_processRepeatedJobs(server, now, &dispatched);
  345. UA_UInt16 timeout = 0;
  346. if(waitInternal)
  347. timeout = (UA_UInt16)((nextRepeated - now) / UA_MSEC_TO_DATETIME);
  348. /* Get work from the networklayer */
  349. for(size_t i = 0; i < server->config.networkLayersSize; ++i) {
  350. UA_ServerNetworkLayer *nl = &server->config.networkLayers[i];
  351. UA_Job *jobs = NULL;
  352. size_t jobsSize;
  353. /* only the last networklayer waits on the tieout */
  354. if(i == server->config.networkLayersSize-1)
  355. jobsSize = nl->getJobs(nl, &jobs, timeout);
  356. else
  357. jobsSize = nl->getJobs(nl, &jobs, 0);
  358. for(size_t k = 0; k < jobsSize; ++k) {
  359. #ifdef UA_ENABLE_MULTITHREADING
  360. /* Filter out delayed work */
  361. if(jobs[k].type == UA_JOBTYPE_METHODCALL_DELAYED) {
  362. addDelayedJob(server, &jobs[k]);
  363. jobs[k].type = UA_JOBTYPE_NOTHING;
  364. continue;
  365. }
  366. #endif
  367. /* Merge half-received messages */
  368. if(jobs[k].type == UA_JOBTYPE_BINARYMESSAGE_NETWORKLAYER)
  369. completeMessages(server, &jobs[k]);
  370. }
  371. /* Dispatch/process jobs */
  372. for(size_t j = 0; j < jobsSize; ++j) {
  373. #ifdef UA_ENABLE_MULTITHREADING
  374. UA_Server_dispatchJob(server, &jobs[j]);
  375. dispatched = true;
  376. #else
  377. UA_Server_processJob(server, &jobs[j]);
  378. #endif
  379. }
  380. /* Clean up jobs list */
  381. if(jobsSize > 0)
  382. UA_free(jobs);
  383. }
  384. #ifdef UA_ENABLE_MULTITHREADING
  385. /* Wake up worker threads */
  386. if(dispatched)
  387. pthread_cond_broadcast(&server->dispatchQueue_condition);
  388. #else
  389. processDelayedCallbacks(server);
  390. #endif
  391. now = UA_DateTime_nowMonotonic();
  392. timeout = 0;
  393. if(nextRepeated > now)
  394. timeout = (UA_UInt16)((nextRepeated - now) / UA_MSEC_TO_DATETIME);
  395. return timeout;
  396. }
  397. UA_StatusCode UA_Server_run_shutdown(UA_Server *server) {
  398. for(size_t i = 0; i < server->config.networkLayersSize; ++i) {
  399. UA_ServerNetworkLayer *nl = &server->config.networkLayers[i];
  400. UA_Job *stopJobs = NULL;
  401. size_t stopJobsSize = nl->stop(nl, &stopJobs);
  402. for(size_t j = 0; j < stopJobsSize; ++j)
  403. UA_Server_processJob(server, &stopJobs[j]);
  404. UA_free(stopJobs);
  405. }
  406. #ifdef UA_ENABLE_MULTITHREADING
  407. /* Ensure that run_shutdown can be called multiple times */
  408. if(server->workers) {
  409. UA_LOG_INFO(server->config.logger, UA_LOGCATEGORY_SERVER,
  410. "Shutting down %u worker thread(s)", server->config.nThreads);
  411. /* Wait for all worker threads to finish */
  412. for(size_t i = 0; i < server->config.nThreads; ++i)
  413. server->workers[i].running = false;
  414. pthread_cond_broadcast(&server->dispatchQueue_condition);
  415. for(size_t i = 0; i < server->config.nThreads; ++i)
  416. pthread_join(server->workers[i].thr, NULL);
  417. /* Free the worker structures */
  418. UA_free(server->workers);
  419. server->workers = NULL;
  420. }
  421. /* Manually finish the work still enqueued */
  422. emptyDispatchQueue(server);
  423. UA_ASSERT_RCU_UNLOCKED();
  424. rcu_barrier(); // wait for all scheduled call_rcu work to complete
  425. #else
  426. processDelayedCallbacks(server);
  427. #endif
  428. return UA_STATUSCODE_GOOD;
  429. }
  430. UA_StatusCode UA_Server_run(UA_Server *server, volatile UA_Boolean *running) {
  431. UA_StatusCode retval = UA_Server_run_startup(server);
  432. if(retval != UA_STATUSCODE_GOOD)
  433. return retval;
  434. while(*running)
  435. UA_Server_run_iterate(server, true);
  436. return UA_Server_run_shutdown(server);
  437. }