Merge branch 'master' into next

* master:
  gnuradio-core: update copyrights
  gnuradio-core: allow swig to handle exceptions in UDP source/sink
  grc: update UDP source and sink block wrappers
  Simplify USE_SELECT usage
  Return immediately when using d_residual.
  Defend against a peer that sends an invalid message length.
  Move initialization of select timeout
  Correct update of d_temp_offset (parallel construction)
  Identify memory leaks that occur on error conditions
  Use -1 as file descriptor "not open" value instead of 0
  Add additional conditionalization of networking includes
  Flush pending errors in gr_udp_sink on disconnect()
  Rework UDP source and sink, with incompatible API changes
  Updates to udp source/sink (select(), wait, cleanup)
  Discard data in gr_udp_sink until receiver is started.
  Use getaddrinfo in gr_udp_{source,sink}
  Changes to gr_udp_{source,sink} for MinGW
  Ignore ENOPROTOOPT return from setsockopt(SO_LINGER)

18 files changed, 709 insertions, 382 deletions

diff --git a/configure.ac b/configure.ac
index 124485bea..c912d9891 100644
--- a/configure.ac
+++ b/configure.ac
@@ -170,9 +170,9 @@ AC_HEADER_SYS_WAIT
 AC_CHECK_HEADERS(fcntl.h limits.h strings.h time.h sys/ioctl.h sys/time.h unistd.h)
 AC_CHECK_HEADERS(linux/ppdev.h dev/ppbus/ppi.h sys/mman.h sys/select.h sys/types.h)
 AC_CHECK_HEADERS(sys/resource.h stdint.h sched.h signal.h sys/syscall.h malloc.h)
-AC_CHECK_HEADERS(netinet/in.h)
 AC_CHECK_HEADERS(windows.h)
 AC_CHECK_HEADERS(vec_types.h)
+AC_CHECK_HEADERS(netdb.h netinet/in.h arpa/inet.h sys/types.h sys/socket.h)
 
 dnl Checks for typedefs, structures, and compiler characteristics.
 AC_C_CONST
