[Commit-gnuradio] r4355 - gnuradio/branches/developers/trondeau/udp/gnuradio-core/src/lib/io

[trondeau]

Author: trondeau
Date: 2007-02-01 20:24:57 -0700 (Thu, 01 Feb 2007)
New Revision: 4355

Modified:
   
gnuradio/branches/developers/trondeau/udp/gnuradio-core/src/lib/io/gr_udp_sink.cc
   
gnuradio/branches/developers/trondeau/udp/gnuradio-core/src/lib/io/gr_udp_sink.h
   
gnuradio/branches/developers/trondeau/udp/gnuradio-core/src/lib/io/gr_udp_sink.i
   
gnuradio/branches/developers/trondeau/udp/gnuradio-core/src/lib/io/gr_udp_source.cc
   
gnuradio/branches/developers/trondeau/udp/gnuradio-core/src/lib/io/gr_udp_source.h
   
gnuradio/branches/developers/trondeau/udp/gnuradio-core/src/lib/io/gr_udp_source.i
Log:
trial fix for ticket:131. Fixed byte dropping, cleaned up code, cleaned up 
error handling. Yep, a lot of stuff in this one. Needs external verification 
before merge

Modified: 
gnuradio/branches/developers/trondeau/udp/gnuradio-core/src/lib/io/gr_udp_sink.cc
===================================================================
--- 
gnuradio/branches/developers/trondeau/udp/gnuradio-core/src/lib/io/gr_udp_sink.cc
   2007-02-01 19:41:35 UTC (rev 4354)
+++ 
gnuradio/branches/developers/trondeau/udp/gnuradio-core/src/lib/io/gr_udp_sink.cc
   2007-02-02 03:24:57 UTC (rev 4355)
@@ -1,6 +1,6 @@
 /* -*- c++ -*- */
 /*
- * Copyright 2006 Free Software Foundation, Inc.
+ * Copyright 2007 Free Software Foundation, Inc.
  * 
  * This file is part of GNU Radio
  * 
@@ -20,16 +20,8 @@
  * Boston, MA 02110-1301, USA.
  */
 
-#ifdef HAVE_CONFIG_H
-#include "config.h"
-#endif
-
 #include <gr_udp_sink.h>
 #include <gr_io_signature.h>
-#include <cstdio>
-#include <sys/types.h>
-#include <sys/stat.h>
-#include <fcntl.h>
 #include <stdexcept>
 
 #define SNK_VERBOSE 0
@@ -37,7 +29,7 @@
 gr_udp_sink::gr_udp_sink (size_t itemsize, 
                          const char *ipaddrl, unsigned short portl,
                          const char *ipaddrr, unsigned short portr,
-                         unsigned int mtu)
+                         int mtu)
   : gr_sync_block ("udp_sink",
                   gr_make_io_signature (1, 1, itemsize),
                   gr_make_io_signature (0, 0, 0)),
@@ -66,12 +58,12 @@
 gr_make_udp_sink (size_t itemsize, 
                  const char *ipaddrl, unsigned short portl,
                  const char *ipaddrr, unsigned short portr,
-                 unsigned int mtu)
+                 int mtu)
 {
   return gr_udp_sink_sptr (new gr_udp_sink (itemsize, 
-                                                     ipaddrl, portl,
-                                                     ipaddrr, portr,
-                                                     mtu));
+                                           ipaddrl, portl,
+                                           ipaddrr, portr,
+                                           mtu));
 }
 
 gr_udp_sink::~gr_udp_sink ()
