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