CMakeLists.txt

@@ -2,7 +2,7 @@ project(cellframe-sdk C)
 cmake_minimum_required(VERSION 3.0)
 
 set(CMAKE_C_STANDARD 11)
-set(CELLFRAME_SDK_NATIVE_VERSION "2.8-18")
+set(CELLFRAME_SDK_NATIVE_VERSION "2.8-19")
 add_definitions ("-DCELLFRAME_SDK_VERSION=\"${CELLFRAME_SDK_NATIVE_VERSION}\"")
 set(DAPSDK_MODULES "")
 

cmake/OS_Detection.cmake

@@ -69,12 +69,13 @@ if(UNIX)
     endif()
     if (BSD)
         if(DAP_DEBUG)
-	  set(_CCOPT "-L/usr/local/lib -I/usr/local/include -DDAP_DEBUG -Wall -Wno-deprecated-declarations -Wno-unused-local-typedefs -Wno-unused-function -Wno-implicit-fallthrough -Wno-unused-variable -Wno-unused-parameter -pg -g3 -ggdb -fno-eliminate-unused-debug-symbols -fno-strict-aliasing")
+	  set(_CCOPT "-L/usr/local/lib -I/usr/local/include -DDAP_DEBUG -Wall -Wno-unused-command-line-argument -Wno-deprecated-declarations -Wno-unused-local-typedefs -Wno-unused-function -Wno-implicit-fallthrough -Wno-unused-variable -Wno-unused-parameter -pg -g3 -ggdb -fno-eliminate-unused-debug-symbols -fno-strict-aliasing")
           set(_LOPT "-pg -L/usr/local/lib ")
 	  SET(CMAKE_SHARED_LINKER_FLAGS "${CMAKE_SHARED_LINKER_FLAGS} -pg")
         else()
-          set(_CCOPT "-L/usr/local/lib -I/usr/local/include -Wno-deprecated-declarations -Wno-unused-local-typedefs -Wno-unused-function -Wno-implicit-fallthrough -Wno-unused-variable -Wno-unused-parameter -O3 -fPIC -fno-strict-aliasing -fno-ident -ffast-math -ftree-vectorize -fno-asynchronous-unwind-tables -ffunction-sections -std=gnu11")
-          set(_LOPT "-L/usr/local/lib ")
+          set(_CCOPT "-L /usr/local/lib -I/usr/local/include -Wno-deprecated-declarations -Wno-unused-command-line-argument -Wno-unused-local-typedefs -Wno-unused-function -Wno-implicit-fallthrough -Wno-unused-variable -Wno-unused-parameter -O3 -fPIC -fno-strict-aliasing -fno-ident -ffast-math -ftree-vectorize -fno-asynchronous-unwind-tables -ffunction-sections -std=gnu11")
+          set(_LOPT "-L /usr/local/lib ")
+	  SET(CMAKE_SHARED_LINKER_FLAGS "${CMAKE_SHARED_LINKER_FLAGS} -L/usr/local/lib")
         endif()
     endif()
 

dap-sdk/net/core/dap_proc_thread.c

@@ -218,7 +218,7 @@ int dap_proc_thread_assign_esocket_unsafe(dap_proc_thread_t * a_thread, dap_even
             l_fflags |= NOTE_WRITE;
 
         EV_SET(&a_esocket->kqueue_event , a_esocket->socket, l_filter, EV_ADD| l_flags | EV_CLEAR, l_fflags,0, a_esocket);
-        return kevent ( a_worker->kqueue_fd,&a_esocket->kqueue_event,1,NULL,0,NULL)==1 ? 0 : -1 ;
+        return kevent ( a_thread->kqueue_fd,&a_esocket->kqueue_event,1,NULL,0,NULL)==1 ? 0 : -1 ;
 
 #else
 #error "Unimplemented new esocket on worker callback for current platform"

dap-sdk/net/core/dap_server.c

@@ -32,11 +32,13 @@
 #include <sys/epoll.h>
 #include <netdb.h>
 #include <sys/timerfd.h>
+#include <sys/un.h>
 #elif defined (DAP_OS_BSD)
 #include <arpa/inet.h>
 #include <netinet/in.h>
 #include <sys/socket.h>
 #include <sys/event.h>
+#include <sys/un.h>
 #include <netdb.h>
 #endif
 
@@ -73,7 +75,7 @@
 
 static dap_events_socket_t * s_es_server_create(dap_events_t * a_events, int a_sock,
                                              dap_events_socket_callbacks_t * a_callbacks, dap_server_t * a_server);
-
+static int s_server_run(dap_server_t * a_server, dap_events_socket_callbacks_t *a_callbacks );
 static void s_es_server_accept(dap_events_socket_t *a_es, int a_remote_socket, struct sockaddr* a_remote_addr);
 static void s_es_server_error(dap_events_socket_t *a_es, int a_arg);
 static void s_es_server_new(dap_events_socket_t *a_es, void * a_arg);
@@ -118,6 +120,58 @@ void dap_server_delete(dap_server_t *a_server)
     DAP_DELETE(a_server);
 }
 
