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_socket(-1), 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
 132   // FIXME leaks if report_error throws below
 133   ret = getaddrinfo( host, port_str, &hints, &ip_src );
 134   if( ret != 0 )
 135     report_error("gr_udp_source/getaddrinfo",
 136                  "can't initialize source socket" );
 137
 138   // FIXME leaks if report_error throws below
 139   d_temp_buff = new char[d_payload_size];   // allow it to hold up to payload_size bytes
 140
 141   // create socket
 142   d_socket = socket(ip_src->ai_family, ip_src->ai_socktype,
 143                     ip_src->ai_protocol);
 144   if(d_socket == -1) {
 145     report_error("socket open","can't open socket");
 146   }
 147
 148   // Turn on reuse address
 149   int opt_val = 1;
 150   if(setsockopt(d_socket, SOL_SOCKET, SO_REUSEADDR, (optval_t)&opt_val, sizeof(int)) == -1) {
 151     report_error("SO_REUSEADDR","can't set socket option SO_REUSEADDR");
 152   }
 153
 154   // Don't wait when shutting down
 155   linger lngr;
 156   lngr.l_onoff  = 1;
 157   lngr.l_linger = 0;
 158   if(setsockopt(d_socket, SOL_SOCKET, SO_LINGER, (optval_t)&lngr, sizeof(linger)) == -1) {
 159     if( !is_error(ENOPROTOOPT) ) {  // no SO_LINGER for SOCK_DGRAM on Windows
 160       report_error("SO_LINGER","can't set socket option SO_LINGER");
 161     }
 162   }
 163
 164 #if USE_RCV_TIMEO
 165   // Set a timeout on the receive function to not block indefinitely
 166   // This value can (and probably should) be changed
 167   // Ignored on Cygwin
 168 #if defined(USING_WINSOCK)
 169   DWORD timeout = 1000;  // milliseconds
 170 #else
 171   timeval timeout;
 172   timeout.tv_sec = 1;
 173   timeout.tv_usec = 0;
 174 #endif
 175   if(setsockopt(d_socket, SOL_SOCKET, SO_RCVTIMEO, (optval_t)&timeout, sizeof(timeout)) == -1) {
 176     report_error("SO_RCVTIMEO","can't set socket option SO_RCVTIMEO");
 177   }
 178 #endif // USE_RCV_TIMEO
 179
 180   // bind socket to an address and port number to listen on
 181   if(bind (d_socket, ip_src->ai_addr, ip_src->ai_addrlen) == -1) {
 182     report_error("socket bind","can't bind socket");
 183   }
 184   freeaddrinfo(ip_src);
 185
 186 }
 187
 188 gr_udp_source_sptr
 189 gr_make_udp_source (size_t itemsize, const char *ipaddr,
 190                     unsigned short port, int payload_size, bool eof, bool wait)
 191 {
 192   return gr_udp_source_sptr (new gr_udp_source (itemsize, ipaddr,
 193                                                 port, payload_size, eof, wait));
 194 }
 195
 196 gr_udp_source::~gr_udp_source ()
 197 {
 198   delete [] d_temp_buff;
 199
 200   if (d_socket != -1){
 201     shutdown(d_socket, SHUT_RDWR);
 202 #if defined(USING_WINSOCK)
 203     closesocket(d_socket);
 204 #else
 205     ::close(d_socket);
 206 #endif
 207     d_socket = -1;
 208   }
 209
 210 #if defined(USING_WINSOCK) // for Windows (with MinGW)
 211   // free winsock resources
 212   WSACleanup();
 213 #endif
 214 }
 215
 216 int
 217 gr_udp_source::work (int noutput_items,
 218                      gr_vector_const_void_star &input_items,
 219                      gr_vector_void_star &output_items)
 220 {
 221   char *out = (char *) output_items[0];
 222   ssize_t r=0, nbytes=0, bytes_received=0;
 223   ssize_t total_bytes = (ssize_t)(d_itemsize*noutput_items);
 224
 225   #if SRC_VERBOSE
 226   printf("\nEntered udp_source\n");
 227   #endif
 228
 229   // Remove items from temp buffer if they are in there
 230   if(d_residual) {
 231     nbytes = std::min(d_residual, total_bytes);
 232     memcpy(out, d_temp_buff+d_temp_offset, nbytes);
 233     bytes_received = nbytes;
 234
 235     #if SRC_VERBOSE
 236     printf("\tTemp buff size: %d  offset: %d (bytes_received: %d) (noutput_items: %d)\n",
 237            d_residual, d_temp_offset, bytes_received, noutput_items);
 238     #endif
 239
 240     // Increment pointer
 241     out += bytes_received;
 242
 243     // Update indexing of amount of bytes left in the buffer
 244     d_residual -= nbytes;
 245     d_temp_offset = d_temp_offset+d_residual;
 246   }
 247
 248 #if USE_SELECT
 249   // Use select() to determine when socket is readable
 250   fd_set readfds;
 251   timeval timeout;
 252   timeout.tv_sec = 1;
 253   timeout.tv_usec = 0;
 254 #endif
 255
 256   while(1) {
 257     // get the data into our output buffer and record the number of bytes
 258
 259 #if USE_SELECT
 260     // RCV_TIMEO doesn't work on all systems (e.g., Cygwin)
 261     // use select() instead of, or in addition to RCV_TIMEO
 262     FD_ZERO(&readfds);
 263     FD_SET(d_socket, &readfds);
 264     r = select(FD_SETSIZE, &readfds, NULL, NULL, &timeout);
 265     if(r < 0) {
 266         report_error("udp_source/select",NULL);
 267         return -1;
 268     }
 269     else if(r == 0 ) {  // timed out
 270       if( d_wait ) {
 271         // Allow boost thread interrupt, then try again
 272         boost::this_thread::interruption_point();
 273         continue;
 274       }
 275       else
 276         return -1;
 277     }
 278 #endif // USE_SELECT
 279
 280     // This is a non-blocking call with a timeout set in the constructor
 281     r = recv(d_socket, d_temp_buff, d_payload_size, 0);  // get the entire payload or the what's available
 282
 283     // Check if there was a problem; forget it if the operation just timed out
 284     if(r == -1) {
 285       if( is_error(EAGAIN) ) {  // handle non-blocking call timeout
 286         #if SRC_VERBOSE
 287         printf("UDP receive timed out\n");
 288         #endif
 289
 290         if( d_wait ) {
 291           // Allow boost thread interrupt, then try again
 292           boost::this_thread::interruption_point();
 293           continue;
 294         }
 295         else
 296           return -1;
 297       }
 298       else {
 299         report_error("udp_source/recv",NULL);
 300         return -1;
 301       }
 302     }
 303     else if(r==0) {
 304       if(d_eof) {
 305         // zero-length packet interpreted as EOF
 306
 307         #if SNK_VERBOSE
 308         printf("\tzero-length packet received; returning EOF\n");
 309         #endif
 310
 311         return -1;
 312       }
 313       else{
 314         // do we need to allow boost thread interrupt?
 315         boost::this_thread::interruption_point();
 316         continue;
 317       }
 318     }
 319     else {
 320       // Calculate the number of bytes we can take from the buffer in this call
 321       nbytes = std::min(r, total_bytes-bytes_received);
 322
 323       // adjust the total number of bytes we have to round down to nearest integer of an itemsize
 324       nbytes -= ((bytes_received+nbytes) % d_itemsize);
 325
 326       // copy the number of bytes we want to look at here
 327       memcpy(out, d_temp_buff, nbytes);
 328
 329       d_residual = r - nbytes;                      // save the number of bytes stored
 330       d_temp_offset=nbytes;                         // reset buffer index
 331
 332       // keep track of the total number of bytes received
 333       bytes_received += nbytes;
 334
 335       // increment the pointer
 336       out += nbytes;
 337
 338       // Immediately return when data comes in
 339       break;
 340     }
 341
 342     #if SNK_VERBOSE
 343     printf("\tbytes received: %d bytes (nbytes: %d)\n", bytes, nbytes);
 344     #endif
 345   }
 346
 347   #if SRC_VERBOSE
 348   printf("Total Bytes Received: %d (bytes_received / noutput_items = %d / %d)\n",
 349          bytes_received, bytes_received, noutput_items);
 350   #endif
 351
 352   // bytes_received is already set to some integer multiple of itemsize
 353   return bytes_received/d_itemsize;
 354 }
 355
 356 // Return port number of d_socket
 357 int gr_udp_source::get_port(void)
 358 {
 359   sockaddr_in name;
 360   socklen_t len = sizeof(name);
 361   int ret = getsockname( d_socket, (sockaddr*)&name, &len );
 362   if( ret ) {
 363     report_error("gr_udp_source/getsockname",NULL);
 364     return -1;
 365   }
 366   return ntohs(name.sin_port);
 367 }