transport_split.c 5.08 KB
Newer Older
laurent's avatar
laurent committed
1 2 3 4 5
#include <executables/split_headers.h>
#include <sys/types.h>          /* See NOTES */
#include <sys/socket.h>
#include <netinet/in.h>
#include <netinet/udp.h>
laurent's avatar
laurent committed
6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55
#include <netdb.h>

bool createUDPsock (char *sourceIP, char *sourcePort, char *destIP, char *destPort, UDPsock_t *result) {
  struct addrinfo hints= {0}, *servinfo, *p;
  hints.ai_family = AF_UNSPEC;
  hints.ai_socktype = SOCK_DGRAM;
  hints.ai_flags = AI_PASSIVE;
  int status;

  if ((status = getaddrinfo(sourceIP, sourcePort, &hints, &servinfo)) != 0) {
    LOG_E(GTPU,"getaddrinfo error: %s\n", gai_strerror(status));
    return false;
  }

  // loop through all the results and bind to the first we can
  for(p = servinfo; p != NULL; p = p->ai_next) {
    if ((result->sockHandler = socket(p->ai_family, p->ai_socktype,
                                      p->ai_protocol)) == -1) {
      LOG_W(GTPU,"socket: %s\n", strerror(errno));
      continue;
    }

    if (bind(result->sockHandler, p->ai_addr, p->ai_addrlen) == -1) {
      close(result->sockHandler);
      LOG_W(GTPU,"bind: %s\n", strerror(errno));
      continue;
    }

    break; // if we get here, we must have connected successfully
  }

  if (p == NULL) {
    // looped off the end of the list with no successful bind
    LOG_E(GTPU,"failed to bind socket: %s %s \n",sourceIP,sourcePort);
    return false;
  }

  freeaddrinfo(servinfo); // all done with this structure

  if ((status = getaddrinfo(destIP, destPort, &hints, &servinfo)) != 0) {
    LOG_E(GTPU,"getaddrinfo error: %s\n", gai_strerror(status));
    return false;
  }

  if (servinfo) {
    result->destAddr=servinfo;
  } else {
    LOG_E(PHY,"No valid UDP addr: %s:%s\n",destIP, destPort);
    return false;
  }
laurent's avatar
laurent committed
56

laurent's avatar
laurent committed
57
  int enable=1;
laurent's avatar
laurent committed
58
  AssertFatal(setsockopt(result->sockHandler, SOL_SOCKET, SO_REUSEADDR, &enable, sizeof(enable))==0,"");
laurent's avatar
laurent committed
59 60
  //struct timeval tv= {0,UDP_TIMEOUT};
  struct timeval tv= {2,UDP_TIMEOUT}; //debug: wait 2 seconds for human understanding
laurent's avatar
laurent committed
61
  AssertFatal(setsockopt(result->sockHandler, SOL_SOCKET, SO_RCVTIMEO,&tv,sizeof(tv)) ==0,"");
laurent's avatar
laurent committed
62 63 64
  // Make a send/recv buffer larger than a a couple of subframe
  // so the kernel will store for us in and out paquets
  int buff=1000*1000*10;
laurent's avatar
laurent committed
65 66 67
  AssertFatal ( setsockopt(result->sockHandler, SOL_SOCKET, SO_SNDBUF, &buff, sizeof(buff)) == 0, "");
  AssertFatal ( setsockopt(result->sockHandler, SOL_SOCKET, SO_RCVBUF, &buff, sizeof(buff)) == 0, "");
  return true;
laurent's avatar
laurent committed
68 69 70 71 72
}