@@ -140,13 +132,23 @@
                   gr_vector_void_star &output_items)
 {
   char *in = (char *) input_items[0];
-  socklen_t bytes=0, bytes_sent=0, bytes_to_send=0;
-  unsigned int total_size = noutput_items*d_itemsize;
+  ssize_t bytes=0, bytes_sent=0, bytes_to_send=0;
+  ssize_t total_size = noutput_items*d_itemsize;
 
-  while(bytes_sent < total_size) {
-    bytes_to_send = (bytes_sent+d_mtu < total_size ? d_mtu : 
total_size-bytes_sent);
-    bytes = send(d_socket, (in+bytes_sent), bytes_to_send, MSG_DONTWAIT);
+  #if SNK_VERBOSE
+  printf("Entered upd_sink\n");
+  #endif
+
+  while(bytes_sent <  total_size) {
+    //bytes_to_send = (bytes_sent+d_mtu < total_size ? d_mtu : 
total_size-bytes_sent);
+    bytes_to_send = std::min(d_mtu, (total_size-bytes_sent));
+  
+    bytes = send(d_socket, (in+bytes_sent), bytes_to_send, 0);
     bytes_sent += bytes;
+    
+    #if SNK_VERBOSE
+    printf("\tbyte sent: %d bytes\n", bytes);
+    #endif
   }
 
   #if SNK_VERBOSE

Modified: 
gnuradio/branches/developers/trondeau/udp/gnuradio-core/src/lib/io/gr_udp_sink.h
===================================================================
--- 
gnuradio/branches/developers/trondeau/udp/gnuradio-core/src/lib/io/gr_udp_sink.h
    2007-02-01 19:41:35 UTC (rev 4354)
+++ 
gnuradio/branches/developers/trondeau/udp/gnuradio-core/src/lib/io/gr_udp_sink.h
    2007-02-02 03:24:57 UTC (rev 4355)
@@ -1,6 +1,6 @@
 /* -*- c++ -*- */
 /*
- * Copyright 2006 Free Software Foundation, Inc.
+ * Copyright 2007 Free Software Foundation, Inc.
  * 
  * This file is part of GNU Radio
  * 
@@ -40,20 +40,20 @@
 gr_make_udp_sink (size_t itemsize, 
                  const char *ipaddrl, unsigned short portl,
                  const char *ipaddrr, unsigned short portr,
-                 unsigned int mtu=540);
+                 int mtu=540);
 
 class gr_udp_sink : public gr_sync_block
 {
   friend gr_udp_sink_sptr gr_make_udp_sink (size_t itemsize, 
                                            const char *ipaddrl, unsigned short 
portl,
                                            const char *ipaddrr, unsigned short 
portr,
-                                           unsigned int mtu);
+                                           int mtu);
  private:
   size_t       d_itemsize;
   bool         d_updated;
   omni_mutex   d_mutex;
 
-  unsigned int   d_mtu;             // maximum transmission unit (packet 
length)
+  int            d_mtu;             // 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_ipaddr_local;    // store the local IP address to use
@@ -65,9 +65,9 @@
 
  protected:
   gr_udp_sink (size_t itemsize, 
-                   const char *ipaddrl, unsigned short portl,
-                   const char *ipaddrr, unsigned short portr,
-                   unsigned int mtu);
+              const char *ipaddrl, unsigned short portl,
+              const char *ipaddrr, unsigned short portr,
+              int mtu);
 
  public:
   ~gr_udp_sink ();
@@ -88,10 +88,10 @@
   void close();
 
   /*! \brief set the MTU of the socket */
-  void set_mtu(unsigned int mtu) { d_mtu = mtu; }
+  void set_mtu(int mtu) { d_mtu = mtu; }
 
   /*! \brief return the MTU of the socket */
-  unsigned int mtu() { return d_mtu; }
+  int mtu() { return d_mtu; }
 
   // should we export anything else?
 

Modified: 
gnuradio/branches/developers/trondeau/udp/gnuradio-core/src/lib/io/gr_udp_sink.i
===================================================================
--- 
gnuradio/branches/developers/trondeau/udp/gnuradio-core/src/lib/io/gr_udp_sink.i
    2007-02-01 19:41:35 UTC (rev 4354)
+++ 
gnuradio/branches/developers/trondeau/udp/gnuradio-core/src/lib/io/gr_udp_sink.i
    2007-02-02 03:24:57 UTC (rev 4355)
@@ -1,6 +1,6 @@
 /* -*- c++ -*- */
 /*
- * Copyright 2006 Free Software Foundation, Inc.
+ * Copyright 2007 Free Software Foundation, Inc.
  * 
  * This file is part of GNU Radio
  * 
@@ -27,7 +27,7 @@
 gr_make_udp_sink (size_t itemsize, 
                  const char *ipaddrl, unsigned short portl,
                  const char *ipaddrr, unsigned short portr,
-                 unsigned int mtu=540);
+                 int mtu=540);
 
 class gr_udp_sink : public gr_sync_block
 {
@@ -35,12 +35,12 @@
   gr_udp_sink (size_t itemsize, 
               const char *ipaddrl, unsigned short portl,
               const char *ipaddrr, unsigned short portr,
-              unsigned int mtu);
+              int mtu);
 
   bool open();
   void close();
-  void set_mtu(unsigned int mtu) { d_mtu = mtu; }
-  unsigned int mtu() { return d_mtu; }
+  void set_mtu(int mtu) { d_mtu = mtu; }
+  int mtu() { return d_mtu; }
 
  public:
   ~gr_udp_sink ();

Modified: 
gnuradio/branches/developers/trondeau/udp/gnuradio-core/src/lib/io/gr_udp_source.cc
===================================================================
--- 
gnuradio/branches/developers/trondeau/udp/gnuradio-core/src/lib/io/gr_udp_source.cc
 2007-02-01 19:41:35 UTC (rev 4354)
+++ 
gnuradio/branches/developers/trondeau/udp/gnuradio-core/src/lib/io/gr_udp_source.cc
 2007-02-02 03:24:57 UTC (rev 4355)
@@ -1,6 +1,6 @@
 /* -*- c++ -*- */
 /*
- * Copyright 2006 Free Software Foundation, Inc.
+ * Copyright 2007 Free Software Foundation, Inc.
  * 
  * This file is part of GNU Radio
  * 
@@ -20,26 +20,19 @@
  * Boston, MA 02110-1301, USA.
  */
 
-#ifdef HAVE_CONFIG_H
-#include "config.h"
-#endif
-
 #include <gr_udp_source.h>
 #include <gr_io_signature.h>
-#include <cstdio>
-#include <sys/types.h>
-#include <sys/stat.h>
-#include <fcntl.h>
 #include <stdexcept>
+#include <errno.h>
 
 #define SRC_VERBOSE 0
 
 gr_udp_source::gr_udp_source(size_t itemsize, const char *ipaddr, 
-                            unsigned short port, unsigned int mtu)
+                            unsigned short port, int mtu)
   : 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_mtu(mtu)
