From 2b4ba10b3d04215cff2e00bca7b11532e97ba017 Mon Sep 17 00:00:00 2001 From: "Roger A. Light" Date: Wed, 19 Sep 2018 14:40:22 +0100 Subject: [PATCH] Fix "round_robin false" behaviour. Closes #481. --- ChangeLog.txt | 1 + src/bridge.c | 31 ++++++++++++---------- src/conf.c | 1 + src/loop.c | 46 +++++++++++++++++++++++++++------ src/mosquitto_broker_internal.h | 1 + 5 files changed, 58 insertions(+), 22 deletions(-) diff --git a/ChangeLog.txt b/ChangeLog.txt index 6e7bd99b..4ce9123a 100644 --- a/ChangeLog.txt +++ b/ChangeLog.txt @@ -7,6 +7,7 @@ Broker: - Fix excessive CPU usage when the number of sockets exceeds the system limit. Closes #948. - Fix for bridge connections when using WITH_ADNS=yes. +- Fix round_robin false behaviour. Closes #481. Library: - Fix situation where username and password is used with SOCKS5 proxy. Closes diff --git a/src/bridge.c b/src/bridge.c index e3e856f6..7f98d399 100644 --- a/src/bridge.c +++ b/src/bridge.c @@ -249,6 +249,10 @@ int bridge__connect_step3(struct mosquitto_db *db, struct mosquitto *context) return rc; } + if(context->bridge->round_robin == false && context->bridge->cur_address != 0){ + context->bridge->primary_retry = mosquitto_time() + 5; + } + rc = send__connect(context, context->keepalive, context->clean_session); if(rc == MOSQ_ERR_SUCCESS){ return MOSQ_ERR_SUCCESS; @@ -270,7 +274,7 @@ int bridge__connect_step3(struct mosquitto_db *db, struct mosquitto *context) int bridge__connect(struct mosquitto_db *db, struct mosquitto *context) { - int rc; + int rc, rc2; int i; char *notification_topic; int notification_topic_len; @@ -348,7 +352,7 @@ int bridge__connect(struct mosquitto_db *db, struct mosquitto *context) log__printf(NULL, MOSQ_LOG_NOTICE, "Connecting bridge %s (%s:%d)", context->bridge->name, context->bridge->addresses[context->bridge->cur_address].address, context->bridge->addresses[context->bridge->cur_address].port); rc = net__socket_connect(context, context->bridge->addresses[context->bridge->cur_address].address, context->bridge->addresses[context->bridge->cur_address].port, NULL, false); - if(rc > 0 ){ + if(rc > 0){ if(rc == MOSQ_ERR_TLS){ net__socket_close(db, context); return rc; /* Error already printed */ @@ -359,28 +363,27 @@ int bridge__connect(struct mosquitto_db *db, struct mosquitto *context) } return rc; + }else if(rc == MOSQ_ERR_CONN_PENDING){ + context->state = mosq_cs_connect_pending; } HASH_ADD(hh_sock, db->contexts_by_sock, sock, sizeof(context->sock), context); - if(rc == MOSQ_ERR_CONN_PENDING){ - context->state = mosq_cs_connect_pending; - } - rc = send__connect(context, context->keepalive, context->clean_session); - if(rc == MOSQ_ERR_SUCCESS){ - return MOSQ_ERR_SUCCESS; - }else if(rc == MOSQ_ERR_ERRNO && errno == ENOTCONN){ + rc2 = send__connect(context, context->keepalive, context->clean_session); + if(rc2 == MOSQ_ERR_SUCCESS){ + return rc; + }else if(rc2 == MOSQ_ERR_ERRNO && errno == ENOTCONN){ return MOSQ_ERR_SUCCESS; }else{ - if(rc == MOSQ_ERR_TLS){ - return rc; /* Error already printed */ - }else if(rc == MOSQ_ERR_ERRNO){ + if(rc2 == MOSQ_ERR_TLS){ + return rc2; /* Error already printed */ + }else if(rc2 == MOSQ_ERR_ERRNO){ log__printf(NULL, MOSQ_LOG_ERR, "Error creating bridge: %s.", strerror(errno)); - }else if(rc == MOSQ_ERR_EAI){ + }else if(rc2 == MOSQ_ERR_EAI){ log__printf(NULL, MOSQ_LOG_ERR, "Error creating bridge: %s.", gai_strerror(errno)); } net__socket_close(db, context); - return rc; + return rc2; } } #endif diff --git a/src/conf.c b/src/conf.c index f5e989ff..27a7bacd 100644 --- a/src/conf.c +++ b/src/conf.c @@ -1157,6 +1157,7 @@ int config__read_file_core(struct mosquitto__config *config, bool reload, struct cur_bridge->try_private = true; cur_bridge->attempt_unsubscribe = true; cur_bridge->protocol_version = mosq_p_mqtt311; + cur_bridge->primary_retry_sock = INVALID_SOCKET; }else{ log__printf(NULL, MOSQ_LOG_ERR, "Error: Empty connection value in configuration."); return MOSQ_ERR_INVAL; diff --git a/src/loop.c b/src/loop.c index 7ab5b7b2..8df41760 100644 --- a/src/loop.c +++ b/src/loop.c @@ -125,11 +125,12 @@ int mosquitto_main_loop(struct mosquitto_db *db, mosq_sock_t *listensock, int li int pollfd_max; #endif #ifdef WITH_BRIDGE - mosq_sock_t bridge_sock; int rc; #endif time_t expiration_check_time = 0; char *id; + int err; + socklen_t len; #ifndef WIN32 sigemptyset(&sigblock); @@ -231,12 +232,40 @@ int mosquitto_main_loop(struct mosquitto_db *db, mosq_sock_t *listensock, int li mosquitto__check_keepalive(db, context); if(context->bridge->round_robin == false && context->bridge->cur_address != 0 + && context->bridge->primary_retry && now > context->bridge->primary_retry){ - if(net__try_connect(context, context->bridge->addresses[0].address, context->bridge->addresses[0].port, &bridge_sock, NULL, false) <= 0){ - COMPAT_CLOSE(bridge_sock); - net__socket_close(db, context); - context->bridge->cur_address = context->bridge->address_count-1; + if(context->bridge->primary_retry_sock == INVALID_SOCKET){ + rc = net__try_connect(context, context->bridge->addresses[0].address, + context->bridge->addresses[0].port, + &context->bridge->primary_retry_sock, NULL, false); + + if(rc == 0){ + COMPAT_CLOSE(context->bridge->primary_retry_sock); + context->bridge->primary_retry_sock = INVALID_SOCKET; + context->bridge->primary_retry = 0; + net__socket_close(db, context); + context->bridge->cur_address = 0; + } + }else{ + len = sizeof(int); + if(!getsockopt(context->bridge->primary_retry_sock, SOL_SOCKET, SO_ERROR, (char *)&err, &len)){ + if(err == 0){ + COMPAT_CLOSE(context->bridge->primary_retry_sock); + context->bridge->primary_retry_sock = INVALID_SOCKET; + context->bridge->primary_retry = 0; + net__socket_close(db, context); + context->bridge->cur_address = context->bridge->address_count-1; + }else{ + COMPAT_CLOSE(context->bridge->primary_retry_sock); + context->bridge->primary_retry_sock = INVALID_SOCKET; + context->bridge->primary_retry = now+5; + } + }else{ + COMPAT_CLOSE(context->bridge->primary_retry_sock); + context->bridge->primary_retry_sock = INVALID_SOCKET; + context->bridge->primary_retry = now+5; + } } } } @@ -324,9 +353,6 @@ int mosquitto_main_loop(struct mosquitto_db *db, mosq_sock_t *listensock, int li if(context->bridge->cur_address == context->bridge->address_count){ context->bridge->cur_address = 0; } - if(context->bridge->round_robin == false && context->bridge->cur_address != 0){ - context->bridge->primary_retry = now + 5; - } }else{ if((context->bridge->start_type == bst_lazy && context->bridge->lazy_reconnect) || (context->bridge->start_type == bst_automatic && now > context->bridge->restart_t)){ @@ -397,6 +423,10 @@ int mosquitto_main_loop(struct mosquitto_db *db, mosq_sock_t *listensock, int li { rc = bridge__connect(db, context); if(rc == MOSQ_ERR_SUCCESS){ + context->bridge->restart_t = 0; + if(context->bridge->round_robin == false && context->bridge->cur_address != 0){ + context->bridge->primary_retry = now + 5; + } #ifdef WITH_EPOLL ev.data.fd = context->sock; ev.events = EPOLLIN; diff --git a/src/mosquitto_broker_internal.h b/src/mosquitto_broker_internal.h index dce5efd2..67a2924c 100644 --- a/src/mosquitto_broker_internal.h +++ b/src/mosquitto_broker_internal.h @@ -428,6 +428,7 @@ struct mosquitto__bridge{ int cur_address; int address_count; time_t primary_retry; + mosq_sock_t primary_retry_sock; bool round_robin; bool try_private; bool try_private_accepted;