transport_split.c 5.22 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
}

// sock: udp socket
// bufferZone: a reception area of bufferSize
72
int receiveSubFrame(UDPsock_t *sock, void *bufferZone,  int bufferSize, uint16_t contentType) {
laurent's avatar
laurent committed
73
  int rcved=0;
laurent's avatar
laurent committed
74
  commonUDP_t *bufOrigin=(commonUDP_t *)bufferZone;
75 76 77 78 79 80 81 82 83 84
  static uint8_t crossData[65536];
  static int crossDataSize=0;

  if (crossDataSize) {
    LOG_D(HW,"copy a block received in previous subframe\n");
    memcpy(bufferZone, crossData, crossDataSize);
    rcved=1;
    bufferZone+=crossDataSize;
    crossDataSize=0;
  }
laurent's avatar
laurent committed
85

laurent's avatar
laurent committed
86 87
  do {
    //read all subframe data from the control unit
laurent's avatar
laurent committed
88
    int ret=recv(sock->sockHandler, bufferZone, bufferSize, 0);
laurent's avatar
laurent committed
89

laurent's avatar
laurent committed
90 91
    if ( ret==-1) {
      if ( errno == EWOULDBLOCK || errno== EINTR ) {
92 93
        LOG_I(HW,"Received: Timeout, subframe incomplete\n");
        return  rcved;
laurent's avatar
laurent committed
94
      } else {
laurent's avatar
laurent committed
95 96
        LOG_E(HW,"Critical issue in socket: %s\n", strerror(errno));
        return -1;
laurent's avatar
laurent committed
97 98
      }
    } else {
laurent's avatar
laurent committed
99 100 101
      if (hUDP(bufferZone)->contentType != contentType)
        abort();

102 103 104 105 106 107 108 109
      if (rcved && bufOrigin->timestamp != hUDP(bufferZone)->timestamp ) {
        if ( hUDP(bufferZone)->timestamp > bufOrigin->timestamp ) {
          LOG_W(HW,"Received data for TS: %lu before end of TS : %lu completion\n",
                hUDP(bufferZone)->timestamp,
                bufOrigin->timestamp);
          memcpy(crossData, bufferZone, ret );
          crossDataSize=ret;
          return rcved;
laurent's avatar
laurent committed
110
        } else {
111 112
          LOG_W(HW,"Dropping late packet\n");
          continue;
laurent's avatar
laurent committed
113
        }
laurent's avatar
laurent committed
114
      }
115 116 117

      rcved++;
      bufferZone+=ret;
laurent's avatar
laurent committed
118
    }
119
  } while ( rcved == 0 || rcved < bufOrigin->nbBlocks );
laurent's avatar
laurent committed
120

121
  LOG_D(HW,"Received: nb_blocks: %d, TS: %lu\n",rcved, bufOrigin->timestamp);
laurent's avatar
laurent committed
122
  return rcved;
laurent's avatar
laurent committed
123 124
}

laurent's avatar
laurent committed
125
int sendSubFrame(UDPsock_t *sock, void *bufferZone, ssize_t secondHeaderSize, uint16_t contentType) {
laurent's avatar
laurent committed
126
  commonUDP_t *UDPheader=(commonUDP_t *)bufferZone ;
laurent's avatar
laurent committed
127
  UDPheader->contentType=contentType;
laurent's avatar
laurent committed
128 129
  int nbBlocks=UDPheader->nbBlocks;
  int blockId=0;
laurent's avatar
laurent committed
130

laurent's avatar
laurent committed
131 132 133 134 135
  if (nbBlocks <= 0 ) {
    LOG_E(PHY,"FS6: can't send blocks: %d\n", nbBlocks);
    return 0;
  }

laurent's avatar
laurent committed
136
  do {
laurent's avatar
laurent committed
137 138 139 140 141 142 143
    if (blockId > 0 ) {
      commonUDP_t *currentHeader=(commonUDP_t *)bufferZone;
      *currentHeader=*UDPheader;
      currentHeader->blockID=blockId;
      memcpy(commonUDPdata((void *)currentHeader), commonUDPdata(bufferZone), secondHeaderSize);
    }

144
    blockId++;
laurent's avatar
laurent committed
145
    int sz=alignedSize(bufferZone);
laurent's avatar
laurent committed
146 147 148
    // 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
149

laurent's avatar
laurent committed
150
    if ( ret != sz )
laurent's avatar
laurent committed
151 152
      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
153

laurent's avatar
laurent committed
154 155 156
    bufferZone+=sz;
    nbBlocks--;
  } while (nbBlocks);
laurent's avatar
laurent committed
157

158 159
  LOG_D(HW,"Sent: TS: %lu, nb blocks %d, size of first block: %lu \n",
        UDPheader->timestamp, UDPheader->nbBlocks, alignedSize((void *)UDPheader));
laurent's avatar
laurent committed
160 161
  return 0;
}