+/**
+ * @brief dap_server_new_local
+ * @param a_events
+ * @param a_path
+ * @param a_mode
+ * @param a_callbacks
+ * @return
+ */
+dap_server_t* dap_server_new_local(dap_events_t *a_events, const char * a_path, const char* a_mode, dap_events_socket_callbacks_t *a_callbacks)
+{
+    assert(a_events);
+#ifdef DAP_OS_UNIX
+    dap_server_t *l_server =  DAP_NEW_Z(dap_server_t);
+    l_server->socket_listener=-1; // To diff it from 0 fd
+    l_server->type = SERVER_LOCAL;
+    l_server->events = a_events;
+    l_server->socket_listener = socket(AF_LOCAL, SOCK_STREAM, 0);
+    if (l_server->socket_listener < 0) {
+        int l_errno = errno;
+        log_it (L_ERROR,"Socket error %s (%d)",strerror(l_errno), l_errno);
+        DAP_DELETE(l_server);
+        return NULL;
+    }
+
+    log_it(L_NOTICE,"Listen socket %d created...", l_server->socket_listener);
+
+    // Set path
+    if(a_path){
+        l_server->listener_path.sun_family =  AF_UNIX;
+        strncpy(l_server->listener_path.sun_path,a_path,sizeof(l_server->listener_path.sun_path)-1);
+        if ( access( a_path , R_OK) != -1 )
+            unlink( a_path );
+    }
+
+    mode_t l_listen_unix_socket_permissions = 0770;
+    if (a_mode){
+        dap_sscanf(a_mode,"%ou", &l_listen_unix_socket_permissions );
+    }
+
+
+    if(s_server_run(l_server,a_callbacks)==0){
+        if(a_path)
+            chmod(a_path,l_listen_unix_socket_permissions);
+        return l_server;
+    }else
+        return NULL;
+#else
+    log_it(L_ERROR, "Local server is not implemented for your platform");
+    return NULL;
+#endif
+}
+
 /**
  * @brief dap_server_new
  * @param a_events
@@ -136,10 +190,11 @@ dap_server_t* dap_server_new(dap_events_t *a_events, const char * a_addr, uint16
     l_server->address = a_addr ? strdup(a_addr) : strdup("0.0.0.0"); // If NULL we listen everything
     l_server->port = a_port;
     l_server->type = a_type;
+    l_server->events = a_events;
 
-    if(l_server->type == DAP_SERVER_TCP)
+    if(l_server->type == SERVER_TCP)
         l_server->socket_listener = socket(AF_INET, SOCK_STREAM, 0);
-    else if (l_server->type == DAP_SERVER_UDP)
+    else if (l_server->type == SERVER_UDP)
         l_server->socket_listener = socket(AF_INET, SOCK_DGRAM, 0);
 #ifdef DAP_OS_WINDOWS
     if (l_server->socket_listener == INVALID_SOCKET) {
@@ -169,28 +224,52 @@ dap_server_t* dap_server_new(dap_events_t *a_events, const char * a_addr, uint16
     l_server->listener_addr.sin_port = htons(l_server->port);
     inet_pton(AF_INET, l_server->address, &(l_server->listener_addr.sin_addr));
 
-    if(bind (l_server->socket_listener, (struct sockaddr *) &(l_server->listener_addr), sizeof(l_server->listener_addr)) < 0) {
+    if(s_server_run(l_server,a_callbacks)==0)
+        return l_server;
+    else
+        return NULL;
+}
+
+/**
+ * @brief s_server_run
+ * @param a_server
+ * @param a_callbacks
+ */
+static int s_server_run(dap_server_t * a_server, dap_events_socket_callbacks_t *a_callbacks )
+{
+    assert(a_server);
+
+    struct sockaddr * l_listener_addr = a_server->type == SERVER_LOCAL ?
+                                        (struct sockaddr *) &(a_server->listener_path) :
+                                        (struct sockaddr *) &(a_server->listener_addr);
+    socklen_t l_listener_addr_len = a_server->type == SERVER_LOCAL ?
+                                        sizeof(a_server->listener_path) :
+                                        sizeof(a_server->listener_addr);
+
+    if(bind (a_server->socket_listener, l_listener_addr, l_listener_addr_len) < 0) {
 #ifdef DAP_OS_WINDOWS
         log_it(L_ERROR,"Bind error: %d", WSAGetLastError());
-        closesocket(l_server->socket_listener);
+        closesocket(a_server->socket_listener);
 #else
         log_it(L_ERROR,"Bind error: %s",strerror(errno));
-        close(l_server->socket_listener);
+        close(a_server->socket_listener);
+        if ( errno == EACCES ) // EACCES=13
+            log_it( L_ERROR, "Server can't start. Permission denied");
 #endif
-        DAP_DELETE(l_server);
-        return NULL;
+        DAP_DELETE(a_server);
+        return -1;
     } else {
-        log_it(L_INFO,"Binded %s:%u",l_server->address,l_server->port);
-        listen(l_server->socket_listener, SOMAXCONN);
+        log_it(L_INFO,"Binded %s:%u",a_server->address,a_server->port);
+        listen(a_server->socket_listener, SOMAXCONN);
     }
 #ifdef DAP_OS_WINDOWS
      u_long l_mode = 1;
-     ioctlsocket(l_server->socket_listener, (long)FIONBIO, &l_mode);
+     ioctlsocket(a_server->socket_listener, (long)FIONBIO, &l_mode);
 #else
-    fcntl( l_server->socket_listener, F_SETFL, O_NONBLOCK);
+    fcntl( a_server->socket_listener, F_SETFL, O_NONBLOCK);
 #endif
-    pthread_mutex_init(&l_server->started_mutex,NULL);
-    pthread_cond_init(&l_server->started_cond,NULL);
+    pthread_mutex_init(&a_server->started_mutex,NULL);
+    pthread_cond_init(&a_server->started_cond,NULL);
 
 
 
@@ -211,45 +290,45 @@ dap_server_t* dap_server_new(dap_events_t *a_events, const char * a_addr, uint16
     for(size_t l_worker_id = 0; l_worker_id < dap_events_worker_get_count() ; l_worker_id++){
         dap_worker_t *l_w = dap_events_worker_get(l_worker_id);
         assert(l_w);
-        dap_events_socket_t * l_es = dap_events_socket_wrap2( l_server, a_events, l_server->socket_listener, &l_callbacks);
-        l_server->es_listeners = dap_list_append(l_server->es_listeners, l_es);
+        dap_events_socket_t * l_es = dap_events_socket_wrap2( a_server, a_events, a_server->socket_listener, &l_callbacks);
+        a_server->es_listeners = dap_list_append(a_server->es_listeners, l_es);
 
         if (l_es) {
-            l_es->type = l_server->type == DAP_SERVER_TCP ? DESCRIPTOR_TYPE_SOCKET_LISTENING : DESCRIPTOR_TYPE_SOCKET_UDP;
+            l_es->type = a_server->type == SERVER_TCP ? DESCRIPTOR_TYPE_SOCKET_LISTENING : DESCRIPTOR_TYPE_SOCKET_UDP;
             // Prepare for multi thread listening
             l_es->ev_base_flags = EPOLLIN;
 #ifdef EPOLLEXCLUSIVE
             l_es->ev_base_flags |= EPOLLET | EPOLLEXCLUSIVE;
 #endif
-            l_es->_inheritor = l_server;
-            pthread_mutex_lock(&l_server->started_mutex);
+            l_es->_inheritor = a_server;
+            pthread_mutex_lock(&a_server->started_mutex);
             dap_worker_add_events_socket( l_es, l_w );
-            pthread_cond_wait(&l_server->started_cond, &l_server->started_mutex);
-            pthread_mutex_unlock(&l_server->started_mutex);
+            pthread_cond_wait(&a_server->started_cond, &a_server->started_mutex);
+            pthread_mutex_unlock(&a_server->started_mutex);
         } else{
             log_it(L_WARNING, "Can't wrap event socket for %s:%u server", a_addr, a_port);
-            return NULL;
+            return -2;
         }
     }
 #else
     // or not
     dap_worker_t *l_w = dap_events_worker_get_auto();
     assert(l_w);
-    dap_events_socket_t * l_es = dap_events_socket_wrap2( l_server, a_events, l_server->socket_listener, &l_callbacks);
+    dap_events_socket_t * l_es = dap_events_socket_wrap2( a_server, a_server->events, a_server->socket_listener, &l_callbacks);
     if (l_es) {
-        l_server->es_listeners = dap_list_append(l_server->es_listeners, l_es);
-        l_es->type = l_server->type == DAP_SERVER_TCP ? DESCRIPTOR_TYPE_SOCKET_LISTENING : DESCRIPTOR_TYPE_SOCKET_UDP;
-        l_es->_inheritor = l_server;
-        pthread_mutex_lock(&l_server->started_mutex);
+        a_server->es_listeners = dap_list_append(a_server->es_listeners, l_es);
+        l_es->type = a_server->type == SERVER_TCP ? DESCRIPTOR_TYPE_SOCKET_LISTENING : DESCRIPTOR_TYPE_SOCKET_UDP;
+        l_es->_inheritor = a_server;
+        pthread_mutex_lock(&a_server->started_mutex);
         dap_worker_add_events_socket( l_es, l_w );
-        pthread_cond_wait(&l_server->started_cond, &l_server->started_mutex);
-        pthread_mutex_unlock(&l_server->started_mutex);
+        pthread_cond_wait(&a_server->started_cond, &a_server->started_mutex);
+        pthread_mutex_unlock(&a_server->started_mutex);
     } else {
-        log_it(L_WARNING, "Can't wrap event socket for %s:%u server", a_addr, a_port);
-        return NULL;
+        log_it(L_WARNING, "Can't wrap event socket server");
+        return -3;
     }
 #endif
-    return  l_server;
+    return 0;
 }
 
 /**

dap-sdk/net/core/include/dap_events_socket.h

@@ -266,6 +266,13 @@ typedef struct dap_events_socket_handler{
     uint128_t uuid;
 } dap_events_socket_handler_t;
 
+typedef struct dap_events_socket_handler_hh{
+    dap_events_socket_t * esocket;
+    uint128_t uuid;
+    uint32_t worker_id;
+    UT_hash_handle hh;
+} dap_events_socket_handler_hh_t;
+
 #ifdef __cplusplus
 extern "C" {
 #endif

dap-sdk/net/core/include/dap_server.h

@@ -25,6 +25,10 @@
 #pragma once
 
 #include "dap_common.h"
+
+#ifdef DAP_OS_UNIX
+#include <sys/un.h>
+#endif
 #if defined( DAP_OS_LINUX)
 
 #include <netinet/in.h>
@@ -53,7 +57,7 @@
 #include "dap_cpu_monitor.h"
 #include "dap_events_socket.h"
 
-typedef enum dap_server_type {DAP_SERVER_TCP, DAP_SERVER_UDP} dap_server_type_t;
+typedef enum dap_server_type {SERVER_TCP, SERVER_UDP,SERVER_LOCAL} dap_server_type_t;
 
 
 
@@ -73,8 +77,14 @@ typedef struct dap_server {
 #endif
   dap_list_t *es_listeners;
 
+  dap_events_t * events;
+
   struct sockaddr_in listener_addr; // Kernel structure for listener's binded address
 
+#ifdef DAP_OS_UNIX
+  struct sockaddr_un listener_path; // Path to UNIX socket
+#endif
+
   void *_inheritor;  // Pointer to the internal data, HTTP for example
 
   dap_cpu_stats_t cpu_stats;
@@ -91,4 +101,6 @@ int dap_server_init( ); // Init server module
 void  dap_server_deinit( void ); // Deinit server module
 
 dap_server_t* dap_server_new(dap_events_t *a_events, const char * a_addr, uint16_t a_port, dap_server_type_t a_type, dap_events_socket_callbacks_t *a_callbacks);
+dap_server_t* dap_server_new_local(dap_events_t *a_events, const char * a_path, const char * a_mode, dap_events_socket_callbacks_t *a_callbacks);
+
 void dap_server_delete(dap_server_t *a_server);

dap-sdk/net/server/notify_server/CMakeLists.txt

@@ -4,7 +4,6 @@ project(dap_notify_srv C)
 file(GLOB DAP_NOTIFY_SRV_SRCS FILES src/*.c)
 file(GLOB DAP_NOTIFY_SRV_HDRS FILES include/*.h)
 
-
 add_library(${PROJECT_NAME} STATIC ${DAP_NOTIFY_SRV_SRCS} ${DAP_NOTIFY_SRV_HDRS})
 target_include_directories(${PROJECT_NAME} INTERFACE . include/)
 target_include_directories(${PROJECT_NAME} PUBLIC include)

dap-sdk/net/server/notify_server/include/dap_notify_srv.h

@@ -21,9 +21,9 @@
     along with any DAP SDK based project.  If not, see <http://www.gnu.org/licenses/>.
 */
 #pragma once
+#include <stdint.h>
 
-int dap_notify_server_init(const char * a_notify_socket_path);
+int dap_notify_server_init();
 void dap_notify_server_deinit();
-struct dap_events_socket * dap_notify_server_create_inter();
-int dap_notify_server_send_f_inter(struct dap_events_socket * a_input, const char * a_format,...);
+int dap_notify_server_send_f_inter(uint32_t a_worker_id, const char * a_format,...);
 int dap_notify_server_send_f_mt(const char * a_format,...);

dap-sdk/net/server/notify_server/src/dap_notify_srv.c

@@ -21,45 +21,75 @@
     along with any DAP SDK based project.  If not, see <http://www.gnu.org/licenses/>.
 */
 
-#ifdef DAP_OS_UNIX
-#include <sys/socket.h>
-#include <sys/un.h>
-#elif DAP_OS_WINDOWS
-#endif
-
 #include <stdlib.h>
 #include <stdarg.h>
 #include <unistd.h>
 
 #include "dap_common.h"
+#include "dap_config.h"
+#include "dap_list.h"
 #include "dap_strfuncs.h"
 #include "dap_events_socket.h"
+#include "dap_server.h"
+#include "dap_events.h"
 #include "dap_notify_srv.h"
 
-dap_events_socket_t * s_notify_server = NULL;
+#define LOG_TAG "notify_server"
+
+
+dap_server_t * s_notify_server = NULL;
 dap_events_socket_t * s_notify_server_queue = NULL;
-static void s_notify_server_accept(dap_events_socket_t * a_es, int a_remote_socket, struct sockaddr* a_remote_addr );
-static void s_notify_server_inter_queue(dap_events_socket_t * a_es, void * a_arg);
+
+dap_events_socket_t ** s_notify_server_queue_inter = NULL;
+
+
+dap_events_socket_handler_hh_t * s_notify_server_clients = NULL;
+pthread_rwlock_t s_notify_server_clients_mutex = PTHREAD_RWLOCK_INITIALIZER;
+
+static void s_notify_server_callback_queue(dap_events_socket_t * a_es, void * a_arg);
+static void s_notify_server_callback_new(dap_events_socket_t * a_es, void * a_arg);
+static void s_notify_server_callback_delete(dap_events_socket_t * a_es, void * a_arg);
+
 
 /**
  * @brief dap_notify_server_init
  * @param a_notify_socket_path
  * @return
  */
-int dap_notify_server_init(const char * a_notify_socket_path)
+int dap_notify_server_init()
 {
-    dap_events_socket_callbacks_t l_callbacks={0};
-    l_callbacks.accept_callback = s_notify_server_accept;
-    s_notify_server = dap_events_socket_create(DESCRIPTOR_TYPE_SOCKET_LOCAL_LISTENING,&l_callbacks );
+    dap_events_socket_callbacks_t l_callbacks={.new_callback = s_notify_server_callback_new,
+                                              .delete_callback = s_notify_server_callback_delete };
+
+
+    const char * l_notify_socket_path = dap_config_get_item_str_default(g_config, "notify_server", "listen_path",NULL);
+    uint16_t l_notify_socket_path_mode = dap_config_get_item_uint16_default(g_config, "notify_server", "listen_path_mode",0600);
+
+    const char * l_notify_socket_address = dap_config_get_item_str_default(g_config, "notify_server", "listen_address",NULL);
+    uint16_t l_notify_socket_port = dap_config_get_item_uint16_default(g_config, "notify_server", "listen_port",0);
+
+    if(l_notify_socket_path){
+        s_notify_server = dap_server_new_local(dap_events_get_default(),l_notify_socket_path,l_notify_socket_path_mode , &l_callbacks);
+    }else if (l_notify_socket_address && l_notify_socket_port ){
+        s_notify_server = dap_server_new(dap_events_get_default(),l_notify_socket_address,
+                                            l_notify_socket_port, SERVER_TCP, &l_callbacks);
+    }else{
+        log_it(L_INFO,"Notify server is not configured, nothing to init but thats okay");
+        return 0;
+    }
+
     if (!s_notify_server)
         return -1;
 
-#ifdef DAP_OS_UNIX
-    struct sockaddr_un l_sock_un={0};
-    l_sock_un.sun_family = AF_LOCAL;
-    strncpy(l_sock_un.sun_path, a_notify_socket_path, sizeof(l_sock_un.sun_path) - 1);
-#endif
+    s_notify_server_queue = dap_events_socket_create_type_queue_ptr_mt(dap_events_worker_get_auto(),s_notify_server_callback_queue);
+    uint32_t l_workers_count = dap_events_worker_get_count();
+    s_notify_server_queue_inter = DAP_NEW_Z_SIZE(dap_events_socket_t*,sizeof (dap_events_socket_t*)*l_workers_count );
+    for(uint32_t i = 0; i < l_workers_count; i++){
+        s_notify_server_queue_inter[i] = dap_events_socket_queue_ptr_create_input(s_notify_server_queue);
+        dap_events_socket_assign_on_worker_mt(s_notify_server_queue_inter[i], dap_events_worker_get(i));
+    }
 
+    log_it(L_NOTICE,"Notify server initalized");
     return 0;
 }
 
@@ -86,14 +116,20 @@ struct dap_events_socket * dap_notify_server_create_inter()
  * @param a_format
  * @return
  */
-int dap_notify_server_send_f_inter(struct dap_events_socket * a_input, const char * a_format,...)
+int dap_notify_server_send_f_inter(uint32_t a_worker_id, const char * a_format,...)
 {
+    if(a_worker_id>= dap_events_worker_get_count()){
+        log_it(L_ERROR,"Wrong worker id %u for send_f_inter() function", a_worker_id);
+        return -10;
+    }
+    dap_events_socket_t * l_input = s_notify_server_queue_inter[a_worker_id];
     va_list va;
     va_start(va, a_format);
     size_t l_str_size=dap_vsnprintf(NULL,0,a_format,va);
     char * l_str = DAP_NEW_SIZE(char,l_str_size+1);
     dap_vsnprintf(l_str,l_str_size+1,a_format,va);
-    return dap_events_socket_queue_ptr_send_to_input(a_input,l_str);
+    va_end(va);
+    return dap_events_socket_queue_ptr_send_to_input(l_input,l_str);
 }
 
 /**
@@ -103,26 +139,81 @@ int dap_notify_server_send_f_inter(struct dap_events_socket * a_input, const cha
  */
 int dap_notify_server_send_f_mt(const char * a_format,...)
 {
-    return -1;
+    va_list va;
+    va_start(va, a_format);
+    size_t l_str_size=dap_vsnprintf(NULL,0,a_format,va);
+    char * l_str = DAP_NEW_SIZE(char,l_str_size+1);
+    dap_vsnprintf(l_str,l_str_size+1,a_format,va);
+    va_end(va);
+    return dap_events_socket_queue_ptr_send(s_notify_server_callback_queue ,l_str);
 }
 
 /**
- * @brief s_notify_server_accept
+ * @brief s_notify_server_inter_queue
  * @param a_es
- * @param a_remote_socket
- * @param a_remote_addr
+ * @param a_arg
  */
-static void s_notify_server_accept(dap_events_socket_t * a_es, int a_remote_socket, struct sockaddr* a_remote_addr )
+static void s_notify_server_callback_queue(dap_events_socket_t * a_es, void * a_arg)
 {
-
+    pthread_rwlock_rdlock(&s_notify_server_clients_mutex);
+    dap_events_socket_handler_hh_t * l_socket_handler = NULL,* l_tmp = NULL;
+    HASH_ITER(hh, s_notify_server_clients, l_socket_handler, l_tmp){
+        uint32_t l_worker_id = l_socket_handler->worker_id;
+        if(l_worker_id>= dap_events_worker_get_count()){
+            log_it(L_ERROR,"Wrong worker id %u for send_inter() function", l_worker_id);
+            continue;
+        }
+        size_t l_str_len = a_arg? strlen((char*)a_arg): 0;
+        if(l_str_len)
+            dap_events_socket_write_inter(a_es->worker->queue_es_io_input[l_worker_id],l_socket_handler->esocket,
+                                      a_arg,l_str_len);
+        DAP_DELETE(a_arg);
+    }
+    pthread_rwlock_unlock(&s_notify_server_clients_mutex);
 }
 
 /**
- * @brief s_notify_server_inter_queue
+ * @brief s_notify_server_callback_new
  * @param a_es
  * @param a_arg
  */
-static void s_notify_server_inter_queue(dap_events_socket_t * a_es, void * a_arg)
+static void s_notify_server_callback_new(dap_events_socket_t * a_es, void * a_arg)
 {
+    (void) a_arg;
+    dap_events_socket_handler_hh_t * l_hh_new;
+    pthread_rwlock_wrlock(&s_notify_server_clients_mutex);
+    HASH_FIND(hh,s_notify_server_clients, &a_es->uuid, sizeof (a_es->uuid), l_hh_new);
+    if (l_hh_new){
+        uint64_t *l_uuid_u64 =(uint64_t*) &a_es->uuid;
+        log_it(L_WARNING,"Trying to add notify client with uuid 0x%016"DAP_UINT64_FORMAT_X"%016"DAP_UINT64_FORMAT_X" but already present this UUID in list, updating only esocket pointer if so",l_uuid_u64 );
+        l_hh_new->esocket = a_es;
+        l_hh_new->worker_id = a_es->worker->id;
+    }else {
+        l_hh_new = DAP_NEW_Z(dap_events_socket_handler_hh_t);
+        l_hh_new->esocket = a_es;
+        l_hh_new->uuid = a_es->uuid;
+        l_hh_new->worker_id = a_es->worker->id;
+        HASH_ADD(hh, s_notify_server_clients, uuid, sizeof (l_hh_new->uuid), l_hh_new);
+    }
+    pthread_rwlock_unlock(&s_notify_server_clients_mutex);
+}
 
+/**
+ * @brief s_notify_server_callback_delete
+ * @param a_es
+ * @param a_arg
+ */
+static void s_notify_server_callback_delete(dap_events_socket_t * a_es, void * a_arg)
+{
+    (void) a_arg;
+    dap_events_socket_handler_hh_t * l_hh_new = NULL;
+    pthread_rwlock_wrlock(&s_notify_server_clients_mutex);
+    HASH_FIND(hh,s_notify_server_clients, &a_es->uuid, sizeof (a_es->uuid), l_hh_new);
+    if (l_hh_new){
+        HASH_DELETE(hh,s_notify_server_clients, l_hh_new);
+    }else{
+        uint64_t *l_uuid_u64 =(uint64_t*) &a_es->uuid;
+        log_it(L_WARNING,"Trying to remove notify client with uuid 0x%016"DAP_UINT64_FORMAT_X"%016"DAP_UINT64_FORMAT_X" but can't find such client in table",l_uuid_u64 );
+    }
+    pthread_rwlock_unlock(&s_notify_server_clients_mutex);
 }

modules/net/dap_chain_net.c

@@ -1757,10 +1757,10 @@ int s_net_load(const char * a_net_name, uint16_t a_acl_idx)
                     }
                     log_it(L_NOTICE,"GDB Info: node_addr: " NODE_ADDR_FP_STR"  links: %u cell_id: 0x%016X ",
                            NODE_ADDR_FP_ARGS(l_node_addr),
-                           PVT(l_net)->node_info->hdr.links_number,
-                           PVT(l_net)->node_info->hdr.cell_id.uint64);
+                           l_net_pvt->node_info->hdr.links_number,
+                           l_net_pvt->node_info->hdr.cell_id.uint64);
                     // save cell_id