// sock: udp socket
// expectedTS: the expected timestamp, 0 if unknown
// bufferZone: a reception area of bufferSize
laurent's avatar
laurent committed
73
int receiveSubFrame(UDPsock_t *sock, uint64_t expectedTS, void *bufferZone,  int bufferSize, uint16_t contentType) {
laurent's avatar
laurent committed
74
  int rcved=0;
laurent's avatar
laurent committed
75
  commonUDP_t *bufOrigin=(commonUDP_t *)bufferZone;
laurent's avatar
laurent committed
76

laurent's avatar
laurent committed
77 78
  do {
    //read all subframe data from the control unit
laurent's avatar
laurent committed
79
    int ret=recv(sock->sockHandler, bufferZone, bufferSize, 0);
laurent's avatar
laurent committed
80

laurent's avatar
laurent committed
81 82
    if ( ret==-1) {
      if ( errno == EWOULDBLOCK || errno== EINTR ) {
laurent's avatar
laurent committed
83
        return  rcved; // Timeout, subframe incomplete
laurent's avatar
laurent committed
84
      } else {
laurent's avatar
laurent committed
85 86
        LOG_E(HW,"Critical issue in socket: %s\n", strerror(errno));
        return -1;
laurent's avatar
laurent committed
87 88
      }
    } else {
laurent's avatar
laurent committed
89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104
      if (hUDP(bufferZone)->contentType != contentType)
        abort();

      if ( hUDP(bufferZone)->timestamp != expectedTS) {
        if ( hUDP(bufferZone)->timestamp < expectedTS) {
          LOG_W(HW,"Received a paquet from past subframes, dropping it\n");
        } else {
          LOG_W(HW,"Received a paquet in future subframes\n");

          if ( rcved == 0 ) {
            LOG_W(HW,"First paquet in the sub-frame, keeping it\n");
            rcved++;
            bufferZone+=ret;
            expectedTS=hUDP(bufferZone)->timestamp;
          }
        }
laurent's avatar
laurent committed
105
      } else {
laurent's avatar
laurent committed
106 107
        rcved++;
        bufferZone+=ret;
laurent's avatar
laurent committed
108 109
      }
    }
laurent's avatar
laurent committed
110
  } while ( !rcved || rcved < hUDP(bufferZone)->nbBlocks );
laurent's avatar
laurent committed
111

laurent's avatar
laurent committed
112 113
  LOG_D(HW,"Received: nb_blocks: %d, TS: %lu, frame: %d, subframe: %d\n",
        rcved, bufOrigin->timestamp, *(((int *)bufOrigin)+1), *(((int *)bufOrigin)+2));
laurent's avatar
laurent committed
114
  return rcved;
laurent's avatar
laurent committed
115 116
}

laurent's avatar
laurent committed
117
int sendSubFrame(UDPsock_t *sock, void *bufferZone, ssize_t secondHeaderSize, uint16_t contentType) {
laurent's avatar
laurent committed
118
  commonUDP_t *UDPheader=(commonUDP_t *)bufferZone ;
laurent's avatar
laurent committed
119
  UDPheader->contentType=contentType;
laurent's avatar
laurent committed
120 121
  int nbBlocks=UDPheader->nbBlocks;
  int blockId=0;
laurent's avatar
laurent committed
122

laurent's avatar
laurent committed
123 124 125 126 127
  if (nbBlocks <= 0 ) {
    LOG_E(PHY,"FS6: can't send blocks: %d\n", nbBlocks);
    return 0;
  }

laurent's avatar
laurent committed
128
  do {
laurent's avatar
laurent committed
129 130 131 132 133 134 135 136
    if (blockId > 0 ) {
      commonUDP_t *currentHeader=(commonUDP_t *)bufferZone;
      *currentHeader=*UDPheader;
      currentHeader->blockID=blockId;
      memcpy(commonUDPdata((void *)currentHeader), commonUDPdata(bufferZone), secondHeaderSize);
      blockId++;
    }

laurent's avatar
laurent committed
137
    int sz=alignedSize(bufferZone);
laurent's avatar
laurent committed
138 139 140
    // Let's use the first address returned by getaddrinfo()
    int ret=sendto(sock->sockHandler, bufferZone, sz, 0,
                   sock->destAddr->ai_addr, sock->destAddr->ai_addrlen);
laurent's avatar
laurent committed
141

laurent's avatar
laurent committed
142
    if ( ret != sz )
laurent's avatar
laurent committed
143 144
      LOG_W(HW,"Wrote socket doesn't return size %d (val: %d, errno:%d, %s)\n",
            sz, ret, errno, strerror(errno));
laurent's avatar
laurent committed
145

laurent's avatar
laurent committed
146 147 148
    bufferZone+=sz;
    nbBlocks--;
  } while (nbBlocks);
laurent's avatar
laurent committed
149

laurent's avatar
laurent committed
150 151
  LOG_D(HW,"Sent: TS: %lu, frame: %d, subframe: %d\n",
        UDPheader->timestamp, *(((int *)UDPheader)+1), *(((int *)UDPheader)+2));
laurent's avatar
laurent committed
152 153
  return 0;
}