From b1f606e82156f660517dc8a9d4ee404e621aafbf Mon Sep 17 00:00:00 2001 From: Riccardo Elitropi Date: Tue, 30 Sep 2025 11:29:39 +0200 Subject: [PATCH] EgtExecutor : - in redis migliorato ciclo degli eventi e aggiunte funzioni per collegamento asincrono per nodo sentinalla e autenticazione (test). --- EXE_Redis.cpp | 291 +++++++++++++++++++++++++++++++++----------------- 1 file changed, 193 insertions(+), 98 deletions(-) diff --git a/EXE_Redis.cpp b/EXE_Redis.cpp index cdd3381..7535f78 100644 --- a/EXE_Redis.cpp +++ b/EXE_Redis.cpp @@ -28,6 +28,7 @@ #include #include #include +#include using namespace std ; @@ -42,7 +43,8 @@ static redisContext* s_pRedisContext = nullptr ; // Contesto Asincrono static redisAsyncContext* s_pRedisAsyncSubContext = nullptr ; static redisAsyncContext* s_pRedisAsyncPubContext = nullptr ; -static atomic s_nPendingDataBase = REDIS_MIN_DB ; // Default +static atomic s_nPendingDataBase = REDIS_MIN_DB ; // Default +static string s_sPassword = "" ; // Default static atomic s_bSubConnected = false ; static atomic s_bPubConnected = false ; static atomic s_bPubLoopRunning = false ; @@ -399,30 +401,33 @@ RedisEventLoop( redisAsyncContext* ctx, bool bIsPub) // Recupero del file descriptor del socket TCP usato da Redis. SOCKET sock = ctx->c.fd ; + // Imposto condizione di ascolto degli eventi + bool bLoopRunning = bIsPub ? s_bPubLoopRunning : s_bSubLoopRunning ; + // Creazione dei due insiemi per il file descriptor ( lettura e scrittura ) + // r = read, w = write + fd_set rfds, wfds ; + // Timeout per funzione select { secondi, millisecondi} + // La funzione select attende che il socket sia pronto per lettura e scrittura + // Resituisce : + // SOCKET_ERROR in caso di errore + // 0 se scade il timeout + // > 0 se ci sono eventi da gestire + timeval timeout = {0, 100000 } ; // 100ms // Ciclo - while ( bIsPub ? s_bPubLoopRunning : s_bSubLoopRunning) { + while ( bLoopRunning) { // Se contesto non valido, interrompo il ciclo if ( ctx == nullptr || ( ctx->c.flags & REDIS_DISCONNECTING)) { LOG_INFO ( GetCmdLogger(), ( string{ "Redis event loop terminated for Async "} + ( bIsPub ? "Pub" : "Sub") + " Events").c_str()) break ; } - // Creazione dei due insiemi per il file descriptor ( lettura e scrittura ) - // r = read, w = write - fd_set rfds, wfds ; + // Recupero file descriptor di lettura e scrittura FD_ZERO( &rfds) ; FD_ZERO( &wfds) ; FD_SET( sock, &rfds) ; FD_SET( sock, &wfds) ; - - // TimeOut per funzione select { secondi, millisecondi} - // La funzione select attende che il socket sia pronto per lettura e scrittura - // Resituisce : - // SOCKET_ERROR in caso di errore - // 0 se scade il timeout - // > 0 se ci sono eventi da gestire - timeval timeout = {0, 5000} ; + // Controllo il risultato ottenuto int nRet = select( 0, &rfds, &wfds, nullptr, &timeout) ; if ( nRet == SOCKET_ERROR) { // Se errore -> interruzione del ciclo @@ -430,19 +435,19 @@ RedisEventLoop( redisAsyncContext* ctx, bool bIsPub) ToString( WSAGetLastError())).c_str()) ; break ; } - else if ( nRet == 0) { - // Inserisco una breve pausa per non saturare le CPU - this_thread::sleep_for( chrono::milliseconds( 10)) ; + else if ( nRet > 0) { + // Se Socket pronto per lettura/ scrittua + if ( ctx != nullptr && ! ( ctx->c.flags & REDIS_DISCONNECTING) && ctx->c.fd != -1) { + if ( FD_ISSET( sock, &rfds)) + redisAsyncHandleRead( ctx) ; + if ( FD_ISSET( sock, &wfds)) + redisAsyncHandleWrite( ctx) ; + } } - if ( nRet > 0) { - // Se Socket pronto per lettura, leggo il messaggio dal contesto di Redis - if ( FD_ISSET( sock, &rfds) && ctx != nullptr) - redisAsyncHandleRead( ctx) ; - // Se Socket pronto per scrittura, scrivo il messaggio per il contesto di Redis - if ( FD_ISSET( sock, &wfds) && ctx != nullptr) - redisAsyncHandleWrite( ctx) ; - } - this_thread::sleep_for( chrono::milliseconds( 10)) ; + + // Reset del timeOut ad ogni ciclo + timeout.tv_sec = 0 ; + timeout.tv_usec = 100000 ; // 100ms } return ; @@ -609,7 +614,7 @@ WaitMessageCallback( redisAsyncContext* ctx, void* r, void*) } //---------------------------------------------------------------------------- -// Funzione per Memorizzare il DataBase Redis da selezionare +// Funzione per Memorizzare il DataBase Redis da selezionare [Callback] //---------------------------------------------------------------------------- static bool SetPendingDataBase( int nDB) @@ -623,7 +628,7 @@ SetPendingDataBase( int nDB) } //---------------------------------------------------------------------------- -// Funzione per Ricavare il numero di DataBase Redis da selezionare +// Funzione per Ricavare il numero di DataBase Redis da selezionare [Callback] //---------------------------------------------------------------------------- static int GetPendingDataBase() @@ -631,6 +636,24 @@ GetPendingDataBase() return s_nPendingDataBase ; } +// --------------------------------------------------------------------------- +// Funzione per Memorizzare la Password di autenticazione [Callback] +// --------------------------------------------------------------------------- +static bool +SetPendingPassword( string sPsw) +{ + s_sPassword = sPsw ; +} + +// -------------------------------------------------------------------------- +// Funzione per ottenere la Password di autenticazione [Callback] +// -------------------------------------------------------------------------- +static string +GetPendingPassword() +{ + return s_sPassword ; +} + //---------------------------------------------------------------------------- static bool GetConnectionContext( redisAsyncContext*& ctx, const string& sHost, int nPort) @@ -700,6 +723,33 @@ ExeRedisAsyncConnect( const string& sConnection) return false ; } + // Verifico se si tratta di un nodo Sentinella + redisContext* sentinelCtx = redisConnect( ConnectionInfo.sHost.c_str(), ConnectionInfo.nPort) ; + if ( sentinelCtx == nullptr) { + LOG_INFO( GetCmdLogger(), "Error : Can't allocate redis context") + return false ; + } + else if ( sentinelCtx->err != 0) { + LOG_INFO( GetCmdLogger(), ( string{ "Error : "} + sentinelCtx->errstr).c_str()) + redisFree( sentinelCtx) ; + return false ; + } + redisReply* replySentinel = ( redisReply*)redisCommand( sentinelCtx, "SENTINEL get-master-addr-by-name %s", + ConnectionInfo.sServiceName.c_str()) ; + if ( replySentinel == nullptr || replySentinel->type != REDIS_REPLY_ARRAY || replySentinel->elements != 2) { + if ( replySentinel != nullptr) + freeReplyObject( replySentinel) ; + } + else if ( replySentinel->element[0] != nullptr && replySentinel->element[0]->str != nullptr && + replySentinel->element[1] != nullptr && replySentinel->element[1]->str != nullptr) { + // --- Nodo sentinella + ConnectionInfo.sHost = replySentinel->element[0]->str ; + ConnectionInfo.nPort = DEFAULT_SENTINEL_MASTER_PORT ; + FromString( replySentinel->element[1]->str, ConnectionInfo.nPort) ; + freeReplyObject( replySentinel) ; + redisFree( sentinelCtx) ; + } + // Definisco le 2 connessioni asincroni con Host e Porta ( per PUBLISH e SUBSCRIBE/UNSUBSCRIBE) if ( ! GetConnectionContext( s_pRedisAsyncPubContext, ConnectionInfo.sHost, ConnectionInfo.nPort) || ! GetConnectionContext( s_pRedisAsyncSubContext, ConnectionInfo.sHost, ConnectionInfo.nPort)) @@ -711,49 +761,57 @@ ExeRedisAsyncConnect( const string& sConnection) s_pRedisAsyncPubContext, []( const redisAsyncContext* ctx, int nStatus) { // --- Se contesto nullo, errore - if ( ctx == nullptr) { - LOG_INFO( GetCmdLogger(), "Error : Redis null Context in CallBack Async Pub Connection") ; - return ; - } - // --- Se Connessione non valida, errore - if ( nStatus != REDIS_OK) { + if ( ctx == nullptr || nStatus != REDIS_OK) { LOG_INFO( GetCmdLogger(), "Error : Async Connect CallBack Pub failed ") ; return ; } - // --- Se Connessione Ok - else { - // --- Selezione del DataBase + // --- Se Autenticazione richiesta -> AUTH e SELECT + string sPassword = GetPendingPassword() ; + if ( ! sPassword.empty()) { redisAsyncCommand( s_pRedisAsyncPubContext, []( redisAsyncContext* ctx, void* r, void*) { - // --- Se contesto nullo, errore - if ( ctx == nullptr) { - LOG_INFO( GetCmdLogger(), "Error : Redis null Context in CallBack Async Pub Connection") ; - return ; - } - // --- Richiesta di Selezione DB a Redis + // Invio la richiesta redisReply* reply = static_cast( r) ; - if ( reply == nullptr) { - // --- Nulla - LOG_INFO( GetCmdLogger(), "Error : Null Database selection reply") ; + if ( reply == nullptr || reply->type == REDIS_REPLY_ERROR) { + string sErrMsg = ( reply != nullptr && reply->str != nullptr) ? reply->str : "Null asnwer" ; + LOG_INFO( GetCmdLogger(), ( string{ " Error : Authentication -> "} + sErrMsg).c_str()) return ; } - if ( reply->type == REDIS_REPLY_STATUS && strcmp( reply->str, "OK") == 0) { - // --- Valida - LOG_INFO( GetCmdLogger(), ( string{ "Connected to DataBase #"} + ToString( GetPendingDataBase()) + " for Pub commands !").c_str()) - s_bPubConnected = true ; - return ; - } - else { - // --- Errore - string sErrMsg = ( reply->str != nullptr ? reply->str : "Null asnwer") ; - LOG_INFO( GetCmdLogger(), ( string{ "Error : DataBase selection -> "} + sErrMsg).c_str()) - return ; - } + LOG_INFO( GetCmdLogger(), "Valid Authentication for Pub connection") + redisAsyncCommand( + ctx, + []( redisAsyncContext* ctx, void* r, void*) { + redisReply* reply = static_cast( r) ; + if ( reply == nullptr || reply->type == REDIS_REPLY_ERROR) { + string sErrMsg = ( reply != nullptr && reply->str != nullptr) ? reply->str : "Null answer" ; + LOG_INFO( GetCmdLogger(), ( string{ "Error : DataBase selection -> "} + sErrMsg).c_str()) ; + return ; + } + LOG_INFO( GetCmdLogger(), ( string{ "Connected to DataBase #"} + ToString( GetPendingDataBase()) + " for Pub commands !").c_str()) ; + s_bPubConnected = true ; + }, + nullptr, "SELECT %d", GetPendingDataBase() + ) ; }, - nullptr, - "SELECT %d", - GetPendingDataBase() + nullptr, "AUTH %s", GetPendingPassword().c_str() + ) ; + } + // Se nessuna autenticazione -> SELECT + else { + redisAsyncCommand( + s_pRedisAsyncPubContext, + []( redisAsyncContext* ctx, void* r, void*) { + redisReply* reply = static_cast( r) ; + if ( reply == nullptr || reply->type == REDIS_REPLY_ERROR) { + string sErrMsg = ( reply != nullptr && reply->str != nullptr) ? reply->str : "Null answer" ; + LOG_INFO( GetCmdLogger(), ( string{ "Error : DataBase selection -> "} + sErrMsg).c_str()) ; + return ; + } + LOG_INFO( GetCmdLogger(), ( string{ "Connected to DataBase #"} + ToString( GetPendingDataBase()) + " for Pub commands !").c_str()) + s_bPubConnected = true ; + }, + nullptr, "SELECT %d", GetPendingDataBase() ) ; } } @@ -764,49 +822,82 @@ ExeRedisAsyncConnect( const string& sConnection) s_pRedisAsyncSubContext, []( const redisAsyncContext* ctx, int nStatus) { // --- Se contesto nullo, errore - if ( ctx == nullptr) { - LOG_INFO( GetCmdLogger(), "Error : Redis null Context in CallBack Async Sub Connection") ; + if ( ctx == nullptr || nStatus != REDIS_OK) { + LOG_INFO( GetCmdLogger(), "Error : Async Connect CallBack Pub failed ") ; return ; } - // --- Se Connessione non valida, errore - if ( nStatus != REDIS_OK) { - LOG_INFO( GetCmdLogger(), "Error : Async Connect CallBack Sub failed ") ; - return ; - } - // --- Se Connessione Ok - else { - // --- Selezione del DataBase + // --- Se Autenticazione richiesta -> AUTH e SELECT + string sPassword = GetPendingPassword() ; + if ( ! sPassword.empty()) { redisAsyncCommand( s_pRedisAsyncSubContext, []( redisAsyncContext* ctx, void* r, void*) { - // --- Se contesto nullo, errore - if ( ctx == nullptr) { - LOG_INFO( GetCmdLogger(), "Error : Redis null Context in CallBack Async Sub Connection") ; + // Invio la richiesta + redisReply* reply = static_cast( r) ; + if ( reply == nullptr || reply->type == REDIS_REPLY_ERROR) { + string sErrMsg = ( reply != nullptr && reply->str != nullptr) ? reply->str : "Null asnwer" ; + LOG_INFO( GetCmdLogger(), ( string{ " Error : Authentication -> "} + sErrMsg).c_str()) return ; } - // --- Richiesta di Selezione DB a Redis + LOG_INFO( GetCmdLogger(), "Valid Authentication for Pub connection") + redisAsyncCommand( + ctx, + []( redisAsyncContext* ctx, void* r, void*) { + redisReply* reply = static_cast( r) ; + if ( reply == nullptr || reply->type == REDIS_REPLY_ERROR) { + string sErrMsg = ( reply != nullptr && reply->str != nullptr) ? reply->str : "Null answer" ; + LOG_INFO( GetCmdLogger(), ( string{ "Error : DataBase selection -> "} + sErrMsg).c_str()) ; + return ; + } + LOG_INFO( GetCmdLogger(), ( string{ "Connected to DataBase #"} + ToString( GetPendingDataBase()) + " for Pub commands !").c_str()) ; + s_bSubConnected = true ; + }, + nullptr, "SELECT %d", GetPendingDataBase() + ) ; + }, + nullptr, "SELECT %d", GetPendingDataBase() + ) ; + // --- Autenticazione richiesta + redisAsyncCommand( + s_pRedisAsyncSubContext, + []( redisAsyncContext* ctx, void* r, void*) { redisReply* reply = static_cast( r) ; if ( reply == nullptr) { // --- Nulla - LOG_INFO( GetCmdLogger(), "Error : Null Database selection reply") ; + LOG_INFO( GetCmdLogger(), "Error : Null Authentication reply") ; return ; } if ( reply->type == REDIS_REPLY_STATUS && strcmp( reply->str, "OK") == 0) { // --- Valida - LOG_INFO( GetCmdLogger(), ( string{ "Connected to DataBase #"} + ToString( GetPendingDataBase()) + " for Sub commands !").c_str()) - s_bSubConnected = true ; + LOG_INFO( GetCmdLogger(), "Valid Authentication for Pub connection") + s_bPubConnected = true ; return ; } else { // --- Errore - string sErrMsg = ( reply->str != nullptr) ? reply->str : "Null asnwer" ; - LOG_INFO( GetCmdLogger(), ( string{ "Error : DataBase selection -> "} + sErrMsg).c_str()) + string sErrMsg = ( reply->str != nullptr ? reply->str : "Null asnwer") ; + LOG_INFO( GetCmdLogger(), ( string{ "Error : Authentication -> "} + sErrMsg).c_str()) return ; } }, - nullptr, - "SELECT %d", - GetPendingDataBase() + nullptr, "AUTH %s", GetPendingPassword().c_str() + ) ; + } + // Se nessuna autenticazione -> SELECT + else { + redisAsyncCommand( + s_pRedisAsyncSubContext, + []( redisAsyncContext* ctx, void* r, void*) { + redisReply* reply = static_cast( r) ; + if ( reply == nullptr || reply->type == REDIS_REPLY_ERROR) { + string sErrMsg = ( reply != nullptr && reply->str != nullptr) ? reply->str : "Null answer" ; + LOG_INFO( GetCmdLogger(), ( string{ "Error : DataBase selection -> "} + sErrMsg).c_str()) ; + return ; + } + LOG_INFO( GetCmdLogger(), ( string{ "Connected to DataBase #"} + ToString( GetPendingDataBase()) + " for Pub commands !").c_str()) + s_bSubConnected = true ; + }, + nullptr, "SELECT %d", GetPendingDataBase() ) ; } } @@ -845,18 +936,19 @@ ExeRedisAsyncConnect( const string& sConnection) // Se richiesto massimo tempo di connessione, aspetto if ( ConnectionInfo.nAsyncTimeout > 0.) { - const int nStepMillSec = 2 ; - int nWaitedMillSec = 0 ; - while ( ! s_bSubConnected && ! s_bPubConnected && nWaitedMillSec < static_cast( ConnectionInfo.nAsyncTimeout)) { - this_thread::sleep_for( chrono::milliseconds( nStepMillSec)) ; - nWaitedMillSec += nStepMillSec ; + auto tStart = chrono::steady_clock::now() ; + auto tTimeOut = chrono::milliseconds( static_cast( ConnectionInfo.nAsyncTimeout)) ; + + while ( ! s_bSubConnected && ! s_bPubConnected) { + this_thread::sleep_for( chrono::milliseconds( 10)) ; + if ( chrono::steady_clock::now() - tStart > tTimeOut) { + LOG_INFO( GetCmdLogger(), "Error in Connection : TimeOut exceeded") + return false ; + } } - if ( ! s_bSubConnected || ! s_bPubConnected) { - LOG_INFO( GetCmdLogger(), "Error in Connection : TimeOut exceeded") - return false ; - } - else - LOG_INFO( GetCmdLogger(), ( string{ "Connected in "} + ToString( nWaitedMillSec) + " ms").c_str()) + + auto tElapsed = chrono::duration_cast( chrono::steady_clock::now() - tStart).count() ; + LOG_INFO( GetCmdLogger(), ( string{ "Connected in "} + ToString( tElapsed) + " ms").c_str()) } return true ; @@ -993,11 +1085,14 @@ ExeRedisAsyncSubscribeOneMessage( const string& sChannel, double dMaxTimeOut, st } // Attivo il TimeOut per l'attesa del messaggio - const int nStepMillSeconds = 2 ; - int nWaitedMillSeconds = 0 ; - while ( ! s_bMessage && nWaitedMillSeconds < static_cast( dMaxTimeOut)) { - this_thread::sleep_for( chrono::milliseconds( nStepMillSeconds)) ; - nWaitedMillSeconds += nStepMillSeconds ; + auto tStart = chrono::steady_clock::now() ; + auto tTimeOut = chrono::milliseconds( static_cast( dMaxTimeOut)) ; + while ( ! s_bMessage) { + this_thread::sleep_for( chrono::milliseconds( 10)) ; + if ( chrono::steady_clock::now() - tStart > tTimeOut) { + LOG_INFO( GetCmdLogger(), ( string{ "Timeout : No Message received on ["} + sChannel + "]").c_str()) + return false ; + } } // Invio del comando Unsubscribe