X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;f=RIC-E2-TERMINATION%2FsctpThread.h;h=10f572bdd6514e48d289f9309f793f252d585b93;hb=c40ea1f7632d62130943768c82bb4a72e6780073;hp=419a2f64fddabca9ddb4bfce84793a0eb25c61a9;hpb=3aec07fc2d80d69667b40ff0a69817646211ea31;p=ric-plt%2Fe2.git

diff --git a/RIC-E2-TERMINATION/sctpThread.h b/RIC-E2-TERMINATION/sctpThread.h
index 419a2f6..10f572b 100644
--- a/RIC-E2-TERMINATION/sctpThread.h
+++ b/RIC-E2-TERMINATION/sctpThread.h
@@ -15,6 +15,11 @@
  * limitations under the License.
 */
 
+/*
+ * This source code is part of the near-RT RIC (RAN Intelligent Controller)
+ * platform project (RICP).
+ */
+
 #ifndef X2_SCTP_THREAD_H
 #define X2_SCTP_THREAD_H
 
@@ -43,6 +48,9 @@
 #include <shared_mutex>
 #include <iterator>
 #include <map>
+#include <sys/inotify.h>
+#include <csignal>
+#include <future>
 
 #include <rmr/rmr.h>
 #include <rmr/RIC_message_types.h>
@@ -58,26 +66,38 @@
 #include <boost/log/sources/global_logger_storage.hpp>
 #include <boost/log/utility/setup/file.hpp>
 #include <boost/log/utility/setup/common_attributes.hpp>
-
+#include <boost/filesystem.hpp>
 
 #include <mdclog/mdclog.h>
 
-#include "3rdparty/asn1cFiles/E2AP-PDU.h"
-#include <3rdparty/asn1cFiles/ProtocolIE-Container.h>
-#include "3rdparty/asn1cFiles/InitiatingMessage.h"
-#include "3rdparty/asn1cFiles/SuccessfulOutcome.h"
-#include "3rdparty/asn1cFiles/UnsuccessfulOutcome.h"
-#include "3rdparty/asn1cFiles/ProtocolIE-Container.h"
-#include "3rdparty/asn1cFiles/ProtocolIE-Field.h"
-
-#ifdef __TRACING__
-#include "openTracing.h"
-#endif
+#include "oranE2/E2AP-PDU.h"
+#include "oranE2/ProtocolIE-Container.h"
+#include "oranE2/InitiatingMessage.h"
+#include "oranE2/SuccessfulOutcome.h"
+#include "oranE2/UnsuccessfulOutcome.h"
+#include "oranE2/ProtocolIE-Container.h"
+#include "oranE2/ProtocolIE-Field.h"
+#include "oranE2/GlobalE2node-gNB-ID.h"
+#include "oranE2/GlobalE2node-en-gNB-ID.h"
+#include "oranE2/GlobalE2node-ng-eNB-ID.h"
+#include "oranE2/GlobalE2node-eNB-ID.h"
+
+#include "cxxopts.hpp"
+//#include "config-cpp/include/config-cpp/config-cpp.h"
+#include <zlib.h>
+#include <prometheus/counter.h>
+#include <prometheus/exposer.h>
+#include <prometheus/gateway.h>
+#include <prometheus/registry.h>
+
+using namespace prometheus;
 
 #include "mapWrapper.h"
 
 #include "base64.h"
 
+#include "ReadConfigFile.h"
+
 using namespace std;
 namespace logging = boost::log;
 namespace src = boost::log::sources;
@@ -92,31 +112,57 @@ namespace expr = boost::log::expressions;
 
 #define MAXEVENTS 128
 
-#define RECEIVE_SCTP_BUFFER_SIZE (64*1024)
-#define RECEIVE_XAPP_BUFFER_SIZE RECEIVE_SCTP_BUFFER_SIZE 
+#define RECEIVE_SCTP_BUFFER_SIZE (256 * 1024)
+#define RECEIVE_XAPP_BUFFER_SIZE RECEIVE_SCTP_BUFFER_SIZE
 
 typedef mapWrapper Sctp_Map_t;
 
