RIC-E2-TERMINATION/sctpThread.h

   1 /*
   2  * Copyright 2019 AT&T Intellectual Property
   3  * Copyright 2019 Nokia
   4  *
   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
   8  *
   9  *      http://www.apache.org/licenses/LICENSE-2.0
  10  *
  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.
  16 */
  17
  18 /*
  19  * This source code is part of the near-RT RIC (RAN Intelligent Controller)
  20  * platform project (RICP).
  21  */
  22
  23 #ifndef X2_SCTP_THREAD_H
  24 #define X2_SCTP_THREAD_H
  25
  26 #include <algorithm>
  27
  28 #include <cstdio>
  29 #include <cerrno>
  30 #include <cstdlib>
  31 #include <cstring>
  32 #include <random>
  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>
  40 #include <thread>
  41 #include <atomic>
  42 #include <sys/param.h>
  43 #include <sys/file.h>
  44 #include <ctime>
  45 #include <netdb.h>
  46 #include <sys/epoll.h>
  47 #include <mutex>
  48 #include <shared_mutex>
  49 #include <iterator>
  50 #include <map>
  51 #include <sys/inotify.h>
  52 #include <csignal>
  53
  54 #include <rmr/rmr.h>
  55 #include <rmr/RIC_message_types.h>
  56 #include <mdclog/mdclog.h>
  57 #include <functional>
  58 #include <iostream>
  59
  60 #include <boost/algorithm/string/predicate.hpp>
  61 #include <boost/lexical_cast.hpp>
  62 #include <boost/move/utility.hpp>
  63 #include <boost/log/sources/logger.hpp>
  64 #include <boost/log/sources/record_ostream.hpp>
  65 #include <boost/log/sources/global_logger_storage.hpp>
  66 #include <boost/log/utility/setup/file.hpp>
  67 #include <boost/log/utility/setup/common_attributes.hpp>
  68 #include <boost/filesystem.hpp>
  69
  70 #include <mdclog/mdclog.h>
  71
  72 #include "oranE2/E2AP-PDU.h"
  73 #include "oranE2/ProtocolIE-Container.h"
  74 #include "oranE2/InitiatingMessage.h"
  75 #include "oranE2/SuccessfulOutcome.h"
  76 #include "oranE2/UnsuccessfulOutcome.h"
  77 #include "oranE2/ProtocolIE-Container.h"
  78 #include "oranE2/ProtocolIE-Field.h"
  79 #include "oranE2/GlobalE2node-gNB-ID.h"
  80 #include "oranE2/GlobalE2node-en-gNB-ID.h"
  81 #include "oranE2/GlobalE2node-ng-eNB-ID.h"
  82 #include "oranE2/GlobalE2node-eNB-ID.h"
  83
  84 #include "cxxopts.hpp"
  85 //#include "config-cpp/include/config-cpp/config-cpp.h"
  86
  87
  88 #include "mapWrapper.h"
  89 #include "statCollector.h"
  90
  91 #include "base64.h"
  92
  93 #include "ReadConfigFile.h"
  94
  95 using namespace std;
  96 namespace logging = boost::log;
  97 namespace src = boost::log::sources;
  98 namespace keywords = boost::log::keywords;
  99 namespace sinks = boost::log::sinks;
 100 namespace posix_time = boost::posix_time;
 101 namespace expr = boost::log::expressions;
 102
 103 #define SRC_PORT 36422
 104 #define SA      struct sockaddr
 105 #define MAX_ENODB_NAME_SIZE 64
 106
 107 #define MAXEVENTS 128
 108
 109 #define RECEIVE_SCTP_BUFFER_SIZE (128 * 1024)
 110 #define RECEIVE_XAPP_BUFFER_SIZE RECEIVE_SCTP_BUFFER_SIZE
 111
 112 typedef mapWrapper Sctp_Map_t;
 113
 114
 115
 116 #define VOLUME_URL_SIZE 256
 117 #define KA_MESSAGE_SIZE 2048
 118
 119 typedef struct sctp_params {
 120     uint16_t rmrPort = 0;
 121     uint16_t sctpPort = SRC_PORT;
 122     int      epoll_fd = 0;
 123     int      listenFD = 0;
 124     int      rmrListenFd = 0;
 125     int      inotifyFD = 0;
 126     int      inotifyWD = 0;
 127     void     *rmrCtx = nullptr;
 128     Sctp_Map_t *sctpMap = nullptr;
 129     char      ka_message[KA_MESSAGE_SIZE] {};
 130     int       ka_message_length = 0;
 131     char       rmrAddress[256] {}; // "tcp:portnumber" "tcp:5566" listen to all address on port 5566
 132     mdclog_severity_t logLevel = MDCLOG_INFO;
 133     char volume[VOLUME_URL_SIZE];
 134     string myIP {};
 135     string fqdn {};
 136     string podName {};
 137     string configFilePath {};
 138     string configFileName {};
 139     bool trace = true;
 140     //shared_timed_mutex fence; // moved to mapWrapper
 141 } sctp_params_t;
 142
 143 typedef struct ConnectedCU {
 144     int fileDescriptor = 0;
 145     char hostName[NI_MAXHOST] {};
 146     char portNumber[NI_MAXSERV] {};
 147     char enodbName[MAX_ENODB_NAME_SIZE] {};
 148     char asnData[RECEIVE_SCTP_BUFFER_SIZE] {};
 149     size_t asnLength = 0;
 150     int mtype = 0;
 151     bool isConnected = false;
 152     bool gotSetup = false;
 153     sctp_params_t *sctpParams = nullptr;
 154 } ConnectedCU_t ;
 155
 156 #define MAX_RMR_BUFF_ARRY 32
 157 typedef struct RmrMessagesBuffer {
 158     char ka_message[KA_MESSAGE_SIZE] {};
 159     int  ka_message_len = 0;
 160     void *rmrCtx = nullptr;
 161     rmr_mbuf_t *sendMessage= nullptr;
 162     //rmr_mbuf_t *sendBufferedMessages[MAX_RMR_BUFF_ARRY] {};
 163     rmr_mbuf_t *rcvMessage= nullptr;
 164     //rmr_mbuf_t *rcvBufferedMessages[MAX_RMR_BUFF_ARRY] {};
 165 } RmrMessagesBuffer_t;
 166
 167 typedef struct formatedMessage {
 168     char enodbName[MAX_ENODB_NAME_SIZE];
 169     struct timespec time;
 170     int messageType;
 171     char direction;
 172     ssize_t asnLength;
 173     unsigned char *asndata;
 174 } FormatedMessage_t;
 175
 176 typedef struct ReportingMessages {
 177     FormatedMessage_t message {};
 178     ConnectedCU_t *peerInfo = nullptr;
 179     long outLen = 0;
 180     unsigned char base64Data[RECEIVE_SCTP_BUFFER_SIZE * 2] {};
 181     char buffer[RECEIVE_SCTP_BUFFER_SIZE * 8] {};
 182     StatCollector *statCollector = nullptr;
 183 } ReportingMessages_t;
 184
 185 cxxopts::ParseResult parse(int argc, char *argv[], sctp_params_t &pSctpParams);
 186
 187 int buildInotify(sctp_params_t &sctpParams);
 188
 189 void handleTermInit(sctp_params_t &sctpParams);
 190
 191 void handleConfigChange(sctp_params_t *sctpParams);
 192
 193 void listener(sctp_params_t *params);
 194
 195 void sendTermInit(sctp_params_t &sctpParams);
 196
 197 int setSocketNoBlocking(int socket);
 198
 199 void handleEinprogressMessages(struct epoll_event &event,
 200                                ReportingMessages_t &message,
 201                                RmrMessagesBuffer_t &rmrMessageBuffer,
 202                                sctp_params_t *params);
 203
 204 void handlepoll_error(struct epoll_event &event,
 205                       ReportingMessages_t &message,
 206                       RmrMessagesBuffer_t &rmrMessageBuffer,
 207                       sctp_params_t *params);
 208
 209
 210 void cleanHashEntry(ConnectedCU_t *peerInfo, Sctp_Map_t *m);
 211
 212
 213 /**
 214  *
 215  * @param message
 216  * @param rmrMessageBuffer
 217  */
 218 void getRequestMetaData(ReportingMessages_t &message, RmrMessagesBuffer_t &rmrMessageBuffer);
 219
 220 /**
 221  *
 222  * @param sctpMap
 223  * @param messagBuffer
 224  * @param message
 225  * @param failedMesgId
 226  * @return
 227  */
 228 int sendMessagetoCu(Sctp_Map_t *sctpMap,
 229                     RmrMessagesBuffer_t &messagBuffer,
 230                     ReportingMessages_t &message,
 231                     int failedMesgId);
 232
 233 void sendFailedSendingMessagetoXapp(RmrMessagesBuffer_t &rmrMessageBuffer,
 234                                     ReportingMessages_t &message,
 235                                     int failedMesgId);
 236
 237 int sendRequestToXapp(ReportingMessages_t &message,
 238                       int requestId,
 239                       RmrMessagesBuffer_t &rmrMmessageBuffer);
 240
 241 /**
 242  *
 243  * @param message
 244  * @param msgType
 245  * @param requestType
 246  * @param rmrMessageBuffer
 247  * @param sctpMap
 248  * @return
 249  */
 250 /*
 251 int sendResponseToXapp(ReportingMessages_t &message,
 252                        int msgType,
 253                        int requestType,
 254                        RmrMessagesBuffer_t &rmrMessageBuffer,
 255                        Sctp_Map_t *sctpMap);
 256 */
 257
 258 /**
 259  *
 260  * @param peerInfo
 261  * @param message
 262  * @param m
 263  * @return
 264  */
 265 int sendSctpMsg(ConnectedCU_t *peerInfo,
 266                 ReportingMessages_t &message,
 267                 Sctp_Map_t *m);
 268
 269 /**
 270  *
 271  * @param events
 272  * @param sctpMap
 273  * @param numOfMessages
 274  * @param rmrMessageBuffer
 275  * @param ts
 276  * @return
 277  */
 278 int receiveDataFromSctp(struct epoll_event *events,
 279                         Sctp_Map_t *sctpMap,
 280                         int &numOfMessages,
 281                         RmrMessagesBuffer_t &rmrMessageBuffer,
 282                         struct timespec &ts);
 283
 284 /**
 285  *
 286  * @param rmrAddress
 287  * @return
 288  */
 289 void getRmrContext(sctp_params_t &pSctpParams);
 290
 291 /**
 292  *
 293  * @param epoll_fd
 294  * @param rmrCtx
 295  * @param sctpMap
 296  * @param messagBuffer
 297  * @return
 298  */
 299 int receiveXappMessages(Sctp_Map_t *sctpMap,
 300                         RmrMessagesBuffer_t &rmrMessageBuffer,
 301                         struct timespec &ts);
 302
 303 /**
 304  *
 305  * @param messagBuffer
 306  * @param failedMsgId
 307  * @param sctpMap
 308  * @return
 309  */
 310 int sendDirectionalSctpMsg(RmrMessagesBuffer_t &messagBuffer,
 311                            ReportingMessages_t &message,
 312                            int failedMsgId,
 313                            Sctp_Map_t *sctpMap);
 314 /**
 315  *
 316  * @param pdu
 317  * @param message
 318  * @param rmrMessageBuffer
 319  */
 320 void asnInitiatingRequest(E2AP_PDU_t *pdu,
 321                           Sctp_Map_t *sctpMap,
 322                           ReportingMessages_t &message,
 323                           RmrMessagesBuffer_t &rmrMessageBuffer);
 324 /**
 325  *
 326  * @param pdu
 327  * @param message
 328  * @param sctpMap
 329  * @param rmrMessageBuffer
 330  */
 331 void asnSuccsesfulMsg(E2AP_PDU_t *pdu,
 332                       Sctp_Map_t *sctpMap,
 333                       ReportingMessages_t &message,
 334                       RmrMessagesBuffer_t &rmrMessageBuffer);
 335 /**
 336  *
 337  * @param pdu
 338  * @param message
 339  * @param sctpMap
 340  * @param rmrMessageBuffer
 341  */
 342 void asnUnSuccsesfulMsg(E2AP_PDU_t *pdu,
 343                         Sctp_Map_t *sctpMap,
 344                         ReportingMessages_t &message,
 345                         RmrMessagesBuffer_t &rmrMessageBuffer);
 346
 347 /**
 348  *
 349  * @param rmrMessageBuffer
 350  * @param message
 351  * @return
 352  */
 353 int sendRmrMessage(RmrMessagesBuffer_t &rmrMessageBuffer, ReportingMessages_t &message);
 354 /**
 355  *
 356  * @param epoll_fd
 357  * @param peerInfo
 358  * @param events
 359  * @param sctpMap
 360  * @param enodbName
 361  * @param msgType
 362  * @returnsrc::logger_mt& lg = my_logger::get();
 363  */
 364 int addToEpoll(int epoll_fd, ConnectedCU_t *peerInfo, uint32_t events, Sctp_Map_t *sctpMap, char *enodbName, int msgType);
 365 /**
 366  *
 367  * @param epoll_fd
 368  * @param peerInfo
 369  * @param events
 370  * @param sctpMap
 371  * @param enodbName
 372  * @param msgType
 373  * @return
 374  */
 375 int modifyToEpoll(int epoll_fd, ConnectedCU_t *peerInfo, uint32_t events, Sctp_Map_t *sctpMap, char *enodbName, int msgType);
 376
 377 /**
 378  *
 379  * @param message
 380  */
 381 void buildJsonMessage(ReportingMessages_t &message);
 382
 383 /**
 384  *
 385  *
 386  * @param state
 387  * @return
 388  */
 389 string translateRmrErrorMessages(int state);
 390
 391
 392 static inline uint64_t rdtscp(uint32_t &aux) {
 393     uint64_t rax,rdx;
 394     asm volatile ("rdtscp\n" : "=a" (rax), "=d" (rdx), "=c" (aux) : :);
 395     return (rdx << (unsigned)32) + rax;
 396 }
 397 #ifndef RIC_SCTP_CONNECTION_FAILURE
 398 #define RIC_SCTP_CONNECTION_FAILURE  10080
 399 #endif
 400
 401 #endif //X2_SCTP_THREAD_H