[Commit-gnuradio] [gnuradio] 18/148: Added a callback into the channel ring enqueue and dequeue. This will allow the deqeue caller to make an arbitrary call when its done with the data. Currently, the enqueue caller makes this callback a release_frames.

[git]

This is an automated email from the git hooks/post-receive script.

nwest pushed a commit to annotated tag old_usrp_devel_udp
in repository gnuradio.

commit 3cefeb194d568898c853a7d9dc2dfe0799c22704
Author: Josh Blum <address@hidden>
Date:   Wed Nov 4 18:13:39 2009 -0800

    Added a callback into the channel ring enqueue and dequeue.
    This will allow the deqeue caller to make an arbitrary call when its done 
with the data.
    Currently, the enqueue caller makes this callback a release_frames.
    
    Also, added more work into the data transport.
---
 usrp2/host/lib/eth_buffer.cc         | 51 +++++++++++++++++++++++++++++
 usrp2/host/lib/eth_buffer.h          |  2 +-
 usrp2/host/lib/eth_data_transport.cc | 63 ++++++++++++++++++++++++++++++++++--
 usrp2/host/lib/eth_data_transport.h  |  7 ++++
 usrp2/host/lib/ring.cc               |  6 ++--
 usrp2/host/lib/ring.h                | 11 +++++--
 usrp2/host/lib/usrp2_impl.cc         | 14 +++++---
 usrp2/host/lib/usrp2_impl.h          |  2 +-
 8 files changed, 142 insertions(+), 14 deletions(-)

diff --git a/usrp2/host/lib/eth_buffer.cc b/usrp2/host/lib/eth_buffer.cc
index bd37061..0bf59d9 100644
--- a/usrp2/host/lib/eth_buffer.cc
+++ b/usrp2/host/lib/eth_buffer.cc
@@ -234,6 +234,57 @@ namespace usrp2 {
     return EB_OK;
   }
 