-#ifdef __TRACING__
-typedef const std::unique_ptr<opentracing::Span> otSpan;
-#else
-typedef const int otSpan;
-#endif
+
 
 #define VOLUME_URL_SIZE 256
+#define KA_MESSAGE_SIZE 2048
 
 typedef struct sctp_params {
+    int      epollTimeOut = -1;
     uint16_t rmrPort = 0;
+    uint16_t sctpPort = SRC_PORT;
     int      epoll_fd = 0;
+    int      listenFD = 0;
     int      rmrListenFd = 0;
+    int      inotifyFD = 0;
+    int      inotifyWD = 0;
     void     *rmrCtx = nullptr;
     Sctp_Map_t *sctpMap = nullptr;
-    char       rmrAddress[256] {}; // "tcp:portnumber" "tcp:5566" listen to all address on port 5566
+    char      ka_message[KA_MESSAGE_SIZE] {};
+    int       ka_message_length = 0;
+    char       rmrAddress[256] {}; // "tcp:port number" "tcp:5566" listen to all address on port 5566
     mdclog_severity_t logLevel = MDCLOG_INFO;
     char volume[VOLUME_URL_SIZE];
-    //shared_timed_mutex fence; // moved to mapWrapper
+    string myIP {};
+    string fqdn {};
+    string podName {};
+    string configFilePath {};
+    string configFileName {};
+    bool trace = true;
+    shared_ptr<prometheus::Registry> prometheusRegistry;
+    string prometheusPort {"8088"};
+    Family<Counter> *prometheusFamily;
+    Exposer *prometheusExposer = nullptr;
 } sctp_params_t;
 
