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

   1 /* -*- c++ -*- */
   2 /*
   3  * Copyright 2007,2008,2009 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 #if defined(HAVE_SOCKET)
  33 #include <netdb.h>
  34 typedef void* optval_t;
  35 #else
  36 // Not posix, assume winsock
  37 #define USING_WINSOCK
  38 #define SHUT_RDWR 2
  39 #define inet_aton(N,A) ( (A)->s_addr = inet_addr(N), ( (A)->s_addr != INADDR_NONE ) )
  40 typedef char* optval_t;
  41 #define ENOPROTOOPT 109
  42 #endif
  43
  44 #define SRC_VERBOSE 0
  45
  46 static int is_error( int perr )
  47 {
  48   // Compare error to posix error code; return nonzero if match.
  49 #if defined(USING_WINSOCK)
  50   // All codes to be checked for must be defined below
  51   int werr = WSAGetLastError();
  52   switch( werr ) {
  53   case WSAETIMEDOUT:
  54     return( perr == EAGAIN );
  55   case WSAENOPROTOOPT:
  56     return( perr == ENOPROTOOPT );
  57   default:
  58     fprintf(stderr,"gr_udp_source/is_error: unknown error %d\n", perr );
  59     throw std::runtime_error("internal error");
  60   }
  61   return 0;
  62 #else
  63   return( perr == errno );
  64 #endif
  65 }
  66
  67 static void report_error( char *msg1, char *msg2 )
  68 {
  69   // Deal with errors, both posix and winsock
  70 #if defined(USING_WINSOCK)
  71   int werr = WSAGetLastError();
  72   fprintf(stderr, "%s: winsock error %d\n", msg1, werr );
  73 #else
  74   perror(msg1);
  75 #endif
  76   if( msg2 != NULL )
  77     throw std::runtime_error(msg2);
  78   return;
  79 }
  80
  81 gr_udp_source::gr_udp_source(size_t itemsize, const char *src,
  82                              unsigned short port_src, int payload_size)
  83   : gr_sync_block ("udp_source",
  84                    gr_make_io_signature(0, 0, 0),
  85                    gr_make_io_signature(1, 1, itemsize)),
  86     d_itemsize(itemsize), d_updated(false), d_payload_size(payload_size), d_residual(0), d_temp_offset(0)
  87 {
  88   int ret = 0;
  89
  90 #if !defined(HAVE_SOCKET) // for Windows (with MinGW)
  91   // initialize winsock DLL
  92   WSADATA wsaData;
  93   int iResult = WSAStartup( MAKEWORD(2,2), &wsaData );
  94   if( iResult != NO_ERROR ) {
  95     report_error( "gr_udp_source WSAStartup", "can't open socket" );
  96   }
  97 #endif
  98
  99   // Set up the address stucture for the source address and port numbers
 100   // Get the source IP address from the host name
 101   struct hostent *hsrc = gethostbyname(src);
 102   if(hsrc) {   // if the source was provided as a host namex
 103     d_ip_src = *(struct in_addr*)hsrc->h_addr_list[0];
 104   }
 105   else { // assume it was specified as an IP address
 106     if((ret=inet_aton(src, &d_ip_src)) == 0) {            // format IP address
 107       report_error("Not a valid source IP address or host name",
 108                    "can't initialize source socket");
 109     }
 110   }
 111
 112   d_port_src = htons(port_src);     // format port number
 113
 114   d_sockaddr_src.sin_family = AF_INET;
 115   d_sockaddr_src.sin_addr   = d_ip_src;
 116   d_sockaddr_src.sin_port   = d_port_src;
 117
 118   d_temp_buff = new char[d_payload_size];   // allow it to hold up to payload_size bytes
 119
 120   open();
 121 }
 122
 123 gr_udp_source_sptr
 124 gr_make_udp_source (size_t itemsize, const char *ipaddr,
 125                     unsigned short port, int payload_size)
 126 {
 127   return gr_udp_source_sptr (new gr_udp_source (itemsize, ipaddr,
 128                                                 port, payload_size));
 129 }
 130
 131 gr_udp_source::~gr_udp_source ()
 132 {
 133   delete [] d_temp_buff;
 134   close();
 135
 136 #if !defined(HAVE_SOCKET) // for Windows (with MinGW)
 137   // free winsock resources
 138   WSACleanup();
 139 #endif
 140 }
 141
 142 bool
 143 gr_udp_source::open()
 144 {
 145   gruel::scoped_lock guard(d_mutex);    // hold mutex for duration of this function
 146   // create socket
 147   d_socket = socket(PF_INET, SOCK_DGRAM, IPPROTO_UDP);
 148   if(d_socket == -1) {
 149     report_error("socket open","can't open socket");
 150   }
 151
 152   // Turn on reuse address
 153   int opt_val = 1;
 154   if(setsockopt(d_socket, SOL_SOCKET, SO_REUSEADDR, (optval_t)&opt_val, sizeof(int)) == -1) {
 155     report_error("SO_REUSEADDR","can't set socket option SO_REUSEADDR");
 156   }
 157
 158   // Don't wait when shutting down
 159   linger lngr;
 160   lngr.l_onoff  = 1;
 161   lngr.l_linger = 0;
 162   if(setsockopt(d_socket, SOL_SOCKET, SO_LINGER, (optval_t)&lngr, sizeof(linger)) == -1) {
 163     if( !is_error(ENOPROTOOPT) ) {  // no SO_LINGER for SOCK_DGRAM on Windows
 164       report_error("SO_LINGER","can't set socket option SO_LINGER");
 165     }
 166   }
 167
 168   // Set a timeout on the receive function to not block indefinitely
 169   // This value can (and probably should) be changed
 170   // Ignored on Cygwin
 171 #if defined(USING_WINSOCK)
 172   DWORD timeout = 1000;  // milliseconds
 173 #else
 174   timeval timeout;
 175   timeout.tv_sec = 1;
 176   timeout.tv_usec = 0;
 177 #endif
 178   if(setsockopt(d_socket, SOL_SOCKET, SO_RCVTIMEO, (optval_t)&timeout, sizeof(timeout)) == -1) {
 179     report_error("SO_RCVTIMEO","can't set socket option SO_RCVTIMEO");
 180   }
 181
 182   // bind socket to an address and port number to listen on
 183   if(bind (d_socket, (sockaddr*)&d_sockaddr_src, sizeof(struct sockaddr)) == -1) {
 184     report_error("socket bind","can't bind socket");
 185   }
 186
 187   d_updated = true;
 188   return d_socket != 0;
 189 }
 190
 191 void
 192 gr_udp_source::close()
 193 {
 194   gruel::scoped_lock guard(d_mutex);    // hold mutex for duration of this function
 195
 196   if (d_socket){
 197     shutdown(d_socket, SHUT_RDWR);
 198     d_socket = 0;
 199   }
 200   d_updated = true;
 201 }
 202
 203 int
 204 gr_udp_source::work (int noutput_items,
 205                      gr_vector_const_void_star &input_items,
 206                      gr_vector_void_star &output_items)
 207 {
 208   char *out = (char *) output_items[0];
 209   ssize_t r=0, nbytes=0, bytes_received=0;
 210   ssize_t total_bytes = (ssize_t)(d_itemsize*noutput_items);
 211
 212   #if SRC_VERBOSE
 213   printf("\nEntered udp_source\n");
 214   #endif
 215
 216   // Remove items from temp buffer if they are in there
 217   if(d_residual) {
 218     nbytes = std::min(d_residual, total_bytes);
 219     memcpy(out, d_temp_buff+d_temp_offset, nbytes);
 220     bytes_received = nbytes;
 221
 222     #if SRC_VERBOSE
 223     printf("\tTemp buff size: %d  offset: %d (bytes_received: %d) (noutput_items: %d)\n",
 224            d_residual, d_temp_offset, bytes_received, noutput_items);
 225     #endif
 226
 227     // Increment pointer
 228     out += bytes_received;
 229
 230     // Update indexing of amount of bytes left in the buffer
 231     d_residual -= nbytes;
 232     d_temp_offset = d_temp_offset+d_residual;
 233   }
 234
 235   while(1) {
 236     // get the data into our output buffer and record the number of bytes
 237     // This is a non-blocking call with a timeout set in the constructor
 238     r = recv(d_socket, d_temp_buff, d_payload_size, 0);  // get the entire payload or the what's available
 239
 240     // Check if there was a problem; forget it if the operation just timed out
 241     if(r == -1) {
 242       if( is_error(EAGAIN) ) {  // handle non-blocking call timeout
 243         #if SRC_VERBOSE
 244         printf("UDP receive timed out\n");
 245         #endif
 246
 247         // Break here to allow the rest of the flow graph time to run and so ctrl-C breaks
 248         break;
 249       }
 250       else {
 251         report_error("udp_source",NULL);
 252         return -1;
 253       }
 254     }
 255     else {
 256       // Calculate the number of bytes we can take from the buffer in this call
 257       nbytes = std::min(r, total_bytes-bytes_received);
 258
 259       // adjust the total number of bytes we have to round down to nearest integer of an itemsize
 260       nbytes -= ((bytes_received+nbytes) % d_itemsize);
 261
 262       // copy the number of bytes we want to look at here
 263       memcpy(out, d_temp_buff, nbytes);
 264
 265       d_residual = r - nbytes;                      // save the number of bytes stored
 266       d_temp_offset=nbytes;                         // reset buffer index
 267
 268       // keep track of the total number of bytes received
 269       bytes_received += nbytes;
 270
 271       // increment the pointer
 272       out += nbytes;
 273
 274       // Immediately return when data comes in
 275       break;
 276     }
 277
 278     #if SNK_VERBOSE
 279     printf("\tbytes received: %d bytes (nbytes: %d)\n", bytes, nbytes);
 280     #endif
 281   }
 282
 283   #if SRC_VERBOSE
 284   printf("Total Bytes Received: %d (bytes_received / noutput_items = %d / %d)\n",
 285          bytes_received, bytes_received, noutput_items);
 286   #endif
 287
 288   // bytes_received is already set to some integer multiple of itemsize
 289   return bytes_received/d_itemsize;
 290 }
 291