diff --git a/gnuradio-core/src/lib/io/gr_udp_sink.cc b/gnuradio-core/src/lib/io/gr_udp_sink.cc
index d37adfb8a..3084a848b 100644
--- a/gnuradio-core/src/lib/io/gr_udp_sink.cc
+++ b/gnuradio-core/src/lib/io/gr_udp_sink.cc
@@ -1,6 +1,6 @@
 /* -*- c++ -*- */
 /*
- * Copyright 2007,2008,2009 Free Software Foundation, Inc.
+ * Copyright 2007,2008,2009,2010 Free Software Foundation, Inc.
  * 
  * This file is part of GNU Radio
  * 
@@ -26,13 +26,24 @@
 #include <gr_udp_sink.h>
 #include <gr_io_signature.h>
 #include <stdexcept>
-#if defined(HAVE_SOCKET)
-#include <netdb.h>
+#include <errno.h>
 #include <stdio.h>
+#include <string.h>
+#if defined(HAVE_NETDB_H)
+#include <netdb.h>
+#ifdef HAVE_SYS_TYPES_H
+#include <sys/types.h>
+#endif
+#ifdef HAVE_SYS_SOCKET_H
+#include <sys/socket.h>  //usually included by <netdb.h>?
+#endif
 typedef void* optval_t;
-#else
+#elif defined(HAVE_WINDOWS_H)
+// if not posix, assume winsock
+#define USING_WINSOCK
+#include <winsock2.h>
+#include <ws2tcpip.h>
 #define SHUT_RDWR 2
-#define inet_aton(N,A) ( (A)->s_addr = inet_addr(N), ( (A)->s_addr != INADDR_NONE ) )
 typedef char* optval_t;
 #endif
 
@@ -40,91 +51,67 @@ typedef char* optval_t;
 
 #define SNK_VERBOSE 0
 
-gr_udp_sink::gr_udp_sink (size_t itemsize, 
-			  const char *src, unsigned short port_src,
-			  const char *dst, unsigned short port_dst,
-			  int payload_size)
-  : 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)
+static int is_error( int perr )
 {
-  int ret = 0;
-  
-  // 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];    
+  // 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");
   }
-  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");
-    }
-  }
-
-  // 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));
+  return 0;
+#else
+  return( perr == errno );
+#endif
 }
 
-gr_udp_sink::~gr_udp_sink ()
+static void report_error( const char *msg1, const char *msg2 )
 {
-  close();
+  // 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;
 }
 
-bool
-gr_udp_sink::open()
+gr_udp_sink::gr_udp_sink (size_t itemsize, 
+			  const char *host, unsigned short port,
+			  int payload_size, bool eof)
+  : gr_sync_block ("udp_sink",
+		   gr_make_io_signature (1, 1, itemsize),
+		   gr_make_io_signature (0, 0, 0)),
+    d_itemsize (itemsize), d_payload_size(payload_size), d_eof(eof),
+    d_socket(-1), d_connected(false)
 {
-  gruel::scoped_lock guard(d_mutex);	// hold mutex for duration of this function
-
-  // create socket
-  if((d_socket = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP)) == -1) {
-    perror("socket open");
-    throw std::runtime_error("can't open socket");
+#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
 
-  // Turn on reuse address
-  int opt_val = true;
-  if(setsockopt(d_socket, SOL_SOCKET, SO_REUSEADDR, (optval_t)&opt_val, sizeof(int)) == -1) {
-    perror("SO_REUSEADDR");
-    throw std::runtime_error("can't set socket option SO_REUSEADDR");
+  // create socket
+  d_socket = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP);
+  if(d_socket == -1) {
+    report_error("socket open","can't open socket");
   }
 
   // Don't wait when shutting down
@@ -132,36 +119,46 @@ gr_udp_sink::open()
   lngr.l_onoff  = 1;
   lngr.l_linger = 0;
   if(setsockopt(d_socket, SOL_SOCKET, SO_LINGER, (optval_t)&lngr, sizeof(linger)) == -1) {
-    perror("SO_LINGER");
-    throw std::runtime_error("can't set socket option SO_LINGER");
+    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");
-  }
+  // Get the destination address
+  connect(host, port);
+}
 
-  // 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");
-  }
+// public constructor that returns a shared_ptr
 
-  d_updated = true;
-  return d_socket != 0;
+gr_udp_sink_sptr
+gr_make_udp_sink (size_t itemsize, 
+		  const char *host, unsigned short port,
+		  int payload_size, bool eof)
+{
+  return gr_udp_sink_sptr (new gr_udp_sink (itemsize, 
+					    host, port,
+					    payload_size, eof));
 }
 
-void
-gr_udp_sink::close()
+gr_udp_sink::~gr_udp_sink ()
 {
-  gruel::scoped_lock guard(d_mutex);	// hold mutex for duration of this function
+  if (d_connected)
+    disconnect();
 
-  if (d_socket){
+  if (d_socket != -1){
     shutdown(d_socket, SHUT_RDWR);
-    d_socket = 0;
+#if defined(USING_WINSOCK)
+    closesocket(d_socket);
+#else
+    ::close(d_socket);
+#endif
+    d_socket = -1;
   }
-  d_updated = true;
+
+#if defined(USING_WINSOCK) // for Windows (with MinGW)
+  // free winsock resources
+  WSACleanup();
+#endif
 }
 
 int 
@@ -174,21 +171,31 @@ 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
 
+  gruel::scoped_lock guard(d_mutex);  // protect d_socket
+
   while(bytes_sent <  total_size) {
     bytes_to_send = std::min((ssize_t)d_payload_size, (total_size-bytes_sent));
   
-    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(d_connected) {
+      r = send(d_socket, (in+bytes_sent), bytes_to_send, 0);
+      if(r == -1) {         // error on send command
+	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
+	}
+      }
     }
+    else
+      r = bytes_to_send;  // discarded for lack of connection
     bytes_sent += r;
     
     #if SNK_VERBOSE
-    printf("\tbyte sent: %d bytes\n", bytes);
+    printf("\tbyte sent: %d bytes\n", r);
     #endif
   }
 
@@ -198,3 +205,98 @@ gr_udp_sink::work (int noutput_items,
 
   return noutput_items;
 }
+
+void gr_udp_sink::connect( const char *host, unsigned short port )
+{
+  if(d_connected)
+    disconnect();
+
+  if(host != NULL ) {
+    // Get the destination address
+    struct addrinfo *ip_dst;
+    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[12];
+    sprintf( port_str, "%d", port );
+
+    // FIXME leaks if report_error throws below
+    int ret = getaddrinfo( host, port_str, &hints, &ip_dst );
+    if( ret != 0 )
+      report_error("gr_udp_source/getaddrinfo",
+		   "can't initialize destination socket" );
+
+    // don't need d_mutex lock when !d_connected
+    if(::connect(d_socket, ip_dst->ai_addr, ip_dst->ai_addrlen) == -1) {
+      report_error("socket connect","can't connect to socket");
+    }
+    d_connected = true;
+
+    freeaddrinfo(ip_dst);
+  }
+
+  return;
+}
+
+void gr_udp_sink::disconnect()
+{
+  if(!d_connected)
+    return;
+
+  #if SNK_VERBOSE
+  printf("gr_udp_sink disconnecting\n");
+  #endif
+
+  gruel::scoped_lock guard(d_mutex);  // protect d_socket from work()
+
+  // Send a few zero-length packets to signal receiver we are done
+  if(d_eof) {
+    int i;
+    for( i = 0; i < 3; i++ )
+      (void) send( d_socket, NULL, 0, 0 );  // ignore errors
+  }
+
+  // Sending EOF can produce ERRCONNREFUSED errors that won't show up
+  //  until the next send or recv, which might confuse us if it happens
+  //  on a new connection.  The following does a nonblocking recv to
+  //  clear any such errors.
+  timeval timeout;
+  timeout.tv_sec = 0;    // zero time for immediate return
+  timeout.tv_usec = 0;
+  fd_set readfds;
+  FD_ZERO(&readfds);
+  FD_SET(d_socket, &readfds);
+  int r = select(FD_SETSIZE, &readfds, NULL, NULL, &timeout);
+  if(r < 0) {
+      #if SNK_VERBOSE
+      report_error("udp_sink/select",NULL);
+      #endif
+  }
+  else if(r > 0) {  // call recv() to get error return
+    r = recv(d_socket, (char*)&readfds, sizeof(readfds), 0);
+    if(r < 0) {
+	#if SNK_VERBOSE
+	report_error("udp_sink/recv",NULL);
+	#endif
+    }
+  }
+
+  // Since I can't find any way to disconnect a datagram socket in Cygwin,
+  // we just leave it connected but disable sending.
+#if 0
+  // zeroed address structure should reset connection
+  struct sockaddr addr;
+  memset( (void*)&addr, 0, sizeof(addr) );
+  // addr.sa_family = AF_UNSPEC;  // doesn't work on Cygwin
+  // addr.sa_family = AF_INET;  // doesn't work on Cygwin
+
+  if(::connect(d_socket, &addr, sizeof(addr)) == -1)
+    report_error("socket connect","can't connect to socket");
+#endif
+
+  d_connected = false;
+
+  return;
+}
diff --git a/gnuradio-core/src/lib/io/gr_udp_sink.h b/gnuradio-core/src/lib/io/gr_udp_sink.h
index f22b92dd0..421d514a4 100644
--- a/gnuradio-core/src/lib/io/gr_udp_sink.h
+++ b/gnuradio-core/src/lib/io/gr_udp_sink.h
@@ -1,6 +1,6 @@
 /* -*- c++ -*- */
 /*
- * Copyright 2007,2008,2009 Free Software Foundation, Inc.
+ * Copyright 2007,2008,2009,2010 Free Software Foundation, Inc.
  * 
  * This file is part of GNU Radio
  * 
@@ -24,18 +24,6 @@
 #define INCLUDED_GR_UDP_SINK_H
 
 #include <gr_sync_block.h>
-#include <boost/thread.hpp>
-#if defined(HAVE_SOCKET)
-#include <sys/socket.h>
-#include <arpa/inet.h>
-#elif defined(HAVE_WINDOWS_H)
-#include <winsock2.h>
-#include <windows.h>
-#endif
-#if defined(HAVE_NETINET_IN_H)
-#include <netinet/in.h>
-#endif
-
 #include <gruel/thread.h>
 
 class gr_udp_sink;
@@ -43,85 +31,75 @@ typedef boost::shared_ptr<gr_udp_sink> gr_udp_sink_sptr;
 
 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=1472);
+		  const char *host, unsigned short port,
+		  int payload_size=1472, bool eof=true);
 
 /*!
  * \brief Write stream to an UDP socket.
  * \ingroup sink_blk
  * 
  * \param itemsize     The size (in bytes) of the item datatype
- * \param src          The source address as either the host name or the 'numbers-and-dots'
- *                     IP address
- * \param port_src     Destination port to bind to (0 allows socket to choose an appropriate port)
- * \param dst          The destination address as either the host name or the 'numbers-and-dots'
- *                     IP address
- * \param port_dst     Destination port to connect to
- * \param payload_size UDP payload size by default set to 
- *                     1472 = (1500 MTU - (8 byte UDP header) - (20 byte IP header))
+ * \param host         The name or IP address of the receiving host; use
+ *                     NULL or None for no connection
+ * \param port         Destination port to connect to on receiving host
+ * \param payload_size UDP payload size by default set to 1472 =
+ *                     (1500 MTU - (8 byte UDP header) - (20 byte IP header))
+ * \param eof          Send zero-length packet on disconnect
  */
 
 class gr_udp_sink : public gr_sync_block
 {
   friend 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);
+					    const char *host,
+					    unsigned short port,
+					    int payload_size, bool eof);
  private:
   size_t	d_itemsize;
-  bool		d_updated;
-  gruel::mutex	d_mutex;
 
-  int            d_payload_size;    // maximum transmission unit (packet length)
-  int            d_socket;          // handle to socket
-  int            d_socket_rcv;      // handle to socket retuned in the accept call
-  struct in_addr d_ip_src;          // store the source ip info
-  struct in_addr d_ip_dst;          // store the destination ip info
-  unsigned short d_port_src;        // the port number to open for connections to this service
-  unsigned short d_port_dst;        // port number of the remove system
-  struct sockaddr_in    d_sockaddr_src;    // store the source sockaddr data (formatted IP address and port number)
-  struct sockaddr_in    d_sockaddr_dst;    // store the destination sockaddr data (formatted IP address and port number)
+  int           d_payload_size;    // maximum transmission unit (packet length)
+  bool          d_eof;             // send zero-length packet on disconnect
+  int           d_socket;          // handle to socket
+  bool          d_connected;       // are we connected?
+  gruel::mutex  d_mutex;           // protects d_socket and d_connected
 
  protected:
   /*!
    * \brief UDP Sink Constructor
    * 
    * \param itemsize     The size (in bytes) of the item datatype
-   * \param src          The source address as either the host name or the 'numbers-and-dots'
-   *                     IP address
-   * \param port_src     Destination port to bind to (0 allows socket to choose an appropriate port)
-   * \param dst          The destination address as either the host name or the 'numbers-and-dots'
-   *                     IP address
-   * \param port_dst     Destination port to connect to
+   * \param host         The name or IP address of the receiving host; use
+   *                     NULL or None for no connection
+   * \param port         Destination port to connect to on receiving host
    * \param payload_size UDP payload size by default set to 
    *                     1472 = (1500 MTU - (8 byte UDP header) - (20 byte IP header))
+   * \param eof          Send zero-length packet on disconnect
    */
   gr_udp_sink (size_t itemsize, 
-	       const char *src, unsigned short port_src,
-	       const char *dst, unsigned short port_dst,
-	       int payload_size);
+	       const char *host, unsigned short port,
+	       int payload_size, bool eof);
 
  public:
   ~gr_udp_sink ();
 