+// RAN to RIC
+#define IN_INITI 0 //INITIATING
+#define IN_SUCC 1 //SUCCESSFUL
+#define IN_UN_SUCC 2 //UN-Successful
+
+// RIC To RAN
+#define OUT_INITI 3 //INITIATING
+#define OUT_SUCC 4 //SUCCESSFUL
+#define OUT_UN_SUCC 5 //UN-Successful
+
+#define MSG_COUNTER 0
+#define BYTES_COUNTER 1
+
 typedef struct ConnectedCU {
     int fileDescriptor = 0;
     char hostName[NI_MAXHOST] {};
@@ -126,15 +172,21 @@ typedef struct ConnectedCU {
     size_t asnLength = 0;
     int mtype = 0;
     bool isConnected = false;
+    bool gotSetup = false;
+    sctp_params_t *sctpParams = nullptr;
+    Counter *counters[6][2][ProcedureCode_id_RICsubscriptionDelete + 1] {};
 } ConnectedCU_t ;
 
-#define MAX_RMR_BUFF_ARRY 32
+
+#define MAX_RMR_BUFF_ARRAY 32
 typedef struct RmrMessagesBuffer {
-    void *rmrCtx;
-    rmr_mbuf_t *sendMessage;
-    rmr_mbuf_t *sendBufferedMessages[MAX_RMR_BUFF_ARRY];
-    rmr_mbuf_t *rcvMessage;
-    rmr_mbuf_t *rcvBufferedMessages[MAX_RMR_BUFF_ARRY];
+    char ka_message[KA_MESSAGE_SIZE] {};
+    int  ka_message_len = 0;
+    void *rmrCtx = nullptr;
+    rmr_mbuf_t *sendMessage= nullptr;
+    //rmr_mbuf_t *sendBufferedMessages[MAX_RMR_BUFF_ARRAY] {};
+    rmr_mbuf_t *rcvMessage= nullptr;
+    //rmr_mbuf_t *rcvBufferedMessages[MAX_RMR_BUFF_ARRAY] {};
 } RmrMessagesBuffer_t;
 
 typedef struct formatedMessage {
@@ -147,53 +199,68 @@ typedef struct formatedMessage {
 } FormatedMessage_t;
 
 typedef struct ReportingMessages {
-    FormatedMessage_t message;
-    int outLen;
-    unsigned char base64Data[RECEIVE_SCTP_BUFFER_SIZE * 2];
-    char buffer[RECEIVE_SCTP_BUFFER_SIZE * 8];
-    size_t bufferLen;
+    FormatedMessage_t message {};
+    ConnectedCU_t *peerInfo = nullptr;
+    long outLen = 0;
+    unsigned char base64Data[RECEIVE_SCTP_BUFFER_SIZE * 2] {};
+    char buffer[RECEIVE_SCTP_BUFFER_SIZE * 8] {};
 } ReportingMessages_t;
 
+cxxopts::ParseResult parse(int argc, char *argv[], sctp_params_t &pSctpParams);
+
+int buildInotify(sctp_params_t &sctpParams);
+
+void handleTermInit(sctp_params_t &sctpParams);
+
+void handleConfigChange(sctp_params_t *sctpParams);
 
 void listener(sctp_params_t *params);
 
+void sendTermInit(sctp_params_t &sctpParams);
+
 int setSocketNoBlocking(int socket);
 
-void cleanHashEntry(ConnectedCU_t *peerInfo, Sctp_Map_t *m, otSpan *pSpan);
+void handleEinprogressMessages(struct epoll_event &event,
+                               ReportingMessages_t &message,
+                               RmrMessagesBuffer_t &rmrMessageBuffer,
+                               sctp_params_t *params);
+
+void handlepoll_error(struct epoll_event &event,
+                      ReportingMessages_t &message,
+                      RmrMessagesBuffer_t &rmrMessageBuffer,
+                      sctp_params_t *params);
+
+
+void cleanHashEntry(ConnectedCU_t *peerInfo, Sctp_Map_t *m);
 
-int getSetupRequestMetaData(ReportingMessages_t &message, char *data, char *host, uint16_t &port, otSpan *pSpan);
 
 /**
  *
  * @param message
  * @param rmrMessageBuffer
- * @param pSpan
  */
-void getRequestMetaData(ReportingMessages_t &message, RmrMessagesBuffer_t &rmrMessageBuffer, otSpan *pSpan);
+void getRequestMetaData(ReportingMessages_t &message, RmrMessagesBuffer_t &rmrMessageBuffer);
 
 /**
  *
  * @param sctpMap
- * @param messagBuffer
+ * @param messageBuffer
  * @param message
  * @param failedMesgId
- * @param pSpan
  * @return
  */
 int sendMessagetoCu(Sctp_Map_t *sctpMap,
-                    RmrMessagesBuffer_t &messagBuffer,
+                    RmrMessagesBuffer_t &messageBuffer,
                     ReportingMessages_t &message,
-                    int failedMesgId, otSpan *pSpan);
+                    int failedMesgId);
 
 void sendFailedSendingMessagetoXapp(RmrMessagesBuffer_t &rmrMessageBuffer,
                                     ReportingMessages_t &message,
-                                    int failedMesgId,
-                                    otSpan *pSpan);
+                                    int failedMesgId);
 
 int sendRequestToXapp(ReportingMessages_t &message,
                       int requestId,
-                      RmrMessagesBuffer_t &rmrMmessageBuffer,
-                      otSpan *pSpan);
+                      RmrMessagesBuffer_t &rmrMmessageBuffer);
 
 /**
  *
@@ -202,28 +269,26 @@ int sendRequestToXapp(ReportingMessages_t &message,
  * @param requestType
  * @param rmrMessageBuffer
  * @param sctpMap
- * @param pSpan
  * @return
  */
