Browse Source

Server: Asynchronous Operations API / Implementation for methods

Klaus Schick 5 years ago
parent
commit
cc94b6aa0f

+ 5 - 1
CMakeLists.txt

@@ -609,7 +609,9 @@ set(internal_headers ${PROJECT_SOURCE_DIR}/deps/open62541_queue.h
                      ${PROJECT_SOURCE_DIR}/src/pubsub/ua_pubsub_ns0.h
                      ${PROJECT_SOURCE_DIR}/src/pubsub/ua_pubsub_ns0.h
                      ${PROJECT_SOURCE_DIR}/src/server/ua_server_internal.h
                      ${PROJECT_SOURCE_DIR}/src/server/ua_server_internal.h
                      ${PROJECT_SOURCE_DIR}/src/server/ua_services.h
                      ${PROJECT_SOURCE_DIR}/src/server/ua_services.h
-                     ${PROJECT_SOURCE_DIR}/src/client/ua_client_internal.h)
+                     ${PROJECT_SOURCE_DIR}/src/client/ua_client_internal.h
+                     ${PROJECT_SOURCE_DIR}/src/server/ua_server_methodqueue.h
+					 ${PROJECT_SOURCE_DIR}/src/server/ua_asyncmethod_manager.h)
 
 
 # TODO: make client optional
 # TODO: make client optional
 set(lib_sources ${PROJECT_SOURCE_DIR}/src/ua_types.c
 set(lib_sources ${PROJECT_SOURCE_DIR}/src/ua_types.c
@@ -636,6 +638,8 @@ set(lib_sources ${PROJECT_SOURCE_DIR}/src/ua_types.c
                 ${PROJECT_SOURCE_DIR}/src/pubsub/ua_pubsub.c
                 ${PROJECT_SOURCE_DIR}/src/pubsub/ua_pubsub.c
                 ${PROJECT_SOURCE_DIR}/src/pubsub/ua_pubsub_manager.c
                 ${PROJECT_SOURCE_DIR}/src/pubsub/ua_pubsub_manager.c
                 ${PROJECT_SOURCE_DIR}/src/pubsub/ua_pubsub_ns0.c
                 ${PROJECT_SOURCE_DIR}/src/pubsub/ua_pubsub_ns0.c
+                ${PROJECT_SOURCE_DIR}/src/server/ua_server_methodqueue.c
+				${PROJECT_SOURCE_DIR}/src/server/ua_asyncmethod_manager.c
                 # services
                 # services
                 ${PROJECT_SOURCE_DIR}/src/server/ua_services_view.c
                 ${PROJECT_SOURCE_DIR}/src/server/ua_services_view.c
                 ${PROJECT_SOURCE_DIR}/src/server/ua_services_method.c
                 ${PROJECT_SOURCE_DIR}/src/server/ua_services_method.c

+ 1 - 0
arch/posix/ua_architecture.h

@@ -149,6 +149,7 @@ extern void * (*UA_globalRealloc)(void *ptr, size_t size);
 
 
 #if UA_MULTITHREADING >= 100
 #if UA_MULTITHREADING >= 100
 #include <pthread.h>
 #include <pthread.h>
+#define Sleep(x) sleep(x / 1000)
 #define UA_LOCK_TYPE_NAME pthread_mutex_t
 #define UA_LOCK_TYPE_NAME pthread_mutex_t
 #define UA_LOCK_TYPE(mutexName) pthread_mutex_t mutexName; \
 #define UA_LOCK_TYPE(mutexName) pthread_mutex_t mutexName; \
                                         pthread_mutexattr_t mutexName##_attr; \
                                         pthread_mutexattr_t mutexName##_attr; \

+ 1 - 0
doc/building.rst

@@ -198,6 +198,7 @@ Detailed SDK Features
         - 0: Multithreading support disabled.
         - 0: Multithreading support disabled.
         - 100: Functions marked with the UA_THREADSAFE-macro are protected with a lock-based enhancement using mutexes.
         - 100: Functions marked with the UA_THREADSAFE-macro are protected with a lock-based enhancement using mutexes.
         Multiple threads are allowed to call these functions of the SDK at the same time without causing race conditions.
         Multiple threads are allowed to call these functions of the SDK at the same time without causing race conditions.
+        Furthermore, this level supports the feature of adding async methods to objects.
         - 200: Work is distributed to a number of worker threads. Those worker threads are created within the SDK.
         - 200: Work is distributed to a number of worker threads. Those worker threads are created within the SDK.
 
 
 **UA_ENABLE_IMMUTABLE_NODES**
 **UA_ENABLE_IMMUTABLE_NODES**

+ 9 - 0
examples/CMakeLists.txt

@@ -87,6 +87,9 @@ add_example(tutorial_server_object tutorial_server_object.c)
 
 
 if(UA_ENABLE_METHODCALLS)
 if(UA_ENABLE_METHODCALLS)
     add_example(tutorial_server_method tutorial_server_method.c)
     add_example(tutorial_server_method tutorial_server_method.c)
+    if (UA_MULTITHREADING EQUAL 100)
+        add_example(tutorial_server_method_async tutorial_server_method_async.c)
+    endif()
 endif()
 endif()
 
 
 add_example(tutorial_client_firststeps tutorial_client_firststeps.c)
 add_example(tutorial_client_firststeps tutorial_client_firststeps.c)
@@ -125,6 +128,12 @@ install(PROGRAMS $<TARGET_FILE:client>
 
 
 add_example(client_async client_async.c)
 add_example(client_async client_async.c)
 
 
+if(UA_ENABLE_METHODCALLS)
+    if (UA_MULTITHREADING EQUAL 100)
+        add_example(client_method_async client_method_async.c)
+    endif()
+endif()
+
 add_example(client_connect_loop client_connect_loop.c)
 add_example(client_connect_loop client_connect_loop.c)
 
 
 add_example(client_connectivitycheck_loop client_connectivitycheck_loop.c)
 add_example(client_connectivitycheck_loop client_connectivitycheck_loop.c)

+ 270 - 0
examples/client_method_async.c

@@ -0,0 +1,270 @@
+/* This work is licensed under a Creative Commons CCZero 1.0 Universal License.
+ * See http://creativecommons.org/publicdomain/zero/1.0/ for more information. */
+
+#include <open62541/client_config_default.h>
+#include <open62541/client_highlevel_async.h>
+#include <open62541/client_subscriptions.h>
+#include <open62541/plugin/log_stdout.h>
+#include <open62541/server_config_default.h>
+#include <open62541/client_subscriptions.h>
+
+#include <stdlib.h>
+#include <signal.h>
+
+UA_Boolean running = true;
+static void InitCallMulti(UA_Client* client);
+
+#ifdef UA_ENABLE_METHODCALLS
+
+static void
+methodCalled(UA_Client *client, void *userdata, UA_UInt32 requestId,
+    UA_CallResponse *response) {
+    UA_UInt32 i;
+    UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_USERLAND, "**** CallRequest Response - Req:%u with %u results",
+        requestId, (UA_UInt32)response->resultsSize);
+    UA_StatusCode retval = response->responseHeader.serviceResult;
+    if (retval == UA_STATUSCODE_GOOD) {
+        for (i = 0; i < response->resultsSize; i++) {
+            if (response->resultsSize >= i)
+                retval = response->results[i].statusCode;
+            else
+                retval = UA_STATUSCODE_BADUNEXPECTEDERROR;
+            if (retval != UA_STATUSCODE_GOOD) {
+                UA_CallResponse_clear(response);
+                UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_USERLAND, "**** CallRequest Response - Req: %u (%u) failed", requestId,i);
+                if (i == response->resultsSize)
+                    return;
+                else
+                    continue;
+            }
+
+            /* Move the output arguments */
+            UA_Variant *output = response->results[i].outputArguments;
+            size_t outputSize = response->results[i].outputArgumentsSize;
+            response->results[i].outputArguments = NULL;
+            response->results[i].outputArgumentsSize = 0;
+
+            if (retval == UA_STATUSCODE_GOOD) {
+                printf("---Method call was successful, returned %lu values.\n",
+                    (unsigned long)outputSize);
+                UA_Array_delete(output, outputSize, &UA_TYPES[UA_TYPES_VARIANT]);
+            }
+            else {
+                printf("---Method call was unsuccessful, returned %x values.\n",
+                    retval);
+            }
+        }
+    }
+    else
+    {
+        UA_LOG_ERROR(UA_Log_Stdout, UA_LOGCATEGORY_USERLAND, "**** CallRequest Response - Req:%u FAILED", requestId);
+    }
+    UA_CallResponse_clear(response);
+    
+    /* We initiate the MultiCall (2 methods within one CallRequest) */
+    InitCallMulti(client);
+}
+
+#ifdef UA_ENABLE_METHODCALLS
+/* Workaround because we do not have an API for that yet */
+
+static UA_StatusCode
+UA_Client_call_asyncMulti(UA_Client *client,
+    const UA_NodeId objectId1, const UA_NodeId methodId1, size_t inputSize1, const UA_Variant *input1,
+    const UA_NodeId objectId2, const UA_NodeId methodId2, size_t inputSize2, const UA_Variant *input2,
+    UA_ClientAsyncServiceCallback callback, void *userdata, UA_UInt32 *reqId) {
+    UA_CallRequest request;
+    UA_CallRequest_init(&request);
+    UA_CallMethodRequest item[2];
+    UA_CallMethodRequest_init(&item[0]);
+    item[0].methodId = methodId1;
+    item[0].objectId = objectId1;
+    item[0].inputArguments = (UA_Variant *)(void*)(uintptr_t)input1; // cast const...
+    item[0].inputArgumentsSize = inputSize1;
+
+    UA_CallMethodRequest_init(&item[1]);
+    item[1].methodId = methodId2;
+    item[1].objectId = objectId2;
+    item[1].inputArguments = (UA_Variant *)(void*)(uintptr_t)input2; // cast const...
+    item[1].inputArgumentsSize = inputSize2;
+
+    request.methodsToCall = &item[0];
+    request.methodsToCallSize = 2;
+
+    return __UA_Client_AsyncService(client, &request,
+        &UA_TYPES[UA_TYPES_CALLREQUEST], callback,
+        &UA_TYPES[UA_TYPES_CALLRESPONSE], userdata, reqId);
+}
+/* End Workaround */
+
+static void InitCallMulti(UA_Client* client) {
+    UA_UInt32 reqId = 0;
+    UA_Variant input;
+    UA_Variant_init(&input);
+    UA_String stringValue = UA_String_fromChars("World 3 (multi)");
+    UA_Variant_setScalar(&input, &stringValue, &UA_TYPES[UA_TYPES_STRING]);
+
+    UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_USERLAND, "**** Initiating CallRequest 3");
+    UA_Client_call_asyncMulti(client,
+        UA_NODEID_NUMERIC(0, UA_NS0ID_OBJECTSFOLDER),
+        UA_NODEID_NUMERIC(1, 62542), 1, &input,
+        UA_NODEID_NUMERIC(0, UA_NS0ID_OBJECTSFOLDER),
+        UA_NODEID_NUMERIC(1, 62541), 1, &input,
+        (UA_ClientAsyncServiceCallback)methodCalled, NULL, &reqId);
+    UA_String_clear(&stringValue);
+}
+#endif
+#endif /* UA_ENABLE_METHODCALLS */
+
+static void stopHandler(int sign) {
+    UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_USERLAND, "Received Ctrl-C");
+    running = 0;
+}
+
+static void
+handler_currentTimeChanged(UA_Client *client, UA_UInt32 subId, void *subContext,
+    UA_UInt32 monId, void *monContext, UA_DataValue *value) {
+    //UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_USERLAND, "currentTime has changed!");
+    if (UA_Variant_hasScalarType(&value->value, &UA_TYPES[UA_TYPES_DATETIME])) {
+        UA_DateTime raw_date = *(UA_DateTime *)value->value.data;
+        UA_DateTimeStruct dts = UA_DateTime_toStruct(raw_date);
+        UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_USERLAND,
+            "date is: %02u-%02u-%04u %02u:%02u:%02u.%03u",
+            dts.day, dts.month, dts.year, dts.hour, dts.min, dts.sec, dts.milliSec);
+    }
+}
+
+static void
+deleteSubscriptionCallback(UA_Client *client, UA_UInt32 subscriptionId, void *subscriptionContext) {
+    UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_USERLAND,
+        "Subscription Id %u was deleted", subscriptionId);
+}
+
+static void
+subscriptionInactivityCallback(UA_Client *client, UA_UInt32 subId, void *subContext) {
+    UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_USERLAND, "Inactivity for subscription %u", subId);
+}
+
+static void
+stateCallback(UA_Client *client, UA_ClientState clientState) {
+    switch (clientState) {
+    case UA_CLIENTSTATE_DISCONNECTED:
+        UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_USERLAND, "The client is disconnected");
+        break;
+    case UA_CLIENTSTATE_WAITING_FOR_ACK:
+        UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_USERLAND, "Waiting for ack");
+        break;
+    case UA_CLIENTSTATE_CONNECTED:
+        UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_USERLAND,
+            "A TCP connection to the server is open");
+        break;
+    case UA_CLIENTSTATE_SECURECHANNEL:
+        UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_USERLAND,
+            "A SecureChannel to the server is open");
+        break;
+    case UA_CLIENTSTATE_SESSION: {
+        UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_USERLAND, "A session with the server is open");
+        /* A new session was created. We need to create the subscription. */
+        /* Create a subscription */
+        UA_CreateSubscriptionRequest request = UA_CreateSubscriptionRequest_default();
+        UA_CreateSubscriptionResponse response = UA_Client_Subscriptions_create(client, request,
+            NULL, NULL, deleteSubscriptionCallback);
+
+        if (response.responseHeader.serviceResult == UA_STATUSCODE_GOOD)
+            UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_USERLAND,
+                "Create subscription succeeded, id %u", response.subscriptionId);
+        else
+            return;
+
+        /* Add a MonitoredItem */
+        UA_MonitoredItemCreateRequest monRequest =
+            UA_MonitoredItemCreateRequest_default(UA_NODEID_NUMERIC(0, UA_NS0ID_SERVER_SERVERSTATUS_CURRENTTIME));
+
+        UA_MonitoredItemCreateResult monResponse =
+            UA_Client_MonitoredItems_createDataChange(client, response.subscriptionId,
+                UA_TIMESTAMPSTORETURN_BOTH,
+                monRequest, NULL, handler_currentTimeChanged, NULL);
+        if (monResponse.statusCode == UA_STATUSCODE_GOOD)
+            UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_USERLAND,
+                "Monitoring UA_NS0ID_SERVER_SERVERSTATUS_CURRENTTIME', id %u",
+                monResponse.monitoredItemId);
+
+        //TODO: check the existance of the nodes inside these functions (otherwise seg faults)
+#ifdef UA_ENABLE_METHODCALLS		
+        UA_UInt32 reqId = 0;
+        UA_Variant input;
+        UA_Variant_init(&input);
+        UA_String stringValue = UA_String_fromChars("World 1");
+        UA_Variant_setScalar(&input, &stringValue, &UA_TYPES[UA_TYPES_STRING]);
+
+        /* Initiate Call 1 */
+        UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_USERLAND, "**** Initiating CallRequest 1");
+        UA_Client_call_async(client,
+            UA_NODEID_NUMERIC(0, UA_NS0ID_OBJECTSFOLDER),
+            UA_NODEID_NUMERIC(1, 62541), 1, &input,
+            methodCalled, NULL, &reqId);
+        UA_String_clear(&stringValue);
+
+        /* Initiate Call 2 */
+        UA_Variant_init(&input);
+        stringValue = UA_String_fromChars("World 2");
+        UA_Variant_setScalar(&input, &stringValue, &UA_TYPES[UA_TYPES_STRING]);
+
+        UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_USERLAND, "**** Initiating CallRequest 2");
+        UA_Client_call_async(client,
+            UA_NODEID_NUMERIC(0, UA_NS0ID_OBJECTSFOLDER),
+            UA_NODEID_NUMERIC(1, 62542), 1, &input,
+            methodCalled, NULL, &reqId);
+        UA_String_clear(&stringValue);
+
+#endif /* UA_ENABLE_METHODCALLS */
+    }
+    break;
+    case UA_CLIENTSTATE_SESSION_RENEWED:
+        UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_USERLAND,
+            "A session with the server is open (renewed)");
+        /* The session was renewed. We don't need to recreate the subscription. */
+        break;
+    case UA_CLIENTSTATE_SESSION_DISCONNECTED:
+        UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_USERLAND, "Session disconnected");
+        break;
+    }
+    return;
+}
+
+int
+main(int argc, char *argv[]) {
+    signal(SIGINT, stopHandler); /* catches ctrl-c */
+
+    UA_Client *client = UA_Client_new();
+    UA_ClientConfig *cc = UA_Client_getConfig(client);
+    UA_ClientConfig_setDefault(cc);
+    /* we use a high timeout because there may be other client and
+    * processing may take long if many method calls are waiting */
+    cc->timeout = 60000;
+
+    /* Set stateCallback */
+    cc->stateCallback = stateCallback;
+    cc->subscriptionInactivityCallback = subscriptionInactivityCallback;
+
+    UA_StatusCode retval = UA_Client_connect(client, "opc.tcp://localhost:4840");
+    if(retval != UA_STATUSCODE_GOOD) {
+        UA_LOG_ERROR(UA_Log_Stdout, UA_LOGCATEGORY_USERLAND,
+                     "Not connected. Retrying to connect in 1 second");
+        UA_Client_delete(client);
+        return EXIT_SUCCESS;
+    }
+
+    /* Endless loop runAsync */
+    while (running) {
+        UA_Client_run_iterate(client, 100);
+    }
+
+    /* Clean up */
+    /* Async disconnect kills unprocessed requests */
+    // UA_Client_disconnect_async (client, &reqId); //can only be used when connected = true
+    // UA_Client_run_iterate (client, &timedOut);
+    UA_Client_disconnect(client);
+    UA_Client_delete(client);
+    return EXIT_SUCCESS;
+}

