[Commit-gnuradio] r8848 - in usrp2/branches/features/host-ng/host-ng: apps lib

[jcorgan]

Author: jcorgan
Date: 2008-07-09 18:16:30 -0600 (Wed, 09 Jul 2008)
New Revision: 8848

Modified:
   usrp2/branches/features/host-ng/host-ng/apps/test_usrp2.cc
   usrp2/branches/features/host-ng/host-ng/lib/eth_buffer.cc
   usrp2/branches/features/host-ng/host-ng/lib/usrp2_impl.cc
   usrp2/branches/features/host-ng/host-ng/lib/usrp2_impl.h
Log:
work-in-progress with threaded enet driver.  deadlocks occurring at low 
decimation rates.

Modified: usrp2/branches/features/host-ng/host-ng/apps/test_usrp2.cc
===================================================================
--- usrp2/branches/features/host-ng/host-ng/apps/test_usrp2.cc  2008-07-09 
21:33:39 UTC (rev 8847)
+++ usrp2/branches/features/host-ng/host-ng/apps/test_usrp2.cc  2008-07-10 
00:16:30 UTC (rev 8848)
@@ -134,21 +134,9 @@
     exit(1);
   }
 
-  // Wait while streaming
-  struct timespec ts;
-  ts.tv_sec = 4;
-  ts.tv_nsec = 0;
-  int r = nanosleep(&ts, 0);
-  if (r == -1)
-    perror("nanosleep");
-  u2->stop_rx_streaming();
+  while(1)
+    u2->rx_samples(0, NULL);
 
-  // Wait to be able to see stray packets
-  ts.tv_sec = 1;
-  ts.tv_nsec = 0;
-  r = nanosleep(&ts, 0);
-  if (r == -1)
-    perror("nanosleep");
-
+  u2->stop_rx_streaming();
   return 0;
 }