+    d_itemsize(itemsize), d_updated(false), d_mtu(mtu), d_residual(0), 
d_temp_offset(0)
 {
   // Set up the address stucture for the local address and port numbers
   inet_aton(ipaddr, &d_ipaddr_local);     // format IP address
@@ -48,13 +41,15 @@
   d_sockaddr_local.sin_family = AF_INET;
   d_sockaddr_local.sin_addr   = d_ipaddr_local;
   d_sockaddr_local.sin_port   = d_port_local;
+
+  d_temp_buff = new char[d_mtu];   // allow it to hold up to mtu bytes
   
   open();
 }
 
 gr_udp_source_sptr
 gr_make_udp_source (size_t itemsize, const char *ipaddr, 
-                   unsigned short port, unsigned int mtu)
+                   unsigned short port, int mtu)
 {
   return gr_udp_source_sptr (new gr_udp_source (itemsize, ipaddr, 
                                                port, mtu));
@@ -62,6 +57,7 @@
 
 gr_udp_source::~gr_udp_source ()
 {
+  delete [] d_temp_buff;
   close();
 }
 
@@ -69,9 +65,8 @@
 gr_udp_source::open()
 {
   omni_mutex_lock l(d_mutex);  // hold mutex for duration of this function
-   
   // create socket
-  d_socket = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP);
+  d_socket = socket(PF_INET, SOCK_DGRAM, IPPROTO_UDP);
   if(d_socket == -1) {
     perror("socket open");
     throw std::runtime_error("can't open socket");
@@ -131,33 +126,80 @@
                     gr_vector_void_star &output_items)
 {
   char *out = (char *) output_items[0];
-  socklen_t bytes_to_receive=0, bytes_received=0;
-  int bytes=0;
+  ssize_t bytes_to_receive=0, bytes_received=0, bytes=0;
+  ssize_t total_bytes = (ssize_t)(d_itemsize*noutput_items);
 
-  while((bytes_received < (unsigned)noutput_items) && (bytes>-1)) {
-    // caclulate the number of byte left if we can fit in all d_mtu bytes
-    bytes_to_receive = (bytes_received+d_mtu < noutput_items ? 
-                       d_mtu : noutput_items-bytes_received);
+  char *buff = new char[d_mtu];
+
+  #if SRC_VERBOSE
+  printf("\nEntered udp_source\n");
+  #endif
+
+  // Remove items from temp buffer if they are in there
+  if(d_residual) {
+    bytes_to_receive = std::min(d_residual, total_bytes);
+    memcpy(out, d_temp_buff+d_temp_offset, bytes_to_receive);
+    bytes_received = bytes_to_receive;
+
+    #if SRC_VERBOSE
+    printf("\tTemp buff size: %d (bytes_to_receive: %d) (noutput_items: 
%d)\n", 
+          d_residual, bytes_received, noutput_items);
+    #endif
+
+    // Increment pointer
+    out += bytes_received;
     
+    // Update indexing of amount of bytes left in the buffer
+    d_residual -= bytes_to_receive;
+    d_temp_offset = d_temp_offset+d_residual;
+  }
+
+  while((bytes_received < total_bytes) && (bytes>-1)) {
     // get the data into our output buffer and record the number of bytes
-    // This is a blocking call, but it's timeout has been set in the 
constructor
-    bytes = recv(d_socket, out, bytes_to_receive, 0);
+    // This is a non-blocking call with a timeout set in the constructor
+    bytes = recv(d_socket, buff, d_mtu, 0);  // get the entire MTU or the rest 
of what's available
 
-    // FIXME if bytes < 0 bail
+    // Check if there was a problem; forget it if the operation just timed out
+    if(bytes == -1) {
+      if(errno == EAGAIN) {  // handle non-blocking call timeout
+        #if SRC_VERBOSE
+       printf("UDP receive timed out\n"); 
+        #endif
+       break;
+      }
+      else {
+       perror("udp_source");
+       return -1;
+      }
+    }
+    else {
+      // Calculate the number of bytes we can take from the buffer in this call
+      bytes_to_receive = std::min(bytes, total_bytes-bytes_received);
 
-    if(bytes > 0) {
+      // copy the number of bytes we want to look at here
+      memcpy(out, buff, bytes_to_receive);    
+
+      d_residual=bytes-bytes_to_receive;                      // save the 
number of bytes stored
+      d_temp_offset=0;                                        // reset buffer 
index
+      memcpy(d_temp_buff, buff+bytes_to_receive, d_residual); // save the 
rest, if we need to
+
       // keep track of the total number of bytes received
-      bytes_received += bytes;
+      bytes_received += bytes_to_receive;
 
       // increment the pointer
-      out += bytes;
+      out += bytes_to_receive;
     }
+
+    #if SNK_VERBOSE
+    printf("\tbytes received: %d bytes (bytes_to_receive: %d)\n", bytes, 
bytes_to_receive);
+    #endif
   }
 
   #if SRC_VERBOSE
-  printf("\nTotal Bytes Received: %d (noutput_items=%d)\n", bytes_received, 
noutput_items); 
+  printf("Total Bytes Received: %d (noutput_items=%d)\n", bytes_received, 
noutput_items);
   #endif
 
-  // FIXME what if (bytes_received % d_itemsize) != 0 ???
-  return int(bytes_received / d_itemsize);
+  delete [] buff;
+
+  return noutput_items;
 }

Modified: 
gnuradio/branches/developers/trondeau/udp/gnuradio-core/src/lib/io/gr_udp_source.h
===================================================================
--- 
gnuradio/branches/developers/trondeau/udp/gnuradio-core/src/lib/io/gr_udp_source.h
  2007-02-01 19:41:35 UTC (rev 4354)
+++ 
gnuradio/branches/developers/trondeau/udp/gnuradio-core/src/lib/io/gr_udp_source.h
  2007-02-02 03:24:57 UTC (rev 4355)
@@ -1,6 +1,6 @@
 /* -*- c++ -*- */
 /*
- * Copyright 2006 Free Software Foundation, Inc.
+ * Copyright 2007 Free Software Foundation, Inc.
  * 
  * This file is part of GNU Radio
  * 
@@ -32,19 +32,19 @@
 typedef boost::shared_ptr<gr_udp_source> gr_udp_source_sptr;
 
 gr_udp_source_sptr gr_make_udp_source(size_t itemsize, const char *ipaddr, 
-                                     unsigned short port, unsigned int 
mtu=540);
+                                     unsigned short port, int mtu=540);
 
 class gr_udp_source : public gr_sync_block
 {
   friend gr_udp_source_sptr gr_make_udp_source(size_t itemsize, const char 
*ipaddr, 
-                                              unsigned short port, unsigned 
int mtu);
+                                              unsigned short port, int mtu);
 
  private:
   size_t       d_itemsize;
   bool         d_updated;
   omni_mutex   d_mutex;
 
-  unsigned int   d_mtu;           // maximum transmission unit (packet length)
+  int            d_mtu;           // 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_ipaddr_local;  // store the local IP address to use
@@ -53,9 +53,13 @@
   unsigned short d_port_remote;   // port number of the remove system
   sockaddr_in    d_sockaddr_local;  // store the local sockaddr data 
(formatted IP address and port number)
   sockaddr_in    d_sockaddr_remote; // store the remote sockaddr data 
(formatted IP address and port number)
-  
+
+  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
+
  protected:
-  gr_udp_source(size_t itemsize, const char *ipaddr, unsigned short port, 
unsigned int mtu);
+  gr_udp_source(size_t itemsize, const char *ipaddr, unsigned short port, int 
mtu);
 
  public:
   ~gr_udp_source();
@@ -76,10 +80,10 @@
   void close();
 
   /*! \brief set the MTU of the socket */
-  void set_mtu(unsigned int mtu) { d_mtu = mtu; }
+  void set_mtu(int mtu) { d_mtu = mtu; }
 
   /*! \brief return the MTU of the socket */
-  unsigned int mtu() { return d_mtu; }
+  int mtu() { return d_mtu; }
 
   // should we export anything else?
 

Modified: 
gnuradio/branches/developers/trondeau/udp/gnuradio-core/src/lib/io/gr_udp_source.i
===================================================================
--- 
gnuradio/branches/developers/trondeau/udp/gnuradio-core/src/lib/io/gr_udp_source.i
  2007-02-01 19:41:35 UTC (rev 4354)
+++ 
gnuradio/branches/developers/trondeau/udp/gnuradio-core/src/lib/io/gr_udp_source.i
  2007-02-02 03:24:57 UTC (rev 4355)
@@ -1,6 +1,6 @@
 /* -*- c++ -*- */
 /*
- * Copyright 2006 Free Software Foundation, Inc.
+ * Copyright 2007 Free Software Foundation, Inc.
  * 
  * This file is part of GNU Radio
  * 
@@ -24,20 +24,20 @@
 
 gr_udp_source_sptr 
 gr_make_udp_source (size_t itemsize, const char *ipaddr, 
-                   unsigned short port, unsigned int mtu=540);
+                   unsigned short port, int mtu=540);
 
 class gr_udp_source : public gr_sync_block
 {
  protected:
   gr_udp_source (size_t itemsize, const char *ipaddr, 
-                unsigned short port, unsigned int mtu);
+                unsigned short port, int mtu);
 
  public:
   ~gr_udp_source ();
 
   bool open();
   void close();
-  void set_mtu(unsigned int mtu) { d_mtu = mtu; }
-  unsigned int mtu() { return d_mtu; }
+  void set_mtu(int mtu) { d_mtu = mtu; }
+  int mtu() { return d_mtu; }
 
 };