-  /*!
-   * \brief open a socket specified by the port and ip address info
-   *
-   * Opens a socket, binds to the address, and makes connectionless association
-   * over UDP. If any of these fail, the fuction retuns the error and exits.
-   */
-  bool open();
+  /*! \brief return the PAYLOAD_SIZE of the socket */
+  int payload_size() { return d_payload_size; }
 
-  /*!
-   * \brief Close current socket.
+  /*! \brief Change the connection to a new destination
+   *
+   * \param host         The name or IP address of the receiving host; use
+   *                     NULL or None to break the connection without closing
+   * \param port         Destination port to connect to on receiving host
    *
-   * Shuts down read/write on the socket
+   * Calls disconnect() to terminate any current connection first.
    */
-  void close();
+  void connect( const char *host, unsigned short port );
 
-  /*! \brief return the PAYLOAD_SIZE of the socket */
-  int payload_size() { return d_payload_size; }
+  /*! \brief Send zero-length packet (if eof is requested) then stop sending
+   *
+   * Zero-byte packets can be interpreted as EOF by gr_udp_source.  Note that
+   * disconnect occurs automatically when the sink is destroyed, but not when
+   * its top_block stops.*/
+  void disconnect();
 
   // should we export anything else?
 
diff --git a/gnuradio-core/src/lib/io/gr_udp_sink.i b/gnuradio-core/src/lib/io/gr_udp_sink.i
index 0f37b477b..a71006ae0 100644
--- a/gnuradio-core/src/lib/io/gr_udp_sink.i
+++ b/gnuradio-core/src/lib/io/gr_udp_sink.i
@@ -1,6 +1,6 @@
 /* -*- c++ -*- */
 /*
- * Copyright 2007 Free Software Foundation, Inc.
+ * Copyright 2007,2010 Free Software Foundation, Inc.
  * 
  * This file is part of GNU Radio
  * 
@@ -25,22 +25,22 @@ GR_SWIG_BLOCK_MAGIC(gr,udp_sink)
 
 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=1472);
+		  const char *host, unsigned short port,
+		  int payload_size=1472, bool eof=true) throw (std::runtime_error);
 
 class gr_udp_sink : public gr_sync_block
 {
  protected:
   gr_udp_sink (size_t itemsize, 
-	       const char *src, unsigned short port_src,
-	       const char *dst, unsigned short port_dst,
-	       int payload_size);
-
-  bool open();
-  void close();
-  int payload_size() { return d_payload_size; }
+	       const char *host, unsigned short port,
+	       int payload_size, bool eof) 
+    throw (std::runtime_error);
 
  public:
   ~gr_udp_sink ();
+
+  int payload_size() { return d_payload_size; }
+  void connect( const char *host, unsigned short port );
+  void disconnect();
+
 };
diff --git a/gnuradio-core/src/lib/io/gr_udp_source.cc b/gnuradio-core/src/lib/io/gr_udp_source.cc
index d76d0ee32..fea9a26ba 100644
--- a/gnuradio-core/src/lib/io/gr_udp_source.cc
+++ b/gnuradio-core/src/lib/io/gr_udp_source.cc
@@ -1,6 +1,6 @@
 /* -*- c++ -*- */
 /*
- * Copyright 2007,2008,2009 Free Software Foundation, Inc.
+ * Copyright 2007,2008,2009,2010 Free Software Foundation, Inc.
  * 
  * This file is part of GNU Radio
  * 
@@ -29,80 +29,126 @@
 #include <errno.h>
 #include <stdio.h>
 #include <string.h>
-#if defined(HAVE_SOCKET)
+
+#if defined(HAVE_NETDB_H)
 #include <netdb.h>
+#ifdef HAVE_SYS_TYPES_H
+#include <sys/types.h>
+#endif
+#ifdef HAVE_SYS_SOCKET_H
+#include <sys/socket.h>
+#endif
 typedef void* optval_t;
-#else
+
+// ntohs() on FreeBSD may require both netinet/in.h and arpa/inet.h, in order
+#if defined(HAVE_NETINET_IN_H)
+#include <netinet/in.h>
+#endif
+#if defined(HAVE_ARPA_INET_H)
+#include <arpa/inet.h>
+#endif
+
+#elif defined(HAVE_WINDOWS_H)
+// if not posix, assume winsock
+#define USING_WINSOCK
+#include <winsock2.h>
+#include <ws2tcpip.h>
 #define SHUT_RDWR 2
-#define inet_aton(N,A) ( (A)->s_addr = inet_addr(N), ( (A)->s_addr != INADDR_NONE ) )
 typedef char* optval_t;
 #endif
 
+#define USE_SELECT    1  // non-blocking receive on all platforms
+#define USE_RCV_TIMEO 0  // non-blocking receive on all but Cygwin
 #define SRC_VERBOSE 0
 
-gr_udp_source::gr_udp_source(size_t itemsize, const char *src, 
-			     unsigned short port_src, int payload_size)
+static int is_error( int perr )
+{
+  // Compare error to posix error code; return nonzero if match.
+#if defined(USING_WINSOCK)
+#define ENOPROTOOPT 109
+  // 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 );
+  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_source::gr_udp_source(size_t itemsize, const char *host, 
+			     unsigned short port, int payload_size,
+			     bool eof, bool wait)
   : gr_sync_block ("udp_source",
 		   gr_make_io_signature(0, 0, 0),
 		   gr_make_io_signature(1, 1, itemsize)),
-    d_itemsize(itemsize), d_updated(false), d_payload_size(payload_size), d_residual(0), d_temp_offset(0)
+    d_itemsize(itemsize), d_payload_size(payload_size),
+    d_eof(eof), d_wait(wait), d_socket(-1), d_residual(0), d_temp_offset(0)
 {
   int ret = 0;
+
+#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 *ip_src;      // store the source IP address to use
+  struct addrinfo hints;
+  memset( (void*)&hints, 0, sizeof(hints) );
+  hints.ai_family = AF_INET;
+  hints.ai_socktype = SOCK_DGRAM;
+  hints.ai_protocol = IPPROTO_UDP;
+  hints.ai_flags = AI_PASSIVE;
+  char port_str[12];
+  sprintf( port_str, "%d", port );
 
-  d_port_src = htons(port_src);     // 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;
+  // FIXME leaks if report_error throws below
+  ret = getaddrinfo( host, port_str, &hints, &ip_src );
+  if( ret != 0 )
+    report_error("gr_udp_source/getaddrinfo",
+		 "can't initialize source socket" );
 
+  // FIXME leaks if report_error throws below
   d_temp_buff = new char[d_payload_size];   // allow it to hold up to payload_size bytes
-  
-  open();
-}
 
-gr_udp_source_sptr
-gr_make_udp_source (size_t itemsize, const char *ipaddr, 
-		    unsigned short port, int payload_size)
-{
-  return gr_udp_source_sptr (new gr_udp_source (itemsize, ipaddr, 
-						port, payload_size));
-}
-
-gr_udp_source::~gr_udp_source ()
-{
-  delete [] d_temp_buff;
-  close();
-}
-
-bool
-gr_udp_source::open()
-{
-  gruel::scoped_lock guard(d_mutex);	// hold mutex for duration of this function
   // create socket
-  d_socket = socket(PF_INET, SOCK_DGRAM, IPPROTO_UDP);
+  d_socket = socket(ip_src->ai_family, ip_src->ai_socktype,
+		    ip_src->ai_protocol);
   if(d_socket == -1) {
-    perror("socket open");
-    throw std::runtime_error("can't open socket");
+    report_error("socket open","can't open socket");
   }
 
   // Turn on reuse address
   int opt_val = 1;
   if(setsockopt(d_socket, SOL_SOCKET, SO_REUSEADDR, (optval_t)&opt_val, sizeof(int)) == -1) {
-    perror("SO_REUSEADDR");
-    throw std::runtime_error("can't set socket option SO_REUSEADDR");
+    report_error("SO_REUSEADDR","can't set socket option SO_REUSEADDR");
   }
 
   // Don't wait when shutting down
@@ -110,40 +156,61 @@ gr_udp_source::open()
   lngr.l_onoff  = 1;
   lngr.l_linger = 0;
   if(setsockopt(d_socket, SOL_SOCKET, SO_LINGER, (optval_t)&lngr, sizeof(linger)) == -1) {
-    perror("SO_LINGER");
-    throw std::runtime_error("can't set socket option SO_LINGER");
+    if( !is_error(ENOPROTOOPT) ) {  // no SO_LINGER for SOCK_DGRAM on Windows
+      report_error("SO_LINGER","can't set socket option SO_LINGER");
+    }
   }
 
+#if USE_RCV_TIMEO
   // Set a timeout on the receive function to not block indefinitely
   // This value can (and probably should) be changed
+  // Ignored on Cygwin
+#if defined(USING_WINSOCK)
+  DWORD timeout = 1000;  // milliseconds
+#else
   timeval timeout;
   timeout.tv_sec = 1;
   timeout.tv_usec = 0;
+#endif
   if(setsockopt(d_socket, SOL_SOCKET, SO_RCVTIMEO, (optval_t)&timeout, sizeof(timeout)) == -1) {
-    perror("SO_RCVTIMEO");
-    throw std::runtime_error("can't set socket option SO_RCVTIMEO");
+    report_error("SO_RCVTIMEO","can't set socket option SO_RCVTIMEO");
   }
+#endif // USE_RCV_TIMEO
 
   // 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");
   }
-  
-  d_updated = true;
-  return d_socket != 0;
+  freeaddrinfo(ip_src);
+
 }
 
-void
-gr_udp_source::close()
+gr_udp_source_sptr
+gr_make_udp_source (size_t itemsize, const char *ipaddr, 
+		    unsigned short port, int payload_size, bool eof, bool wait)
 {
-  gruel::scoped_lock guard(d_mutex);	// hold mutex for duration of this function
+  return gr_udp_source_sptr (new gr_udp_source (itemsize, ipaddr, 
+						port, payload_size, eof, wait));
+}
+
+gr_udp_source::~gr_udp_source ()
+{
+  delete [] d_temp_buff;
 
-  if (d_socket){
+  if (d_socket != -1){
     shutdown(d_socket, SHUT_RDWR);
-    d_socket = 0;
+#if defined(USING_WINSOCK)
+    closesocket(d_socket);
+#else
+    ::close(d_socket);
+#endif
+    d_socket = -1;
   }
-  d_updated = true;
+
+#if defined(USING_WINSOCK) // for Windows (with MinGW)
+  // free winsock resources
+  WSACleanup();
+#endif
 }
 
 int 
@@ -175,29 +242,85 @@ gr_udp_source::work (int noutput_items,
     
     // Update indexing of amount of bytes left in the buffer
     d_residual -= nbytes;
-    d_temp_offset = d_temp_offset+d_residual;
+    d_temp_offset += nbytes;
+
+    // Return now with what we've got.
+    assert(nbytes % d_itemsize == 0);
+    return nbytes/d_itemsize;
   }
 
   while(1) {
     // get the data into our output buffer and record the number of bytes
+
+#if USE_SELECT
+    // RCV_TIMEO doesn't work on all systems (e.g., Cygwin)
+    // use select() instead of, or in addition to RCV_TIMEO
+    fd_set readfds;
+    timeval timeout;
+    timeout.tv_sec = 1;	  // Init timeout each iteration.  Select can modify it.
+    timeout.tv_usec = 0;
+    FD_ZERO(&readfds);
+    FD_SET(d_socket, &readfds);
+    r = select(FD_SETSIZE, &readfds, NULL, NULL, &timeout);
+    if(r < 0) {
+	report_error("udp_source/select",NULL);
+	return -1;
+    }
+    else if(r == 0 ) {  // timed out
+      if( d_wait ) {
+	// Allow boost thread interrupt, then try again
+	boost::this_thread::interruption_point();
+	continue;
+      }
+      else
+	return -1;
+    }
+#endif // USE_SELECT
+
     // This is a non-blocking call with a timeout set in the constructor
     r = recv(d_socket, d_temp_buff, d_payload_size, 0);  // get the entire payload or the what's available
 
+    // If r > 0, round it down to a multiple of d_itemsize 
+    // (If sender is broken, don't propagate problem)
+    if (r > 0)
+      r = (r/d_itemsize) * d_itemsize;
+
     // Check if there was a problem; forget it if the operation just timed out
     if(r == -1) {
-      if(errno == EAGAIN) {  // handle non-blocking call timeout
+      if( is_error(EAGAIN) ) {  // handle non-blocking call timeout
         #if SRC_VERBOSE
 	printf("UDP receive timed out\n"); 
         #endif
 
-	// Break here to allow the rest of the flow graph time to run and so ctrl-C breaks
-	break;
+	if( d_wait ) {
+	  // Allow boost thread interrupt, then try again
+	  boost::this_thread::interruption_point();
+	  continue;
+	}
+	else
+	  return -1;
       }
       else {
-	perror("udp_source");
+	report_error("udp_source/recv",NULL);
 	return -1;
       }
     }
+    else if(r==0) {
+      if(d_eof) {
+	// zero-length packet interpreted as EOF
+
+	#if SNK_VERBOSE
+	printf("\tzero-length packet received; returning EOF\n");
+	#endif
+
+	return -1;
+      }
+      else{
+	// do we need to allow boost thread interrupt?
+	boost::this_thread::interruption_point();
+	continue;
+      }
+    }
     else {
       // Calculate the number of bytes we can take from the buffer in this call
       nbytes = std::min(r, total_bytes-bytes_received);
@@ -235,3 +358,15 @@ gr_udp_source::work (int noutput_items,
   return bytes_received/d_itemsize;
 }
 
+// Return port number of d_socket
+int gr_udp_source::get_port(void)
+{
+  sockaddr_in name;
+  socklen_t len = sizeof(name);
+  int ret = getsockname( d_socket, (sockaddr*)&name, &len );
+  if( ret ) {
+    report_error("gr_udp_source/getsockname",NULL);
+    return -1;
+  }
+  return ntohs(name.sin_port);
+}
diff --git a/gnuradio-core/src/lib/io/gr_udp_source.h b/gnuradio-core/src/lib/io/gr_udp_source.h
index 61d719e4d..e23231aa7 100644
--- a/gnuradio-core/src/lib/io/gr_udp_source.h
+++ b/gnuradio-core/src/lib/io/gr_udp_source.h
@@ -1,6 +1,6 @@
 /* -*- c++ -*- */
 /*
- * Copyright 2007,2008,2009 Free Software Foundation, Inc.
+ * Copyright 2007,2008,2009,2010 Free Software Foundation, Inc.
  * 
  * This file is part of GNU Radio
  * 
@@ -24,54 +24,48 @@
 #define INCLUDED_GR_UDP_SOURCE_H
 
 #include <gr_sync_block.h>
-#if defined(HAVE_SOCKET)
-#include <sys/socket.h>
-#include <arpa/inet.h>
-#elif defined(HAVE_WINDOWS_H)
-#include <winsock2.h>
-#include <windows.h>
-#endif
-#if defined(HAVE_NETINET_IN_H)
-#include <netinet/in.h>
-#endif
-
 #include <gruel/thread.h>
 
 class gr_udp_source;
 typedef boost::shared_ptr<gr_udp_source> gr_udp_source_sptr;
 
-gr_udp_source_sptr gr_make_udp_source(size_t itemsize, const char *src, 
-				      unsigned short port_src, int payload_size=1472);
+gr_udp_source_sptr gr_make_udp_source(size_t itemsize, const char *host, 
+				      unsigned short port,
+				      int payload_size=1472,
+				      bool eof=true, bool wait=true);
 
 /*! 
  * \brief Read stream from an UDP socket.
  * \ingroup source_blk
  *
  * \param itemsize     The size (in bytes) of the item datatype
- * \param src          The source address as either the host name or the 'numbers-and-dots'
- *                     IP address
- * \param port_src     The port number on which the socket listens for data
- * \param payload_size UDP payload size by default set to 
- *                     1472 = (1500 MTU - (8 byte UDP header) - (20 byte IP header))
+ * \param host         The name or IP address of the receiving host; can be
+ *                     NULL, None, or "0.0.0.0" to allow reading from any
+ *                     interface on the host
+ * \param port         The port number on which to receive data; use 0 to
+ *                     have the system assign an unused port number
+ * \param payload_size UDP payload size by default set to 1472 =
+ *                     (1500 MTU - (8 byte UDP header) - (20 byte IP header))
+ * \param eof          Interpret zero-length packet as EOF (default: true)
+ * \param wait         Wait for data if not immediately available
+ *                     (default: true)
  *
 */
 
 class gr_udp_source : public gr_sync_block
 {
-  friend gr_udp_source_sptr gr_make_udp_source(size_t itemsize, const char *src, 
-					       unsigned short port_src, int payload_size);
+  friend gr_udp_source_sptr gr_make_udp_source(size_t itemsize,
+					       const char *host, 
+					       unsigned short port,
+					       int payload_size,
+					       bool eof, bool wait);
 
  private:
   size_t	d_itemsize;
-  bool		d_updated;
-  gruel::mutex	d_mutex;
-
-  int            d_payload_size;  // maximum transmission unit (packet length)
-  int            d_socket;        // handle to socket
-  int            d_socket_rcv;    // handle to socket retuned in the accept call
-  struct in_addr d_ip_src;        // store the source IP address to use
-  unsigned short d_port_src;      // the port number to open for connections to this service
-  struct sockaddr_in    d_sockaddr_src;  // store the source sockaddr data (formatted IP address and port number)
+  int           d_payload_size;  // maximum transmission unit (packet length)
+  bool          d_eof;           // zero-length packet is EOF
+  bool          d_wait;          // wait if data if not immediately available
+  int           d_socket;        // handle to socket
   char *d_temp_buff;    // hold buffer between calls
   ssize_t d_residual;   // hold information about number of bytes stored in the temp buffer
   size_t d_temp_offset; // point to temp buffer location offset
@@ -81,35 +75,29 @@ class gr_udp_source : public gr_sync_block
    * \brief UDP Source Constructor
    * 
    * \param itemsize     The size (in bytes) of the item datatype
-   * \param src          The source address as either the host name or the 'numbers-and-dots'
-   *                     IP address
-   * \param port_src     The port number on which the socket listens for data
-   * \param payload_size UDP payload size by default set to 
-   *                     1472 = (1500 MTU - (8 byte UDP header) - (20 byte IP header))
+   * \param host         The name or IP address of the receiving host; can be
+   *                     NULL, None, or "0.0.0.0" to allow reading from any
+   *                     interface on the host
+   * \param port         The port number on which to receive data; use 0 to
+   *                     have the system assign an unused port number
+   * \param payload_size UDP payload size by default set to 1472 =
+   *                     (1500 MTU - (8 byte UDP header) - (20 byte IP header))
+   * \param eof          Interpret zero-length packet as EOF (default: true)
+   * \param wait         Wait for data if not immediately available
+   *                     (default: true)
    */
-  gr_udp_source(size_t itemsize, const char *src, unsigned short port_src, int payload_size);
+  gr_udp_source(size_t itemsize, const char *host, unsigned short port,
+		int payload_size, bool eof, bool wait);
 
  public:
   ~gr_udp_source();
 
-  /*!
-   * \brief open a socket specified by the port and ip address info
-   *
-   * Opens a socket, binds to the address, and waits for a connection
-   * over UDP. If any of these fail, the fuction retuns the error and exits.
-   */
-  bool open();
-
-  /*!
-   * \brief Close current socket.
-   *
-   * Shuts down read/write on the socket
-   */
-  void close();
-
   /*! \brief return the PAYLOAD_SIZE of the socket */
   int payload_size() { return d_payload_size; }
 
+  /*! \breif return the port number of the socket */
+  int get_port();
+
   // should we export anything else?
 
   int work(int noutput_items,
diff --git a/gnuradio-core/src/lib/io/gr_udp_source.i b/gnuradio-core/src/lib/io/gr_udp_source.i
index fb39dad68..2001f33e9 100644
--- a/gnuradio-core/src/lib/io/gr_udp_source.i
+++ b/gnuradio-core/src/lib/io/gr_udp_source.i
@@ -1,6 +1,6 @@
 /* -*- c++ -*- */
 /*
- * Copyright 2007 Free Software Foundation, Inc.
+ * Copyright 2007,2010 Free Software Foundation, Inc.
  * 
  * This file is part of GNU Radio
  * 
@@ -23,20 +23,19 @@
 GR_SWIG_BLOCK_MAGIC(gr,udp_source)
 
 gr_udp_source_sptr 
-gr_make_udp_source (size_t itemsize, const char *src, 
-		    unsigned short port_src, int payload_size=1472);
+gr_make_udp_source (size_t itemsize, const char *host, 
+		    unsigned short port, int payload_size=1472,
+		    bool eof=true, bool wait=true) throw (std::runtime_error);
 
 class gr_udp_source : public gr_sync_block
 {
  protected:
-  gr_udp_source (size_t itemsize, const char *src, 
-		 unsigned short port_src, int payload_size);
+  gr_udp_source (size_t itemsize, const char *host, 
+		 unsigned short port, int payload_size, bool eof, bool wait) throw (std::runtime_error);
 
  public:
   ~gr_udp_source ();
 
-  bool open();
-  void close();
   int payload_size() { return d_payload_size; }
-
+  int get_port();
 };
diff --git a/gnuradio-core/src/python/gnuradio/gr/Makefile.am b/gnuradio-core/src/python/gnuradio/gr/Makefile.am
index 3aff89ee7..341f58812 100644..100755
--- a/gnuradio-core/src/python/gnuradio/gr/Makefile.am
+++ b/gnuradio-core/src/python/gnuradio/gr/Makefile.am
@@ -1,5 +1,5 @@
 #
-# Copyright 2004,2005,2006,2008 Free Software Foundation, Inc.
+# Copyright 2004,2005,2006,2008,2010 Free Software Foundation, Inc.
 # 
 # This file is part of GNU Radio
 # 
@@ -97,4 +97,5 @@ noinst_PYTHON = 			\
 	qa_unpack_k_bits.py		\
 	qa_repeat.py                    \
 	qa_scrambler.py			\
+	qa_udp_sink_source.py		\
 	qa_vector_sink_source.py	
diff --git a/gnuradio-core/src/python/gnuradio/gr/qa_udp_sink_source.py b/gnuradio-core/src/python/gnuradio/gr/qa_udp_sink_source.py
new file mode 100755
index 000000000..b00b26bbe
--- /dev/null
+++ b/gnuradio-core/src/python/gnuradio/gr/qa_udp_sink_source.py
@@ -0,0 +1,99 @@
+#!/usr/bin/env python
+#
+# Copyright 2008,2010 Free Software Foundation, Inc.
+# 
+# This file is part of GNU Radio
+# 
+# GNU Radio is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 3, or (at your option)
+# any later version.
+# 
+# GNU Radio is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+# 
+# You should have received a copy of the GNU General Public License
+# along with GNU Radio; see the file COPYING.  If not, write to
+# the Free Software Foundation, Inc., 51 Franklin Street,
+# Boston, MA 02110-1301, USA.
+# 
+
+from gnuradio import gr, gr_unittest
+from threading import Timer
+
+class test_sink_source(gr_unittest.TestCase):
+
+    def setUp(self):
+        self.tb_snd = gr.top_block()
+        self.tb_rcv = gr.top_block()
+
+    def tearDown(self):
+        self.tb_rcv = None
+        self.tb_snd = None
+        
+    def test_001(self):
+        port = 65500
+
+        n_data = 16
+        src_data = [float(x) for x in range(n_data)]
+        expected_result = tuple(src_data)
+        src = gr.vector_source_f(src_data)
+        udp_snd = gr.udp_sink( gr.sizeof_float, 'localhost', port )
+        self.tb_snd.connect( src, udp_snd )
+
+        udp_rcv = gr.udp_source( gr.sizeof_float, 'localhost', port )
+        dst = gr.vector_sink_f()
+        self.tb_rcv.connect( udp_rcv, dst )
+
+        self.tb_rcv.start()
+        self.tb_snd.run()
+        udp_snd.disconnect()
+        self.timeout = False
+        q = Timer(3.0,self.stop_rcv)
+        q.start()
+        self.tb_rcv.wait()
+        q.cancel()
+
+        result_data = dst.data()
+        self.assertEqual(expected_result, result_data)
+        self.assert_(not self.timeout)
+        
+    def test_002(self):
+        udp_rcv = gr.udp_source( gr.sizeof_float, '0.0.0.0', 0, eof=False )
+        rcv_port = udp_rcv.get_port()
+
+        udp_snd = gr.udp_sink( gr.sizeof_float, '127.0.0.1', 65500 )
+        udp_snd.connect( 'localhost', rcv_port )
+
+        n_data = 16
+        src_data = [float(x) for x in range(n_data)]
+        expected_result = tuple(src_data)
+        src = gr.vector_source_f(src_data)
+        dst = gr.vector_sink_f()
+
+        self.tb_snd.connect( src, udp_snd )
+        self.tb_rcv.connect( udp_rcv, dst )
+
+        self.tb_rcv.start()
+        self.tb_snd.run()
+        udp_snd.disconnect()
+        self.timeout = False
+        q = Timer(3.0,self.stop_rcv)
+        q.start()
+        self.tb_rcv.wait()
+        q.cancel()
+
+        result_data = dst.data()
+        self.assertEqual(expected_result, result_data)
+        self.assert_(self.timeout)  # source ignores EOF?
+
+    def stop_rcv(self):
+        self.timeout = True
+        self.tb_rcv.stop()
+        #print "tb_rcv stopped by Timer"
+        
+if __name__ == '__main__':
+    gr_unittest.main ()
+    
diff --git a/gnuradio-examples/python/network/audio_sink.py b/gnuradio-examples/python/network/audio_sink.py
index e59d50834..eb18a75aa 100755
--- a/gnuradio-examples/python/network/audio_sink.py
+++ b/gnuradio-examples/python/network/audio_sink.py
@@ -1,6 +1,6 @@
 #!/usr/bin/env python
 #
-# Copyright 2006,2007 Free Software Foundation, Inc.
+# Copyright 2006,2007,2010 Free Software Foundation, Inc.
 # 
 # This file is part of GNU Radio
 # 
@@ -25,30 +25,36 @@ from gnuradio.eng_option import eng_option
 from optparse import OptionParser
 
 class audio_sink(gr.top_block):
-    def __init__(self, src, port, pkt_size, sample_rate):
+    def __init__(self, host, port, pkt_size, sample_rate, eof, wait):
         gr.top_block.__init__(self, "audio_sink")
-        src = gr.udp_source(gr.sizeof_float, src, port, pkt_size)
+        src = gr.udp_source(gr.sizeof_float, host, port, pkt_size,
+                            eof=eof, wait=wait)
         dst = audio.sink(sample_rate)
         self.connect(src, dst)
         
 if __name__ == '__main__':
     parser = OptionParser(option_class=eng_option)
-    parser.add_option("", "--src-name", type="string", default="localhost",
+    parser.add_option("", "--host", type="string", default="0.0.0.0",
                       help="local host name (domain name or IP address)")
-    parser.add_option("", "--src-port", type="int", default=65500,
+    parser.add_option("", "--port", type="int", default=65500,
                       help="port value to listen to for connection")
     parser.add_option("", "--packet-size", type="int", default=1472,
                       help="packet size.")
     parser.add_option("-r", "--sample-rate", type="int", default=32000,
                       help="audio signal sample rate [default=%default]")
+    parser.add_option("", "--no-eof", action="store_true", default=False,
+                      help="don't send EOF on disconnect")
+    parser.add_option("", "--no-wait", action="store_true", default=False,
+                      help="don't wait for source")
     (options, args) = parser.parse_args()
     if len(args) != 0:
         parser.print_help()
         raise SystemExit, 1
 
     # Create an instance of a hierarchical block
-    top_block = audio_sink(options.src_name, options.src_port,
-                           options.packet_size, options.sample_rate)
+    top_block = audio_sink(options.host, options.port,
+                           options.packet_size, options.sample_rate,
+                           not options.no_eof, not options.no_wait)
     
     try:    
         # Run forever
diff --git a/gnuradio-examples/python/network/audio_source.py b/gnuradio-examples/python/network/audio_source.py
index d7f4f6d93..5818ccbd8 100755
--- a/gnuradio-examples/python/network/audio_source.py
+++ b/gnuradio-examples/python/network/audio_source.py
@@ -1,6 +1,6 @@
 #!/usr/bin/env python
 #
-# Copyright 2006,2007 Free Software Foundation, Inc.
+# Copyright 2006,2007,2010 Free Software Foundation, Inc.
 # 
 # This file is part of GNU Radio
 # 
@@ -25,32 +25,33 @@ from gnuradio.eng_option import eng_option
 from optparse import OptionParser
 
 class audio_source(gr.top_block):
-    def __init__(self, src, dst, port, pkt_size, sample_rate):
+    def __init__(self, host, port, pkt_size, sample_rate, eof):
         gr.top_block.__init__(self, "audio_source")
         self.audio = audio.source(sample_rate)
-	self.sink = gr.udp_sink(gr.sizeof_float, src, 0, dst, port, pkt_size)
+	self.sink = gr.udp_sink(gr.sizeof_float, host, port, pkt_size, eof=eof)
         self.connect(self.audio, self.sink)
 
 if __name__ == '__main__':
     parser = OptionParser(option_class=eng_option)
-    parser.add_option("", "--src-name", type="string", default="localhost",
-                      help="local host name (domain name or IP address)")
-    parser.add_option("", "--dst-name", type="string", default="localhost",
+    parser.add_option("", "--host", type="string", default="localhost",
                       help="Remote host name (domain name or IP address")
-    parser.add_option("", "--dst-port", type="int", default=65500,
-                      help="port value to connect to")
+    parser.add_option("", "--port", type="int", default=65500,
+                      help="port number to connect to")
     parser.add_option("", "--packet-size", type="int", default=1472,
                       help="packet size.")
     parser.add_option("-r", "--sample-rate", type="int", default=32000 ,
                       help="audio signal sample rate [default=%default]")
+    parser.add_option("", "--no-eof", action="store_true", default=False,
+                      help="don't send EOF on disconnect")
     (options, args) = parser.parse_args()
     if len(args) != 0:
         parser.print_help()
         raise SystemExit, 1
 
     # Create an instance of a hierarchical block
-    top_block = audio_source(options.src_name, options.dst_name, options.dst_port,
-                             options.packet_size, options.sample_rate)
+    top_block = audio_source(options.host, options.port,
+                             options.packet_size, options.sample_rate,
+                             not options.no_eof)
     
     try:    
         # Run forever
diff --git a/gnuradio-examples/python/network/dial_tone_sink.py b/gnuradio-examples/python/network/dial_tone_sink.py
index 47d24b9bc..1b9009552 100755
--- a/gnuradio-examples/python/network/dial_tone_sink.py
+++ b/gnuradio-examples/python/network/dial_tone_sink.py
@@ -1,6 +1,6 @@
 #!/usr/bin/env python
 #
-# Copyright 2006,2007 Free Software Foundation, Inc.
+# Copyright 2006,2007,2010 Free Software Foundation, Inc.
 # 
 # This file is part of GNU Radio
 # 
@@ -25,30 +25,36 @@ from gnuradio.eng_option import eng_option
 from optparse import OptionParser
 
 class dial_tone_sink(gr.top_block):
-    def __init__(self, src, port, pkt_size, sample_rate):
+    def __init__(self, host, port, pkt_size, sample_rate, eof, wait):
         gr.top_block.__init__(self, "dial_tone_sink")
-        udp = gr.udp_source(gr.sizeof_float, src, port, pkt_size)
+        udp = gr.udp_source(gr.sizeof_float, host, port, pkt_size,
+                            eof=eof, wait=wait)
         sink = audio.sink(sample_rate)
         self.connect(udp, sink)
         
 if __name__ == '__main__':
     parser = OptionParser(option_class=eng_option)
-    parser.add_option("", "--src-name", type="string", default="localhost",
+    parser.add_option("", "--host", type="string", default="0.0.0.0",
                       help="local host name (domain name or IP address)")
-    parser.add_option("", "--src-port", type="int", default=65500,
+    parser.add_option("", "--port", type="int", default=65500,
                       help="port value to listen to for connection")
     parser.add_option("", "--packet-size", type="int", default=1472,
                       help="packet size.")
     parser.add_option("-r", "--sample-rate", type="int", default=8000,
                       help="audio signal sample rate [default=%default]")
+    parser.add_option("", "--no-eof", action="store_true", default=False,
+                      help="don't send EOF on disconnect")
+    parser.add_option("", "--no-wait", action="store_true", default=False,
+                      help="don't wait for source")
     (options, args) = parser.parse_args()
     if len(args) != 0:
         parser.print_help()
         raise SystemExit, 1
 
     # Create an instance of a hierarchical block
-    top_block = dial_tone_sink(options.src_name, options.src_port,
-                               options.packet_size, options.sample_rate)
+    top_block = dial_tone_sink(options.host, options.port,
+                               options.packet_size, options.sample_rate,
+                               not options.no_eof, not options.no_wait)
     
     try:    
         # Run forever
diff --git a/gnuradio-examples/python/network/dial_tone_source.py b/gnuradio-examples/python/network/dial_tone_source.py
index 835f9aafc..766ecf16d 100755
--- a/gnuradio-examples/python/network/dial_tone_source.py
+++ b/gnuradio-examples/python/network/dial_tone_source.py
@@ -1,6 +1,6 @@
 #!/usr/bin/env python
 #
-# Copyright 2006,2007 Free Software Foundation, Inc.
+# Copyright 2006,2007,2010 Free Software Foundation, Inc.
 # 
 # This file is part of GNU Radio
 # 
@@ -25,7 +25,7 @@ from gnuradio.eng_option import eng_option
 from optparse import OptionParser
 
 class dial_tone_source(gr.top_block):
-    def __init__(self, src, dst, port, pkt_size, sample_rate):
+    def __init__(self, host, port, pkt_size, sample_rate, eof):
         gr.top_block.__init__(self, "dial_tone_source")
 
         amplitude = 0.3
@@ -35,31 +35,32 @@ class dial_tone_source(gr.top_block):
 
         # Throttle needed here to account for the other side's audio card sampling rate
 	thr = gr.throttle(gr.sizeof_float, sample_rate)
-	sink = gr.udp_sink(gr.sizeof_float, src, 0, dst, port, pkt_size)
+	sink = gr.udp_sink(gr.sizeof_float, host, port, pkt_size, eof=eof)
 	self.connect(src0, (add, 0))
 	self.connect(src1, (add, 1))
 	self.connect(add, thr, sink)
 
 if __name__ == '__main__':
     parser = OptionParser(option_class=eng_option)
-    parser.add_option("", "--src-name", type="string", default="localhost",
-                      help="local host name (domain name or IP address)")
-    parser.add_option("", "--dst-name", type="string", default="localhost",
+    parser.add_option("", "--host", type="string", default="localhost",
                       help="Remote host name (domain name or IP address")
-    parser.add_option("", "--dst-port", type="int", default=65500,
-                      help="port value to connect to")
+    parser.add_option("", "--port", type="int", default=65500,
+                      help="port number to connect to")
     parser.add_option("", "--packet-size", type="int", default=1472,
                       help="packet size.")
     parser.add_option("-r", "--sample-rate", type="int", default=8000,
                       help="audio signal sample rate [default=%default]")
+    parser.add_option("", "--no-eof", action="store_true", default=False,
+                      help="don't send EOF on disconnect")
     (options, args) = parser.parse_args()
     if len(args) != 0:
         parser.print_help()
         raise SystemExit, 1
 
     # Create an instance of a hierarchical block
-    top_block = dial_tone_source(options.src_name, options.dst_name, options.dst_port,
-                                 options.packet_size, options.sample_rate)
+    top_block = dial_tone_source(options.host, options.port,
+                                 options.packet_size, options.sample_rate,
+                                 not options.no_eof)
     
     try:    
         # Run forever
@@ -67,4 +68,3 @@ if __name__ == '__main__':
     except KeyboardInterrupt:
         # Ctrl-C exits
         pass
-    
diff --git a/gnuradio-examples/python/network/vector_sink.py b/gnuradio-examples/python/network/vector_sink.py
index 981cc598b..5d73858a3 100755
--- a/gnuradio-examples/python/network/vector_sink.py
+++ b/gnuradio-examples/python/network/vector_sink.py
@@ -1,6 +1,6 @@
 #!/usr/bin/env python
 #
-# Copyright 2006 Free Software Foundation, Inc.
+# Copyright 2006,2010 Free Software Foundation, Inc.
 # 
 # This file is part of GNU Radio
 # 
@@ -25,28 +25,35 @@ from gnuradio.eng_option import eng_option
 from optparse import OptionParser
 
 class vector_sink(gr.top_block):
-    def __init__(self, src, port, pkt_size):
+    def __init__(self, host, port, pkt_size, eof, wait):
         gr.top_block.__init__(self, "vector_sink")
 
-        udp = gr.udp_source(gr.sizeof_float, src, port, pkt_size)
+        udp = gr.udp_source(gr.sizeof_float, host, port, pkt_size,
+                            eof=eof, wait=wait)
         sink = gr.file_sink(gr.sizeof_float, "received.dat")
         self.connect(udp, sink)
 
 if __name__ == "__main__":
     parser = OptionParser(option_class=eng_option)
-    parser.add_option("", "--src-name", type="string", default="localhost",
+    parser.add_option("", "--host", type="string", default="0.0.0.0",
                       help="local host name (domain name or IP address)")
-    parser.add_option("", "--src-port", type="int", default=65500,
+    parser.add_option("", "--port", type="int", default=65500,
                       help="port value to listen to for connection")
     parser.add_option("", "--packet-size", type="int", default=1471,
                       help="packet size.")
+    parser.add_option("", "--no-eof", action="store_true", default=False,
+                      help="don't send EOF on disconnect")
+    parser.add_option("", "--no-wait", action="store_true", default=False,
+                      help="don't wait for source")
     (options, args) = parser.parse_args()
     if len(args) != 0:
         parser.print_help()
         raise SystemExit, 1
     
     # Create an instance of a hierarchical block
-    top_block = vector_sink(options.src_name, options.src_port, options.packet_size)
+    top_block = vector_sink(options.host, options.port,
+                            options.packet_size,
+                            not options.no_eof, not options.no_wait)
     
     try:    
         # Run forever
diff --git a/gnuradio-examples/python/network/vector_source.py b/gnuradio-examples/python/network/vector_source.py
index e7ec2a461..0e7d67844 100755
--- a/gnuradio-examples/python/network/vector_source.py
+++ b/gnuradio-examples/python/network/vector_source.py
@@ -1,6 +1,6 @@
 #!/usr/bin/env python
 #
-# Copyright 2006 Free Software Foundation, Inc.
+# Copyright 2006,2010 Free Software Foundation, Inc.
 # 
 # This file is part of GNU Radio
 # 
@@ -25,31 +25,31 @@ from gnuradio.eng_option import eng_option
 from optparse import OptionParser
 
 class vector_source(gr.top_block):
-    def __init__(self, src, dst, port, pkt_size):
+    def __init__(self, host, port, pkt_size, eof):
         gr.top_block.__init__(self, "vector_source")
         data = [i*0.01 for i in range(1000)]
         vec = gr.vector_source_f(data, True)
-        udp = gr.udp_sink(gr.sizeof_float, src, 0, dst, port, pkt_size)
+        udp = gr.udp_sink(gr.sizeof_float, host, port, pkt_size, eof=eof)
         self.connect(vec, udp)
 
 if __name__ == '__main__':
     parser = OptionParser(option_class=eng_option)
-    parser.add_option("", "--src-name", type="string", default="localhost",
-                      help="local host name (domain name or IP address)")
-    parser.add_option("", "--dst-name", type="string", default="localhost",
+    parser.add_option("", "--host", type="string", default="localhost",
                       help="Remote host name (domain name or IP address")
-    parser.add_option("", "--dst-port", type="int", default=65500,
-                      help="port value to connect to")
+    parser.add_option("", "--port", type="int", default=65500,
+                      help="port number to connect to")
     parser.add_option("", "--packet-size", type="int", default=1471,
                       help="packet size.")
+    parser.add_option("", "--no-eof", action="store_true", default=False,
+                      help="don't send EOF on disconnect")
     (options, args) = parser.parse_args()
     if len(args) != 0:
         parser.print_help()
         raise SystemExit, 1
 
 # Create an instance of a hierarchical block
-    top_block = vector_source(options.src_name, options.dst_name,
-                              options.dst_port, options.packet_size)
+    top_block = vector_source(options.host, options.port, options.packet_size,
+                              not options.no_eof)
     
     try:    
         # Run forever
diff --git a/grc/blocks/gr_udp_sink.xml b/grc/blocks/gr_udp_sink.xml
index e9f6c2be8..45f81075f 100644
--- a/grc/blocks/gr_udp_sink.xml
+++ b/grc/blocks/gr_udp_sink.xml
@@ -8,7 +8,7 @@
 	<name>UDP Sink</name>
 	<key>gr_udp_sink</key>
 	<import>from gnuradio import gr</import>
-	<make>gr.udp_sink($type.size*$vlen, $ipaddr_local, $port_local, $ipaddr_remote, $port_remote, $mtu)</make>
+	<make>gr.udp_sink($type.size*$vlen, $ipaddr, $port, $psize, $eof)</make>
 	<callback>set_mtu($mtu)</callback>
 	<param>
 		<name>Input Type</name>
@@ -41,34 +41,26 @@
 		</option>
 	</param>
 	<param>
-		<name>Local IP Address</name>
-		<key>ipaddr_local</key>
-		<value>127.0.0.1</value>
+		<name>Destination IP Address</name>
+		<key>ipaddr</key>
 		<type>string</type>
 	</param>
 	<param>
-		<name>Local Port</name>
-		<key>port_local</key>
-		<value>0</value>
+		<name>Destination Port</name>
+		<key>port</key>
 		<type>int</type>
 	</param>
 	<param>
-		<name>Remote IP Address</name>
-		<key>ipaddr_remote</key>
-		<value>127.0.0.1</value>
-		<type>string</type>
-	</param>
-	<param>
-		<name>Remote Port</name>
-		<key>port_remote</key>
-		<value>1234</value>
+		<name>Payload Size</name>
+		<key>psize</key>
+		<value>1472</value>
 		<type>int</type>
 	</param>
 	<param>
-		<name>MTU</name>
-		<key>mtu</key>
-		<value>1024</value>
-		<type>int</type>
+	        <name>Send Null Pkt as EOF</name>
+		<key>eof</key>
+		<value>True</value>
+		<type>bool</type>
 	</param>
 	<param>
 		<name>Vec Length</name>
diff --git a/grc/blocks/gr_udp_source.xml b/grc/blocks/gr_udp_source.xml
index f03adf802..a1b961651 100644
--- a/grc/blocks/gr_udp_source.xml
+++ b/grc/blocks/gr_udp_source.xml
@@ -8,7 +8,7 @@
 	<name>UDP Source</name>
 	<key>gr_udp_source</key>
 	<import>from gnuradio import gr</import>
-	<make>gr.udp_source($type.size*$vlen, $ipaddr, $port, $mtu)</make>
+	<make>gr.udp_source($type.size*$vlen, $ipaddr, $port, $psize, $eof, $wait)</make>
 	<callback>set_mtu($mtu)</callback>
 	<param>
 		<name>Output Type</name>
@@ -53,12 +53,24 @@
 		<type>int</type>
 	</param>
 	<param>
-		<name>MTU</name>
-		<key>mtu</key>
-		<value>1024</value>
+		<name>Payload Size</name>
+		<key>psize</key>
+		<value>1472</value>
 		<type>int</type>
 	</param>
 	<param>
+	        <name>Null Pkt is EOF</name>
+		<key>eof</key>
+		<value>True</value>
+		<type>bool</type>
+	</param>
+	<param>
+	        <name>Wait for Data</name>
+		<key>wait</key>
+		<value>True</value>
+		<type>bool</type>
+	</param>
+	<param>
 		<name>Vec Length</name>
 		<key>vlen</key>
 		<value>1</value>
diff --git a/usrp2/host/include/usrp2/usrp2.h b/usrp2/host/include/usrp2/usrp2.h
index 7069507cf..e29caa33d 100644
--- a/usrp2/host/include/usrp2/usrp2.h
+++ b/usrp2/host/include/usrp2/usrp2.h
@@ -21,6 +21,7 @@
 
 #include <boost/shared_ptr.hpp>
 #include <boost/utility.hpp>
+#include <boost/bind.hpp>
 #include <vector>
 #include <complex>
 #include <usrp2/rx_sample_handler.h>