ua_server_worker.c 25 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700
  1. #include "ua_util.h"
  2. #include "ua_server_internal.h"
  3. /**
  4. * There are four types of job execution:
  5. *
  6. * 1. Normal jobs (dispatched to worker threads if multithreading is activated)
  7. *
  8. * 2. Repeated jobs with a repetition interval (dispatched to worker threads)
  9. *
  10. * 3. Mainloop jobs are executed (once) from the mainloop and not in the worker threads. The server
  11. * contains a stack structure where all threads can add mainloop jobs for the next mainloop
  12. * iteration. This is used e.g. to trigger adding and removing repeated jobs without blocking the
  13. * mainloop.
  14. *
  15. * 4. Delayed jobs are executed once in a worker thread. But only when all normal jobs that were
  16. * dispatched earlier have been executed. This is achieved by a counter in the worker threads. We
  17. * compute from the counter if all previous jobs have finished. The delay can be very long, since we
  18. * try to not interfere too much with normal execution. A use case is to eventually free obsolete
  19. * structures that _could_ still be accessed from concurrent threads.
  20. *
  21. * - Remove the entry from the list
  22. * - mark it as "dead" with an atomic operation
  23. * - add a delayed job that frees the memory when all concurrent operations have completed
  24. *
  25. * This approach to concurrently accessible memory is known as epoch based reclamation [1]. According to
  26. * [2], it performs competitively well on many-core systems. Our version of EBR does however not require
  27. * a global epoch. Instead, every worker thread has its own epoch counter that we observe for changes.
  28. *
  29. * [1] Fraser, K. 2003. Practical lock freedom. Ph.D. thesis. Computer Laboratory, University of Cambridge.
  30. * [2] Hart, T. E., McKenney, P. E., Brown, A. D., & Walpole, J. (2007). Performance of memory reclamation
  31. * for lockless synchronization. Journal of Parallel and Distributed Computing, 67(12), 1270-1285.
  32. *
  33. * Future Plans: Use work-stealing to load-balance between cores.
  34. * [3] Le, Nhat Minh, et al. "Correct and efficient work-stealing for weak
  35. * memory models." ACM SIGPLAN Notices. Vol. 48. No. 8. ACM, 2013.
  36. */
  37. #define MAXTIMEOUT 50 // max timeout in millisec until the next main loop iteration
  38. static void
  39. processJob(UA_Server *server, UA_Job *job) {
  40. UA_ASSERT_RCU_UNLOCKED();
  41. UA_RCU_LOCK();
  42. switch(job->type) {
  43. case UA_JOBTYPE_NOTHING:
  44. break;
  45. case UA_JOBTYPE_DETACHCONNECTION:
  46. UA_Connection_detachSecureChannel(job->job.closeConnection);
  47. break;
  48. case UA_JOBTYPE_BINARYMESSAGE_NETWORKLAYER:
  49. UA_Server_processBinaryMessage(server, job->job.binaryMessage.connection,
  50. &job->job.binaryMessage.message);
  51. UA_Connection *connection = job->job.binaryMessage.connection;
  52. connection->releaseRecvBuffer(connection, &job->job.binaryMessage.message);
  53. break;
  54. case UA_JOBTYPE_BINARYMESSAGE_ALLOCATED:
  55. UA_Server_processBinaryMessage(server, job->job.binaryMessage.connection,
  56. &job->job.binaryMessage.message);
  57. UA_ByteString_deleteMembers(&job->job.binaryMessage.message);
  58. break;
  59. case UA_JOBTYPE_METHODCALL:
  60. case UA_JOBTYPE_METHODCALL_DELAYED:
  61. job->job.methodCall.method(server, job->job.methodCall.data);
  62. break;
  63. default:
  64. UA_LOG_WARNING(server->config.logger, UA_LOGCATEGORY_SERVER,
  65. "Trying to execute a job of unknown type");
  66. break;
  67. }
  68. UA_RCU_UNLOCK();
  69. }
  70. /*******************************/
  71. /* Worker Threads and Dispatch */
  72. /*******************************/
  73. #ifdef UA_ENABLE_MULTITHREADING
  74. struct MainLoopJob {
  75. struct cds_lfs_node node;
  76. UA_Job job;
  77. };
  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. 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. return NULL;
  108. }
  109. static void
  110. dispatchJob(UA_Server *server, const UA_Job *job) {
  111. struct DispatchJob *dj = UA_malloc(sizeof(struct DispatchJob));
  112. dj->job = *job;
  113. cds_wfcq_node_init(&dj->node);
  114. cds_wfcq_enqueue(&server->dispatchQueue_head, &server->dispatchQueue_tail, &dj->node);
  115. }
  116. static void
  117. emptyDispatchQueue(UA_Server *server) {
  118. while(!cds_wfcq_empty(&server->dispatchQueue_head, &server->dispatchQueue_tail)) {
  119. struct DispatchJob *dj = (struct DispatchJob*)
  120. cds_wfcq_dequeue_blocking(&server->dispatchQueue_head, &server->dispatchQueue_tail);
  121. processJob(server, &dj->job);
  122. UA_free(dj);
  123. }
  124. }
  125. #endif
  126. /*****************/
  127. /* Repeated Jobs */
  128. /*****************/
  129. /* The linked list of jobs is sorted according to the next execution timestamp */
  130. struct RepeatedJob {
  131. LIST_ENTRY(RepeatedJob) next; /* Next element in the list */
  132. UA_DateTime nextTime; /* The next time when the jobs are to be executed */
  133. UA_UInt64 interval; /* Interval in 100ns resolution */
  134. UA_Guid id; /* Id of the repeated job */
  135. UA_Job job; /* The job description itself */
  136. };
  137. /* internal. call only from the main loop. */
  138. static void
  139. addRepeatedJob(UA_Server *server, struct RepeatedJob * UA_RESTRICT rj) {
  140. /* Search for the best position on the repeatedJobs sorted list. The goal is
  141. * to have many repeated jobs with the same repetition interval in a
  142. * "block". This helps to reduce the (linear) search to find the next entry
  143. * in the repeatedJobs list when dispatching the repeated jobs.
  144. * For this, we search between "nexttime_max - 1s" and "nexttime_max" for
  145. * entries with the same repetition interval and adjust the "nexttime".
  146. * Otherwise, add entry after the first element before "nexttime_max". */
  147. UA_DateTime nextTime_max = UA_DateTime_nowMonotonic() + (UA_Int64) rj->interval;
  148. struct RepeatedJob *afterRj = NULL;
  149. struct RepeatedJob *tmpRj;
  150. LIST_FOREACH(tmpRj, &server->repeatedJobs, next) {
  151. if(tmpRj->nextTime >= nextTime_max)
  152. break;
  153. if(tmpRj->interval == rj->interval &&
  154. tmpRj->nextTime > (nextTime_max - UA_SEC_TO_DATETIME))
  155. nextTime_max = tmpRj->nextTime; /* break in the next iteration */
  156. afterRj = tmpRj;
  157. }
  158. /* add the repeated job */
  159. rj->nextTime = nextTime_max;
  160. if(afterRj)
  161. LIST_INSERT_AFTER(afterRj, rj, next);
  162. else
  163. LIST_INSERT_HEAD(&server->repeatedJobs, rj, next);
  164. }
  165. UA_StatusCode
  166. UA_Server_addRepeatedJob(UA_Server *server, UA_Job job,
  167. UA_UInt32 interval, UA_Guid *jobId) {
  168. /* the interval needs to be at least 5ms */
  169. if(interval < 5)
  170. return UA_STATUSCODE_BADINTERNALERROR;
  171. UA_UInt64 interval_dt =
  172. (UA_UInt64)interval * (UA_UInt64)UA_MSEC_TO_DATETIME; // from ms to 100ns resolution
  173. /* Create and fill the repeated job structure */
  174. struct RepeatedJob *rj = UA_malloc(sizeof(struct RepeatedJob));
  175. if(!rj)
  176. return UA_STATUSCODE_BADOUTOFMEMORY;
  177. /* done inside addRepeatedJob:
  178. * rj->nextTime = UA_DateTime_nowMonotonic() + interval_dt; */
  179. rj->interval = interval_dt;
  180. rj->id = UA_Guid_random();
  181. rj->job = job;
  182. #ifdef UA_ENABLE_MULTITHREADING
  183. /* Call addRepeatedJob from the main loop */
  184. struct MainLoopJob *mlw = UA_malloc(sizeof(struct MainLoopJob));
  185. if(!mlw) {
  186. UA_free(rj);
  187. return UA_STATUSCODE_BADOUTOFMEMORY;
  188. }
  189. mlw->job = (UA_Job) {
  190. .type = UA_JOBTYPE_METHODCALL,
  191. .job.methodCall = {.data = rj, .method = (void (*)(UA_Server*, void*))addRepeatedJob}};
  192. cds_lfs_push(&server->mainLoopJobs, &mlw->node);
  193. #else
  194. /* Add directly */
  195. addRepeatedJob(server, rj);
  196. #endif
  197. if(jobId)
  198. *jobId = rj->id;
  199. return UA_STATUSCODE_GOOD;
  200. }
  201. /* - Dispatches all repeated jobs that have timed out
  202. * - Reinserts dispatched job at their new position in the sorted list
  203. * - Returns the next datetime when a repeated job is scheduled */
  204. static UA_DateTime
  205. processRepeatedJobs(UA_Server *server, UA_DateTime current, UA_Boolean *dispatched) {
  206. /* Find the last job that is executed in this iteration */
  207. struct RepeatedJob *lastNow = NULL, *tmp;
  208. LIST_FOREACH(tmp, &server->repeatedJobs, next) {
  209. if(tmp->nextTime > current)
  210. break;
  211. lastNow = tmp;
  212. }
  213. /* Keep pointer to the previously dispatched job to avoid linear search for
  214. * "batched" jobs with the same nexttime and interval */
  215. struct RepeatedJob tmp_last;
  216. tmp_last.nextTime = current-1; /* never matches. just to avoid if(last_added && ...) */
  217. struct RepeatedJob *last_dispatched = &tmp_last;
  218. /* Iterate over the list of elements (sorted according to the nextTime timestamp) */
  219. struct RepeatedJob *rj, *tmp_rj;
  220. LIST_FOREACH_SAFE(rj, &server->repeatedJobs, next, tmp_rj) {
  221. if(rj->nextTime > current)
  222. break;
  223. /* Dispatch/process job */
  224. #ifdef UA_ENABLE_MULTITHREADING
  225. dispatchJob(server, &rj->job);
  226. *dispatched = true;
  227. #else
  228. struct RepeatedJob **previousNext = rj->next.le_prev;
  229. processJob(server, &rj->job);
  230. /* See if the current job was deleted during processJob. That means the
  231. * le_next field of the previous repeated job (could also be the list
  232. * head) does no longer point to the current repeated job */
  233. if((void*)*previousNext != (void*)rj) {
  234. UA_LOG_DEBUG(server->config.logger, UA_LOGCATEGORY_SERVER,
  235. "The current repeated job removed itself");
  236. continue;
  237. }
  238. #endif
  239. /* Set the time for the next execution */
  240. rj->nextTime += (UA_Int64)rj->interval;
  241. /* Prevent an infinite loop when the repeated jobs took more time than
  242. * rj->interval */
  243. if(rj->nextTime < current)
  244. rj->nextTime = current + 1;
  245. /* Find new position for rj to keep the list sorted */
  246. struct RepeatedJob *prev_rj;
  247. if(last_dispatched->nextTime == rj->nextTime) {
  248. /* We "batch" repeatedJobs with the same interval in
  249. * addRepeatedJobs. So this might occur quite often. */
  250. UA_assert(last_dispatched != &tmp_last);
  251. prev_rj = last_dispatched;
  252. } else {
  253. /* Find the position by a linear search starting at the first
  254. * possible job */
  255. UA_assert(lastNow); /* Not NULL. Otherwise, we never reach this point. */
  256. prev_rj = lastNow;
  257. while(true) {
  258. struct RepeatedJob *n = LIST_NEXT(prev_rj, next);
  259. if(!n || n->nextTime >= rj->nextTime)
  260. break;
  261. prev_rj = n;
  262. }
  263. }
  264. /* Add entry */
  265. if(prev_rj != rj) {
  266. LIST_REMOVE(rj, next);
  267. LIST_INSERT_AFTER(prev_rj, rj, next);
  268. }
  269. /* Update last_dispatched and loop */
  270. last_dispatched = rj;
  271. }
  272. /* Check if the next repeated job is sooner than the usual timeout */
  273. struct RepeatedJob *first = LIST_FIRST(&server->repeatedJobs);
  274. UA_DateTime next = current + (MAXTIMEOUT * UA_MSEC_TO_DATETIME);
  275. if(first && first->nextTime < next)
  276. next = first->nextTime;
  277. return next;
  278. }
  279. /* Call this function only from the main loop! */
  280. static void
  281. removeRepeatedJob(UA_Server *server, UA_Guid *jobId) {
  282. struct RepeatedJob *rj;
  283. LIST_FOREACH(rj, &server->repeatedJobs, next) {
  284. if(!UA_Guid_equal(jobId, &rj->id))
  285. continue;
  286. LIST_REMOVE(rj, next);
  287. UA_free(rj);
  288. break;
  289. }
  290. #ifdef UA_ENABLE_MULTITHREADING
  291. UA_free(jobId);
  292. #endif
  293. }
  294. UA_StatusCode UA_Server_removeRepeatedJob(UA_Server *server, UA_Guid jobId) {
  295. #ifdef UA_ENABLE_MULTITHREADING
  296. UA_Guid *idptr = UA_malloc(sizeof(UA_Guid));
  297. if(!idptr)
  298. return UA_STATUSCODE_BADOUTOFMEMORY;
  299. *idptr = jobId;
  300. // dispatch to the mainloopjobs stack
  301. struct MainLoopJob *mlw = UA_malloc(sizeof(struct MainLoopJob));
  302. mlw->job = (UA_Job) {
  303. .type = UA_JOBTYPE_METHODCALL,
  304. .job.methodCall = {.data = idptr, .method = (void (*)(UA_Server*, void*))removeRepeatedJob}};
  305. cds_lfs_push(&server->mainLoopJobs, &mlw->node);
  306. #else
  307. removeRepeatedJob(server, &jobId);
  308. #endif
  309. return UA_STATUSCODE_GOOD;
  310. }
  311. void UA_Server_deleteAllRepeatedJobs(UA_Server *server) {
  312. struct RepeatedJob *current, *temp;
  313. LIST_FOREACH_SAFE(current, &server->repeatedJobs, next, temp) {
  314. LIST_REMOVE(current, next);
  315. UA_free(current);
  316. }
  317. }
  318. /****************/
  319. /* Delayed Jobs */
  320. /****************/
  321. #ifndef UA_ENABLE_MULTITHREADING
  322. typedef struct UA_DelayedJob {
  323. SLIST_ENTRY(UA_DelayedJob) next;
  324. UA_Job job;
  325. } UA_DelayedJob;
  326. UA_StatusCode
  327. UA_Server_delayedCallback(UA_Server *server, UA_ServerCallback callback, void *data) {
  328. UA_DelayedJob *dj = UA_malloc(sizeof(UA_DelayedJob));
  329. if(!dj)
  330. return UA_STATUSCODE_BADOUTOFMEMORY;
  331. dj->job.type = UA_JOBTYPE_METHODCALL;
  332. dj->job.job.methodCall.data = data;
  333. dj->job.job.methodCall.method = callback;
  334. SLIST_INSERT_HEAD(&server->delayedCallbacks, dj, next);
  335. return UA_STATUSCODE_GOOD;
  336. }
  337. static void
  338. processDelayedCallbacks(UA_Server *server) {
  339. UA_DelayedJob *dj, *dj_tmp;
  340. SLIST_FOREACH_SAFE(dj, &server->delayedCallbacks, next, dj_tmp) {
  341. SLIST_REMOVE(&server->delayedCallbacks, dj, UA_DelayedJob, next);
  342. processJob(server, &dj->job);
  343. UA_free(dj);
  344. }
  345. }
  346. #else
  347. #define DELAYEDJOBSSIZE 100 // Collect delayed jobs until we have DELAYEDWORKSIZE items
  348. struct DelayedJobs {
  349. struct DelayedJobs *next;
  350. UA_UInt32 *workerCounters; // initially NULL until the counter are set
  351. UA_UInt32 jobsCount; // the size of the array is DELAYEDJOBSSIZE, the count may be less
  352. UA_Job jobs[DELAYEDJOBSSIZE]; // when it runs full, a new delayedJobs entry is created
  353. };
  354. /* Dispatched as an ordinary job when the DelayedJobs list is full */
  355. static void getCounters(UA_Server *server, struct DelayedJobs *delayed) {
  356. UA_UInt32 *counters = UA_malloc(server->config.nThreads * sizeof(UA_UInt32));
  357. for(UA_UInt16 i = 0; i < server->config.nThreads; ++i)
  358. counters[i] = server->workers[i].counter;
  359. delayed->workerCounters = counters;
  360. }
  361. /* Call from the main thread only. This is the only function that modifies */
  362. /* server->delayedWork. processDelayedWorkQueue modifies the "next" (after the */
  363. /* head). */
  364. static void addDelayedJob(UA_Server *server, UA_Job *job) {
  365. struct DelayedJobs *dj = server->delayedJobs;
  366. if(!dj || dj->jobsCount >= DELAYEDJOBSSIZE) {
  367. /* create a new DelayedJobs and add it to the linked list */
  368. dj = UA_malloc(sizeof(struct DelayedJobs));
  369. if(!dj) {
  370. UA_LOG_ERROR(server->config.logger, UA_LOGCATEGORY_SERVER,
  371. "Not enough memory to add a delayed job");
  372. return;
  373. }
  374. dj->jobsCount = 0;
  375. dj->workerCounters = NULL;
  376. dj->next = server->delayedJobs;
  377. server->delayedJobs = dj;
  378. /* dispatch a method that sets the counter for the full list that comes afterwards */
  379. if(dj->next) {
  380. UA_Job setCounter = (UA_Job){
  381. .type = UA_JOBTYPE_METHODCALL, .job.methodCall =
  382. {.method = (void (*)(UA_Server*, void*))getCounters, .data = dj->next}};
  383. dispatchJob(server, &setCounter);
  384. }
  385. }
  386. dj->jobs[dj->jobsCount] = *job;
  387. ++dj->jobsCount;
  388. }
  389. static void
  390. delayed_free(UA_Server *server, void *data) {
  391. UA_free(data);
  392. }
  393. UA_StatusCode UA_Server_delayedFree(UA_Server *server, void *data) {
  394. return UA_Server_delayedCallback(server, delayed_free, data);
  395. }
  396. static void
  397. addDelayedJobAsync(UA_Server *server, UA_Job *job) {
  398. addDelayedJob(server, job);
  399. UA_free(job);
  400. }
  401. UA_StatusCode
  402. UA_Server_delayedCallback(UA_Server *server, UA_ServerCallback callback, void *data) {
  403. UA_Job *j = UA_malloc(sizeof(UA_Job));
  404. if(!j)
  405. return UA_STATUSCODE_BADOUTOFMEMORY;
  406. j->type = UA_JOBTYPE_METHODCALL;
  407. j->job.methodCall.data = data;
  408. j->job.methodCall.method = callback;
  409. struct MainLoopJob *mlw = UA_malloc(sizeof(struct MainLoopJob));
  410. mlw->job = (UA_Job) {.type = UA_JOBTYPE_METHODCALL, .job.methodCall =
  411. {.data = j, .method = (UA_ServerCallback)addDelayedJobAsync}};
  412. cds_lfs_push(&server->mainLoopJobs, &mlw->node);
  413. return UA_STATUSCODE_GOOD;
  414. }
  415. /* Find out which delayed jobs can be executed now */
  416. static void
  417. dispatchDelayedJobs(UA_Server *server, void *_) {
  418. /* start at the second */
  419. struct DelayedJobs *dw = server->delayedJobs, *beforedw = dw;
  420. if(dw)
  421. dw = dw->next;
  422. /* find the first delayedwork where the counters have been set and have moved */
  423. while(dw) {
  424. if(!dw->workerCounters) {
  425. beforedw = dw;
  426. dw = dw->next;
  427. continue;
  428. }
  429. UA_Boolean allMoved = true;
  430. for(size_t i = 0; i < server->config.nThreads; ++i) {
  431. if(dw->workerCounters[i] == server->workers[i].counter) {
  432. allMoved = false;
  433. break;
  434. }
  435. }
  436. if(allMoved)
  437. break;
  438. beforedw = dw;
  439. dw = dw->next;
  440. }
  441. /* process and free all delayed jobs from here on */
  442. while(dw) {
  443. for(size_t i = 0; i < dw->jobsCount; ++i)
  444. processJob(server, &dw->jobs[i]);
  445. struct DelayedJobs *next = UA_atomic_xchg((void**)&beforedw->next, NULL);
  446. UA_free(dw->workerCounters);
  447. UA_free(dw);
  448. dw = next;
  449. }
  450. }
  451. #endif
  452. /********************/
  453. /* Main Server Loop */
  454. /********************/
  455. #ifdef UA_ENABLE_MULTITHREADING
  456. static void processMainLoopJobs(UA_Server *server) {
  457. /* no synchronization required if we only use push and pop_all */
  458. struct cds_lfs_head *head = __cds_lfs_pop_all(&server->mainLoopJobs);
  459. if(!head)
  460. return;
  461. struct MainLoopJob *mlw = (struct MainLoopJob*)&head->node;
  462. struct MainLoopJob *next;
  463. do {
  464. processJob(server, &mlw->job);
  465. next = (struct MainLoopJob*)mlw->node.next;
  466. UA_free(mlw);
  467. //cppcheck-suppress unreadVariable
  468. } while((mlw = next));
  469. }
  470. #endif
  471. UA_StatusCode UA_Server_run_startup(UA_Server *server) {
  472. #ifdef UA_ENABLE_MULTITHREADING
  473. /* Spin up the worker threads */
  474. UA_LOG_INFO(server->config.logger, UA_LOGCATEGORY_SERVER,
  475. "Spinning up %u worker thread(s)", server->config.nThreads);
  476. pthread_cond_init(&server->dispatchQueue_condition, 0);
  477. pthread_mutex_init(&server->dispatchQueue_mutex, 0);
  478. server->workers = UA_malloc(server->config.nThreads * sizeof(UA_Worker));
  479. if(!server->workers)
  480. return UA_STATUSCODE_BADOUTOFMEMORY;
  481. for(size_t i = 0; i < server->config.nThreads; ++i) {
  482. UA_Worker *worker = &server->workers[i];
  483. worker->server = server;
  484. worker->counter = 0;
  485. worker->running = true;
  486. pthread_create(&worker->thr, NULL, (void* (*)(void*))workerLoop, worker);
  487. }
  488. /* Try to execute delayed callbacks every 10 sec */
  489. UA_Job processDelayed = {.type = UA_JOBTYPE_METHODCALL,
  490. .job.methodCall = {.method = dispatchDelayedJobs, .data = NULL} };
  491. UA_Server_addRepeatedJob(server, processDelayed, 10000, NULL);
  492. #endif
  493. /* Start the networklayers */
  494. UA_StatusCode result = UA_STATUSCODE_GOOD;
  495. for(size_t i = 0; i < server->config.networkLayersSize; ++i) {
  496. UA_ServerNetworkLayer *nl = &server->config.networkLayers[i];
  497. result |= nl->start(nl, server->config.logger);
  498. }
  499. return result;
  500. }
  501. /* completeMessages is run synchronous on the jobs returned from the network
  502. layer, so that the order for processing TCP packets is never mixed up. */
  503. static void
  504. completeMessages(UA_Server *server, UA_Job *job) {
  505. UA_Boolean realloced = UA_FALSE;
  506. UA_StatusCode retval = UA_Connection_completeMessages(job->job.binaryMessage.connection,
  507. &job->job.binaryMessage.message, &realloced);
  508. if(retval != UA_STATUSCODE_GOOD) {
  509. if(retval == UA_STATUSCODE_BADOUTOFMEMORY)
  510. UA_LOG_WARNING(server->config.logger, UA_LOGCATEGORY_NETWORK,
  511. "Lost message(s) from Connection %i as memory could not be allocated",
  512. job->job.binaryMessage.connection->sockfd);
  513. else if(retval != UA_STATUSCODE_GOOD)
  514. UA_LOG_INFO(server->config.logger, UA_LOGCATEGORY_NETWORK,
  515. "Could not merge half-received messages on Connection %i with error 0x%08x",
  516. job->job.binaryMessage.connection->sockfd, retval);
  517. job->type = UA_JOBTYPE_NOTHING;
  518. return;
  519. }
  520. if(realloced)
  521. job->type = UA_JOBTYPE_BINARYMESSAGE_ALLOCATED;
  522. /* discard the job if message is empty - also no leak is possible here */
  523. if(job->job.binaryMessage.message.length == 0)
  524. job->type = UA_JOBTYPE_NOTHING;
  525. }
  526. UA_UInt16 UA_Server_run_iterate(UA_Server *server, UA_Boolean waitInternal) {
  527. #ifdef UA_ENABLE_MULTITHREADING
  528. /* Run work assigned for the main thread */
  529. processMainLoopJobs(server);
  530. #endif
  531. /* Process repeated work */
  532. UA_DateTime now = UA_DateTime_nowMonotonic();
  533. UA_Boolean dispatched = false; /* to wake up worker threads */
  534. UA_DateTime nextRepeated = processRepeatedJobs(server, now, &dispatched);
  535. UA_UInt16 timeout = 0;
  536. if(waitInternal)
  537. timeout = (UA_UInt16)((nextRepeated - now) / UA_MSEC_TO_DATETIME);
  538. /* Get work from the networklayer */
  539. for(size_t i = 0; i < server->config.networkLayersSize; ++i) {
  540. UA_ServerNetworkLayer *nl = &server->config.networkLayers[i];
  541. UA_Job *jobs;
  542. size_t jobsSize;
  543. /* only the last networklayer waits on the tieout */
  544. if(i == server->config.networkLayersSize-1)
  545. jobsSize = nl->getJobs(nl, &jobs, timeout);
  546. else
  547. jobsSize = nl->getJobs(nl, &jobs, 0);
  548. for(size_t k = 0; k < jobsSize; ++k) {
  549. #ifdef UA_ENABLE_MULTITHREADING
  550. /* Filter out delayed work */
  551. if(jobs[k].type == UA_JOBTYPE_METHODCALL_DELAYED) {
  552. addDelayedJob(server, &jobs[k]);
  553. jobs[k].type = UA_JOBTYPE_NOTHING;
  554. continue;
  555. }
  556. #endif
  557. /* Merge half-received messages */
  558. if(jobs[k].type == UA_JOBTYPE_BINARYMESSAGE_NETWORKLAYER)
  559. completeMessages(server, &jobs[k]);
  560. }
  561. /* Dispatch/process jobs */
  562. for(size_t j = 0; j < jobsSize; ++j) {
  563. #ifdef UA_ENABLE_MULTITHREADING
  564. dispatchJob(server, &jobs[j]);
  565. dispatched = true;
  566. #else
  567. processJob(server, &jobs[j]);
  568. #endif
  569. }
  570. #ifdef UA_ENABLE_MULTITHREADING
  571. /* Wake up worker threads */
  572. if(dispatched)
  573. pthread_cond_broadcast(&server->dispatchQueue_condition);
  574. #endif
  575. /* Clean up jobs list */
  576. if(jobsSize > 0)
  577. UA_free(jobs);
  578. }
  579. #ifndef UA_ENABLE_MULTITHREADING
  580. processDelayedCallbacks(server);
  581. #endif
  582. now = UA_DateTime_nowMonotonic();
  583. timeout = 0;
  584. if(nextRepeated > now)
  585. timeout = (UA_UInt16)((nextRepeated - now) / UA_MSEC_TO_DATETIME);
  586. return timeout;
  587. }
  588. UA_StatusCode UA_Server_run_shutdown(UA_Server *server) {
  589. for(size_t i = 0; i < server->config.networkLayersSize; ++i) {
  590. UA_ServerNetworkLayer *nl = &server->config.networkLayers[i];
  591. UA_Job *stopJobs;
  592. size_t stopJobsSize = nl->stop(nl, &stopJobs);
  593. for(size_t j = 0; j < stopJobsSize; ++j)
  594. processJob(server, &stopJobs[j]);
  595. UA_free(stopJobs);
  596. }
  597. #ifdef UA_ENABLE_MULTITHREADING
  598. UA_LOG_INFO(server->config.logger, UA_LOGCATEGORY_SERVER,
  599. "Shutting down %u worker thread(s)", server->config.nThreads);
  600. /* Wait for all worker threads to finish */
  601. for(size_t i = 0; i < server->config.nThreads; ++i)
  602. server->workers[i].running = false;
  603. pthread_cond_broadcast(&server->dispatchQueue_condition);
  604. for(size_t i = 0; i < server->config.nThreads; ++i)
  605. pthread_join(server->workers[i].thr, NULL);
  606. UA_free(server->workers);
  607. /* Manually finish the work still enqueued.
  608. This especially contains delayed frees */
  609. emptyDispatchQueue(server);
  610. UA_ASSERT_RCU_UNLOCKED();
  611. rcu_barrier(); // wait for all scheduled call_rcu work to complete
  612. #else
  613. processDelayedCallbacks(server);
  614. #endif
  615. return UA_STATUSCODE_GOOD;
  616. }
  617. UA_StatusCode UA_Server_run(UA_Server *server, volatile UA_Boolean *running) {
  618. UA_StatusCode retval = UA_Server_run_startup(server);
  619. if(retval != UA_STATUSCODE_GOOD)
  620. return retval;
  621. while(*running)
  622. UA_Server_run_iterate(server, true);
  623. return UA_Server_run_shutdown(server);
  624. }