123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490 |
- /* This Source Code Form is subject to the terms of the Mozilla Public
- * License, v. 2.0. If a copy of the MPL was not distributed with this
- * file, You can obtain one at http://mozilla.org/MPL/2.0/. */
- #include "ua_util.h"
- #include "ua_server_internal.h"
- /**
- * There are four types of job execution:
- *
- * 1. Normal jobs (dispatched to worker threads if multithreading is activated)
- *
- * 2. Repeated jobs with a repetition interval (dispatched to worker threads)
- *
- * 3. Mainloop jobs are executed (once) from the mainloop and not in the worker threads. The server
- * contains a stack structure where all threads can add mainloop jobs for the next mainloop
- * iteration. This is used e.g. to trigger adding and removing repeated jobs without blocking the
- * mainloop.
- *
- * 4. Delayed jobs are executed once in a worker thread. But only when all normal jobs that were
- * dispatched earlier have been executed. This is achieved by a counter in the worker threads. We
- * compute from the counter if all previous jobs have finished. The delay can be very long, since we
- * try to not interfere too much with normal execution. A use case is to eventually free obsolete
- * structures that _could_ still be accessed from concurrent threads.
- *
- * - Remove the entry from the list
- * - mark it as "dead" with an atomic operation
- * - add a delayed job that frees the memory when all concurrent operations have completed
- *
- * This approach to concurrently accessible memory is known as epoch based reclamation [1]. According to
- * [2], it performs competitively well on many-core systems. Our version of EBR does however not require
- * a global epoch. Instead, every worker thread has its own epoch counter that we observe for changes.
- *
- * [1] Fraser, K. 2003. Practical lock freedom. Ph.D. thesis. Computer Laboratory, University of Cambridge.
- * [2] Hart, T. E., McKenney, P. E., Brown, A. D., & Walpole, J. (2007). Performance of memory reclamation
- * for lockless synchronization. Journal of Parallel and Distributed Computing, 67(12), 1270-1285.
- *
- * Future Plans: Use work-stealing to load-balance between cores.
- * [3] Le, Nhat Minh, et al. "Correct and efficient work-stealing for weak
- * memory models." ACM SIGPLAN Notices. Vol. 48. No. 8. ACM, 2013.
- */
- void
- UA_Server_processJob(UA_Server *server, UA_Job *job) {
- UA_ASSERT_RCU_UNLOCKED();
- UA_RCU_LOCK();
- switch(job->type) {
- case UA_JOBTYPE_NOTHING:
- break;
- case UA_JOBTYPE_DETACHCONNECTION:
- UA_Connection_detachSecureChannel(job->job.closeConnection);
- break;
- case UA_JOBTYPE_BINARYMESSAGE_NETWORKLAYER:
- {
- UA_Server_processBinaryMessage(server, job->job.binaryMessage.connection,
- &job->job.binaryMessage.message);
- UA_Connection *connection = job->job.binaryMessage.connection;
- connection->releaseRecvBuffer(connection, &job->job.binaryMessage.message);
- }
- break;
- case UA_JOBTYPE_BINARYMESSAGE_ALLOCATED:
- UA_Server_processBinaryMessage(server, job->job.binaryMessage.connection,
- &job->job.binaryMessage.message);
- UA_ByteString_deleteMembers(&job->job.binaryMessage.message);
- break;
- case UA_JOBTYPE_METHODCALL:
- case UA_JOBTYPE_METHODCALL_DELAYED:
- job->job.methodCall.method(server, job->job.methodCall.data);
- break;
- default:
- UA_LOG_WARNING(server->config.logger, UA_LOGCATEGORY_SERVER,
- "Trying to execute a job of unknown type");
- break;
- }
- UA_RCU_UNLOCK();
- }
- /*******************************/
- /* Worker Threads and Dispatch */
- /*******************************/
- #ifdef UA_ENABLE_MULTITHREADING
- struct DispatchJob {
- struct cds_wfcq_node node; // node for the queue
- UA_Job job;
- };
- static void *
- workerLoop(UA_Worker *worker) {
- UA_Server *server = worker->server;
- UA_UInt32 *counter = &worker->counter;
- volatile UA_Boolean *running = &worker->running;
- /* Initialize the (thread local) random seed with the ram address of worker */
- UA_random_seed((uintptr_t)worker);
- rcu_register_thread();
- while(*running) {
- struct DispatchJob *dj = (struct DispatchJob*)
- cds_wfcq_dequeue_blocking(&server->dispatchQueue_head, &server->dispatchQueue_tail);
- if(dj) {
- UA_Server_processJob(server, &dj->job);
- UA_free(dj);
- } else {
- /* nothing to do. sleep until a job is dispatched (and wakes up all worker threads) */
- pthread_mutex_lock(&server->dispatchQueue_mutex);
- pthread_cond_wait(&server->dispatchQueue_condition, &server->dispatchQueue_mutex);
- pthread_mutex_unlock(&server->dispatchQueue_mutex);
- }
- UA_atomic_add(counter, 1);
- }
- UA_ASSERT_RCU_UNLOCKED();
- rcu_barrier(); // wait for all scheduled call_rcu work to complete
- rcu_unregister_thread();
- UA_LOG_DEBUG(server->config.logger, UA_LOGCATEGORY_SERVER, "Worker shut down");
- return NULL;
- }
- void
- UA_Server_dispatchJob(UA_Server *server, const UA_Job *job) {
- struct DispatchJob *dj = UA_malloc(sizeof(struct DispatchJob));
- // todo: check malloc
- dj->job = *job;
- cds_wfcq_node_init(&dj->node);
- cds_wfcq_enqueue(&server->dispatchQueue_head, &server->dispatchQueue_tail, &dj->node);
- }
- static void
- emptyDispatchQueue(UA_Server *server) {
- while(!cds_wfcq_empty(&server->dispatchQueue_head, &server->dispatchQueue_tail)) {
- struct DispatchJob *dj = (struct DispatchJob*)
- cds_wfcq_dequeue_blocking(&server->dispatchQueue_head, &server->dispatchQueue_tail);
- UA_Server_processJob(server, &dj->job);
- UA_free(dj);
- }
- }
- #endif
- /****************/
- /* Delayed Jobs */
- /****************/
- #ifndef UA_ENABLE_MULTITHREADING
- typedef struct UA_DelayedJob {
- SLIST_ENTRY(UA_DelayedJob) next;
- UA_Job job;
- } UA_DelayedJob;
- UA_StatusCode
- UA_Server_delayedCallback(UA_Server *server, UA_ServerCallback callback, void *data) {
- UA_DelayedJob *dj = (UA_DelayedJob *)UA_malloc(sizeof(UA_DelayedJob));
- if(!dj)
- return UA_STATUSCODE_BADOUTOFMEMORY;
- dj->job.type = UA_JOBTYPE_METHODCALL;
- dj->job.job.methodCall.data = data;
- dj->job.job.methodCall.method = callback;
- SLIST_INSERT_HEAD(&server->delayedCallbacks, dj, next);
- return UA_STATUSCODE_GOOD;
- }
- static void
- processDelayedCallbacks(UA_Server *server) {
- UA_DelayedJob *dj, *dj_tmp;
- SLIST_FOREACH_SAFE(dj, &server->delayedCallbacks, next, dj_tmp) {
- SLIST_REMOVE(&server->delayedCallbacks, dj, UA_DelayedJob, next);
- UA_Server_processJob(server, &dj->job);
- UA_free(dj);
- }
- }
- #else
- #define DELAYEDJOBSSIZE 100 // Collect delayed jobs until we have DELAYEDWORKSIZE items
- struct DelayedJobs {
- struct DelayedJobs *next;
- UA_UInt32 *workerCounters; // initially NULL until the counter are set
- UA_UInt32 jobsCount; // the size of the array is DELAYEDJOBSSIZE, the count may be less
- UA_Job jobs[DELAYEDJOBSSIZE]; // when it runs full, a new delayedJobs entry is created
- };
- /* Dispatched as an ordinary job when the DelayedJobs list is full */
- static void getCounters(UA_Server *server, struct DelayedJobs *delayed) {
- UA_UInt32 *counters = UA_malloc(server->config.nThreads * sizeof(UA_UInt32));
- for(UA_UInt16 i = 0; i < server->config.nThreads; ++i)
- counters[i] = server->workers[i].counter;
- delayed->workerCounters = counters;
- }
- /* Call from the main thread only. This is the only function that modifies */
- /* server->delayedWork. processDelayedWorkQueue modifies the "next" (after the */
- /* head). */
- static void addDelayedJob(UA_Server *server, UA_Job *job) {
- struct DelayedJobs *dj = server->delayedJobs;
- if(!dj || dj->jobsCount >= DELAYEDJOBSSIZE) {
- /* create a new DelayedJobs and add it to the linked list */
- dj = UA_malloc(sizeof(struct DelayedJobs));
- if(!dj) {
- UA_LOG_ERROR(server->config.logger, UA_LOGCATEGORY_SERVER,
- "Not enough memory to add a delayed job");
- return;
- }
- dj->jobsCount = 0;
- dj->workerCounters = NULL;
- dj->next = server->delayedJobs;
- server->delayedJobs = dj;
- /* dispatch a method that sets the counter for the full list that comes afterwards */
- if(dj->next) {
- UA_Job setCounter = (UA_Job){
- .type = UA_JOBTYPE_METHODCALL, .job.methodCall =
- {.method = (void (*)(UA_Server*, void*))getCounters, .data = dj->next}};
- UA_Server_dispatchJob(server, &setCounter);
- }
- }
- dj->jobs[dj->jobsCount] = *job;
- ++dj->jobsCount;
- }
- static void
- delayed_free(UA_Server *server, void *data) {
- UA_free(data);
- }
- UA_StatusCode UA_Server_delayedFree(UA_Server *server, void *data) {
- return UA_Server_delayedCallback(server, delayed_free, data);
- }
- static void
- addDelayedJobAsync(UA_Server *server, UA_Job *job) {
- addDelayedJob(server, job);
- UA_free(job);
- }
- UA_StatusCode
- UA_Server_delayedCallback(UA_Server *server, UA_ServerCallback callback, void *data) {
- UA_Job *j = UA_malloc(sizeof(UA_Job));
- if(!j)
- return UA_STATUSCODE_BADOUTOFMEMORY;
- j->type = UA_JOBTYPE_METHODCALL;
- j->job.methodCall.data = data;
- j->job.methodCall.method = callback;
- struct MainLoopJob *mlw = UA_malloc(sizeof(struct MainLoopJob));
- mlw->job = (UA_Job) {.type = UA_JOBTYPE_METHODCALL, .job.methodCall =
- {.data = j, .method = (UA_ServerCallback)addDelayedJobAsync}};
- cds_lfs_push(&server->mainLoopJobs, &mlw->node);
- return UA_STATUSCODE_GOOD;
- }
- /* Find out which delayed jobs can be executed now */
- static void
- dispatchDelayedJobs(UA_Server *server, void *_) {
- /* start at the second */
- struct DelayedJobs *dw = server->delayedJobs, *beforedw = dw;
- if(dw)
- dw = dw->next;
- /* find the first delayedwork where the counters have been set and have moved */
- while(dw) {
- if(!dw->workerCounters) {
- beforedw = dw;
- dw = dw->next;
- continue;
- }
- UA_Boolean allMoved = true;
- for(size_t i = 0; i < server->config.nThreads; ++i) {
- if(dw->workerCounters[i] == server->workers[i].counter) {
- allMoved = false;
- break;
- }
- }
- if(allMoved)
- break;
- beforedw = dw;
- dw = dw->next;
- }
- /* process and free all delayed jobs from here on */
- while(dw) {
- for(size_t i = 0; i < dw->jobsCount; ++i)
- UA_Server_processJob(server, &dw->jobs[i]);
- struct DelayedJobs *next = UA_atomic_xchg((void**)&beforedw->next, NULL);
- UA_free(dw->workerCounters);
- UA_free(dw);
- dw = next;
- }
- }
- #endif
- /********************/
- /* Main Server Loop */
- /********************/
- #ifdef UA_ENABLE_MULTITHREADING
- static void processMainLoopJobs(UA_Server *server) {
- /* no synchronization required if we only use push and pop_all */
- struct cds_lfs_head *head = __cds_lfs_pop_all(&server->mainLoopJobs);
- if(!head)
- return;
- struct MainLoopJob *mlw = (struct MainLoopJob*)&head->node;
- struct MainLoopJob *next;
- do {
- UA_Server_processJob(server, &mlw->job);
- next = (struct MainLoopJob*)mlw->node.next;
- UA_free(mlw);
- //cppcheck-suppress unreadVariable
- } while((mlw = next));
- }
- #endif
- UA_StatusCode UA_Server_run_startup(UA_Server *server) {
- #ifdef UA_ENABLE_MULTITHREADING
- /* Spin up the worker threads */
- UA_LOG_INFO(server->config.logger, UA_LOGCATEGORY_SERVER,
- "Spinning up %u worker thread(s)", server->config.nThreads);
- pthread_cond_init(&server->dispatchQueue_condition, 0);
- pthread_mutex_init(&server->dispatchQueue_mutex, 0);
- server->workers = UA_malloc(server->config.nThreads * sizeof(UA_Worker));
- if(!server->workers)
- return UA_STATUSCODE_BADOUTOFMEMORY;
- for(size_t i = 0; i < server->config.nThreads; ++i) {
- UA_Worker *worker = &server->workers[i];
- worker->server = server;
- worker->counter = 0;
- worker->running = true;
- pthread_create(&worker->thr, NULL, (void* (*)(void*))workerLoop, worker);
- }
- /* Try to execute delayed callbacks every 10 sec */
- UA_Job processDelayed = {.type = UA_JOBTYPE_METHODCALL,
- .job.methodCall = {.method = dispatchDelayedJobs, .data = NULL} };
- UA_Server_addRepeatedJob(server, processDelayed, 10000, NULL);
- #endif
- /* Start the networklayers */
- UA_StatusCode result = UA_STATUSCODE_GOOD;
- for(size_t i = 0; i < server->config.networkLayersSize; ++i) {
- UA_ServerNetworkLayer *nl = &server->config.networkLayers[i];
- result |= nl->start(nl, server->config.logger);
- }
- return result;
- }
- /* completeMessages is run synchronous on the jobs returned from the network
- layer, so that the order for processing TCP packets is never mixed up. */
- static void
- completeMessages(UA_Server *server, UA_Job *job) {
- UA_Boolean realloced = UA_FALSE;
- UA_StatusCode retval = UA_Connection_completeMessages(job->job.binaryMessage.connection,
- &job->job.binaryMessage.message, &realloced);
- if(retval != UA_STATUSCODE_GOOD) {
- if(retval == UA_STATUSCODE_BADOUTOFMEMORY)
- UA_LOG_WARNING(server->config.logger, UA_LOGCATEGORY_NETWORK,
- "Lost message(s) from Connection %i as memory could not be allocated",
- job->job.binaryMessage.connection->sockfd);
- else if(retval != UA_STATUSCODE_GOOD)
- UA_LOG_INFO(server->config.logger, UA_LOGCATEGORY_NETWORK,
- "Could not merge half-received messages on Connection %i with error 0x%08x",
- job->job.binaryMessage.connection->sockfd, retval);
- job->type = UA_JOBTYPE_NOTHING;
- return;
- }
- if(realloced)
- job->type = UA_JOBTYPE_BINARYMESSAGE_ALLOCATED;
- /* discard the job if message is empty - also no leak is possible here */
- if(job->job.binaryMessage.message.length == 0)
- job->type = UA_JOBTYPE_NOTHING;
- }
- UA_UInt16 UA_Server_run_iterate(UA_Server *server, UA_Boolean waitInternal) {
- #ifdef UA_ENABLE_MULTITHREADING
- /* Run work assigned for the main thread */
- processMainLoopJobs(server);
- #endif
- /* Process repeated work */
- UA_DateTime now = UA_DateTime_nowMonotonic();
- UA_Boolean dispatched = false; /* to wake up worker threads */
- UA_DateTime nextRepeated = UA_Server_processRepeatedJobs(server, now, &dispatched);
- UA_UInt16 timeout = 0;
- if(waitInternal)
- timeout = (UA_UInt16)((nextRepeated - now) / UA_MSEC_TO_DATETIME);
- /* Get work from the networklayer */
- for(size_t i = 0; i < server->config.networkLayersSize; ++i) {
- UA_ServerNetworkLayer *nl = &server->config.networkLayers[i];
- UA_Job *jobs = NULL;
- size_t jobsSize;
- /* only the last networklayer waits on the tieout */
- if(i == server->config.networkLayersSize-1)
- jobsSize = nl->getJobs(nl, &jobs, timeout);
- else
- jobsSize = nl->getJobs(nl, &jobs, 0);
- for(size_t k = 0; k < jobsSize; ++k) {
- #ifdef UA_ENABLE_MULTITHREADING
- /* Filter out delayed work */
- if(jobs[k].type == UA_JOBTYPE_METHODCALL_DELAYED) {
- addDelayedJob(server, &jobs[k]);
- jobs[k].type = UA_JOBTYPE_NOTHING;
- continue;
- }
- #endif
- /* Merge half-received messages */
- if(jobs[k].type == UA_JOBTYPE_BINARYMESSAGE_NETWORKLAYER)
- completeMessages(server, &jobs[k]);
- }
- /* Dispatch/process jobs */
- for(size_t j = 0; j < jobsSize; ++j) {
- #ifdef UA_ENABLE_MULTITHREADING
- UA_Server_dispatchJob(server, &jobs[j]);
- dispatched = true;
- #else
- UA_Server_processJob(server, &jobs[j]);
- #endif
- }
- /* Clean up jobs list */
- if(jobsSize > 0)
- UA_free(jobs);
- }
- #ifdef UA_ENABLE_MULTITHREADING
- /* Wake up worker threads */
- if(dispatched)
- pthread_cond_broadcast(&server->dispatchQueue_condition);
- #else
- processDelayedCallbacks(server);
- #endif
- now = UA_DateTime_nowMonotonic();
- timeout = 0;
- if(nextRepeated > now)
- timeout = (UA_UInt16)((nextRepeated - now) / UA_MSEC_TO_DATETIME);
- return timeout;
- }
- UA_StatusCode UA_Server_run_shutdown(UA_Server *server) {
- for(size_t i = 0; i < server->config.networkLayersSize; ++i) {
- UA_ServerNetworkLayer *nl = &server->config.networkLayers[i];
- UA_Job *stopJobs = NULL;
- size_t stopJobsSize = nl->stop(nl, &stopJobs);
- for(size_t j = 0; j < stopJobsSize; ++j)
- UA_Server_processJob(server, &stopJobs[j]);
- UA_free(stopJobs);
- }
- #ifdef UA_ENABLE_MULTITHREADING
- /* Ensure that run_shutdown can be called multiple times */
- if(server->workers) {
- UA_LOG_INFO(server->config.logger, UA_LOGCATEGORY_SERVER,
- "Shutting down %u worker thread(s)", server->config.nThreads);
- /* Wait for all worker threads to finish */
- for(size_t i = 0; i < server->config.nThreads; ++i)
- server->workers[i].running = false;
- pthread_cond_broadcast(&server->dispatchQueue_condition);
- for(size_t i = 0; i < server->config.nThreads; ++i)
- pthread_join(server->workers[i].thr, NULL);
- /* Free the worker structures */
- UA_free(server->workers);
- server->workers = NULL;
- }
- /* Manually finish the work still enqueued */
- emptyDispatchQueue(server);
- UA_ASSERT_RCU_UNLOCKED();
- rcu_barrier(); // wait for all scheduled call_rcu work to complete
- #else
- processDelayedCallbacks(server);
- #endif
- return UA_STATUSCODE_GOOD;
- }
- UA_StatusCode UA_Server_run(UA_Server *server, volatile UA_Boolean *running) {
- UA_StatusCode retval = UA_Server_run_startup(server);
- if(retval != UA_STATUSCODE_GOOD)
- return retval;
- while(*running)
- UA_Server_run_iterate(server, true);
- return UA_Server_run_shutdown(server);
- }
|