diff --git a/dap-sdk/net/client/dap_client_http.c b/dap-sdk/net/client/dap_client_http.c
index e47609b4fb53fe9c30afc377cae9bff0b00daca6..bfdee7bd165cef0ff9a06c6046e18b41cc3d414a 100644
--- a/dap-sdk/net/client/dap_client_http.c
+++ b/dap-sdk/net/client/dap_client_http.c
@@ -504,6 +504,7 @@ void* dap_client_http_request_custom(dap_worker_t * a_worker,const char *a_uplin
dap_events_socket_t *l_ev_socket = dap_events_socket_wrap_no_add(dap_events_get_default(), l_socket, &l_s_callbacks);
+ log_it(L_DEBUG,"Created client request socket %d", l_socket);
// create private struct
dap_client_http_pvt_t *l_http_pvt = DAP_NEW_Z(dap_client_http_pvt_t);
l_ev_socket->_inheritor = l_http_pvt;
diff --git a/dap-sdk/net/core/dap_events_socket.c b/dap-sdk/net/core/dap_events_socket.c
index 96af0b83be22ea0fce0f01e057e8a4a9b7d13558..19db40fb3df359acad8110093facb34023f31cb0 100644
--- a/dap-sdk/net/core/dap_events_socket.c
+++ b/dap-sdk/net/core/dap_events_socket.c
@@ -184,7 +184,7 @@ dap_events_socket_t *dap_events_socket_wrap_no_add( dap_events_t *a_events,
#elif defined(DAP_EVENTS_CAPS_KQUEUE)
l_ret->kqueue_event_catched_data.esocket = l_ret;
l_ret->kqueue_base_flags = 0;
- l_ret->kqueue_base_filter = EVFILT_EXCEPT;
+ l_ret->kqueue_base_filter = 0;
#endif
if ( a_sock!= 0 && a_sock != -1){
@@ -386,7 +386,8 @@ dap_events_socket_t * dap_events_socket_create(dap_events_desc_type_t a_type, da
return NULL;
}
l_es->type = a_type ;
-
+ if(s_debug_reactor)
+ log_it(L_DEBUG,"Created socket %d type %d", l_sock,l_es->type);
return l_es;
}
@@ -1278,19 +1279,24 @@ int dap_events_socket_queue_ptr_send( dap_events_socket_t * a_es, void* a_arg)
EV_SET(&l_event,a_es->socket, EVFILT_USER,0, NOTE_TRIGGER ,0, l_es_w_data);
int l_n;
if(a_es->pipe_out){ // If we have pipe out - we send events directly to the pipe out kqueue fd
- if(a_es->pipe_out->worker)
+ if(a_es->pipe_out->worker){
+ if( s_debug_reactor) log_it(L_DEBUG, "Sent kevent() with ptr %p to pipe_out worker on esocket %d",a_arg,a_es);
l_n = kevent(a_es->pipe_out->worker->kqueue_fd,&l_event,1,NULL,0,NULL);
- else if (a_es->pipe_out->proc_thread)
+ }else if (a_es->pipe_out->proc_thread){
l_n = kevent(a_es->pipe_out->proc_thread->kqueue_fd,&l_event,1,NULL,0,NULL);
+ if( s_debug_reactor) log_it(L_DEBUG, "Sent kevent() with ptr %p to pipe_out proc_thread on esocket %d",a_arg,a_es);
+ }
else {
log_it(L_WARNING,"Trying to send pointer in pipe out queue thats not assigned to any worker or proc thread");
l_n = 0;
}
- }else if(a_es->worker)
+ }else if(a_es->worker){
l_n = kevent(a_es->worker->kqueue_fd,&l_event,1,NULL,0,NULL);
- else if (a_es->proc_thread)
+ if( s_debug_reactor) log_it(L_DEBUG, "Sent kevent() with ptr %p to worker on esocket %d",a_arg,a_es);
+ }else if (a_es->proc_thread){
l_n = kevent(a_es->proc_thread->kqueue_fd,&l_event,1,NULL,0,NULL);
- else {
+ if( s_debug_reactor) log_it(L_DEBUG, "Sent kevent() with ptr %p to proc_thread on esocket %d",a_arg,a_es);
+ }else {
log_it(L_WARNING,"Trying to send pointer in queue thats not assigned to any worker or proc thread");
l_n = 0;
}
@@ -1499,29 +1505,45 @@ void dap_events_socket_worker_poll_update_unsafe(dap_events_socket_t * a_esocket
u_short l_flags =a_esocket->kqueue_base_flags;
u_int l_fflags =a_esocket->kqueue_base_fflags;
-
- // Check & add
- if( a_esocket->flags & DAP_SOCK_READY_TO_READ )
- EV_SET(l_event, a_esocket->socket, EVFILT_READ,l_flags| EV_ADD,l_fflags,a_esocket->kqueue_data,a_esocket);
-
- if( a_esocket->flags & DAP_SOCK_READY_TO_WRITE || a_esocket->flags &DAP_SOCK_CONNECTING )
- l_filter |= EVFILT_WRITE;
-
- if (a_esocket->type == DESCRIPTOR_TYPE_EVENT || a_esocket->type == DESCRIPTOR_TYPE_QUEUE)
+ int l_kqueue_fd = a_esocket->worker? a_esocket->worker->kqueue_fd :
+ a_esocket->proc_thread ? a_esocket->proc_thread->kqueue_fd : -1;
+ if ( l_kqueue_fd == -1 ){
+ log_it(L_ERROR, "Esocket is not assigned with anything ,exit");
+ }
+ // Check & add
+ bool l_is_error=false;
+ int l_errno=0;
+ if (a_esocket->type == DESCRIPTOR_TYPE_EVENT || a_esocket->type == DESCRIPTOR_TYPE_QUEUE){
EV_SET(l_event, a_esocket->socket, EVFILT_USER,EV_ADD| EV_CLEAR ,0,0, &a_esocket->kqueue_event_catched_data );
- else
+ if( kevent( l_kqueue_fd,l_event,1,NULL,0,NULL)!=0){
+ l_is_error = true;
+ l_errno = errno;
+ }
+ }else{
EV_SET(l_event, a_esocket->socket, l_filter,l_flags| EV_ADD,l_fflags,a_esocket->kqueue_data,a_esocket);
-
-
- if( a_esocket->worker){
- if ( kevent(a_esocket->worker->kqueue_fd,l_event,1,NULL,0,NULL)!=1 ){
- int l_errno = errno;
- char l_errbuf[128];
- l_errbuf[0]=0;
- strerror_r(l_errno, l_errbuf, sizeof (l_errbuf));
- log_it(L_ERROR,"Can't update client socket state on kqueue fd %d: \"%s\" (%d)",
- a_esocket->worker->kqueue_fd, l_errbuf, l_errno);
- }
+ if( a_esocket->flags & DAP_SOCK_READY_TO_READ ){
+ EV_SET(l_event, a_esocket->socket, EVFILT_READ,l_flags| EV_ADD,l_fflags,a_esocket->kqueue_data,a_esocket);
+ if( kevent( l_kqueue_fd,l_event,1,NULL,0,NULL) != 1 ){
+ l_is_error = true;
+ l_errno = errno;
+ }
+ }
+ if( !l_is_error){
+ if( a_esocket->flags & DAP_SOCK_READY_TO_WRITE || a_esocket->flags &DAP_SOCK_CONNECTING ){
+ EV_SET(l_event, a_esocket->socket, EVFILT_WRITE,l_flags| EV_ADD,l_fflags,a_esocket->kqueue_data,a_esocket);
+ if(kevent( l_kqueue_fd,l_event,1,NULL,0,NULL) != 1){
+ l_is_error = true;
+ l_errno = errno;
+ }
+ }
+ }
+ }
+ if ( l_is_error ){
+ char l_errbuf[128];
+ l_errbuf[0]=0;
+ strerror_r(l_errno, l_errbuf, sizeof (l_errbuf));
+ log_it(L_ERROR,"Can't update client socket state on kqueue fd %d: \"%s\" (%d)",
+ l_kqueue_fd, l_errbuf, l_errno);
}
}
@@ -1562,7 +1584,7 @@ void dap_events_socket_set_readable_unsafe( dap_events_socket_t *a_esocket, bool
char l_errbuf[128];
l_errbuf[0]=0;
strerror_r(l_errno, l_errbuf, sizeof (l_errbuf));
- log_it(L_ERROR,"Can't update client socket state on kqueue fd for set_read op %d: \"%s\" (%d)",
+ log_it(L_ERROR," for set_read op %d: \"%s\" (%d)",
l_kqueue_fd, l_errbuf, l_errno);
}
}
@@ -1587,6 +1609,7 @@ void dap_events_socket_set_writable_unsafe( dap_events_socket_t *a_esocket, bool
if ( a_is_ready == (bool)(a_esocket->flags & DAP_SOCK_READY_TO_WRITE)) {
return;
}
+
if ( a_is_ready ) {
a_esocket->flags |= DAP_SOCK_READY_TO_WRITE;
#ifdef DAP_EVENTS_CAPS_EPOLL
@@ -1609,18 +1632,20 @@ void dap_events_socket_set_writable_unsafe( dap_events_socket_t *a_esocket, bool
a_esocket->type != DESCRIPTOR_TYPE_TIMER ){
struct kevent l_event;
uint16_t l_op_flag = a_is_ready? EV_ADD : EV_DELETE;
+ int l_expected_reply = a_is_ready? 1: 0;
EV_SET(&l_event, a_esocket->socket, EVFILT_WRITE,
a_esocket->kqueue_base_flags | l_op_flag,a_esocket->kqueue_base_fflags ,
a_esocket->kqueue_data,a_esocket);
int l_kqueue_fd = a_esocket->worker? a_esocket->worker->kqueue_fd :
a_esocket->proc_thread ? a_esocket->proc_thread->kqueue_fd : -1;
if( l_kqueue_fd>0 ){
- if ( kevent(l_kqueue_fd,&l_event,1,NULL,0,NULL)!=1 ){
+ int l_kevent_ret=kevent(l_kqueue_fd,&l_event,1,NULL,0,NULL);
+ if ( l_kevent_ret!=l_expected_reply ){
int l_errno = errno;
char l_errbuf[128];
l_errbuf[0]=0;
strerror_r(l_errno, l_errbuf, sizeof (l_errbuf));
- log_it(L_ERROR,"Can't update client socket state on kqueue fd for set_read op %d: \"%s\" (%d)",
+ log_it(L_ERROR,"Can't update client socket state on kqueue fd for set_write op %d: \"%s\" (%d)",
l_kqueue_fd, l_errbuf, l_errno);
}
}
@@ -1662,8 +1687,13 @@ void dap_events_socket_remove_and_delete_unsafe( dap_events_socket_t *a_es, bool
*/
void dap_events_socket_delete_unsafe( dap_events_socket_t * a_esocket , bool a_preserve_inheritor)
{
+#ifdef DAP_EVENTS_CAPS_KQUEUE
+ if( a_esocket->type != DESCRIPTOR_TYPE_EVENT
+ && a_esocket->type != DESCRIPTOR_TYPE_TIMER
+ && a_esocket->type != DESCRIPTOR_TYPE_QUEUE )
+#endif
if (a_esocket->events){ // It could be socket NOT from events
- if (a_esocket->socket != -1){
+ if (a_esocket->socket != -1 && a_esocket->socket != 0 ){
if(!dap_events_socket_find_unsafe(a_esocket->socket, a_esocket->events)){
if(s_debug_reactor)
log_it(L_ERROR, "esocket %d type %d already deleted", a_esocket->socket, a_esocket->type);
@@ -1727,14 +1757,42 @@ void dap_events_socket_remove_from_worker_unsafe( dap_events_socket_t *a_es, dap
#elif defined(DAP_EVENTS_CAPS_KQUEUE)
if (a_es->socket != -1 && a_es->type != DESCRIPTOR_TYPE_TIMER){
struct kevent * l_event = &a_es->kqueue_event;
- EV_SET(l_event, a_es->socket, 0 ,EV_DELETE, 0,0,a_es);
- if ( kevent( a_worker->kqueue_fd,l_event,1,NULL,0,NULL) != 1 ) {
- int l_errno = errno;
- char l_errbuf[128];
- strerror_r(l_errno, l_errbuf, sizeof (l_errbuf));
- log_it( L_ERROR,"Can't remove event socket's handler %d from the epoll_fd %d \"%s\" (%d)", a_es->socket,
- a_worker->kqueue_fd, l_errbuf, l_errno);
- }
+ if (a_es->kqueue_base_filter){
+ EV_SET(l_event, a_es->socket, a_es->kqueue_base_filter ,EV_DELETE, 0,0,a_es);
+ if ( kevent( a_worker->kqueue_fd,l_event,1,NULL,0,NULL) != 1 ) {
+ int l_errno = errno;
+ char l_errbuf[128];
+ strerror_r(l_errno, l_errbuf, sizeof (l_errbuf));
+ log_it( L_ERROR,"Can't remove event socket's handler %d from the kqueue %d filter %d \"%s\" (%d)", a_es->socket,
+ a_worker->kqueue_fd,a_es->kqueue_base_filter, l_errbuf, l_errno);
+ }
+ }else{
+ EV_SET(l_event, a_es->socket, EVFILT_EXCEPT ,EV_DELETE, 0,0,a_es);
+ kevent( a_worker->kqueue_fd,l_event,1,NULL,0,NULL); // If this filter is not set up - no warnings
+
+
+ if(a_es->flags & DAP_SOCK_READY_TO_WRITE){
+ EV_SET(l_event, a_es->socket, EVFILT_WRITE ,EV_DELETE, 0,0,a_es);
+ if ( kevent( a_worker->kqueue_fd,l_event,1,NULL,0,NULL) != 0 ) {
+ int l_errno = errno;
+ char l_errbuf[128];
+ strerror_r(l_errno, l_errbuf, sizeof (l_errbuf));
+ log_it( L_ERROR,"Can't remove event socket's handler %d from the kqueue %d filter EVFILT_WRITE \"%s\" (%d)", a_es->socket,
+ a_worker->kqueue_fd, l_errbuf, l_errno);
+ }
+ }
+ if(a_es->flags & DAP_SOCK_READY_TO_READ){
+ EV_SET(l_event, a_es->socket, EVFILT_READ ,EV_DELETE, 0,0,a_es);
+ if ( kevent( a_worker->kqueue_fd,l_event,1,NULL,0,NULL) != 0 ) {
+ int l_errno = errno;
+ char l_errbuf[128];
+ strerror_r(l_errno, l_errbuf, sizeof (l_errbuf));
+ log_it( L_ERROR,"Can't remove event socket's handler %d from the kqueue %d filter EVFILT_READ \"%s\" (%d)", a_es->socket,
+ a_worker->kqueue_fd, l_errbuf, l_errno);
+ }
+ }
+
+ }
}
#elif defined (DAP_EVENTS_CAPS_POLL)
if (a_es->poll_index < a_worker->poll_count ){
diff --git a/dap-sdk/net/core/dap_proc_thread.c b/dap-sdk/net/core/dap_proc_thread.c
index a9222a590aa0419cd267fa65a6fbafb7ca592cef..e64b8db9fd66c14e825d352040efd0fecad10630 100644
--- a/dap-sdk/net/core/dap_proc_thread.c
+++ b/dap-sdk/net/core/dap_proc_thread.c
@@ -293,26 +293,46 @@ int dap_proc_thread_esocket_update_poll_flags(dap_proc_thread_t * a_thread, dap_
u_short l_flags = a_esocket->kqueue_base_flags;
u_int l_fflags = a_esocket->kqueue_base_fflags;
short l_filter = a_esocket->kqueue_base_filter;
- if(a_esocket->flags & DAP_SOCK_READY_TO_READ )
- l_filter |= EVFILT_READ;
- if(a_esocket->flags & DAP_SOCK_READY_TO_WRITE )
- l_filter |= EVFILT_WRITE;
- if (a_esocket->type == DESCRIPTOR_TYPE_EVENT || a_esocket->type == DESCRIPTOR_TYPE_QUEUE)
- EV_SET(&a_esocket->kqueue_event , a_esocket->socket, EVFILT_USER,EV_ADD|EV_CLEAR ,0,
- a_esocket->kqueue_data,
- &a_esocket->kqueue_event_catched_data );
- else
- EV_SET(&a_esocket->kqueue_event , a_esocket->socket, l_filter, EV_ADD| l_flags, l_fflags,
- a_esocket->kqueue_data, a_esocket);
- int l_ret = kevent ( a_thread->kqueue_fd,&a_esocket->kqueue_event,1,NULL,0,NULL);
- if((l_ret != -1) && ( l_ret != 0 &&
- (a_esocket->type == DESCRIPTOR_TYPE_QUEUE||(a_esocket->type == DESCRIPTOR_TYPE_EVENT )) ) ) {
- int l_errno = errno;
+ int l_kqueue_fd = a_esocket->proc_thread ? a_esocket->proc_thread->kqueue_fd : -1;
+ if ( l_kqueue_fd == -1 ){
+ log_it(L_ERROR, "Esocket is not assigned with anything ,exit");
+ }
+ struct kevent * l_event = &a_esocket->kqueue_event;
+ // Check & add
+ bool l_is_error=false;
+ int l_errno=0;
+ if (a_esocket->type == DESCRIPTOR_TYPE_EVENT || a_esocket->type == DESCRIPTOR_TYPE_QUEUE){
+ EV_SET(l_event, a_esocket->socket, EVFILT_USER,EV_ADD| EV_CLEAR ,0,0, &a_esocket->kqueue_event_catched_data );
+ if( kevent( l_kqueue_fd,l_event,1,NULL,0,NULL)!=0){
+ l_is_error = true;
+ l_errno = errno;
+ }
+ }else{
+ EV_SET(l_event, a_esocket->socket, l_filter,l_flags| EV_ADD,l_fflags,a_esocket->kqueue_data,a_esocket);
+ if( a_esocket->flags & DAP_SOCK_READY_TO_READ ){
+ EV_SET(l_event, a_esocket->socket, EVFILT_READ,l_flags| EV_ADD,l_fflags,a_esocket->kqueue_data,a_esocket);
+ if( kevent( l_kqueue_fd,l_event,1,NULL,0,NULL) != 1 ){
+ l_is_error = true;
+ l_errno = errno;
+ }
+ }
+ if( !l_is_error){
+ if( a_esocket->flags & DAP_SOCK_READY_TO_WRITE || a_esocket->flags &DAP_SOCK_CONNECTING ){
+ EV_SET(l_event, a_esocket->socket, EVFILT_WRITE,l_flags| EV_ADD,l_fflags,a_esocket->kqueue_data,a_esocket);
+ if(kevent( l_kqueue_fd,l_event,1,NULL,0,NULL) != 1){
+ l_is_error = true;
+ l_errno = errno;
+ }
+ }
+ }
+ }
+
+ if ( l_is_error){
char l_errbuf[128];
- l_errbuf[0]='\0';
- strerror_r(l_errno,l_errbuf,sizeof (l_errbuf));
- log_it(L_CRITICAL, "Can't add descriptor in proc thread kqueue , err: %d (%s)", l_errno, l_errbuf);
- return -1;
+ l_errbuf[0]=0;
+ strerror_r(l_errno, l_errbuf, sizeof (l_errbuf));
+ log_it(L_ERROR,"Can't update client socket state on kqueue fd %d: \"%s\" (%d)",
+ l_kqueue_fd, l_errbuf, l_errno);
}
#else
diff --git a/dap-sdk/net/core/dap_worker.c b/dap-sdk/net/core/dap_worker.c
index ee913ea3498f6bd9c3b97b446a2c2177b933b591..973748a25fe8f450818c6c9eca508ee701b02497 100644
--- a/dap-sdk/net/core/dap_worker.c
+++ b/dap-sdk/net/core/dap_worker.c
@@ -798,12 +798,24 @@ static void s_queue_add_es_callback( dap_events_socket_t * a_es, void * a_arg)
}
if(s_debug_reactor)
- log_it(L_DEBUG, "Received event socket %p to add on worker", l_es_new);
+ log_it(L_DEBUG, "Received event socket %p (ident %d type %d) to add on worker", l_es_new, l_es_new->socket, l_es_new->type);
+#ifdef DAP_EVENTS_CAPS_KQUEUE
+ if(l_es_new->socket!=0 && l_es_new->socket != -1 &&
+ l_es_new->type != DESCRIPTOR_TYPE_EVENT &&
+ l_es_new->type != DESCRIPTOR_TYPE_QUEUE &&
+ l_es_new->type != DESCRIPTOR_TYPE_TIMER
+ )
+#else
+ if(l_es_new->socket!=0 && l_es_new->socket != -1)
+
+#endif
if(dap_events_socket_check_unsafe( l_worker, l_es_new)){
// Socket already present in worker, it's OK
return;
}
+
+
switch( l_es_new->type){
case DESCRIPTOR_TYPE_SOCKET_UDP:
@@ -993,6 +1005,15 @@ static bool s_socket_all_check_activity( void * a_arg)
*/
void dap_worker_add_events_socket(dap_events_socket_t * a_events_socket, dap_worker_t * a_worker)
{
+/*
+#ifdef DAP_EVENTS_CAPS_KQUEUE
+ a_events_socket->worker = a_worker;
+ if(dap_worker_add_events_socket_unsafe(a_events_socket, a_worker)!=0)
+ a_events_socket->worker = NULL;
+
+#else*/
+ if(s_debug_reactor)
+ log_it(L_DEBUG,"Worker add esocket %d", a_events_socket->socket);
int l_ret = dap_events_socket_queue_ptr_send( a_worker->queue_es_new, a_events_socket );
if(l_ret != 0 ){
char l_errbuf[128];
@@ -1000,6 +1021,7 @@ void dap_worker_add_events_socket(dap_events_socket_t * a_events_socket, dap_wor
strerror_r(l_ret, l_errbuf, sizeof(l_errbuf));
log_it(L_ERROR, "Can't send pointer in queue: \"%s\"(code %d)", l_errbuf, l_ret);
}
+//#endif
}
/**
@@ -1060,41 +1082,67 @@ int dap_worker_add_events_socket_unsafe( dap_events_socket_t * a_esocket, dap_wo
if ( a_esocket->type == DESCRIPTOR_TYPE_EVENT && a_esocket->pipe_out){
return 0;
}
-
+ struct kevent l_event;
u_short l_flags = a_esocket->kqueue_base_flags;
u_int l_fflags = a_esocket->kqueue_base_fflags;
short l_filter = a_esocket->kqueue_base_filter;
- if(a_esocket->flags & DAP_SOCK_READY_TO_READ )
- l_filter |= EVFILT_READ;
- if(a_esocket->flags & DAP_SOCK_READY_TO_WRITE )
- l_fflags |= EVFILT_WRITE;
-
- EV_SET(&a_esocket->kqueue_event , a_esocket->socket, l_filter, EV_ADD | l_flags , l_fflags,
- a_esocket->kqueue_data, a_esocket);
- if(kevent ( a_worker->kqueue_fd,&a_esocket->kqueue_event,1,NULL,0,NULL) != -1 ){
- a_worker->kqueue_events_count++;
-/* if ( a_worker->kqueue_events_count == (size_t) a_worker->kqueue_events_count_max ){ // realloc
- if(a_worker->kqueue_events_count_max > (INT32_MAX -a_worker->kqueue_events_count_max ) ){
- log_it(L_CRITICAL,"Reached esocket count limit, can't add more than %d", a_worker->kqueue_events_count_max);
- a_worker->kqueue_events_count--;
- return -999;
+ int l_kqueue_fd =a_worker->kqueue_fd;
+ if ( l_kqueue_fd == -1 ){
+ log_it(L_ERROR, "Esocket is not assigned with anything ,exit");
+ }
+ // Check & add
+ bool l_is_error=false;
+ int l_errno=0;
+ if (a_esocket->type == DESCRIPTOR_TYPE_EVENT || a_esocket->type == DESCRIPTOR_TYPE_QUEUE){
+ EV_SET(&l_event, a_esocket->socket, EVFILT_USER,EV_ADD| EV_CLEAR ,0,0, &a_esocket->kqueue_event_catched_data );
+ if( kevent( l_kqueue_fd,&l_event,1,NULL,0,NULL)!=0){
+ l_is_error = true;
+ l_errno = errno;
+ }
+ }else{
+ if( l_filter){
+ EV_SET(&l_event, a_esocket->socket, l_filter,l_flags| EV_ADD,l_fflags,a_esocket->kqueue_data,a_esocket);
+ if( kevent( l_kqueue_fd,&l_event,1,NULL,0,NULL) != 0 ){
+ l_is_error = true;
+ l_errno = errno;
+ }else if (s_debug_reactor){
+ log_it(L_DEBUG, "kevent set custom filter %d on fd %d",l_filter, a_esocket->socket);
}
- a_worker->kqueue_events_count_max *= 2;
- log_it(L_WARNING, "Too many descriptors (%u), resizing array twice to %u", a_worker->poll_count, a_worker->poll_count_max);
- a_worker->poll =DAP_REALLOC(a_worker->poll, a_worker->poll_count_max * sizeof(*a_worker->poll));
- a_worker->poll_esocket =DAP_REALLOC(a_worker->poll_esocket, a_worker->poll_count_max * sizeof(*a_worker->poll_esocket));
- }*/
- if(s_debug_reactor)
- log_it(L_DEBUG,"Added ident %d in kqueue()", a_esocket->socket);
+ }else{
+ if( a_esocket->flags & DAP_SOCK_READY_TO_READ ){
+ EV_SET(&l_event, a_esocket->socket, EVFILT_READ,l_flags| EV_ADD,l_fflags,a_esocket->kqueue_data,a_esocket);
+ if( kevent( l_kqueue_fd,&l_event,1,NULL,0,NULL) != 0 ){
+ l_is_error = true;
+ l_errno = errno;
+ }else if (s_debug_reactor){
+ log_it(L_DEBUG, "kevent set EVFILT_READ on fd %d", a_esocket->socket);
+ }
- return 0;
- }else{
- int l_errno = errno;
- char l_errbuf[128]={};
- strerror_r(errno,l_errbuf,sizeof (l_errbuf));
- log_it(L_ERROR,"Can't add socket on kqueue: %s (code %d)", l_errbuf,l_errno);
- return l_errno;
+ }
+ if( !l_is_error){
+ if( a_esocket->flags & DAP_SOCK_READY_TO_WRITE || a_esocket->flags &DAP_SOCK_CONNECTING ){
+ EV_SET(&l_event, a_esocket->socket, EVFILT_WRITE,l_flags| EV_ADD,l_fflags,a_esocket->kqueue_data,a_esocket);
+ if(kevent( l_kqueue_fd,&l_event,1,NULL,0,NULL) != 0){
+ l_is_error = true;
+ l_errno = errno;
+ }else if (s_debug_reactor){
+ log_it(L_DEBUG, "kevent set EVFILT_WRITE on fd %d", a_esocket->socket);
+ }
+ }
+ }
+ }
}
+
+ if ( l_is_error ){
+ char l_errbuf[128];
+ l_errbuf[0]=0;
+ strerror_r(l_errno, l_errbuf, sizeof (l_errbuf));
+ log_it(L_ERROR,"Can't update client socket state on kqueue fd %d: \"%s\" (%d)",
+ a_esocket->socket, l_errbuf, l_errno);
+ return l_errno;
+ }else
+ return 0;
+
#else
#error "Unimplemented new esocket on worker callback for current platform"
#endif
diff --git a/modules/net/dap_chain_net.c b/modules/net/dap_chain_net.c
index c8aa4d25e17ccc2860566b7d571c969df8006d14..db35af96360c27cc5d92780c87d6598a662b2e62 100644
--- a/modules/net/dap_chain_net.c
+++ b/modules/net/dap_chain_net.c
@@ -837,61 +837,63 @@ static bool s_net_states_proc(dap_proc_thread_t *a_thread, void *a_arg)
case NODE_ROLE_LIGHT:
default: {
// Get DNS request result from root nodes as synchronization links
- bool l_sync_fill_root_nodes = true;
+ bool l_sync_fill_root_nodes = false;
uint32_t l_link_id=0;
- for (size_t n=0; n< s_required_links_count;n++ ) {
- struct in_addr l_addr = {};
- uint16_t i, l_port;
- if (l_net_pvt->seed_aliases_count) {
- i = rand() % l_net_pvt->seed_aliases_count;
- dap_chain_node_addr_t *l_remote_addr = dap_chain_node_alias_find(l_net, l_net_pvt->seed_aliases[i]);
- if (l_remote_addr){
- dap_chain_node_info_t *l_remote_node_info = dap_chain_node_info_read(l_net, l_remote_addr);
- if(l_remote_node_info){
- l_addr.s_addr = l_remote_node_info ? l_remote_node_info->hdr.ext_addr_v4.s_addr : 0;
- DAP_DELETE(l_remote_node_info);
- l_port = DNS_LISTEN_PORT;
+ if (!l_sync_fill_root_nodes){
+ for (size_t n=0; n< s_required_links_count;n++ ) {
+ struct in_addr l_addr = {};
+ uint16_t i, l_port;
+ if (l_net_pvt->seed_aliases_count) {
+ i = rand() % l_net_pvt->seed_aliases_count;
+ dap_chain_node_addr_t *l_remote_addr = dap_chain_node_alias_find(l_net, l_net_pvt->seed_aliases[i]);
+ if (l_remote_addr){
+ dap_chain_node_info_t *l_remote_node_info = dap_chain_node_info_read(l_net, l_remote_addr);
+ if(l_remote_node_info){
+ l_addr.s_addr = l_remote_node_info ? l_remote_node_info->hdr.ext_addr_v4.s_addr : 0;
+ DAP_DELETE(l_remote_node_info);
+ l_port = DNS_LISTEN_PORT;
+ }else{
+ log_it(L_WARNING,"Can't find node info for node addr "NODE_ADDR_FP_STR,
+ NODE_ADDR_FP_ARGS(l_remote_addr));
+ }
}else{
- log_it(L_WARNING,"Can't find node info for node addr "NODE_ADDR_FP_STR,
- NODE_ADDR_FP_ARGS(l_remote_addr));
+ log_it(L_WARNING,"Can't find alias info for seed alias %s",l_net_pvt->seed_aliases[i]);
+ }
+ } else if (l_net_pvt->bootstrap_nodes_count) {
+ i = rand() % l_net_pvt->bootstrap_nodes_count;
+ l_addr = l_net_pvt->bootstrap_nodes_addrs[i];
+ l_port = l_net_pvt->bootstrap_nodes_ports[i];
+ } else {
+ log_it(L_ERROR, "No root servers present in configuration file. Can't establish DNS requests");
+ if (!dap_list_length(l_net_pvt->links_info)) { // No links can be prepared, go offline
+ l_net_pvt->state_target = NET_STATE_OFFLINE;
}
- }else{
- log_it(L_WARNING,"Can't find alias info for seed alias %s",l_net_pvt->seed_aliases[i]);
- }
- } else if (l_net_pvt->bootstrap_nodes_count) {
- i = rand() % l_net_pvt->bootstrap_nodes_count;
- l_addr = l_net_pvt->bootstrap_nodes_addrs[i];
- l_port = l_net_pvt->bootstrap_nodes_ports[i];
- } else {
- log_it(L_ERROR, "No root servers present in configuration file. Can't establish DNS requests");
- if (!dap_list_length(l_net_pvt->links_info)) { // No links can be prepared, go offline
- l_net_pvt->state_target = NET_STATE_OFFLINE;
- }
- }
- if (l_addr.s_addr) {
- dap_chain_node_info_t *l_link_node_info = DAP_NEW_Z(dap_chain_node_info_t);
- if(! l_link_node_info){
- log_it(L_CRITICAL,"Can't allocate memory for node link info");
- break;
}
-#ifdef DAP_OS_UNIX
- struct in_addr _in_addr = { .s_addr = l_addr.s_addr };
-#else
- struct in_addr _in_addr = { { .S_addr = l_addr.S_un.S_addr } };
-#endif
-
- l_sync_fill_root_nodes = false;
- if (l_net_pvt->state_target != NET_STATE_OFFLINE) {
- l_net_pvt->links_dns_requests++;
- struct link_dns_request * l_dns_request = DAP_NEW_Z(struct link_dns_request);
- l_dns_request->net = l_net;
- l_dns_request->link_id = l_link_id;
- dap_chain_node_info_dns_request(l_addr, l_port, l_net->pub.name, l_link_node_info,
- s_net_state_link_prepare_success,
- s_net_state_link_prepare_error,l_dns_request);
+ if (l_addr.s_addr) {
+ dap_chain_node_info_t *l_link_node_info = DAP_NEW_Z(dap_chain_node_info_t);
+ if(! l_link_node_info){
+ log_it(L_CRITICAL,"Can't allocate memory for node link info");
+ break;
+ }
+ #ifdef DAP_OS_UNIX
+ struct in_addr _in_addr = { .s_addr = l_addr.s_addr };
+ #else
+ struct in_addr _in_addr = { { .S_addr = l_addr.S_un.S_addr } };
+ #endif
+
+ l_sync_fill_root_nodes = false;
+ if (l_net_pvt->state_target != NET_STATE_OFFLINE) {
+ l_net_pvt->links_dns_requests++;
+ struct link_dns_request * l_dns_request = DAP_NEW_Z(struct link_dns_request);
+ l_dns_request->net = l_net;
+ l_dns_request->link_id = l_link_id;
+ dap_chain_node_info_dns_request(l_addr, l_port, l_net->pub.name, l_link_node_info,
+ s_net_state_link_prepare_success,
+ s_net_state_link_prepare_error,l_dns_request);
+ }
}
+ l_link_id++;
}
- l_link_id++;
}
if (l_sync_fill_root_nodes){
log_it(L_ATT,"Not found bootstrap addresses, fill seed nodelist from root aliases");
diff --git a/modules/net/dap_chain_node_dns_client.c b/modules/net/dap_chain_node_dns_client.c
index df023da94b325ea5b790d3f768a6fa50bfd0ef2b..3b7df7f5ace3373c4369affc61cfbf2cb5f67c07 100644
--- a/modules/net/dap_chain_node_dns_client.c
+++ b/modules/net/dap_chain_node_dns_client.c
@@ -52,6 +52,7 @@ static void s_dns_client_esocket_read_callback(dap_events_socket_t * a_esocket,
static void s_dns_client_esocket_error_callback(dap_events_socket_t * a_esocket, int a_error);
static bool s_dns_client_esocket_timeout_callback( void * a_arg);
static void s_dns_client_esocket_delete_callback(dap_events_socket_t * a_esocket, void * a_arg);
+static void s_dns_client_esocket_worker_assign_callback(dap_events_socket_t * a_esocket, dap_worker_t * a_worker);
/**
* @brief s_dns_client_esocket_read_callback
@@ -179,6 +180,24 @@ static void s_dns_client_esocket_delete_callback(dap_events_socket_t * a_esocket
DAP_DEL_Z(l_dns_client->buf);
}
+/**
+ * @brief s_dns_client_esocket_worker_assign_callback
+ * @param a_esocket
+ * @param a_worker
+ */
+static void s_dns_client_esocket_worker_assign_callback(dap_events_socket_t * a_esocket, dap_worker_t * a_worker)
+{
+ struct dns_client * l_dns_client = (struct dns_client*) a_esocket->_inheritor;
+ dap_events_socket_write_unsafe(a_esocket,l_dns_client->dns_request->data, l_dns_client->dns_request->size );
+
+ dap_events_socket_handler_t * l_es_handler = DAP_NEW_Z(dap_events_socket_handler_t);
+ l_es_handler->esocket = a_esocket;
+ l_es_handler->uuid = a_esocket->uuid;
+ dap_timerfd_start_on_worker(a_worker, dap_config_get_item_uint64_default(g_config,"dns_client","request_timeout",10)*1000,
+ s_dns_client_esocket_timeout_callback,l_es_handler);
+
+}
+
/**
* @brief dap_chain_node_info_dns_request
* @param a_addr
@@ -236,6 +255,7 @@ void dap_chain_node_info_dns_request(struct in_addr a_addr, uint16_t a_port, cha
dap_events_socket_callbacks_t l_esocket_callbacks={0};
+ l_esocket_callbacks.worker_assign_callback = s_dns_client_esocket_worker_assign_callback;
l_esocket_callbacks.delete_callback = s_dns_client_esocket_delete_callback; // Delete client callback
l_esocket_callbacks.read_callback = s_dns_client_esocket_read_callback; // Read function
l_esocket_callbacks.error_callback = s_dns_client_esocket_error_callback; // Error processing function
@@ -248,14 +268,7 @@ void dap_chain_node_info_dns_request(struct in_addr a_addr, uint16_t a_port, cha
l_esocket->_inheritor = l_dns_client;
dap_worker_t * l_worker = dap_events_worker_get_auto();
- dap_events_socket_write_unsafe(l_esocket,l_dns_client->dns_request->data, l_dns_client->dns_request->size );
dap_events_socket_assign_on_worker_mt(l_esocket,l_worker);
-
- dap_events_socket_handler_t * l_es_handler = DAP_NEW_Z(dap_events_socket_handler_t);
- l_es_handler->esocket = l_esocket;
- l_es_handler->uuid = l_esocket->uuid;
- dap_timerfd_start_on_worker(l_worker, dap_config_get_item_uint64_default(g_config,"dns_client","request_timeout",10)*1000,
- s_dns_client_esocket_timeout_callback,l_es_handler);
}