Merge branch 'next'

5 files changed, 74 insertions, 77 deletions

diff --git a/host/lib/transport/gen_vrt_if_packet.py b/host/lib/transport/gen_vrt_if_packet.py
index 245a7ddbd..e28ce3aae 100755..100644
--- a/host/lib/transport/gen_vrt_if_packet.py
+++ b/host/lib/transport/gen_vrt_if_packet.py
@@ -137,7 +137,7 @@ void vrt::if_hdr_pack_$(suffix)(
         #if $pred & $tlr_p
             {
                 const size_t empty_bytes = if_packet_info.num_payload_words32*sizeof(boost::uint32_t) - if_packet_info.num_payload_bytes;
-                if_packet_info.tlr |= (0x3 << 22) | (occ_table[empty_bytes & 0x3] << 10);
+                if_packet_info.tlr = (0x3 << 22) | (occ_table[empty_bytes & 0x3] << 10);
             }
             packet_buff[$num_header_words+if_packet_info.num_payload_words32] = $(XE_MACRO)(if_packet_info.tlr);
             #set $flags |= (0x1 << 26);
diff --git a/host/lib/transport/libusb1_zero_copy.cpp b/host/lib/transport/libusb1_zero_copy.cpp
index 28d6cdd5b..3e67264cd 100644
--- a/host/lib/transport/libusb1_zero_copy.cpp
+++ b/host/lib/transport/libusb1_zero_copy.cpp
@@ -80,13 +80,14 @@ UHD_INLINE bool wait_for_completion(libusb_context *ctx, const double timeout, b
  **********************************************************************/
 class libusb_zero_copy_mrb : public managed_recv_buffer{
 public:
-    libusb_zero_copy_mrb(libusb_transfer *lut):
+    libusb_zero_copy_mrb(libusb_transfer *lut, const size_t frame_size):
         _ctx(libusb::session::get_global_session()->get_context()),
-        _lut(lut), _expired(false) { /* NOP */ }
+        _lut(lut), _expired(false), _frame_size(frame_size) { /* NOP */ }
 
     void release(void){
         if (_expired) return;
         completed = false;
+        _lut->length = _frame_size; //always reset length
         UHD_ASSERT_THROW(libusb_submit_transfer(_lut) == 0);
         _expired = true;
     }
@@ -109,6 +110,7 @@ private:
     libusb_context *_ctx;
     libusb_transfer *_lut;
     bool _expired;
+    const size_t _frame_size;
 };
 
 /***********************************************************************
@@ -118,9 +120,9 @@ private:
  **********************************************************************/
 class libusb_zero_copy_msb : public managed_send_buffer{
 public:
-    libusb_zero_copy_msb(libusb_transfer *lut):
+    libusb_zero_copy_msb(libusb_transfer *lut, const size_t frame_size):
         _ctx(libusb::session::get_global_session()->get_context()),
-        _lut(lut), _expired(false) { /* NOP */ }
+        _lut(lut), _expired(false), _frame_size(frame_size) { /* NOP */ }
 
     void commit(size_t len){
         if (_expired) return;
@@ -144,11 +146,12 @@ public:
 
 private:
     void *get_buff(void) const{return _lut->buffer;}
-    size_t get_size(void) const{return _lut->length;}
+    size_t get_size(void) const{return _frame_size;}
 
     libusb_context *_ctx;
     libusb_transfer *_lut;
     bool _expired;
+    const size_t _frame_size;
 };
 
 /***********************************************************************
@@ -184,7 +187,7 @@ public:
             libusb_transfer *lut = libusb_alloc_transfer(0);
             UHD_ASSERT_THROW(lut != NULL);
 
-            _mrb_pool.push_back(boost::shared_ptr<libusb_zero_copy_mrb>(new libusb_zero_copy_mrb(lut)));
+            _mrb_pool.push_back(boost::shared_ptr<libusb_zero_copy_mrb>(new libusb_zero_copy_mrb(lut, this->get_recv_frame_size())));
 
             libusb_fill_bulk_transfer(
                 lut,                                                    // transfer
@@ -207,7 +210,7 @@ public:
             libusb_transfer *lut = libusb_alloc_transfer(0);
             UHD_ASSERT_THROW(lut != NULL);
 
-            _msb_pool.push_back(boost::shared_ptr<libusb_zero_copy_msb>(new libusb_zero_copy_msb(lut)));
+            _msb_pool.push_back(boost::shared_ptr<libusb_zero_copy_msb>(new libusb_zero_copy_msb(lut, this->get_send_frame_size())));
 
             libusb_fill_bulk_transfer(
                 lut,                                                    // transfer
diff --git a/host/lib/transport/super_recv_packet_handler.hpp b/host/lib/transport/super_recv_packet_handler.hpp
index a5876c8bf..74fbe82fb 100644
--- a/host/lib/transport/super_recv_packet_handler.hpp
+++ b/host/lib/transport/super_recv_packet_handler.hpp
@@ -1,5 +1,5 @@
 //
-// Copyright 2011 Ettus Research LLC
+// Copyright 2011-2012 Ettus Research LLC
 //
 // This program is free software: you can redistribute it and/or modify
 // it under the terms of the GNU General Public License as published by
@@ -283,7 +283,7 @@ private:
         info.ifpi.num_packet_words32 = num_packet_words32 - _header_offset_words32;
         info.vrt_hdr = buff->cast<const boost::uint32_t *>() + _header_offset_words32;
         _vrt_unpacker(info.vrt_hdr, info.ifpi);
-        info.time = time_spec_t(time_t(info.ifpi.tsi), size_t(info.ifpi.tsf), _tick_rate); //assumes has_tsi and has_tsf are true
+        info.time = time_spec_t::from_ticks(info.ifpi.tsf, _tick_rate); //assumes has_tsf is true
         info.copy_buff = reinterpret_cast<const char *>(info.vrt_hdr + info.ifpi.num_header_words32);
 
         //--------------------------------------------------------------
@@ -408,7 +408,7 @@ private:
 
             case PACKET_INLINE_MESSAGE:
                 std::swap(curr_info, next_info); //save progress from curr -> next
-                curr_info.metadata.has_time_spec = next_info[index].ifpi.has_tsi and next_info[index].ifpi.has_tsf;
+                curr_info.metadata.has_time_spec = next_info[index].ifpi.has_tsf;
                 curr_info.metadata.time_spec = next_info[index].time;
                 curr_info.metadata.more_fragments = false;
                 curr_info.metadata.fragment_offset = 0;
@@ -436,7 +436,7 @@ private:
                 alignment_check(index, curr_info);
                 std::swap(curr_info, next_info); //save progress from curr -> next
                 curr_info.metadata.has_time_spec = prev_info.metadata.has_time_spec;
-                curr_info.metadata.time_spec = prev_info.metadata.time_spec + time_spec_t(0,
+                curr_info.metadata.time_spec = prev_info.metadata.time_spec + time_spec_t::from_ticks(
                     prev_info[index].ifpi.num_payload_words32*sizeof(boost::uint32_t)/_bytes_per_otw_item, _samp_rate);
                 curr_info.metadata.more_fragments = false;
                 curr_info.metadata.fragment_offset = 0;
@@ -469,7 +469,7 @@ private:
         }
 
         //set the metadata from the buffer information at index zero
-        curr_info.metadata.has_time_spec = curr_info[0].ifpi.has_tsi and curr_info[0].ifpi.has_tsf;
+        curr_info.metadata.has_time_spec = curr_info[0].ifpi.has_tsf;
         curr_info.metadata.time_spec = curr_info[0].time;
         curr_info.metadata.more_fragments = false;
         curr_info.metadata.fragment_offset = 0;
@@ -508,7 +508,7 @@ private:
         metadata = info.metadata;
 
         //interpolate the time spec (useful when this is a fragment)
-        metadata.time_spec += time_spec_t(0, info.fragment_offset_in_samps, _samp_rate);
+        metadata.time_spec += time_spec_t::from_ticks(info.fragment_offset_in_samps, _samp_rate);
 
         //extract the number of samples available to copy
         const size_t nsamps_available = info.data_bytes_to_copy/_bytes_per_otw_item;
diff --git a/host/lib/transport/super_send_packet_handler.hpp b/host/lib/transport/super_send_packet_handler.hpp
index 5ed8e0143..3d68507ed 100644
--- a/host/lib/transport/super_send_packet_handler.hpp
+++ b/host/lib/transport/super_send_packet_handler.hpp
@@ -1,5 +1,5 @@
 //
-// Copyright 2011 Ettus Research LLC
+// Copyright 2011-2012 Ettus Research LLC
 //
 // This program is free software: you can redistribute it and/or modify
 // it under the terms of the GNU General Public License as published by
@@ -134,11 +134,10 @@ public:
         vrt::if_packet_info_t if_packet_info;
         if_packet_info.has_sid = false;
         if_packet_info.has_cid = false;
-        if_packet_info.has_tlr = false;
-        if_packet_info.has_tsi = metadata.has_time_spec;
+        if_packet_info.has_tlr = true;
+        if_packet_info.has_tsi = false;
         if_packet_info.has_tsf = metadata.has_time_spec;
-        if_packet_info.tsi     = boost::uint32_t(metadata.time_spec.get_full_secs());
-        if_packet_info.tsf     = boost::uint64_t(metadata.time_spec.get_tick_count(_tick_rate));
+        if_packet_info.tsf     = metadata.time_spec.to_ticks(_tick_rate);
         if_packet_info.sob     = metadata.start_of_burst;
         if_packet_info.eob     = metadata.end_of_burst;
 
@@ -174,9 +173,8 @@ public:
             if (num_samps_sent == 0) return total_num_samps_sent;
 
             //setup metadata for the next fragment
-            const time_spec_t time_spec = metadata.time_spec + time_spec_t(0, total_num_samps_sent, _samp_rate);
-            if_packet_info.tsi = boost::uint32_t(time_spec.get_full_secs());
-            if_packet_info.tsf = boost::uint64_t(time_spec.get_tick_count(_tick_rate));
+            const time_spec_t time_spec = metadata.time_spec + time_spec_t::from_ticks(total_num_samps_sent, _samp_rate);
+            if_packet_info.tsf = time_spec.to_ticks(_tick_rate);
             if_packet_info.sob = false;
 
         }
diff --git a/host/lib/transport/usb_zero_copy_wrapper.cpp b/host/lib/transport/usb_zero_copy_wrapper.cpp
index 227c4b392..690e5aaa2 100644
--- a/host/lib/transport/usb_zero_copy_wrapper.cpp
+++ b/host/lib/transport/usb_zero_copy_wrapper.cpp
@@ -1,5 +1,5 @@
 //
-// Copyright 2011 Ettus Research LLC
+// Copyright 2011-2012 Ettus Research LLC
 //
 // This program is free software: you can redistribute it and/or modify
 // it under the terms of the GNU General Public License as published by
@@ -18,17 +18,13 @@
 #include <uhd/transport/usb_zero_copy.hpp>
 #include <uhd/transport/bounded_buffer.hpp>
 #include <uhd/transport/buffer_pool.hpp>
+#include <uhd/utils/byteswap.hpp>
+#include <uhd/utils/msg.hpp>
 #include <boost/foreach.hpp>
 #include <vector>
 #include <iostream>
 
 using namespace uhd::transport;
-bool debug = true;
-
-static inline size_t next_boundary(size_t length, size_t boundary){
-    //pad to the boundary, assumes boundary is a power of 2
-    return (length + (boundary-1)) & ~(boundary-1);
-}
 
 /***********************************************************************
  * USB zero copy wrapper - managed receive buffer
@@ -45,7 +41,7 @@ public:
         _mrb.reset();
     }
 
-    sptr get_new(managed_recv_buffer::sptr mrb, const void *mem, size_t len){
+    UHD_INLINE sptr get_new(managed_recv_buffer::sptr mrb, const void *mem, size_t len){
         _mrb = mrb;
         _mem = mem;
         _len = len;
@@ -67,28 +63,46 @@ private:
  **********************************************************************/
 class usb_zero_copy_wrapper_msb : public managed_send_buffer{
 public:
-    usb_zero_copy_wrapper_msb(bounded_buffer<usb_zero_copy_wrapper_msb *> &queue, size_t boundary):
-        _queue(queue), _boundary(boundary){/*NOP*/}
+    usb_zero_copy_wrapper_msb(const usb_zero_copy::sptr internal, const size_t fragmentation_size):
+        _internal(internal), _fragmentation_size(fragmentation_size){/*NOP*/}
 
     void commit(size_t len){
-        if (_msb.get() == NULL) return;
-        _msb->commit(next_boundary(len, _boundary));
-        _queue.push_with_haste(this);
-        _msb.reset();
+        if (len == 0) return;
+
+        //get a reference to the VITA header before incrementing
+        const boost::uint32_t vita_header = reinterpret_cast<const boost::uint32_t *>(_mem_buffer_tip)[0];
+
+        _bytes_in_buffer += len;
+        _mem_buffer_tip += len;
+
+        //extract VITA end of packet flag, we must force flush under eof conditions
+        const bool eop = (uhd::wtohx(vita_header) & (0x1 << 24)) != 0;
+        const bool full = _bytes_in_buffer >= (_last_send_buff->size() - _fragmentation_size);
+        if (eop or full){
+            _last_send_buff->commit(_bytes_in_buffer);
+            _last_send_buff.reset();
+        }
     }
 
-    sptr get_new(managed_send_buffer::sptr msb){
-        _msb = msb;
+    UHD_INLINE sptr get_new(const double timeout){
+        if (not _last_send_buff){
+            _last_send_buff = _internal->get_send_buff(timeout);
+            if (not _last_send_buff) return sptr();
+            _mem_buffer_tip = _last_send_buff->cast<char *>();
+            _bytes_in_buffer = 0;
+        }
         return make_managed_buffer(this);
     }
 
 private:
-    void *get_buff(void) const{return _msb->cast<void *>();}
-    size_t get_size(void) const{return _msb->size();}
-
-    bounded_buffer<usb_zero_copy_wrapper_msb *> &_queue;
-    size_t _boundary;
-    managed_send_buffer::sptr _msb;
+    void *get_buff(void) const{return reinterpret_cast<void *>(_mem_buffer_tip);}
+    size_t get_size(void) const{return _fragmentation_size;}
+
+    usb_zero_copy::sptr _internal;
+    const size_t _fragmentation_size;
+    managed_send_buffer::sptr _last_send_buff;
+    size_t _bytes_in_buffer;
+    char *_mem_buffer_tip;
 };
 
 /***********************************************************************
@@ -96,23 +110,16 @@ private:
  **********************************************************************/
 class usb_zero_copy_wrapper : public usb_zero_copy{
 public:
-    usb_zero_copy_wrapper(
-        sptr usb_zc, size_t usb_frame_boundary
-    ):
+    usb_zero_copy_wrapper(sptr usb_zc, const size_t frame_boundary):
         _internal_zc(usb_zc),
-        _usb_frame_boundary(usb_frame_boundary),
+        _frame_boundary(frame_boundary),
         _available_recv_buffs(this->get_num_recv_frames()),
-        _available_send_buffs(this->get_num_send_frames()),
         _mrb_pool(this->get_num_recv_frames(), usb_zero_copy_wrapper_mrb(_available_recv_buffs)),
-        _msb_pool(this->get_num_send_frames(), usb_zero_copy_wrapper_msb(_available_send_buffs, usb_frame_boundary))
+        _the_only_msb(usb_zero_copy_wrapper_msb(usb_zc, frame_boundary))
     {
         BOOST_FOREACH(usb_zero_copy_wrapper_mrb &mrb, _mrb_pool){
             _available_recv_buffs.push_with_haste(&mrb);
         }
-
-        BOOST_FOREACH(usb_zero_copy_wrapper_msb &msb, _msb_pool){
-            _available_send_buffs.push_with_haste(&msb);
-        }
     }
 
     managed_recv_buffer::sptr get_recv_buff(double timeout){
@@ -128,18 +135,17 @@ public:
             //extract this packet's memory address and length in bytes
             const char *mem = _last_recv_buff->cast<const char *>() + _last_recv_offset;
             const boost::uint32_t *mem32 = reinterpret_cast<const boost::uint32_t *>(mem);
-            size_t len = (mem32[0] & 0xffff)*sizeof(boost::uint32_t); //length in bytes (from VRT header)
-            
+            const size_t len = (uhd::wtohx(mem32[0]) & 0xffff)*sizeof(boost::uint32_t); //length in bytes (from VRT header)
+
             managed_recv_buffer::sptr recv_buff; //the buffer to be returned to the user
-            
             recv_buff = wmrb->get_new(_last_recv_buff, mem, len);
-            _last_recv_offset = next_boundary(_last_recv_offset + len, _usb_frame_boundary);
-            
+            _last_recv_offset += len;
+
             //check if this receive buffer has been exhausted
             if (_last_recv_offset >= _last_recv_buff->size()) {
                 _last_recv_buff.reset();
             }
-            
+
             return recv_buff;
         }
 
@@ -152,20 +158,11 @@ public:
     }
 
     size_t get_recv_frame_size(void) const{
-        return _internal_zc->get_recv_frame_size();
+        return std::min(_frame_boundary, _internal_zc->get_recv_frame_size());
     }
 
     managed_send_buffer::sptr get_send_buff(double timeout){
-        managed_send_buffer::sptr send_buff = _internal_zc->get_send_buff(timeout);
-        
-        //attempt to get a wrapper for a managed send buffer
-        usb_zero_copy_wrapper_msb *wmsb = NULL;
-        if (send_buff.get() and _available_send_buffs.pop_with_haste(wmsb)){
-            return wmsb->get_new(send_buff);
-        }
-
-        //otherwise return a null sptr for failure
-        return managed_send_buffer::sptr();
+        return _the_only_msb.get_new(timeout);
     }
 
     size_t get_num_send_frames(void) const{
@@ -173,20 +170,19 @@ public:
     }
 
     size_t get_send_frame_size(void) const{
-        return _internal_zc->get_send_frame_size();
+        return std::min(_frame_boundary, _internal_zc->get_send_frame_size());
     }
 
 private:
     sptr _internal_zc;
-    size_t _usb_frame_boundary;
+    size_t _frame_boundary;
     bounded_buffer<usb_zero_copy_wrapper_mrb *> _available_recv_buffs;
-    bounded_buffer<usb_zero_copy_wrapper_msb *> _available_send_buffs;
     std::vector<usb_zero_copy_wrapper_mrb> _mrb_pool;
-    std::vector<usb_zero_copy_wrapper_msb> _msb_pool;
-    
+    usb_zero_copy_wrapper_msb _the_only_msb;
+
     //buffer to store partially-received VRT packets in
     buffer_pool::sptr _fragment_mem;
-    
+
     //state for last recv buffer to create multiple managed buffers
     managed_recv_buffer::sptr _last_recv_buff;
     size_t _last_recv_offset;