+ 232 - 0
examples/tutorial_server_method_async.c

@@ -0,0 +1,232 @@
+/* This work is licensed under a Creative Commons CCZero 1.0 Universal License.
+ * See http://creativecommons.org/publicdomain/zero/1.0/ for more information. */
+
+/**
+ * Adding Async Methods to Objects
+ * -------------------------
+ *
+ * An object in an OPC UA information model may contain methods similar to
+ * objects in a programming language. Methods are represented by a MethodNode.
+ * Note that several objects may reference the same MethodNode. When an object
+ * type is instantiated, a reference to the method is added instead of copying
+ * the MethodNode. Therefore, the identifier of the context object is always
+ * explicitly stated when a method is called.
+ *
+ * The method callback takes as input a custom data pointer attached to the
+ * method node, the identifier of the object from which the method is called,
+ * and two arrays for the input and output arguments. The input and output
+ * arguments are all of type :ref:`variant`. Each variant may in turn contain a
+ * (multi-dimensional) array or scalar of any data type.
+ *
+ * Constraints for the method arguments are defined in terms of data type, value
+ * rank and array dimension (similar to variable definitions). The argument
+ * definitions are stored in child VariableNodes of the MethodNode with the
+ * respective BrowseNames ``(0, "InputArguments")`` and ``(0,
+ * "OutputArguments")``.
+ *
+ * Example: Hello World Method
+ * ^^^^^^^^^^^^^^^^^^^^^^^^^^^
+ * The method takes a string scalar and returns a string scalar with "Hello "
+ * prepended. The type and length of the input arguments is checked internally
+ * by the SDK, so that we don't have to verify the arguments in the callback. */
+
+#include <open62541/client_config_default.h>
+#include <open62541/plugin/log_stdout.h>
+#include <open62541/server.h>
+#include <open62541/server_config_default.h>
+
+#include <signal.h>
+#include <stdlib.h>
+
+#ifndef WIN32
+#include <pthread.h>
+#define THREAD_HANDLE pthread_t
+#define THREAD_CREATE(handle, callback) pthread_create(&handle, NULL, callback, NULL)
+#define THREAD_JOIN(handle) pthread_join(handle, NULL)
+#define THREAD_CALLBACK(name) static void * name(void *_)
+#else
+#include <windows.h>
+#define THREAD_HANDLE HANDLE
+#define THREAD_CREATE(handle, callback) { handle = CreateThread( NULL, 0, callback, NULL, 0, NULL); }
+#define THREAD_JOIN(handle) WaitForSingleObject(handle, INFINITE)
+#define THREAD_CALLBACK(name) static DWORD WINAPI name( LPVOID lpParam )
+#endif
+
+static UA_Server* globalServer;
+static volatile UA_Boolean running = true;
+
+static void stopHandler(int sign) {
+    UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_SERVER, "received ctrl-c");
+    running = false;
+}
+
+static UA_StatusCode
+helloWorldMethodCallback1(UA_Server *server,
+                         const UA_NodeId *sessionId, void *sessionHandle,
+                         const UA_NodeId *methodId, void *methodContext,
+                         const UA_NodeId *objectId, void *objectContext,
+                         size_t inputSize, const UA_Variant *input,
+                         size_t outputSize, UA_Variant *output) {
+    UA_String *inputStr = (UA_String*)input->data;
+    UA_String tmp = UA_STRING_ALLOC("Hello ");
+    if(inputStr->length > 0) {
+        tmp.data = (UA_Byte *)UA_realloc(tmp.data, tmp.length + inputStr->length);
+        memcpy(&tmp.data[tmp.length], inputStr->data, inputStr->length);
+        tmp.length += inputStr->length;
+    }
+	UA_Variant_setScalarCopy(output, &tmp, &UA_TYPES[UA_TYPES_STRING]);   
+    char* test = (char*)calloc(1,tmp.length+1);
+    memcpy(test, tmp.data, tmp.length);    
+    UA_String_clear(&tmp);
+    UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_SERVER, "'Hello World 1 (async)' was called and will take 3 seconds");
+    UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_SERVER, "    Data 1: %s", test);
+    free(test);
+	UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_SERVER, "'Hello World 1 (async)' has ended");
+    return UA_STATUSCODE_GOOD;
+}
+
+static void
+addHellWorldMethod1(UA_Server *server) {
+    UA_Argument inputArgument;
+    UA_Argument_init(&inputArgument);
+    inputArgument.description = UA_LOCALIZEDTEXT("en-US", "A String");
+    inputArgument.name = UA_STRING("MyInput");
+    inputArgument.dataType = UA_TYPES[UA_TYPES_STRING].typeId;
+    inputArgument.valueRank = UA_VALUERANK_SCALAR;
+
+    UA_Argument outputArgument;
+    UA_Argument_init(&outputArgument);
+    outputArgument.description = UA_LOCALIZEDTEXT("en-US", "A String");
+    outputArgument.name = UA_STRING("MyOutput");
+    outputArgument.dataType = UA_TYPES[UA_TYPES_STRING].typeId;
+    outputArgument.valueRank = UA_VALUERANK_SCALAR;
+
+    UA_MethodAttributes helloAttr = UA_MethodAttributes_default;
+    helloAttr.description = UA_LOCALIZEDTEXT("en-US","Say `Hello World` async");
+    helloAttr.displayName = UA_LOCALIZEDTEXT("en-US","Hello World async");
+    helloAttr.executable = true;
+    helloAttr.userExecutable = true;
+    UA_Server_addMethodNode(server, UA_NODEID_NUMERIC(1,62541),
+                            UA_NODEID_NUMERIC(0, UA_NS0ID_OBJECTSFOLDER),
+                            UA_NODEID_NUMERIC(0, UA_NS0ID_HASORDEREDCOMPONENT),
+                            UA_QUALIFIEDNAME(1, "hello world"),
+                            helloAttr, &helloWorldMethodCallback1,
+                            1, &inputArgument, 1, &outputArgument, NULL, NULL);	
+	/* Get the method node */
+	UA_NodeId id = UA_NODEID_NUMERIC(1, 62541);
+	UA_Server_setMethodNodeAsync(server, id, UA_TRUE);	
+}
+
+static UA_StatusCode
+helloWorldMethodCallback2(UA_Server *server,
+	const UA_NodeId *sessionId, void *sessionHandle,
+	const UA_NodeId *methodId, void *methodContext,
+	const UA_NodeId *objectId, void *objectContext,
+	size_t inputSize, const UA_Variant *input,
+	size_t outputSize, UA_Variant *output) {
+	UA_String *inputStr = (UA_String*)input->data;
+	UA_String tmp = UA_STRING_ALLOC("Hello ");
+	if (inputStr->length > 0) {
+		tmp.data = (UA_Byte *)UA_realloc(tmp.data, tmp.length + inputStr->length);
+		memcpy(&tmp.data[tmp.length], inputStr->data, inputStr->length);
+		tmp.length += inputStr->length;
+	}
+	UA_Variant_setScalarCopy(output, &tmp, &UA_TYPES[UA_TYPES_STRING]);
+    char* test = (char*)calloc(1, tmp.length + 1);
+    memcpy(test, tmp.data, tmp.length);
+	UA_String_clear(&tmp);
+    UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_SERVER, "'Hello World 2 (async)' was called and will take 1 seconds");
+    UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_SERVER, "    Data 2: %s", test);
+    free(test);
+    UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_SERVER, "'Hello World 2 (async)' has ended");
+	return UA_STATUSCODE_GOOD;
+}
+
+static void
+addHellWorldMethod2(UA_Server *server) {
+	UA_Argument inputArgument;
+	UA_Argument_init(&inputArgument);
+	inputArgument.description = UA_LOCALIZEDTEXT("en-US", "A String");
+	inputArgument.name = UA_STRING("MyInput");
+	inputArgument.dataType = UA_TYPES[UA_TYPES_STRING].typeId;
+	inputArgument.valueRank = UA_VALUERANK_SCALAR;
+
+	UA_Argument outputArgument;
+	UA_Argument_init(&outputArgument);
+	outputArgument.description = UA_LOCALIZEDTEXT("en-US", "A String");
+	outputArgument.name = UA_STRING("MyOutput");
+	outputArgument.dataType = UA_TYPES[UA_TYPES_STRING].typeId;
+	outputArgument.valueRank = UA_VALUERANK_SCALAR;
+
+	UA_MethodAttributes helloAttr = UA_MethodAttributes_default;
+	helloAttr.description = UA_LOCALIZEDTEXT("en-US", "Say `Hello World` sync");
+	helloAttr.displayName = UA_LOCALIZEDTEXT("en-US", "Hello World sync");
+	helloAttr.executable = true;
+	helloAttr.userExecutable = true;
+	UA_Server_addMethodNode(server, UA_NODEID_NUMERIC(1, 62542),
+		UA_NODEID_NUMERIC(0, UA_NS0ID_OBJECTSFOLDER),
+		UA_NODEID_NUMERIC(0, UA_NS0ID_HASORDEREDCOMPONENT),
+		UA_QUALIFIEDNAME(1, "hello world 2"),
+		helloAttr, &helloWorldMethodCallback2,
+		1, &inputArgument, 1, &outputArgument, NULL, NULL);
+	/* Get the method node */
+	UA_NodeId id = UA_NODEID_NUMERIC(1, 62542);
+	UA_Server_setMethodNodeAsync(server, id, UA_TRUE);
+}
+
+THREAD_CALLBACK(ThreadWorker) {
+    while(running) {
+        UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_SERVER,
+                    "Try to dequeue an async operation");
+        const UA_AsyncOperationRequest* request = NULL;
+        void *context = NULL;
+        UA_AsyncOperationType type;
+        if(UA_Server_getAsyncOperation(globalServer, &type, &request, &context) == true) {
+            UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_SERVER, "AsyncMethod_Testing: Got entry: OKAY");
+            UA_CallMethodResult response = UA_Server_call(globalServer, &request->callMethodRequest);
+            UA_Server_setAsyncOperationResult(globalServer, (UA_AsyncOperationResponse*)&response,
+                                              context);
+            UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_SERVER, "AsyncMethod_Testing: Call done: OKAY");
+            UA_CallMethodResult_deleteMembers(&response);
+        } else {
+            /* not a good style, but done for simplicity :-) */
+            Sleep(5000);
+        }
+    }
+    return 0;
+}
+
+/* This callback will be called when a new entry is added to the Callrequest queue */
+static void
+TestCallback(UA_Server *server) {
+    UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_SERVER,
+                "Dispatched an async method");
+}
+
+int main(void) {
+    signal(SIGINT, stopHandler);
+    signal(SIGTERM, stopHandler);
+
+    globalServer = UA_Server_new();
+    UA_ServerConfig *config = UA_Server_getConfig(globalServer);
+    UA_ServerConfig_setDefault(config);
+
+    /* Set the NotifyCallback */
+    config->asyncOperationNotifyCallback = TestCallback;
+
+    /* Start the Worker-Thread */
+    THREAD_HANDLE hThread;
+    THREAD_CREATE(hThread, ThreadWorker);
+    
+    /* Add methods */
+    addHellWorldMethod1(globalServer);
+	addHellWorldMethod2(globalServer);
+
+    UA_StatusCode retval = UA_Server_run(globalServer, &running);
+
+    /* Shutdown the thread */
+    THREAD_JOIN(hThread);
+
+    UA_Server_delete(globalServer);
+    return retval == UA_STATUSCODE_GOOD ? EXIT_SUCCESS : EXIT_FAILURE;
+}