-                    l_net->pub.cell_id.uint64 = PVT(l_net)->node_info->hdr.cell_id.uint64;
+                    l_net->pub.cell_id.uint64 = l_net_pvt->node_info->hdr.cell_id.uint64;
                 }
             }
             else{
@@ -1829,12 +1829,12 @@ int s_net_load(const char * a_net_name, uint16_t a_acl_idx)
 
         } else {
             log_it(L_ERROR,"Can't any chains for network %s",l_net->pub.name);
-            PVT(l_net)->load_mode = false;
+            l_net_pvt->load_mode = false;
 
             return -2;
         }
         // Do specific role actions post-chain created
-        switch ( PVT( l_net )->node_role.enums ) {
+        switch ( l_net_pvt->node_role.enums ) {
             case NODE_ROLE_ROOT_MASTER:{
                 // Set to process everything in datum pool
                 dap_chain_t * l_chain = NULL;
@@ -1848,7 +1848,7 @@ int s_net_load(const char * a_net_name, uint16_t a_acl_idx)
                 if (l_chain )
                    l_chain->is_datum_pool_proc = true;
 
-                PVT(l_net)->state_target = NET_STATE_ONLINE;
+                l_net_pvt->state_target = NET_STATE_ONLINE;
                 log_it(L_INFO,"Root node role established");
             } break;
             case NODE_ROLE_CELL_MASTER:
@@ -1873,12 +1873,12 @@ int s_net_load(const char * a_net_name, uint16_t a_acl_idx)
                 //    DAP_DELETE (l_proc_chains);
                 //l_proc_chains = NULL;
 
-                PVT(l_net)->state_target = NET_STATE_ONLINE;
+                l_net_pvt->state_target = NET_STATE_ONLINE;
                 log_it(L_INFO,"Master node role established");
             } break;
             case NODE_ROLE_FULL:{
                 log_it(L_INFO,"Full node role established");
-                PVT(l_net)->state_target = NET_STATE_ONLINE;
+                l_net_pvt->state_target = NET_STATE_ONLINE;
             } break;
             case NODE_ROLE_LIGHT:
             default:
@@ -1887,13 +1887,13 @@ int s_net_load(const char * a_net_name, uint16_t a_acl_idx)
         }
 
         if (s_seed_mode || !dap_config_get_item_bool_default(g_config ,"general", "auto_online",false ) ) { // If we seed we do everything manual. First think - prefil list of node_addrs and its aliases
-            PVT(l_net)->state_target = NET_STATE_OFFLINE;
+            l_net_pvt->state_target = NET_STATE_OFFLINE;
         }else{
-            PVT(l_net)->state = NET_STATE_LINKS_PREPARE;
+            l_net_pvt->state = NET_STATE_LINKS_PREPARE;
         }
 
-        PVT(l_net)->load_mode = false;
-        PVT(l_net)->flags |= F_DAP_CHAIN_NET_GO_SYNC;
+        l_net_pvt->load_mode = false;
+        l_net_pvt->flags |= F_DAP_CHAIN_NET_GO_SYNC;
 
         // Start the proc thread
         log_it(L_NOTICE, "Сhain network \"%s\" initialized",l_net_item->name);

modules/net/dap_chain_node_dns_server.c

@@ -261,7 +261,7 @@ void dap_dns_server_start(dap_events_t *a_ev, uint16_t a_port)
     s_dns_server = DAP_NEW_Z(dap_dns_server_t);
     dap_events_socket_callbacks_t l_cb = {};
     l_cb.read_callback = dap_dns_client_read;
-    s_dns_server->instance = dap_server_new(a_ev, NULL, a_port, DAP_SERVER_UDP, &l_cb);
+    s_dns_server->instance = dap_server_new(a_ev, NULL, a_port, SERVER_UDP, &l_cb);
     if (!s_dns_server->instance) {
         log_it(L_ERROR, "Can't start DNS server");
         return;