From 92f1ecd7c55d8197a0ec8d26d4b6c17e22ddd988 Mon Sep 17 00:00:00 2001 From: David Weir Date: Tue, 2 Dec 2025 16:37:23 -0500 Subject: [PATCH 01/17] Support ehanced client info and expanded module id ranges --- include/RTMA_types.h | 91 ++- include/internal/UPipe.h | 2 + src/core/PipeLib/SocketPipe.cpp | 14 + src/core/PipeLib/SocketPipe.h | 2 + src/modules/MessageManager/MessageManager.cpp | 623 +++++++++++------- src/modules/MessageManager/MessageManager.h | 242 +++++-- 6 files changed, 652 insertions(+), 322 deletions(-) diff --git a/include/RTMA_types.h b/include/RTMA_types.h index 6efe093..239b1f8 100644 --- a/include/RTMA_types.h +++ b/include/RTMA_types.h @@ -1,25 +1,29 @@ #ifndef _RTMA_TYPES_H_ // this is commented out becuase make_config fails with it #define _RTMA_TYPES_H_ +#include // Types used in the RTMA system typedef short MODULE_ID; typedef short HOST_ID; typedef int MSG_TYPE; typedef int MSG_COUNT; +typedef int32_t UID; // Maximums for the entire RTMA system -#define MAX_MODULES 200 //maximal number of modules in the system -#define DYN_MOD_ID_START 100 //module ID where pool of dynamic IDs begin +#define MAX_MODULES 256 //maximal number of modules in the system +#define DYN_MOD_ID_START 10000 //module ID where pool of dynamic IDs begin (10000 - 32000) #define MAX_HOSTS 5 //maximal number of hosts in the system #define MAX_MESSAGE_TYPES 10000 //maximal number of message types in the system #define MIN_STREAM_TYPE 9000 //minimal number type for a data stream #define MAX_TIMERS 100 //maximal number of total timers TimerModule can handle #define MAX_INTERNAL_TIMERS 20 //maximal number of internal timers that can be set by an RTMA module +#define MAX_NAME_LEN 32 // Module name length limit // Maximums for core modules #define MAX_RTMA_MSG_TYPE 99 // Message types below 100 are reserved for RTMA core #define MAX_RTMA_MODULE_ID 9 // Module ID-s below 10 are reserved for RTMA core +#define MAX_MODULE_ID 32000 // Header fields for all messages passed through RTMA // Following macro was commented out for compatibility with ctypesgen2 v2.2.2 python package (used for Python3 compatiblity) All references to this macro in this file and in RTMA.h were replaced with actual values @@ -199,16 +203,16 @@ typedef STRING_DATA MDF_AM_CONFIG_FILE_DATA; typedef char MDF_AM_APP_NAME[]; //sent by SLAVE AMs -#define MT_SLAVE_ALL_MODULES_READY 69 -#define MT_SLAVE_FAIL_START_APP 70 -typedef STRING_DATA MDF_SLAVE_FAIL_START_APP; -#define MT_SLAVE_FAIL_STOP_APP 71 -#define MT_SLAVE_FAIL_KILL_APP 72 -#define MT_SLAVE_APP_SHUTODWN_COMPLETE 74 -#define MT_SLAVE_APP_RESTART_COMPLETE 75 -#define MT_SLAVE_APP_KILL_COMPLETE 76 -#define MT_SLAVE_AM_ERROR 77 -typedef STRING_DATA MDF_SLAVE_AM_ERROR; +// #define MT_SLAVE_ALL_MODULES_READY 69 +// #define MT_SLAVE_FAIL_START_APP 70 +// typedef STRING_DATA MDF_SLAVE_FAIL_START_APP; +// #define MT_SLAVE_FAIL_STOP_APP 71 +// #define MT_SLAVE_FAIL_KILL_APP 72 +// #define MT_SLAVE_APP_SHUTODWN_COMPLETE 74 +// #define MT_SLAVE_APP_RESTART_COMPLETE 75 +// #define MT_SLAVE_APP_KILL_COMPLETE 76 +// #define MT_SLAVE_AM_ERROR 77 +// typedef STRING_DATA MDF_SLAVE_AM_ERROR; // Error in Module @@ -271,4 +275,67 @@ typedef struct { double send_time; } MDF_TIMING_MESSAGE; +#define MT_CONNECT_V2 4 +typedef struct { + int16_t logger_status; + int16_t daemon_status; + int16_t allow_multiple; + MODULE_ID mod_id; + int32_t pid; + char name[MAX_NAME_LEN]; +} MDF_CONNECT_V2; + +#define MT_SUBSCRIPTION_OPTION 70 +typedef struct { + MSG_TYPE msg_type; + int32_t option; + int32_t value; +} MDF_SUBSCRIPTION_OPTION; + +#define MT_PING 29 +typedef struct { + int32_t uid; + MODULE_ID dest_id; +} MDF_PING; + +#define MT_PONG 30 +typedef struct { + int32_t uid; + MODULE_ID src_id; +} MDF_PONG; + +#define MT_INTRODUCE 31 + +#define MT_HELLO 32 +typedef struct { + int32_t uid; + int32_t pid; + MODULE_ID mod_id; + uint16_t port; + char addr[32]; + char name[MAX_NAME_LEN]; +} MDF_HELLO; + +#define MT_GOODBYE 33 +typedef struct { + int32_t uid; + int32_t pid; + MODULE_ID mod_id; + uint16_t port; + char addr[32]; + char name[MAX_NAME_LEN]; +} MDF_GOODBYE; + +#define MT_CLIENT_SET_NAME 34 +typedef struct { + char name[MAX_NAME_LEN]; +} MDF_CLIENT_SET_NAME; + +#define MT_MESSAGE_TIMING 69 +typedef struct { + MSG_TYPE msg_type; + MODULE_ID src_id; + double send_time; +} MDF_MESSAGE_TIMING; + #endif //_RTMA_TYPES_H_ diff --git a/include/internal/UPipe.h b/include/internal/UPipe.h index e7ba13f..c70ce22 100644 --- a/include/internal/UPipe.h +++ b/include/internal/UPipe.h @@ -151,6 +151,8 @@ class UPipe // bytes written. virtual int Write( void *data_buffer, int n_bytes, double timeout) = 0; + void GetIpAddress(char *addr, uint16_t *port, int bufsz) {}; + }; diff --git a/src/core/PipeLib/SocketPipe.cpp b/src/core/PipeLib/SocketPipe.cpp index 8bd4dba..8d42cde 100644 --- a/src/core/PipeLib/SocketPipe.cpp +++ b/src/core/PipeLib/SocketPipe.cpp @@ -337,6 +337,20 @@ SocketPipe::GetCapacity( void) } CATCH_and_THROW( "SocketPipe::GetCapacity"); } +void +SocketPipe::GetIpAddress(char* addr, uint16_t *port, int bufsz) { + // windows: https://learn.microsoft.com/en-us/windows/win32/api/winsock2/nf-winsock2-getsockname + // linux: https://man7.org/linux/man-pages/man2/getsockname.2.html + struct sockaddr_in name = { 0 }; + int namelen = sizeof(name); + if (getsockname(_hPipe.id, (struct sockaddr*)&name, &namelen) == SOCKET_ERROR) { + return; + } + + *port = ntohs(name.sin_port); + inet_ntop(AF_INET, &(name.sin_addr), &(addr[0]), bufsz); +} + int SocketPipe::Read( void *data_buffer, int nbytes_to_read, double timeout) { diff --git a/src/core/PipeLib/SocketPipe.h b/src/core/PipeLib/SocketPipe.h index 5bfa057..213409c 100644 --- a/src/core/PipeLib/SocketPipe.h +++ b/src/core/PipeLib/SocketPipe.h @@ -16,6 +16,7 @@ #define WINSOCK_H #define FD_SETSIZE UPipeServer::MAX_CLIENTS #include + #include #endif #else //LINUX only h files @@ -71,6 +72,7 @@ class SocketPipe : public UPipe int GetCapacity( void); int Read( void *data_buffer, int n_bytes, double timeout); int Write( void *data_buffer, int n_bytes, double timeout); + void GetIpAddress(char *addr, uint16_t *port, int bufsz); private: SocketHandle _hPipe; // Handle to underlying OS socket diff --git a/src/modules/MessageManager/MessageManager.cpp b/src/modules/MessageManager/MessageManager.cpp index e4199e1..dd57e65 100644 --- a/src/modules/MessageManager/MessageManager.cpp +++ b/src/modules/MessageManager/MessageManager.cpp @@ -42,23 +42,10 @@ int main( int argc, char *argv[]) CMessageManager::CMessageManager( ) { m_Version = "2.11bci"; - m_NextDynamicModIdOffset = 0; + m_NextDynamicModId = DYN_MOD_ID_START; InitializeAbsTime(); - // from RP3 RTMA (for timing message) - #ifdef __unix__ - ftime(&timebuffer); - #else - _ftime(&timebuffer); // C4996 - #endif - - m_LastMessageCount = timebuffer.time; - m_LastMessageCountmsec = timebuffer.millitm; - for(int i=0;i 0) && (mod_id < MAX_MODULES)) ? 1 : 0; bool message_from_this_host = (M->src_host_id == HID_LOCAL_HOST) ? 1 : 0; - //For keeping track of message timing: - if (M->msg_type>0 && M->msg_typemsg_type]++; - - #ifdef __unix__ - ftime(&timebuffer); // C4996 - #else - _ftime(&timebuffer); // C4996 - #endif - - time_t t = timebuffer.time; - unsigned short tmsec = timebuffer.millitm; - - if ((t-m_LastMessageCount)>1 || (tmsec-m_LastMessageCountmsec)>900) //time to send out message with timing info - { - SendMessageTiming(); - #ifdef __unix__ - ftime(&timebuffer); // C4996 - #else - _ftime(&timebuffer); // C4996 - #endif - - m_LastMessageCount = timebuffer.time; - m_LastMessageCountmsec = timebuffer.millitm; - } - switch( M->msg_type) { - ////////DEBUG++++++++++++ - //case 1750:{ - //double t = GetAbsTime(); - //printf("%.3f\n",t); - //fflush(stdout); - //break; - //} - ////////DEBUG------------ case MT_CONNECT: MDF_CONNECT data; memset( &data, 0, sizeof(data)); @@ -184,11 +144,34 @@ CMessageManager::ProcessMessage( CMessage *M, UPipe *pSourcePipe) SetMyPriority(NORMAL_PRIORITY_CLASS); mod_id = ConnectModule( mod_id, pSourcePipe, data.logger_status, data.daemon_status); if (mod_id > 0) { - SendAcknowledge( mod_id); SetMyPriority(prev_priority_class); } break; - + case MT_CONNECT_V2: + MDF_CONNECT_V2 connect_v2; + M->GetData((void*) &connect_v2); + prev_priority_class = GetMyPriority(); + SetMyPriority(NORMAL_PRIORITY_CLASS); + mod_id = ConnectModuleV2( mod_id, pSourcePipe, &connect_v2); + if (mod_id > 0) { + SendAcknowledge(mod_id); + SetMyPriority(prev_priority_class); + } + break; + case MT_CLIENT_SET_NAME: + MDF_CLIENT_SET_NAME set_name; + M->GetData((void*) &set_name); + m_ConnectedModules[M->src_mod_id].SetName(set_name.name); + break; + case MT_SUBSCRIPTION_OPTION: + MDF_SUBSCRIPTION_OPTION sub_opt; + M->GetData((void*) &sub_opt); + SubscriptionOption(mod_id, &sub_opt); + break; + case MT_PONG: + break; + case MT_INTRODUCE: + break; case MT_FORCE_DISCONNECT: MDF_FORCE_DISCONNECT data_MDF_FORCE_DISCONNECT; M->GetData( &data_MDF_FORCE_DISCONNECT); @@ -198,30 +181,33 @@ CMessageManager::ProcessMessage( CMessage *M, UPipe *pSourcePipe) MyCString err("MM got MT_FORCE_DISCONNECT with invalid module id ["); err += (int)mod_id; err += "]"; - CMessage R(MT_MM_ERROR, (void*)err.GetContent(), err.GetLen()); - DispatchMessage(&R); + m_OutMsg.Set(MT_MM_ERROR, (void*)err.GetContent(), err.GetLen()); + DispatchMessage(&m_OutMsg); } else{ MyCString info("MM forcing disconnect on module ["); info += (int)mod_id; info += "]"; - CMessage R(MT_MM_INFO, (void*)info.GetContent(), info.GetLen()); - DispatchMessage(&R); + m_OutMsg.Set(MT_MM_INFO, (void*)info.GetContent(), info.GetLen()); + DispatchMessage(&m_OutMsg); ShutdownModule(mod_id); } break; case MT_DISCONNECT: prev_priority_class = GetMyPriority(); SetMyPriority(NORMAL_PRIORITY_CLASS); - DisconnectModule( mod_id); + DisconnectModule(mod_id); SetMyPriority(prev_priority_class); break; case MT_MODULE_READY: //store pids so that application module can kill processes later MDF_MODULE_READY m; + MDF_HELLO hello; M->GetData(&m); - if (mod_id>=0 && mod_idsrc_mod_id].pid = m.pid; + + // Notify that the module has joined and ready + SendHello(M->src_mod_id); break; case MT_SUBSCRIBE: MSG_TYPE msg_type_to_subscribe; @@ -274,6 +260,39 @@ CMessageManager::ProcessMessage( CMessage *M, UPipe *pSourcePipe) DEBUG_TEXT( "Processed!"); } +void CMessageManager::SendHello(MODULE_ID mod_id) { + MDF_HELLO hello; + CModuleRecord mod = m_ConnectedModules[mod_id]; + mod.SetHello(&hello); + m_OutMsg.Set(MT_HELLO, &hello, sizeof(hello)); + DispatchMessage(&m_OutMsg); +} + +void CMessageManager::SendGoodbye(MODULE_ID mod_id) { + MDF_GOODBYE goodbye; + CModuleRecord *mod = GetRecord(mod_id); + SendGoodbye(mod); +} + +void CMessageManager::SendGoodbye(CModuleRecord *mod) { + MDF_GOODBYE goodbye; + if (mod) { + mod->SetGoodbye(&goodbye); + m_OutMsg.Set(MT_GOODBYE, &goodbye, sizeof(goodbye)); + DispatchMessage(&m_OutMsg); + } +} + +void CMessageManager::SendPing(MODULE_ID mod_id) { + MDF_PING ping; + CModuleRecord *mod = GetRecord(mod_id); + if (mod) { + ping.uid = 0; + m_OutMsg.Set(MT_PING, &ping, sizeof(ping)); + DispatchMessage(&m_OutMsg); + } +} + /* * Should be called when forwarding a message from other modules * The given message will be forwarded to: @@ -283,27 +302,29 @@ CMessageManager::ProcessMessage( CMessage *M, UPipe *pSourcePipe) * - if the message has no destination address, it will be forwarded to all subscribed modules */ void -CMessageManager::DistributeMessage( CMessage *M) +CMessageManager::DistributeMessage(CMessage *M) { DEBUG_TEXT( "Distributing Message..."); CSubscriberList * SL; - SL = GetSubscriberList( M->msg_type); - if( SL != NULL) + SL = GetSubscriberList(M->msg_type); + if (SL != NULL) { - MODULE_ID mod_id = SL->GetFirstSubscriber(); - while( mod_id > 0) + UID uid = SL->GetFirstSubscriber(); + while (uid >= 0) { /* the order of the code in this while loop is important don't modify it unless you know what you're doing */ + CModuleRecord mod = m_ConnectedModules[uid]; + int send_it = 0; - int has_specific_dest = ( M->dest_mod_id == 0)? 0 : 1; + int has_specific_dest = (M->dest_mod_id == 0)? 0 : 1; if(has_specific_dest) { //send only to the specific destination - if(mod_id == M->dest_mod_id) + if(mod.ModuleID == M->dest_mod_id) send_it = 1; else send_it = 0; @@ -313,7 +334,7 @@ CMessageManager::DistributeMessage( CMessage *M) } //forward everything to logger modules - if( m_ConnectedModules[mod_id].LoggerStatus) + if( mod.LoggerStatus) send_it = 1; if( SL->SubscriptionPaused()) @@ -324,9 +345,20 @@ CMessageManager::DistributeMessage( CMessage *M) DEBUG_TEXT_( "Forwarding message to module " << mod_id << "... "); try { - int status = ForwardMessage(M, mod_id); - if( status == 0) { - LogFailedMessage( M, mod_id); + int status = 0; + if (SL->TimingOnly()) { + MDF_MESSAGE_TIMING msg_time; + msg_time.msg_type = M->msg_type; + msg_time.src_id = M->src_mod_id; + msg_time.send_time = M->send_time; + m_OutMsg.Set(MT_MESSAGE_TIMING, &msg_time, sizeof(msg_time)); + status = SendMessage(&m_OutMsg, &mod); + } else { + status = ForwardMessage(M, &mod); + } + + if(status == 0) { + LogFailedMessage(M, mod.ModuleID); DEBUG_TEXT( "Failed to Forward Message!"); } else { DEBUG_TEXT( "Forwarded!"); @@ -337,7 +369,7 @@ CMessageManager::DistributeMessage( CMessage *M) } } - mod_id = SL->GetNextSubscriber(); + uid = SL->GetNextSubscriber(); } } DEBUG_TEXT( "Done distributing!"); @@ -348,19 +380,20 @@ CMessageManager::DistributeMessage( CMessage *M) * The message will be sent to all subscribed modules including loggers */ void -CMessageManager::DispatchMessage( CMessage *M) +CMessageManager::DispatchMessage(CMessage *M) { CSubscriberList * SL; - SL = GetSubscriberList( M->msg_type); + SL = GetSubscriberList(M->msg_type); if( SL != NULL) { // Send message to all subscribers - MODULE_ID mod_id = SL->GetFirstSubscriber(); - while( mod_id > 0) + UID uid = SL->GetFirstSubscriber(); + while(uid > 0) { - SendMessage(M, mod_id); - mod_id = SL->GetNextSubscriber(); + CModuleRecord mod = m_ConnectedModules[uid]; + SendMessage(M, &mod); + uid = SL->GetNextSubscriber(); } } } @@ -372,27 +405,28 @@ CMessageManager::DispatchMessage( CMessage *M) * The message will also be forwarded to all subscribed logger modules */ void -CMessageManager::DispatchMessage( CMessage *M, MODULE_ID mod_id) +CMessageManager::DispatchMessage(CMessage *M, CModuleRecord *dest_mod) { CSubscriberList * SL; //send the message to the module it is intended to, disregarding subscriptions //(enables MM to send system message to modules) - SendMessage(M, mod_id); + SendMessage(M, dest_mod); //CC all logger modules - SL = GetSubscriberList( M->msg_type); + SL = GetSubscriberList(M->msg_type); if( SL != NULL) { - MODULE_ID cc_mod_id = SL->GetFirstSubscriber(); + UID uid = SL->GetFirstSubscriber(); - while( cc_mod_id > 0) + while(uid >= 0) { - if( m_ConnectedModules[cc_mod_id].LoggerStatus) - if(cc_mod_id != mod_id)//don't send to destination module again - ForwardMessage(M, cc_mod_id); + CModuleRecord mod = m_ConnectedModules[uid]; + if(mod.LoggerStatus) + if(mod.ModuleID != dest_mod->ModuleID)//don't send to destination module again + ForwardMessage(M, &mod); - cc_mod_id = SL->GetNextSubscriber(); + uid = SL->GetNextSubscriber(); } } } @@ -403,10 +437,10 @@ CMessageManager::DispatchMessage( CMessage *M, MODULE_ID mod_id) * The signal will also be forwarded to all subscribed logger modules */ void -CMessageManager::DispatchSignal( MSG_TYPE sig, MODULE_ID mod_id) +CMessageManager::DispatchSignal(MSG_TYPE sig, CModuleRecord *dest_mod) { - CMessage R(sig); - DispatchMessage(&R, mod_id); + m_OutMsg.Set(sig); + DispatchMessage(&m_OutMsg, dest_mod); } /* @@ -415,12 +449,14 @@ CMessageManager::DispatchSignal( MSG_TYPE sig, MODULE_ID mod_id) * The message will also be forwarded to all subscribed logger modules */ void -CMessageManager::DispatchMessageToAll( CMessage *M) +CMessageManager::DispatchMessageToAll(CMessage *M) { - for(int mod_id=0; mod_id= 0 && mod.ModuleID > 0){ + DispatchMessage(M, &mod); + } } } @@ -433,13 +469,11 @@ CMessageManager::DispatchSignalToAll( MSG_TYPE sig) int -CMessageManager::SendMessage( CMessage *M, MODULE_ID mod_id) +CMessageManager::SendMessage(CMessage *M, CModuleRecord* mod) //overloaded function for RTMA_Module::SendMessage() //Sends a message to a module, specifying the MessageManager itself as the source module { - if( !ModuleIsConnected( mod_id)) return 0; - - UPipe *mod_pipe = GetModPipe( mod_id); + UPipe *mod_pipe = mod->pModulePipe; if(mod_pipe == NULL) return 0; // Assume that msg_type, num_data_bytes, data - have been filled in @@ -448,50 +482,66 @@ CMessageManager::SendMessage( CMessage *M, MODULE_ID mod_id) M->recv_time = 0.0; M->src_host_id = HID_LOCAL_HOST; M->src_mod_id = MID_MESSAGE_MANAGER; - M->dest_mod_id = mod_id; + M->dest_mod_id = mod->ModuleID; double timeout = 0; // By default use non-blocking write so MM does not get stuck on a frozen module - if( m_ConnectedModules[mod_id].LoggerStatus) timeout = -1; // Logger modules should not lose data, so use blocking write + if(mod->LoggerStatus) timeout = -1; // Logger modules should not lose data, so use blocking write - int status = M->Send( mod_pipe, timeout); + int status = M->Send(mod_pipe, timeout); return status; } int -CMessageManager::ForwardMessage( CMessage *M, MODULE_ID mod_id) +CMessageManager::ForwardMessage(CMessage *M, CModuleRecord* dest_mod) //Forward a message where the header is already filled in //Source module field in the header is unaltered { - if( !ModuleIsConnected( mod_id)) return 0; - - UPipe *mod_pipe = GetModPipe( mod_id); + UPipe *mod_pipe = dest_mod->pModulePipe; if(mod_pipe == NULL) return 0; double timeout = 0; // By default use non-blocking write so MM does not get stuck on a frozen module - if( m_ConnectedModules[mod_id].LoggerStatus) timeout = -1; // Logger modules should not lose data, so use blocking write + if(dest_mod->LoggerStatus) timeout = -1; // Logger modules should not lose data, so use blocking write - int status = M->Send( mod_pipe, timeout); + int status = M->Send(mod_pipe, timeout); return status; } int -CMessageManager::SendSignal( MSG_TYPE sig, MODULE_ID mod_id) +CMessageManager::SendSignal( MSG_TYPE sig, CModuleRecord *dest_mod) //overloaded function for RTMA_Module::SendSignal() //returns 0 if module is not connected, or failed to send message to it; 1 on success { - CMessage M(sig); - return SendMessage(&M, mod_id); + m_OutMsg.Set(sig); + return SendMessage(&m_OutMsg, dest_mod); +} + +CModuleRecord* +CMessageManager::GetOpenRecord(){ + for (int i=0;i= 0) && !ModuleIsConnected(module_id) ) + if( ( module_id < MAX_MODULE_ID) && ( module_id >= 0) && !ModuleIsConnected(module_id) ) { if( pSourcePipe != NULL) { + CModuleRecord* mod = GetOpenRecord(); + if (mod == NULL) { + printf("MessageManager has reached the MAX_MODULES allowed\n."); + return -1; + } // get the next available module ID from "dynamic" pool if (module_id == 0) @@ -500,17 +550,61 @@ CMessageManager::ConnectModule( MODULE_ID module_id, UPipe *pSourcePipe, short l if (module_id > 0) { DEBUG_TEXT( "Connecting module " << module_id << " on pipe " << pSourcePipe); - m_ConnectedModules[module_id].Reset(); // Create a module record - m_ConnectedModules[module_id].ModuleID = module_id; - m_ConnectedModules[module_id].pModulePipe = pSourcePipe; - m_ConnectedModules[module_id].LoggerStatus = logger_status; - m_ConnectedModules[module_id].DaemonStatus = daemon_status; + mod->ModuleID = module_id; + mod->pModulePipe = pSourcePipe; + mod->LoggerStatus = logger_status; + mod->DaemonStatus = daemon_status; + + // Add default subscription to TIMER_EXPIRED + if( !logger_status) AddSubscription( mod->uid, MT_TIMER_EXPIRED); + + SendAcknowledge(mod); + } + } + } + else + module_id = 0; // something went wrong, don't allow the new connection + + return module_id; +} + +MODULE_ID +CMessageManager::ConnectModuleV2( MODULE_ID module_id, UPipe *pSourcePipe, MDF_CONNECT_V2 *data) +{ + + if( ( module_id < MAX_MODULE_ID) && ( module_id >= 0) && !ModuleIsConnected(module_id) ) + { + if( pSourcePipe != NULL) { + CModuleRecord* mod = GetOpenRecord(); + if (mod == NULL) { + printf("MessageManager has reached the MAX_MODULES allowed\n."); + return -1; + } + + // get the next available module ID from "dynamic" pool + if (module_id == 0) + module_id = GetDynamicModuleId(); + + if (module_id > 0) { + DEBUG_TEXT( "Connecting module " << module_id << " on pipe " << pSourcePipe); + + mod->ModuleID = module_id; + mod->pModulePipe = pSourcePipe; + mod->LoggerStatus = data->logger_status; + mod->DaemonStatus = data->daemon_status; + mod->AllowMultiple = data->allow_multiple; + mod->pid = data->pid; + mod->SetName(data->name); // Add default subscription to TIMER_EXPIRED - if( !logger_status) AddSubscription( module_id, MT_TIMER_EXPIRED); + if( !(data->logger_status)) AddSubscription( mod->uid, MT_TIMER_EXPIRED); + + SendAcknowledge(mod); } } + // Notify that a V2 client has connected + SendHello(module_id); } else module_id = 0; // something went wrong, don't allow the new connection @@ -518,34 +612,53 @@ CMessageManager::ConnectModule( MODULE_ID module_id, UPipe *pSourcePipe, short l return module_id; } +CModuleRecord* CMessageManager::GetRecord(MODULE_ID module_id) { + for (int i=0;iDisconnectClient( module_pipe); + UPipeAutoServer::_server->DisconnectClient(mod->pModulePipe); + + // Make sure not to send anything to disconnected client + RemoveSubscription(mod, ALL_MESSAGE_TYPES); + + // Notify that the module has left + SendGoodbye(mod); // Clean up module record - CleanUpModuleRecord( module_id); + CleanUpModuleRecord(mod); } void -CMessageManager::CleanUpModuleRecord( MODULE_ID module_id) +CMessageManager::CleanUpModuleRecord(CModuleRecord *mod) { - RemoveSubscription( module_id, ALL_MESSAGE_TYPES); - m_ConnectedModules[module_id].Reset(); + RemoveSubscription(mod, ALL_MESSAGE_TYPES); + mod->Reset(); } void -CMessageManager::ShutdownModule( MODULE_ID mod_id) +CMessageManager::ShutdownModule(MODULE_ID mod_id) { - if( ModuleIsConnected( mod_id)) + CModuleRecord* mod = GetRecord(mod_id); + if(mod != NULL && mod->ModuleID > 0) { + MODULE_ID mod_id = mod->ModuleID; switch(mod_id) { case MID_COMMAND_MODULE: @@ -553,28 +666,28 @@ CMessageManager::ShutdownModule( MODULE_ID mod_id) break; case MID_APPLICATION_MODULE: - DispatchSignal( MT_AM_EXIT, mod_id); + DispatchSignal(MT_AM_EXIT, mod); break; case MID_STATUS_MODULE: - DispatchSignal( MT_SM_EXIT, mod_id); + DispatchSignal(MT_SM_EXIT, mod); break; case MID_QUICKLOGGER: - DispatchSignal( MT_LM_EXIT, mod_id); + DispatchSignal(MT_LM_EXIT, mod); break; default: - DispatchSignal( MT_EXIT, mod_id); + DispatchSignal(MT_EXIT, mod); break; } - DisconnectModule( mod_id); + DisconnectModule(mod_id); } } void -CMessageManager::ShutdownAllModules( int shutdown_RTMA, int shutdown_daemons) +CMessageManager::ShutdownAllModules(int shutdown_RTMA, int shutdown_daemons) { int mod_id; int start_mod_id; @@ -586,8 +699,9 @@ CMessageManager::ShutdownAllModules( int shutdown_RTMA, int shutdown_daemons) start_mod_id = MAX_RTMA_MODULE_ID+1; //shutdown only application modules } - for( mod_id = start_mod_id; mod_id < MAX_MODULES; mod_id++) { - if( IsDaemon(mod_id)) { + for (int uid = 0; uid < MAX_MODULES; uid++) { + CModuleRecord mod = m_ConnectedModules[uid]; + if(mod.DaemonStatus) { if( shutdown_daemons) ShutdownModule(mod_id); } else { ShutdownModule(mod_id); @@ -596,145 +710,193 @@ CMessageManager::ShutdownAllModules( int shutdown_RTMA, int shutdown_daemons) } void -CMessageManager::AddSubscription( MODULE_ID module_id, MSG_TYPE message_type) +CMessageManager::AddSubscription(MODULE_ID mod_id, MSG_TYPE message_type) { MSG_TYPE mt; - if( ((message_type < 0) || (message_type > MAX_MESSAGE_TYPES)) && (message_type != ALL_MESSAGE_TYPES) ) - { - //send MDF_FAIL_SUBSCRIBE instead of ACK so the module's subscribe function will fail - MDF_FAIL_SUBSCRIBE data; - data.mod_id = module_id; - data.msg_type = message_type; - CMessage R(MT_FAIL_SUBSCRIBE, (void*)&data, sizeof(data)); - DispatchMessage(&R,module_id); - return; - } + CModuleRecord *mod = GetRecord(mod_id); + AddSubscription(mod, message_type); +} - switch( message_type) - { - case ALL_MESSAGE_TYPES: - for( mt = 0; mt < MAX_MESSAGE_TYPES; mt++) { - GetSubscriberList( mt)->AddSubscriber( module_id); +void +CMessageManager::AddSubscription(CModuleRecord *mod, MSG_TYPE message_type) +{ + MSG_TYPE mt; + + if (mod) { + if( ((message_type < 0) || (message_type > MAX_MESSAGE_TYPES)) && (message_type != ALL_MESSAGE_TYPES) ) + { + //send MDF_FAIL_SUBSCRIBE instead of ACK so the module's subscribe function will fail + MDF_FAIL_SUBSCRIBE data; + data.mod_id = mod->ModuleID; + data.msg_type = message_type; + CMessage R(MT_FAIL_SUBSCRIBE, (void*)&data, sizeof(data)); + DispatchMessage(&R, mod); + return; + } + + switch( message_type) + { + case ALL_MESSAGE_TYPES: + for( mt = 0; mt < MAX_MESSAGE_TYPES; mt++) { + GetSubscriberList(mt)->AddSubscriber(mod->uid); + } + break; + default: + CSubscriberList* list = GetSubscriberList(message_type); + if(!list->IsSubscribed(mod->uid)) + list->AddSubscriber(mod->uid); } - break; - default: - CSubscriberList* list = GetSubscriberList( message_type); - if(!list->IsSubscribed(module_id)) - list->AddSubscriber( module_id); } } void -CMessageManager::RemoveSubscription( MODULE_ID module_id, MSG_TYPE message_type) +CMessageManager::RemoveSubscription(MODULE_ID mod_id, MSG_TYPE message_type) { // Might wanna add checks here for valid module_id, connected module MSG_TYPE mt; - switch( message_type) { - case ALL_MESSAGE_TYPES: - for( mt = 0; mt < MAX_MESSAGE_TYPES; mt++) { - GetSubscriberList( mt)->RemoveSubscriber( module_id); + CModuleRecord *mod = GetRecord(mod_id); + RemoveSubscription(mod, message_type); +} + +void +CMessageManager::RemoveSubscription(CModuleRecord *mod, MSG_TYPE message_type) +{ + // Might wanna add checks here for valid module_id, connected module + MSG_TYPE mt; + + if (mod) { + switch( message_type) { + case ALL_MESSAGE_TYPES: + for( mt = 0; mt < MAX_MESSAGE_TYPES; mt++) { + GetSubscriberList(mt)->RemoveSubscriber(mod->uid); + } + break; + default: + GetSubscriberList(message_type)->RemoveSubscriber(mod->uid); } - break; - default: - GetSubscriberList( message_type)->RemoveSubscriber( module_id); } } void -CMessageManager::PauseSubscription( MODULE_ID mod_id, MSG_TYPE message_type) +CMessageManager::PauseSubscription(MODULE_ID mod_id, MSG_TYPE message_type) { // Might wanna add checks here for valid module_id, connected module MSG_TYPE mt; - switch( message_type) { - case ALL_MESSAGE_TYPES: - for( mt = 0; mt < MAX_MESSAGE_TYPES; mt++) { - GetSubscriberList( mt)->PauseSubscriber( mod_id); + CModuleRecord *mod = GetRecord(mod_id); + if (mod) { + switch( message_type) { + case ALL_MESSAGE_TYPES: + for( mt = 0; mt < MAX_MESSAGE_TYPES; mt++) { + GetSubscriberList(mt)->PauseSubscriber(mod->uid); + } + break; + default: + GetSubscriberList(message_type)->PauseSubscriber(mod->uid); } - break; - default: - GetSubscriberList( message_type)->PauseSubscriber( mod_id); } } void -CMessageManager::ResumeSubscription( MODULE_ID mod_id, MSG_TYPE message_type) +CMessageManager::ResumeSubscription(MODULE_ID mod_id, MSG_TYPE message_type) { // Might wanna add checks here for valid module_id, connected module MSG_TYPE mt; - switch( message_type) { - case ALL_MESSAGE_TYPES: - for( mt = 0; mt < MAX_MESSAGE_TYPES; mt++) { - GetSubscriberList( mt)->ResumeSubscriber( mod_id); + CModuleRecord *mod = GetRecord(mod_id); + if (mod) { + switch( message_type) { + case ALL_MESSAGE_TYPES: + for( mt = 0; mt < MAX_MESSAGE_TYPES; mt++) { + GetSubscriberList(mt)->ResumeSubscriber(mod->uid); + } + break; + default: + GetSubscriberList(message_type)->ResumeSubscriber(mod->uid); + } + } +} + +void +CMessageManager::SubscriptionOption(MODULE_ID mod_id, MDF_SUBSCRIPTION_OPTION *sub_opt) +{ + CModuleRecord *mod = GetRecord(mod_id); + if (mod) { + if (sub_opt->value) { + GetSubscriberList(sub_opt->msg_type)->SetSubscriberOption(mod->uid, sub_opt->option); + } else { + GetSubscriberList(sub_opt->msg_type)->ClearSubscriberOption(mod->uid, sub_opt->option); } - break; - default: - GetSubscriberList( message_type)->ResumeSubscriber( mod_id); } } CSubscriberList * -CMessageManager::GetSubscriberList( MSG_TYPE message_type) +CMessageManager::GetSubscriberList(MSG_TYPE message_type) { - if( message_type >= MAX_MESSAGE_TYPES || message_type < 0) + if (message_type >= MAX_MESSAGE_TYPES || message_type < 0) return &m_EmptySubscriberList; else return &(m_SubscribersToMessageType[message_type]); } bool -CMessageManager::IsModuleSubscribed( MODULE_ID mod_id, MSG_TYPE message_type) +CMessageManager::IsModuleSubscribed(UID uid, MSG_TYPE message_type) { - if( message_type >= MAX_MESSAGE_TYPES || message_type < 0 || - mod_id < MID_MESSAGE_MANAGER || mod_id > MAX_MODULES ) + if (message_type >= MAX_MESSAGE_TYPES || message_type < 0) return false; - return GetSubscriberList( message_type)->IsSubscribed(mod_id); + return GetSubscriberList(message_type)->IsSubscribed(uid); } void -CMessageManager::SendAcknowledge( MODULE_ID mod_id) +CMessageManager::SendAcknowledge(MODULE_ID mod_id) { - DEBUG_TEXT_( "Sending ACK to module " << mod_id << "... "); - DispatchSignal( MT_ACKNOWLEDGE, mod_id); - DEBUG_TEXT( "Sent!"); + CModuleRecord *mod = GetRecord(mod_id); + SendAcknowledge(mod); } - - -int -CMessageManager::ModuleIsConnected( MODULE_ID mod_id) +void +CMessageManager::SendAcknowledge(CModuleRecord * mod) { - if( mod_id < MID_MESSAGE_MANAGER) - return 0; - if( mod_id < MAX_MODULES) { - if( m_ConnectedModules[mod_id].ModuleID == mod_id) { - return 1; - } + if (mod) { + DEBUG_TEXT_( "Sending ACK to module " << mod_id << "... "); + DispatchSignal( MT_ACKNOWLEDGE, mod); + DEBUG_TEXT( "Sent!"); } - return 0; } -UPipe* -CMessageManager::GetModPipe( MODULE_ID mod_id) -{ - if( !ModuleIsConnected( mod_id)) return NULL; - UPipe *mod_pipe = m_ConnectedModules[mod_id].pModulePipe; - return mod_pipe; +void +CMessageManager::SendIntroductions(MODULE_ID mod_id) +{ + MDF_HELLO hello; + CModuleRecord *dest_mod = GetRecord(mod_id); + if (dest_mod) { + for (int i=0; i< MAX_MODULES; i++){ + CModuleRecord mod = m_ConnectedModules[i]; + if (mod.uid >= 0 && mod.ModuleID > 0){ + mod.SetHello(&hello); + m_OutMsg.Set(MT_HELLO, &hello, sizeof(hello)); + SendMessage(&m_OutMsg, dest_mod); + } + } + } } + int -CMessageManager::IsDaemon( MODULE_ID mod_id) +CMessageManager::ModuleIsConnected(MODULE_ID mod_id) { - if( !ModuleIsConnected( mod_id)) return 0; - if( m_ConnectedModules[mod_id].DaemonStatus) { - return 1; - } else { + if( mod_id < MID_MESSAGE_MANAGER) return 0; + + for (int i=0; iuid = uid; + hello->mod_id = ModuleID; + hello->pid = pid; + pModulePipe->GetIpAddress(&(hello->addr[0]), &(hello->port), sizeof(hello->addr)); + memcpy(hello->name, name, sizeof(hello->name)); + } + + void SetGoodbye(MDF_GOODBYE *goodbye) { + goodbye->uid = uid; + goodbye->mod_id = ModuleID; + goodbye->pid = pid; + pModulePipe->GetIpAddress(&(goodbye->addr[0]), &(goodbye->port), sizeof(goodbye->addr)); + memcpy(goodbye->name, name, sizeof(goodbye->name)); + } + void Reset( void) { ModuleID = -1; pModulePipe = NULL; LoggerStatus = 0; DaemonStatus = 0; + AllowMultiple = 0; + pid = 0; + uid = -1; + if (name != NULL) { + free(name); + name = NULL; + } + } + + ~CModuleRecord(){ + Reset(); } }; @@ -124,7 +165,8 @@ class CList } }; -#define SUBSCRIBER_FLAG_PAUSE 0x01 +#define SUBSCRIBER_FLAG_PAUSE 0x01 +#define SUBSCRIBER_FLAG_TIMING_ONLY 0x02 class CSubscriberList : protected CList { @@ -139,28 +181,28 @@ class CSubscriberList : protected CList } void - AddSubscriber( MODULE_ID ModuleID) { - CListItem *subscriber = new CListItem( ModuleID); + AddSubscriber(UID uid) { + CListItem *subscriber = new CListItem(uid); AppendItem( subscriber); } void - RemoveSubscriber( MODULE_ID ModuleID) { + RemoveSubscriber(UID uid) { CListItem *current_item = GetFirstItem(); while( current_item != NULL) { - if( current_item->data == ModuleID) { - RemoveItem( current_item); + if( current_item->data == uid) { + RemoveItem(current_item); break; } - current_item = GetNextItem( current_item); + current_item = GetNextItem(current_item); } } void - PauseSubscriber( MODULE_ID ModuleID) { + PauseSubscriber(UID uid) { CListItem *current_item = GetFirstItem(); while( current_item != NULL) { - if( current_item->data == ModuleID) { + if( current_item->data == uid) { set_flag_bits( current_item->flags, SUBSCRIBER_FLAG_PAUSE); break; } @@ -169,10 +211,10 @@ class CSubscriberList : protected CList } void - ResumeSubscriber( MODULE_ID ModuleID) { + ResumeSubscriber(UID uid) { CListItem *current_item = GetFirstItem(); while( current_item != NULL) { - if( current_item->data == ModuleID) { + if( current_item->data == uid) { clear_flag_bits( current_item->flags, SUBSCRIBER_FLAG_PAUSE); break; } @@ -180,8 +222,52 @@ class CSubscriberList : protected CList } } + void + SetSubscriberOption(UID uid, int option) { + // Only act on valid options + switch(option) { + case SUBSCRIBER_FLAG_PAUSE: + break; + case SUBSCRIBER_FLAG_TIMING_ONLY: + break; + default: + return; + } + + CListItem *current_item = GetFirstItem(); + while( current_item != NULL) { + if( current_item->data == uid) { + set_flag_bits( current_item->flags, option); + break; + } + current_item = GetNextItem( current_item); + } + } + + void + ClearSubscriberOption(UID uid, int option) { + // Only act on valid options + switch(option) { + case SUBSCRIBER_FLAG_PAUSE: + break; + case SUBSCRIBER_FLAG_TIMING_ONLY: + break; + default: + return; + } + + CListItem *current_item = GetFirstItem(); + while( current_item != NULL) { + if( current_item->data == uid) { + clear_flag_bits( current_item->flags, option); + break; + } + current_item = GetNextItem( current_item); + } + } + int - SubscriptionPaused( void) { + SubscriptionPaused(void) { int is_paused = 0; if( m_CurrentItem != NULL) { if( check_flag_bits( m_CurrentItem->flags, SUBSCRIBER_FLAG_PAUSE)) { @@ -191,8 +277,19 @@ class CSubscriberList : protected CList return is_paused; } - MODULE_ID - GetFirstSubscriber( void) { + int + TimingOnly(void) { + int is_timing_only = 0; + if( m_CurrentItem != NULL) { + if( check_flag_bits( m_CurrentItem->flags, SUBSCRIBER_FLAG_TIMING_ONLY)) { + is_timing_only = 1; + } + } + return is_timing_only; + } + + UID + GetFirstSubscriber(void) { m_CurrentItem = GetFirstItem(); if( m_CurrentItem == NULL) { return -1; @@ -201,8 +298,8 @@ class CSubscriberList : protected CList } } - MODULE_ID - GetNextSubscriber( void) { + UID + GetNextSubscriber(void) { m_CurrentItem = GetNextItem( m_CurrentItem); if( m_CurrentItem == NULL) { return -1; @@ -212,8 +309,8 @@ class CSubscriberList : protected CList } bool - IsSubscribed(MODULE_ID ModuleID){ - CListItem subscriber = CListItem( ModuleID); + IsSubscribed(UID uid){ + CListItem subscriber = CListItem(uid); return DoesItemExist(&subscriber); } @@ -231,43 +328,28 @@ class CMessageManager : public UPipeAutoServer MainLoop( char *cmd_line_options); private: - MODULE_ID m_NextDynamicModIdOffset; + MODULE_ID m_NextDynamicModId; CModuleRecord m_ConnectedModules[MAX_MODULES]; CSubscriberList m_SubscribersToMessageType[MAX_MESSAGE_TYPES]; CSubscriberList m_EmptySubscriberList; + CMessage m_OutMsg; MyCString m_Version; - unsigned short m_MessageCounts[MAX_MESSAGE_TYPES]; - int m_ModulePIDs[MAX_MODULES]; - time_t m_LastMessageCount; - unsigned short m_LastMessageCountmsec; - #ifdef __unix__ - struct timeb timebuffer; - #else - struct _timeb timebuffer; - #endif MODULE_ID GetDynamicModuleId() { - for(MODULE_ID id=0; id < (MAX_MODULES - DYN_MOD_ID_START); id++) - { - MODULE_ID curr_id = m_NextDynamicModIdOffset + DYN_MOD_ID_START; + MODULE_ID curr_id = m_NextDynamicModId; + if (curr_id > MAX_MODULE_ID) { + printf("All dynamic IDs are in use.\n"); + return 0; + } + else { // update offset to next available dynamic module ID - m_NextDynamicModIdOffset++; - if (m_NextDynamicModIdOffset == MAX_MODULES-DYN_MOD_ID_START) - m_NextDynamicModIdOffset = 0; - - if (m_ConnectedModules[curr_id].ModuleID != curr_id) - return curr_id; + m_NextDynamicModId++; + return curr_id; } - DEBUG_TEXT("CMessageBuffer::GetDynamicModuleId(): All dynamic IDs are in use"); - return 0; } - void - SendMessageTiming(); - // Sends timing information to subscribed modules and resets the counter - void HandleData( UPipe *pClientPipe); // Override of abstract base class method. Gets called whenever data is ready on any input pipe. @@ -288,7 +370,7 @@ class CMessageManager : public UPipeAutoServer //Sends the message to all subscribers and Logger modules, headers specifying the message came from MM void - DispatchMessage( CMessage *M, MODULE_ID mod_id); + DispatchMessage( CMessage *M, CModuleRecord *dest_mod); //Sends the message only to the specified mod_id and Logger modules, headers specifying the message came from MM void @@ -296,7 +378,7 @@ class CMessageManager : public UPipeAutoServer //Sends the signal to all subscribers and Logger modules, headers specifying the signal came from MM void - DispatchSignal( MSG_TYPE sig, MODULE_ID mod_id); + DispatchSignal( MSG_TYPE sig, CModuleRecord *dest_mod); //Sends the signal only to the specified mod_id and Logger modules, headers specifying the message came from MM void @@ -310,63 +392,97 @@ class CMessageManager : public UPipeAutoServer MODULE_ID ConnectModule( MODULE_ID module_id, UPipe *pSourcePipe, short logger_status, short daemon_status); + MODULE_ID + ConnectModuleV2( MODULE_ID module_id, UPipe *pSourcePipe, MDF_CONNECT_V2 *connect); + + CModuleRecord* GetOpenRecord(); + + CModuleRecord* GetRecord(MODULE_ID module_id); + void - DisconnectModule( MODULE_ID module_id); + DisconnectModule(MODULE_ID module_id); void - CleanUpModuleRecord( MODULE_ID module_id); + CleanUpModuleRecord(CModuleRecord *mod); void - ShutdownModule( MODULE_ID mod_id); + ShutdownModule(MODULE_ID module_id); void - ShutdownAllModules( int shutdown_RTMA=1, int shutdown_daemons=1); + ShutdownAllModules(int shutdown_RTMA=1, int shutdown_daemons=1); void - ShutdownStatusModule( void); + ShutdownStatusModule(void); void - ShutdownLoggerModule( void); + ShutdownLoggerModule(void); void - AddSubscription( MODULE_ID module_id, MSG_TYPE message_type); + AddSubscription(MODULE_ID mod_id, MSG_TYPE message_type); void - RemoveSubscription( MODULE_ID module_id, MSG_TYPE message_type); + AddSubscription(CModuleRecord *mod, MSG_TYPE message_type); + + void + RemoveSubscription(MODULE_ID mod_id, MSG_TYPE message_type); + + void + RemoveSubscription(CModuleRecord *mod, MSG_TYPE message_type); void - PauseSubscription( MODULE_ID mod_id, MSG_TYPE msg_type_to_pause); + PauseSubscription(MODULE_ID mod_id, MSG_TYPE msg_type_to_pause); void - ResumeSubscription( MODULE_ID mod_id, MSG_TYPE msg_type_to_resume); + ResumeSubscription(MODULE_ID mod_id, MSG_TYPE msg_type_to_resume); + + void + SubscriptionOption(MODULE_ID mod_id, MDF_SUBSCRIPTION_OPTION *sub_opt); CSubscriberList * GetSubscriberList( MSG_TYPE message_type); bool - IsModuleSubscribed( MODULE_ID mod_id, MSG_TYPE message_type); + IsModuleSubscribed(UID uid, MSG_TYPE message_type); void - SendAcknowledge( MODULE_ID mod_id); + SendAcknowledge(MODULE_ID mod_id); + + void + SendAcknowledge(CModuleRecord* mod); + void + SendIntroductions(MODULE_ID mod_id); + + void + SendHello(MODULE_ID mod_id); + + void + SendGoodbye(MODULE_ID mod_id); + + void + SendGoodbye(CModuleRecord* mod); + + void + SendPing(MODULE_ID mod_id); + int - SendMessage( CMessage *m, MODULE_ID mod_id); + SendMessage( CMessage *m, CModuleRecord *dest_mod); int - SendSignal( MSG_TYPE sig, MODULE_ID dest_mod_id); + SendSignal( MSG_TYPE sig, CModuleRecord *dest_mod); //dest_mod_id= module id that should be put in dest_mod_id in message header int - ForwardMessage( CMessage *m, MODULE_ID mod_id); + ForwardMessage(CMessage *m, CModuleRecord* mod); int - ModuleIsConnected( MODULE_ID mod_id); + ModuleIsConnected(MODULE_ID mod_id); UPipe* - GetModPipe( MODULE_ID mod_id); + GetModPipe(UID uid); int - IsDaemon( MODULE_ID mod_id); + IsDaemon(UID uid); void ReportLoad( void); From 67c5ce2212d7f73402292a6d989e8f69e46904bb Mon Sep 17 00:00:00 2001 From: David Weir Date: Wed, 3 Dec 2025 10:24:48 -0500 Subject: [PATCH 02/17] Remove timer thread code --- src/core/RTMA.cpp | 290 ---------------------------------------------- 1 file changed, 290 deletions(-) diff --git a/src/core/RTMA.cpp b/src/core/RTMA.cpp index c6fa958..1863884 100644 --- a/src/core/RTMA.cpp +++ b/src/core/RTMA.cpp @@ -17,35 +17,6 @@ #endif #endif -//TIMER THREAD STUFF -typedef struct { - int timer_id; //unique timer_id - unsigned int snooze_time; //expiration time in ms - int cancel_timer; //can be set by the RTMA module to signal the thread to cancel this timer - double start_time; //time at which this timer was set by the RTMA module -}TIMER_INFO; -//global array accessed by the timer thread & the RTMA module -TIMER_INFO Gm_Timers[MAX_INTERNAL_TIMERS]; - -//TimerThread struct decleration -typedef struct { - int thread_exists; - int keep_running; //should the timer thread keep running - RTMA_Module *p; //pointer to this - int last_error; //used by the thread to signal error - TIMER_INFO* timers; //pointer to the global struct above -#ifdef _UNIX_C - pthread_t thread_handle; //handle to the thread - pthread_mutex_t tMutex; //protects Gm_Timers -#else - HANDLE thread_handle; //handle to the thread - HANDLE tMutex; //protects Gm_Timers -#endif -}TIMER_THREAD_INFO; -//the struct that is actually passed to the TimerThread -TIMER_THREAD_INFO Gm_TimerThreadInfo; //information that the thread needs for startup & operation - - ////////////////////////////////////////////////////////////////////////////// //////////////////// CMessage //////////////////////////////////////////////// ////////////////////////////////////////////////////////////////////////////// @@ -890,267 +861,6 @@ GetMyPriority() } CATCH_and_THROW( "GetMyPriority()"); } -////////////////////////////////////////////////////////////////////////////// -//////////////////// Internal Timer ////////////////////////////////////////// -////////////////////////////////////////////////////////////////////////////// -/* - * The global variables referred in this section are defined in the GLOBALS section of the code - */ - -//The timer thread -#ifdef _UNIX_C -void *TimerThread(void* lpParam) -#else -DWORD WINAPI -TimerThread( LPVOID lpParam) -#endif -{ - TIMER_THREAD_INFO *info = (TIMER_THREAD_INFO*) lpParam; - RTMA_Module *module = info->p; - TIMER_INFO *timers = info->timers; - int timers_expiring_soon=0; - int have_expired_timers =0; - int expired_timers[MAX_INTERNAL_TIMERS]; - - for(int i=0; i< MAX_INTERNAL_TIMERS; i++) - expired_timers[i]=0; - - while( info->keep_running ) - { - //lock mutex - #ifdef _UNIX_C - pthread_mutex_lock(&info->tMutex); - #else - DWORD status = WaitForSingleObject( info->tMutex, INFINITE ); - if(status == WAIT_ABANDONED) return 0; //RTMA module terminated while locking the mutex - if(status == WAIT_FAILED){ info->last_error = GetLastError(); return 0;} - #endif - - //look for expired timers - int slot=0; - timers_expiring_soon = 0; - while( (slot < MAX_INTERNAL_TIMERS) && info->keep_running ) - { - if(timers[slot].timer_id > 0) - { - if(timers[slot].cancel_timer) // a canceled timer - { - timers[slot].cancel_timer = 0; - timers[slot].snooze_time = 0; - timers[slot].start_time = 0; - timers[slot].timer_id = 0; - } - else //a valid timer - { - double snooze_time_sec = ((double)timers[slot].snooze_time)/1000; - if(timers[slot].start_time + snooze_time_sec < GetAbsTime() ) - { - expired_timers[slot] = 1;//this timer expired! - have_expired_timers = 1; - }else if(timers[slot].start_time + snooze_time_sec + 0.02 > GetAbsTime() ){ - timers_expiring_soon = 1;//a timer will expire within the next 20ms - } - } - } - slot++; - } - #ifdef _UNIX_C - pthread_mutex_unlock(&info->tMutex); - #else - ReleaseMutex(info->tMutex); - #endif - - //notify the RTMA Module of any expired timers. By releasing the mutex above- it can notify us of any canceled timers - if(have_expired_timers) - { - #ifdef _UNIX_C - pthread_mutex_lock(&info->tMutex); - #else - DWORD status = WaitForSingleObject( info->tMutex, INFINITE ); - if(status == WAIT_ABANDONED) return 0; //RTMA module terminated while locking the mutex - if(status == WAIT_FAILED){ info->last_error = GetLastError(); return 0;} - #endif - for(int i=0; i < MAX_INTERNAL_TIMERS; i++) - { - if(expired_timers[i] == 1) - { - if(!timers[i].cancel_timer)//check if the module canceled the timer in the meanwhile - module->SelfNotifyExpiredTimer(timers[i].timer_id); - timers[i].cancel_timer = 0; - timers[i].snooze_time = 0; - timers[i].start_time = 0; - timers[i].timer_id = 0; - expired_timers[i] = 0; - } - } - #ifdef _UNIX_C - pthread_mutex_unlock(&info->tMutex); - #else - ReleaseMutex(info->tMutex); - #endif - } - have_expired_timers = 0; - - //try to sleep in 10ms increments until the last 10ms of any timer - if(timers_expiring_soon || !info->keep_running) - Sleep(1); - else - Sleep(10); - } - - return 0; -} - - -int -RTMA_Module::SetTimer(unsigned int time_ms) -//sets a local timer to expire within the stated time in ms. -//Returns: [timer_id] OR [-1: no available timers, -2: failed to spawn thread, -3: failed to create mutex] -{ - TRY { - int slot=0; - - if(Gm_TimerThreadInfo.thread_exists == 0)//m_TimerCount == 1)//we never spawned TimerThread! - { - //reset the global timers array - while(slot < MAX_INTERNAL_TIMERS) - { - Gm_Timers[slot].cancel_timer = 0; - Gm_Timers[slot].snooze_time = 0; - Gm_Timers[slot].timer_id = 0; - slot++; - } - //slot to be used for the first timer - slot=0; - - //initialize mutex & thread info - Gm_TimerThreadInfo.keep_running = 1; - Gm_TimerThreadInfo.last_error = 0; - Gm_TimerThreadInfo.p = this; - Gm_TimerThreadInfo.thread_handle = 0; - Gm_TimerThreadInfo.timers = Gm_Timers; - #ifdef _UNIX_C - pthread_mutex_init(&Gm_TimerThreadInfo.tMutex, NULL); - //Gm_TimerThreadInfo.tMutex = PTHREAD_MUTEX_INITIALIZER; - #else - Gm_TimerThreadInfo.tMutex = CreateMutex(NULL, FALSE, NULL);//mutex with default security attr, not owned, unnamed - if(Gm_TimerThreadInfo.tMutex == NULL) - return -3; - #endif - - //create timer info - Gm_Timers[slot].timer_id = m_TimerCount; - Gm_Timers[slot].snooze_time = time_ms; - Gm_Timers[slot].cancel_timer = 0; - Gm_Timers[slot].start_time = GetAbsTime(); - - //spawn TimerThread - #ifdef _UNIX_C - int status = pthread_create(&Gm_TimerThreadInfo.thread_handle, NULL, TimerThread, (void*) &Gm_TimerThreadInfo); - if(status) - return -2; - #else - Gm_TimerThreadInfo.thread_handle = CreateThread( NULL, 0, TimerThread, (LPVOID) &Gm_TimerThreadInfo, 0, NULL); - if(Gm_TimerThreadInfo.thread_handle == NULL) - return -2; - #endif - Gm_TimerThreadInfo.thread_exists = 1; - - - } - else - {//TimerThread was already spawned before - #ifdef _UNIX_C - pthread_mutex_lock( &Gm_TimerThreadInfo.tMutex); - #else - WaitForSingleObject( Gm_TimerThreadInfo.tMutex, INFINITE ); - #endif - while(slot < MAX_INTERNAL_TIMERS) - { - //find the first available slot - if((Gm_Timers[slot].timer_id == 0) && (Gm_Timers[slot].cancel_timer == 0)) - break; - else - slot++; - } - - if(slot == MAX_INTERNAL_TIMERS) - { - #ifdef _UNIX_C - pthread_mutex_unlock(&Gm_TimerThreadInfo.tMutex); - #else - ReleaseMutex(Gm_TimerThreadInfo.tMutex); - #endif - return -1; - }else{ - //create timer info - Gm_Timers[slot].timer_id = m_TimerCount; - Gm_Timers[slot].snooze_time = time_ms; - Gm_Timers[slot].cancel_timer = 0; - Gm_Timers[slot].start_time = GetAbsTime(); - #ifdef _UNIX_C - pthread_mutex_unlock(&Gm_TimerThreadInfo.tMutex); - #else - ReleaseMutex(Gm_TimerThreadInfo.tMutex); - #endif - } - } - - m_TimerCount++; - return (m_TimerCount - 1); - } CATCH_and_THROW( "RTMA_Module::SetTimer(unsigned int time_ms)"); -} - -int -RTMA_Module::CancelTimer(int timer_id) -//cancels a timer- returns 1 on success, 0 on failure (return value does not gurantee that the timer was actually canceled!) -{ - TRY { - int slot=0; - int timer_canceled=0; - #ifdef _UNIX_C - pthread_mutex_lock( &Gm_TimerThreadInfo.tMutex); - #else - WaitForSingleObject( Gm_TimerThreadInfo.tMutex, INFINITE ); - #endif - while(slot < MAX_INTERNAL_TIMERS) - { - //find the timer - if(Gm_Timers[slot].timer_id == timer_id) - { - if(Gm_Timers[slot].cancel_timer == 1) - timer_canceled = 1; - else - Gm_Timers[slot].cancel_timer = 1; - break; - }else{ - slot++; - } - } - - #ifdef _UNIX_C - pthread_mutex_unlock(&Gm_TimerThreadInfo.tMutex); - #else - ReleaseMutex(Gm_TimerThreadInfo.tMutex); - #endif - - return ((slot == MAX_INTERNAL_TIMERS)|| (timer_canceled == 1) ) ? 0 : 1; - } CATCH_and_THROW( "RTMA_Module::CancelTimer(int timer_id)"); -} - -int -RTMA_Module::SelfNotifyExpiredTimer(int timer_id) -//sends MT_TIMER_EXPIRED to this module itself -{ - TRY { - MDF_TIMER_EXPIRED data; - data.timer_id = timer_id; - - CMessage M(MT_TIMER_EXPIRED, (void*)&data, sizeof(data)); - return SendMessage( &M, m_ModuleID, m_HostID); - } CATCH_and_THROW( "RTMA_Module::SelfNotifyExpiredTimer(int timer_id)"); -} - ////////////////////////////////////////////////////////////////////////////// //////////////////// GLOBAL FUNCTIONS///////////////////////////////////////// ////////////////////////////////////////////////////////////////////////////// From 3650d06bba70d2d696f1bc5c1b526d9d210cca9b Mon Sep 17 00:00:00 2001 From: David Weir Date: Wed, 3 Dec 2025 10:25:08 -0500 Subject: [PATCH 03/17] Remove old core defs not is use --- include/RTMA_types.h | 126 +++++++------------------------------------ 1 file changed, 19 insertions(+), 107 deletions(-) diff --git a/include/RTMA_types.h b/include/RTMA_types.h index 239b1f8..45183f9 100644 --- a/include/RTMA_types.h +++ b/include/RTMA_types.h @@ -16,8 +16,6 @@ typedef int32_t UID; #define MAX_HOSTS 5 //maximal number of hosts in the system #define MAX_MESSAGE_TYPES 10000 //maximal number of message types in the system #define MIN_STREAM_TYPE 9000 //minimal number type for a data stream -#define MAX_TIMERS 100 //maximal number of total timers TimerModule can handle -#define MAX_INTERNAL_TIMERS 20 //maximal number of internal timers that can be set by an RTMA module #define MAX_NAME_LEN 32 // Module name length limit // Maximums for core modules @@ -92,10 +90,6 @@ typedef struct { // Module ID-s of core modules #define MID_MESSAGE_MANAGER 0 -#define MID_COMMAND_MODULE 1 -#define MID_APPLICATION_MODULE 2 -#define MID_NETWORK_RELAY 3 -#define MID_STATUS_MODULE 4 #define MID_QUICKLOGGER 5 #define HID_LOCAL_HOST 0 @@ -111,8 +105,14 @@ typedef char STRING_DATA[]; //message data type for variable length string mes #define MT_EXIT 0 #define MT_KILL 1 #define MT_ACKNOWLEDGE 2 + #define MT_FAIL_SUBSCRIBE 6 -typedef struct { MODULE_ID mod_id; short reserved; MSG_TYPE msg_type;} MDF_FAIL_SUBSCRIBE; +typedef struct { + MODULE_ID mod_id; + short reserved; + MSG_TYPE msg_type; +} MDF_FAIL_SUBSCRIBE; + #define MT_FAILED_MESSAGE 8 // Sent by MM when it cannot forward a message to a module typedef struct { MODULE_ID dest_mod_id; @@ -120,14 +120,14 @@ typedef struct { double time_of_failure; RTMA_MSG_HEADER msg_header; } MDF_FAILED_MESSAGE; -#define MT_MM_ERROR 83 -typedef STRING_DATA MDF_MM_ERROR; -#define MT_MM_INFO 84 -typedef STRING_DATA MDF_MM_INFO; // Messages sent by modules to MessageManager #define MT_CONNECT 13 -typedef struct { short logger_status; short daemon_status;} MDF_CONNECT; +typedef struct { + short logger_status; + short daemon_status; +} MDF_CONNECT; + #define MT_DISCONNECT 14 // Subscription messages @@ -140,115 +140,27 @@ typedef MSG_TYPE MDF_UNSUBSCRIBE; typedef MSG_TYPE MDF_PAUSE_SUBSCRIPTION; typedef MSG_TYPE MDF_RESUME_SUBSCRIPTION; -#define MT_SHUTDOWN_RTMA 17 -#define MT_SHUTDOWN_APP 18 #define MT_FORCE_DISCONNECT 82 -typedef struct { int mod_id; } MDF_FORCE_DISCONNECT; - +typedef struct { + int mod_id; +} MDF_FORCE_DISCONNECT; -//Messages sent by all core modules -#define MT_CORE_MODULE_REINIT_ACK 25 //Messages sent by all modules #define MT_MODULE_READY 26 -typedef struct { int pid; } MDF_MODULE_READY; -#define MT_DYNAMIC_DD_READ_ERR 90 -typedef STRING_DATA MDF_DYNAMIC_DD_READ_ERR; +typedef struct { + int pid; +} MDF_MODULE_READY; + #define MT_DEBUG_TEXT 91 typedef STRING_DATA MDF_DEBUG_TEXT; -// Messages sent to ApplicationModule -#define MT_AM_EXIT 30 -#define MT_START_APP 31 -typedef STRING_DATA MDF_START_APP; -#define MT_STOP_APP 32 -typedef STRING_DATA MDF_STOP_APP; -#define MT_RESTART_APP 33 -typedef STRING_DATA MDF_RESTART_APP; -#define MT_KILL_APP 34 -typedef STRING_DATA MDF_KILL_APP; -#define MT_AM_RE_READ_CONFIG_FILE 89 -#define MT_AM_GET_APP_NAME 92 - //sent by MASTER AM to SLAVE AMs -#define MT_SLAVE_START_APP 64 -typedef STRING_DATA MDF_SLAVE_START_APP; -#define MT_SLAVE_START_APP_ACK 65 -typedef struct {int num_remote_hosts; } MDF_SLAVE_START_APP_ACK; -#define MT_SLAVE_STOP_APP 66 -typedef STRING_DATA MDF_SLAVE_STOP_APP; -#define MT_SLAVE_KILL_APP 67 -typedef STRING_DATA MDF_SLAVE_KILL_APP; -#define MT_SLAVE_RESTART_APP 68 -typedef STRING_DATA MDF_SLAVE_RESTART_APP; - -//Messages sent by ApplicationModule -#define MT_AM_ERROR 35 //sends out if a module started not by application manager -typedef STRING_DATA MDF_AM_ERROR; -#define MT_AM_ACKNOWLEDGE 36 //sent as an immediate reply to MT_START_APP, MT_STOP_APP, MT_KILL_APP -#define MT_FAIL_START_APP 37 //sent if got MT_START_APP and failed to start ANY module -typedef STRING_DATA MDF_FAIL_START_APP; -#define MT_FAIL_STOP_APP 38 //sent if got MT_STOP_APP and failed to stop ANY module -typedef STRING_DATA MDF_FAIL_STOP_APP; -#define MT_FAIL_KILL_APP 39 //sent if got MT_KILL_APP and failed to kill ANY module -typedef STRING_DATA MDF_FAIL_KILL_APP; -#define MT_APP_START_COMPLETE 40 //a final reply to MT_START_APP when done starting application -#define MT_APP_SHUTODWN_COMPLETE 41 //a final reply to MT_STOP_APP when done with application shutdown -#define MT_APP_RESTART_COMPLETE 42 //a final reply to MT_RESTART_APP when done with application restart -#define MT_APP_KILL_COMPLETE 43 //a final reply to MT_KILL_APP when done with killing application -#define MT_ALL_MODULES_READY 44 //sent if got MT_MODULE_READY from all modules -#define MT_CORE_MODULE_REINIT 45 //sent after an application is shutdown/killed - to indicate core modules that they have to reinitialize -#define MT_AM_CONFIG_FILE_DATA 46 //data about what was read from the AM config file -typedef STRING_DATA MDF_AM_CONFIG_FILE_DATA; -#define MT_AM_APP_NAME 93 -typedef char MDF_AM_APP_NAME[]; - -//sent by SLAVE AMs -// #define MT_SLAVE_ALL_MODULES_READY 69 -// #define MT_SLAVE_FAIL_START_APP 70 -// typedef STRING_DATA MDF_SLAVE_FAIL_START_APP; -// #define MT_SLAVE_FAIL_STOP_APP 71 -// #define MT_SLAVE_FAIL_KILL_APP 72 -// #define MT_SLAVE_APP_SHUTODWN_COMPLETE 74 -// #define MT_SLAVE_APP_RESTART_COMPLETE 75 -// #define MT_SLAVE_APP_KILL_COMPLETE 76 -// #define MT_SLAVE_AM_ERROR 77 -// typedef STRING_DATA MDF_SLAVE_AM_ERROR; - - -// Error in Module -#define MT_APP_ERROR 47 -typedef STRING_DATA MDF_APP_ERROR; - -//Messages sent to StatusModule -#define MT_SM_EXIT 48 - -//Messages sent by TimerModule -#define MT_CLOCK_SYNC 49 -#define MT_TIMER_EXPIRED 50 -typedef struct { int timer_id; } MDF_TIMER_EXPIRED; -#define MT_TIMED_OUT 73 -typedef MDF_TIMER_EXPIRED MDF_TIMED_OUT; -#define MT_SET_TIMER_FAILED 51 -typedef struct { MODULE_ID mod_id; int timer_id; int snooze_time; } MDF_SET_TIMER_FAILED; - -//typedef struct { int a; int b; double x; } MDF_TEST_DATA; - -//Messages sent to TimerModule -#define MT_TM_EXIT 52 -#define MT_SET_TIMER 53 -typedef struct { int timer_id; int snooze_time; } MDF_SET_TIMER; - -#define MT_CANCEL_TIMER 54 -typedef struct { int timer_id; } MDF_CANCEL_TIMER; - //Messages sent to LoggerModule #define MT_LM_EXIT 55 // Messages sent by core modules when they have finished initializing and are ready to serve #define MT_MM_READY 94 -//#define MT_SM_READY 95 #define MT_LM_READY 96 -//#define MT_AM_READY 97 // Messages for QuickLogger Module // MT_SAVE_MESSAGE_LOG - Tells QuickLogger to dump its current message buffer From 3fbb66ea89ecdc64e65150a73a446d21ada73092 Mon Sep 17 00:00:00 2001 From: David Weir Date: Wed, 3 Dec 2025 10:25:41 -0500 Subject: [PATCH 04/17] Refactor and cleanup --- src/modules/MessageManager/MessageManager.cpp | 872 +++++++++--------- src/modules/MessageManager/MessageManager.h | 49 +- 2 files changed, 491 insertions(+), 430 deletions(-) diff --git a/src/modules/MessageManager/MessageManager.cpp b/src/modules/MessageManager/MessageManager.cpp index dd57e65..aa79614 100644 --- a/src/modules/MessageManager/MessageManager.cpp +++ b/src/modules/MessageManager/MessageManager.cpp @@ -10,28 +10,28 @@ #include "Debug.h" #ifdef _UNIX_C -int main( int argc, char *argv[]) +int main(int argc, char *argv[]) #else -int main( int argc, char *argv[]) -//int APIENTRY WinMain( HINSTANCE hInstance, HINSTANCE hPrevInstance, LPSTR lpCmdLine, int nShowCmd) +int main(int argc, char *argv[]) +// int APIENTRY WinMain( HINSTANCE hInstance, HINSTANCE hPrevInstance, LPSTR lpCmdLine, int nShowCmd) #endif { try { char *options; char empty_str[] = ""; - #ifdef _UNIX_C +#ifdef _UNIX_C options = (argc > 1) ? argv[1] : empty_str; - #else +#else options = (argc > 1) ? argv[1] : empty_str; - //options = (char*) lpCmdLine; - #endif +// options = (char*) lpCmdLine; +#endif CMessageManager MM; - MM.MainLoop( options); + MM.MainLoop(options); return 0; } - catch(MyCException &E) + catch (MyCException &E) { E.AddToStack("MM WinMain- aborting"); E.ReportToFile(); @@ -39,74 +39,82 @@ int main( int argc, char *argv[]) } } -CMessageManager::CMessageManager( ) +CMessageManager::CMessageManager() { - m_Version = "2.11bci"; - m_NextDynamicModId = DYN_MOD_ID_START; + m_Version = "2.11bci"; + m_NextDynamicModId = DYN_MOD_ID_START; - InitializeAbsTime(); + // Initialize the module record array + // uid=-1 indicates an unused slot + for (int i = 0; i < MAX_MODULES; i++) + { + m_ConnectedModules[i].Reset(); + } + InitializeAbsTime(); } -CMessageManager::~CMessageManager( ) +CMessageManager::~CMessageManager() { } -void -CMessageManager::MainLoop( char *cmd_line_options) +void CMessageManager::MainLoop(char *cmd_line_options) { try { - DEBUG_TEXT( "Entering Main Loop"); + DEBUG_TEXT("Entering Main Loop"); // Elevate process priority #ifdef _WINDOWS_C - BOOL success = SetPriorityClass( GetCurrentProcess(), REALTIME_PRIORITY_CLASS); - //if( success) printf("Yay!\n"); - //else printf("Too bad!\n"); - success = SetThreadPriority( GetCurrentThread(), THREAD_PRIORITY_TIME_CRITICAL); - //if( success) printf("Yay!\n"); - //else printf("Too bad!\n"); + BOOL success = SetPriorityClass(GetCurrentProcess(), REALTIME_PRIORITY_CLASS); + // if( success) printf("Yay!\n"); + // else printf("Too bad!\n"); + success = SetThreadPriority(GetCurrentThread(), THREAD_PRIORITY_TIME_CRITICAL); + // if( success) printf("Yay!\n"); + // else printf("Too bad!\n"); #endif // Start managing messages - if (strlen( cmd_line_options) > 0) { - char *server_name = cmd_line_options; - UPipeAutoServer::Run( server_name); - } else { - UPipeAutoServer::Run( DEFAULT_PIPE_SERVER_NAME_FOR_MM); + if (strlen(cmd_line_options) > 0) + { + char *server_name = cmd_line_options; + UPipeAutoServer::Run(server_name); + } + else + { + UPipeAutoServer::Run(DEFAULT_PIPE_SERVER_NAME_FOR_MM); } - DEBUG_TEXT( "Main Loop Finished"); + DEBUG_TEXT("Main Loop Finished"); } - catch(MyCException &E) + catch (MyCException &E) { E.AddToStack("CMessageManager::MainLoop()"); throw E; } } -void -CMessageManager::HandleData( UPipe *pSourcePipe) +void CMessageManager::HandleData(UPipe *pSourcePipe) { SetMyPriority(THIS_MODULE_BASE_PRIORITY); - DEBUG_TEXT_( "Receiving message from pipe " << pSourcePipe << "... "); + DEBUG_TEXT_("Receiving message from pipe " << pSourcePipe << "... "); CMessage M; - M.Receive( pSourcePipe); - DEBUG_TEXT( "Received message of type " << M.msg_type); + M.Receive(pSourcePipe); + DEBUG_TEXT("Received message of type " << M.msg_type); - ProcessMessage( &M, pSourcePipe); - DistributeMessage( &M); + ProcessMessage(&M, pSourcePipe); + DistributeMessage(&M); SetMyPriority(NORMAL_PRIORITY_CLASS); } -void -CMessageManager::HandleDisconnect( UPipe *pModulePipe) +void CMessageManager::HandleDisconnect(UPipe *pModulePipe) { - DEBUG_TEXT_( "Pipe " << pModulePipe << " broken"); + DEBUG_TEXT_("Pipe " << pModulePipe << " broken"); // Find module ID - for( int uid = 0; uid < MAX_MODULES; uid++) { + for (int uid = 0; uid < MAX_MODULES; uid++) + { CModuleRecord mod = m_ConnectedModules[uid]; - if(mod.pModulePipe == pModulePipe) { + if (mod.pModulePipe == pModulePipe) + { // Make sure not to send anything to disconnected client RemoveSubscription(&mod, ALL_MESSAGE_TYPES); @@ -115,152 +123,80 @@ CMessageManager::HandleDisconnect( UPipe *pModulePipe) // Delete module record CleanUpModuleRecord(&mod); - DEBUG_TEXT_( ", disconnected module " << mod_id); + DEBUG_TEXT_(", disconnected module " << mod_id); break; } } DEBUG_TEXT("!"); } -void -CMessageManager::ProcessMessage( CMessage *M, UPipe *pSourcePipe) -{ - DEBUG_TEXT_( "Processing message... "); - MODULE_ID mod_id = M->src_mod_id; - int prev_priority_class; - - //perform sanity checks before processing the message - bool is_connect_message = (M->msg_type == MT_CONNECT)? 1 : 0; - bool self_message = (mod_id == MID_MESSAGE_MANAGER) ? 1 : 0; - bool is_valid_mod_id = ((mod_id > 0) && (mod_id < MAX_MODULES)) ? 1 : 0; - bool message_from_this_host = (M->src_host_id == HID_LOCAL_HOST) ? 1 : 0; - - switch( M->msg_type) { - case MT_CONNECT: - MDF_CONNECT data; - memset( &data, 0, sizeof(data)); - M->GetData((void*) & data); - prev_priority_class = GetMyPriority(); - SetMyPriority(NORMAL_PRIORITY_CLASS); - mod_id = ConnectModule( mod_id, pSourcePipe, data.logger_status, data.daemon_status); - if (mod_id > 0) { - SetMyPriority(prev_priority_class); - } - break; +void CMessageManager::ProcessMessage(CMessage *M, UPipe *pSourcePipe) +{ + DEBUG_TEXT_("Processing message... "); + int prev_priority_class; + + switch (M->msg_type) + { + case MT_CONNECT: + HandleConnect(M, pSourcePipe); + break; + case MT_CONNECT_V2: - MDF_CONNECT_V2 connect_v2; - M->GetData((void*) &connect_v2); - prev_priority_class = GetMyPriority(); - SetMyPriority(NORMAL_PRIORITY_CLASS); - mod_id = ConnectModuleV2( mod_id, pSourcePipe, &connect_v2); - if (mod_id > 0) { - SendAcknowledge(mod_id); - SetMyPriority(prev_priority_class); - } + HandleConnectV2(M, pSourcePipe); break; + case MT_CLIENT_SET_NAME: - MDF_CLIENT_SET_NAME set_name; - M->GetData((void*) &set_name); - m_ConnectedModules[M->src_mod_id].SetName(set_name.name); + HandleSetName(M); break; + case MT_SUBSCRIPTION_OPTION: - MDF_SUBSCRIPTION_OPTION sub_opt; - M->GetData((void*) &sub_opt); - SubscriptionOption(mod_id, &sub_opt); + HandleSubscriptionControl(M); break; + case MT_PONG: break; + case MT_INTRODUCE: + SendIntroductions(M->src_mod_id); break; - case MT_FORCE_DISCONNECT: - MDF_FORCE_DISCONNECT data_MDF_FORCE_DISCONNECT; - M->GetData( &data_MDF_FORCE_DISCONNECT); - mod_id = data_MDF_FORCE_DISCONNECT.mod_id; - if( (mod_id < 0) || (mod_id > MAX_MODULES) || !ModuleIsConnected(mod_id) ) - { - MyCString err("MM got MT_FORCE_DISCONNECT with invalid module id ["); - err += (int)mod_id; - err += "]"; - m_OutMsg.Set(MT_MM_ERROR, (void*)err.GetContent(), err.GetLen()); - DispatchMessage(&m_OutMsg); - } - else{ - MyCString info("MM forcing disconnect on module ["); - info += (int)mod_id; - info += "]"; - m_OutMsg.Set(MT_MM_INFO, (void*)info.GetContent(), info.GetLen()); - DispatchMessage(&m_OutMsg); - ShutdownModule(mod_id); - } - break; - case MT_DISCONNECT: - prev_priority_class = GetMyPriority(); - SetMyPriority(NORMAL_PRIORITY_CLASS); - DisconnectModule(mod_id); - SetMyPriority(prev_priority_class); - break; - case MT_MODULE_READY: //store pids so that application module can kill processes later - MDF_MODULE_READY m; - MDF_HELLO hello; - M->GetData(&m); - m_ConnectedModules[M->src_mod_id].pid = m.pid; + case MT_FORCE_DISCONNECT: + HandleForceDisconnect(M); + break; - // Notify that the module has joined and ready - SendHello(M->src_mod_id); - break; - case MT_SUBSCRIBE: - MSG_TYPE msg_type_to_subscribe; - M->GetData( &msg_type_to_subscribe); - AddSubscription( mod_id, msg_type_to_subscribe); - DEBUG_TEXT_(" Added subscription to msg type " << msg_type_to_subscribe << " for module " << mod_id << "... "); - SendAcknowledge( mod_id); - break; + case MT_DISCONNECT: + HandleModuleDisconnect(M); + break; - case MT_UNSUBSCRIBE: - MSG_TYPE msg_type_to_unsubscribe; - M->GetData( &msg_type_to_unsubscribe); - RemoveSubscription( mod_id, msg_type_to_unsubscribe); - SendAcknowledge( mod_id); - break; + case MT_MODULE_READY: + HandleModuleReady(M); + break; - case MT_PAUSE_SUBSCRIPTION: - MSG_TYPE msg_type_to_pause; - M->GetData( &msg_type_to_pause); - PauseSubscription( mod_id, msg_type_to_pause); - SendAcknowledge( mod_id); - break; + case MT_SUBSCRIBE: + HandleSubscribe(M); + break; - case MT_RESUME_SUBSCRIPTION: - MSG_TYPE msg_type_to_resume; - M->GetData( &msg_type_to_resume); - ResumeSubscription( mod_id, msg_type_to_resume); - SendAcknowledge( mod_id); - break; + case MT_UNSUBSCRIBE: + HandleUnsubscribe(M); + break; - case MT_SHUTDOWN_RTMA: - prev_priority_class = GetMyPriority(); - SetMyPriority(NORMAL_PRIORITY_CLASS); - DistributeMessage(M);//since we will exit the loop after that- we need to forward this message first - ShutdownAllModules( ); - UPipeAutoServer::_keepRunning = false; - break; + case MT_PAUSE_SUBSCRIPTION: + HandlePauseSubscription(M); + break; - case MT_SHUTDOWN_APP: - prev_priority_class = GetMyPriority(); - SetMyPriority(NORMAL_PRIORITY_CLASS); - ShutdownAllModules(0,0); //shutdown only application modules (not core modules or daemons) - SetMyPriority(prev_priority_class); - break; + case MT_RESUME_SUBSCRIPTION: + HandleResumeSubscription(M); + break; - default: - DEBUG_TEXT( "Nothing to do!"); - return; + default: + DEBUG_TEXT("Nothing to do!"); + return; } - DEBUG_TEXT( "Processed!"); + DEBUG_TEXT("Processed!"); } -void CMessageManager::SendHello(MODULE_ID mod_id) { +void CMessageManager::SendHello(MODULE_ID mod_id) +{ MDF_HELLO hello; CModuleRecord mod = m_ConnectedModules[mod_id]; mod.SetHello(&hello); @@ -268,31 +204,58 @@ void CMessageManager::SendHello(MODULE_ID mod_id) { DispatchMessage(&m_OutMsg); } -void CMessageManager::SendGoodbye(MODULE_ID mod_id) { +void CMessageManager::SendGoodbye(MODULE_ID mod_id) +{ MDF_GOODBYE goodbye; CModuleRecord *mod = GetRecord(mod_id); SendGoodbye(mod); } -void CMessageManager::SendGoodbye(CModuleRecord *mod) { +void CMessageManager::SendGoodbye(CModuleRecord *mod) +{ MDF_GOODBYE goodbye; - if (mod) { + if (mod) + { mod->SetGoodbye(&goodbye); m_OutMsg.Set(MT_GOODBYE, &goodbye, sizeof(goodbye)); DispatchMessage(&m_OutMsg); } } -void CMessageManager::SendPing(MODULE_ID mod_id) { +void CMessageManager::SendPing(MODULE_ID mod_id) +{ MDF_PING ping; CModuleRecord *mod = GetRecord(mod_id); - if (mod) { + if (mod) + { ping.uid = 0; m_OutMsg.Set(MT_PING, &ping, sizeof(ping)); DispatchMessage(&m_OutMsg); } } +void CMessageManager::HandleSetName(CMessage *M) +{ + MDF_CLIENT_SET_NAME set_name; + M->GetData((void *)&set_name); + CModuleRecord *mod = GetRecord(M->src_mod_id); + if (mod) + { + mod->SetName(set_name.name); + } +} + +void CMessageManager::HandleSubscriptionControl(CMessage *M) +{ + MDF_SUBSCRIPTION_OPTION sub_opt; + M->GetData((void *)&sub_opt); + CModuleRecord *mod = GetRecord(M->src_mod_id); + if (mod) + { + SubscriptionOption(mod->uid, &sub_opt); + } +} + /* * Should be called when forwarding a message from other modules * The given message will be forwarded to: @@ -301,12 +264,11 @@ void CMessageManager::SendPing(MODULE_ID mod_id) { * it will be forwarded only there * - if the message has no destination address, it will be forwarded to all subscribed modules */ -void -CMessageManager::DistributeMessage(CMessage *M) +void CMessageManager::DistributeMessage(CMessage *M) { - DEBUG_TEXT( "Distributing Message..."); + DEBUG_TEXT("Distributing Message..."); - CSubscriberList * SL; + CSubscriberList *SL; SL = GetSubscriberList(M->msg_type); if (SL != NULL) { @@ -319,111 +281,117 @@ CMessageManager::DistributeMessage(CMessage *M) CModuleRecord mod = m_ConnectedModules[uid]; int send_it = 0; - int has_specific_dest = (M->dest_mod_id == 0)? 0 : 1; + int has_specific_dest = (M->dest_mod_id == 0) ? 0 : 1; - if(has_specific_dest) + if (has_specific_dest) { - //send only to the specific destination - if(mod.ModuleID == M->dest_mod_id) + // send only to the specific destination + if (mod.ModuleID == M->dest_mod_id) send_it = 1; else send_it = 0; - }else{ - //no specific destination address- so should be forwarded to all subscribers for this MT + } + else + { + // no specific destination address- so should be forwarded to all subscribers for this MT send_it = 1; } - //forward everything to logger modules - if( mod.LoggerStatus) + // forward everything to logger modules + if (mod.LoggerStatus) send_it = 1; - - if( SL->SubscriptionPaused()) + + if (SL->SubscriptionPaused()) send_it = 0; - if( send_it) + if (send_it) { - DEBUG_TEXT_( "Forwarding message to module " << mod_id << "... "); + DEBUG_TEXT_("Forwarding message to module " << mod_id << "... "); try { int status = 0; - if (SL->TimingOnly()) { + if (SL->TimingOnly()) + { MDF_MESSAGE_TIMING msg_time; msg_time.msg_type = M->msg_type; msg_time.src_id = M->src_mod_id; msg_time.send_time = M->send_time; m_OutMsg.Set(MT_MESSAGE_TIMING, &msg_time, sizeof(msg_time)); status = SendMessage(&m_OutMsg, &mod); - } else { + } + else + { status = ForwardMessage(M, &mod); } - if(status == 0) { - LogFailedMessage(M, mod.ModuleID); - DEBUG_TEXT( "Failed to Forward Message!"); - } else { - DEBUG_TEXT( "Forwarded!"); + if (status == 0) + { + LogFailedMessage(M, mod.ModuleID); + DEBUG_TEXT("Failed to Forward Message!"); + } + else + { + DEBUG_TEXT("Forwarded!"); } } - catch( UPipeClosedException &E) { - DEBUG_TEXT( "Failed to Forward Message, destination module socket is closed/dead"); - } + catch (UPipeClosedException &E) + { + DEBUG_TEXT("Failed to Forward Message, destination module socket is closed/dead"); + } } - + uid = SL->GetNextSubscriber(); } } - DEBUG_TEXT( "Done distributing!"); + DEBUG_TEXT("Done distributing!"); } /* * Should be called internally in MM when sending a message to anybody that cares * The message will be sent to all subscribed modules including loggers */ -void -CMessageManager::DispatchMessage(CMessage *M) +void CMessageManager::DispatchMessage(CMessage *M) { - CSubscriberList * SL; - + CSubscriberList *SL; + SL = GetSubscriberList(M->msg_type); - if( SL != NULL) + if (SL != NULL) { // Send message to all subscribers UID uid = SL->GetFirstSubscriber(); - while(uid > 0) + while (uid > 0) { CModuleRecord mod = m_ConnectedModules[uid]; - SendMessage(M, &mod); + SendMessage(M, &mod); uid = SL->GetNextSubscriber(); } } } - /* * Should be called internally in MM when sending a message to a module * The message will be sent, even if the module is not subscribed to it * The message will also be forwarded to all subscribed logger modules */ -void -CMessageManager::DispatchMessage(CMessage *M, CModuleRecord *dest_mod) +void CMessageManager::DispatchMessage(CMessage *M, CModuleRecord *dest_mod) { - CSubscriberList * SL; - - //send the message to the module it is intended to, disregarding subscriptions + CSubscriberList *SL; + + // send the message to the module it is intended to, disregarding subscriptions //(enables MM to send system message to modules) SendMessage(M, dest_mod); - //CC all logger modules + // CC all logger modules SL = GetSubscriberList(M->msg_type); - if( SL != NULL) + if (SL != NULL) { UID uid = SL->GetFirstSubscriber(); - while(uid >= 0) + while (uid >= 0) { CModuleRecord mod = m_ConnectedModules[uid]; - if(mod.LoggerStatus) - if(mod.ModuleID != dest_mod->ModuleID)//don't send to destination module again + if (mod.LoggerStatus) + if (mod.ModuleID != dest_mod->ModuleID) // don't send to destination module again ForwardMessage(M, &mod); uid = SL->GetNextSubscriber(); @@ -436,8 +404,7 @@ CMessageManager::DispatchMessage(CMessage *M, CModuleRecord *dest_mod) * The signal will be sent, even if the module is not subscribed to it * The signal will also be forwarded to all subscribed logger modules */ -void -CMessageManager::DispatchSignal(MSG_TYPE sig, CModuleRecord *dest_mod) +void CMessageManager::DispatchSignal(MSG_TYPE sig, CModuleRecord *dest_mod) { m_OutMsg.Set(sig); DispatchMessage(&m_OutMsg, dest_mod); @@ -448,97 +415,101 @@ CMessageManager::DispatchSignal(MSG_TYPE sig, CModuleRecord *dest_mod) * The message will be sent to all connected modules, even if the module is not subscribed to it * The message will also be forwarded to all subscribed logger modules */ -void -CMessageManager::DispatchMessageToAll(CMessage *M) +void CMessageManager::DispatchMessageToAll(CMessage *M) { - for(int uid=0; uid= 0 && mod.ModuleID > 0){ + if (uid >= 0 && mod.ModuleID > 0) + { DispatchMessage(M, &mod); } } } -void -CMessageManager::DispatchSignalToAll( MSG_TYPE sig) +void CMessageManager::DispatchSignalToAll(MSG_TYPE sig) { CMessage R(sig); DispatchMessageToAll(&R); } - -int -CMessageManager::SendMessage(CMessage *M, CModuleRecord* mod) -//overloaded function for RTMA_Module::SendMessage() -//Sends a message to a module, specifying the MessageManager itself as the source module +int CMessageManager::SendMessage(CMessage *M, CModuleRecord *mod) +// overloaded function for RTMA_Module::SendMessage() +// Sends a message to a module, specifying the MessageManager itself as the source module { UPipe *mod_pipe = mod->pModulePipe; - if(mod_pipe == NULL) return 0; - + if (mod_pipe == NULL) + return 0; + // Assume that msg_type, num_data_bytes, data - have been filled in - M->msg_count = 0; - M->send_time = GetAbsTime(); - M->recv_time = 0.0; + M->msg_count = 0; + M->send_time = GetAbsTime(); + M->recv_time = 0.0; M->src_host_id = HID_LOCAL_HOST; - M->src_mod_id = MID_MESSAGE_MANAGER; + M->src_mod_id = MID_MESSAGE_MANAGER; M->dest_mod_id = mod->ModuleID; double timeout = 0; // By default use non-blocking write so MM does not get stuck on a frozen module - if(mod->LoggerStatus) timeout = -1; // Logger modules should not lose data, so use blocking write + if (mod->LoggerStatus) + timeout = -1; // Logger modules should not lose data, so use blocking write int status = M->Send(mod_pipe, timeout); return status; } -int -CMessageManager::ForwardMessage(CMessage *M, CModuleRecord* dest_mod) -//Forward a message where the header is already filled in -//Source module field in the header is unaltered +int CMessageManager::ForwardMessage(CMessage *M, CModuleRecord *dest_mod) +// Forward a message where the header is already filled in +// Source module field in the header is unaltered { UPipe *mod_pipe = dest_mod->pModulePipe; - if(mod_pipe == NULL) return 0; + if (mod_pipe == NULL) + return 0; double timeout = 0; // By default use non-blocking write so MM does not get stuck on a frozen module - if(dest_mod->LoggerStatus) timeout = -1; // Logger modules should not lose data, so use blocking write + if (dest_mod->LoggerStatus) + timeout = -1; // Logger modules should not lose data, so use blocking write int status = M->Send(mod_pipe, timeout); return status; } -int -CMessageManager::SendSignal( MSG_TYPE sig, CModuleRecord *dest_mod) -//overloaded function for RTMA_Module::SendSignal() -//returns 0 if module is not connected, or failed to send message to it; 1 on success +int CMessageManager::SendSignal(MSG_TYPE sig, CModuleRecord *dest_mod) +// overloaded function for RTMA_Module::SendSignal() +// returns 0 if module is not connected, or failed to send message to it; 1 on success { m_OutMsg.Set(sig); return SendMessage(&m_OutMsg, dest_mod); } -CModuleRecord* -CMessageManager::GetOpenRecord(){ - for (int i=0;i= 0) && !ModuleIsConnected(module_id) ) + if ((module_id < MAX_MODULE_ID) && (module_id >= 0) && !ModuleIsConnected(module_id)) { - if( pSourcePipe != NULL) { - CModuleRecord* mod = GetOpenRecord(); - if (mod == NULL) { + if (pSourcePipe != NULL) + { + CModuleRecord *mod = GetOpenRecord(); + if (mod == NULL) + { printf("MessageManager has reached the MAX_MODULES allowed\n."); return -1; } @@ -547,37 +518,37 @@ CMessageManager::ConnectModule( MODULE_ID module_id, UPipe *pSourcePipe, short l if (module_id == 0) module_id = GetDynamicModuleId(); - if (module_id > 0) { - DEBUG_TEXT( "Connecting module " << module_id << " on pipe " << pSourcePipe); + if (module_id > 0) + { + DEBUG_TEXT("Connecting module " << module_id << " on pipe " << pSourcePipe); // Create a module record - mod->ModuleID = module_id; - mod->pModulePipe = pSourcePipe; + mod->ModuleID = module_id; + mod->pModulePipe = pSourcePipe; mod->LoggerStatus = logger_status; mod->DaemonStatus = daemon_status; - // Add default subscription to TIMER_EXPIRED - if( !logger_status) AddSubscription( mod->uid, MT_TIMER_EXPIRED); - SendAcknowledge(mod); } } } else - module_id = 0; // something went wrong, don't allow the new connection + module_id = 0; // something went wrong, don't allow the new connection return module_id; } MODULE_ID -CMessageManager::ConnectModuleV2( MODULE_ID module_id, UPipe *pSourcePipe, MDF_CONNECT_V2 *data) +CMessageManager::ConnectModuleV2(MODULE_ID module_id, UPipe *pSourcePipe, MDF_CONNECT_V2 *data) { - if( ( module_id < MAX_MODULE_ID) && ( module_id >= 0) && !ModuleIsConnected(module_id) ) + if ((module_id < MAX_MODULE_ID) && (module_id >= 0) && !ModuleIsConnected(module_id)) { - if( pSourcePipe != NULL) { - CModuleRecord* mod = GetOpenRecord(); - if (mod == NULL) { + if (pSourcePipe != NULL) + { + CModuleRecord *mod = GetOpenRecord(); + if (mod == NULL) + { printf("MessageManager has reached the MAX_MODULES allowed\n."); return -1; } @@ -586,20 +557,18 @@ CMessageManager::ConnectModuleV2( MODULE_ID module_id, UPipe *pSourcePipe, MDF_C if (module_id == 0) module_id = GetDynamicModuleId(); - if (module_id > 0) { - DEBUG_TEXT( "Connecting module " << module_id << " on pipe " << pSourcePipe); + if (module_id > 0) + { + DEBUG_TEXT("Connecting module " << module_id << " on pipe " << pSourcePipe); - mod->ModuleID = module_id; - mod->pModulePipe = pSourcePipe; + mod->ModuleID = module_id; + mod->pModulePipe = pSourcePipe; mod->LoggerStatus = data->logger_status; mod->DaemonStatus = data->daemon_status; mod->AllowMultiple = data->allow_multiple; mod->pid = data->pid; mod->SetName(data->name); - // Add default subscription to TIMER_EXPIRED - if( !(data->logger_status)) AddSubscription( mod->uid, MT_TIMER_EXPIRED); - SendAcknowledge(mod); } } @@ -607,26 +576,28 @@ CMessageManager::ConnectModuleV2( MODULE_ID module_id, UPipe *pSourcePipe, MDF_C SendHello(module_id); } else - module_id = 0; // something went wrong, don't allow the new connection + module_id = 0; // something went wrong, don't allow the new connection return module_id; } -CModuleRecord* CMessageManager::GetRecord(MODULE_ID module_id) { - for (int i=0;iReset(); } -void -CMessageManager::ShutdownModule(MODULE_ID mod_id) +void CMessageManager::ShutdownModule(MODULE_ID mod_id) { - CModuleRecord* mod = GetRecord(mod_id); - if(mod != NULL && mod->ModuleID > 0) + CModuleRecord *mod = GetRecord(mod_id); + if (mod != NULL && mod->ModuleID > 0) { MODULE_ID mod_id = mod->ModuleID; - switch(mod_id) + switch (mod_id) { - case MID_COMMAND_MODULE: - //never shutdown Command Module - break; - - case MID_APPLICATION_MODULE: - DispatchSignal(MT_AM_EXIT, mod); + case MID_QUICKLOGGER: + DispatchSignal(MT_LM_EXIT, mod); break; - case MID_STATUS_MODULE: - DispatchSignal(MT_SM_EXIT, mod); - break; - - case MID_QUICKLOGGER: - DispatchSignal(MT_LM_EXIT, mod); - break; - - default: - DispatchSignal(MT_EXIT, mod); + default: + DispatchSignal(MT_EXIT, mod); break; } @@ -686,90 +643,71 @@ CMessageManager::ShutdownModule(MODULE_ID mod_id) } } -void -CMessageManager::ShutdownAllModules(int shutdown_RTMA, int shutdown_daemons) +void CMessageManager::ShutdownAllModules(int shutdown_RTMA, int shutdown_daemons) { int mod_id; int start_mod_id; - if(shutdown_RTMA) { - start_mod_id = MID_COMMAND_MODULE; //shutdown RTMA modules as well - shutdown_daemons = 1; - } else { - start_mod_id = MAX_RTMA_MODULE_ID+1; //shutdown only application modules - } - - for (int uid = 0; uid < MAX_MODULES; uid++) { + for (int uid = 0; uid < MAX_MODULES; uid++) + { CModuleRecord mod = m_ConnectedModules[uid]; - if(mod.DaemonStatus) { - if( shutdown_daemons) ShutdownModule(mod_id); - } else { + if (mod.DaemonStatus) + { + if (shutdown_daemons) + ShutdownModule(mod_id); + } + else + { ShutdownModule(mod_id); } } } -void -CMessageManager::AddSubscription(MODULE_ID mod_id, MSG_TYPE message_type) -{ - MSG_TYPE mt; - - CModuleRecord *mod = GetRecord(mod_id); - AddSubscription(mod, message_type); -} - -void -CMessageManager::AddSubscription(CModuleRecord *mod, MSG_TYPE message_type) +void CMessageManager::AddSubscription(CModuleRecord *mod, MSG_TYPE message_type) { MSG_TYPE mt; - if (mod) { - if( ((message_type < 0) || (message_type > MAX_MESSAGE_TYPES)) && (message_type != ALL_MESSAGE_TYPES) ) + if (mod) + { + if (((message_type < 0) || (message_type > MAX_MESSAGE_TYPES)) && (message_type != ALL_MESSAGE_TYPES)) { - //send MDF_FAIL_SUBSCRIBE instead of ACK so the module's subscribe function will fail + // send MDF_FAIL_SUBSCRIBE instead of ACK so the module's subscribe function will fail MDF_FAIL_SUBSCRIBE data; data.mod_id = mod->ModuleID; data.msg_type = message_type; - CMessage R(MT_FAIL_SUBSCRIBE, (void*)&data, sizeof(data)); + CMessage R(MT_FAIL_SUBSCRIBE, (void *)&data, sizeof(data)); DispatchMessage(&R, mod); return; - } + } - switch( message_type) + switch (message_type) { case ALL_MESSAGE_TYPES: - for( mt = 0; mt < MAX_MESSAGE_TYPES; mt++) { + for (mt = 0; mt < MAX_MESSAGE_TYPES; mt++) + { GetSubscriberList(mt)->AddSubscriber(mod->uid); } break; default: - CSubscriberList* list = GetSubscriberList(message_type); - if(!list->IsSubscribed(mod->uid)) + CSubscriberList *list = GetSubscriberList(message_type); + if (!list->IsSubscribed(mod->uid)) list->AddSubscriber(mod->uid); } } } -void -CMessageManager::RemoveSubscription(MODULE_ID mod_id, MSG_TYPE message_type) +void CMessageManager::RemoveSubscription(CModuleRecord *mod, MSG_TYPE message_type) { // Might wanna add checks here for valid module_id, connected module MSG_TYPE mt; - - CModuleRecord *mod = GetRecord(mod_id); - RemoveSubscription(mod, message_type); -} -void -CMessageManager::RemoveSubscription(CModuleRecord *mod, MSG_TYPE message_type) -{ - // Might wanna add checks here for valid module_id, connected module - MSG_TYPE mt; - - if (mod) { - switch( message_type) { + if (mod) + { + switch (message_type) + { case ALL_MESSAGE_TYPES: - for( mt = 0; mt < MAX_MESSAGE_TYPES; mt++) { + for (mt = 0; mt < MAX_MESSAGE_TYPES; mt++) + { GetSubscriberList(mt)->RemoveSubscriber(mod->uid); } break; @@ -779,56 +717,161 @@ CMessageManager::RemoveSubscription(CModuleRecord *mod, MSG_TYPE message_type) } } -void -CMessageManager::PauseSubscription(MODULE_ID mod_id, MSG_TYPE message_type) +void CMessageManager::PauseSubscription(CModuleRecord *mod, MSG_TYPE message_type) { // Might wanna add checks here for valid module_id, connected module MSG_TYPE mt; - - CModuleRecord *mod = GetRecord(mod_id); - if (mod) { - switch( message_type) { - case ALL_MESSAGE_TYPES: - for( mt = 0; mt < MAX_MESSAGE_TYPES; mt++) { - GetSubscriberList(mt)->PauseSubscriber(mod->uid); - } - break; - default: - GetSubscriberList(message_type)->PauseSubscriber(mod->uid); + + switch (message_type) + { + case ALL_MESSAGE_TYPES: + for (mt = 0; mt < MAX_MESSAGE_TYPES; mt++) + { + GetSubscriberList(mt)->PauseSubscriber(mod->uid); } + break; + default: + GetSubscriberList(message_type)->PauseSubscriber(mod->uid); } } -void -CMessageManager::ResumeSubscription(MODULE_ID mod_id, MSG_TYPE message_type) +void CMessageManager::ResumeSubscription(CModuleRecord *mod, MSG_TYPE message_type) { // Might wanna add checks here for valid module_id, connected module MSG_TYPE mt; - - CModuleRecord *mod = GetRecord(mod_id); - if (mod) { - switch( message_type) { - case ALL_MESSAGE_TYPES: - for( mt = 0; mt < MAX_MESSAGE_TYPES; mt++) { - GetSubscriberList(mt)->ResumeSubscriber(mod->uid); - } - break; - default: - GetSubscriberList(message_type)->ResumeSubscriber(mod->uid); + + switch (message_type) + { + case ALL_MESSAGE_TYPES: + for (mt = 0; mt < MAX_MESSAGE_TYPES; mt++) + { + GetSubscriberList(mt)->ResumeSubscriber(mod->uid); } + break; + default: + GetSubscriberList(message_type)->ResumeSubscriber(mod->uid); } } -void -CMessageManager::SubscriptionOption(MODULE_ID mod_id, MDF_SUBSCRIPTION_OPTION *sub_opt) +void CMessageManager::SubscriptionOption(UID uid, MDF_SUBSCRIPTION_OPTION *sub_opt) { - CModuleRecord *mod = GetRecord(mod_id); - if (mod) { - if (sub_opt->value) { - GetSubscriberList(sub_opt->msg_type)->SetSubscriberOption(mod->uid, sub_opt->option); - } else { - GetSubscriberList(sub_opt->msg_type)->ClearSubscriberOption(mod->uid, sub_opt->option); - } + if (sub_opt->value) + { + GetSubscriberList(sub_opt->msg_type)->SetSubscriberOption(uid, sub_opt->option); + } + else + { + GetSubscriberList(sub_opt->msg_type)->ClearSubscriberOption(uid, sub_opt->option); + } +} + +void CMessageManager::HandleConnect(CMessage *M, UPipe *pSourcePipe) +{ + MDF_CONNECT data; + memset(&data, 0, sizeof(data)); + M->GetData((void *)&data); + int prev_priority_class = GetMyPriority(); + SetMyPriority(NORMAL_PRIORITY_CLASS); + MODULE_ID mod_id = ConnectModule(mod_id, pSourcePipe, data.logger_status, data.daemon_status); + if (mod_id > 0) + { + SetMyPriority(prev_priority_class); + } +} + +void CMessageManager::HandleConnectV2(CMessage *M, UPipe *pSourcePipe) +{ + MDF_CONNECT_V2 connect_v2; + M->GetData((void *)&connect_v2); + int prev_priority_class = GetMyPriority(); + SetMyPriority(NORMAL_PRIORITY_CLASS); + MODULE_ID mod_id = ConnectModuleV2(mod_id, pSourcePipe, &connect_v2); + if (mod_id > 0) + { + SetMyPriority(prev_priority_class); + } +} + +void CMessageManager::HandleForceDisconnect(CMessage *M) +{ + MDF_FORCE_DISCONNECT data_MDF_FORCE_DISCONNECT; + M->GetData(&data_MDF_FORCE_DISCONNECT); + MODULE_ID mod_id = data_MDF_FORCE_DISCONNECT.mod_id; + ShutdownModule(mod_id); +} + +void CMessageManager::HandleModuleDisconnect(CMessage *M) +{ + int prev_priority_class = GetMyPriority(); + SetMyPriority(NORMAL_PRIORITY_CLASS); + DisconnectModule(M->src_mod_id); + SetMyPriority(prev_priority_class); +} + +void CMessageManager::HandleModuleReady(CMessage *M) +{ + MDF_MODULE_READY module_ready; + MDF_HELLO hello; + M->GetData(&module_ready); + + CModuleRecord *mod = GetRecord(M->src_mod_id); + if (mod) + { + mod->pid = module_ready.pid; + + // Notify that the module has joined and ready + SendHello(M->src_mod_id); + } +} + +void CMessageManager::HandleSubscribe(CMessage *M) +{ + MSG_TYPE msg_type_to_subscribe; + M->GetData(&msg_type_to_subscribe); + + CModuleRecord *mod = GetRecord(M->src_mod_id); + if (mod) + { + AddSubscription(mod, msg_type_to_subscribe); + DEBUG_TEXT_(" Added subscription to msg type " << msg_type_to_subscribe << " for module " << M->src_mod_id << "... "); + SendAcknowledge(mod); + } +} + +void CMessageManager::HandleUnsubscribe(CMessage *M) +{ + MSG_TYPE msg_type_to_unsubscribe; + M->GetData(&msg_type_to_unsubscribe); + CModuleRecord *mod = GetRecord(M->src_mod_id); + if (mod) + { + RemoveSubscription(mod, msg_type_to_unsubscribe); + SendAcknowledge(mod); + } +} + +void CMessageManager::HandlePauseSubscription(CMessage *M) +{ + MSG_TYPE msg_type_to_pause; + M->GetData(&msg_type_to_pause); + CModuleRecord *mod = GetRecord(M->src_mod_id); + if (mod) + { + PauseSubscription(mod, msg_type_to_pause); + SendAcknowledge(mod); + } +} + +void CMessageManager::HandleResumeSubscription(CMessage *M) +{ + + MSG_TYPE msg_type_to_resume; + M->GetData(&msg_type_to_resume); + CModuleRecord *mod = GetRecord(M->src_mod_id); + if (mod) + { + ResumeSubscription(mod, msg_type_to_resume); + SendAcknowledge(mod); } } @@ -841,41 +884,35 @@ CMessageManager::GetSubscriberList(MSG_TYPE message_type) return &(m_SubscribersToMessageType[message_type]); } -bool -CMessageManager::IsModuleSubscribed(UID uid, MSG_TYPE message_type) +bool CMessageManager::IsModuleSubscribed(UID uid, MSG_TYPE message_type) { if (message_type >= MAX_MESSAGE_TYPES || message_type < 0) return false; - - return GetSubscriberList(message_type)->IsSubscribed(uid); -} -void -CMessageManager::SendAcknowledge(MODULE_ID mod_id) -{ - CModuleRecord *mod = GetRecord(mod_id); - SendAcknowledge(mod); + return GetSubscriberList(message_type)->IsSubscribed(uid); } -void -CMessageManager::SendAcknowledge(CModuleRecord * mod) +void CMessageManager::SendAcknowledge(CModuleRecord *mod) { - if (mod) { - DEBUG_TEXT_( "Sending ACK to module " << mod_id << "... "); - DispatchSignal( MT_ACKNOWLEDGE, mod); - DEBUG_TEXT( "Sent!"); + if (mod) + { + DEBUG_TEXT_("Sending ACK to module " << mod_id << "... "); + DispatchSignal(MT_ACKNOWLEDGE, mod); + DEBUG_TEXT("Sent!"); } } -void -CMessageManager::SendIntroductions(MODULE_ID mod_id) +void CMessageManager::SendIntroductions(MODULE_ID mod_id) { MDF_HELLO hello; CModuleRecord *dest_mod = GetRecord(mod_id); - if (dest_mod) { - for (int i=0; i< MAX_MODULES; i++){ + if (dest_mod) + { + for (int i = 0; i < MAX_MODULES; i++) + { CModuleRecord mod = m_ConnectedModules[i]; - if (mod.uid >= 0 && mod.ModuleID > 0){ + if (mod.uid >= 0 && mod.ModuleID > 0) + { mod.SetHello(&hello); m_OutMsg.Set(MT_HELLO, &hello, sizeof(hello)); SendMessage(&m_OutMsg, dest_mod); @@ -884,29 +921,28 @@ CMessageManager::SendIntroductions(MODULE_ID mod_id) } } - -int -CMessageManager::ModuleIsConnected(MODULE_ID mod_id) +int CMessageManager::ModuleIsConnected(MODULE_ID mod_id) { - if( mod_id < MID_MESSAGE_MANAGER) + if (mod_id < MID_MESSAGE_MANAGER) return 0; - for (int i=0; i Date: Wed, 3 Dec 2025 10:28:20 -0500 Subject: [PATCH 05/17] Only set timing data once --- src/modules/MessageManager/MessageManager.cpp | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/src/modules/MessageManager/MessageManager.cpp b/src/modules/MessageManager/MessageManager.cpp index aa79614..ebc7080 100644 --- a/src/modules/MessageManager/MessageManager.cpp +++ b/src/modules/MessageManager/MessageManager.cpp @@ -267,6 +267,7 @@ void CMessageManager::HandleSubscriptionControl(CMessage *M) void CMessageManager::DistributeMessage(CMessage *M) { DEBUG_TEXT("Distributing Message..."); + int timing_set = 0; CSubscriberList *SL; SL = GetSubscriberList(M->msg_type); @@ -313,10 +314,13 @@ void CMessageManager::DistributeMessage(CMessage *M) if (SL->TimingOnly()) { MDF_MESSAGE_TIMING msg_time; - msg_time.msg_type = M->msg_type; - msg_time.src_id = M->src_mod_id; - msg_time.send_time = M->send_time; - m_OutMsg.Set(MT_MESSAGE_TIMING, &msg_time, sizeof(msg_time)); + if (!timing_set) { + timing_set = 1; + msg_time.msg_type = M->msg_type; + msg_time.src_id = M->src_mod_id; + msg_time.send_time = M->send_time; + m_OutMsg.Set(MT_MESSAGE_TIMING, &msg_time, sizeof(msg_time)); + } status = SendMessage(&m_OutMsg, &mod); } else From ff9d7524c5effb50b13519e2b218b380d9e15afa Mon Sep 17 00:00:00 2001 From: David Weir Date: Wed, 3 Dec 2025 10:37:37 -0500 Subject: [PATCH 06/17] Heap allocate MM instance --- src/modules/MessageManager/MessageManager.cpp | 28 +++++++++---------- 1 file changed, 14 insertions(+), 14 deletions(-) diff --git a/src/modules/MessageManager/MessageManager.cpp b/src/modules/MessageManager/MessageManager.cpp index ebc7080..0a566e5 100644 --- a/src/modules/MessageManager/MessageManager.cpp +++ b/src/modules/MessageManager/MessageManager.cpp @@ -27,8 +27,8 @@ int main(int argc, char *argv[]) // options = (char*) lpCmdLine; #endif - CMessageManager MM; - MM.MainLoop(options); + CMessageManager* MM = new CMessageManager(); + MM->MainLoop(options); return 0; } catch (MyCException &E) @@ -649,20 +649,20 @@ void CMessageManager::ShutdownModule(MODULE_ID mod_id) void CMessageManager::ShutdownAllModules(int shutdown_RTMA, int shutdown_daemons) { - int mod_id; - int start_mod_id; - for (int uid = 0; uid < MAX_MODULES; uid++) { CModuleRecord mod = m_ConnectedModules[uid]; - if (mod.DaemonStatus) + if (mod.uid >= 0 && mod.ModuleID > 0) { - if (shutdown_daemons) - ShutdownModule(mod_id); - } - else - { - ShutdownModule(mod_id); + if (mod.DaemonStatus) + { + if (shutdown_daemons) + ShutdownModule(mod.ModuleID); + } + else + { + ShutdownModule(mod.ModuleID); + } } } } @@ -776,7 +776,7 @@ void CMessageManager::HandleConnect(CMessage *M, UPipe *pSourcePipe) M->GetData((void *)&data); int prev_priority_class = GetMyPriority(); SetMyPriority(NORMAL_PRIORITY_CLASS); - MODULE_ID mod_id = ConnectModule(mod_id, pSourcePipe, data.logger_status, data.daemon_status); + MODULE_ID mod_id = ConnectModule(M->src_mod_id, pSourcePipe, data.logger_status, data.daemon_status); if (mod_id > 0) { SetMyPriority(prev_priority_class); @@ -789,7 +789,7 @@ void CMessageManager::HandleConnectV2(CMessage *M, UPipe *pSourcePipe) M->GetData((void *)&connect_v2); int prev_priority_class = GetMyPriority(); SetMyPriority(NORMAL_PRIORITY_CLASS); - MODULE_ID mod_id = ConnectModuleV2(mod_id, pSourcePipe, &connect_v2); + MODULE_ID mod_id = ConnectModuleV2(M->src_mod_id, pSourcePipe, &connect_v2); if (mod_id > 0) { SetMyPriority(prev_priority_class); From 4bd6425387c2a7418fabdc9d1eb7bf0cb2033e96 Mon Sep 17 00:00:00 2001 From: David Weir Date: Wed, 3 Dec 2025 10:37:48 -0500 Subject: [PATCH 07/17] Guard against malloc null --- src/modules/MessageManager/MessageManager.h | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/src/modules/MessageManager/MessageManager.h b/src/modules/MessageManager/MessageManager.h index 0fb63c4..906b983 100644 --- a/src/modules/MessageManager/MessageManager.h +++ b/src/modules/MessageManager/MessageManager.h @@ -35,8 +35,10 @@ class CModuleRecord free(name); } name = (char*) malloc(MAX_NAME_LEN); - memcpy(name, src_name, MAX_NAME_LEN); - name[MAX_NAME_LEN - 1] = '\0'; + if (name != NULL) { + memcpy(name, src_name, MAX_NAME_LEN); + name[MAX_NAME_LEN - 1] = '\0'; + } } void SetHello(MDF_HELLO *hello) { From 0074f3068f9367a52807a5270e7483a297a2b477 Mon Sep 17 00:00:00 2001 From: David Weir Date: Wed, 3 Dec 2025 10:41:25 -0500 Subject: [PATCH 08/17] Remove more timer code --- include/RTMA.h | 12 ------------ src/core/RTMA.cpp | 20 ++------------------ 2 files changed, 2 insertions(+), 30 deletions(-) diff --git a/include/RTMA.h b/include/RTMA.h index 8455970..7a620e7 100644 --- a/include/RTMA.h +++ b/include/RTMA.h @@ -119,7 +119,6 @@ class RTMA_Module HOST_ID m_HostID; double m_StartTime; int m_Pid; - int m_TimerCount; #if (RTMA_PROFILE == TRUE) int m_NumProfiledMsgs; @@ -252,17 +251,6 @@ class RTMA_Module //if MsgType is specified- will not return until the requested msg type was received (and will discard all other messages received) //if MsgType is not specified- will return the first message received (in this case just a wrapper for ReadMessage) - int - SetTimer(unsigned int time_ms); - //sets a local timer to expire within the time stated (in ms). Returns timer_id or -1 on failure - - int - CancelTimer(int timer_id); - - int - SelfNotifyExpiredTimer(int timer_id); - //sends MT_TIMER_EXPIRED to m_WrtInputPipe (self input pipe). Returns 0 on failure, 1 on success - double UpTime( void); int GetPid( void); diff --git a/src/core/RTMA.cpp b/src/core/RTMA.cpp index 1863884..395d3c2 100644 --- a/src/core/RTMA.cpp +++ b/src/core/RTMA.cpp @@ -323,9 +323,7 @@ void RTMA_Module::InitVariables( MODULE_ID ModuleID, HOST_ID HostID) #else m_Pid = _getpid(); #endif - m_TimerCount=1; - Gm_TimerThreadInfo.thread_exists = 0; - + InitializeAbsTime(); } CATCH_and_THROW( "void RTMA_Module::InitVariables( MODULE_ID ModuleID, HOST_ID HostID)"); @@ -341,21 +339,7 @@ RTMA_Module::~RTMA_Module( ) void RTMA_Module::Cleanup( void) { - TRY { - if(Gm_TimerThreadInfo.thread_exists == 1) // m_TimerCount > 1) - { - #ifdef _UNIX_C - Gm_TimerThreadInfo.keep_running = 0; - pthread_join(Gm_TimerThreadInfo.thread_handle, NULL); - pthread_mutex_destroy(&Gm_TimerThreadInfo.tMutex); - #else - TerminateThread(Gm_TimerThreadInfo.thread_handle, 0); - CloseHandle(Gm_TimerThreadInfo.thread_handle); - //The system closes the mutex handle automatically when the process terminates - #endif - Gm_TimerThreadInfo.thread_exists = 0; - } - + TRY { if( m_Connected) { DisconnectFromMMM( ); } From 6b6a986bf439f6b8771b4359c38d6eb8bf029864 Mon Sep 17 00:00:00 2001 From: David Weir Date: Wed, 3 Dec 2025 10:52:30 -0500 Subject: [PATCH 09/17] Return -1 always for timer functions --- lang/matlab/MatlabRTMA.cpp | 14 ++------------ 1 file changed, 2 insertions(+), 12 deletions(-) diff --git a/lang/matlab/MatlabRTMA.cpp b/lang/matlab/MatlabRTMA.cpp index 19f0f8a..9b093f2 100644 --- a/lang/matlab/MatlabRTMA.cpp +++ b/lang/matlab/MatlabRTMA.cpp @@ -253,23 +253,13 @@ mexFunction( case SET_TIMER: if( num_input_args < 2) Error( "incorrect number of arguments"); - if( TheModule.IsConnected( )) { - SnoozeTime = (unsigned int) mxGetScalar( input_arg[1]); - TimerID = TheModule.SetTimer( SnoozeTime); - } else { - TimerID = -1; - } + TimerID = -1; output_arg[0] = mxCreateDoubleScalar( (double) TimerID); break; case CANCEL_TIMER: if( num_input_args < 2) Error( "incorrect number of arguments"); - if( TheModule.IsConnected( )) { - TimerID = (unsigned int) mxGetScalar( input_arg[1]); - status = TheModule.CancelTimer( TimerID); - } else { - status = -1; - } + status = -1; output_arg[0] = mxCreateDoubleScalar( (double) status); break; From 0718b656780da08aa6ce9750188d5570c20f23b7 Mon Sep 17 00:00:00 2001 From: David Weir Date: Wed, 3 Dec 2025 11:16:12 -0500 Subject: [PATCH 10/17] Update some core MT values --- include/RTMA_types.h | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/include/RTMA_types.h b/include/RTMA_types.h index 45183f9..e973b14 100644 --- a/include/RTMA_types.h +++ b/include/RTMA_types.h @@ -197,7 +197,7 @@ typedef struct { char name[MAX_NAME_LEN]; } MDF_CONNECT_V2; -#define MT_SUBSCRIPTION_OPTION 70 +#define MT_SUBSCRIPTION_OPTION 7 typedef struct { MSG_TYPE msg_type; int32_t option; @@ -243,10 +243,11 @@ typedef struct { char name[MAX_NAME_LEN]; } MDF_CLIENT_SET_NAME; -#define MT_MESSAGE_TIMING 69 +#define MT_MESSAGE_TIMING 9 typedef struct { MSG_TYPE msg_type; MODULE_ID src_id; + int16_t reserved; double send_time; } MDF_MESSAGE_TIMING; From 3687a4845fd3fd8b2d1a28932129438819027d10 Mon Sep 17 00:00:00 2001 From: David Weir Date: Thu, 4 Dec 2025 13:05:03 -0500 Subject: [PATCH 11/17] Remove ip address --- include/RTMA_types.h | 6 ++---- include/internal/UPipe.h | 2 -- src/core/PipeLib/SocketPipe.cpp | 14 -------------- src/core/PipeLib/SocketPipe.h | 1 - 4 files changed, 2 insertions(+), 21 deletions(-) diff --git a/include/RTMA_types.h b/include/RTMA_types.h index e973b14..34a374a 100644 --- a/include/RTMA_types.h +++ b/include/RTMA_types.h @@ -223,8 +223,7 @@ typedef struct { int32_t uid; int32_t pid; MODULE_ID mod_id; - uint16_t port; - char addr[32]; + int16_t reserved; char name[MAX_NAME_LEN]; } MDF_HELLO; @@ -233,8 +232,7 @@ typedef struct { int32_t uid; int32_t pid; MODULE_ID mod_id; - uint16_t port; - char addr[32]; + int16_t reserved; char name[MAX_NAME_LEN]; } MDF_GOODBYE; diff --git a/include/internal/UPipe.h b/include/internal/UPipe.h index c70ce22..e7ba13f 100644 --- a/include/internal/UPipe.h +++ b/include/internal/UPipe.h @@ -151,8 +151,6 @@ class UPipe // bytes written. virtual int Write( void *data_buffer, int n_bytes, double timeout) = 0; - void GetIpAddress(char *addr, uint16_t *port, int bufsz) {}; - }; diff --git a/src/core/PipeLib/SocketPipe.cpp b/src/core/PipeLib/SocketPipe.cpp index 8d42cde..8bd4dba 100644 --- a/src/core/PipeLib/SocketPipe.cpp +++ b/src/core/PipeLib/SocketPipe.cpp @@ -337,20 +337,6 @@ SocketPipe::GetCapacity( void) } CATCH_and_THROW( "SocketPipe::GetCapacity"); } -void -SocketPipe::GetIpAddress(char* addr, uint16_t *port, int bufsz) { - // windows: https://learn.microsoft.com/en-us/windows/win32/api/winsock2/nf-winsock2-getsockname - // linux: https://man7.org/linux/man-pages/man2/getsockname.2.html - struct sockaddr_in name = { 0 }; - int namelen = sizeof(name); - if (getsockname(_hPipe.id, (struct sockaddr*)&name, &namelen) == SOCKET_ERROR) { - return; - } - - *port = ntohs(name.sin_port); - inet_ntop(AF_INET, &(name.sin_addr), &(addr[0]), bufsz); -} - int SocketPipe::Read( void *data_buffer, int nbytes_to_read, double timeout) { diff --git a/src/core/PipeLib/SocketPipe.h b/src/core/PipeLib/SocketPipe.h index 213409c..ffbe62f 100644 --- a/src/core/PipeLib/SocketPipe.h +++ b/src/core/PipeLib/SocketPipe.h @@ -72,7 +72,6 @@ class SocketPipe : public UPipe int GetCapacity( void); int Read( void *data_buffer, int n_bytes, double timeout); int Write( void *data_buffer, int n_bytes, double timeout); - void GetIpAddress(char *addr, uint16_t *port, int bufsz); private: SocketHandle _hPipe; // Handle to underlying OS socket From dfb6cc3977585a585778f2f4d5018b32fa153dea Mon Sep 17 00:00:00 2001 From: David Weir Date: Thu, 4 Dec 2025 13:05:45 -0500 Subject: [PATCH 12/17] Use potinters to access module records --- src/modules/MessageManager/MessageManager.cpp | 146 +++++++++--------- 1 file changed, 72 insertions(+), 74 deletions(-) diff --git a/src/modules/MessageManager/MessageManager.cpp b/src/modules/MessageManager/MessageManager.cpp index 0a566e5..c999aac 100644 --- a/src/modules/MessageManager/MessageManager.cpp +++ b/src/modules/MessageManager/MessageManager.cpp @@ -13,7 +13,6 @@ int main(int argc, char *argv[]) #else int main(int argc, char *argv[]) -// int APIENTRY WinMain( HINSTANCE hInstance, HINSTANCE hPrevInstance, LPSTR lpCmdLine, int nShowCmd) #endif { try @@ -112,17 +111,17 @@ void CMessageManager::HandleDisconnect(UPipe *pModulePipe) // Find module ID for (int uid = 0; uid < MAX_MODULES; uid++) { - CModuleRecord mod = m_ConnectedModules[uid]; - if (mod.pModulePipe == pModulePipe) + CModuleRecord *mod = &m_ConnectedModules[uid]; + if (mod->pModulePipe == pModulePipe) { // Make sure not to send anything to disconnected client - RemoveSubscription(&mod, ALL_MESSAGE_TYPES); + RemoveSubscription(mod, ALL_MESSAGE_TYPES); // Notify that the module has left - SendGoodbye(&mod); + SendGoodbye(mod); // Delete module record - CleanUpModuleRecord(&mod); + CleanUpModuleRecord(mod); DEBUG_TEXT_(", disconnected module " << mod_id); break; } @@ -133,7 +132,6 @@ void CMessageManager::HandleDisconnect(UPipe *pModulePipe) void CMessageManager::ProcessMessage(CMessage *M, UPipe *pSourcePipe) { DEBUG_TEXT_("Processing message... "); - int prev_priority_class; switch (M->msg_type) { @@ -195,28 +193,23 @@ void CMessageManager::ProcessMessage(CMessage *M, UPipe *pSourcePipe) DEBUG_TEXT("Processed!"); } -void CMessageManager::SendHello(MODULE_ID mod_id) +void CMessageManager::SendHello(CModuleRecord *mod) { MDF_HELLO hello; - CModuleRecord mod = m_ConnectedModules[mod_id]; - mod.SetHello(&hello); - m_OutMsg.Set(MT_HELLO, &hello, sizeof(hello)); - DispatchMessage(&m_OutMsg); -} - -void CMessageManager::SendGoodbye(MODULE_ID mod_id) -{ - MDF_GOODBYE goodbye; - CModuleRecord *mod = GetRecord(mod_id); - SendGoodbye(mod); + if ((mod != NULL) && (mod->uid >= 0)){ + mod->SetHello(&hello); + printf("Hello: %d\n", hello.mod_id); + m_OutMsg.Set(MT_HELLO, &hello, sizeof(hello)); + DispatchMessage(&m_OutMsg); + } } void CMessageManager::SendGoodbye(CModuleRecord *mod) { MDF_GOODBYE goodbye; - if (mod) - { + if ((mod != NULL) && (mod->uid >= 0)) { mod->SetGoodbye(&goodbye); + printf("Goodbye: %d\n", goodbye.mod_id); m_OutMsg.Set(MT_GOODBYE, &goodbye, sizeof(goodbye)); DispatchMessage(&m_OutMsg); } @@ -226,7 +219,7 @@ void CMessageManager::SendPing(MODULE_ID mod_id) { MDF_PING ping; CModuleRecord *mod = GetRecord(mod_id); - if (mod) + if ((mod != NULL) && (mod->uid >= 0)) { ping.uid = 0; m_OutMsg.Set(MT_PING, &ping, sizeof(ping)); @@ -239,9 +232,9 @@ void CMessageManager::HandleSetName(CMessage *M) MDF_CLIENT_SET_NAME set_name; M->GetData((void *)&set_name); CModuleRecord *mod = GetRecord(M->src_mod_id); - if (mod) + if ((mod != NULL) && (mod->uid >= 0)) { - mod->SetName(set_name.name); + mod->SetName(&(set_name.name[0])); } } @@ -250,7 +243,7 @@ void CMessageManager::HandleSubscriptionControl(CMessage *M) MDF_SUBSCRIPTION_OPTION sub_opt; M->GetData((void *)&sub_opt); CModuleRecord *mod = GetRecord(M->src_mod_id); - if (mod) + if ((mod != NULL) && (mod->uid >= 0)) { SubscriptionOption(mod->uid, &sub_opt); } @@ -279,7 +272,7 @@ void CMessageManager::DistributeMessage(CMessage *M) /* the order of the code in this while loop is important don't modify it unless you know what you're doing */ - CModuleRecord mod = m_ConnectedModules[uid]; + CModuleRecord *mod = &m_ConnectedModules[uid]; int send_it = 0; int has_specific_dest = (M->dest_mod_id == 0) ? 0 : 1; @@ -287,7 +280,7 @@ void CMessageManager::DistributeMessage(CMessage *M) if (has_specific_dest) { // send only to the specific destination - if (mod.ModuleID == M->dest_mod_id) + if (mod->ModuleID == M->dest_mod_id) send_it = 1; else send_it = 0; @@ -299,7 +292,7 @@ void CMessageManager::DistributeMessage(CMessage *M) } // forward everything to logger modules - if (mod.LoggerStatus) + if (mod->LoggerStatus) send_it = 1; if (SL->SubscriptionPaused()) @@ -321,16 +314,16 @@ void CMessageManager::DistributeMessage(CMessage *M) msg_time.send_time = M->send_time; m_OutMsg.Set(MT_MESSAGE_TIMING, &msg_time, sizeof(msg_time)); } - status = SendMessage(&m_OutMsg, &mod); + status = SendMessage(&m_OutMsg, mod); } else { - status = ForwardMessage(M, &mod); + status = ForwardMessage(M, mod); } if (status == 0) { - LogFailedMessage(M, mod.ModuleID); + LogFailedMessage(M, mod->ModuleID); DEBUG_TEXT("Failed to Forward Message!"); } else @@ -365,8 +358,8 @@ void CMessageManager::DispatchMessage(CMessage *M) UID uid = SL->GetFirstSubscriber(); while (uid > 0) { - CModuleRecord mod = m_ConnectedModules[uid]; - SendMessage(M, &mod); + CModuleRecord *mod = &m_ConnectedModules[uid]; + SendMessage(M, mod); uid = SL->GetNextSubscriber(); } } @@ -393,10 +386,10 @@ void CMessageManager::DispatchMessage(CMessage *M, CModuleRecord *dest_mod) while (uid >= 0) { - CModuleRecord mod = m_ConnectedModules[uid]; - if (mod.LoggerStatus) - if (mod.ModuleID != dest_mod->ModuleID) // don't send to destination module again - ForwardMessage(M, &mod); + CModuleRecord *mod = &m_ConnectedModules[uid]; + if (mod->LoggerStatus) + if (mod->ModuleID != dest_mod->ModuleID) // don't send to destination module again + ForwardMessage(M, mod); uid = SL->GetNextSubscriber(); } @@ -423,10 +416,10 @@ void CMessageManager::DispatchMessageToAll(CMessage *M) { for (int uid = 0; uid < MAX_MODULES; uid++) { - CModuleRecord mod = m_ConnectedModules[uid]; - if (uid >= 0 && mod.ModuleID > 0) + CModuleRecord *mod = &m_ConnectedModules[uid]; + if (uid >= 0 && mod->ModuleID > 0) { - DispatchMessage(M, &mod); + DispatchMessage(M, mod); } } } @@ -492,12 +485,13 @@ CMessageManager::GetOpenRecord() { for (int i = 0; i < MAX_MODULES; i++) { - CModuleRecord mod = m_ConnectedModules[i]; - if (mod.uid == -1) + CModuleRecord* mod = &m_ConnectedModules[i]; + if (mod->uid == -1) { - mod.Reset(); - mod.uid = i; - return &mod; + mod->Reset(); + mod->uid = i; + printf("CreateNewRecord(%d)\n", mod->uid); + return mod; } } return NULL; @@ -571,13 +565,14 @@ CMessageManager::ConnectModuleV2(MODULE_ID module_id, UPipe *pSourcePipe, MDF_CO mod->DaemonStatus = data->daemon_status; mod->AllowMultiple = data->allow_multiple; mod->pid = data->pid; - mod->SetName(data->name); + mod->SetName(&(data->name[0])); SendAcknowledge(mod); + + // Notify that a V2 client has connected + SendHello(mod); } } - // Notify that a V2 client has connected - SendHello(module_id); } else module_id = 0; // something went wrong, don't allow the new connection @@ -589,10 +584,10 @@ CModuleRecord *CMessageManager::GetRecord(MODULE_ID module_id) { for (int i = 0; i < MAX_MODULES; i++) { - CModuleRecord mod = m_ConnectedModules[i]; - if (mod.ModuleID == module_id) + CModuleRecord* mod = &m_ConnectedModules[i]; + if ((mod->uid >= 0) && (mod->ModuleID == module_id)) { - return &mod; + return mod; } } return NULL; @@ -601,27 +596,28 @@ CModuleRecord *CMessageManager::GetRecord(MODULE_ID module_id) void CMessageManager::DisconnectModule(MODULE_ID module_id) { CModuleRecord *mod = GetRecord(module_id); - if (mod == NULL) + if ((mod == NULL) || (mod->uid < 0)) return; // Send ACK SendAcknowledge(mod); - // Close module's pipe - UPipeAutoServer::_server->DisconnectClient(mod->pModulePipe); - // Make sure not to send anything to disconnected client RemoveSubscription(mod, ALL_MESSAGE_TYPES); // Notify that the module has left SendGoodbye(mod); + // Close module's pipe + UPipeAutoServer::_server->DisconnectClient(mod->pModulePipe); + // Clean up module record CleanUpModuleRecord(mod); } void CMessageManager::CleanUpModuleRecord(CModuleRecord *mod) { + printf("CleanUpModuleRecord(%d)\n", mod->ModuleID); RemoveSubscription(mod, ALL_MESSAGE_TYPES); mod->Reset(); } @@ -651,17 +647,17 @@ void CMessageManager::ShutdownAllModules(int shutdown_RTMA, int shutdown_daemons { for (int uid = 0; uid < MAX_MODULES; uid++) { - CModuleRecord mod = m_ConnectedModules[uid]; - if (mod.uid >= 0 && mod.ModuleID > 0) + CModuleRecord *mod = &m_ConnectedModules[uid]; + if (mod->uid >= 0 && mod->ModuleID > 0) { - if (mod.DaemonStatus) + if (mod->DaemonStatus) { if (shutdown_daemons) - ShutdownModule(mod.ModuleID); + ShutdownModule(mod->ModuleID); } else { - ShutdownModule(mod.ModuleID); + ShutdownModule(mod->ModuleID); } } } @@ -671,7 +667,7 @@ void CMessageManager::AddSubscription(CModuleRecord *mod, MSG_TYPE message_type) { MSG_TYPE mt; - if (mod) + if ((mod != NULL) && (mod->uid >= 0)) { if (((message_type < 0) || (message_type > MAX_MESSAGE_TYPES)) && (message_type != ALL_MESSAGE_TYPES)) { @@ -705,7 +701,7 @@ void CMessageManager::RemoveSubscription(CModuleRecord *mod, MSG_TYPE message_ty // Might wanna add checks here for valid module_id, connected module MSG_TYPE mt; - if (mod) + if ((mod != NULL) && (mod->uid >= 0)) { switch (message_type) { @@ -777,6 +773,7 @@ void CMessageManager::HandleConnect(CMessage *M, UPipe *pSourcePipe) int prev_priority_class = GetMyPriority(); SetMyPriority(NORMAL_PRIORITY_CLASS); MODULE_ID mod_id = ConnectModule(M->src_mod_id, pSourcePipe, data.logger_status, data.daemon_status); + printf("Connect: %d\n", mod_id); if (mod_id > 0) { SetMyPriority(prev_priority_class); @@ -790,6 +787,7 @@ void CMessageManager::HandleConnectV2(CMessage *M, UPipe *pSourcePipe) int prev_priority_class = GetMyPriority(); SetMyPriority(NORMAL_PRIORITY_CLASS); MODULE_ID mod_id = ConnectModuleV2(M->src_mod_id, pSourcePipe, &connect_v2); + printf("ConnectV2: %d\n", mod_id); if (mod_id > 0) { SetMyPriority(prev_priority_class); @@ -815,16 +813,15 @@ void CMessageManager::HandleModuleDisconnect(CMessage *M) void CMessageManager::HandleModuleReady(CMessage *M) { MDF_MODULE_READY module_ready; - MDF_HELLO hello; M->GetData(&module_ready); CModuleRecord *mod = GetRecord(M->src_mod_id); - if (mod) + if ((mod != NULL) && (mod->uid >= 0)) { mod->pid = module_ready.pid; // Notify that the module has joined and ready - SendHello(M->src_mod_id); + SendHello(mod); } } @@ -834,7 +831,7 @@ void CMessageManager::HandleSubscribe(CMessage *M) M->GetData(&msg_type_to_subscribe); CModuleRecord *mod = GetRecord(M->src_mod_id); - if (mod) + if ((mod != NULL) && (mod->uid >= 0)) { AddSubscription(mod, msg_type_to_subscribe); DEBUG_TEXT_(" Added subscription to msg type " << msg_type_to_subscribe << " for module " << M->src_mod_id << "... "); @@ -847,7 +844,7 @@ void CMessageManager::HandleUnsubscribe(CMessage *M) MSG_TYPE msg_type_to_unsubscribe; M->GetData(&msg_type_to_unsubscribe); CModuleRecord *mod = GetRecord(M->src_mod_id); - if (mod) + if ((mod != NULL) && (mod->uid >= 0)) { RemoveSubscription(mod, msg_type_to_unsubscribe); SendAcknowledge(mod); @@ -859,7 +856,7 @@ void CMessageManager::HandlePauseSubscription(CMessage *M) MSG_TYPE msg_type_to_pause; M->GetData(&msg_type_to_pause); CModuleRecord *mod = GetRecord(M->src_mod_id); - if (mod) + if ((mod != NULL) && (mod->uid >= 0)) { PauseSubscription(mod, msg_type_to_pause); SendAcknowledge(mod); @@ -872,7 +869,7 @@ void CMessageManager::HandleResumeSubscription(CMessage *M) MSG_TYPE msg_type_to_resume; M->GetData(&msg_type_to_resume); CModuleRecord *mod = GetRecord(M->src_mod_id); - if (mod) + if ((mod != NULL) && (mod->uid >= 0)) { ResumeSubscription(mod, msg_type_to_resume); SendAcknowledge(mod); @@ -898,7 +895,7 @@ bool CMessageManager::IsModuleSubscribed(UID uid, MSG_TYPE message_type) void CMessageManager::SendAcknowledge(CModuleRecord *mod) { - if (mod) + if ((mod != NULL) && (mod->uid >= 0)) { DEBUG_TEXT_("Sending ACK to module " << mod_id << "... "); DispatchSignal(MT_ACKNOWLEDGE, mod); @@ -914,10 +911,11 @@ void CMessageManager::SendIntroductions(MODULE_ID mod_id) { for (int i = 0; i < MAX_MODULES; i++) { - CModuleRecord mod = m_ConnectedModules[i]; - if (mod.uid >= 0 && mod.ModuleID > 0) - { - mod.SetHello(&hello); + CModuleRecord *mod = &m_ConnectedModules[i]; + if (mod->uid >= 0 && mod->ModuleID > 0) + { + mod->SetHello(&hello); + printf("Hello: %d\n", hello.mod_id); m_OutMsg.Set(MT_HELLO, &hello, sizeof(hello)); SendMessage(&m_OutMsg, dest_mod); } From 1fd71efb4ea1d2450bb8647f7779a4d9ca54858b Mon Sep 17 00:00:00 2001 From: David Weir Date: Thu, 4 Dec 2025 13:06:18 -0500 Subject: [PATCH 13/17] Remove code no longer needed --- src/modules/MessageManager/MessageManager.h | 67 ++++++++------------- 1 file changed, 26 insertions(+), 41 deletions(-) diff --git a/src/modules/MessageManager/MessageManager.h b/src/modules/MessageManager/MessageManager.h index 906b983..7995ca4 100644 --- a/src/modules/MessageManager/MessageManager.h +++ b/src/modules/MessageManager/MessageManager.h @@ -31,8 +31,9 @@ class CModuleRecord } void SetName(char* src_name) { + printf("SetName: %d\n", ModuleID); if (name != NULL) { - free(name); + myfree(name); } name = (char*) malloc(MAX_NAME_LEN); if (name != NULL) { @@ -45,20 +46,37 @@ class CModuleRecord hello->uid = uid; hello->mod_id = ModuleID; hello->pid = pid; - pModulePipe->GetIpAddress(&(hello->addr[0]), &(hello->port), sizeof(hello->addr)); - memcpy(hello->name, name, sizeof(hello->name)); + if (name != NULL) { + memcpy(hello->name, name, sizeof(hello->name)); + } + else { + hello->name[0] = '\0'; + } + } void SetGoodbye(MDF_GOODBYE *goodbye) { goodbye->uid = uid; goodbye->mod_id = ModuleID; goodbye->pid = pid; - pModulePipe->GetIpAddress(&(goodbye->addr[0]), &(goodbye->port), sizeof(goodbye->addr)); - memcpy(goodbye->name, name, sizeof(goodbye->name)); + if (name != NULL) { + memcpy(goodbye->name, name, sizeof(goodbye->name)); + } + else { + goodbye->name[0] = '\0'; + } + } + + void myfree(char* p) { + free(p); } void Reset( void) { + if (name != NULL) { + myfree(name); + name = NULL; + } ModuleID = -1; pModulePipe = NULL; LoggerStatus = 0; @@ -66,13 +84,11 @@ class CModuleRecord AllowMultiple = 0; pid = 0; uid = -1; - if (name != NULL) { - free(name); - name = NULL; - } + } ~CModuleRecord(){ + printf("~CModuleRecord(%d)\n", ModuleID); Reset(); } }; @@ -376,10 +392,6 @@ class CMessageManager : public UPipeAutoServer DispatchMessage( CMessage *M, CModuleRecord *dest_mod); //Sends the message only to the specified mod_id and Logger modules, headers specifying the message came from MM - void - DispatchSignal( MSG_TYPE sig); - //Sends the signal to all subscribers and Logger modules, headers specifying the signal came from MM - void DispatchSignal( MSG_TYPE sig, CModuleRecord *dest_mod); //Sends the signal only to the specified mod_id and Logger modules, headers specifying the message came from MM @@ -414,12 +426,6 @@ class CMessageManager : public UPipeAutoServer void ShutdownAllModules(int shutdown_RTMA=1, int shutdown_daemons=1); - void - ShutdownStatusModule(void); - - void - ShutdownLoggerModule(void); - void AddSubscription(CModuleRecord *mod, MSG_TYPE message_type); @@ -448,10 +454,7 @@ class CMessageManager : public UPipeAutoServer SendIntroductions(MODULE_ID mod_id); void - SendHello(MODULE_ID mod_id); - - void - SendGoodbye(MODULE_ID mod_id); + SendHello(CModuleRecord *mod); void SendGoodbye(CModuleRecord* mod); @@ -505,24 +508,6 @@ class CMessageManager : public UPipeAutoServer int ModuleIsConnected(MODULE_ID mod_id); - UPipe* - GetModPipe(UID uid); - - int - IsDaemon(UID uid); - - void - ReportLoad( void); - //sends an MT_MM_LOAD_REPORT message with the load data of all connected modules - - void - AskUsageReport( void); - //sends MT_MOD_REPORT_USAGE to all connected modules - - void - ReportMMStatus( void); - //sends MT_MM_STATUS_REPORT message - void LogFailedMessage( CMessage *m, MODULE_ID mod_id); //sends a message to logger modules indicating that a message failed to be forwarded to one of the modules From b00b800a9bf1301df420e2ffb6b9da94703db4d9 Mon Sep 17 00:00:00 2001 From: David Weir Date: Thu, 4 Dec 2025 13:06:52 -0500 Subject: [PATCH 14/17] Update visual studio porj --- src/modules/MessageManager/MessageManager.vcxproj | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/modules/MessageManager/MessageManager.vcxproj b/src/modules/MessageManager/MessageManager.vcxproj index 7d11bd7..7c2e033 100644 --- a/src/modules/MessageManager/MessageManager.vcxproj +++ b/src/modules/MessageManager/MessageManager.vcxproj @@ -162,12 +162,12 @@ ../../../lib/RTMAd.lib;%(AdditionalDependencies) - MessageManager.exe + .\MessageManager.exe true ..\PipeLib\debug;%(AdditionalLibraryDirectories) true .\Debug/MessageManager.pdb - Windows + Console copy MessageManager.exe ..\..\..\bin From 2611ec0c64c7be9b0b671e14b828607e80053f05 Mon Sep 17 00:00:00 2001 From: David Weir Date: Thu, 4 Dec 2025 16:08:51 -0500 Subject: [PATCH 15/17] Add ip address to hello and goodbye --- include/RTMA_types.h | 7 ++- include/internal/UPipe.h | 2 +- src/core/PipeLib/SocketPipe.cpp | 18 ++++++- src/core/PipeLib/SocketPipe.h | 1 + src/modules/MessageManager/MessageManager.cpp | 20 ++++---- src/modules/MessageManager/MessageManager.h | 48 +++++++++++++++++-- 6 files changed, 81 insertions(+), 15 deletions(-) diff --git a/include/RTMA_types.h b/include/RTMA_types.h index 34a374a..f68326e 100644 --- a/include/RTMA_types.h +++ b/include/RTMA_types.h @@ -23,6 +23,7 @@ typedef int32_t UID; #define MAX_RTMA_MODULE_ID 9 // Module ID-s below 10 are reserved for RTMA core #define MAX_MODULE_ID 32000 +#define SUBSCRIBER_FLAG_TIMING_ONLY 0x02 // Header fields for all messages passed through RTMA // Following macro was commented out for compatibility with ctypesgen2 v2.2.2 python package (used for Python3 compatiblity) All references to this macro in this file and in RTMA.h were replaced with actual values //#define RTMA_MSG_HEADER_FIELDS \ @@ -223,7 +224,8 @@ typedef struct { int32_t uid; int32_t pid; MODULE_ID mod_id; - int16_t reserved; + uint16_t port; + char addr[MAX_NAME_LEN]; char name[MAX_NAME_LEN]; } MDF_HELLO; @@ -232,7 +234,8 @@ typedef struct { int32_t uid; int32_t pid; MODULE_ID mod_id; - int16_t reserved; + uint16_t port; + char addr[MAX_NAME_LEN]; char name[MAX_NAME_LEN]; } MDF_GOODBYE; diff --git a/include/internal/UPipe.h b/include/internal/UPipe.h index e7ba13f..e68bce6 100644 --- a/include/internal/UPipe.h +++ b/include/internal/UPipe.h @@ -150,7 +150,7 @@ class UPipe // never blocks (if no space in pipe, then returns with 0). Returns number of // bytes written. virtual int Write( void *data_buffer, int n_bytes, double timeout) = 0; - + virtual int GetIpAddress(char* addr, uint16_t* port, int bufsz) = 0; }; diff --git a/src/core/PipeLib/SocketPipe.cpp b/src/core/PipeLib/SocketPipe.cpp index 8bd4dba..7f02c5f 100644 --- a/src/core/PipeLib/SocketPipe.cpp +++ b/src/core/PipeLib/SocketPipe.cpp @@ -443,8 +443,24 @@ SocketPipe::Write( void *data_buffer, int n_bytes, double timeout) } return nbytes_writ; } CATCH_and_THROW( "SocketPipe::Write"); -} +} +int SocketPipe::GetIpAddress(char* addr, uint16_t* port, int bufsz) +{ + // windows: https://learn.microsoft.com/en-us/windows/win32/api/winsock2/nf-winsock2-getsockname + // linux: https://man7.org/linux/man-pages/man2/getsockname.2.html + struct sockaddr_in name = { 0 }; + int namelen = sizeof(name); + if (getpeername(_hPipe.id, (struct sockaddr*)&name, &namelen) == SOCKET_ERROR) { + return 0; + } + + *port = ntohs(name.sin_port); + inet_ntop(AF_INET, &(name.sin_addr), addr, bufsz); + //printf("%s:%u\n", addr, *port); + + return 0; +} // ////////////////////////////////////////////////////////////////////// diff --git a/src/core/PipeLib/SocketPipe.h b/src/core/PipeLib/SocketPipe.h index ffbe62f..91200aa 100644 --- a/src/core/PipeLib/SocketPipe.h +++ b/src/core/PipeLib/SocketPipe.h @@ -72,6 +72,7 @@ class SocketPipe : public UPipe int GetCapacity( void); int Read( void *data_buffer, int n_bytes, double timeout); int Write( void *data_buffer, int n_bytes, double timeout); + int GetIpAddress(char* addr, uint16_t* port, int bufsz); private: SocketHandle _hPipe; // Handle to underlying OS socket diff --git a/src/modules/MessageManager/MessageManager.cpp b/src/modules/MessageManager/MessageManager.cpp index c999aac..0363ff4 100644 --- a/src/modules/MessageManager/MessageManager.cpp +++ b/src/modules/MessageManager/MessageManager.cpp @@ -198,7 +198,7 @@ void CMessageManager::SendHello(CModuleRecord *mod) MDF_HELLO hello; if ((mod != NULL) && (mod->uid >= 0)){ mod->SetHello(&hello); - printf("Hello: %d\n", hello.mod_id); + //printf("Hello: %d\n", hello.mod_id); m_OutMsg.Set(MT_HELLO, &hello, sizeof(hello)); DispatchMessage(&m_OutMsg); } @@ -209,7 +209,7 @@ void CMessageManager::SendGoodbye(CModuleRecord *mod) MDF_GOODBYE goodbye; if ((mod != NULL) && (mod->uid >= 0)) { mod->SetGoodbye(&goodbye); - printf("Goodbye: %d\n", goodbye.mod_id); + //printf("Goodbye: %d\n", goodbye.mod_id); m_OutMsg.Set(MT_GOODBYE, &goodbye, sizeof(goodbye)); DispatchMessage(&m_OutMsg); } @@ -235,6 +235,8 @@ void CMessageManager::HandleSetName(CMessage *M) if ((mod != NULL) && (mod->uid >= 0)) { mod->SetName(&(set_name.name[0])); + //printf("SetName: %d -> %s\n", mod->ModuleID, mod->name); + SendHello(mod); } } @@ -356,7 +358,7 @@ void CMessageManager::DispatchMessage(CMessage *M) { // Send message to all subscribers UID uid = SL->GetFirstSubscriber(); - while (uid > 0) + while (uid >= 0) { CModuleRecord *mod = &m_ConnectedModules[uid]; SendMessage(M, mod); @@ -490,7 +492,7 @@ CMessageManager::GetOpenRecord() { mod->Reset(); mod->uid = i; - printf("CreateNewRecord(%d)\n", mod->uid); + //printf("CreateNewRecord(%d)\n", mod->uid); return mod; } } @@ -522,7 +524,7 @@ CMessageManager::ConnectModule(MODULE_ID module_id, UPipe *pSourcePipe, short lo // Create a module record mod->ModuleID = module_id; - mod->pModulePipe = pSourcePipe; + mod->SetUPipe(pSourcePipe); mod->LoggerStatus = logger_status; mod->DaemonStatus = daemon_status; @@ -560,7 +562,7 @@ CMessageManager::ConnectModuleV2(MODULE_ID module_id, UPipe *pSourcePipe, MDF_CO DEBUG_TEXT("Connecting module " << module_id << " on pipe " << pSourcePipe); mod->ModuleID = module_id; - mod->pModulePipe = pSourcePipe; + mod->SetUPipe(pSourcePipe); mod->LoggerStatus = data->logger_status; mod->DaemonStatus = data->daemon_status; mod->AllowMultiple = data->allow_multiple; @@ -617,7 +619,7 @@ void CMessageManager::DisconnectModule(MODULE_ID module_id) void CMessageManager::CleanUpModuleRecord(CModuleRecord *mod) { - printf("CleanUpModuleRecord(%d)\n", mod->ModuleID); + //printf("CleanUpModuleRecord(%d)\n", mod->ModuleID); RemoveSubscription(mod, ALL_MESSAGE_TYPES); mod->Reset(); } @@ -758,10 +760,12 @@ void CMessageManager::SubscriptionOption(UID uid, MDF_SUBSCRIPTION_OPTION *sub_o if (sub_opt->value) { GetSubscriberList(sub_opt->msg_type)->SetSubscriberOption(uid, sub_opt->option); + printf("Enabling subscription option: %d\n", sub_opt->option); } else { GetSubscriberList(sub_opt->msg_type)->ClearSubscriberOption(uid, sub_opt->option); + printf("Disabling subscription option: %d\n", sub_opt->option); } } @@ -915,7 +919,7 @@ void CMessageManager::SendIntroductions(MODULE_ID mod_id) if (mod->uid >= 0 && mod->ModuleID > 0) { mod->SetHello(&hello); - printf("Hello: %d\n", hello.mod_id); + //printf("Hello: %d\n", hello.mod_id); m_OutMsg.Set(MT_HELLO, &hello, sizeof(hello)); SendMessage(&m_OutMsg, dest_mod); } diff --git a/src/modules/MessageManager/MessageManager.h b/src/modules/MessageManager/MessageManager.h index 7995ca4..c495254 100644 --- a/src/modules/MessageManager/MessageManager.h +++ b/src/modules/MessageManager/MessageManager.h @@ -20,18 +20,19 @@ class CModuleRecord short LoggerStatus; short DaemonStatus; short AllowMultiple; + uint16_t port; int32_t uid; int32_t pid; - + char* addr; char* name; CModuleRecord( ) { name = NULL; + addr = NULL; Reset(); } void SetName(char* src_name) { - printf("SetName: %d\n", ModuleID); if (name != NULL) { myfree(name); } @@ -42,10 +43,36 @@ class CModuleRecord } } + void SetUPipe(UPipe* pipe) { + if (addr != NULL) { + myfree(addr); + } + if (pipe == NULL) { + return; + } + + pModulePipe = pipe; + + addr = (char*)malloc(MAX_NAME_LEN); + if (addr != NULL) { + pipe->GetIpAddress(addr, &port, MAX_NAME_LEN); + } + + } + void SetHello(MDF_HELLO *hello) { hello->uid = uid; hello->mod_id = ModuleID; hello->pid = pid; + hello->port = port; + + if (addr != NULL) { + memcpy(hello->addr, addr, sizeof(hello->addr)); + } + else { + hello->addr[0] = '\0'; + } + if (name != NULL) { memcpy(hello->name, name, sizeof(hello->name)); } @@ -59,6 +86,15 @@ class CModuleRecord goodbye->uid = uid; goodbye->mod_id = ModuleID; goodbye->pid = pid; + goodbye->port = port; + + if (addr != NULL) { + memcpy(goodbye->addr, addr, sizeof(goodbye->addr)); + } + else { + goodbye->addr[0] = '\0'; + } + if (name != NULL) { memcpy(goodbye->name, name, sizeof(goodbye->name)); } @@ -77,6 +113,12 @@ class CModuleRecord myfree(name); name = NULL; } + + if (addr != NULL) { + myfree(addr); + addr = NULL; + } + ModuleID = -1; pModulePipe = NULL; LoggerStatus = 0; @@ -185,7 +227,7 @@ class CList }; #define SUBSCRIBER_FLAG_PAUSE 0x01 -#define SUBSCRIBER_FLAG_TIMING_ONLY 0x02 + class CSubscriberList : protected CList { From 32d921beb193b16eb36792bef5188713e440093f Mon Sep 17 00:00:00 2001 From: David Weir Date: Thu, 4 Dec 2025 16:23:22 -0500 Subject: [PATCH 16/17] Comment debug prints --- src/modules/MessageManager/MessageManager.cpp | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/modules/MessageManager/MessageManager.cpp b/src/modules/MessageManager/MessageManager.cpp index 0363ff4..d4db18c 100644 --- a/src/modules/MessageManager/MessageManager.cpp +++ b/src/modules/MessageManager/MessageManager.cpp @@ -777,7 +777,7 @@ void CMessageManager::HandleConnect(CMessage *M, UPipe *pSourcePipe) int prev_priority_class = GetMyPriority(); SetMyPriority(NORMAL_PRIORITY_CLASS); MODULE_ID mod_id = ConnectModule(M->src_mod_id, pSourcePipe, data.logger_status, data.daemon_status); - printf("Connect: %d\n", mod_id); + //printf("Connect: %d\n", mod_id); if (mod_id > 0) { SetMyPriority(prev_priority_class); @@ -791,7 +791,7 @@ void CMessageManager::HandleConnectV2(CMessage *M, UPipe *pSourcePipe) int prev_priority_class = GetMyPriority(); SetMyPriority(NORMAL_PRIORITY_CLASS); MODULE_ID mod_id = ConnectModuleV2(M->src_mod_id, pSourcePipe, &connect_v2); - printf("ConnectV2: %d\n", mod_id); + //printf("ConnectV2: %d\n", mod_id); if (mod_id > 0) { SetMyPriority(prev_priority_class); From 1a1aa8c032fed755e69ae962517fe3a442ed446e Mon Sep 17 00:00:00 2001 From: David Weir Date: Fri, 5 Dec 2025 09:36:09 -0500 Subject: [PATCH 17/17] Remove sub option. --- include/RTMA_types.h | 15 ----- src/modules/MessageManager/MessageManager.cpp | 65 ++++--------------- src/modules/MessageManager/MessageManager.h | 61 ----------------- 3 files changed, 11 insertions(+), 130 deletions(-) diff --git a/include/RTMA_types.h b/include/RTMA_types.h index f68326e..9a76640 100644 --- a/include/RTMA_types.h +++ b/include/RTMA_types.h @@ -23,7 +23,6 @@ typedef int32_t UID; #define MAX_RTMA_MODULE_ID 9 // Module ID-s below 10 are reserved for RTMA core #define MAX_MODULE_ID 32000 -#define SUBSCRIBER_FLAG_TIMING_ONLY 0x02 // Header fields for all messages passed through RTMA // Following macro was commented out for compatibility with ctypesgen2 v2.2.2 python package (used for Python3 compatiblity) All references to this macro in this file and in RTMA.h were replaced with actual values //#define RTMA_MSG_HEADER_FIELDS \ @@ -181,13 +180,6 @@ typedef struct { #define MT_RESET_MESSAGE_LOG 60 #define MT_DUMP_MESSAGE_LOG 61 -#define MT_TIMING_MESSAGE 80 -typedef struct { - unsigned short timing[MAX_MESSAGE_TYPES]; - int ModulePID[MAX_MODULES]; //0 if not connected - double send_time; -} MDF_TIMING_MESSAGE; - #define MT_CONNECT_V2 4 typedef struct { int16_t logger_status; @@ -244,12 +236,5 @@ typedef struct { char name[MAX_NAME_LEN]; } MDF_CLIENT_SET_NAME; -#define MT_MESSAGE_TIMING 9 -typedef struct { - MSG_TYPE msg_type; - MODULE_ID src_id; - int16_t reserved; - double send_time; -} MDF_MESSAGE_TIMING; #endif //_RTMA_TYPES_H_ diff --git a/src/modules/MessageManager/MessageManager.cpp b/src/modules/MessageManager/MessageManager.cpp index d4db18c..cde4394 100644 --- a/src/modules/MessageManager/MessageManager.cpp +++ b/src/modules/MessageManager/MessageManager.cpp @@ -147,10 +147,6 @@ void CMessageManager::ProcessMessage(CMessage *M, UPipe *pSourcePipe) HandleSetName(M); break; - case MT_SUBSCRIPTION_OPTION: - HandleSubscriptionControl(M); - break; - case MT_PONG: break; @@ -240,17 +236,6 @@ void CMessageManager::HandleSetName(CMessage *M) } } -void CMessageManager::HandleSubscriptionControl(CMessage *M) -{ - MDF_SUBSCRIPTION_OPTION sub_opt; - M->GetData((void *)&sub_opt); - CModuleRecord *mod = GetRecord(M->src_mod_id); - if ((mod != NULL) && (mod->uid >= 0)) - { - SubscriptionOption(mod->uid, &sub_opt); - } -} - /* * Should be called when forwarding a message from other modules * The given message will be forwarded to: @@ -305,41 +290,21 @@ void CMessageManager::DistributeMessage(CMessage *M) DEBUG_TEXT_("Forwarding message to module " << mod_id << "... "); try { - int status = 0; - if (SL->TimingOnly()) - { - MDF_MESSAGE_TIMING msg_time; - if (!timing_set) { - timing_set = 1; - msg_time.msg_type = M->msg_type; - msg_time.src_id = M->src_mod_id; - msg_time.send_time = M->send_time; - m_OutMsg.Set(MT_MESSAGE_TIMING, &msg_time, sizeof(msg_time)); - } - status = SendMessage(&m_OutMsg, mod); - } - else - { - status = ForwardMessage(M, mod); - } - - if (status == 0) - { + int status = ForwardMessage(M, mod); + if (status == 0) { LogFailedMessage(M, mod->ModuleID); DEBUG_TEXT("Failed to Forward Message!"); } - else - { + else { DEBUG_TEXT("Forwarded!"); } } - catch (UPipeClosedException &E) - { + catch (UPipeClosedException& E) { DEBUG_TEXT("Failed to Forward Message, destination module socket is closed/dead"); } } - uid = SL->GetNextSubscriber(); + uid = SL->GetNextSubscriber(); } } DEBUG_TEXT("Done distributing!"); @@ -529,6 +494,12 @@ CMessageManager::ConnectModule(MODULE_ID module_id, UPipe *pSourcePipe, short lo mod->DaemonStatus = daemon_status; SendAcknowledge(mod); + + // Notify that a client has connected + // Note: PID and Name will not be filled in + // Additional HELLO msgs will be sent with + // CLIENT_SET_NAME and MODULE_READY + SendHello(mod); } } } @@ -755,20 +726,6 @@ void CMessageManager::ResumeSubscription(CModuleRecord *mod, MSG_TYPE message_ty } } -void CMessageManager::SubscriptionOption(UID uid, MDF_SUBSCRIPTION_OPTION *sub_opt) -{ - if (sub_opt->value) - { - GetSubscriberList(sub_opt->msg_type)->SetSubscriberOption(uid, sub_opt->option); - printf("Enabling subscription option: %d\n", sub_opt->option); - } - else - { - GetSubscriberList(sub_opt->msg_type)->ClearSubscriberOption(uid, sub_opt->option); - printf("Disabling subscription option: %d\n", sub_opt->option); - } -} - void CMessageManager::HandleConnect(CMessage *M, UPipe *pSourcePipe) { MDF_CONNECT data; diff --git a/src/modules/MessageManager/MessageManager.h b/src/modules/MessageManager/MessageManager.h index c495254..86ade55 100644 --- a/src/modules/MessageManager/MessageManager.h +++ b/src/modules/MessageManager/MessageManager.h @@ -283,50 +283,6 @@ class CSubscriberList : protected CList } } - void - SetSubscriberOption(UID uid, int option) { - // Only act on valid options - switch(option) { - case SUBSCRIBER_FLAG_PAUSE: - break; - case SUBSCRIBER_FLAG_TIMING_ONLY: - break; - default: - return; - } - - CListItem *current_item = GetFirstItem(); - while( current_item != NULL) { - if( current_item->data == uid) { - set_flag_bits( current_item->flags, option); - break; - } - current_item = GetNextItem( current_item); - } - } - - void - ClearSubscriberOption(UID uid, int option) { - // Only act on valid options - switch(option) { - case SUBSCRIBER_FLAG_PAUSE: - break; - case SUBSCRIBER_FLAG_TIMING_ONLY: - break; - default: - return; - } - - CListItem *current_item = GetFirstItem(); - while( current_item != NULL) { - if( current_item->data == uid) { - clear_flag_bits( current_item->flags, option); - break; - } - current_item = GetNextItem( current_item); - } - } - int SubscriptionPaused(void) { int is_paused = 0; @@ -338,17 +294,6 @@ class CSubscriberList : protected CList return is_paused; } - int - TimingOnly(void) { - int is_timing_only = 0; - if( m_CurrentItem != NULL) { - if( check_flag_bits( m_CurrentItem->flags, SUBSCRIBER_FLAG_TIMING_ONLY)) { - is_timing_only = 1; - } - } - return is_timing_only; - } - UID GetFirstSubscriber(void) { m_CurrentItem = GetFirstItem(); @@ -480,9 +425,6 @@ class CMessageManager : public UPipeAutoServer void ResumeSubscription(CModuleRecord *mod, MSG_TYPE msg_type_to_resume); - void - SubscriptionOption(UID uid, MDF_SUBSCRIPTION_OPTION *sub_opt); - CSubscriberList * GetSubscriberList( MSG_TYPE message_type); @@ -507,9 +449,6 @@ class CMessageManager : public UPipeAutoServer void HandleSetName(CMessage *m); - void - HandleSubscriptionControl(CMessage *m); - void HandleForceDisconnect(CMessage *m);