+/*
 int sendResponseToXapp(ReportingMessages_t &message,
                        int msgType,
                        int requestType,
                        RmrMessagesBuffer_t &rmrMessageBuffer,
-                       Sctp_Map_t *sctpMap,
-                       otSpan *pSpan);
+                       Sctp_Map_t *sctpMap);
+*/
 
 /**
  *
  * @param peerInfo
  * @param message
  * @param m
- * @param pSpan
  * @return
  */
 int sendSctpMsg(ConnectedCU_t *peerInfo,
                 ReportingMessages_t &message,
-                Sctp_Map_t *m,
-                otSpan *pSpan);
+                Sctp_Map_t *m);
 
 /**
  *
@@ -232,114 +297,83 @@ int sendSctpMsg(ConnectedCU_t *peerInfo,
  * @param numOfMessages
  * @param rmrMessageBuffer
  * @param ts
- * @param pSpan
  * @return
  */
 int receiveDataFromSctp(struct epoll_event *events,
                         Sctp_Map_t *sctpMap,
                         int &numOfMessages,
                         RmrMessagesBuffer_t &rmrMessageBuffer,
-                        struct timespec &ts,
-                        otSpan *pSpan);
+                        struct timespec &ts);
 
 /**
  *
  * @param rmrAddress
- * @param pSpan
  * @return
  */
-void *getRmrContext(char *rmrAddress, otSpan *pSpan);
+void getRmrContext(sctp_params_t &pSctpParams);
 
 /**
  *
- * @param epoll_fd
- * @param rmrCtx
  * @param sctpMap
- * @param messagBuffer
- * @param pSpan
- * @return
- */
-int receiveXappMessages(int epoll_fd,
-                        Sctp_Map_t *sctpMap,
-                        RmrMessagesBuffer_t &rmrMessageBuffer,
-                        struct timespec &ts,
-                        otSpan *pSpan);
-
-/**
- *
  * @param rmrMessageBuffer
- * @param message
- * @param epoll_fd
- * @param sctpMap
- * @param pSpan
+ * @param ts
  * @return
  */
-int connectToCUandSetUp(RmrMessagesBuffer_t &rmrMessageBuffer,
-                           ReportingMessages_t &message,
-                           int epoll_fd,
-                           Sctp_Map_t *sctpMap,
-                           otSpan *pSpan);
+int receiveXappMessages(Sctp_Map_t *sctpMap,
+                        RmrMessagesBuffer_t &rmrMessageBuffer,
+                        struct timespec &ts);
 
 /**
  *
- * @param messagBuffer
+ * @param messageBuffer
  * @param failedMsgId
  * @param sctpMap
- * @param pSpan
  * @return
  */
-int sendDirectionalSctpMsg(RmrMessagesBuffer_t &messagBuffer,
+int sendDirectionalSctpMsg(RmrMessagesBuffer_t &messageBuffer,
                            ReportingMessages_t &message,
                            int failedMsgId,
-                           Sctp_Map_t *sctpMap,
-                           otSpan *pSpan);
+                           Sctp_Map_t *sctpMap);
 /**
  *
  * @param pdu
  * @param message
  * @param rmrMessageBuffer
- * @param pSpan
  */
 void asnInitiatingRequest(E2AP_PDU_t *pdu,
+                          Sctp_Map_t *sctpMap,
                           ReportingMessages_t &message,
-                          RmrMessagesBuffer_t &rmrMessageBuffer,
-                          otSpan *pSpan);
+                          RmrMessagesBuffer_t &rmrMessageBuffer);
 /**
  *
  * @param pdu
  * @param message
  * @param sctpMap
  * @param rmrMessageBuffer
- * @param pSpan
  */
