2 * Copyright 2019 AT&T Intellectual Property
5 * Licensed under the Apache License, Version 2.0 (the "License");
6 * you may not use this file except in compliance with the License.
7 * You may obtain a copy of the License at
9 * http://www.apache.org/licenses/LICENSE-2.0
11 * Unless required by applicable law or agreed to in writing, software
12 * distributed under the License is distributed on an "AS IS" BASIS,
13 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 * See the License for the specific language governing permissions and
15 * limitations under the License.
19 * This source code is part of the near-RT RIC (RAN Intelligent Controller)
20 * platform project (RICP).
23 #ifndef X2_SCTP_THREAD_H
24 #define X2_SCTP_THREAD_H
33 #include <sys/socket.h>
34 #include <arpa/inet.h>
35 #include <netinet/in_systm.h>
36 #include <netinet/in.h>
37 #include <netinet/ip.h>
38 #include <netinet/ip_icmp.h>
39 #include <netinet/sctp.h>
42 #include <sys/param.h>
44 #include <sys/types.h>
48 #include <sys/epoll.h>
50 #include <shared_mutex>
53 #include <sys/inotify.h>
59 #include <rmr/RIC_message_types.h>
60 #include <mdclog/mdclog.h>
64 #include <boost/algorithm/string/predicate.hpp>
65 #include <boost/lexical_cast.hpp>
66 #include <boost/move/utility.hpp>
67 #include <boost/log/sources/logger.hpp>
68 #include <boost/log/sources/record_ostream.hpp>
69 #include <boost/log/sources/global_logger_storage.hpp>
70 #include <boost/log/utility/setup/file.hpp>
71 #include <boost/log/utility/setup/common_attributes.hpp>
72 #include <boost/filesystem.hpp>
74 #include <mdclog/mdclog.h>
76 #include "oranE2/E2AP-PDU.h"
77 #include "oranE2/ProtocolIE-Container.h"
78 #include "oranE2/InitiatingMessage.h"
79 #include "oranE2/SuccessfulOutcome.h"
80 #include "oranE2/UnsuccessfulOutcome.h"
81 #include "oranE2/ProtocolIE-Container.h"
82 #include "oranE2/ProtocolIE-Field.h"
83 #include "oranE2/GlobalE2node-gNB-ID.h"
84 #include "oranE2/GlobalE2node-en-gNB-ID.h"
85 #include "oranE2/GlobalE2node-ng-eNB-ID.h"
86 #include "oranE2/GlobalE2node-eNB-ID.h"
88 #include "cxxopts.hpp"
89 //#include "config-cpp/include/config-cpp/config-cpp.h"
91 #include <prometheus/counter.h>
92 #include <prometheus/exposer.h>
93 #include <prometheus/gateway.h>
94 #include <prometheus/registry.h>
96 using namespace prometheus;
98 #include "mapWrapper.h"
102 #include "ReadConfigFile.h"
105 namespace logging = boost::log;
106 namespace src = boost::log::sources;
107 namespace keywords = boost::log::keywords;
108 namespace sinks = boost::log::sinks;
109 namespace posix_time = boost::posix_time;
110 namespace expr = boost::log::expressions;
112 #define SRC_PORT 36422
113 #define SA struct sockaddr
114 #define MAX_ENODB_NAME_SIZE 64
116 #define MAXEVENTS 128
118 #define RECEIVE_SCTP_BUFFER_SIZE (8 * 1024)
119 #define RECEIVE_XAPP_BUFFER_SIZE RECEIVE_SCTP_BUFFER_SIZE
121 typedef mapWrapper Sctp_Map_t;
125 #define VOLUME_URL_SIZE 256
126 #define KA_MESSAGE_SIZE 2048
128 enum E2T_Internal_Counters
130 SCTP_ABORT_INITIATED_BY_E2NODE = 0,
131 INVALID_MESSAGE_RECEIVED = 1,
132 E2T_MAX_INTERNAL_COUNTER = 2,
135 typedef struct sctp_params {
136 int epollTimeOut = -1;
137 uint16_t rmrPort = 0;
138 uint16_t sctpPort = SRC_PORT;
144 void *rmrCtx = nullptr;
145 Sctp_Map_t *sctpMap = nullptr;
146 char ka_message[KA_MESSAGE_SIZE] {};
147 int ka_message_length = 0;
148 char rmrAddress[256] {}; // "tcp:port number" "tcp:5566" listen to all address on port 5566
149 char volume[VOLUME_URL_SIZE];
153 string configFilePath {};
154 string configFileName {};
156 shared_ptr<prometheus::Registry> prometheusRegistry;
157 string prometheusPort {"8088"};
158 Family<Counter> *prometheusFamily;
159 Exposer *prometheusExposer = nullptr;
160 Counter *e2tCounters[6][2][ProcedureCode_id_RICsubscriptionDeleteRequired + 1] {};
161 Counter *e2tInternalCounters[E2T_Internal_Counters::E2T_MAX_INTERNAL_COUNTER] {};
165 #define IN_INITI 0 //INITIATING
166 #define IN_SUCC 1 //SUCCESSFUL
167 #define IN_UN_SUCC 2 //UN-Successful
170 #define OUT_INITI 3 //INITIATING
171 #define OUT_SUCC 4 //SUCCESSFUL
172 #define OUT_UN_SUCC 5 //UN-Successful
174 #define MSG_COUNTER 0
175 #define BYTES_COUNTER 1
177 #define INVALID_STREAM_ID -1
179 typedef struct ConnectedCU {
180 int fileDescriptor = 0;
181 char hostName[NI_MAXHOST] {};
182 char portNumber[NI_MAXSERV] {};
183 char enodbName[MAX_ENODB_NAME_SIZE] {};
184 char asnData[RECEIVE_SCTP_BUFFER_SIZE] {};
185 size_t asnLength = 0;
187 bool isConnected = false;
188 bool gotSetup = false;
189 sctp_params_t *sctpParams = nullptr;
190 Counter *counters[6][2][ProcedureCode_id_RICsubscriptionDeleteRequired + 1] {};
191 bool isSingleStream = false;
192 int singleStreamId = 0;
193 Counter *e2tInternalCounters[E2T_Internal_Counters::E2T_MAX_INTERNAL_COUNTER] {};
197 #define MAX_RMR_BUFF_ARRAY 32
198 typedef struct RmrMessagesBuffer {
199 char ka_message[KA_MESSAGE_SIZE] {};
200 int ka_message_len = 0;
201 void *rmrCtx = nullptr;
202 rmr_mbuf_t *sendMessage= nullptr;
203 //rmr_mbuf_t *sendBufferedMessages[MAX_RMR_BUFF_ARRAY] {};
204 rmr_mbuf_t *rcvMessage= nullptr;
205 //rmr_mbuf_t *rcvBufferedMessages[MAX_RMR_BUFF_ARRAY] {};
206 } RmrMessagesBuffer_t;
208 typedef struct formatedMessage {
209 char enodbName[MAX_ENODB_NAME_SIZE];
210 struct timespec time;
214 unsigned char *asndata;
217 typedef struct ReportingMessages {
218 FormatedMessage_t message {};
219 ConnectedCU_t *peerInfo = nullptr;
221 unsigned char base64Data[RECEIVE_SCTP_BUFFER_SIZE * 2] {};
222 char buffer[RECEIVE_SCTP_BUFFER_SIZE * 8] {};
223 } ReportingMessages_t;
225 enum E2T_Procedure_States
227 E2_SETUP_PROCEDURE_NOT_INITIATED = 0,
228 E2_SETUP_PROCEDURE_ONGOING = 1,
229 E2_SETUP_PROCEDURE_COMPLETED = 2,
230 RIC_SERVICE_UPDATE_PROCEDURE_ONGOING = 3,
231 RIC_SERVICE_UPDATE_PROCEDURE_COMPLETED = 4,
232 RIC_SUBS_PROCEDURE_ONGOING = 5,
233 RIC_SUBS_PROCEDURE_COMPLETED = 6,
234 RIC_INDICATION_PROCEDURE_ONGOING = 7,
235 RIC_INDICATION_PROCEDURE_COMPLETED = 8,
236 RIC_SUBS_DEL_PROCEDURE_ONGOING = 9,
237 RIC_SUBS_DEL_PROCEDURE_COMPLETED = 10,
238 CONTROL_PROCEDURE_ONGOING = 11,
239 CONTROL_PROCEDURE_COMPLETED = 12,
240 E2_NODE_CONF_UPDATE_PROCEDURE_ONGOING = 13,
241 E2_NODE_CONF_UPDATE_PROCEDURE_COMPLETED = 14,
242 RESET_PROCEDURE_ONGOING = 15,
243 RESET_PROCEDURE_COMPLETED = 16,
246 struct E2NodeConnectionHandling
248 E2T_Procedure_States e2tProcedureOngoingStatus;
249 long e2SetupProcedureTransactionId;
252 constexpr int negativeOne = -1;
253 constexpr int negativeSix = -6;
254 constexpr int negativeSeven = -7;
255 constexpr int numberZero = 0;
256 constexpr int numberOne = 1;
257 constexpr int numberTwo = 2;
258 constexpr int numberThree = 3;
259 constexpr int numberFour = 4;
260 constexpr int numberFive = 5;
261 constexpr int numberTwenty = 20;
263 constexpr uint8_t sendMsgMaxBitPosition = 2;
264 constexpr uint8_t sendMsgToE2MBitSetPosition = 0;
265 constexpr uint8_t sendMsgToSubMgrBitSetPosition = 1;
267 constexpr uint8_t requiredIePresentMaxBitSetPosition = 3;
268 constexpr uint8_t transactionIdIeBitSetPosition = 0;
269 constexpr uint8_t ricRequestIdIeBitSetPosition = 1;
270 constexpr uint8_t causeIeBitSetPosition = 2;
272 cxxopts::ParseResult parse(int argc, char *argv[], sctp_params_t &pSctpParams);
274 int buildInotify(sctp_params_t &sctpParams);
276 void handleTermInit(sctp_params_t &sctpParams);
278 void handleConfigChange(sctp_params_t *sctpParams);
280 void listener(sctp_params_t *params);
282 void sendTermInit(sctp_params_t &sctpParams);
284 int setSocketNoBlocking(int socket);
286 void handleEinprogressMessages(struct epoll_event &event,
287 ReportingMessages_t &message,
288 RmrMessagesBuffer_t &rmrMessageBuffer,
289 sctp_params_t *params);
291 void handlepoll_error(struct epoll_event &event,
292 ReportingMessages_t &message,
293 RmrMessagesBuffer_t &rmrMessageBuffer,
294 sctp_params_t *params);
297 void cleanHashEntry(ConnectedCU_t *peerInfo, Sctp_Map_t *m);
303 * @param rmrMessageBuffer
305 void getRequestMetaData(ReportingMessages_t &message, RmrMessagesBuffer_t &rmrMessageBuffer);
310 * @param messageBuffer
312 * @param failedMesgId
315 int sendMessagetoCu(Sctp_Map_t *sctpMap,
316 RmrMessagesBuffer_t &messageBuffer,
317 ReportingMessages_t &message,
320 void sendFailedSendingMessagetoXapp(RmrMessagesBuffer_t &rmrMessageBuffer,
321 ReportingMessages_t &message,
324 int sendRequestToXapp(ReportingMessages_t &message,
326 RmrMessagesBuffer_t &rmrMmessageBuffer);
333 * @param rmrMessageBuffer
338 int sendResponseToXapp(ReportingMessages_t &message,
341 RmrMessagesBuffer_t &rmrMessageBuffer,
342 Sctp_Map_t *sctpMap);
352 int sendSctpMsg(ConnectedCU_t *peerInfo,
353 ReportingMessages_t &message,
360 * @param numOfMessages
361 * @param rmrMessageBuffer
365 int receiveDataFromSctp(struct epoll_event *events,
368 RmrMessagesBuffer_t &rmrMessageBuffer,
369 struct timespec &ts);
376 void getRmrContext(sctp_params_t &pSctpParams);
381 * @param rmrMessageBuffer
385 int receiveXappMessages(Sctp_Map_t *sctpMap,
386 RmrMessagesBuffer_t &rmrMessageBuffer,
387 struct timespec &ts);
391 * @param messageBuffer
396 int sendDirectionalSctpMsg(RmrMessagesBuffer_t &messageBuffer,
397 ReportingMessages_t &message,
399 Sctp_Map_t *sctpMap);
404 * @param rmrMessageBuffer
406 void asnInitiatingRequest(E2AP_PDU_t *pdu,
408 ReportingMessages_t &message,
409 RmrMessagesBuffer_t &rmrMessageBuffer,int streamId);
415 * @param rmrMessageBuffer
417 void asnSuccessfulMsg(E2AP_PDU_t *pdu,
419 ReportingMessages_t &message,
420 RmrMessagesBuffer_t &rmrMessageBuffer);
426 * @param rmrMessageBuffer
428 void asnUnSuccsesfulMsg(E2AP_PDU_t *pdu,
430 ReportingMessages_t &message,
431 RmrMessagesBuffer_t &rmrMessageBuffer);
435 * @param rmrMessageBuffer
439 int sendRmrMessage(RmrMessagesBuffer_t &rmrMessageBuffer, ReportingMessages_t &message);
448 * @returnsrc::logger_mt& lg = my_logger::get();
450 int addToEpoll(int epoll_fd, ConnectedCU_t *peerInfo, uint32_t events, Sctp_Map_t *sctpMap, char *enodbName, int msgType);
461 int modifyToEpoll(int epoll_fd, ConnectedCU_t *peerInfo, uint32_t events, Sctp_Map_t *sctpMap, char *enodbName, int msgType);
467 void buildJsonMessage(ReportingMessages_t &message);
475 string translateRmrErrorMessages(int state);
477 int buildConfiguration(sctp_params_t &sctpParams);
478 void startPrometheus(sctp_params_t &sctpParams);
479 static int enable_log_change_notify(const char* fileName);
480 static int register_log_change_notify(const char *fileName);
481 static void * monitor_loglevel_change_handler(void* arg);
482 void update_mdc_log_level_severity(char* log_level);
483 char* getinterfaceip();
484 static char* parse_file(char* filename);
487 static inline uint64_t rdtscp(uint32_t &aux) {
489 asm volatile ("rdtscp\n" : "=a" (rax), "=d" (rdx), "=c" (aux) : :);
490 return (rdx << (unsigned)32) + rax;
492 #ifndef RIC_SCTP_CONNECTION_FAILURE
493 #define RIC_SCTP_CONNECTION_FAILURE 10080
497 #define FILE_DESCRIPTOR 53424 /*Dummy value for file descriptor only when UT is defined*/
500 int buildListeningPort(sctp_params_t &sctpParams);
501 void buildE2TPrometheusCounters(sctp_params_t &sctpParams);
503 int fetchStreamId(ConnectedCU_t *peerInfo, ReportingMessages_t &message);
504 void removeE2ConnectionEntryFromMap(char* eNBName);
505 bool getE2tProcedureOngoingStatus(char *enbName, E2NodeConnectionHandling &e2NodeConnectionHandling);
506 void setE2ProcedureOngoingStatus(char *enbName, E2T_Procedure_States state);
507 void insertE2ProcedureOngoing(char *enbName, long &transactionID);
508 E2T_Procedure_States currentE2tProcedureOngoingStatus(char *enbName);
509 void printEntryPresentInMap();
510 void handleE2SetupReq(ReportingMessages_t &message, RmrMessagesBuffer_t &rmrMessageBuffer, E2AP_PDU_t *pdu, long &transactionID, int streamId, Sctp_Map_t *sctpMap);
511 bitset<sendMsgMaxBitPosition> getSendMsgBitSetValue(int procedureCode, bitset<requiredIePresentMaxBitSetPosition> isRequiredIesPresent, char* enbName);
512 #endif //X2_SCTP_THREAD_H