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/include/RTMA_types.h b/include/RTMA_types.h index 6efe093..9a76640 100644 --- a/include/RTMA_types.h +++ b/include/RTMA_types.h @@ -1,25 +1,27 @@ #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 @@ -88,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 @@ -107,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; @@ -116,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 @@ -136,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 @@ -264,11 +180,61 @@ typedef struct { #define MT_RESET_MESSAGE_LOG 60 #define MT_DUMP_MESSAGE_LOG 61 -#define MT_TIMING_MESSAGE 80 +#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 7 typedef struct { - unsigned short timing[MAX_MESSAGE_TYPES]; - int ModulePID[MAX_MODULES]; //0 if not connected - double send_time; -} MDF_TIMING_MESSAGE; + 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[MAX_NAME_LEN]; + 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[MAX_NAME_LEN]; + char name[MAX_NAME_LEN]; +} MDF_GOODBYE; + +#define MT_CLIENT_SET_NAME 34 +typedef struct { + char name[MAX_NAME_LEN]; +} MDF_CLIENT_SET_NAME; + #endif //_RTMA_TYPES_H_ 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/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; 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 5bfa057..91200aa 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); + int GetIpAddress(char* addr, uint16_t* port, int bufsz); private: SocketHandle _hPipe; // Handle to underlying OS socket diff --git a/src/core/RTMA.cpp b/src/core/RTMA.cpp index c6fa958..395d3c2 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 //////////////////////////////////////////////// ////////////////////////////////////////////////////////////////////////////// @@ -352,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)"); @@ -370,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( ); } @@ -890,267 +845,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///////////////////////////////////////// ////////////////////////////////////////////////////////////////////////////// diff --git a/src/modules/MessageManager/MessageManager.cpp b/src/modules/MessageManager/MessageManager.cpp index e4199e1..cde4394 100644 --- a/src/modules/MessageManager/MessageManager.cpp +++ b/src/modules/MessageManager/MessageManager.cpp @@ -10,28 +10,27 @@ #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[]) #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); + CMessageManager* MM = new CMessageManager(); + MM->MainLoop(options); return 0; } - catch(MyCException &E) + catch (MyCException &E) { E.AddToStack("MM WinMain- aborting"); E.ReportToFile(); @@ -39,239 +38,202 @@ int main( int argc, char *argv[]) } } -CMessageManager::CMessageManager( ) +CMessageManager::CMessageManager() { - m_Version = "2.11bci"; - m_NextDynamicModIdOffset = 0; - - InitializeAbsTime(); + m_Version = "2.11bci"; + m_NextDynamicModId = DYN_MOD_ID_START; - // from RP3 RTMA (for timing message) - #ifdef __unix__ - ftime(&timebuffer); - #else - _ftime(&timebuffer); // C4996 - #endif + // Initialize the module record array + // uid=-1 indicates an unused slot + for (int i = 0; i < MAX_MODULES; i++) + { + m_ConnectedModules[i].Reset(); + } - m_LastMessageCount = timebuffer.time; - m_LastMessageCountmsec = timebuffer.millitm; - for(int i=0;i 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 mod_id = 0; mod_id < MAX_MODULES; mod_id++) { - if( m_ConnectedModules[mod_id].pModulePipe == pModulePipe) { + for (int uid = 0; uid < MAX_MODULES; uid++) + { + CModuleRecord *mod = &m_ConnectedModules[uid]; + if (mod->pModulePipe == pModulePipe) + { + // Make sure not to send anything to disconnected client + RemoveSubscription(mod, ALL_MESSAGE_TYPES); + + // Notify that the module has left + SendGoodbye(mod); + // Delete module record - CleanUpModuleRecord( mod_id); - DEBUG_TEXT_( ", disconnected module " << mod_id); + CleanUpModuleRecord(mod); + 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; - - //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)); - 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) { - SendAcknowledge( mod_id); - SetMyPriority(prev_priority_class); - } - 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 += "]"; - CMessage R(MT_MM_ERROR, (void*)err.GetContent(), err.GetLen()); - DispatchMessage(&R); - } - 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); - ShutdownModule(mod_id); - } - break; - case MT_DISCONNECT: - prev_priority_class = GetMyPriority(); - SetMyPriority(NORMAL_PRIORITY_CLASS); - DisconnectModule( mod_id); - SetMyPriority(prev_priority_class); - break; +void CMessageManager::ProcessMessage(CMessage *M, UPipe *pSourcePipe) +{ + DEBUG_TEXT_("Processing message... "); - case MT_MODULE_READY: //store pids so that application module can kill processes later - MDF_MODULE_READY m; - M->GetData(&m); - if (mod_id>=0 && mod_idGetData( &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; + switch (M->msg_type) + { + case MT_CONNECT: + HandleConnect(M, pSourcePipe); + 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_CONNECT_V2: + HandleConnectV2(M, pSourcePipe); + 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_CLIENT_SET_NAME: + HandleSetName(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_PONG: + 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_INTRODUCE: + SendIntroductions(M->src_mod_id); + 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_FORCE_DISCONNECT: + HandleForceDisconnect(M); + break; - default: - DEBUG_TEXT( "Nothing to do!"); - return; + case MT_DISCONNECT: + HandleModuleDisconnect(M); + break; + + case MT_MODULE_READY: + HandleModuleReady(M); + break; + + case MT_SUBSCRIBE: + HandleSubscribe(M); + break; + + case MT_UNSUBSCRIBE: + HandleUnsubscribe(M); + break; + + case MT_PAUSE_SUBSCRIPTION: + HandlePauseSubscription(M); + break; + + case MT_RESUME_SUBSCRIPTION: + HandleResumeSubscription(M); + break; + + default: + DEBUG_TEXT("Nothing to do!"); + return; + } + DEBUG_TEXT("Processed!"); +} + +void CMessageManager::SendHello(CModuleRecord *mod) +{ + MDF_HELLO hello; + 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 != 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); + } +} + +void CMessageManager::SendPing(MODULE_ID mod_id) +{ + MDF_PING ping; + CModuleRecord *mod = GetRecord(mod_id); + if ((mod != NULL) && (mod->uid >= 0)) + { + 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 != NULL) && (mod->uid >= 0)) + { + mod->SetName(&(set_name.name[0])); + //printf("SetName: %d -> %s\n", mod->ModuleID, mod->name); + SendHello(mod); } - DEBUG_TEXT( "Processed!"); } /* @@ -282,117 +244,121 @@ CMessageManager::ProcessMessage( CMessage *M, UPipe *pSourcePipe) * 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..."); + int timing_set = 0; - CSubscriberList * SL; - SL = GetSubscriberList( M->msg_type); - if( SL != NULL) + CSubscriberList *SL; + 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) + if (has_specific_dest) { - //send only to the specific destination - if(mod_id == 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( m_ConnectedModules[mod_id].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 = ForwardMessage(M, mod_id); - if( status == 0) { - LogFailedMessage( M, mod_id); - DEBUG_TEXT( "Failed to Forward Message!"); - } else { - DEBUG_TEXT( "Forwarded!"); + int status = ForwardMessage(M, mod); + 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"); + } } - - mod_id = SL->GetNextSubscriber(); + + 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; - - SL = GetSubscriberList( M->msg_type); - if( SL != NULL) + CSubscriberList *SL; + + 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(); } } } - /* * 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, MODULE_ID mod_id) +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, mod_id); + SendMessage(M, dest_mod); - //CC all logger modules - SL = GetSubscriberList( M->msg_type); - if( SL != NULL) + // CC all logger modules + 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(); } } } @@ -402,11 +368,10 @@ CMessageManager::DispatchMessage( CMessage *M, MODULE_ID mod_id) * 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, MODULE_ID mod_id) +void CMessageManager::DispatchSignal(MSG_TYPE sig, CModuleRecord *dest_mod) { - CMessage R(sig); - DispatchMessage(&R, mod_id); + m_OutMsg.Set(sig); + DispatchMessage(&m_OutMsg, dest_mod); } /* @@ -414,370 +379,533 @@ CMessageManager::DispatchSignal( MSG_TYPE sig, MODULE_ID mod_id) * 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 mod_id=0; mod_id= 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, MODULE_ID mod_id) -//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 { - if( !ModuleIsConnected( mod_id)) return 0; - - UPipe *mod_pipe = GetModPipe( mod_id); - if(mod_pipe == NULL) return 0; - + UPipe *mod_pipe = mod->pModulePipe; + 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->dest_mod_id = mod_id; + 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( 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) -//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 { - if( !ModuleIsConnected( mod_id)) return 0; - - UPipe *mod_pipe = GetModPipe( mod_id); - if(mod_pipe == NULL) return 0; + 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) -//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 { - 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 < MAX_MODULES; i++) + { + CModuleRecord* mod = &m_ConnectedModules[i]; + if (mod->uid == -1) + { + mod->Reset(); + mod->uid = i; + //printf("CreateNewRecord(%d)\n", mod->uid); + return mod; + } + } + return NULL; } MODULE_ID -CMessageManager::ConnectModule( MODULE_ID module_id, UPipe *pSourcePipe, short logger_status, short daemon_status) +CMessageManager::ConnectModule(MODULE_ID module_id, UPipe *pSourcePipe, short logger_status, short daemon_status) { - if( ( module_id < MAX_MODULES) && ( module_id >= 0) && !ModuleIsConnected(module_id) ) + if ((module_id < MAX_MODULE_ID) && (module_id >= 0) && !ModuleIsConnected(module_id)) { - if( pSourcePipe != NULL) { + 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); + 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->SetUPipe(pSourcePipe); + mod->LoggerStatus = logger_status; + 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); + } + } + } + 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->SetUPipe(pSourcePipe); + mod->LoggerStatus = data->logger_status; + mod->DaemonStatus = data->daemon_status; + mod->AllowMultiple = data->allow_multiple; + mod->pid = data->pid; + mod->SetName(&(data->name[0])); - // Add default subscription to TIMER_EXPIRED - if( !logger_status) AddSubscription( module_id, MT_TIMER_EXPIRED); + SendAcknowledge(mod); + + // Notify that a V2 client has connected + SendHello(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; } -void -CMessageManager::DisconnectModule( MODULE_ID module_id) +CModuleRecord *CMessageManager::GetRecord(MODULE_ID module_id) { - if( !ModuleIsConnected( module_id)) return; + for (int i = 0; i < MAX_MODULES; i++) + { + CModuleRecord* mod = &m_ConnectedModules[i]; + if ((mod->uid >= 0) && (mod->ModuleID == module_id)) + { + return mod; + } + } + return NULL; +} + +void CMessageManager::DisconnectModule(MODULE_ID module_id) +{ + CModuleRecord *mod = GetRecord(module_id); + if ((mod == NULL) || (mod->uid < 0)) + return; // Send ACK - SendAcknowledge( module_id); + SendAcknowledge(mod); + + // 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 - UPipe *module_pipe = GetModPipe( module_id); - UPipeAutoServer::_server->DisconnectClient( module_pipe); + UPipeAutoServer::_server->DisconnectClient(mod->pModulePipe); // Clean up module record - CleanUpModuleRecord( module_id); + CleanUpModuleRecord(mod); } -void -CMessageManager::CleanUpModuleRecord( MODULE_ID module_id) +void CMessageManager::CleanUpModuleRecord(CModuleRecord *mod) { - RemoveSubscription( module_id, ALL_MESSAGE_TYPES); - m_ConnectedModules[module_id].Reset(); + //printf("CleanUpModuleRecord(%d)\n", mod->ModuleID); + RemoveSubscription(mod, ALL_MESSAGE_TYPES); + mod->Reset(); } -void -CMessageManager::ShutdownModule( MODULE_ID mod_id) +void CMessageManager::ShutdownModule(MODULE_ID mod_id) { - if( ModuleIsConnected( mod_id)) + CModuleRecord *mod = GetRecord(mod_id); + if (mod != NULL && mod->ModuleID > 0) { - switch(mod_id) + MODULE_ID mod_id = mod->ModuleID; + switch (mod_id) { - case MID_COMMAND_MODULE: - //never shutdown Command Module - break; - - case MID_APPLICATION_MODULE: - DispatchSignal( MT_AM_EXIT, mod_id); - break; - - case MID_STATUS_MODULE: - DispatchSignal( MT_SM_EXIT, mod_id); + case MID_QUICKLOGGER: + DispatchSignal(MT_LM_EXIT, mod); break; - case MID_QUICKLOGGER: - DispatchSignal( MT_LM_EXIT, mod_id); - break; - - default: - DispatchSignal( MT_EXIT, mod_id); + default: + DispatchSignal(MT_EXIT, mod); break; } - DisconnectModule( mod_id); + DisconnectModule(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( mod_id = start_mod_id; mod_id < MAX_MODULES; mod_id++) { - if( IsDaemon(mod_id)) { - if( shutdown_daemons) ShutdownModule(mod_id); - } else { - ShutdownModule(mod_id); + for (int uid = 0; uid < MAX_MODULES; uid++) + { + CModuleRecord *mod = &m_ConnectedModules[uid]; + if (mod->uid >= 0 && mod->ModuleID > 0) + { + if (mod->DaemonStatus) + { + if (shutdown_daemons) + ShutdownModule(mod->ModuleID); + } + else + { + ShutdownModule(mod->ModuleID); + } } } } -void -CMessageManager::AddSubscription( MODULE_ID module_id, MSG_TYPE message_type) +void CMessageManager::AddSubscription(CModuleRecord *mod, MSG_TYPE message_type) { MSG_TYPE mt; - if( ((message_type < 0) || (message_type > MAX_MESSAGE_TYPES)) && (message_type != ALL_MESSAGE_TYPES) ) + if ((mod != NULL) && (mod->uid >= 0)) { - //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; - } + 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( module_id); + 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) +void CMessageManager::RemoveSubscription(CModuleRecord *mod, 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); + + if ((mod != NULL) && (mod->uid >= 0)) + { + 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) +void CMessageManager::PauseSubscription(CModuleRecord *mod, MSG_TYPE message_type) { // Might wanna add checks here for valid module_id, connected module MSG_TYPE mt; - - switch( message_type) { + + switch (message_type) + { case ALL_MESSAGE_TYPES: - for( mt = 0; mt < MAX_MESSAGE_TYPES; mt++) { - GetSubscriberList( mt)->PauseSubscriber( mod_id); + for (mt = 0; mt < MAX_MESSAGE_TYPES; mt++) + { + GetSubscriberList(mt)->PauseSubscriber(mod->uid); } break; default: - GetSubscriberList( message_type)->PauseSubscriber( mod_id); + 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; - - switch( message_type) { + + switch (message_type) + { case ALL_MESSAGE_TYPES: - for( mt = 0; mt < MAX_MESSAGE_TYPES; mt++) { - GetSubscriberList( mt)->ResumeSubscriber( mod_id); + for (mt = 0; mt < MAX_MESSAGE_TYPES; mt++) + { + GetSubscriberList(mt)->ResumeSubscriber(mod->uid); } break; default: - GetSubscriberList( message_type)->ResumeSubscriber( mod_id); + GetSubscriberList(message_type)->ResumeSubscriber(mod->uid); } } -CSubscriberList * -CMessageManager::GetSubscriberList( MSG_TYPE message_type) +void CMessageManager::HandleConnect(CMessage *M, UPipe *pSourcePipe) { - if( message_type >= MAX_MESSAGE_TYPES || message_type < 0) - return &m_EmptySubscriberList; - else - return &(m_SubscribersToMessageType[message_type]); + 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(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); + } } -bool -CMessageManager::IsModuleSubscribed( MODULE_ID mod_id, MSG_TYPE message_type) +void CMessageManager::HandleConnectV2(CMessage *M, UPipe *pSourcePipe) { - if( message_type >= MAX_MESSAGE_TYPES || message_type < 0 || - mod_id < MID_MESSAGE_MANAGER || mod_id > MAX_MODULES ) - return false; - - return GetSubscriberList( message_type)->IsSubscribed(mod_id); + MDF_CONNECT_V2 connect_v2; + M->GetData((void *)&connect_v2); + 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); + } +} + +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::SendAcknowledge( MODULE_ID mod_id) +void CMessageManager::HandleModuleDisconnect(CMessage *M) { - DEBUG_TEXT_( "Sending ACK to module " << mod_id << "... "); - DispatchSignal( MT_ACKNOWLEDGE, mod_id); - DEBUG_TEXT( "Sent!"); + 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; + M->GetData(&module_ready); + + CModuleRecord *mod = GetRecord(M->src_mod_id); + if ((mod != NULL) && (mod->uid >= 0)) + { + mod->pid = module_ready.pid; + // Notify that the module has joined and ready + SendHello(mod); + } +} -int -CMessageManager::ModuleIsConnected( MODULE_ID mod_id) +void CMessageManager::HandleSubscribe(CMessage *M) { - if( mod_id < MID_MESSAGE_MANAGER) - return 0; - if( mod_id < MAX_MODULES) { - if( m_ConnectedModules[mod_id].ModuleID == mod_id) { - return 1; - } + MSG_TYPE msg_type_to_subscribe; + M->GetData(&msg_type_to_subscribe); + + CModuleRecord *mod = GetRecord(M->src_mod_id); + 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 << "... "); + 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 != NULL) && (mod->uid >= 0)) + { + RemoveSubscription(mod, msg_type_to_unsubscribe); + SendAcknowledge(mod); } - return 0; } -UPipe* -CMessageManager::GetModPipe( MODULE_ID mod_id) +void CMessageManager::HandlePauseSubscription(CMessage *M) { - if( !ModuleIsConnected( mod_id)) return NULL; - UPipe *mod_pipe = m_ConnectedModules[mod_id].pModulePipe; - return mod_pipe; + MSG_TYPE msg_type_to_pause; + M->GetData(&msg_type_to_pause); + CModuleRecord *mod = GetRecord(M->src_mod_id); + if ((mod != NULL) && (mod->uid >= 0)) + { + PauseSubscription(mod, msg_type_to_pause); + SendAcknowledge(mod); + } } -int -CMessageManager::IsDaemon( MODULE_ID mod_id) +void CMessageManager::HandleResumeSubscription(CMessage *M) { - if( !ModuleIsConnected( mod_id)) return 0; - if( m_ConnectedModules[mod_id].DaemonStatus) { - return 1; - } else { - return 0; + + MSG_TYPE msg_type_to_resume; + M->GetData(&msg_type_to_resume); + CModuleRecord *mod = GetRecord(M->src_mod_id); + if ((mod != NULL) && (mod->uid >= 0)) + { + ResumeSubscription(mod, msg_type_to_resume); + SendAcknowledge(mod); } } -void -CMessageManager::LogFailedMessage( CMessage *M, MODULE_ID mod_id) +CSubscriberList * +CMessageManager::GetSubscriberList(MSG_TYPE message_type) { - MDF_FAILED_MESSAGE data; - memset( &data, 0, sizeof(data)); - data.dest_mod_id = mod_id; - data.time_of_failure = GetAbsTime(); - memcpy( &data.msg_header, M, sizeof(RTMA_MSG_HEADER)); - CMessage F(MT_FAILED_MESSAGE, &data, sizeof(data)); - DispatchMessage(&F); + if (message_type >= MAX_MESSAGE_TYPES || message_type < 0) + return &m_EmptySubscriberList; + else + return &(m_SubscribersToMessageType[message_type]); +} - m_MessageCounts[MT_FAILED_MESSAGE]++; +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::SendMessageTiming() +void CMessageManager::SendAcknowledge(CModuleRecord *mod) { - MDF_TIMING_MESSAGE data; - memset( &data, 0, sizeof(data)); - data.send_time = GetAbsTime(); - for (int i=0;iuid >= 0)) { - data.timing[i] = m_MessageCounts[i]; - m_MessageCounts[i] = 0; + DEBUG_TEXT_("Sending ACK to module " << mod_id << "... "); + DispatchSignal(MT_ACKNOWLEDGE, mod); + DEBUG_TEXT("Sent!"); } +} - //add IsConnected stuff here - #ifdef __unix__ - data.ModulePID[0] = getpid(); //MM - #else - data.ModulePID[0] = _getpid(); //MM - #endif - for (int i=1; iuid >= 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); + } + } } +} - CMessage F(MT_TIMING_MESSAGE, &data, sizeof(data)); - DispatchMessage(&F); +int CMessageManager::ModuleIsConnected(MODULE_ID mod_id) +{ + if (mod_id < MID_MESSAGE_MANAGER) + return 0; + for (int i = 0; i < MAX_MODULES; i++) + { + if (m_ConnectedModules[i].ModuleID == mod_id) + { + return 1; + } + } + return 0; } + +void CMessageManager::LogFailedMessage(CMessage *M, MODULE_ID mod_id) +{ + MDF_FAILED_MESSAGE data; + memset(&data, 0, sizeof(data)); + data.dest_mod_id = mod_id; + data.time_of_failure = GetAbsTime(); + memcpy(&data.msg_header, M, sizeof(RTMA_MSG_HEADER)); + CMessage F(MT_FAILED_MESSAGE, &data, sizeof(data)); + DispatchMessage(&F); +} \ No newline at end of file diff --git a/src/modules/MessageManager/MessageManager.h b/src/modules/MessageManager/MessageManager.h index 49e921b..86ade55 100644 --- a/src/modules/MessageManager/MessageManager.h +++ b/src/modules/MessageManager/MessageManager.h @@ -19,17 +19,119 @@ class CModuleRecord UPipe *pModulePipe; 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) { + if (name != NULL) { + myfree(name); + } + name = (char*) malloc(MAX_NAME_LEN); + if (name != NULL) { + memcpy(name, src_name, MAX_NAME_LEN); + name[MAX_NAME_LEN - 1] = '\0'; + } + } + + 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)); + } + else { + hello->name[0] = '\0'; + } + + } + + void SetGoodbye(MDF_GOODBYE *goodbye) { + 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)); + } + else { + goodbye->name[0] = '\0'; + } + } + + void myfree(char* p) { + free(p); + } + void Reset( void) { + if (name != NULL) { + myfree(name); + name = NULL; + } + + if (addr != NULL) { + myfree(addr); + addr = NULL; + } + ModuleID = -1; pModulePipe = NULL; LoggerStatus = 0; DaemonStatus = 0; + AllowMultiple = 0; + pid = 0; + uid = -1; + + } + + ~CModuleRecord(){ + printf("~CModuleRecord(%d)\n", ModuleID); + Reset(); } }; @@ -124,7 +226,8 @@ class CList } }; -#define SUBSCRIBER_FLAG_PAUSE 0x01 +#define SUBSCRIBER_FLAG_PAUSE 0x01 + class CSubscriberList : protected CList { @@ -139,28 +242,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 +272,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; } @@ -181,7 +284,7 @@ class CSubscriberList : protected CList } 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 +294,8 @@ class CSubscriberList : protected CList return is_paused; } - MODULE_ID - GetFirstSubscriber( void) { + UID + GetFirstSubscriber(void) { m_CurrentItem = GetFirstItem(); if( m_CurrentItem == NULL) { return -1; @@ -201,8 +304,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 +315,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 +334,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,15 +376,11 @@ 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 - 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, 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,75 +394,100 @@ class CMessageManager : public UPipeAutoServer MODULE_ID ConnectModule( MODULE_ID module_id, UPipe *pSourcePipe, short logger_status, short daemon_status); - void - DisconnectModule( MODULE_ID module_id); + MODULE_ID + ConnectModuleV2( MODULE_ID module_id, UPipe *pSourcePipe, MDF_CONNECT_V2 *connect); - void - CleanUpModuleRecord( MODULE_ID module_id); + CModuleRecord* GetOpenRecord(); + + CModuleRecord* GetRecord(MODULE_ID module_id); void - ShutdownModule( MODULE_ID mod_id); + DisconnectModule(MODULE_ID module_id); void - ShutdownAllModules( int shutdown_RTMA=1, int shutdown_daemons=1); + CleanUpModuleRecord(CModuleRecord *mod); void - ShutdownStatusModule( void); + ShutdownModule(MODULE_ID module_id); - void - ShutdownLoggerModule( void); + void + ShutdownAllModules(int shutdown_RTMA=1, int shutdown_daemons=1); void - AddSubscription( MODULE_ID module_id, MSG_TYPE message_type); + AddSubscription(CModuleRecord *mod, MSG_TYPE message_type); void - RemoveSubscription( MODULE_ID module_id, MSG_TYPE message_type); + RemoveSubscription(CModuleRecord *mod, MSG_TYPE message_type); void - PauseSubscription( MODULE_ID mod_id, MSG_TYPE msg_type_to_pause); + PauseSubscription(CModuleRecord *mod, MSG_TYPE msg_type_to_pause); void - ResumeSubscription( MODULE_ID mod_id, MSG_TYPE msg_type_to_resume); + ResumeSubscription(CModuleRecord *mod, MSG_TYPE msg_type_to_resume); 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); + void + SendAcknowledge(CModuleRecord* mod); + void + SendIntroductions(MODULE_ID mod_id); + + void + SendHello(CModuleRecord *mod); + + void + SendGoodbye(CModuleRecord* mod); + + void + SendPing(MODULE_ID mod_id); + + void + HandleSetName(CMessage *m); + + void + HandleForceDisconnect(CMessage *m); + + void + HandleModuleDisconnect(CMessage *m); + + void + HandleModuleReady(CMessage *m); + + void + HandleSubscribe(CMessage *m); + + void + HandleUnsubscribe(CMessage *m); + + void + HandlePauseSubscription(CMessage *m); + + void + HandleResumeSubscription(CMessage *m); + + void + HandleConnect(CMessage *m, UPipe* pSourcePipe); + + void + HandleConnectV2(CMessage *m, UPipe* PSourcePipe); + 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); - - int - ModuleIsConnected( MODULE_ID mod_id); - - UPipe* - GetModPipe( MODULE_ID mod_id); + ForwardMessage(CMessage *m, CModuleRecord* mod); int - IsDaemon( MODULE_ID mod_id); - - 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 + ModuleIsConnected(MODULE_ID mod_id); void LogFailedMessage( CMessage *m, MODULE_ID mod_id); 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