Updates to udp source/sink (select(), wait, cleanup)
[debian/gnuradio] / gnuradio-core / src / lib / io / gr_udp_sink.cc
old mode 100644 (file)
new mode 100755 (executable)
index 15bcf9a..263d3dd
@@ -1,6 +1,6 @@
 /* -*- c++ -*- */
 /*
- * Copyright 2007,2008 Free Software Foundation, Inc.
+ * Copyright 2007,2008,2009,2010 Free Software Foundation, Inc.
  * 
  * This file is part of GNU Radio
  * 
 #include <gr_udp_sink.h>
 #include <gr_io_signature.h>
 #include <stdexcept>
-#include <netdb.h>
+#include <errno.h>
+#include <stdio.h>
+#include <string.h>
+#if defined(HAVE_NETDB_H)
+typedef void* optval_t;
+#elif defined(HAVE_WINDOWS_H)
+// if not posix, assume winsock
+#define USING_WINSOCK
+#include <winsock2.h>
+#include <ws2tcpip.h>
+#define SHUT_RDWR 2
+typedef char* optval_t;
+#endif
+
+#include <gruel/thread.h>
 
 #define SNK_VERBOSE 0
 
+static int is_error( int perr )
+{
+  // Compare error to posix error code; return nonzero if match.
+#if defined(USING_WINSOCK)
+#define ENOPROTOOPT 109
+#define ECONNREFUSED 111
+  // All codes to be checked for must be defined below
+  int werr = WSAGetLastError();
+  switch( werr ) {
+  case WSAETIMEDOUT:
+    return( perr == EAGAIN );
+  case WSAENOPROTOOPT:
+    return( perr == ENOPROTOOPT );
+  case WSAECONNREFUSED:
+    return( perr == ECONNREFUSED );
+  default:
+    fprintf(stderr,"gr_udp_source/is_error: unknown error %d\n", perr );
+    throw std::runtime_error("internal error");
+  }
+  return 0;
+#else
+  return( perr == errno );
+#endif
+}
+
+static void report_error( const char *msg1, const char *msg2 )
+{
+  // Deal with errors, both posix and winsock
+#if defined(USING_WINSOCK)
+  int werr = WSAGetLastError();
+  fprintf(stderr, "%s: winsock error %d\n", msg1, werr );
+#else
+  perror(msg1);
+#endif
+  if( msg2 != NULL )
+    throw std::runtime_error(msg2);
+  return;
+}
+
 gr_udp_sink::gr_udp_sink (size_t itemsize, 
                          const char *src, unsigned short port_src,
                          const char *dst, unsigned short port_dst,
@@ -37,121 +90,109 @@ gr_udp_sink::gr_udp_sink (size_t itemsize,
   : gr_sync_block ("udp_sink",
                   gr_make_io_signature (1, 1, itemsize),
                   gr_make_io_signature (0, 0, 0)),
-    d_itemsize (itemsize), d_updated(false), d_payload_size(payload_size)
+    d_itemsize (itemsize), d_payload_size(payload_size)
 {
   int ret = 0;
+  struct addrinfo *ip_src;        // store the source ip info
+  struct addrinfo *ip_dst;        // store the destination ip info
+
+#if defined(USING_WINSOCK) // for Windows (with MinGW)
+  // initialize winsock DLL
+  WSADATA wsaData;
+  int iResult = WSAStartup( MAKEWORD(2,2), &wsaData );
+  if( iResult != NO_ERROR ) {
+    report_error( "gr_udp_source WSAStartup", "can't open socket" );
+  }
+#endif
   
   // Set up the address stucture for the source address and port numbers
   // Get the source IP address from the host name
-  struct hostent *hsrc = gethostbyname(src);
-  if(hsrc) {   // if the source was provided as a host namex
-    d_ip_src = *(struct in_addr*)hsrc->h_addr_list[0];    
-  }
-  else { // assume it was specified as an IP address
-    if((ret=inet_aton(src, &d_ip_src)) == 0) {            // format IP address
-      perror("Not a valid source IP address or host name");
-      throw std::runtime_error("can't initialize source socket");
-    }
-  }
+  struct addrinfo hints;
+  memset( (void*)&hints, 0, sizeof(hints) );
+  hints.ai_family = AF_INET;
+  hints.ai_socktype = SOCK_DGRAM;
+  hints.ai_protocol = IPPROTO_UDP;
+  char port_str[7];
+  sprintf( port_str, "%d", port_src );
+  ret = getaddrinfo( src, port_str, &hints, &ip_src );
+  if( ret != 0 )
+    report_error("gr_udp_source/getaddrinfo",
+                "can't initialize source socket" );
 
   // Get the destination IP address from the host name
-  struct hostent *hdst = gethostbyname(dst);
-  if(hdst) {   // if the source was provided as a host namex
-    d_ip_dst = *(struct in_addr*)hdst->h_addr_list[0];    
-  }
-  else { // assume it was specified as an IP address
-    if((ret=inet_aton(dst, &d_ip_dst)) == 0) {            // format IP address
-      perror("Not a valid destination IP address or host name");
-      throw std::runtime_error("can't initialize destination socket");
-    }
-  }
-
-  d_port_src = htons(port_src);           // format port number
-  d_port_dst = htons(port_dst);           // format port number
-
-  d_sockaddr_src.sin_family = AF_INET;
-  d_sockaddr_src.sin_addr   = d_ip_src;
-  d_sockaddr_src.sin_port   = d_port_src;
-
-  d_sockaddr_dst.sin_family = AF_INET;
-  d_sockaddr_dst.sin_addr   = d_ip_dst;
-  d_sockaddr_dst.sin_port   = d_port_dst;
-  
-  open();
-}
-
-// public constructor that returns a shared_ptr
-
-gr_udp_sink_sptr
-gr_make_udp_sink (size_t itemsize, 
-                 const char *src, unsigned short port_src,
-                 const char *dst, unsigned short port_dst,
-                 int payload_size)
-{
-  return gr_udp_sink_sptr (new gr_udp_sink (itemsize, 
-                                           src, port_src,
-                                           dst, port_dst,
-                                           payload_size));
-}
-
-gr_udp_sink::~gr_udp_sink ()
-{
-  close();
-}
-
-bool
-gr_udp_sink::open()
-{
-  omni_mutex_lock l(d_mutex);  // hold mutex for duration of this function
+  sprintf( port_str, "%d", port_dst );
+  ret = getaddrinfo( dst, port_str, &hints, &ip_dst );
+  if( ret != 0 )
+    report_error("gr_udp_source/getaddrinfo",
+                "can't initialize destination socket" );
 
   // create socket
-  if((d_socket = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP)) == -1) {
-    perror("socket open");
-    throw std::runtime_error("can't open socket");
+  d_socket = socket(ip_src->ai_family, ip_src->ai_socktype,
+                   ip_src->ai_protocol);
+  if(d_socket == -1) {
+    report_error("socket open","can't open socket");
   }
 
   // Turn on reuse address
   int opt_val = true;
-  if(setsockopt(d_socket, SOL_SOCKET, SO_REUSEADDR, (void*)&opt_val, sizeof(int)) == -1) {
-    perror("SO_REUSEADDR");
-    throw std::runtime_error("can't set socket option SO_REUSEADDR");
+  if(setsockopt(d_socket, SOL_SOCKET, SO_REUSEADDR, (optval_t)&opt_val, sizeof(int)) == -1) {
+    report_error("SO_REUSEADDR","can't set socket option SO_REUSEADDR");
   }
 
   // Don't wait when shutting down
   linger lngr;
   lngr.l_onoff  = 1;
   lngr.l_linger = 0;
-  if(setsockopt(d_socket, SOL_SOCKET, SO_LINGER, (void*)&lngr, sizeof(linger)) == -1) {
-    perror("SO_LINGER");
-    throw std::runtime_error("can't set socket option SO_LINGER");
+  if(setsockopt(d_socket, SOL_SOCKET, SO_LINGER, (optval_t)&lngr, sizeof(linger)) == -1) {
+    if( !is_error(ENOPROTOOPT) ) {  // no SO_LINGER for SOCK_DGRAM on Windows
+      report_error("SO_LINGER","can't set socket option SO_LINGER");
+    }
   }
 
   // bind socket to an address and port number to listen on
-  if(bind (d_socket, (sockaddr*)&d_sockaddr_src, sizeof(struct sockaddr)) == -1) {
-    perror("socket bind");
-    throw std::runtime_error("can't bind socket");
+  if(bind (d_socket, ip_src->ai_addr, ip_src->ai_addrlen) == -1) {
+    report_error("socket bind","can't bind socket");
   }
 
   // Not sure if we should throw here or allow retries
-  if(connect(d_socket, (sockaddr*)&d_sockaddr_dst, sizeof(struct sockaddr)) == -1) {
-    perror("socket connect");
-    throw std::runtime_error("can't connect to socket");
+  if(connect(d_socket, ip_dst->ai_addr, ip_dst->ai_addrlen) == -1) {
+    report_error("socket connect","can't connect to socket");
   }
 
-  d_updated = true;
-  return d_socket != 0;
+  freeaddrinfo(ip_src);
+  freeaddrinfo(ip_dst);
 }
 
-void
-gr_udp_sink::close()
+// public constructor that returns a shared_ptr
+
+gr_udp_sink_sptr
+gr_make_udp_sink (size_t itemsize, 
+                 const char *src, unsigned short port_src,
+                 const char *dst, unsigned short port_dst,
+                 int payload_size)
 {
-  omni_mutex_lock l(d_mutex);  // hold mutex for duration of this function
+  return gr_udp_sink_sptr (new gr_udp_sink (itemsize, 
+                                           src, port_src,
+                                           dst, port_dst,
+                                           payload_size));
+}
 
+gr_udp_sink::~gr_udp_sink ()
+{
   if (d_socket){
     shutdown(d_socket, SHUT_RDWR);
+#if defined(USING_WINSOCK)
+    closesocket(d_socket);
+#else
+    ::close(d_socket);
+#endif
     d_socket = 0;
   }
-  d_updated = true;
+
+#if defined(USING_WINSOCK) // for Windows (with MinGW)
+  // free winsock resources
+  WSACleanup();
+#endif
 }
 
 int 
@@ -164,7 +205,7 @@ gr_udp_sink::work (int noutput_items,
   ssize_t total_size = noutput_items*d_itemsize;
 
   #if SNK_VERBOSE
-  printf("Entered upd_sink\n");
+  printf("Entered udp_sink\n");
   #endif
 
   while(bytes_sent <  total_size) {
@@ -172,8 +213,12 @@ gr_udp_sink::work (int noutput_items,
   
     r = send(d_socket, (in+bytes_sent), bytes_to_send, 0);
     if(r == -1) {         // error on send command
-      perror("udp_sink"); // there should be no error case where this function 
-      return -1;          // should not exit immediately
+      if( is_error(ECONNREFUSED) )
+       r = bytes_to_send;  // discard data until receiver is started
+      else {
+       report_error("udp_sink",NULL); // there should be no error case where
+       return -1;                  // this function should not exit immediately
+      }
     }
     bytes_sent += r;