-void asnSuccsesfulMsg(E2AP_PDU_t *pdu,
-                      ReportingMessages_t &message,
+void asnSuccessfulMsg(E2AP_PDU_t *pdu,
                       Sctp_Map_t *sctpMap,
-                      RmrMessagesBuffer_t &rmrMessageBuffer,
-                      otSpan *pSpan);
+                      ReportingMessages_t &message,
+                      RmrMessagesBuffer_t &rmrMessageBuffer);
 /**
  *
  * @param pdu
  * @param message
  * @param sctpMap
  * @param rmrMessageBuffer
- * @param pSpan
  */
 void asnUnSuccsesfulMsg(E2AP_PDU_t *pdu,
-                        ReportingMessages_t &message,
                         Sctp_Map_t *sctpMap,
-                        RmrMessagesBuffer_t &rmrMessageBuffer,
-                        otSpan *pSpan);
+                        ReportingMessages_t &message,
+                        RmrMessagesBuffer_t &rmrMessageBuffer);
 
 /**
  *
  * @param rmrMessageBuffer
  * @param message
- * @param pSpan
  * @return
  */
-int sendRmrMessage(RmrMessagesBuffer_t &rmrMessageBuffer, ReportingMessages_t &message, otSpan *pSpan);
-
+int sendRmrMessage(RmrMessagesBuffer_t &rmrMessageBuffer, ReportingMessages_t &message);
 /**
  *
  * @param epoll_fd
@@ -348,10 +382,9 @@ int sendRmrMessage(RmrMessagesBuffer_t &rmrMessageBuffer, ReportingMessages_t &m
  * @param sctpMap
  * @param enodbName
  * @param msgType
- * @param pSpan
  * @returnsrc::logger_mt& lg = my_logger::get();
  */
-int addToEpoll(int epoll_fd, ConnectedCU_t *peerInfo, uint32_t events, Sctp_Map_t *sctpMap, char *enodbName, int msgType, otSpan *pSpan);
+int addToEpoll(int epoll_fd, ConnectedCU_t *peerInfo, uint32_t events, Sctp_Map_t *sctpMap, char *enodbName, int msgType);
 /**
  *
  * @param epoll_fd
@@ -360,10 +393,9 @@ int addToEpoll(int epoll_fd, ConnectedCU_t *peerInfo, uint32_t events, Sctp_Map_
  * @param sctpMap
  * @param enodbName
  * @param msgType
- * @param pSpan
  * @return
  */
-int modifyToEpoll(int epoll_fd, ConnectedCU_t *peerInfo, uint32_t events, Sctp_Map_t *sctpMap, char *enodbName, int msgType, otSpan *pSpan);
+int modifyToEpoll(int epoll_fd, ConnectedCU_t *peerInfo, uint32_t events, Sctp_Map_t *sctpMap, char *enodbName, int msgType);
 
 /**
  *
@@ -379,8 +411,28 @@ void buildJsonMessage(ReportingMessages_t &message);
  */
 string translateRmrErrorMessages(int state);
 
+int buildConfiguration(sctp_params_t &sctpParams);
+void startPrometheus(sctp_params_t &sctpParams);
+static int enable_log_change_notify(const char* fileName);
+static int register_log_change_notify(const char *fileName);
+static void * monitor_loglevel_change_handler(void* arg);
+void  update_mdc_log_level_severity(char* log_level);
+static char* parse_file(char* filename);
+
+
+static inline uint64_t rdtscp(uint32_t &aux) {
+    uint64_t rax,rdx;
+    asm volatile ("rdtscp\n" : "=a" (rax), "=d" (rdx), "=c" (aux) : :);
+    return (rdx << (unsigned)32) + rax;
+}
 #ifndef RIC_SCTP_CONNECTION_FAILURE
 #define RIC_SCTP_CONNECTION_FAILURE  10080
 #endif
 
+#ifdef UNIT_TEST
+    #define FILE_DESCRIPTOR 53424 /*Dummy value for file descriptor only when UT is defined*/
+#endif
+
+int buildListeningPort(sctp_params_t &sctpParams);
+
 #endif //X2_SCTP_THREAD_H