clock_gettime(CLOCK_MONOTONIC, &end);
if (er.encoded == -1) {
mdclog_write(MDCLOG_ERR, "encoding of %s failed, %s", asn_DEF_E2SM_gNB_X2_eventTriggerDefinition.name, strerror(errno));
- } else if (er.encoded > buffer_size) {
+ } else if (er.encoded > (ssize_t)buffer_size) {
mdclog_write(MDCLOG_ERR, "Buffer of size %d is to small for %s", (int) buffer_size,
typeDescriptor->name);
} else if (mdclog_level_get() >= MDCLOG_DEBUG) {
}
PLMN_Identity_t *createPLMN_ID(const unsigned char *data) {
- printEntry("PLMN_Identity_t", __func__);
+ printEntry("PLMN_Identity_t", __func__)
PLMN_Identity_t *plmnId = calloc(1, sizeof(PLMN_Identity_t));
ASN_STRUCT_RESET(asn_DEF_PLMN_Identity, plmnId);
plmnId->size = 3;
}
ENB_ID_t *createENB_ID(ENB_ID_PR enbType, unsigned char *data) {
- printEntry("ENB_ID_t", __func__);
+ printEntry("ENB_ID_t", __func__)
ENB_ID_t *enb = calloc(1, sizeof(ENB_ID_t));
ASN_STRUCT_RESET(asn_DEF_ENB_ID, enb);
switch (enbType) {
}
GNB_ID_t *createGnb_id(const unsigned char *data, int numOfBits) {
- printEntry("GNB_ID_t", __func__);
+ printEntry("GNB_ID_t", __func__)
if (numOfBits < 22 || numOfBits > 32) {
mdclog_write(MDCLOG_ERR, "GNB_ID_t number of bits = %d, needs to be 22 .. 32", numOfBits);
return NULL;
gnb->present = GNB_ID_PR_gNB_ID;
gnb->choice.gNB_ID.size = numOfBits % 8 == 0 ? (unsigned int)(numOfBits / 8) : (unsigned int)(numOfBits / 8 + 1);
- gnb->choice.gNB_ID.bits_unused = gnb->choice.gNB_ID.size * 8 - numOfBits;
+ gnb->choice.gNB_ID.bits_unused = (int)gnb->choice.gNB_ID.size * 8 - numOfBits;
gnb->choice.gNB_ID.buf = calloc(1, gnb->choice.gNB_ID.size);
memcpy(gnb->choice.gNB_ID.buf, data, gnb->choice.gNB_ID.size);
gnb->choice.gNB_ID.buf[gnb->choice.gNB_ID.size - 1] =
}
GlobalENB_ID_t *createGlobalENB_ID(PLMN_Identity_t *plmnIdentity, ENB_ID_t *enbId) {
- printEntry("GlobalENB_ID_t", __func__);
+ printEntry("GlobalENB_ID_t", __func__)
GlobalENB_ID_t *genbId = calloc(1, sizeof(GlobalENB_ID_t));
ASN_STRUCT_RESET(asn_DEF_GlobalENB_ID, genbId);
memcpy(&genbId->pLMN_Identity, plmnIdentity, sizeof(PLMN_Identity_t));
}
GlobalGNB_ID_t *createGlobalGNB_ID(PLMN_Identity_t *plmnIdentity, GNB_ID_t *gnb) {
- printEntry("GlobalGNB_ID_t", __func__);
+ printEntry("GlobalGNB_ID_t", __func__)
GlobalGNB_ID_t *ggnbId = calloc(1, sizeof(GlobalGNB_ID_t));
ASN_STRUCT_RESET(asn_DEF_GlobalGNB_ID, ggnbId);
Interface_ID_t *createInterfaceIDForGnb(GlobalGNB_ID_t *gnb) {
- printEntry("Interface_ID_t", __func__);
+ printEntry("Interface_ID_t", __func__)
Interface_ID_t *interfaceId = calloc(1, sizeof(Interface_ID_t));
ASN_STRUCT_RESET(asn_DEF_Interface_ID, interfaceId);
}
Interface_ID_t *createInterfaceIDForEnb(GlobalENB_ID_t *enb) {
- printEntry("Interface_ID_t", __func__);
+ printEntry("Interface_ID_t", __func__)
Interface_ID_t *interfaceId = calloc(1, sizeof(Interface_ID_t));
ASN_STRUCT_RESET(asn_DEF_Interface_ID, interfaceId);
InterfaceMessageType_t *createInterfaceMessageInitiating(long procedureCode) {
- printEntry("InterfaceMessageType_t", __func__);
+ printEntry("InterfaceMessageType_t", __func__)
InterfaceMessageType_t *intMsgT = calloc(1, sizeof(InterfaceMessageType_t));
ASN_STRUCT_RESET(asn_DEF_InterfaceMessageType, intMsgT);
}
InterfaceMessageType_t *createInterfaceMessageSuccsesful(long procedureCode) {
- printEntry("InterfaceMessageType_t", __func__);
+ printEntry("InterfaceMessageType_t", __func__)
InterfaceMessageType_t *intMsgT = calloc(1, sizeof(InterfaceMessageType_t));
ASN_STRUCT_RESET(asn_DEF_InterfaceMessageType, intMsgT);
}
InterfaceMessageType_t *createInterfaceMessageUnsuccessful(long procedureCode) {
- printEntry("InterfaceMessageType_t", __func__);
+ printEntry("InterfaceMessageType_t", __func__)
InterfaceMessageType_t *intMsgT = calloc(1, sizeof(InterfaceMessageType_t));
ASN_STRUCT_RESET(asn_DEF_InterfaceMessageType, intMsgT);
}
InterfaceProtocolIE_Value_t *createInterfaceProtocolValueInt(long number) {
- printEntry("InterfaceProtocolIE_Value_t", __func__);
+ printEntry("InterfaceProtocolIE_Value_t", __func__)
InterfaceProtocolIE_Value_t *value = calloc(1, sizeof(InterfaceProtocolIE_Value_t));
ASN_STRUCT_RESET(asn_DEF_InterfaceProtocolIE_Value, value);
}
InterfaceProtocolIE_Value_t *createInterfaceProtocolValueEnum(long number) {
- printEntry("InterfaceProtocolIE_Value_t", __func__);
+ printEntry("InterfaceProtocolIE_Value_t", __func__)
InterfaceProtocolIE_Value_t *value = calloc(1, sizeof(InterfaceProtocolIE_Value_t));
ASN_STRUCT_RESET(asn_DEF_InterfaceProtocolIE_Value, value);
}
InterfaceProtocolIE_Value_t *createInterfaceProtocolValueBool(int val) {
- printEntry("InterfaceProtocolIE_Value_t", __func__);
+ printEntry("InterfaceProtocolIE_Value_t", __func__)
InterfaceProtocolIE_Value_t *value = calloc(1, sizeof(InterfaceProtocolIE_Value_t));
ASN_STRUCT_RESET(asn_DEF_InterfaceProtocolIE_Value, value);
InterfaceProtocolIE_Value_t *createInterfaceProtocolValueBitString(unsigned char *buf, int numOfBits) {
- printEntry("InterfaceProtocolIE_Value_t", __func__);
+ printEntry("InterfaceProtocolIE_Value_t", __func__)
size_t size = numOfBits % 8 == 0 ? (unsigned int)(numOfBits / 8) : (unsigned int)(numOfBits / 8 + 1);
if (strlen((const char *)buf) < size) {
mdclog_write(MDCLOG_ERR, "size of buffer is small : %d needs to be %d in %s", (int)strlen((const char *)buf), (int)size, __func__);
value->present = InterfaceProtocolIE_Value_PR_valueBitS;
value->choice.valueBitS.size = numOfBits % 8 == 0 ? (unsigned int)(numOfBits / 8) : (unsigned int)(numOfBits / 8 + 1);
value->choice.valueBitS.buf = calloc(1, value->choice.valueBitS.size);
- int bits_unused = value->choice.valueBitS.size * 8 - numOfBits;
+ int bits_unused = (int)value->choice.valueBitS.size * 8 - numOfBits;
value->choice.valueBitS.bits_unused = bits_unused;
memcpy(value->choice.valueBitS.buf, buf, value->choice.valueBitS.size);
InterfaceProtocolIE_Value_t *createInterfaceProtocolValueOCTETS(uint8_t *buf) {
- printEntry("InterfaceProtocolIE_Value_t", __func__);
+ printEntry("InterfaceProtocolIE_Value_t", __func__)
size_t size = strlen((const char *)buf);
InterfaceProtocolIE_Value_t *value = calloc(1, sizeof(InterfaceProtocolIE_Value_t));
ASN_STRUCT_RESET(asn_DEF_InterfaceProtocolIE_Value, value);
InterfaceProtocolIE_Item_t *createInterfaceProtocolIE_Item(long id, long test, InterfaceProtocolIE_Value_t *value) {
- printEntry("InterfaceProtocolIE_Item_t", __func__);
+ printEntry("InterfaceProtocolIE_Item_t", __func__)
if (test < InterfaceProtocolIE_Test_equal || test > InterfaceProtocolIE_Test_present) {
mdclog_write(MDCLOG_ERR, "InterfaceProtocolIE_Item_t test value is %ld, out of scope %d .. %d ",
test, InterfaceProtocolIE_Test_equal, InterfaceProtocolIE_Test_present);
ActionParameter_Value_t *createActionParameterValue_Int(long number) {
- printEntry("ActionParameter_Value_t", __func__);
+ printEntry("ActionParameter_Value_t", __func__)
ActionParameter_Value_t *value = calloc(1, sizeof(ActionParameter_Value_t));
ASN_STRUCT_RESET(asn_DEF_ActionParameter_Value, value);
}
ActionParameter_Value_t *createActionParameterValue_Enum(long number) {
- printEntry("ActionParameter_Value_t", __func__);
+ printEntry("ActionParameter_Value_t", __func__)
ActionParameter_Value_t *value = calloc(1, sizeof(ActionParameter_Value_t));
ASN_STRUCT_RESET(asn_DEF_ActionParameter_Value, value);
}
ActionParameter_Value_t *createActionParameterValue_Bool(int val) {
- printEntry("ActionParameter_Value_t", __func__);
+ printEntry("ActionParameter_Value_t", __func__)
ActionParameter_Value_t *value = calloc(1, sizeof(ActionParameter_Value_t));
ASN_STRUCT_RESET(asn_DEF_ActionParameter_Value, value);
ActionParameter_Value_t *createActionParameterValue_Bit_String(unsigned char *buf, int numOfBits) {
- printEntry("ActionParameter_Value_t", __func__);
+ printEntry("ActionParameter_Value_t", __func__)
size_t size = numOfBits % 8 == 0 ? (unsigned int)(numOfBits / 8) : (unsigned int)(numOfBits / 8 + 1);
if (strlen((const char *)buf) < size) {
mdclog_write(MDCLOG_ERR, "size of buffer is small : %d needs to be %d in %s", (int)strlen((const char *)buf), (int)size, __func__);
}
- int bits_unused = size * 8 - numOfBits;
+ int bits_unused = (int)size * 8 - numOfBits;
ActionParameter_Value_t *value = calloc(1, sizeof(ActionParameter_Value_t));
ASN_STRUCT_RESET(asn_DEF_ActionParameter_Value, value);
ActionParameter_Value_t *createActionParameterValue_OCTETS(uint8_t *buf) {
- printEntry("ActionParameter_Value_t", __func__);
+ printEntry("ActionParameter_Value_t", __func__)
size_t size = strlen((const char *)buf);
ActionParameter_Value_t *value = calloc(1, sizeof(ActionParameter_Value_t));
ASN_STRUCT_RESET(asn_DEF_ActionParameter_Value, value);
* @return ActionParameter_Value_t *
*/
ActionParameter_Value_t *createActionParameterValue_PRINTS(char *buf) {
- printEntry("ActionParameter_Value_t", __func__);
+ printEntry("ActionParameter_Value_t", __func__)
size_t size = strlen((const char *)buf);
ActionParameter_Value_t *value = calloc(1, sizeof(ActionParameter_Value_t));
ASN_STRUCT_RESET(asn_DEF_ActionParameter_Value, value);
}
ActionParameter_Item_t *creatActionParameter_Item(long id, ActionParameter_Value_t *val) {
- printEntry("ActionParameter_Item_t", __func__);
+ printEntry("ActionParameter_Item_t", __func__)
if (id < 0 || id > 255) {
mdclog_write(MDCLOG_ERR, "ActionParameter_Item_t id = %ld, values are 0 .. 255", id);
return NULL;
int listSize,
uint8_t *buffer,
size_t buffer_size) {
- printEntry("E2SM_gNB_X2_eventTriggerDefinition_t", __func__);
+ printEntry("E2SM_gNB_X2_eventTriggerDefinition_t", __func__)
if (direction < InterfaceDirection_incoming || direction > InterfaceDirection_outgoing) {
mdclog_write(MDCLOG_ERR, "E2SM_gNB_X2_eventTriggerDefinition_t direction = %ld, values are %d .. %d",
direction, InterfaceDirection_incoming, InterfaceDirection_outgoing);
if (mdclog_level_get() >= MDCLOG_INFO) {
uint8_t buf1[4096];
- asn_enc_rval_t er1;
+ //asn_enc_rval_t er1;
encodebuff(ATS_BASIC_XER, &asn_DEF_E2SM_gNB_X2_eventTriggerDefinition,
eventTrigger,
buf1,
size_t createActionDefinition(long styleId, ActionParameter_Item_t actionParamList[], int listSize,
uint8_t *buffer,
size_t buffer_size) {
- printEntry("E2SM_gNB_X2_actionDefinition_t", __func__);
+ printEntry("E2SM_gNB_X2_actionDefinition_t", __func__)
E2SM_gNB_X2_actionDefinition_t *actDef = calloc(1, sizeof(E2SM_gNB_X2_actionDefinition_t));
ASN_STRUCT_RESET(asn_DEF_E2SM_gNB_X2_actionDefinition, actDef);
if (mdclog_level_get() >= MDCLOG_INFO) {
uint8_t buf1[4096];
- asn_enc_rval_t er1;
+ //asn_enc_rval_t er1;
encodebuff(ATS_BASIC_XER, &asn_DEF_E2SM_gNB_X2_actionDefinition,
actDef,
buf1,
int size,
uint8_t *buffer,
size_t buffer_size) {
- printEntry("E2SM_gNB_X2_indicationHeader_t", __func__);
+ printEntry("E2SM_gNB_X2_indicationHeader_t", __func__)
if (direction < InterfaceDirection_incoming || direction > InterfaceDirection_outgoing) {
mdclog_write(MDCLOG_ERR, "E2SM_gNB_X2_indicationHeader_t direction = %ld, values are %d .. %d",
direction, InterfaceDirection_incoming, InterfaceDirection_outgoing);
if (mdclog_level_get() >= MDCLOG_INFO) {
uint8_t buf1[4096];
- asn_enc_rval_t er1;
+ //asn_enc_rval_t er1;
encodebuff(ATS_BASIC_XER, &asn_DEF_E2SM_gNB_X2_indicationHeader,
indiHead,
buf1,
size_t createE2SM_gNB_X2_indicationMessage(uint8_t *message, uint msgSize,
uint8_t *buffer,
size_t buffer_size) {
- printEntry("E2SM_gNB_X2_indicationMessage_t", __func__);
+ printEntry("E2SM_gNB_X2_indicationMessage_t", __func__)
if (msgSize <= 0) {
mdclog_write(MDCLOG_ERR, "E2SM_gNB_X2_indicationMessage_t failed messsage size = %d", msgSize);
return -1;
if (mdclog_level_get() >= MDCLOG_INFO) {
uint8_t buf1[4096];
- asn_enc_rval_t er1;
+ //asn_enc_rval_t er1;
encodebuff(ATS_BASIC_XER, &asn_DEF_E2SM_gNB_X2_indicationMessage,
indicationMessage,
buf1,
size_t createE2SM_gNB_X2_callProcessID(long callProcess_Id,
uint8_t *buffer,
size_t buffer_size) {
- printEntry("E2SM_gNB_X2_callProcessID_t", __func__);
+ printEntry("E2SM_gNB_X2_callProcessID_t", __func__)
E2SM_gNB_X2_callProcessID_t *callProcessId = calloc(1, sizeof(E2SM_gNB_X2_callProcessID_t));
ASN_STRUCT_RESET(asn_DEF_E2SM_gNB_X2_callProcessID, callProcessId);
if (mdclog_level_get() >= MDCLOG_INFO) {
uint8_t buf1[4096];
- asn_enc_rval_t er1;
+ //asn_enc_rval_t er1;
encodebuff(ATS_BASIC_XER, &asn_DEF_E2SM_gNB_X2_callProcessID,
callProcessId,
buf1,
size_t createE2SM_gNB_X2_controlHeader(Interface_ID_t *interfaceId, long direction,
uint8_t *buffer,
size_t buffer_size) {
- printEntry("E2SM_gNB_X2_controlHeader_t", __func__);
+ printEntry("E2SM_gNB_X2_controlHeader_t", __func__)
if (direction < InterfaceDirection_incoming || direction > InterfaceDirection_outgoing) {
mdclog_write(MDCLOG_ERR, "E2SM_gNB_X2_controlHeader_t direction = %ld, values are %d .. %d",
direction, InterfaceDirection_incoming, InterfaceDirection_outgoing);
if (mdclog_level_get() >= MDCLOG_INFO) {
uint8_t buf1[4096];
- asn_enc_rval_t er1;
+ //asn_enc_rval_t er1;
encodebuff(ATS_BASIC_XER, &asn_DEF_E2SM_gNB_X2_controlHeader,
controlHeader,
buf1,
size_t createE2SM_gNB_X2_controlMessage(uint8_t *message, uint msgSize,
uint8_t *buffer,
size_t buffer_size) {
- printEntry("E2SM_gNB_X2_controlMessage_t", __func__);
+ printEntry("E2SM_gNB_X2_controlMessage_t", __func__)
E2SM_gNB_X2_controlMessage_t *controlMsg = calloc(1, sizeof(E2SM_gNB_X2_controlMessage_t));
ASN_STRUCT_RESET(asn_DEF_E2SM_gNB_X2_controlMessage, controlMsg);
if (mdclog_level_get() >= MDCLOG_INFO) {
uint8_t buf1[4096];
- asn_enc_rval_t er1;
+ //asn_enc_rval_t er1;
encodebuff(ATS_BASIC_XER, &asn_DEF_E2SM_gNB_X2_controlMessage,
controlMsg,
buf1,
}
return len;
-}
\ No newline at end of file
+}
using namespace std::placeholders;
+using namespace boost::filesystem;
+
#ifdef __TRACING__
using namespace opentracing;
#endif
BOOST_LOG_INLINE_GLOBAL_LOGGER_DEFAULT(my_logger, src::logger_mt)
boost::shared_ptr<sinks::synchronous_sink<sinks::text_file_backend>> boostLogger;
+double cpuClock = 0.0;
+bool jsonTrace = true;
void init_log() {
mdclog_attr_t *attr;
mdclog_init(attr);
mdclog_attr_destroy(attr);
}
+auto start_time = std::chrono::high_resolution_clock::now();
+typedef std::chrono::duration<double, std::ratio<1,1>> seconds_t;
+double age() {
+ return seconds_t(std::chrono::high_resolution_clock::now() - start_time).count();
+}
+
+double approx_CPU_MHz(unsigned sleeptime) {
+ using namespace std::chrono_literals;
+ uint32_t aux = 0;
+ uint64_t cycles_start = rdtscp(aux);
+ double time_start = age();
+ std::this_thread::sleep_for(sleeptime * 1ms);
+ uint64_t elapsed_cycles = rdtscp(aux) - cycles_start;
+ double elapsed_time = age() - time_start;
+ return elapsed_cycles / elapsed_time;
+}
//std::atomic<int64_t> rmrCounter{0};
std::atomic<int64_t> num_of_messages{0};
+std::atomic<int64_t> num_of_XAPP_messages{0};
static long transactionCounter = 0;
int main(const int argc, char **argv) {
- sctp_params_t pSctpParams;
+ sctp_params_t sctpParams;
#ifdef __TRACING__
opentracing::Tracer::InitGlobal(tracelibcpp::createTracer("E2 Terminator"));
auto span = opentracing::Tracer::Global()->StartSpan(__FUNCTION__);
otSpan span = 0;
#endif
+ {
+ std::random_device device{};
+ std::mt19937 generator(device());
+ std::uniform_int_distribution<long> distribution(1, (long) 1e12);
+ transactionCounter = distribution(generator);
+ }
+
+ uint64_t st = 0,en = 0;
+ uint32_t aux1 = 0;
+ uint32_t aux2 = 0;
+ st = rdtscp(aux1);
+
unsigned num_cpus = std::thread::hardware_concurrency();
-#ifdef ERROR_LEVEL
- mdclog_severity_t loglevel = MDCLOG_ERR;
-#else
- mdclog_severity_t loglevel = MDCLOG_INFO;
-#endif
init_log();
- mdclog_level_set(loglevel);
+ mdclog_level_set(MDCLOG_INFO);
- if (argc < 7) {
- mdclog_mdc_add("app", argv[0]);
- mdclog_write(MDCLOG_ERR, "Usage nano <rmr port> logLevel <debug/warning/info/error> volume <PATH to log file location>");
- return -1;
+ cpuClock = approx_CPU_MHz(100);
+
+ mdclog_write(MDCLOG_ERR, "CPU speed %11.11f", cpuClock);
+ auto result = parse(argc, argv, sctpParams);
+
+ path p = (sctpParams.configFilePath + "/" + sctpParams.configFileName).c_str();
+ if (exists(p)) {
+ const int size = 2048;
+ auto fileSize = file_size(p);
+ if (fileSize > size) {
+ mdclog_write(MDCLOG_ERR, "File %s larger than %d", p.string().c_str(), size);
+ exit(-1);
+ }
+ } else {
+ mdclog_write(MDCLOG_ERR, "Configuration File %s not exists", p.string().c_str());
+ exit(-1);
}
- {
- std::random_device device{};
- std::mt19937 generator(device());
- std::uniform_int_distribution<long> distribution(1, (long) 1e12);
- transactionCounter = distribution(generator);
+ ReadConfigFile conf;
+ if (conf.openConfigFile(p.string()) == -1) {
+ mdclog_write(MDCLOG_ERR, "Filed to open config file %s, %s",
+ p.string().c_str(), strerror(errno));
+ exit(-1);
+ }
+ int rmrPort = conf.getIntValue("nano");
+ if (rmrPort == -1) {
+ mdclog_write(MDCLOG_ERR, "illigal RMR port ");
+ exit(-1);
+ }
+ sctpParams.rmrPort = (uint16_t)rmrPort;
+ snprintf(sctpParams.rmrAddress, sizeof(sctpParams.rmrAddress), "%d", (int) (sctpParams.rmrPort));
+
+ auto tmpStr = conf.getStringValue("loglevel");
+ if (tmpStr.length() == 0) {
+ mdclog_write(MDCLOG_ERR, "illigal loglevel. Set loglevel to MDCLOG_INFO");
+ tmpStr = "info";
+ }
+ transform(tmpStr.begin(), tmpStr.end(), tmpStr.begin(), ::tolower);
+
+ if ((tmpStr.compare("debug")) == 0) {
+ sctpParams.logLevel = MDCLOG_DEBUG;
+ } else if ((tmpStr.compare("info")) == 0) {
+ sctpParams.logLevel = MDCLOG_INFO;
+ } else if ((tmpStr.compare("warning")) == 0) {
+ sctpParams.logLevel = MDCLOG_WARN;
+ } else if ((tmpStr.compare("error")) == 0) {
+ sctpParams.logLevel = MDCLOG_ERR;
+ } else {
+ mdclog_write(MDCLOG_ERR, "illigal loglevel = %s. Set loglevel to MDCLOG_INFO", tmpStr.c_str());
+ sctpParams.logLevel = MDCLOG_INFO;
+ }
+ mdclog_level_set(sctpParams.logLevel);
+
+ tmpStr = conf.getStringValue("volume");
+ if (tmpStr.length() == 0) {
+ mdclog_write(MDCLOG_ERR, "illigal volume.");
+ exit(-1);
}
char tmpLogFilespec[VOLUME_URL_SIZE];
tmpLogFilespec[0] = 0;
- pSctpParams.volume[0] = 0;
- //read paramters from CLI
- for (auto i = 1; i < argc; i += 2) {
- char *dummy;
- if (strcasecmp("nano", argv[i]) == 0) {
- pSctpParams.rmrPort = (uint16_t) (uint16_t) strtol(argv[i + 1], &dummy, 10);
- } else if (strcasecmp("loglevel", argv[i]) == 0) {
- if (strcasecmp("debug", argv[i + 1]) == 0) {
- loglevel = MDCLOG_DEBUG;
- } else if (strcasecmp("info", argv[i + 1]) == 0) {
- loglevel = MDCLOG_INFO;
- } else if (strcasecmp("warning", argv[i + 1]) == 0) {
- loglevel = MDCLOG_WARN;
- } else if (strcasecmp("error", argv[i + 1]) == 0) {
- loglevel = MDCLOG_ERR;
- }
- } else if (strcasecmp("volume", argv[i]) == 0) {
- snprintf(pSctpParams.volume, VOLUME_URL_SIZE, "%s", argv[i + 1]);
- snprintf(tmpLogFilespec, VOLUME_URL_SIZE, "%s", argv[i + 1]);
- }
+ sctpParams.volume[0] = 0;
+ snprintf(sctpParams.volume, VOLUME_URL_SIZE, "%s", tmpStr.c_str());
+ // copy the name to temp file as well
+ snprintf(tmpLogFilespec, VOLUME_URL_SIZE, "%s", tmpStr.c_str());
+
+
+ // define the file name in the tmp directory under the volume
+ strcat(tmpLogFilespec,"/tmp/E2Term_%Y-%m-%d_%H-%M-%S.%N.tmpStr");
+
+ sctpParams.myIP = conf.getStringValue("local-ip");
+ if (sctpParams.myIP.length() == 0) {
+ mdclog_write(MDCLOG_ERR, "illigal local-ip.");
+ exit(-1);
}
+ tmpStr = conf.getStringValue("trace");
+ transform(tmpStr.begin(), tmpStr.end(), tmpStr.begin(), ::tolower);
+ if ((tmpStr.compare("start")) == 0) {
+ mdclog_write(MDCLOG_INFO, "Trace set to: start");
+ sctpParams.trace = true;
+ } else if ((tmpStr.compare("stop")) == 0) {
+ mdclog_write(MDCLOG_INFO, "Trace set to: stop");
+ sctpParams.trace = false;
+ }
+ jsonTrace = sctpParams.trace;
- pSctpParams.logLevel = loglevel;
- snprintf(pSctpParams.rmrAddress, sizeof(pSctpParams.rmrAddress) - 1, "%d", (int) (pSctpParams.rmrPort));
+ en = rdtscp(aux2);
- strcat(tmpLogFilespec,"/tmp/E2Term_%Y-%m-%d_%H-%M-%S.%N.log");
+ mdclog_write(MDCLOG_INFO, "start = %lx end = %lx diff = %lx\n", st, en, en - st);
+ mdclog_write(MDCLOG_INFO, "start high = %lx start lo = %lx end high = %lx end lo = %lx\n",
+ st >> 32, st & 0xFFFFFFFF, (int64_t)en >> 32, en & 0xFFFFFFFF);
+ mdclog_write(MDCLOG_INFO, "ellapsed time = %5.9f\n", (double)(en - st)/cpuClock);
if (mdclog_level_get() >= MDCLOG_INFO) {
- mdclog_mdc_add("RMR Port", to_string(pSctpParams.rmrPort).c_str());
- mdclog_mdc_add("LogLevel", to_string(pSctpParams.logLevel).c_str());
- mdclog_mdc_add("volume", pSctpParams.volume);
+ mdclog_mdc_add("RMR Port", to_string(sctpParams.rmrPort).c_str());
+ mdclog_mdc_add("LogLevel", to_string(sctpParams.logLevel).c_str());
+ mdclog_mdc_add("volume", sctpParams.volume);
mdclog_mdc_add("tmpLogFilespec", tmpLogFilespec);
+ mdclog_mdc_add("my ip", sctpParams.myIP.c_str());
mdclog_write(MDCLOG_INFO, "running parameters");
}
// Files written to the current working directory
boostLogger = logging::add_file_log(
- keywords::file_name = tmpLogFilespec,
+ keywords::file_name = tmpLogFilespec, // to temp directory
keywords::rotation_size = 10 * 1024 * 1024,
keywords::time_based_rotation = sinks::file::rotation_at_time_interval(posix_time::hours(1)),
keywords::format = "%Message%"
- //keywords::format = "[%TimeStamp%]: %Message%" // use each log with time stamp
+ //keywords::format = "[%TimeStamp%]: %Message%" // use each tmpStr with time stamp
);
// Setup a destination folder for collecting rotated (closed) files --since the same volumn can use rename()
boostLogger->locked_backend()->set_file_collector(sinks::file::make_collector(
- keywords::target = pSctpParams.volume
- //keywords::max_size = 16 * 1024 * 1024,
- //keywords::min_free_space = 100 * 1024 * 1024
+ keywords::target = sctpParams.volume
));
// Upon restart, scan the directory for files matching the file_name pattern
boostLogger->locked_backend()->scan_for_files();
- // Enable auto-flushing after each log record written
+ // Enable auto-flushing after each tmpStr record written
if (mdclog_level_get() >= MDCLOG_DEBUG) {
boostLogger->locked_backend()->auto_flush(true);
}
// start epoll
- pSctpParams.epoll_fd = epoll_create1(0);
- if (pSctpParams.epoll_fd == -1) {
+ sctpParams.epoll_fd = epoll_create1(0);
+ if (sctpParams.epoll_fd == -1) {
mdclog_write(MDCLOG_ERR, "failed to open epoll descriptor");
exit(-1);
}
- pSctpParams.rmrCtx = getRmrContext(pSctpParams.rmrAddress, &span);
- if (pSctpParams.rmrCtx == nullptr) {
- mdclog_write(MDCLOG_ERR, "Failed to initialize RMR");
- close(pSctpParams.epoll_fd);
+ getRmrContext(sctpParams, &span);
+ if (sctpParams.rmrCtx == nullptr) {
+ close(sctpParams.epoll_fd);
exit(-1);
}
- rmr_init_trace(pSctpParams.rmrCtx, 200);
- // get the RMR fd for the epoll
- pSctpParams.rmrListenFd = rmr_get_rcvfd(pSctpParams.rmrCtx);
- struct epoll_event event{};
- // add RMR fd to epoll
- event.events = (EPOLLIN);
- event.data.fd = pSctpParams.rmrListenFd;
- // add listening RMR FD to epoll
- if (epoll_ctl(pSctpParams.epoll_fd, EPOLL_CTL_ADD, pSctpParams.rmrListenFd, &event)) {
- mdclog_write(MDCLOG_ERR, "Failed to add RMR descriptor to epoll");
- close(pSctpParams.rmrListenFd);
- rmr_close(pSctpParams.rmrCtx);
- close(pSctpParams.epoll_fd);
+
+ if (buildInotify(sctpParams) == -1) {
+ close(sctpParams.rmrListenFd);
+ rmr_close(sctpParams.rmrCtx);
+ close(sctpParams.epoll_fd);
exit(-1);
- }
+ }
- pSctpParams.sctpMap = new mapWrapper();
+ sctpParams.sctpMap = new mapWrapper();
std::vector<std::thread> threads(num_cpus);
// std::vector<std::thread> threads;
num_cpus = 1;
for (unsigned int i = 0; i < num_cpus; i++) {
- threads[i] = std::thread(listener, &pSctpParams);
+ threads[i] = std::thread(listener, &sctpParams);
cpu_set_t cpuset;
CPU_ZERO(&cpuset);
if (rc != 0) {
mdclog_write(MDCLOG_ERR, "Error calling pthread_setaffinity_np: %d", rc);
}
+ }
-// threads.emplace_back(std::thread(listener, &pSctpParams));
+ //loop over term_init until first message from xApp
+ handleTermInit(sctpParams);
+
+ for (auto &t : threads) {
+ t.join();
}
+#ifdef __TRACING__
+ opentracing::Tracer::Global()->Close();
+#endif
+ return 0;
+}
+
+void handleTermInit(sctp_params_t &sctpParams) {
+ sendTermInit(sctpParams);
//send to e2 manager init of e2 term
//E2_TERM_INIT
+
+ int count = 0;
+ auto exitCond = true;
+ while (exitCond) {
+ auto xappMessages = num_of_XAPP_messages.load(std::memory_order_acquire);
+ if (xappMessages > 0) {
+ exitCond = false;
+ continue;
+ }
+ usleep(10000);
+ count++;
+ if (count % 100 == 0) {
+ mdclog_write(MDCLOG_ERR, "No messages from any xApp : %ld", xappMessages);
+ sendTermInit(sctpParams);
+ }
+ }
+}
+
+void sendTermInit(sctp_params_t &sctpParams) {
auto term_init = false;
- char buff[128]{};
- auto len = snprintf(buff, 128, "E2 terminator started");
- rmr_mbuf_t *msg = rmr_alloc_msg(pSctpParams.rmrCtx, 200);
+ char *buff = (char *)calloc(1, sctpParams.myIP.length());
+ auto len = snprintf(buff, sctpParams.myIP.length(), "%s:%d", (const char *)sctpParams.myIP.c_str(),sctpParams.rmrPort);
+ rmr_mbuf_t *msg = rmr_alloc_msg(sctpParams.rmrCtx, sctpParams.myIP.length());
auto count = 0;
while (!term_init) {
msg->mtype = E2_TERM_INIT;
static unsigned char tx[32];
auto txLen = snprintf((char *) tx, sizeof tx, "%15ld", transactionCounter++);
rmr_bytes2xact(msg, tx, txLen);
- msg = rmr_send_msg(pSctpParams.rmrCtx, msg);
+ msg = rmr_send_msg(sctpParams.rmrCtx, msg);
if (msg == nullptr) {
- msg = rmr_alloc_msg(pSctpParams.rmrCtx, 200);
+ msg = rmr_alloc_msg(sctpParams.rmrCtx, sctpParams.myIP.length());
} else if (msg->state == 0) {
term_init = true;
rmr_free_msg(msg);
count++;
}
- for (auto &t : threads) {
- t.join();
+ free(buff);
+}
+
+/**
+ *
+ * @param argc
+ * @param argv
+ * @param sctpParams
+ * @return
+ */
+cxxopts::ParseResult parse(int argc, char *argv[], sctp_params_t &sctpParams) {
+ cxxopts::Options options(argv[0], "e2 term help");
+ options.positional_help("[optional args]").show_positional_help();
+ options.allow_unrecognised_options().add_options()
+ ("p,path", "config file path", cxxopts::value<std::string>(sctpParams.configFilePath)->default_value("config"))
+ ("f,file", "config file name", cxxopts::value<std::string>(sctpParams.configFileName)->default_value("config.conf"))
+ ("h,help", "Print help");
+
+ auto result = options.parse(argc, argv);
+
+ if (result.count("help")) {
+ std::cout << options.help({""}) << std::endl;
+ exit(0);
}
+ return result;
+}
-#ifdef __TRACING__
- opentracing::Tracer::Global()->Close();
-#endif
+/**
+ *
+ * @param sctpParams
+ * @return -1 failed 0 success
+ */
+int buildInotify(sctp_params_t &sctpParams) {
+ sctpParams.inotifyFD = inotify_init1(IN_NONBLOCK);
+ if (sctpParams.inotifyFD == -1) {
+ mdclog_write(MDCLOG_ERR, "Failed to init inotify (inotify_init1) %s", strerror(errno));
+ close(sctpParams.rmrListenFd);
+ rmr_close(sctpParams.rmrCtx);
+ close(sctpParams.epoll_fd);
+ return -1;
+ }
+
+ sctpParams.inotifyWD = inotify_add_watch(sctpParams.inotifyFD,
+ (const char *)sctpParams.configFilePath.c_str(),
+ IN_OPEN | IN_CLOSE);
+ if (sctpParams.inotifyWD == -1) {
+ mdclog_write(MDCLOG_ERR, "Failed to add directory : %s to inotify (inotify_add_watch) %s",
+ sctpParams.configFilePath.c_str(),
+ strerror(errno));
+ close(sctpParams.inotifyFD);
+ return -1;
+ }
+
+ struct epoll_event event{};
+ event.events = (EPOLLIN);
+ event.data.fd = sctpParams.inotifyFD;
+ // add listening RMR FD to epoll
+ if (epoll_ctl(sctpParams.epoll_fd, EPOLL_CTL_ADD, sctpParams.inotifyFD, &event)) {
+ mdclog_write(MDCLOG_ERR, "Failed to add inotify FD to epoll");
+ close(sctpParams.inotifyFD);
+ return -1;
+ }
return 0;
}
otSpan span = 0;
#endif
int num_of_SCTP_messages = 0;
- int num_of_XAPP_messages = 0;
auto totalTime = 0.0;
mdclog_mdc_clean();
mdclog_level_set(params->logLevel);
clock_gettime(CLOCK_MONOTONIC, &message.message.time);
start.tv_sec = message.message.time.tv_sec;
start.tv_nsec = message.message.time.tv_nsec;
- if ((events[i].events & EPOLLERR) || (events[i].events & EPOLLHUP)) {
- if (events[i].data.fd != params->rmrListenFd) {
- auto *peerInfo = (ConnectedCU_t *)events[i].data.ptr;
- mdclog_write(MDCLOG_ERR, "epoll error, events %0x on fd %d, RAN NAME : %s",
- events[i].events, peerInfo->fileDescriptor, peerInfo->enodbName);
- rmrMessageBuffer.sendMessage->len = snprintf((char *)rmrMessageBuffer.sendMessage->payload, 256,
- "%s|Failed SCTP Connection",
- peerInfo->enodbName);
- message.message.asndata = rmrMessageBuffer.sendMessage->payload;
- message.message.asnLength = rmrMessageBuffer.sendMessage->len;
- memcpy(message.message.enodbName, peerInfo->enodbName, sizeof(peerInfo->enodbName));
- message.message.direction = 'N';
- if (sendRequestToXapp(message, RIC_SCTP_CONNECTION_FAILURE, rmrMessageBuffer, &span) != 0) {
- mdclog_write(MDCLOG_ERR, "SCTP_CONNECTION_FAIL message failed to send to xAPP");
- }
-
- close(peerInfo->fileDescriptor);
- cleanHashEntry((ConnectedCU_t *) events[i].data.ptr, params->sctpMap, &span);
- } else {
- mdclog_write(MDCLOG_ERR, "epoll error, events %0x on RMR FD", events[i].events);
- }
+ if ((events[i].events & EPOLLERR) || (events[i].events & EPOLLHUP)) {
+ handlepoll_error(events[i], message, rmrMessageBuffer, params, &span);
} else if (events[i].events & EPOLLOUT) {
- // this need to send waiting message from connection EINPROGRESS
- auto *peerInfo = (ConnectedCU_t *) events[i].data.ptr;
-
- memcpy(message.message.enodbName, peerInfo->enodbName, sizeof(peerInfo->enodbName));
-
- mdclog_write(MDCLOG_INFO, "file descriptor %d got EPOLLOUT", peerInfo->fileDescriptor);
- auto retVal = 0;
- socklen_t retValLen = 0;
- auto rc = getsockopt(peerInfo->fileDescriptor, SOL_SOCKET, SO_ERROR, &retVal, &retValLen);
- if (rc != 0 || retVal != 0) {
- if (rc != 0) {
- rmrMessageBuffer.sendMessage->len = snprintf((char *)rmrMessageBuffer.sendMessage->payload, 256,
- "%s|Failed SCTP Connection, after EINPROGRESS the getsockopt%s",
- peerInfo->enodbName, strerror(errno));
- } else if (retVal != 0) {
- rmrMessageBuffer.sendMessage->len = snprintf((char *)rmrMessageBuffer.sendMessage->payload, 256,
- "%s|Failed SCTP Connection after EINPROGRESS, SO_ERROR",
- peerInfo->enodbName);
- }
-
- message.message.asndata = rmrMessageBuffer.sendMessage->payload;
- message.message.asnLength = rmrMessageBuffer.sendMessage->len;
- mdclog_write(MDCLOG_ERR, "%s", rmrMessageBuffer.sendMessage->payload);
- message.message.direction = 'N';
- if (sendRequestToXapp(message, RIC_SCTP_CONNECTION_FAILURE, rmrMessageBuffer, &span) != 0) {
- mdclog_write(MDCLOG_ERR, "SCTP_CONNECTION_FAIL message failed to send to xAPP");
- }
- memset(peerInfo->asnData, 0, peerInfo->asnLength);
- peerInfo->asnLength = 0;
- peerInfo->mtype = 0;
- continue;
- }
-
- peerInfo->isConnected = true;
-
- if (modifyToEpoll(params->epoll_fd, peerInfo, (EPOLLIN | EPOLLET), params->sctpMap, peerInfo->enodbName,
- peerInfo->mtype, &span) != 0) {
- mdclog_write(MDCLOG_ERR, "epoll_ctl EPOLL_CTL_MOD");
- continue;
- }
-
- message.message.asndata = (unsigned char *)peerInfo->asnData;
- message.message.asnLength = peerInfo->asnLength;
- message.message.messageType = peerInfo->mtype;
- memcpy(message.message.enodbName, peerInfo->enodbName, sizeof(peerInfo->enodbName));
- num_of_messages.fetch_add(1, std::memory_order_release);
- if (mdclog_level_get() >= MDCLOG_DEBUG) {
- mdclog_write(MDCLOG_DEBUG, "send the delayed SETUP/ENDC SETUP to sctp for %s",
- message.message.enodbName);
- }
- if (sendSctpMsg(peerInfo, message, params->sctpMap, &span) != 0) {
- if (mdclog_level_get() >= MDCLOG_DEBUG) {
- mdclog_write(MDCLOG_DEBUG, "Error write to SCTP %s %d", __func__, __LINE__);
- }
- continue;
- }
-
- memset(peerInfo->asnData, 0, peerInfo->asnLength);
- peerInfo->asnLength = 0;
- peerInfo->mtype = 0;
-
+ handleEinprogressMessages(events[i], message, rmrMessageBuffer, params, &span);
} else if (params->rmrListenFd == events[i].data.fd) {
// got message from XAPP
- num_of_XAPP_messages++;
+ num_of_XAPP_messages.fetch_add(1, std::memory_order_release);
num_of_messages.fetch_add(1, std::memory_order_release);
if (mdclog_level_get() >= MDCLOG_DEBUG) {
mdclog_write(MDCLOG_DEBUG, "new message from RMR");
&span) != 0) {
mdclog_write(MDCLOG_ERR, "Error handling Xapp message");
}
+ } else if (params->inotifyFD == events[i].data.fd) {
+ mdclog_write(MDCLOG_INFO, "Got event from inotify (configuration update)");
+ handleConfigChange(params);
} else {
/* We RMR_ERR_RETRY have data on the fd waiting to be read. Read and display it.
* We must read whatever data is available completely, as we are running
#endif
}
+/**
+ *
+ * @param sctpParams
+ */
+void handleConfigChange(sctp_params_t *sctpParams) {
+ char buf[4096] __attribute__ ((aligned(__alignof__(struct inotify_event))));
+ const struct inotify_event *event;
+ char *ptr;
+
+ path p = (sctpParams->configFilePath + "/" + sctpParams->configFileName).c_str();
+ auto endlessLoop = true;
+ while (endlessLoop) {
+ auto len = read(sctpParams->inotifyFD, buf, sizeof buf);
+ if (len == -1) {
+ if (errno != EAGAIN) {
+ mdclog_write(MDCLOG_ERR, "read %s ", strerror(errno));
+ endlessLoop = false;
+ continue;
+ }
+ else {
+ endlessLoop = false;
+ continue;
+ }
+ }
+
+ for (ptr = buf; ptr < buf + len; ptr += sizeof(struct inotify_event) + event->len) {
+ event = (const struct inotify_event *)ptr;
+ if (event->mask & (uint32_t)IN_ISDIR) {
+ continue;
+ }
+
+ // the directory name
+ if (sctpParams->inotifyWD == event->wd) {
+ // not the directory
+ }
+ if (event->len) {
+ if (!(sctpParams->configFileName.compare(event->name))) {
+ continue;
+ }
+ }
+ // only the file we want
+ if (event->mask & (uint32_t)IN_CLOSE_WRITE) {
+ if (exists(p)) {
+ const int size = 2048;
+ auto fileSize = file_size(p);
+ if (fileSize > size) {
+ mdclog_write(MDCLOG_ERR, "File %s larger than %d", p.string().c_str(), size);
+ return;
+ }
+ } else {
+ mdclog_write(MDCLOG_ERR, "Configuration File %s not exists", p.string().c_str());
+ return;
+ }
+
+ ReadConfigFile conf;
+ if (conf.openConfigFile(p.string()) == -1) {
+ mdclog_write(MDCLOG_ERR, "Filed to open config file %s, %s",
+ p.string().c_str(), strerror(errno));
+ return;
+ }
+
+ auto tmpStr = conf.getStringValue("loglevel");
+ if (tmpStr.length() == 0) {
+ mdclog_write(MDCLOG_ERR, "illigal loglevel. Set loglevel to MDCLOG_INFO");
+ tmpStr = "info";
+ }
+ transform(tmpStr.begin(), tmpStr.end(), tmpStr.begin(), ::tolower);
+
+ if ((tmpStr.compare("debug")) == 0) {
+ mdclog_write(MDCLOG_INFO, "Log level set to MDCLOG_DEBUG");
+ sctpParams->logLevel = MDCLOG_DEBUG;
+ } else if ((tmpStr.compare("info")) == 0) {
+ mdclog_write(MDCLOG_INFO, "Log level set to MDCLOG_INFO");
+ sctpParams->logLevel = MDCLOG_INFO;
+ } else if ((tmpStr.compare("warning")) == 0) {
+ mdclog_write(MDCLOG_INFO, "Log level set to MDCLOG_WARN");
+ sctpParams->logLevel = MDCLOG_WARN;
+ } else if ((tmpStr.compare("error")) == 0) {
+ mdclog_write(MDCLOG_INFO, "Log level set to MDCLOG_ERR");
+ sctpParams->logLevel = MDCLOG_ERR;
+ } else {
+ mdclog_write(MDCLOG_ERR, "illigal loglevel = %s. Set loglevel to MDCLOG_INFO", tmpStr.c_str());
+ sctpParams->logLevel = MDCLOG_INFO;
+ }
+ mdclog_level_set(sctpParams->logLevel);
+
+
+ tmpStr = conf.getStringValue("trace");
+ if (tmpStr.length() == 0) {
+ mdclog_write(MDCLOG_ERR, "illigal trace. Set trace to stop");
+ tmpStr = "stop";
+ }
+
+ transform(tmpStr.begin(), tmpStr.end(), tmpStr.begin(), ::tolower);
+ if ((tmpStr.compare("start")) == 0) {
+ mdclog_write(MDCLOG_INFO, "Trace set to: start");
+ sctpParams->trace = true;
+ } else if ((tmpStr.compare("stop")) == 0) {
+ mdclog_write(MDCLOG_INFO, "Trace set to: stop");
+ sctpParams->trace = false;
+ } else {
+ mdclog_write(MDCLOG_ERR, "Trace was set to wrong value %s, set to stop", tmpStr.c_str());
+ sctpParams->trace = false;
+ }
+ jsonTrace = sctpParams->trace;
+ endlessLoop = false;
+ }
+ }
+ }
+}
+
+/**
+ *
+ * @param event
+ * @param message
+ * @param rmrMessageBuffer
+ * @param params
+ * @param pSpan
+ */
+void handleEinprogressMessages(struct epoll_event &event,
+ ReportingMessages_t &message,
+ RmrMessagesBuffer_t &rmrMessageBuffer,
+ sctp_params_t *params,
+ otSpan *pSpan) {
+#ifdef __TRACING__
+ auto lspan = opentracing::Tracer::Global()->StartSpan(
+ __FUNCTION__, { opentracing::ChildOf(&pSpan->get()->context()) });
+#else
+ otSpan lspan = 0;
+#endif
+ auto *peerInfo = (ConnectedCU_t *)event.data.ptr;
+ memcpy(message.message.enodbName, peerInfo->enodbName, sizeof(peerInfo->enodbName));
+
+ mdclog_write(MDCLOG_INFO, "file descriptor %d got EPOLLOUT", peerInfo->fileDescriptor);
+ auto retVal = 0;
+ socklen_t retValLen = 0;
+ auto rc = getsockopt(peerInfo->fileDescriptor, SOL_SOCKET, SO_ERROR, &retVal, &retValLen);
+ if (rc != 0 || retVal != 0) {
+ if (rc != 0) {
+ rmrMessageBuffer.sendMessage->len = snprintf((char *)rmrMessageBuffer.sendMessage->payload, 256,
+ "%s|Failed SCTP Connection, after EINPROGRESS the getsockopt%s",
+ peerInfo->enodbName, strerror(errno));
+ } else if (retVal != 0) {
+ rmrMessageBuffer.sendMessage->len = snprintf((char *)rmrMessageBuffer.sendMessage->payload, 256,
+ "%s|Failed SCTP Connection after EINPROGRESS, SO_ERROR",
+ peerInfo->enodbName);
+ }
+
+ message.message.asndata = rmrMessageBuffer.sendMessage->payload;
+ message.message.asnLength = rmrMessageBuffer.sendMessage->len;
+ mdclog_write(MDCLOG_ERR, "%s", rmrMessageBuffer.sendMessage->payload);
+ message.message.direction = 'N';
+ if (sendRequestToXapp(message, RIC_SCTP_CONNECTION_FAILURE, rmrMessageBuffer, &lspan) != 0) {
+ mdclog_write(MDCLOG_ERR, "SCTP_CONNECTION_FAIL message failed to send to xAPP");
+ }
+ memset(peerInfo->asnData, 0, peerInfo->asnLength);
+ peerInfo->asnLength = 0;
+ peerInfo->mtype = 0;
+#ifdef __TRACING__
+ lspan->Finish();
+#endif
+ return;
+ }
+
+ peerInfo->isConnected = true;
+
+ if (modifyToEpoll(params->epoll_fd, peerInfo, (EPOLLIN | EPOLLET), params->sctpMap, peerInfo->enodbName,
+ peerInfo->mtype, &lspan) != 0) {
+ mdclog_write(MDCLOG_ERR, "epoll_ctl EPOLL_CTL_MOD");
+#ifdef __TRACING__
+ lspan->Finish();
+#endif
+ return;
+ }
+
+ message.message.asndata = (unsigned char *)peerInfo->asnData;
+ message.message.asnLength = peerInfo->asnLength;
+ message.message.messageType = peerInfo->mtype;
+ memcpy(message.message.enodbName, peerInfo->enodbName, sizeof(peerInfo->enodbName));
+ num_of_messages.fetch_add(1, std::memory_order_release);
+ if (mdclog_level_get() >= MDCLOG_DEBUG) {
+ mdclog_write(MDCLOG_DEBUG, "send the delayed SETUP/ENDC SETUP to sctp for %s",
+ message.message.enodbName);
+ }
+ if (sendSctpMsg(peerInfo, message, params->sctpMap, &lspan) != 0) {
+ if (mdclog_level_get() >= MDCLOG_DEBUG) {
+ mdclog_write(MDCLOG_DEBUG, "Error write to SCTP %s %d", __func__, __LINE__);
+ }
+#ifdef __TRACING__
+ lspan->Finish();
+#endif
+ return;
+ }
+
+ memset(peerInfo->asnData, 0, peerInfo->asnLength);
+ peerInfo->asnLength = 0;
+ peerInfo->mtype = 0;
+#ifdef __TRACING__
+ lspan->Finish();
+#endif
+}
+
+
+void handlepoll_error(struct epoll_event &event,
+ ReportingMessages_t &message,
+ RmrMessagesBuffer_t &rmrMessageBuffer,
+ sctp_params_t *params,
+ otSpan *pSpan) {
+#ifdef __TRACING__
+ auto lspan = opentracing::Tracer::Global()->StartSpan(
+ __FUNCTION__, { opentracing::ChildOf(&pSpan->get()->context()) });
+#else
+ otSpan lspan = 0;
+#endif
+ if (event.data.fd != params->rmrListenFd) {
+ auto *peerInfo = (ConnectedCU_t *)event.data.ptr;
+ mdclog_write(MDCLOG_ERR, "epoll error, events %0x on fd %d, RAN NAME : %s",
+ event.events, peerInfo->fileDescriptor, peerInfo->enodbName);
+
+ rmrMessageBuffer.sendMessage->len = snprintf((char *)rmrMessageBuffer.sendMessage->payload, 256,
+ "%s|Failed SCTP Connection",
+ peerInfo->enodbName);
+ message.message.asndata = rmrMessageBuffer.sendMessage->payload;
+ message.message.asnLength = rmrMessageBuffer.sendMessage->len;
+
+ memcpy(message.message.enodbName, peerInfo->enodbName, sizeof(peerInfo->enodbName));
+ message.message.direction = 'N';
+ if (sendRequestToXapp(message, RIC_SCTP_CONNECTION_FAILURE, rmrMessageBuffer, &lspan) != 0) {
+ mdclog_write(MDCLOG_ERR, "SCTP_CONNECTION_FAIL message failed to send to xAPP");
+ }
+
+ close(peerInfo->fileDescriptor);
+ cleanHashEntry((ConnectedCU_t *) event.data.ptr, params->sctpMap, &lspan);
+ } else {
+ mdclog_write(MDCLOG_ERR, "epoll error, events %0x on RMR FD", event.events);
+ }
+#ifdef __TRACING__
+ lspan->Finish();
+#endif
+
+}
/**
*
* @param socket
static unsigned char tx[32];
snprintf((char *) tx, sizeof tx, "%15ld", transactionCounter++);
rmr_bytes2xact(rmrMessageBuffer.sendMessage, tx, strlen((const char *) tx));
- rmr_bytes2meid(rmrMessageBuffer.sendMessage, (unsigned char *)message.message.enodbName, strlen(message.message.enodbName));
+ rmr_bytes2meid(rmrMessageBuffer.sendMessage,
+ (unsigned char *)message.message.enodbName,
+ strlen(message.message.enodbName));
sendRmrMessage(rmrMessageBuffer, message, &lspan);
messageSent = true;
}
-void *getRmrContext(char *rmrAddress, otSpan *pSpan) {
+void getRmrContext(sctp_params_t &pSctpParams, otSpan *pSpan) {
#ifdef __TRACING__
auto lspan = opentracing::Tracer::Global()->StartSpan(
__FUNCTION__, { opentracing::ChildOf(&pSpan->get()->context()) });
#else
// otSpan lspan = 0;
#endif
- void *rmrCtx = rmr_init(rmrAddress, RMR_MAX_RCV_BYTES, RMRFL_NONE);
-
-
- if (rmrCtx == nullptr) {
- mdclog_write(MDCLOG_ERR, "RMR failed to initialise : %s", strerror(errno));
+ pSctpParams.rmrCtx = nullptr;
+ pSctpParams.rmrCtx = rmr_init(pSctpParams.rmrAddress, RMR_MAX_RCV_BYTES, RMRFL_NONE);
+ if (pSctpParams.rmrCtx == nullptr) {
+ mdclog_write(MDCLOG_ERR, "Failed to initialize RMR");
#ifdef __TRACING__
lspan->Finish();
#endif
-
- return (nullptr);
+ return;
}
- rmr_set_stimeout(rmrCtx, 0); // disable retries for any send operation
+ rmr_set_stimeout(pSctpParams.rmrCtx, 0); // disable retries for any send operation
// we need to find that routing table exist and we can run
if (mdclog_level_get() >= MDCLOG_INFO) {
mdclog_write(MDCLOG_INFO, "We are after RMR INIT wait for RMR_Ready");
int rmrReady = 0;
int count = 0;
while (!rmrReady) {
- if ((rmrReady = rmr_ready(rmrCtx)) == 0) {
+ if ((rmrReady = rmr_ready(pSctpParams.rmrCtx)) == 0) {
sleep(1);
}
count++;
#ifdef __TRACING__
lspan->Finish();
#endif
-
- return rmrCtx;
+ rmr_init_trace(pSctpParams.rmrCtx, 200);
+ // get the RMR fd for the epoll
+ pSctpParams.rmrListenFd = rmr_get_rcvfd(pSctpParams.rmrCtx);
+ struct epoll_event event{};
+ // add RMR fd to epoll
+ event.events = (EPOLLIN);
+ event.data.fd = pSctpParams.rmrListenFd;
+ // add listening RMR FD to epoll
+ if (epoll_ctl(pSctpParams.epoll_fd, EPOLL_CTL_ADD, pSctpParams.rmrListenFd, &event)) {
+ mdclog_write(MDCLOG_ERR, "Failed to add RMR descriptor to epoll");
+ close(pSctpParams.rmrListenFd);
+ rmr_close(pSctpParams.rmrCtx);
+ pSctpParams.rmrCtx = nullptr;
+ }
}
/**
}
void buildJsonMessage(ReportingMessages_t &message) {
- message.outLen = sizeof(message.base64Data);
- base64::encode((const unsigned char *)message.message.asndata,
- (const int)message.message.asnLength,
- message.base64Data,
- message.outLen);
- if (mdclog_level_get() >= MDCLOG_DEBUG) {
- mdclog_write(MDCLOG_DEBUG, "asn data length = %d, base64 message length = %d ",
- (int)message.message.asnLength,
- (int)message.outLen);
- }
+ if (jsonTrace) {
+ message.outLen = sizeof(message.base64Data);
+ base64::encode((const unsigned char *) message.message.asndata,
+ (const int) message.message.asnLength,
+ message.base64Data,
+ message.outLen);
+ if (mdclog_level_get() >= MDCLOG_DEBUG) {
+ mdclog_write(MDCLOG_DEBUG, "asn data length = %d, base64 message length = %d ",
+ (int) message.message.asnLength,
+ (int) message.outLen);
+ }
// char buff[256];
// // build day time to seconds from epoc
// // add nanosecond
// snprintf(buff, sizeof buff, "%s.%09ld UTC\n", buff, message.message.time.tv_nsec);
- message.bufferLen = snprintf(message.buffer, sizeof(message.buffer),
- "{\"header\": {\"ts\": \"%ld.%09ld\","
- "\"ranName\": \"%s\","
- "\"messageType\": %d,"
- "\"direction\": \"%c\"},"
- "\"base64Length\": %d,"
- "\"asnBase64\": \"%s\"}",
- message.message.time.tv_sec,
- message.message.time.tv_nsec,
- message.message.enodbName,
- message.message.messageType,
- message.message.direction,
- (int)message.outLen,
- message.base64Data);
- static src::logger_mt& lg = my_logger::get();
-
- BOOST_LOG(lg) << message.buffer;
-
+ message.bufferLen = snprintf(message.buffer, sizeof(message.buffer),
+ "{\"header\": {\"ts\": \"%ld.%09ld\","
+ "\"ranName\": \"%s\","
+ "\"messageType\": %d,"
+ "\"direction\": \"%c\"},"
+ "\"base64Length\": %d,"
+ "\"asnBase64\": \"%s\"}",
+ message.message.time.tv_sec,
+ message.message.time.tv_nsec,
+ message.message.enodbName,
+ message.message.messageType,
+ message.message.direction,
+ (int) message.outLen,
+ message.base64Data);
+ static src::logger_mt &lg = my_logger::get();
+
+ BOOST_LOG(lg) << message.buffer;
+ }
}
Code Review / ric-plt / e2.git / commitdiff