aboutsummaryrefslogtreecommitdiffstats
path: root/host
diff options
context:
space:
mode:
Diffstat (limited to 'host')
-rw-r--r--host/lib/usrp/b100/b100_impl.cpp2
-rw-r--r--host/lib/usrp/b100/b100_impl.hpp4
-rw-r--r--host/lib/usrp/b100/io_impl.cpp8
-rw-r--r--host/lib/usrp/common/recv_packet_demuxer_3000.hpp114
4 files changed, 119 insertions, 9 deletions
diff --git a/host/lib/usrp/b100/b100_impl.cpp b/host/lib/usrp/b100/b100_impl.cpp
index b3237d547..1673f5f3c 100644
--- a/host/lib/usrp/b100/b100_impl.cpp
+++ b/host/lib/usrp/b100/b100_impl.cpp
@@ -472,7 +472,7 @@ b100_impl::b100_impl(const device_addr_t &device_addr){
}
//initialize io handling
- _recv_demuxer = recv_packet_demuxer::make(_data_transport, _rx_dsps.size(), B100_RX_SID_BASE);
+ _recv_demuxer.reset(new recv_packet_demuxer_3000(_data_transport));
//allocate streamer weak ptrs containers
_rx_streamers.resize(_rx_dsps.size());
diff --git a/host/lib/usrp/b100/b100_impl.hpp b/host/lib/usrp/b100/b100_impl.hpp
index 68d7043a1..5f4899eb3 100644
--- a/host/lib/usrp/b100/b100_impl.hpp
+++ b/host/lib/usrp/b100/b100_impl.hpp
@@ -29,7 +29,7 @@
#include "time64_core_200.hpp"
#include "fifo_ctrl_excelsior.hpp"
#include "user_settings_core_200.hpp"
-#include "recv_packet_demuxer.hpp"
+#include "recv_packet_demuxer_3000.hpp"
#include <uhd/device.hpp>
#include <uhd/property_tree.hpp>
#include <uhd/types/dict.hpp>
@@ -105,7 +105,7 @@ private:
//transports
uhd::transport::zero_copy_if::sptr _ctrl_transport;
uhd::transport::zero_copy_if::sptr _data_transport;
- uhd::usrp::recv_packet_demuxer::sptr _recv_demuxer;
+ boost::shared_ptr<uhd::usrp::recv_packet_demuxer_3000> _recv_demuxer;
//dboard stuff
uhd::usrp::dboard_manager::sptr _dboard_manager;
diff --git a/host/lib/usrp/b100/io_impl.cpp b/host/lib/usrp/b100/io_impl.cpp
index bcf712cd9..9b5d4d25c 100644
--- a/host/lib/usrp/b100/io_impl.cpp
+++ b/host/lib/usrp/b100/io_impl.cpp
@@ -1,5 +1,5 @@
//
-// Copyright 2011-2012 Ettus Research LLC
+// Copyright 2011-2013 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
@@ -153,17 +153,13 @@ rx_streamer::sptr b100_impl::get_rx_stream(const uhd::stream_args_t &args_){
id.num_outputs = 1;
my_streamer->set_converter(id);
- //flush stuff
- _data_transport->get_recv_buff(-1.0); //negative flushes!!
- _recv_demuxer = recv_packet_demuxer::make(_data_transport, _rx_dsps.size(), B100_RX_SID_BASE);
-
//bind callbacks for the handler
for (size_t chan_i = 0; chan_i < args.channels.size(); chan_i++){
const size_t dsp = args.channels[chan_i];
_rx_dsps[dsp]->set_nsamps_per_packet(spp); //seems to be a good place to set this
_rx_dsps[dsp]->setup(args);
my_streamer->set_xport_chan_get_buff(chan_i, boost::bind(
- &recv_packet_demuxer::get_recv_buff, _recv_demuxer, dsp, _1
+ &recv_packet_demuxer_3000::get_recv_buff, _recv_demuxer, B100_RX_SID_BASE + dsp, _1
), true /*flush*/);
my_streamer->set_overflow_handler(chan_i, boost::bind(
&rx_dsp_core_200::handle_overflow, _rx_dsps[dsp]
diff --git a/host/lib/usrp/common/recv_packet_demuxer_3000.hpp b/host/lib/usrp/common/recv_packet_demuxer_3000.hpp
new file mode 100644
index 000000000..d4c899f1c
--- /dev/null
+++ b/host/lib/usrp/common/recv_packet_demuxer_3000.hpp
@@ -0,0 +1,114 @@
+//
+// Copyright 2013 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
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program 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 this program. If not, see <http://www.gnu.org/licenses/>.
+//
+
+#ifndef INCLUDED_LIBUHD_USRP_COMMON_RECV_PACKET_DEMUXER_3000_HPP
+#define INCLUDED_LIBUHD_USRP_COMMON_RECV_PACKET_DEMUXER_3000_HPP
+
+#include <uhd/config.hpp>
+#include <uhd/transport/zero_copy.hpp>
+#include <boost/cstdint.hpp>
+#include <boost/thread.hpp>
+#include <uhd/utils/msg.hpp>
+#include <uhd/utils/atomic.hpp>
+#include <uhd/types/time_spec.hpp>
+#include <uhd/utils/byteswap.hpp>
+#include <queue>
+#include <map>
+
+namespace uhd{ namespace usrp{
+
+ struct recv_packet_demuxer_3000
+ {
+ recv_packet_demuxer_3000(transport::zero_copy_if::sptr xport):
+ _xport(xport)
+ {/*NOP*/}
+
+ transport::managed_recv_buffer::sptr get_recv_buff(const boost::uint32_t sid, const double timeout)
+ {
+ const time_spec_t exit_time = time_spec_t(timeout) + time_spec_t::get_system_time();
+ transport::managed_recv_buffer::sptr buff;
+ buff = _internal_get_recv_buff(sid, timeout);
+ while (not buff) //loop until timeout
+ {
+ const time_spec_t delta = exit_time - time_spec_t::get_system_time();
+ const double new_timeout = delta.get_real_secs();
+ if (new_timeout < 0.0) break;
+ buff = _internal_get_recv_buff(sid, new_timeout);
+ }
+ return buff;
+ }
+
+ transport::managed_recv_buffer::sptr _internal_get_recv_buff(const boost::uint32_t sid, const double timeout)
+ {
+ transport::managed_recv_buffer::sptr buff;
+
+ //----------------------------------------------------------
+ //-- Check the queue to see if we already have a buffer
+ //----------------------------------------------------------
+ {
+ boost::mutex::scoped_lock l(mutex);
+ queue_type_t &queue = _queues[sid];
+ if (not queue.empty())
+ {
+ buff = queue.front();
+ queue.pop();
+ return buff;
+ }
+ }
+
+ //----------------------------------------------------------
+ //-- Try to claim the transport or wait patiently
+ //----------------------------------------------------------
+ if (_claimed.cas(1, 0))
+ {
+ boost::mutex::scoped_lock l(mutex);
+ cond.timed_wait(l, boost::posix_time::microseconds(long(timeout*1e6)));
+ }
+
+ //----------------------------------------------------------
+ //-- Wait on the transport for input buffers
+ //----------------------------------------------------------
+ else
+ {
+ buff = _xport->get_recv_buff(timeout);
+ if (buff)
+ {
+ const boost::uint32_t new_sid = uhd::wtohx(buff->cast<const boost::uint32_t *>()[1]);
+ if (new_sid != sid)
+ {
+ boost::mutex::scoped_lock l(mutex);
+ _queues[new_sid].push(buff);
+ buff.reset();
+ }
+ }
+ _claimed.write(0);
+ cond.notify_all();
+ }
+ return buff;
+ }
+
+ typedef std::queue<transport::managed_recv_buffer::sptr> queue_type_t;
+ std::map<boost::uint32_t, queue_type_t> _queues;
+ transport::zero_copy_if::sptr _xport;
+ uhd::atomic_uint32_t _claimed;
+ boost::condition_variable cond;
+ boost::mutex mutex;
+ };
+
+}} //namespace uhd::usrp
+
+#endif /* INCLUDED_LIBUHD_USRP_COMMON_RECV_PACKET_DEMUXER_3000_HPP */
generated by cgit v1.2.3 (git 2.39.1) at 2024-12-28 00:39:13 +0000