ua_server_worker.c 25 KB

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