git.gag.com Git - debian/gnuradio/blob - gnuradio-core/src/lib/io/gr_udp_source.cc

   1 /* -*- c++ -*- */
   2 /*
   3  * Copyright 2007,2008,2009,2010 Free Software Foundation, Inc.
   4  *
   5  * This file is part of GNU Radio
   6  *
   7  * GNU Radio is free software; you can redistribute it and/or modify
   8  * it under the terms of the GNU General Public License as published by
   9  * the Free Software Foundation; either version 3, or (at your option)
  10  * any later version.
  11  *
  12  * GNU Radio is distributed in the hope that it will be useful,
  13  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  15  * GNU General Public License for more details.
  16  *
  17  * You should have received a copy of the GNU General Public License
  18  * along with GNU Radio; see the file COPYING.  If not, write to
  19  * the Free Software Foundation, Inc., 51 Franklin Street,
  20  * Boston, MA 02110-1301, USA.
  21  */
  22
  23 #ifdef HAVE_CONFIG_H
  24 #include "config.h"
  25 #endif
  26 #include <gr_udp_source.h>
  27 #include <gr_io_signature.h>
  28 #include <stdexcept>
  29 #include <errno.h>
  30 #include <stdio.h>
  31 #include <string.h>
  32
  33 #if defined(HAVE_NETDB_H)
  34 #include <netdb.h>
  35 #ifdef HAVE_SYS_TYPES_H
  36 #include <sys/types.h>
  37 #endif
  38 #ifdef HAVE_SYS_SOCKET_H
  39 #include <sys/socket.h>
  40 #endif
  41 typedef void* optval_t;
  42
  43 // ntohs() on FreeBSD may require both netinet/in.h and arpa/inet.h, in order
  44 #if defined(HAVE_NETINET_IN_H)
  45 #include <netinet/in.h>
  46 #endif
  47 #if defined(HAVE_ARPA_INET_H)
  48 #include <arpa/inet.h>
  49 #endif
  50
  51 #elif defined(HAVE_WINDOWS_H)
  52 // if not posix, assume winsock
  53 #define USING_WINSOCK
  54 #include <winsock2.h>
  55 #include <ws2tcpip.h>
  56 #define SHUT_RDWR 2
  57 typedef char* optval_t;
  58 #endif
  59
  60 #define USE_SELECT    1  // non-blocking receive on all platforms
  61 #define USE_RCV_TIMEO 0  // non-blocking receive on all but Cygwin
  62 #define SRC_VERBOSE 0
  63
  64 static int is_error( int perr )
  65 {
  66   // Compare error to posix error code; return nonzero if match.
  67 #if defined(USING_WINSOCK)
  68 #define ENOPROTOOPT 109
  69   // All codes to be checked for must be defined below
  70   int werr = WSAGetLastError();
  71   switch( werr ) {
  72   case WSAETIMEDOUT:
  73     return( perr == EAGAIN );
  74   case WSAENOPROTOOPT:
  75     return( perr == ENOPROTOOPT );
  76   default:
  77     fprintf(stderr,"gr_udp_source/is_error: unknown error %d\n", perr );
  78     throw std::runtime_error("internal error");
  79   }
  80   return 0;
  81 #else
  82   return( perr == errno );
  83 #endif
  84 }
  85
  86 static void report_error( const char *msg1, const char *msg2 )
  87 {
  88   // Deal with errors, both posix and winsock
  89 #if defined(USING_WINSOCK)
  90   int werr = WSAGetLastError();
  91   fprintf(stderr, "%s: winsock error %d\n", msg1, werr );
  92 #else
  93   perror(msg1);
  94 #endif
  95   if( msg2 != NULL )
  96     throw std::runtime_error(msg2);
  97   return;
  98 }
  99
 100 gr_udp_source::gr_udp_source(size_t itemsize, const char *host,
 101                              unsigned short port, int payload_size,
 102                              bool eof, bool wait)
 103   : gr_sync_block ("udp_source",
 104                    gr_make_io_signature(0, 0, 0),
 105                    gr_make_io_signature(1, 1, itemsize)),
 106     d_itemsize(itemsize), d_payload_size(payload_size),
 107     d_eof(eof), d_wait(wait), d_residual(0), d_temp_offset(0)
 108 {
 109   int ret = 0;
 110
 111 #if defined(USING_WINSOCK) // for Windows (with MinGW)
 112   // initialize winsock DLL
 113   WSADATA wsaData;
 114   int iResult = WSAStartup( MAKEWORD(2,2), &wsaData );
 115   if( iResult != NO_ERROR ) {
 116     report_error( "gr_udp_source WSAStartup", "can't open socket" );
 117   }
 118 #endif
 119
 120   // Set up the address stucture for the source address and port numbers
 121   // Get the source IP address from the host name
 122   struct addrinfo *ip_src;      // store the source IP address to use
 123   struct addrinfo hints;
 124   memset( (void*)&hints, 0, sizeof(hints) );
 125   hints.ai_family = AF_INET;
 126   hints.ai_socktype = SOCK_DGRAM;
 127   hints.ai_protocol = IPPROTO_UDP;
 128   hints.ai_flags = AI_PASSIVE;
 129   char port_str[12];
 130   sprintf( port_str, "%d", port );
 131   ret = getaddrinfo( host, port_str, &hints, &ip_src );
 132   if( ret != 0 )
 133     report_error("gr_udp_source/getaddrinfo",
 134                  "can't initialize source socket" );
 135
 136   d_temp_buff = new char[d_payload_size];   // allow it to hold up to payload_size bytes
 137
 138   // create socket
 139   d_socket = socket(ip_src->ai_family, ip_src->ai_socktype,
 140                     ip_src->ai_protocol);
 141   if(d_socket == -1) {
 142     report_error("socket open","can't open socket");
 143   }
 144
 145   // Turn on reuse address
 146   int opt_val = 1;
 147   if(setsockopt(d_socket, SOL_SOCKET, SO_REUSEADDR, (optval_t)&opt_val, sizeof(int)) == -1) {
 148     report_error("SO_REUSEADDR","can't set socket option SO_REUSEADDR");
 149   }
 150
 151   // Don't wait when shutting down
 152   linger lngr;
 153   lngr.l_onoff  = 1;
 154   lngr.l_linger = 0;
 155   if(setsockopt(d_socket, SOL_SOCKET, SO_LINGER, (optval_t)&lngr, sizeof(linger)) == -1) {
 156     if( !is_error(ENOPROTOOPT) ) {  // no SO_LINGER for SOCK_DGRAM on Windows
 157       report_error("SO_LINGER","can't set socket option SO_LINGER");
 158     }
 159   }
 160
 161 #if USE_RCV_TIMEO
 162   // Set a timeout on the receive function to not block indefinitely
 163   // This value can (and probably should) be changed
 164   // Ignored on Cygwin
 165 #if defined(USING_WINSOCK)
 166   DWORD timeout = 1000;  // milliseconds
 167 #else
 168   timeval timeout;
 169   timeout.tv_sec = 1;
 170   timeout.tv_usec = 0;
 171 #endif
 172   if(setsockopt(d_socket, SOL_SOCKET, SO_RCVTIMEO, (optval_t)&timeout, sizeof(timeout)) == -1) {
 173     report_error("SO_RCVTIMEO","can't set socket option SO_RCVTIMEO");
 174   }
 175 #endif // USE_RCV_TIMEO
 176
 177   // bind socket to an address and port number to listen on
 178   if(bind (d_socket, ip_src->ai_addr, ip_src->ai_addrlen) == -1) {
 179     report_error("socket bind","can't bind socket");
 180   }
 181   freeaddrinfo(ip_src);
 182
 183 }
 184
 185 gr_udp_source_sptr
 186 gr_make_udp_source (size_t itemsize, const char *ipaddr,
 187                     unsigned short port, int payload_size, bool eof, bool wait)
 188 {
 189   return gr_udp_source_sptr (new gr_udp_source (itemsize, ipaddr,
 190                                                 port, payload_size, eof, wait));
 191 }
 192
 193 gr_udp_source::~gr_udp_source ()
 194 {
 195   delete [] d_temp_buff;
 196
 197   if (d_socket){
 198     shutdown(d_socket, SHUT_RDWR);
 199 #if defined(USING_WINSOCK)
 200     closesocket(d_socket);
 201 #else
 202     ::close(d_socket);
 203 #endif
 204     d_socket = 0;
 205   }
 206
 207 #if defined(USING_WINSOCK) // for Windows (with MinGW)
 208   // free winsock resources
 209   WSACleanup();
 210 #endif
 211 }
 212
 213 int
 214 gr_udp_source::work (int noutput_items,
 215                      gr_vector_const_void_star &input_items,
 216                      gr_vector_void_star &output_items)
 217 {
 218   char *out = (char *) output_items[0];
 219   ssize_t r=0, nbytes=0, bytes_received=0;
 220   ssize_t total_bytes = (ssize_t)(d_itemsize*noutput_items);
 221
 222   #if SRC_VERBOSE
 223   printf("\nEntered udp_source\n");
 224   #endif
 225
 226   // Remove items from temp buffer if they are in there
 227   if(d_residual) {
 228     nbytes = std::min(d_residual, total_bytes);
 229     memcpy(out, d_temp_buff+d_temp_offset, nbytes);
 230     bytes_received = nbytes;
 231
 232     #if SRC_VERBOSE
 233     printf("\tTemp buff size: %d  offset: %d (bytes_received: %d) (noutput_items: %d)\n",
 234            d_residual, d_temp_offset, bytes_received, noutput_items);
 235     #endif
 236
 237     // Increment pointer
 238     out += bytes_received;
 239
 240     // Update indexing of amount of bytes left in the buffer
 241     d_residual -= nbytes;
 242     d_temp_offset = d_temp_offset+d_residual;
 243   }
 244
 245 #if USE_SELECT
 246   // Use select() to determine when socket is readable
 247   fd_set readfds;
 248   timeval timeout;
 249   timeout.tv_sec = 1;
 250   timeout.tv_usec = 0;
 251 #endif
 252
 253   while(1) {
 254     // get the data into our output buffer and record the number of bytes
 255
 256 #if USE_SELECT
 257     // RCV_TIMEO doesn't work on all systems (e.g., Cygwin)
 258     // use select() instead of, or in addition to RCV_TIMEO
 259     FD_ZERO(&readfds);
 260     FD_SET(d_socket, &readfds);
 261     r = select(FD_SETSIZE, &readfds, NULL, NULL, &timeout);
 262     if(r < 0) {
 263         report_error("udp_source/select",NULL);
 264         return -1;
 265     }
 266     else if(r == 0 ) {  // timed out
 267       if( d_wait ) {
 268         // Allow boost thread interrupt, then try again
 269         boost::this_thread::interruption_point();
 270         continue;
 271       }
 272       else
 273         return -1;
 274     }
 275 #endif // USE_SELECT
 276
 277     // This is a non-blocking call with a timeout set in the constructor
 278     r = recv(d_socket, d_temp_buff, d_payload_size, 0);  // get the entire payload or the what's available
 279
 280     // Check if there was a problem; forget it if the operation just timed out
 281     if(r == -1) {
 282       if( is_error(EAGAIN) ) {  // handle non-blocking call timeout
 283         #if SRC_VERBOSE
 284         printf("UDP receive timed out\n");
 285         #endif
 286
 287         if( d_wait ) {
 288           // Allow boost thread interrupt, then try again
 289           boost::this_thread::interruption_point();
 290           continue;
 291         }
 292         else
 293           return -1;
 294       }
 295       else {
 296         report_error("udp_source/recv",NULL);
 297         return -1;
 298       }
 299     }
 300     else if(r==0) {
 301       if(d_eof) {
 302         // zero-length packet interpreted as EOF
 303
 304         #if SNK_VERBOSE
 305         printf("\tzero-length packet received; returning EOF\n");
 306         #endif
 307
 308         return -1;
 309       }
 310       else{
 311         // do we need to allow boost thread interrupt?
 312         boost::this_thread::interruption_point();
 313         continue;
 314       }
 315     }
 316     else {
 317       // Calculate the number of bytes we can take from the buffer in this call
 318       nbytes = std::min(r, total_bytes-bytes_received);
 319
 320       // adjust the total number of bytes we have to round down to nearest integer of an itemsize
 321       nbytes -= ((bytes_received+nbytes) % d_itemsize);
 322
 323       // copy the number of bytes we want to look at here
 324       memcpy(out, d_temp_buff, nbytes);
 325
 326       d_residual = r - nbytes;                      // save the number of bytes stored
 327       d_temp_offset=nbytes;                         // reset buffer index
 328
 329       // keep track of the total number of bytes received
 330       bytes_received += nbytes;
 331
 332       // increment the pointer
 333       out += nbytes;
 334
 335       // Immediately return when data comes in
 336       break;
 337     }
 338
 339     #if SNK_VERBOSE
 340     printf("\tbytes received: %d bytes (nbytes: %d)\n", bytes, nbytes);
 341     #endif
 342   }
 343
 344   #if SRC_VERBOSE
 345   printf("Total Bytes Received: %d (bytes_received / noutput_items = %d / %d)\n",
 346          bytes_received, bytes_received, noutput_items);
 347   #endif
 348
 349   // bytes_received is already set to some integer multiple of itemsize
 350   return bytes_received/d_itemsize;
 351 }
 352
 353 // Return port number of d_socket
 354 int gr_udp_source::get_port(void)
 355 {
 356   sockaddr_in name;
 357   socklen_t len = sizeof(name);
 358   int ret = getsockname( d_socket, (sockaddr*)&name, &len );
 359   if( ret ) {
 360     report_error("gr_udp_source/getsockname",NULL);
 361     return -1;
 362   }
 363   return ntohs(name.sin_port);
 364 }