Modified: usrp2/branches/features/host-ng/host-ng/lib/eth_buffer.cc
===================================================================
--- usrp2/branches/features/host-ng/host-ng/lib/eth_buffer.cc   2008-07-09 
21:33:39 UTC (rev 8847)
+++ usrp2/branches/features/host-ng/host-ng/lib/eth_buffer.cc   2008-07-10 
00:16:30 UTC (rev 8848)
@@ -173,7 +173,7 @@
   int 
   eth_buffer::rx_frames(data_handler *f, int timeout)
   {
-    DEBUG_LOG("'");
+    DEBUG_LOG("\n");
       
     while (!frame_available()) {
       if (timeout == 0) {
@@ -191,18 +191,15 @@
       int pres = poll(&pfd, 1, timeout);
       if (pres == -1) {
         perror("poll");
-        DEBUG_LOG("E");
        return -1;
       }
 
       if (pres == 0) {
-        DEBUG_LOG("-");
+        DEBUG_LOG("t");
        return 2;
       }
     }
 
-    DEBUG_LOG("R");
-    
     // Iterate through available packets
     while (frame_available()) {
       // Get start of ethernet frame and length
@@ -212,10 +209,10 @@
       
       // Invoke data handler
       data_handler::result r = (*f)(base, len);
-      if (!(r & data_handler::KEEP))
+      if (!(r & data_handler::KEEP)) {
         hdr->tp_status = TP_STATUS_KERNEL; // mark it free
-      else
-        DEBUG_LOG("K");
+      }
+
       inc_head();
 
       if (r & data_handler::DONE)
@@ -243,13 +240,8 @@
   eth_buffer::release_frame(void *base)
   {
     // Get d_frame_size aligned header
-    
-    tpacket_hdr *hdr = (tpacket_hdr *)
-      ((((size_t)base+d_frame_size-1) & ~(d_frame_size-1))); 
-
+    tpacket_hdr *hdr = (tpacket_hdr *)((size_t)base & ~(d_frame_size-1));
     hdr->tp_status = TP_STATUS_KERNEL; // mark it free
-    DEBUG_LOG("f");
   }
   
 } // namespace usrp2
-

Modified: usrp2/branches/features/host-ng/host-ng/lib/usrp2_impl.cc
===================================================================
--- usrp2/branches/features/host-ng/host-ng/lib/usrp2_impl.cc   2008-07-09 
21:33:39 UTC (rev 8847)
+++ usrp2/branches/features/host-ng/host-ng/lib/usrp2_impl.cc   2008-07-10 
00:16:30 UTC (rev 8848)
@@ -78,7 +78,8 @@
   usrp2::impl::impl(const std::string &ifc, const std::string &addr)
     : d_buffer(new eth_buffer()), d_pf(0), d_bg_thread(0), d_bg_running(false),
       d_rx_decim(0), d_rx_seqno(-1), d_tx_seqno(0), d_next_rid(0),
-      d_num_rx_frames(0), d_num_rx_lost(0), d_num_rx_bytes(0)
+      d_num_rx_frames(0), d_num_rx_lost(0), d_num_rx_bytes(0), 
d_num_enqueued(0),
+      d_enqueued_mutex(), d_bg_pending_mutex(), 
d_bg_pending_cond(&d_bg_pending_mutex)
   {
     props_vector_t u2s = find(ifc, addr);
     if (u2s.size() != 1)
@@ -95,7 +96,7 @@
     
     if (USRP2_IMPL_DEBUG)
       std::cerr << "usrp2 constructor: using USRP2 at " << d_addr << std::endl;
-    
+
     memset(d_pending_replies, 0, sizeof(d_pending_replies));
     memset(d_channel_rings, 0, sizeof(d_channel_rings));
 
@@ -188,9 +189,6 @@
   bool 
   usrp2::impl::set_rx_gain(double gain)
   {
-    if (USRP2_IMPL_DEBUG)
-      std::cerr << "usrp2: setting rx gain to " << gain << "...";
-
     op_config_rx_v2_cmd cmd;
     op_config_rx_reply_v2_t reply;
 
@@ -203,17 +201,12 @@
       return false;
 
     bool success = (ntohx(reply.ok) == 1);
-    if (USRP2_IMPL_DEBUG)
-      std::cerr << (success ? "ok." : "fail.") << std::endl;
     return success;
   }
   
   bool
   usrp2::impl::set_rx_center_freq(double frequency, tune_result *result)
   {
-    if (USRP2_IMPL_DEBUG)
-      std::cerr << "usrp2: setting rx freq to " << frequency << "...";
-
     op_config_rx_v2_cmd cmd;
     op_config_rx_reply_v2_t reply;
 
@@ -228,9 +221,6 @@
       return false;
 
     bool success = (ntohx(reply.ok) == 1);
-    if (USRP2_IMPL_DEBUG)
-      std::cerr << (success ? "ok." : "fail.") << std::endl;
-
     if (result && success) {
       result->baseband_freq =
         u2_fxpt_freq_to_double( 
@@ -256,9 +246,6 @@
   bool
   usrp2::impl::set_rx_decim(int decimation_factor)
   {
-    if (USRP2_IMPL_DEBUG)
-      std::cerr << "usrp2: setting rx decimation to " << decimation_factor << 
"...";
-
     op_config_rx_v2_cmd cmd;
     op_config_rx_reply_v2_t reply;
 
@@ -271,18 +258,12 @@
       return false;
 
     bool success = (ntohx(reply.ok) == 1);
-    if (USRP2_IMPL_DEBUG)
-      std::cerr << (success ? "ok." : "fail.") << std::endl;
     return success;
   }
   
   bool
   usrp2::impl::set_rx_scale_iq(int scale_i, int scale_q)
   {
-    if (USRP2_IMPL_DEBUG)
-      std::cerr << "usrp2: setting rx scale to ("
-                << scale_i << ", " << scale_q << ")...";
-
     op_config_rx_v2_cmd cmd;
     op_config_rx_reply_v2_t reply;
 
@@ -295,18 +276,12 @@
       return false;
 
     bool success = (ntohx(reply.ok) == 1);
-    if (USRP2_IMPL_DEBUG)
-      std::cerr << (success ? "ok." : "fail.") << std::endl;
     return success;
   }
   
   bool
   usrp2::impl::start_rx_streaming(unsigned int channel, unsigned int 
items_per_frame)
   {
-    if (USRP2_IMPL_DEBUG)
-      std::cerr << "usrp2: starting rx streaming on channel "
-               << channel << "...";
-
     if (channel > MAX_CHAN) {
       std::cerr << "usrp2: invalid channel number (" << channel
                << ")" << std::endl;
@@ -347,18 +322,12 @@
       return false;
 
     bool success = (ntohx(reply.ok) == 1);
-    if (USRP2_IMPL_DEBUG)
-      std::cerr << (success ? "ok." : "fail.") << std::endl;
     return success;
   }
   
   bool
   usrp2::impl::stop_rx_streaming(unsigned int channel)
   {
-    if (USRP2_IMPL_DEBUG)
-      std::cerr << "usrp2: stopping rx streaming on channel "
-               << channel << std::endl;
-
     if (channel > MAX_CHAN) {
       std::cerr << "usrp2: invalid channel number (" << channel
                << ")" << std::endl;
@@ -393,9 +362,6 @@
       return false;
 
     bool success = (ntohx(reply.ok) == 1);
-    if (USRP2_IMPL_DEBUG)
-      std::cerr << (success ? "ok." : "fail.") << std::endl;
-
     if (success) {
       ring *rp = d_channel_rings[channel];
       d_channel_rings[channel] = 0;
@@ -405,14 +371,44 @@
     return success;
   }
   
+  bool
+  usrp2::impl::transmit_cmd(void *cmd, int len, pending_reply *p, double secs)
+  {
+    if (p)    
+      d_pending_replies[p->rid()] = p;
+    
+    // Transmit command
+    if (d_buffer->tx_frame(cmd, len) != 1) {
+      d_pending_replies[p->rid()] = 0;
+      return false;
+    }
+
+    int res = 1;
+    if (p)
+      res = p->wait(secs);
+      
+    d_pending_replies[p->rid()] = 0;
+    return res == 1;
+  }
+
   void
   usrp2::impl::bg_loop()
   {
     d_bg_running = true;
     while(d_bg_running) {
-      int res = d_buffer->rx_frames(this, 100);
+      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 res = d_buffer->rx_frames(this, 100); // FIXME magic timeout
       if (res == -1)
        break;  
+
+      // Wait for user API thread(s) to process all enqueued packets.
+      // The channel ring runner that decrements d_num_enqueued to zero 
+      // will signal this thread to continue.
+      while(d_num_enqueued)
+       d_bg_pending_cond.wait();
     }
   }
   
@@ -423,37 +419,16 @@
     int chan = u2p_chan(&pkt->hdrs.fixed);
 
     if (chan == CONTROL_CHAN) {                // control packets
-      DEBUG_LOG("C");
+      DEBUG_LOG("c");
       return handle_control_packet(base, len);
     }
     else {                             // data packets
-      DEBUG_LOG("D");
       return handle_data_packet(base, len);
     }
 
     // not reached
   }
 
-  bool
-  usrp2::impl::transmit_cmd(void *cmd, int len, pending_reply *p, double secs)
-  {
-    if (p)    
-      d_pending_replies[p->rid()] = p;
-    
-    // Transmit command
-    if (d_buffer->tx_frame(cmd, len) != 1) {
-      d_pending_replies[p->rid()] = 0;
-      return false;
-    }
-
-    int res = 1;
-    if (p)
-      res = p->wait(secs);
-      
-    d_pending_replies[p->rid()] = 0;
-    return res == 1;
-  }
-
   data_handler::result
   usrp2::impl::handle_control_packet(const void *base, size_t len)
   {
@@ -478,7 +453,6 @@
     
       // Copy reply into caller's buffer
       memcpy(rp->buffer(), p, std::min(oplen, buflen));
-      DEBUG_LOG("s"); // pending blocked thread condition variable signaled
       rp->signal();
       d_pending_replies[rid] = 0;
       return 0; // release packet
@@ -511,23 +485,23 @@
     }
 
     d_rx_seqno = pkt->hdrs.thdr.seqno;
-// incomplete channel ring functionality disabled
-#if 0
+
     unsigned int chan = u2p_chan(&pkt->hdrs.fixed);
     if (d_channel_rings[chan] == 0) {
-      DEBUG_LOG("?");
+      DEBUG_LOG("!");
       return 0; // discard packet, no channel handler
     }
-    
+
     if (d_channel_rings[chan]->enqueue(&pkt->samples, 0)) {
-      DEBUG_LOG("e");
+      inc_enqueued();
+      DEBUG_LOG("+");
       return KEEP; // channel ring runner will mark frame done
     }
     else {
       DEBUG_LOG("!");
       return 0; // discard, no room in channel ring
     }          
-#endif
+
     return 0;
   }
 
@@ -555,7 +529,7 @@
     
     // Wait for frames available in channel ring
     while (!rp->not_empty()) {
-      DEBUG_LOG("w");
+      DEBUG_LOG("W");
       rp->wait();
     }
     
@@ -563,11 +537,13 @@
     void *p;
     size_t len;
     while (rp->dequeue(&p, &len)) {
-      DEBUG_LOG("d");
-      // Invoke user callback, handle KEEP and DONE
 
+      // TODO: Invoke user callback, handle KEEP and DONE
+
       // Release frame
       d_buffer->release_frame(p);
+      DEBUG_LOG("-");
+      dec_enqueued();
     }
 
     return true;
@@ -577,6 +553,7 @@
   usrp2::impl::stop_bg()
   {
     d_bg_running = false;
+    d_bg_pending_cond.signal();
     
     void *dummy_status;
     d_bg_thread->join(&dummy_status);  

Modified: usrp2/branches/features/host-ng/host-ng/lib/usrp2_impl.h
===================================================================
--- usrp2/branches/features/host-ng/host-ng/lib/usrp2_impl.h    2008-07-09 
21:33:39 UTC (rev 8847)
+++ usrp2/branches/features/host-ng/host-ng/lib/usrp2_impl.h    2008-07-10 
00:16:30 UTC (rev 8848)
@@ -39,23 +39,39 @@
     static const size_t NRIDS = 256;
     static const size_t NCHANS = 32;
 
-    eth_buffer   *d_buffer;
-    pktfilter    *d_pf;
-    std::string   d_addr;
-    usrp2_thread *d_bg_thread;
-    volatile bool d_bg_running; // TODO: multistate if needed
+    eth_buffer    *d_buffer;
+    pktfilter     *d_pf;
+    std::string    d_addr;
+    usrp2_thread  *d_bg_thread;
+    volatile bool  d_bg_running; // TODO: multistate if needed
     
-    int           d_rx_decim;
-    int           d_rx_seqno;
-    int           d_tx_seqno;
-    int           d_next_rid;
-    unsigned int  d_num_rx_frames;
-    unsigned int  d_num_rx_lost;
-    unsigned int  d_num_rx_bytes;
-    
+    int            d_rx_decim;
+    int            d_rx_seqno;
+    int            d_tx_seqno;
+    int            d_next_rid;
+    unsigned int   d_num_rx_frames;
+    unsigned int   d_num_rx_lost;
+    unsigned int   d_num_rx_bytes;
+
+    unsigned int   d_num_enqueued;
+    omni_mutex     d_enqueued_mutex;
+    omni_mutex     d_bg_pending_mutex;
+    omni_condition d_bg_pending_cond;
+
     pending_reply *d_pending_replies[NRIDS]; // indexed by 8-bit reply id
     ring          *d_channel_rings[NCHANS]; // indexed by 5-bit channel number
 
+    void inc_enqueued() {
+      omni_mutex_lock l(d_enqueued_mutex);
+      d_num_enqueued++;
+    }
+    
+    void dec_enqueued() {
+      omni_mutex_lock l(d_enqueued_mutex);
+      if (--d_num_enqueued == 0)
+        d_bg_pending_cond.signal();
+    }
+    
     static bool parse_mac_addr(const std::string &s, u2_mac_addr_t *p);
     void init_et_hdrs(u2_eth_packet_t *p, const std::string &dst);
     void init_etf_hdrs(u2_eth_packet_t *p, const std::string &dst,