+int
+  eth_buffer::rx_frame(void **buff, int timeout_in_ms)
+  {
+    DEBUG_LOG("\n");
+      
+    while (!frame_available()) {
+      if (timeout_in_ms == 0) {
+        DEBUG_LOG("w");
+        return -1;
+      }
+      
+      struct pollfd pfd;
+      pfd.fd = d_fd;
+      pfd.revents = 0;
+      pfd.events = POLLIN;
+
+      DEBUG_LOG("P");
+
+      int pres = poll(&pfd, 1, timeout_in_ms);
+      if (pres == -1) {
+        perror("poll");
+        return -1;
+      }
+
+      if (pres == 0) {
+        DEBUG_LOG("t");
+        return -1;
+      }
+    }
+
+    // Iterate through available packets
+    if (frame_available()) {
+        // Get start of ethernet frame and length
+        tpacket_hdr *hdr = (tpacket_hdr *)d_ring[d_head];
+        void *base = (uint8_t *)hdr+hdr->tp_mac;
+        size_t len = hdr->tp_len;
+
+        // FYI, (base % 4 == 2) Not what we want given the current FPGA
+        // code.  This means that our uint32_t samples are not 4-byte
+        // aligned.  We'll have to deal with it downstream.
+
+        *buff = base;
+
+        hdr->tp_status = TP_STATUS_KERNEL; // mark it free
+        inc_head();
+
+        return len;
+    }
+    return -1;
+  }
+
   eth_buffer::result
   eth_buffer::tx_frame(const void *base, size_t len, int flags)
   {
diff --git a/usrp2/host/lib/eth_buffer.h b/usrp2/host/lib/eth_buffer.h
index 8dee9a4..8c9f1d1 100644
--- a/usrp2/host/lib/eth_buffer.h
+++ b/usrp2/host/lib/eth_buffer.h
@@ -147,7 +147,7 @@ namespace usrp2 {
      * \returns EB_ERROR if there was an unrecoverable error.
      */
     result rx_frames(data_handler *f, int timeout=-1);
-
+    int rx_frame(void **buff, int timeout_in_ms);
     /*
      * \brief Release frame from buffer
      *
diff --git a/usrp2/host/lib/eth_data_transport.cc 
b/usrp2/host/lib/eth_data_transport.cc
index a781b6c..d1185cf 100644
--- a/usrp2/host/lib/eth_data_transport.cc
+++ b/usrp2/host/lib/eth_data_transport.cc
@@ -16,6 +16,8 @@
  * along with this program.  If not, see <http://www.gnu.org/licenses/>.
  */
 
+#define DEBUG_LOG(x) ::write(2, (x), 1)
+
 #include "eth_data_transport.h"
 #include <gruel/inet.h>
 #include <gruel/realtime.h>
@@ -23,7 +25,8 @@
 #include <iostream>
 
 usrp2::eth_data_transport::eth_data_transport(const std::string &ifc, 
u2_mac_addr_t mac, size_t rx_bufsize)
- : transport("ethernet control"), d_mac(mac), d_tx_seqno(0){
+ : transport("ethernet control"), d_mac(mac), d_tx_seqno(0), d_rx_seqno(0),
+ d_num_rx_frames(0), d_num_rx_missing(0), d_num_rx_overruns(0), 
d_num_rx_bytes(0){
 
     //create raw ethernet device
     d_eth_data = new eth_buffer(rx_bufsize);
@@ -47,10 +50,66 @@ void usrp2::eth_data_transport::init(){
         std::cerr << "usrp2: failed to enable realtime scheduling" << 
std::endl;
 }
 
+//FIXME clean this up, probably when we get vrt headers
+//eth transport is only responsible for eth headers and transport headers
+//that leaves the u2 fixed headers to be handled by the usrp2 impl
+typedef struct {
+  u2_eth_hdr_t         ehdr;
+  u2_transport_hdr_t   thdr;
+} u2_eth_packet_only_t;
+
 int usrp2::eth_data_transport::sendv(const iovec *iov, size_t iovlen){
-   return 0;
+    //create a new iov array with a space for ethernet header
+    // and move the current iovs to the center of the new array
+    size_t all_iov_len = iovlen + 1;
+    iovec all_iov[all_iov_len];
+    for (size_t i = 0; i < iovlen; i++){
+        all_iov[i+1] = iov[i];
+    }
+    //setup a new ethernet header
+    u2_eth_packet_only_t hdr;
+    hdr.ehdr.ethertype = htons(U2_DATA_ETHERTYPE);
+    memcpy(&hdr.ehdr.dst, d_mac.addr, 6);
+    memcpy(&hdr.ehdr.src, d_eth_data->mac(), 6);
+    hdr.thdr.flags = 0; // FIXME transport header values?
+    hdr.thdr.seqno = d_tx_seqno++;
+    hdr.thdr.ack = 0;
+
+    return d_eth_data->tx_framev(all_iov, all_iov_len);
 }
 
 int usrp2::eth_data_transport::recv(void **buff){
+    void *base;
+
+    DEBUG_LOG(":");
+    // Receive available frames from ethernet buffer.  Handler will
+    // process control frames, enqueue data packets in channel
+    // rings, and signal blocked API threads
+    int len = d_eth_data->rx_frame(&base, 100); // FIXME magic timeout
+    
+    u2_eth_samples_t *pkt = (u2_eth_samples_t *)base;
+    d_num_rx_frames++;
+    d_num_rx_bytes += len;
+    
+    /* --- FIXME start of fake transport layer handler --- */
+
+    if (d_rx_seqno != -1) {
+      int expected_seqno = (d_rx_seqno + 1) & 0xFF;
+      int seqno = pkt->hdrs.thdr.seqno; 
+      
+      if (seqno != expected_seqno) {
+        DEBUG_LOG("S"); // missing sequence number
+        int missing = seqno - expected_seqno;
+        if (missing < 0)
+            missing += 256;
+        d_num_rx_overruns++;
+        d_num_rx_missing += missing;
+      }
+    }
+
+    d_rx_seqno = pkt->hdrs.thdr.seqno;
+
+    /* --- end of fake transport layer handler --- */
+    
     return 0;
 }
diff --git a/usrp2/host/lib/eth_data_transport.h 
b/usrp2/host/lib/eth_data_transport.h
index c624fa1..9780a43 100644
--- a/usrp2/host/lib/eth_data_transport.h
+++ b/usrp2/host/lib/eth_data_transport.h
@@ -19,6 +19,7 @@
 #ifndef INCLUDED_ETH_DATA_TRANSPORT_H
 #define INCLUDED_ETH_DATA_TRANSPORT_H
 
+#include <usrp2/data_handler.h>
 #include "transport.h"
 #include "eth_buffer.h"
 #include "pktfilter.h"
@@ -32,6 +33,12 @@ namespace usrp2{
         pktfilter     *d_pf_data;
         u2_mac_addr_t d_mac;
         int           d_tx_seqno;
+        int           d_rx_seqno;
+        data_handler::result operator()(const void *base, size_t len);
+        unsigned int   d_num_rx_frames;
+        unsigned int   d_num_rx_missing;
+        unsigned int   d_num_rx_overruns;
+        unsigned int   d_num_rx_bytes;
 
     public:
         eth_data_transport(const std::string &ifc, u2_mac_addr_t mac, size_t 
rx_bufsize);
diff --git a/usrp2/host/lib/ring.cc b/usrp2/host/lib/ring.cc
index 71bfadd..9d06fc5 100644
--- a/usrp2/host/lib/ring.cc
+++ b/usrp2/host/lib/ring.cc
@@ -46,7 +46,7 @@ namespace usrp2 {
   }
 
   bool
-  ring::enqueue(void *p, size_t len)
+  ring::enqueue(void *p, size_t len, cb_t cb)
   {
     gruel::scoped_lock l(d_mutex);
     if (full())
@@ -54,6 +54,7 @@ namespace usrp2 {
       
     d_ring[d_write_ind].d_len = len;
     d_ring[d_write_ind].d_base = p;
+    d_ring[d_write_ind].d_cb = cb;
 
     inc_write_ind();
     d_not_empty.notify_one();
@@ -61,7 +62,7 @@ namespace usrp2 {
   }
 
   bool
-  ring::dequeue(void **p, size_t *len)
+  ring::dequeue(void **p, size_t *len, cb_t *cb)
   {
     gruel::scoped_lock l(d_mutex);
     if (empty())
@@ -69,6 +70,7 @@ namespace usrp2 {
       
     *p   = d_ring[d_read_ind].d_base;
     *len = d_ring[d_read_ind].d_len;
+    *cb = d_ring[d_read_ind].d_cb;
 
     inc_read_ind();
     return true;
diff --git a/usrp2/host/lib/ring.h b/usrp2/host/lib/ring.h
index 3437132..5bd78c2 100644
--- a/usrp2/host/lib/ring.h
+++ b/usrp2/host/lib/ring.h
@@ -33,6 +33,10 @@ namespace usrp2 {
 
   class ring
   {
+  public:
+    //typedef for void no argument function
+    typedef boost::function<void()> cb_t;
+
   private:
  
     size_t d_max;
@@ -43,6 +47,7 @@ namespace usrp2 {
     {
       void *d_base;
       size_t d_len;
+      cb_t d_cb;
     };
     std::vector<ring_desc> d_ring;
 
@@ -69,13 +74,13 @@ namespace usrp2 {
     bool full() const { return (d_write_ind+1)%d_max == d_read_ind; }
 
   public:
-    
+
     ring(unsigned int entries);
 
     void wait_for_not_empty();
 
-    bool enqueue(void *p, size_t len);
-    bool dequeue(void **p, size_t *len);
+    bool enqueue(void *p, size_t len, cb_t cb);
+    bool dequeue(void **p, size_t *len, cb_t *cb);
   };
 
 }  // namespace usrp2
diff --git a/usrp2/host/lib/usrp2_impl.cc b/usrp2/host/lib/usrp2_impl.cc
index b4c32e6..075b331 100644
--- a/usrp2/host/lib/usrp2_impl.cc
+++ b/usrp2/host/lib/usrp2_impl.cc
@@ -472,7 +472,9 @@ namespace usrp2 {
       // Strip off ethernet header and transport header and enqueue the rest
       
       size_t offset = offsetof(u2_eth_samples_t, hdrs.fixed);
-      if (d_channel_rings[chan]->enqueue(&pkt->hdrs.fixed, len-offset)) {
+      
+      ring::cb_t callback = boost::bind(&eth_buffer::release_frame, 
d_eth_data, (void*)base);
+      if (d_channel_rings[chan]->enqueue(&pkt->hdrs.fixed, len-offset, 
callback)) {
        inc_enqueued();
        DEBUG_LOG("+");
        return data_handler::KEEP;      // channel ring runner will mark frame 
done
@@ -737,8 +739,9 @@ namespace usrp2 {
     
     // Iterate through frames and present to user
     void *p;
+    ring::cb_t callback;
     size_t frame_len_in_bytes;
-    while (rp->dequeue(&p, &frame_len_in_bytes)) {
+    while (rp->dequeue(&p, &frame_len_in_bytes, &callback)) {
       uint32_t        *items;                  // points to beginning of data 
items
       size_t           nitems_in_uint32s;
       rx_metadata      md;
@@ -746,8 +749,8 @@ namespace usrp2 {
        return false;
 
       bool want_more = (*handler)(items, nitems_in_uint32s, &md);
-      d_eth_data->release_frame(p);
       DEBUG_LOG("-");
+      callback();
       dec_enqueued();
 
       if (!want_more)
@@ -778,9 +781,10 @@ namespace usrp2 {
 
     // Iterate through frames and drop them
     void *p;
+    ring::cb_t callback;
     size_t frame_len_in_bytes;
-    while (rp->dequeue(&p, &frame_len_in_bytes)) {
-      d_eth_data->release_frame(p);
+    while (rp->dequeue(&p, &frame_len_in_bytes, &callback)) {
+      callback();
       dec_enqueued();
     }
     return true;
diff --git a/usrp2/host/lib/usrp2_impl.h b/usrp2/host/lib/usrp2_impl.h
index 3d99915..25c7c60 100644
--- a/usrp2/host/lib/usrp2_impl.h
+++ b/usrp2/host/lib/usrp2_impl.h
@@ -71,7 +71,7 @@ namespace usrp2 {
     int            d_rx_seqno;
     int            d_tx_seqno;
     int            d_next_rid;
-    unsigned int   d_num_rx_frames;
+    unsigned int   d_num_rx_frames; //TODO remove this stuff, its in transport 
data
     unsigned int   d_num_rx_missing;
     unsigned int   d_num_rx_overruns;
     unsigned int   d_num_rx_bytes;