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 "asn1cFiles/E2AP-PDU.h"
  73 #include "asn1cFiles/ProtocolIE-Container.h"
  74 #include "asn1cFiles/InitiatingMessage.h"
  75 #include "asn1cFiles/SuccessfulOutcome.h"
  76 #include "asn1cFiles/UnsuccessfulOutcome.h"
  77 #include "asn1cFiles/ProtocolIE-Container.h"
  78 #include "asn1cFiles/ProtocolIE-Field.h"
  79
  80 #include "cxxopts.hpp"
  81 //#include "config-cpp/include/config-cpp/config-cpp.h"
  82
  83 #ifdef __TRACING__
  84 #include "openTracing.h"
  85 #endif
  86
  87 #include "mapWrapper.h"
  88
  89 #include "base64.h"
  90
  91 #include "ReadConfigFile.h"
  92
  93 using namespace std;
  94 namespace logging = boost::log;
  95 namespace src = boost::log::sources;
  96 namespace keywords = boost::log::keywords;
  97 namespace sinks = boost::log::sinks;
  98 namespace posix_time = boost::posix_time;
  99 namespace expr = boost::log::expressions;
 100
 101 #define SRC_PORT 36422
 102 #define SA      struct sockaddr
 103 #define MAX_ENODB_NAME_SIZE 64
 104
 105 #define MAXEVENTS 128
 106
 107 #define RECEIVE_SCTP_BUFFER_SIZE (64*1024)
 108 #define RECEIVE_XAPP_BUFFER_SIZE RECEIVE_SCTP_BUFFER_SIZE
 109
 110 typedef mapWrapper Sctp_Map_t;
 111
 112 #ifdef __TRACING__
 113 typedef const std::unique_ptr<opentracing::Span> otSpan;
 114 #else
 115 typedef const int otSpan;
 116 #endif
 117
 118 #define VOLUME_URL_SIZE 256
 119
 120 typedef struct sctp_params {
 121     uint16_t rmrPort = 0;
 122     int      epoll_fd = 0;
 123     int      rmrListenFd = 0;
 124     int      inotifyFD = 0;
 125     int      inotifyWD = 0;
 126     void     *rmrCtx = nullptr;
 127     Sctp_Map_t *sctpMap = nullptr;
 128     char      ka_message[4096] {};
 129     int       ka_message_length = 0;
 130     char       rmrAddress[256] {}; // "tcp:portnumber" "tcp:5566" listen to all address on port 5566
 131     mdclog_severity_t logLevel = MDCLOG_INFO;
 132     char volume[VOLUME_URL_SIZE];
 133     string myIP {};
 134     string fqdn {};
 135     string podName {};
 136     string configFilePath {};
 137     string configFileName {};
 138     bool trace = true;
 139     //shared_timed_mutex fence; // moved to mapWrapper
 140 } sctp_params_t;
 141
 142 typedef struct ConnectedCU {
 143     int fileDescriptor = 0;
 144     char hostName[NI_MAXHOST] {};
 145     char portNumber[NI_MAXSERV] {};
 146     char enodbName[MAX_ENODB_NAME_SIZE] {};
 147     char asnData[RECEIVE_SCTP_BUFFER_SIZE] {};
 148     size_t asnLength = 0;
 149     int mtype = 0;
 150     bool isConnected = false;
 151 } ConnectedCU_t ;
 152
 153 #define MAX_RMR_BUFF_ARRY 32
 154 typedef struct RmrMessagesBuffer {
 155     char ka_message[4096] {};
 156     int  ka_message_len = 0;
 157     void *rmrCtx = nullptr;
 158     rmr_mbuf_t *sendMessage= nullptr;
 159     rmr_mbuf_t *sendBufferedMessages[MAX_RMR_BUFF_ARRY] {};
 160     rmr_mbuf_t *rcvMessage= nullptr;
 161     rmr_mbuf_t *rcvBufferedMessages[MAX_RMR_BUFF_ARRY] {};
 162 } RmrMessagesBuffer_t;
 163
 164 typedef struct formatedMessage {
 165     char enodbName[MAX_ENODB_NAME_SIZE];
 166     struct timespec time;
 167     int messageType;
 168     char direction;
 169     ssize_t asnLength;
 170     unsigned char *asndata;
 171 } FormatedMessage_t;
 172
 173 typedef struct ReportingMessages {
 174     FormatedMessage_t message;
 175     long outLen;
 176     unsigned char base64Data[RECEIVE_SCTP_BUFFER_SIZE * 2];
 177     char buffer[RECEIVE_SCTP_BUFFER_SIZE * 8];
 178 } ReportingMessages_t;
 179
 180 cxxopts::ParseResult parse(int argc, char *argv[], sctp_params_t &pSctpParams);
 181
 182 int buildInotify(sctp_params_t &sctpParams);
 183
 184 void handleTermInit(sctp_params_t &sctpParams);
 185
 186 void handleConfigChange(sctp_params_t *sctpParams);
 187
 188 void listener(sctp_params_t *params);
 189
 190 void sendTermInit(sctp_params_t &sctpParams);
 191
 192 int setSocketNoBlocking(int socket);
 193
 194 void handleEinprogressMessages(struct epoll_event &event,
 195                                ReportingMessages_t &message,
 196                                RmrMessagesBuffer_t &rmrMessageBuffer,
 197                                sctp_params_t *params,
 198                                otSpan *pSpan);
 199
 200 void handlepoll_error(struct epoll_event &event,
 201                       ReportingMessages_t &message,
 202                       RmrMessagesBuffer_t &rmrMessageBuffer,
 203                       sctp_params_t *params,
 204                       otSpan *pSpan);
 205
 206
 207 void cleanHashEntry(ConnectedCU_t *peerInfo, Sctp_Map_t *m, otSpan *pSpan);
 208
 209 int getSetupRequestMetaData(ReportingMessages_t &message, char *data, char *host, uint16_t &port, otSpan *pSpan);
 210
 211 /**
 212  *
 213  * @param message
 214  * @param rmrMessageBuffer
 215  * @param pSpan
 216  */
 217 void getRequestMetaData(ReportingMessages_t &message, RmrMessagesBuffer_t &rmrMessageBuffer, otSpan *pSpan);
 218
 219 /**
 220  *
 221  * @param sctpMap
 222  * @param messagBuffer
 223  * @param message
 224  * @param failedMesgId
 225  * @param pSpan
 226  * @return
 227  */
 228 int sendMessagetoCu(Sctp_Map_t *sctpMap,
 229                     RmrMessagesBuffer_t &messagBuffer,
 230                     ReportingMessages_t &message,
 231                     int failedMesgId, otSpan *pSpan);
 232
 233 void sendFailedSendingMessagetoXapp(RmrMessagesBuffer_t &rmrMessageBuffer,
 234                                     ReportingMessages_t &message,
 235                                     int failedMesgId,
 236                                     otSpan *pSpan);
 237
 238 int sendRequestToXapp(ReportingMessages_t &message,
 239                       int requestId,
 240                       RmrMessagesBuffer_t &rmrMmessageBuffer,
 241                       otSpan *pSpan);
 242
 243 /**
 244  *
 245  * @param message
 246  * @param msgType
 247  * @param requestType
 248  * @param rmrMessageBuffer
 249  * @param sctpMap
 250  * @param pSpan
 251  * @return
 252  */
 253 int sendResponseToXapp(ReportingMessages_t &message,
 254                        int msgType,
 255                        int requestType,
 256                        RmrMessagesBuffer_t &rmrMessageBuffer,
 257                        Sctp_Map_t *sctpMap,
 258                        otSpan *pSpan);
 259
 260 /**
 261  *
 262  * @param peerInfo
 263  * @param message
 264  * @param m
 265  * @param pSpan
 266  * @return
 267  */
 268 int sendSctpMsg(ConnectedCU_t *peerInfo,
 269                 ReportingMessages_t &message,
 270                 Sctp_Map_t *m,
 271                 otSpan *pSpan);
 272
 273 /**
 274  *
 275  * @param events
 276  * @param sctpMap
 277  * @param numOfMessages
 278  * @param rmrMessageBuffer
 279  * @param ts
 280  * @param pSpan
 281  * @return
 282  */
 283 int receiveDataFromSctp(struct epoll_event *events,
 284                         Sctp_Map_t *sctpMap,
 285                         int &numOfMessages,
 286                         RmrMessagesBuffer_t &rmrMessageBuffer,
 287                         struct timespec &ts,
 288                         otSpan *pSpan);
 289
 290 /**
 291  *
 292  * @param rmrAddress
 293  * @param pSpan
 294  * @return
 295  */
 296 void getRmrContext(sctp_params_t &pSctpParams, otSpan *pSpan);
 297
 298 /**
 299  *
 300  * @param epoll_fd
 301  * @param rmrCtx
 302  * @param sctpMap
 303  * @param messagBuffer
 304  * @param pSpan
 305  * @return
 306  */
 307 int receiveXappMessages(int epoll_fd,
 308                         Sctp_Map_t *sctpMap,
 309                         RmrMessagesBuffer_t &rmrMessageBuffer,
 310                         struct timespec &ts,
 311                         otSpan *pSpan);
 312
 313 /**
 314  *
 315  * @param rmrMessageBuffer
 316  * @param message
 317  * @param epoll_fd
 318  * @param sctpMap
 319  * @param pSpan
 320  * @return
 321  */
 322 int connectToCUandSetUp(RmrMessagesBuffer_t &rmrMessageBuffer,
 323                            ReportingMessages_t &message,
 324                            int epoll_fd,
 325                            Sctp_Map_t *sctpMap,
 326                            otSpan *pSpan);
 327
 328 /**
 329  *
 330  * @param messagBuffer
 331  * @param failedMsgId
 332  * @param sctpMap
 333  * @param pSpan
 334  * @return
 335  */
 336 int sendDirectionalSctpMsg(RmrMessagesBuffer_t &messagBuffer,
 337                            ReportingMessages_t &message,
 338                            int failedMsgId,
 339                            Sctp_Map_t *sctpMap,
 340                            otSpan *pSpan);
 341 /**
 342  *
 343  * @param pdu
 344  * @param message
 345  * @param rmrMessageBuffer
 346  * @param pSpan
 347  */
 348 void asnInitiatingRequest(E2AP_PDU_t *pdu,
 349                           ReportingMessages_t &message,
 350                           RmrMessagesBuffer_t &rmrMessageBuffer,
 351                           otSpan *pSpan);
 352 /**
 353  *
 354  * @param pdu
 355  * @param message
 356  * @param sctpMap
 357  * @param rmrMessageBuffer
 358  * @param pSpan
 359  */
 360 void asnSuccsesfulMsg(E2AP_PDU_t *pdu,
 361                       ReportingMessages_t &message,
 362                       Sctp_Map_t *sctpMap,
 363                       RmrMessagesBuffer_t &rmrMessageBuffer,
 364                       otSpan *pSpan);
 365 /**
 366  *
 367  * @param pdu
 368  * @param message
 369  * @param sctpMap
 370  * @param rmrMessageBuffer
 371  * @param pSpan
 372  */
 373 void asnUnSuccsesfulMsg(E2AP_PDU_t *pdu,
 374                         ReportingMessages_t &message,
 375                         Sctp_Map_t *sctpMap,
 376                         RmrMessagesBuffer_t &rmrMessageBuffer,
 377                         otSpan *pSpan);
 378
 379 /**
 380  *
 381  * @param rmrMessageBuffer
 382  * @param message
 383  * @param pSpan
 384  * @return
 385  */
 386 int sendRmrMessage(RmrMessagesBuffer_t &rmrMessageBuffer, ReportingMessages_t &message, otSpan *pSpan);
 387
 388 /**
 389  *
 390  * @param epoll_fd
 391  * @param peerInfo
 392  * @param events
 393  * @param sctpMap
 394  * @param enodbName
 395  * @param msgType
 396  * @param pSpan
 397  * @returnsrc::logger_mt& lg = my_logger::get();
 398  */
 399 int addToEpoll(int epoll_fd, ConnectedCU_t *peerInfo, uint32_t events, Sctp_Map_t *sctpMap, char *enodbName, int msgType, otSpan *pSpan);
 400 /**
 401  *
 402  * @param epoll_fd
 403  * @param peerInfo
 404  * @param events
 405  * @param sctpMap
 406  * @param enodbName
 407  * @param msgType
 408  * @param pSpan
 409  * @return
 410  */
 411 int modifyToEpoll(int epoll_fd, ConnectedCU_t *peerInfo, uint32_t events, Sctp_Map_t *sctpMap, char *enodbName, int msgType, otSpan *pSpan);
 412
 413 /**
 414  *
 415  * @param message
 416  */
 417 void buildJsonMessage(ReportingMessages_t &message);
 418
 419 /**
 420  *
 421  *
 422  * @param state
 423  * @return
 424  */
 425 string translateRmrErrorMessages(int state);
 426
 427
 428 static inline uint64_t rdtscp(uint32_t &aux) {
 429     uint64_t rax,rdx;
 430     asm volatile ("rdtscp\n" : "=a" (rax), "=d" (rdx), "=c" (aux) : :);
 431     return (rdx << 32) + rax;
 432 }
 433 #ifndef RIC_SCTP_CONNECTION_FAILURE
 434 #define RIC_SCTP_CONNECTION_FAILURE  10080
 435 #endif
 436
 437 #endif //X2_SCTP_THREAD_H