+ 3 - 0
include/open62541/plugin/nodestore.h

@@ -235,6 +235,9 @@ typedef struct {
 
 
     /* Members specific to open62541 */
     /* Members specific to open62541 */
     UA_MethodCallback method;
     UA_MethodCallback method;
+#if UA_MULTITHREADING >= 100
+    UA_Boolean async; /* Indicates an async method call */
+#endif
 } UA_MethodNode;
 } UA_MethodNode;
 
 
 /**
 /**

+ 69 - 0
include/open62541/server.h

@@ -1368,6 +1368,75 @@ UA_Server_AccessControl_allowHistoryUpdateDeleteRawModified(UA_Server *server,
                                                             bool isDeleteModified);
                                                             bool isDeleteModified);
 #endif // UA_ENABLE_HISTORIZING
 #endif // UA_ENABLE_HISTORIZING
 
 
+/**
+* .. _async_methods:
+*
+* Async Methods
+* -------------
+* Sample on working with async method calls
+* Check client_method_async and tutorial_server_method_async for working samples.
+*
+* Minimal Server:
+*
+* - Create a method node
+* - mark as async: ``UA_Server_MethodNodeAsync``
+* - set callback to be notified: ``config->asyncOperationNotifyCallback = ...``
+* - fetch MethodCall: ``UA_Server_GetNextAsyncMethod``
+* - execute Method: e.g. UA_Server_call(UA_Server*, UA_CallMethodRequest*)
+* - pass back the result: ``UA_Server_SetAsyncMethodResult``
+* - free memory used: free UA_CallMethodResult */
+
+#if UA_MULTITHREADING >= 100
+
+/* Set the async flag in a method node */
+UA_StatusCode UA_EXPORT
+UA_Server_setMethodNodeAsync(UA_Server *server, const UA_NodeId id,
+                             UA_Boolean isAsync);
+
+typedef enum {
+    UA_ASYNCOPERATIONTYPE_INVALID, /* 0, the default */
+    UA_ASYNCOPERATIONTYPE_CALL
+    /* UA_ASYNCOPERATIONTYPE_READ, */
+    /* UA_ASYNCOPERATIONTYPE_WRITE, */
+} UA_AsyncOperationType;
+
+typedef union {
+    UA_CallMethodRequest callMethodRequest;
+    /* UA_ReadValueId readValueId; */
+    /* UA_WriteValue writeValue; */
+} UA_AsyncOperationRequest;
+
+typedef union {
+    UA_CallMethodResult callMethodResult;
+    /* UA_DataValue readResult; */
+    /* UA_StatusCode writeResult; */
+} UA_AsyncOperationResponse;
+
+/* Get and remove next Method Call Request
+ *
+ * @param server The server object
+ * @param type The type of the async operation
+ * @param request Receives pointer to the operation
+ * @param context Receives the pointer to the operation context
+ * @return UA_FALSE if queue is empty, UA_TRUE else
+ * Note: Timeout is doubled for requests which already got fetched by worker via UA_Server_getAsyncOperation */
+UA_Boolean UA_EXPORT
+UA_Server_getAsyncOperation(UA_Server *server, UA_AsyncOperationType *type,
+                            const UA_AsyncOperationRequest **request,
+                            void **context);
+
+/* Worker submits Method Call Response
+ *
+ * @param server The server object
+ * @param response Pointer to the operation result
+ * @param context Pointer to the operation context */
+void UA_EXPORT
+UA_Server_setAsyncOperationResult(UA_Server *server,
+                                  const UA_AsyncOperationResponse *response,
+                                  void *context);
+
+#endif /* !UA_MULTITHREADING >= 100 */
+
 _UA_END_DECLS
 _UA_END_DECLS
 
 
 #endif /* UA_SERVER_H_ */
 #endif /* UA_SERVER_H_ */

+ 15 - 0
include/open62541/server_config.h

@@ -83,6 +83,9 @@ typedef struct {
 
 
 #endif
 #endif
 
 
+typedef void
+(*UA_Server_AsyncOperationNotifyCallback)(UA_Server *server);
+
 struct UA_ServerConfig {
 struct UA_ServerConfig {
     UA_UInt16 nThreads; /* only if multithreading is enabled */
     UA_UInt16 nThreads; /* only if multithreading is enabled */
     UA_Logger logger;
     UA_Logger logger;
@@ -140,6 +143,18 @@ struct UA_ServerConfig {
      * .. note:: See the section for :ref:`access-control
      * .. note:: See the section for :ref:`access-control
      *    handling<access-control>`. */
      *    handling<access-control>`. */
 
 
+    /* Async Operations */
+#if UA_MULTITHREADING >= 100
+    UA_Double asyncOperationTimeout; /* in ms, 0 => unlimited */
+    size_t maxAsyncOperationQueueSize; /* 0 => unlimited */
+    UA_Double asyncCallRequestTimeout; /* in ms, 0 => unlimited */
+    /* Notify workers when an async operation was enqueued */
+    UA_Server_AsyncOperationNotifyCallback asyncOperationNotifyCallback;
+#endif
+    /**
+     * .. note:: See the section for :ref:`async
+     * operations<async-operations>`. */
+
     /* Certificate Verification */
     /* Certificate Verification */
     UA_CertificateVerification certificateVerification;
     UA_CertificateVerification certificateVerification;
 
 

+ 6 - 0
plugins/ua_config_default.c

@@ -216,6 +216,12 @@ setDefaultConfig(UA_ServerConfig *conf) {
     /* conf->deleteAtTimeDataCapability = UA_FALSE; */
     /* conf->deleteAtTimeDataCapability = UA_FALSE; */
 #endif
 #endif
 
 
+#if UA_MULTITHREADING >= 100
+    conf->asyncOperationTimeout = 0;
+    conf->maxAsyncOperationQueueSize = 0;
+    conf->asyncCallRequestTimeout = 120000; /* Call request Timeout in ms (2 minutes) */
+#endif
+
     /* --> Finish setting the default static config <-- */
     /* --> Finish setting the default static config <-- */
 
 
     return UA_STATUSCODE_GOOD;
     return UA_STATUSCODE_GOOD;

+ 146 - 0
src/server/ua_asyncmethod_manager.c

@@ -0,0 +1,146 @@
+/* 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/.
+ *
+ *    Copyright 2019 (c) Fraunhofer IOSB (Author: Klaus Schick)
+ * based on
+ *    Copyright 2014-2017 (c) Fraunhofer IOSB (Author: Julius Pfrommer)
+ *    Copyright 2014, 2017 (c) Florian Palm
+ *    Copyright 2015 (c) Sten Grüner
+ *    Copyright 2015 (c) Oleksiy Vasylyev
+ *    Copyright 2017 (c) Stefan Profanter, fortiss GmbH
+ */
+
+#include "ua_asyncmethod_manager.h"
+#include "ua_server_internal.h"
+#include "ua_subscription.h"
+
+#if UA_MULTITHREADING >= 100
+
+UA_StatusCode
+UA_AsyncMethodManager_init(UA_AsyncMethodManager *amm, UA_Server *server) {
+    LIST_INIT(&amm->asyncmethods);
+    amm->currentCount = 0;
+    amm->server = server;
+    return UA_STATUSCODE_GOOD;
+}
+
+void UA_AsyncMethodManager_deleteMembers(UA_AsyncMethodManager *amm) {
+    asyncmethod_list_entry *current, *temp;
+    LIST_FOREACH_SAFE(current, &amm->asyncmethods, pointers, temp) {
+        UA_AsyncMethodManager_removeEntry(amm, current);
+    }
+}
+
+asyncmethod_list_entry*
+UA_AsyncMethodManager_getById(UA_AsyncMethodManager *amm, const UA_UInt32 requestId, const UA_NodeId *sessionId) {
+    asyncmethod_list_entry *current = NULL;
+    LIST_FOREACH(current, &amm->asyncmethods, pointers) {
+        if ((current->requestId == requestId) && UA_NodeId_equal(current->sessionId, sessionId))
+            return current;
+    }
+    return NULL;
+}
+
+UA_StatusCode
+UA_AsyncMethodManager_createEntry(UA_AsyncMethodManager *amm, const UA_NodeId *sessionId,
+    const UA_UInt32 channelId, const UA_UInt32 requestId, const UA_UInt32 requestHandle,
+    const UA_DataType *responseType, const UA_UInt32 nCountdown) {
+    asyncmethod_list_entry *newentry = (asyncmethod_list_entry*)UA_calloc(1, sizeof(asyncmethod_list_entry));
+    if (!newentry) {
+        UA_LOG_ERROR(&amm->server->config.logger, UA_LOGCATEGORY_SERVER,
+            "UA_AsyncMethodManager_createEntry: Mem alloc failed.");
+        return UA_STATUSCODE_BADOUTOFMEMORY;
+    }
+
+    UA_LOG_DEBUG(&amm->server->config.logger, UA_LOGCATEGORY_SERVER,
+        "UA_AsyncMethodManager_createEntry: Chan: %u. Req# %u", channelId, requestId);
+    UA_atomic_addUInt32(&amm->currentCount, 1);
+    newentry->requestId = requestId;
+    newentry->sessionId = sessionId;
+    newentry->requestHandle = requestHandle;
+    newentry->responseType = responseType;
+    newentry->nCountdown = nCountdown;
+    newentry->m_tDispatchTime = UA_DateTime_now();
+    UA_CallResponse_init(&newentry->response);
+    newentry->response.results = (UA_CallMethodResult*)UA_calloc(nCountdown, sizeof(UA_CallMethodResult));
+    newentry->response.resultsSize = nCountdown;
+    if (newentry->response.results == NULL)
+    {
+        UA_LOG_ERROR(&amm->server->config.logger, UA_LOGCATEGORY_SERVER,
+            "UA_AsyncMethodManager_createEntry: Mem alloc failed.");
+        UA_free(newentry);
+        return UA_STATUSCODE_BADOUTOFMEMORY;
+    }
+
+    /* Set the StatusCode to timeout by default. Will be overwritten when the
+     * result is set. */
+    for(size_t i = 0; i < nCountdown; i++)
+        newentry->response.results[i].statusCode = UA_STATUSCODE_BADTIMEOUT;
+
+    LIST_INSERT_HEAD(&amm->asyncmethods, newentry, pointers);
+    return UA_STATUSCODE_GOOD;
+}
+
+/* remove entry and free all allocated data */
+UA_StatusCode
+UA_AsyncMethodManager_removeEntry(UA_AsyncMethodManager *amm, asyncmethod_list_entry *current) {
+    if (current) {
+        LIST_REMOVE(current, pointers);
+        UA_atomic_subUInt32(&amm->currentCount, 1);
+        UA_CallResponse_deleteMembers(&current->response);
+        UA_free(current);
+    }
+    UA_LOG_DEBUG(&amm->server->config.logger, UA_LOGCATEGORY_SERVER,
+                 "UA_AsyncMethodManager_removeEntry: # of open CallRequests: %u", amm->currentCount);
+    return UA_STATUSCODE_GOOD;
+}
+
+/* Check if CallRequest is waiting way too long (120s) */
+void
+UA_AsyncMethodManager_checkTimeouts(UA_Server *server, UA_AsyncMethodManager *amm) {
+    asyncmethod_list_entry* current = NULL;
+    asyncmethod_list_entry* current_tmp = NULL;
+    LIST_FOREACH_SAFE(current, &amm->asyncmethods, pointers, current_tmp) {
+        UA_DateTime tNow = UA_DateTime_now();
+        UA_DateTime tReq = current->m_tDispatchTime;
+        UA_DateTime diff = tNow - tReq;
+
+        /* The calls are all done or the timeout has not passed */
+        if (current->nCountdown == 0 || server->config.asyncCallRequestTimeout <= 0.0 ||
+            diff <= server->config.asyncCallRequestTimeout * (UA_DateTime)UA_DATETIME_MSEC)
+            continue;
+
+        /* We got an unfinished CallResponse waiting way too long for being finished.
+         * Set the remaining StatusCodes and return. */
+        UA_LOG_WARNING(&server->config.logger, UA_LOGCATEGORY_SERVER, "UA_AsyncMethodManager_checkTimeouts: "
+                       "RequestCall #%u was removed due to a timeout (120s)", current->requestId);
+
+        /* Get the session */
+        UA_LOCK(server->serviceMutex);
+        UA_Session* session = UA_SessionManager_getSessionById(&amm->server->sessionManager, current->sessionId);
+        UA_UNLOCK(server->serviceMutex);
+        if(!session) {
+            UA_LOG_WARNING(&amm->server->config.logger, UA_LOGCATEGORY_SERVER,
+                           "UA_AsyncMethodManager_checkTimeouts: Session is gone");
+            goto remove;
+        }
+
+        /* Check the channel */
+        if(!session->header.channel) {
+            UA_LOG_WARNING(&server->config.logger, UA_LOGCATEGORY_SERVER,
+                           "UA_Server_InsertMethodResponse: Channel is gone");
+            goto remove;
+        }
+
+        /* Okay, here we go, send the UA_CallResponse */
+        sendResponse(session->header.channel, current->requestId, current->requestHandle,
+                     (UA_ResponseHeader*)&current->response.responseHeader, current->responseType);
+        UA_LOG_DEBUG(&server->config.logger, UA_LOGCATEGORY_SERVER,
+                     "UA_Server_SendResponse: Response for Req# %u sent", current->requestId);
+    remove:
+        UA_AsyncMethodManager_removeEntry(amm, current);
+    }
+}
+
+#endif

+ 68 - 0
src/server/ua_asyncmethod_manager.h

@@ -0,0 +1,68 @@
+/* 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/.
+ *
+  *    Copyright 2019 (c) Fraunhofer IOSB (Author: Klaus Schick)
+ * based on
+ *    Copyright 2014-2017 (c) Fraunhofer IOSB (Author: Julius Pfrommer)
+ *    Copyright 2014, 2017 (c) Florian Palm
+ *    Copyright 2015 (c) Sten Grüner
+ *    Copyright 2015 (c) Oleksiy Vasylyev
+ *    Copyright 2017 (c) Stefan Profanter, fortiss GmbH
+ */
+
+#ifndef UA_ASYNCMETHOD_MANAGER_H_
+#define UA_ASYNCMETHOD_MANAGER_H_
+
+#include <open62541/server.h>
+
+#include "open62541_queue.h"
+#include "ua_session.h"
+#include "ua_util_internal.h"
+#include "ua_workqueue.h"
+
+#if UA_MULTITHREADING >= 100
+
+_UA_BEGIN_DECLS
+
+typedef struct asyncmethod_list_entry {
+    LIST_ENTRY(asyncmethod_list_entry) pointers;
+    UA_UInt32 requestId;
+    const UA_NodeId *sessionId;
+    UA_UInt32 requestHandle;
+    const UA_DataType *responseType;
+    UA_DateTime	m_tDispatchTime;	/* Creation time */
+    UA_UInt32 nCountdown;			/* Counter for open UA_CallResults */
+    UA_CallResponse response;		/* The 'collected' CallResponse for our CallMethodResponse(s) */
+} asyncmethod_list_entry;
+
+typedef struct UA_AsyncMethodManager {
+    LIST_HEAD(asyncmethod_list, asyncmethod_list_entry) asyncmethods; // doubly-linked list of CallRequests/Responses
+    UA_UInt32 currentCount;
+    UA_Server *server;
+} UA_AsyncMethodManager;
+
+UA_StatusCode
+UA_AsyncMethodManager_init(UA_AsyncMethodManager *amm, UA_Server *server);
+
+/* Deletes all entries */
+void UA_AsyncMethodManager_deleteMembers(UA_AsyncMethodManager *ammm);
+
+UA_StatusCode
+UA_AsyncMethodManager_createEntry(UA_AsyncMethodManager *amm, const UA_NodeId *sessionId, const UA_UInt32 channelId,
+    const UA_UInt32 requestId, const UA_UInt32 requestHandle, const UA_DataType *responseType, const UA_UInt32 nCountdown);
+
+UA_StatusCode
+UA_AsyncMethodManager_removeEntry(UA_AsyncMethodManager *amm, asyncmethod_list_entry *current);
+
+asyncmethod_list_entry*
+UA_AsyncMethodManager_getById(UA_AsyncMethodManager *amm, const UA_UInt32 requestId, const UA_NodeId *sessionId);
+
+void
+UA_AsyncMethodManager_checkTimeouts(UA_Server *server, UA_AsyncMethodManager *amm);
+
+_UA_END_DECLS
+
+#endif
+
+#endif /* UA_ASYNCMETHOD_MANAGER_H_ */

+ 3 - 0
src/server/ua_nodes.c

@@ -110,6 +110,9 @@ static UA_StatusCode
 UA_MethodNode_copy(const UA_MethodNode *src, UA_MethodNode *dst) {
 UA_MethodNode_copy(const UA_MethodNode *src, UA_MethodNode *dst) {
     dst->executable = src->executable;
     dst->executable = src->executable;
     dst->method = src->method;
     dst->method = src->method;
+#if UA_MULTITHREADING >= 100
+    dst->async = src->async;
+#endif
     return UA_STATUSCODE_GOOD;
     return UA_STATUSCODE_GOOD;
 }
 }
 
 

+ 85 - 0
src/server/ua_server.c

@@ -19,6 +19,10 @@
 
 
 #include "ua_server_internal.h"
 #include "ua_server_internal.h"
 
 
+#if UA_MULTITHREADING >= 100
+#include "server/ua_server_methodqueue.h"
+#endif
+
 #ifdef UA_ENABLE_PUBSUB_INFORMATIONMODEL
 #ifdef UA_ENABLE_PUBSUB_INFORMATIONMODEL
 #include "ua_pubsub_ns0.h"
 #include "ua_pubsub_ns0.h"
 #endif
 #endif
@@ -193,6 +197,12 @@ void UA_Server_delete(UA_Server *server) {
     UA_DiscoveryManager_deleteMembers(&server->discoveryManager, server);
     UA_DiscoveryManager_deleteMembers(&server->discoveryManager, server);
 #endif
 #endif
 
 
+#if UA_MULTITHREADING >= 100
+    UA_Server_removeCallback(server, server->nCBIdResponse);
+    UA_Server_MethodQueues_delete(server);
+    UA_AsyncMethodManager_deleteMembers(&server->asyncMethodManager);
+#endif
+
     /* Clean up the Admin Session */
     /* Clean up the Admin Session */
     UA_LOCK(server->serviceMutex);
     UA_LOCK(server->serviceMutex);
     UA_Session_deleteMembersCleanup(&server->adminSession, server);
     UA_Session_deleteMembersCleanup(&server->adminSession, server);
@@ -278,6 +288,14 @@ UA_Server_init(UA_Server *server) {
     UA_SecureChannelManager_init(&server->secureChannelManager, server);
     UA_SecureChannelManager_init(&server->secureChannelManager, server);
     UA_SessionManager_init(&server->sessionManager, server);
     UA_SessionManager_init(&server->sessionManager, server);
 
 
+#if UA_MULTITHREADING >= 100
+    UA_AsyncMethodManager_init(&server->asyncMethodManager, server);
+    UA_Server_MethodQueues_init(server);
+    /* Add a regular callback for for checking responmses using a 50ms interval. */
+    UA_Server_addRepeatedCallback(server, (UA_ServerCallback)UA_Server_CallMethodResponse, NULL,
+                                  50.0, &server->nCBIdResponse);
+#endif
+
     /* Add a regular callback for cleanup and maintenance. With a 10s interval. */
     /* Add a regular callback for cleanup and maintenance. With a 10s interval. */
     UA_Server_addRepeatedCallback(server, (UA_ServerCallback)UA_Server_cleanup, NULL,
     UA_Server_addRepeatedCallback(server, (UA_ServerCallback)UA_Server_cleanup, NULL,
                                   10000.0, NULL);
                                   10000.0, NULL);
@@ -493,6 +511,73 @@ verifyServerApplicationURI(const UA_Server *server) {
 }
 }
 #endif
 #endif
 
 
+#if UA_MULTITHREADING >= 100
+
+void
+UA_Server_InsertMethodResponse(UA_Server *server, const UA_UInt32 nRequestId,
+                               const UA_NodeId *nSessionId, const UA_UInt32 nIndex,
+                               const UA_CallMethodResult *response) {
+    /* Grab the open Request, so we can continue to construct the response */
+    asyncmethod_list_entry *data =
+        UA_AsyncMethodManager_getById(&server->asyncMethodManager, nRequestId, nSessionId);
+    if(!data) {
+        UA_LOG_WARNING(&server->config.logger, UA_LOGCATEGORY_SERVER,
+                       "UA_Server_InsertMethodResponse: can not find UA_CallRequest/UA_CallResponse "
+                       "for Req# %u", nRequestId);
+        return;
+    }
+
+    /* Add UA_CallMethodResult to UA_CallResponse */
+    UA_CallResponse* pResponse = &data->response;
+    UA_CallMethodResult_copy(response, pResponse->results + nIndex);
+
+    /* Reduce the number of open results. Are we done yet with all requests? */
+    data->nCountdown -= 1;
+    if(data->nCountdown > 0)
+        return;
+    
+    /* Get the session */
+    UA_LOCK(server->serviceMutex);
+    UA_Session* session = UA_SessionManager_getSessionById(&server->sessionManager, data->sessionId);
+    UA_UNLOCK(server->serviceMutex);
+    if(!session) {
+        UA_LOG_WARNING(&server->config.logger, UA_LOGCATEGORY_SERVER, "UA_Server_InsertMethodResponse: Session is gone");
+        UA_AsyncMethodManager_removeEntry(&server->asyncMethodManager, data);
+        return;
+    }
+
+    /* Check the channel */
+    UA_SecureChannel* channel = session->header.channel;
+    if(!channel) {
+        UA_LOG_WARNING(&server->config.logger, UA_LOGCATEGORY_SERVER, "UA_Server_InsertMethodResponse: Channel is gone");
+        UA_AsyncMethodManager_removeEntry(&server->asyncMethodManager, data);
+        return;
+    }
+
+    /* Okay, here we go, send the UA_CallResponse */
+    sendResponse(channel, data->requestId, data->requestHandle,
+                 (UA_ResponseHeader*)&data->response.responseHeader, data->responseType);
+    UA_LOG_DEBUG(&server->config.logger, UA_LOGCATEGORY_SERVER,
+                 "UA_Server_SendResponse: Response for Req# %u sent", data->requestId);
+    /* Remove this job from the UA_AsyncMethodManager */
+    UA_AsyncMethodManager_removeEntry(&server->asyncMethodManager, data);
+}
+
+void
+UA_Server_CallMethodResponse(UA_Server *server, void* data) {
+    /* Server fetches Result from queue */
+    struct AsyncMethodQueueElement* pResponseServer = NULL;
+    while(UA_Server_GetAsyncMethodResult(server, &pResponseServer)) {
+        UA_LOG_DEBUG(&server->config.logger, UA_LOGCATEGORY_SERVER,
+                     "UA_Server_CallMethodResponse: Got Response: OKAY");
+        UA_Server_InsertMethodResponse(server, pResponseServer->m_nRequestId, &pResponseServer->m_nSessionId,
+                                       pResponseServer->m_nIndex, &pResponseServer->m_Response);
+        UA_Server_DeleteMethodQueueElement(server, pResponseServer);
+    }
+}
+
+#endif
+
 /********************/
 /********************/
 /* Main Server Loop */
 /* Main Server Loop */
 /********************/
 /********************/

+ 21 - 1
src/server/ua_server_binary.c

@@ -402,7 +402,7 @@ processOPN(UA_Server *server, UA_SecureChannel *channel,
     return retval;
     return retval;
 }
 }
 
 
-static UA_StatusCode
+UA_StatusCode
 sendResponse(UA_SecureChannel *channel, UA_UInt32 requestId, UA_UInt32 requestHandle,
 sendResponse(UA_SecureChannel *channel, UA_UInt32 requestId, UA_UInt32 requestHandle,
              UA_ResponseHeader *responseHeader, const UA_DataType *responseType) {
              UA_ResponseHeader *responseHeader, const UA_DataType *responseType) {
     /* Prepare the ResponseHeader */
     /* Prepare the ResponseHeader */
@@ -549,6 +549,26 @@ processMSGDecoded(UA_Server *server, UA_SecureChannel *channel, UA_UInt32 reques
     }
     }
 #endif
 #endif
 
 
+#if UA_MULTITHREADING >= 100
+    /* If marked as async, the call request is not answered immediately, unless
+     * there is an error */
+    if(requestType == &UA_TYPES[UA_TYPES_CALLREQUEST]) {
+        responseHeader->serviceResult =
+            UA_AsyncMethodManager_createEntry(&server->asyncMethodManager, &session->sessionId,
+                                              channel->securityToken.channelId, requestId, requestHeader->requestHandle, responseType,
+                                              (UA_UInt32)((const UA_CallRequest*)requestHeader)->methodsToCallSize);
+        if(responseHeader->serviceResult == UA_STATUSCODE_GOOD)
+            Service_CallAsync(server, session, channel, requestId,
+                              (const UA_CallRequest*)requestHeader, (UA_CallResponse*)responseHeader);
+
+        /* We got an error, so send response directly */
+        if(responseHeader->serviceResult != UA_STATUSCODE_GOOD)
+            return sendResponse(channel, requestId, requestHeader->requestHandle,
+                                responseHeader, responseType);
+        return UA_STATUSCODE_GOOD;
+    }
+#endif
+
     /* Dispatch the synchronous service call and send the response */
     /* Dispatch the synchronous service call and send the response */
     UA_LOCK(server->serviceMutex);
     UA_LOCK(server->serviceMutex);
     service(server, session, requestHeader, responseHeader);
     service(server, session, requestHeader, responseHeader);

+ 63 - 1
src/server/ua_server_internal.h

@@ -2,6 +2,7 @@
  * License, v. 2.0. If a copy of the MPL was not distributed with this
  * 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/.
  * file, You can obtain one at http://mozilla.org/MPL/2.0/.
  *
  *
+ *    Copyright 2019 (c) Fraunhofer IOSB (Author: Klaus Schick)
  *    Copyright 2014-2018 (c) Fraunhofer IOSB (Author: Julius Pfrommer)
  *    Copyright 2014-2018 (c) Fraunhofer IOSB (Author: Julius Pfrommer)
  *    Copyright 2014, 2017 (c) Florian Palm
  *    Copyright 2014, 2017 (c) Florian Palm
  *    Copyright 2015-2016 (c) Sten Grüner
  *    Copyright 2015-2016 (c) Sten Grüner
@@ -22,6 +23,7 @@
 #include "ua_connection_internal.h"
 #include "ua_connection_internal.h"
 #include "ua_securechannel_manager.h"
 #include "ua_securechannel_manager.h"
 #include "ua_session_manager.h"
 #include "ua_session_manager.h"
+#include "ua_asyncmethod_manager.h"
 #include "ua_timer.h"
 #include "ua_timer.h"
 #include "ua_util_internal.h"
 #include "ua_util_internal.h"
 #include "ua_workqueue.h"
 #include "ua_workqueue.h"
@@ -60,6 +62,28 @@ typedef enum {
     UA_SERVERLIFECYLE_RUNNING
     UA_SERVERLIFECYLE_RUNNING
 } UA_ServerLifecycle;
 } UA_ServerLifecycle;
 
 
+#if UA_MULTITHREADING >= 100
+struct AsyncMethodQueueElement {
+        UA_CallMethodRequest m_Request;
+        UA_CallMethodResult	m_Response;
+        UA_DateTime	m_tDispatchTime;
+        UA_UInt32	m_nRequestId;
+        UA_NodeId	m_nSessionId;
+        UA_UInt32	m_nIndex;
+
+        SIMPLEQ_ENTRY(AsyncMethodQueueElement) next;
+    };
+	
+/* Internal Helper to transfer info */
+    struct AsyncMethodContextInternal {
+        UA_UInt32 nRequestId;
+        UA_NodeId nSessionId;
+        UA_UInt32 nIndex;
+        const UA_CallRequest* pRequest;
+        UA_SecureChannel* pChannel;
+    };
+#endif	
+	
 struct UA_Server {
 struct UA_Server {
     /* Config */
     /* Config */
     UA_ServerConfig config;
     UA_ServerConfig config;
@@ -75,6 +99,9 @@ struct UA_Server {
     /* Security */
     /* Security */
     UA_SecureChannelManager secureChannelManager;
     UA_SecureChannelManager secureChannelManager;
     UA_SessionManager sessionManager;
     UA_SessionManager sessionManager;
+#if UA_MULTITHREADING >= 100
+    UA_AsyncMethodManager asyncMethodManager;
+#endif
     UA_Session adminSession; /* Local access to the services (for startup and
     UA_Session adminSession; /* Local access to the services (for startup and
                               * maintenance) uses this Session with all possible
                               * maintenance) uses this Session with all possible
                               * access rights (Session Id: 1) */
                               * access rights (Session Id: 1) */
@@ -114,10 +141,24 @@ struct UA_Server {
     UA_PubSubManager pubSubManager;
     UA_PubSubManager pubSubManager;
 #endif
 #endif
 
 
+
 #if UA_MULTITHREADING >= 100
 #if UA_MULTITHREADING >= 100
     UA_LOCK_TYPE(networkMutex)
     UA_LOCK_TYPE(networkMutex)
     UA_LOCK_TYPE(serviceMutex)
     UA_LOCK_TYPE(serviceMutex)
-#endif
+
+	/* Async Method Handling */
+    UA_UInt32	nMQCurSize;		/* actual size of queue */
+    UA_UInt64	nCBIdIntegrity;	/* id of callback queue check callback  */
+    UA_UInt64	nCBIdResponse;	/* id of callback check for a response  */
+
+    UA_LOCK_TYPE(ua_request_queue_lock)
+    UA_LOCK_TYPE(ua_response_queue_lock)
+    UA_LOCK_TYPE(ua_pending_list_lock)
+
+    SIMPLEQ_HEAD(ua_method_request_queue, AsyncMethodQueueElement) ua_method_request_queue;    
+    SIMPLEQ_HEAD(ua_method_response_queue, AsyncMethodQueueElement) ua_method_response_queue;
+    SIMPLEQ_HEAD(ua_method_pending_list, AsyncMethodQueueElement) ua_method_pending_list;
+#endif /* UA_MULTITHREADING >= 100 */
 };
 };
 
 
 /*****************/
 /*****************/
@@ -190,6 +231,18 @@ UA_StatusCode
 writeWithSession(UA_Server *server, UA_Session *session,
 writeWithSession(UA_Server *server, UA_Session *session,
                  const UA_WriteValue *value);
                  const UA_WriteValue *value);
 
 
+#if UA_MULTITHREADING >= 100
+void
+UA_Server_InsertMethodResponse(UA_Server *server, const UA_UInt32 nRequestId,
+                               const UA_NodeId* nSessionId, const UA_UInt32 nIndex,
+                               const UA_CallMethodResult* response);
+void 
+    UA_Server_CallMethodResponse(UA_Server *server, void* data);
+#endif
+
+UA_StatusCode
+sendResponse(UA_SecureChannel *channel, UA_UInt32 requestId, UA_UInt32 requestHandle,
+             UA_ResponseHeader *responseHeader, const UA_DataType *responseType);
 
 
 /* Many services come as an array of operations. This function generalizes the
 /* Many services come as an array of operations. This function generalizes the
  * processing of the operations. */
  * processing of the operations. */
@@ -208,6 +261,15 @@ UA_Server_processServiceOperations(UA_Server *server, UA_Session *session,
                                    const UA_DataType *responseOperationsType)
                                    const UA_DataType *responseOperationsType)
     UA_FUNC_ATTR_WARN_UNUSED_RESULT;
     UA_FUNC_ATTR_WARN_UNUSED_RESULT;
 
 
+UA_StatusCode
+UA_Server_processServiceOperationsAsync(UA_Server *server, UA_Session *session,
+                                        UA_ServiceOperation operationCallback,
+                                        void *context,
+                                        const size_t *requestOperations,
+                                        const UA_DataType *requestOperationsType,
+                                        size_t *responseOperations,
+                                        const UA_DataType *responseOperationsType)
+UA_FUNC_ATTR_WARN_UNUSED_RESULT;
 
 
 /******************************************/
 /******************************************/
 /* Internal function calls, without locks */
 /* Internal function calls, without locks */

+ 339 - 0
src/server/ua_server_methodqueue.c

@@ -0,0 +1,339 @@
+/* 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/.
+*
+*    Copyright 2019 (c) Fraunhofer IOSB (Author: Klaus Schick)
+*/
+
+#include "ua_server_methodqueue.h"
+#include <open62541/types_generated_handling.h>
+#include "ua_server_internal.h"
+#include <open62541/plugin/log.h>
+
+#if UA_MULTITHREADING >= 100
+
+/* Initialize Request Queue */
+void
+UA_Server_MethodQueues_init(UA_Server *server) {
+    server->nMQCurSize = 0;
+
+    UA_LOCK_INIT(server->ua_request_queue_lock);
+    SIMPLEQ_INIT(&server->ua_method_request_queue);
+
+    UA_LOCK_INIT(server->ua_response_queue_lock);
+    SIMPLEQ_INIT(&server->ua_method_response_queue);
+
+    UA_LOCK_INIT(server->ua_pending_list_lock);
+    SIMPLEQ_INIT(&server->ua_method_pending_list);
+
+    /* Add a regular callback for cleanup and maintenance using a 10s interval. */
+    UA_Server_addRepeatedCallback(server, (UA_ServerCallback)UA_Server_CheckQueueIntegrity,
+        NULL, 10000.0, &server->nCBIdIntegrity);
+}
+
+/* Cleanup and terminate queues */
+void
+UA_Server_MethodQueues_delete(UA_Server *server) {
+    UA_Server_removeCallback(server, server->nCBIdIntegrity);
+
+    /* Clean up request queue */
+    UA_LOCK(server->ua_request_queue_lock);
+    while (!SIMPLEQ_EMPTY(&server->ua_method_request_queue)) {
+        struct AsyncMethodQueueElement* request = SIMPLEQ_FIRST(&server->ua_method_request_queue);
+        SIMPLEQ_REMOVE_HEAD(&server->ua_method_request_queue, next);
+        UA_Server_DeleteMethodQueueElement(server, request);
+    }
+    UA_UNLOCK(server->ua_request_queue_lock);
+
+    /* Clean up response queue */
+    UA_LOCK(server->ua_response_queue_lock);
+    while (!SIMPLEQ_EMPTY(&server->ua_method_response_queue)) {
+        struct AsyncMethodQueueElement* response = SIMPLEQ_FIRST(&server->ua_method_response_queue);
+        SIMPLEQ_REMOVE_HEAD(&server->ua_method_response_queue, next);
+        UA_Server_DeleteMethodQueueElement(server, response);
+    }
+    UA_UNLOCK(server->ua_response_queue_lock);
+
+    /* Clear all pending */
+    UA_LOCK(server->ua_pending_list_lock);
+    while (!SIMPLEQ_EMPTY(&server->ua_method_pending_list)) {
+        struct AsyncMethodQueueElement* response = SIMPLEQ_FIRST(&server->ua_method_pending_list);
+        SIMPLEQ_REMOVE_HEAD(&server->ua_method_pending_list, next);
+        UA_Server_DeleteMethodQueueElement(server, response);
+    }
+    UA_UNLOCK(server->ua_pending_list_lock);
+
+    /* delete all locks */
+    /* TODO KS: actually we should make sure the worker is not 'hanging' on this lock anymore */
+    Sleep(100);
+    UA_LOCK_DESTROY(server->ua_response_queue_lock);
+    UA_LOCK_DESTROY(server->ua_request_queue_lock);
+    UA_LOCK_DESTROY(server->ua_pending_list_lock);
+}
+
+/* Checks queue element timeouts */
+void
+UA_Server_CheckQueueIntegrity(UA_Server *server, void *data) {
+    /* for debugging/testing purposes */
+    if(server->config.asyncOperationTimeout <= 0.0) {
+        UA_AsyncMethodManager_checkTimeouts(server, &server->asyncMethodManager);
+        return;
+    }
+
+    /* To prevent a lockup, we remove a maximum 10% of timed out entries */
+    /* on small queues, we do at least 3 */
+    size_t bMaxRemove = server->config.maxAsyncOperationQueueSize / 10;
+    if(bMaxRemove < 3)
+        bMaxRemove = 3;
+    UA_Boolean bCheckQueue = UA_TRUE;
+    UA_LOCK(server->ua_request_queue_lock);
+    /* Check if entry has been in the queue too long time */
+    while(bCheckQueue && bMaxRemove-- && !SIMPLEQ_EMPTY(&server->ua_method_request_queue)) {
+        struct AsyncMethodQueueElement* request_elem = SIMPLEQ_FIRST(&server->ua_method_request_queue);
+        if (request_elem) {
+            UA_DateTime tNow = UA_DateTime_now();
+            UA_DateTime tReq = request_elem->m_tDispatchTime;
+            UA_DateTime diff = tNow - tReq;
+            if(diff > (UA_DateTime)(server->config.asyncOperationTimeout * UA_DATETIME_MSEC)) {
+                /* remove it from the queue */
+                UA_LOG_WARNING(&server->config.logger, UA_LOGCATEGORY_SERVER,
+                    "UA_Server_CheckQueueIntegrity: Request #%u was removed due to a timeout (%f)",
+                    request_elem->m_nRequestId, server->config.asyncOperationTimeout);
+                SIMPLEQ_REMOVE_HEAD(&server->ua_method_request_queue, next);
+                server->nMQCurSize--;
+                /* Notify that we removed this request - e.g. Bad Call Response (UA_STATUSCODE_BADREQUESTTIMEOUT) */
+                UA_CallMethodResult* result = &request_elem->m_Response;
+                UA_CallMethodResult_clear(result);
+                result->statusCode = UA_STATUSCODE_BADREQUESTTIMEOUT;
+                UA_Server_InsertMethodResponse(server, request_elem->m_nRequestId, &request_elem->m_nSessionId, request_elem->m_nIndex, result);
+                UA_CallMethodResult_deleteMembers(result);
+                UA_Server_DeleteMethodQueueElement(server, request_elem);
+            }
+            else {
+                /* queue entry is not older than server->nMQTimeoutSecs, so we stop checking */
+                bCheckQueue = UA_FALSE;
+            }
+        }
+    }
+    UA_UNLOCK(server->ua_request_queue_lock);
+
+    /* Clear all pending */
+    bCheckQueue = UA_TRUE;
+    UA_LOCK(server->ua_pending_list_lock);
+    /* Check if entry has been in the pendig list too long time */
+    while(bCheckQueue && !SIMPLEQ_EMPTY(&server->ua_method_pending_list)) {
+        struct AsyncMethodQueueElement* request_elem = SIMPLEQ_FIRST(&server->ua_method_pending_list);
+        if (request_elem) {
+            UA_DateTime tNow = UA_DateTime_now();
+            UA_DateTime tReq = request_elem->m_tDispatchTime;
+            UA_DateTime diff = tNow - tReq;
+            if (diff > (UA_DateTime)(server->config.asyncOperationTimeout * UA_DATETIME_MSEC)) {
+                /* remove it from the list */
+                UA_LOG_WARNING(&server->config.logger, UA_LOGCATEGORY_SERVER,
+                    "UA_Server_CheckQueueIntegrity: Pending request #%u was removed due to a timeout (%f)",
+                    request_elem->m_nRequestId, server->config.asyncOperationTimeout);
+                SIMPLEQ_REMOVE_HEAD(&server->ua_method_pending_list, next);
+                /* Notify that we removed this request - e.g. Bad Call Response (UA_STATUSCODE_BADREQUESTTIMEOUT) */
+                UA_CallMethodResult* result = &request_elem->m_Response;
+                UA_CallMethodResult_clear(result);
+                result->statusCode = UA_STATUSCODE_BADREQUESTTIMEOUT;
+                UA_Server_InsertMethodResponse(server, request_elem->m_nRequestId, &request_elem->m_nSessionId, request_elem->m_nIndex, result);
+                UA_CallMethodResult_deleteMembers(result);
+                UA_Server_DeleteMethodQueueElement(server, request_elem);
+            }
+            else {
+                /* list entry is not older than server->nMQTimeoutSecs, so we stop checking */
+                bCheckQueue = UA_FALSE;
+            }
+        }
+    }
+    UA_UNLOCK(server->ua_pending_list_lock);
+
+    /* Now we check if we still have pending CallRequests */
+    UA_AsyncMethodManager_checkTimeouts(server, &server->asyncMethodManager);
+}
+
+/* Enqueue next MethodRequest */
+UA_StatusCode
+UA_Server_SetNextAsyncMethod(UA_Server *server,
+    const UA_UInt32 nRequestId,
+    const UA_NodeId *nSessionId,
+    const UA_UInt32 nIndex,
+    const UA_CallMethodRequest *pRequest) {
+    UA_StatusCode result = UA_STATUSCODE_GOOD;
+
+    if (server->config.maxAsyncOperationQueueSize == 0 ||
+        server->nMQCurSize < server->config.maxAsyncOperationQueueSize) {
+        struct AsyncMethodQueueElement* elem = (struct AsyncMethodQueueElement*)UA_calloc(1, sizeof(struct AsyncMethodQueueElement));
+        if (elem)
+        {
+            UA_CallMethodRequest_init(&elem->m_Request);
+            result = UA_CallMethodRequest_copy(pRequest, &elem->m_Request);
+            if (result != UA_STATUSCODE_GOOD) {
+                UA_LOG_ERROR(&server->config.logger, UA_LOGCATEGORY_SERVER,
+                    "UA_Server_SetAsyncMethodResult: UA_CallMethodRequest_copy failed.");                
+            }
+
+            elem->m_nRequestId = nRequestId;
+            elem->m_nSessionId = *nSessionId;
+            elem->m_nIndex = nIndex;
+            UA_CallMethodResult_clear(&elem->m_Response);
+            elem->m_tDispatchTime = UA_DateTime_now();
+            UA_LOCK(server->ua_request_queue_lock);
+            SIMPLEQ_INSERT_TAIL(&server->ua_method_request_queue, elem, next);
+            server->nMQCurSize++;
+            if(server->config.asyncOperationNotifyCallback)
+                server->config.asyncOperationNotifyCallback(server);
+            UA_UNLOCK(server->ua_request_queue_lock);
+        }
+        else {
+            /* notify about error */
+            UA_LOG_ERROR(&server->config.logger, UA_LOGCATEGORY_SERVER,
+                "UA_Server_SetNextAsyncMethod: Mem alloc failed.");
+            result = UA_STATUSCODE_BADOUTOFMEMORY;
+        }
+    }
+    else {
+        /* issue warning */
+        UA_LOG_WARNING(&server->config.logger, UA_LOGCATEGORY_SERVER,
+            "UA_Server_SetNextAsyncMethod: Queue exceeds limit (%d).",
+                       (UA_UInt32)server->config.maxAsyncOperationQueueSize);
+        result = UA_STATUSCODE_BADUNEXPECTEDERROR;
+    }
+    return result;
+}
+
+/* Private API */
+/* Get next Method Call Response */
+UA_Boolean
+UA_Server_GetAsyncMethodResult(UA_Server *server, struct AsyncMethodQueueElement **pResponse) {
+    UA_Boolean bRV = UA_FALSE;    
+    UA_LOCK(server->ua_response_queue_lock);
+    if (!SIMPLEQ_EMPTY(&server->ua_method_response_queue)) {
+        *pResponse = SIMPLEQ_FIRST(&server->ua_method_response_queue);
+        SIMPLEQ_REMOVE_HEAD(&server->ua_method_response_queue, next);
+        bRV = UA_TRUE;
+    }
+    UA_UNLOCK(server->ua_response_queue_lock);
+    return bRV;
+}
+
+/* Deep delete queue Element - only memory we did allocate */
+void
+UA_Server_DeleteMethodQueueElement(UA_Server *server, struct AsyncMethodQueueElement *pElem) {
+    UA_CallMethodRequest_deleteMembers(&pElem->m_Request);
+    UA_CallMethodResult_deleteMembers(&pElem->m_Response);
+    UA_free(pElem);
+}
+
+void UA_Server_AddPendingMethodCall(UA_Server* server, struct AsyncMethodQueueElement *pElem) {
+    UA_LOCK(server->ua_pending_list_lock);
+    pElem->m_tDispatchTime = UA_DateTime_now(); /* reset timestamp for timeout */
+    SIMPLEQ_INSERT_TAIL(&server->ua_method_pending_list, pElem, next);
+    UA_UNLOCK(server->ua_pending_list_lock);
+}
+
+void UA_Server_RmvPendingMethodCall(UA_Server *server, struct AsyncMethodQueueElement *pElem) {
+/* Remove element from pending list */
+/* Do NOT delete it because we still need it */
+    struct AsyncMethodQueueElement* current = NULL;
+    struct AsyncMethodQueueElement* tmp_iter = NULL;
+    struct AsyncMethodQueueElement* previous = NULL;
+    UA_LOCK(server->ua_pending_list_lock);
+    SIMPLEQ_FOREACH_SAFE(current, &server->ua_method_pending_list, next, tmp_iter) {
+        if (pElem == current) {
+            if (previous == NULL)
+                SIMPLEQ_REMOVE_HEAD(&server->ua_method_pending_list, next);
+            else
+                SIMPLEQ_REMOVE_AFTER(&server->ua_method_pending_list, previous, next);
+            break;
+        }
+        previous = current;
+    }
+    UA_UNLOCK(server->ua_pending_list_lock);
+    return;
+}
+
+UA_Boolean UA_Server_IsPendingMethodCall(UA_Server *server, struct AsyncMethodQueueElement *pElem) {
+    UA_Boolean bRV = UA_FALSE;
+    struct AsyncMethodQueueElement* current = NULL;
+    struct AsyncMethodQueueElement* tmp_iter = NULL;
+    UA_LOCK(server->ua_pending_list_lock);
+    SIMPLEQ_FOREACH_SAFE(current, &server->ua_method_pending_list, next, tmp_iter) {
+        if (pElem == current) {
+            bRV = UA_TRUE;
+            break;
+        }
+    }
+    UA_UNLOCK(server->ua_pending_list_lock);
+    return bRV;
+}
+
+/* ----------------------------------------------------------------- */
+/* Public API */
+
+/* Get and remove next Method Call Request */
+UA_Boolean
+UA_Server_getAsyncOperation(UA_Server *server, UA_AsyncOperationType *type,
+                            const UA_AsyncOperationRequest **request,
+                            void **context) {
+    UA_Boolean bRV = UA_FALSE;
+    *type = UA_ASYNCOPERATIONTYPE_INVALID;
+    struct AsyncMethodQueueElement *elem = NULL;
+    UA_LOCK(server->ua_request_queue_lock);
+    if (!SIMPLEQ_EMPTY(&server->ua_method_request_queue)) {
+        elem = SIMPLEQ_FIRST(&server->ua_method_request_queue);
+        SIMPLEQ_REMOVE_HEAD(&server->ua_method_request_queue, next);
+        server->nMQCurSize--;
+        if (elem) {
+            *request = (UA_AsyncOperationRequest*)&elem->m_Request;
+            *context = (void*)elem;            
+            bRV = UA_TRUE;
+        }
+        else {
+            UA_LOG_ERROR(&server->config.logger, UA_LOGCATEGORY_SERVER,
+                "UA_Server_GetNextAsyncMethod: elem is a NULL-Pointer.");
+        }
+    }
+    UA_UNLOCK(server->ua_request_queue_lock);
+    if (bRV && elem) {
+        *type = UA_ASYNCOPERATIONTYPE_CALL;
+        UA_Server_AddPendingMethodCall(server, elem);
+    }
+    return bRV;
+}
+
+/* Worker submits Method Call Response */
+void
+UA_Server_setAsyncOperationResult(UA_Server *server,
+                                  const UA_AsyncOperationResponse *response,
+                                  void *context) {
+    struct AsyncMethodQueueElement* elem = (struct AsyncMethodQueueElement*)context;
+    if (!elem || !UA_Server_IsPendingMethodCall(server, elem) ) {
+        /* Something went wrong, late call? */
+        /* Dismiss response */
+        UA_LOG_WARNING(&server->config.logger, UA_LOGCATEGORY_SERVER,
+            "UA_Server_SetAsyncMethodResult: elem is a NULL-Pointer or not valid anymore.");
+    }
+    else {
+        /* UA_Server_RmvPendingMethodCall MUST be called outside the lock
+        * otherwise we can run into a deadlock */
+        UA_Server_RmvPendingMethodCall(server, elem);
+
+        UA_StatusCode result = UA_CallMethodResult_copy(&response->callMethodResult,
+                                                        &elem->m_Response);
+        if (result != UA_STATUSCODE_GOOD) {
+            UA_LOG_ERROR(&server->config.logger, UA_LOGCATEGORY_SERVER,
+                "UA_Server_SetAsyncMethodResult: UA_CallMethodResult_copy failed.");
+            /* Add failed CallMethodResult to response queue */
+            UA_CallMethodResult_clear(&elem->m_Response);
+            elem->m_Response.statusCode = UA_STATUSCODE_BADOUTOFMEMORY;
+        }
+        /* Insert response in queue */
+        UA_LOCK(server->ua_response_queue_lock);
+        SIMPLEQ_INSERT_TAIL(&server->ua_method_response_queue, elem, next);
+        UA_UNLOCK(server->ua_response_queue_lock);
+    }
+}
+
+#endif /* !UA_MULTITHREADING >= 100 */

+ 93 - 0
src/server/ua_server_methodqueue.h

@@ -0,0 +1,93 @@
+/* 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/.
+*
+*    Copyright 2019 (c) Fraunhofer IOSB (Author: Klaus Schick)
+*/
+
+#ifndef UA_SERVER_METHODQUEUE_H_
+#define	UA_SERVER_METHODQUEUE_H_
+
+#include <open62541/types_generated.h>
+#include "open62541_queue.h"
+#include "ua_server_internal.h"
+
+#if UA_MULTITHREADING >= 100
+
+_UA_BEGIN_DECLS
+
+/* Public API, see server.h  */
+
+/* Private API */
+/* Initialize Request Queue
+*
+* @param server The server object */
+void UA_Server_MethodQueues_init(UA_Server *server);
+
+/* Cleanup and terminate queues
+*
+* @param server The server object */
+void UA_Server_MethodQueues_delete(UA_Server *server);
+
+/* Checks queue element timeouts
+*
+* @param server The server object */
+void UA_Server_CheckQueueIntegrity(UA_Server *server, void *data);
+
+/* Deep delete of queue element (except UA_CallMethodRequest which is managed by the caller)
+*
+* @param server The server object
+* @param pElem Pointer to queue element */
+void UA_Server_DeleteMethodQueueElement(UA_Server *server, struct AsyncMethodQueueElement *pElem);
+
+
+/* Internal functions, declared here to be accessible by unit test(s) */
+
+/* Add element to pending list 
+*
+* @param server The server object
+* @param pElem Pointer to queue element */
+void UA_Server_AddPendingMethodCall(UA_Server* server, struct AsyncMethodQueueElement *pElem);
+
+/* Remove element from pending list
+*
+* @param server The server object
+* @param pElem Pointer to queue element */
+void UA_Server_RmvPendingMethodCall(UA_Server *server, struct AsyncMethodQueueElement *pElem);
+
+/* check if element is in pending list
+*
+* @param server The server object
+* @param pElem Pointer to queue element
+* @return UA_TRUE if element was found, UA_FALSE else */
+UA_Boolean UA_Server_IsPendingMethodCall(UA_Server *server, struct AsyncMethodQueueElement *pElem);
+
+
+/* Enqueue next MethodRequest
+*
+* @param server The server object
+* @param Pointer to UA_NodeId (pSessionIdentifier)
+* @param UA_UInt32 (unique requestId/server)
+* @param Index of request within UA_Callrequest
+* @param Pointer to UA_CallMethodRequest
+* @return UA_STATUSCODE_GOOD if request was enqueue,
+    UA_STATUSCODE_BADUNEXPECTEDERROR if queue full
+*	UA_STATUSCODE_BADOUTOFMEMORY if no memory available */
+UA_StatusCode UA_Server_SetNextAsyncMethod(UA_Server *server,
+    const UA_UInt32 nRequestId,
+    const UA_NodeId *nSessionId,
+    const UA_UInt32 nIndex,
+    const UA_CallMethodRequest* pRequest);
+
+/* Get next Method Call Response, user has to call 'UA_DeleteMethodQueueElement(...)' to cleanup memory
+*
+* @param server The server object
+* @param Receives pointer to AsyncMethodQueueElement
+* @return UA_FALSE if queue is empty, UA_TRUE else */
+UA_Boolean UA_Server_GetAsyncMethodResult(UA_Server *server, struct AsyncMethodQueueElement **pResponse);
+
+_UA_END_DECLS
+
+#endif /* !UA_MULTITHREADING >= 100 */
+
+#endif /* !UA_SERVER_METHODQUEUE_H_ */

+ 36 - 0
src/server/ua_server_utils.c

@@ -293,6 +293,42 @@ UA_Server_processServiceOperations(UA_Server *server, UA_Session *session,
     return UA_STATUSCODE_GOOD;
     return UA_STATUSCODE_GOOD;
 }
 }
 
 
+#if UA_MULTITHREADING >= 100
+
+/* this is a copy of the above + contest.nIndex is set :-( Any ideas for a better solution? */
+UA_StatusCode
+UA_Server_processServiceOperationsAsync(UA_Server *server, UA_Session *session,
+    UA_ServiceOperation operationCallback,
+    void *context, const size_t *requestOperations,
+    const UA_DataType *requestOperationsType,
+    size_t *responseOperations,
+    const UA_DataType *responseOperationsType) {
+    size_t ops = *requestOperations;
+    if (ops == 0)
+        return UA_STATUSCODE_BADNOTHINGTODO;
+
+    struct AsyncMethodContextInternal* pContext = (struct AsyncMethodContextInternal*)context;
+
+    /* No padding after size_t */
+    void **respPos = (void**)((uintptr_t)responseOperations + sizeof(size_t));
+    *respPos = UA_Array_new(ops, responseOperationsType);
+    if (!(*respPos))
+        return UA_STATUSCODE_BADOUTOFMEMORY;
+
+    *responseOperations = ops;
+    uintptr_t respOp = (uintptr_t)*respPos;
+    /* No padding after size_t */
+    uintptr_t reqOp = *(uintptr_t*)((uintptr_t)requestOperations + sizeof(size_t));
+    for (size_t i = 0; i < ops; i++) {
+        pContext->nIndex = (UA_UInt32)i;
+        operationCallback(server, session, context, (void*)reqOp, (void*)respOp);
+        reqOp += requestOperationsType->memSize;
+        respOp += responseOperationsType->memSize;
+    }
+    return UA_STATUSCODE_GOOD;
+}
+#endif
+
 /* A few global NodeId definitions */
 /* A few global NodeId definitions */
 const UA_NodeId subtypeId = {0, UA_NODEIDTYPE_NUMERIC, {UA_NS0ID_HASSUBTYPE}};
 const UA_NodeId subtypeId = {0, UA_NODEIDTYPE_NUMERIC, {UA_NS0ID_HASSUBTYPE}};
 const UA_NodeId hierarchicalReferences = {0, UA_NODEIDTYPE_NUMERIC, {UA_NS0ID_HIERARCHICALREFERENCES}};
 const UA_NodeId hierarchicalReferences = {0, UA_NODEIDTYPE_NUMERIC, {UA_NS0ID_HIERARCHICALREFERENCES}};

+ 6 - 0
src/server/ua_services.h

@@ -356,6 +356,12 @@ Service_HistoryUpdate(UA_Server *server, UA_Session *session,
 void Service_Call(UA_Server *server, UA_Session *session,
 void Service_Call(UA_Server *server, UA_Session *session,
                   const UA_CallRequest *request,
                   const UA_CallRequest *request,
                   UA_CallResponse *response);
                   UA_CallResponse *response);
+
+# if UA_MULTITHREADING >= 100
+void Service_CallAsync(UA_Server *server, UA_Session *session, UA_SecureChannel* channel,
+                       UA_UInt32 requestId, const UA_CallRequest *request,
+                       UA_CallResponse *response);
+#endif
 #endif
 #endif
 
 
 #ifdef UA_ENABLE_SUBSCRIPTIONS
 #ifdef UA_ENABLE_SUBSCRIPTIONS

+ 90 - 0
src/server/ua_services_method.c

@@ -17,6 +17,8 @@
 
 
 #ifdef UA_ENABLE_METHODCALLS /* conditional compilation */
 #ifdef UA_ENABLE_METHODCALLS /* conditional compilation */
 
 
+#include "ua_server_methodqueue.h"
+
 static const UA_VariableNode *
 static const UA_VariableNode *
 getArgumentsVariableNode(UA_Server *server, const UA_MethodNode *ofMethod,
 getArgumentsVariableNode(UA_Server *server, const UA_MethodNode *ofMethod,
                          UA_String withBrowseName) {
                          UA_String withBrowseName) {
@@ -230,6 +232,94 @@ callWithMethodAndObject(UA_Server *server, UA_Session *session,
     /* TODO: Verify Output matches the argument definition */
     /* TODO: Verify Output matches the argument definition */
 }
 }
 
 
+#if UA_MULTITHREADING >= 100
+
+static UA_StatusCode
+setMethodNodeAsync(UA_Server *server, UA_Session *session,
+                   UA_Node *node, UA_Boolean *isAsync) {
+    UA_MethodNode *method = (UA_MethodNode*)node;
+    if(method->nodeClass != UA_NODECLASS_METHOD)
+        return UA_STATUSCODE_BADNODECLASSINVALID;
+    method->async = *isAsync;
+    return UA_STATUSCODE_GOOD;
+}
+
+UA_StatusCode
+UA_Server_setMethodNodeAsync(UA_Server *server, const UA_NodeId id,
+                             UA_Boolean isAsync) {
+    return UA_Server_editNode(server, &server->adminSession, &id,
+                              (UA_EditNodeCallback)setMethodNodeAsync, &isAsync);
+}
+
+static void
+Operation_CallMethodAsync(UA_Server *server, UA_Session *session, void *context,
+    const UA_CallMethodRequest *request, UA_CallMethodResult *result) {
+    struct AsyncMethodContextInternal *pContext = (struct AsyncMethodContextInternal*)context;
+
+    /* Get the method node */
+    const UA_MethodNode *method = (const UA_MethodNode*)
+        UA_Nodestore_getNode(server->nsCtx, &request->methodId);
+    if (!method) {
+        result->statusCode = UA_STATUSCODE_BADNODEIDUNKNOWN;
+        return;
+    }
+
+    /* Get the object node */
+    const UA_ObjectNode *object = (const UA_ObjectNode*)
+        UA_Nodestore_getNode(server->nsCtx, &request->objectId);
+    if (!object) {
+        result->statusCode = UA_STATUSCODE_BADNODEIDUNKNOWN;
+        UA_Nodestore_releaseNode(server->nsCtx, (const UA_Node*)method);
+        return;
+    }
+
+    if (method->async) {
+        /* Async case */        
+        UA_StatusCode res = UA_Server_SetNextAsyncMethod(server, pContext->nRequestId, &pContext->nSessionId, pContext->nIndex, request);
+        if (res != UA_STATUSCODE_GOOD) {
+            UA_LOG_ERROR(&server->config.logger, UA_LOGCATEGORY_SERVER,
+                "Operation_CallMethodAsync: Adding request to queue: FAILED");
+            /* Set this Request as failed */
+            UA_CallMethodResult_clear(result);
+            result->statusCode = res;
+            UA_Server_InsertMethodResponse(server, pContext->nRequestId, &pContext->nSessionId, pContext->nIndex, result);
+            UA_CallMethodResult_deleteMembers(result);
+        }
+    }
+    else {
+        /* Sync execution case, continue with method and object as context */
+        callWithMethodAndObject(server, session, request, result, method, object);
+        UA_Server_InsertMethodResponse(server, pContext->nRequestId, &pContext->nSessionId, pContext->nIndex, result);
+        UA_CallMethodResult_deleteMembers(result);
+    }
+
+    /* Release the method and object node */
+    UA_Nodestore_releaseNode(server->nsCtx, (const UA_Node*)method);
+    UA_Nodestore_releaseNode(server->nsCtx, (const UA_Node*)object);
+}
+
+void Service_CallAsync(UA_Server *server, UA_Session *session, UA_SecureChannel* channel, UA_UInt32 requestId,
+    const UA_CallRequest *request, UA_CallResponse *response) {
+
+    UA_LOG_DEBUG_SESSION(&server->config.logger, session, "Processing CallRequestAsync");
+    if (server->config.maxNodesPerMethodCall != 0 &&
+        request->methodsToCallSize > server->config.maxNodesPerMethodCall) {
+        response->responseHeader.serviceResult = UA_STATUSCODE_BADTOOMANYOPERATIONS;
+        return;
+    }
+
+    struct AsyncMethodContextInternal context;
+    context.nRequestId = requestId;
+    context.nSessionId = session->sessionId;
+    context.pRequest = request;
+    context.pChannel = (UA_SecureChannel*)channel;
+    response->responseHeader.serviceResult =
+        UA_Server_processServiceOperationsAsync(server, session, (UA_ServiceOperation)Operation_CallMethodAsync, &context,
+            &request->methodsToCallSize, &UA_TYPES[UA_TYPES_CALLMETHODREQUEST],
+            &response->resultsSize, &UA_TYPES[UA_TYPES_CALLMETHODRESULT]);
+}
+#endif
+
 static void
 static void
 Operation_CallMethod(UA_Server *server, UA_Session *session, void *context,
 Operation_CallMethod(UA_Server *server, UA_Session *session, void *context,
                      const UA_CallMethodRequest *request, UA_CallMethodResult *result) {
                      const UA_CallMethodRequest *request, UA_CallMethodResult *result) {

+ 4 - 0
tests/CMakeLists.txt

@@ -221,6 +221,10 @@ if (UA_MULTITHREADING EQUAL 100)
     add_executable(check_mt_addDeleteObject multithreading/check_mt_addDeleteObject.c $<TARGET_OBJECTS:open62541-object> $<TARGET_OBJECTS:open62541-testplugins>)
     add_executable(check_mt_addDeleteObject multithreading/check_mt_addDeleteObject.c $<TARGET_OBJECTS:open62541-object> $<TARGET_OBJECTS:open62541-testplugins>)
     target_link_libraries(check_mt_addDeleteObject ${LIBS})
     target_link_libraries(check_mt_addDeleteObject ${LIBS})
     add_test_valgrind(mt_addDeleteObject ${TESTS_BINARY_DIR}/check_mt_addDeleteObject)
     add_test_valgrind(mt_addDeleteObject ${TESTS_BINARY_DIR}/check_mt_addDeleteObject)
+
+    add_executable(check_server_asyncop server/check_server_asyncop.c $<TARGET_OBJECTS:open62541-object> $<TARGET_OBJECTS:open62541-testplugins>)
+    target_link_libraries(check_server_asyncop ${LIBS})
+    add_test_valgrind(server_asyncop ${TESTS_BINARY_DIR}/check_server_asyncop)
 endif()
 endif()
 
 
 add_executable(check_services_call server/check_services_call.c $<TARGET_OBJECTS:open62541-object> $<TARGET_OBJECTS:open62541-testplugins>)
 add_executable(check_services_call server/check_services_call.c $<TARGET_OBJECTS:open62541-object> $<TARGET_OBJECTS:open62541-testplugins>)