Fix race condition that caused commands such as stop_rx_streaming to fail.
[debian/gnuradio] / usrp2 / host / lib / usrp2_impl.cc
index 332e479a59b5d56802efa7892e866fe4ed7c7734..b81b3cd61a5d8969e80e81de5eb7f57da18213e5 100644 (file)
@@ -128,9 +128,9 @@ namespace usrp2 {
   }
 
 
-  usrp2::impl::impl(const std::string &ifc, props *p)
-    : d_eth_buf(new eth_buffer()), d_pf(0), d_bg_thread(0), d_bg_running(false),
-      d_rx_seqno(-1), d_tx_seqno(0), d_next_rid(0),
+  usrp2::impl::impl(const std::string &ifc, props *p, size_t rx_bufsize)
+    : d_eth_buf(new eth_buffer(rx_bufsize)), d_interface_name(ifc), d_pf(0), d_bg_thread(0),
+      d_bg_running(false), d_rx_seqno(-1), d_tx_seqno(0), d_next_rid(0),
       d_num_rx_frames(0), d_num_rx_missing(0), d_num_rx_overruns(0), d_num_rx_bytes(0), 
       d_num_enqueued(0), d_enqueued_mutex(), d_bg_pending_cond(&d_enqueued_mutex),
       d_channel_rings(NCHANS), d_tx_interp(0), d_rx_decim(0)
@@ -138,12 +138,15 @@ namespace usrp2 {
     if (!d_eth_buf->open(ifc, htons(U2_ETHERTYPE)))
       throw std::runtime_error("Unable to register USRP2 protocol");
     
-    d_pf = pktfilter::make_ethertype_inbound(U2_ETHERTYPE, d_eth_buf->mac());
+    d_addr = p->addr;
+
+    // Create a packet filter for U2_ETHERTYPE packets sourced from target USRP2
+    u2_mac_addr_t usrp_mac;
+    parse_mac_addr(d_addr, &usrp_mac);
+    d_pf = pktfilter::make_ethertype_inbound_target(U2_ETHERTYPE, (const unsigned char*)&(usrp_mac.addr));
     if (!d_pf || !d_eth_buf->attach_pktfilter(d_pf))
       throw std::runtime_error("Unable to attach packet filter.");
     
-    d_addr = p->addr;
-    
     if (USRP2_IMPL_DEBUG)
       std::cerr << "usrp2 constructor: using USRP2 at " << d_addr << std::endl;
 
@@ -152,6 +155,10 @@ namespace usrp2 {
     d_bg_thread = new usrp2_thread(this);
     d_bg_thread->start();
 
+    // In case the USRP2 was left streaming RX
+    // FIXME: only one channel right now
+    stop_rx_streaming(0);
+
     if (!dboard_info())                // we're hosed
       throw std::runtime_error("Unable to retrieve daughterboard info");
 
@@ -291,22 +298,24 @@ namespace usrp2 {
     cmd->eop.len = sizeof(cmd->eop);
   }
 
+
   bool
-  usrp2::impl::transmit_cmd(void *cmd, size_t len, pending_reply *p, double secs)
+  usrp2::impl::transmit_cmd(void *cmd, size_t len)
   {
-    if (p)    
-      d_pending_replies[p->rid()] = p;
+    return d_eth_buf->tx_frame(cmd, len) == eth_buffer::EB_OK;
+  }
+
+  bool
+  usrp2::impl::transmit_cmd_and_wait(void *cmd, size_t len, pending_reply *p, double secs)
+  {
+    d_pending_replies[p->rid()] = p;
     
-    // Transmit command
-    if (d_eth_buf->tx_frame(cmd, len) != eth_buffer::EB_OK) {
+    if (!transmit_cmd(cmd, len)){
       d_pending_replies[p->rid()] = 0;
       return false;
     }
 
-    int res = 1;
-    if (p)
-      res = p->wait(secs);
-      
+    int res = p->wait_for_completion(secs);
     d_pending_replies[p->rid()] = 0;
     return res == 1;
   }
@@ -397,7 +406,7 @@ namespace usrp2 {
     
       // Copy reply into caller's buffer
       memcpy(rp->buffer(), p, std::min(oplen, buflen));
-      rp->signal();
+      rp->notify_completion();
       d_pending_replies[rid] = 0;
       return data_handler::RELEASE;
     }
@@ -438,24 +447,28 @@ namespace usrp2 {
     // FIXME unaligned load!
     unsigned int chan = u2p_chan(&pkt->hdrs.fixed);
 
-    if (!d_channel_rings[chan]) {
-      DEBUG_LOG("!");
-      return data_handler::RELEASE;    // discard packet, no channel handler
-    }
-
-    // Strip off ethernet header and transport header and enqueue the rest
+    {
+      omni_mutex_lock l(d_channel_rings_mutex);
 
-    size_t offset = offsetof(u2_eth_samples_t, hdrs.fixed);
-    if (d_channel_rings[chan]->enqueue(&pkt->hdrs.fixed, len-offset)) {
-      inc_enqueued();
-      DEBUG_LOG("+");
-      return data_handler::KEEP;       // channel ring runner will mark frame done
+      if (!d_channel_rings[chan]) {
+       DEBUG_LOG("!");
+       return data_handler::RELEASE;   // discard packet, no channel handler
+      }
+      
+      // 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)) {
+       inc_enqueued();
+       DEBUG_LOG("+");
+       return data_handler::KEEP;      // channel ring runner will mark frame done
+      }
+      else {
+       DEBUG_LOG("!");
+       return data_handler::RELEASE;   // discard, no room in channel ring
+      }        
+      return data_handler::RELEASE;
     }
-    else {
-      DEBUG_LOG("!");
-      return data_handler::RELEASE;    // discard, no room in channel ring
-    }          
-    return data_handler::RELEASE;
   }
 
 
@@ -474,7 +487,7 @@ namespace usrp2 {
     cmd.op.gain = htons(u2_double_to_fxpt_gain(gain));
     
     pending_reply p(cmd.op.rid, &reply, sizeof(reply));
-    if (!transmit_cmd(&cmd, sizeof(cmd), &p, DEF_CMD_TIMEOUT))
+    if (!transmit_cmd_and_wait(&cmd, sizeof(cmd), &p, DEF_CMD_TIMEOUT))
       return false;
 
     bool success = (ntohx(reply.ok) == 1);
@@ -501,7 +514,7 @@ namespace usrp2 {
     cmd.eop.len = sizeof(cmd.eop);
     
     pending_reply p(cmd.op.rid, &reply, sizeof(reply));
-    if (!transmit_cmd(&cmd, sizeof(cmd), &p, DEF_CMD_TIMEOUT))
+    if (!transmit_cmd_and_wait(&cmd, sizeof(cmd), &p, DEF_CMD_TIMEOUT))
       return false;
 
     bool success = (ntohx(reply.ok) == 1);
@@ -521,7 +534,7 @@ namespace usrp2 {
     cmd.op.freq_lo = htonl(u2_fxpt_freq_lo(fxpt));
     
     pending_reply p(cmd.op.rid, &reply, sizeof(reply));
-    if (!transmit_cmd(&cmd, sizeof(cmd), &p, DEF_CMD_TIMEOUT))
+    if (!transmit_cmd_and_wait(&cmd, sizeof(cmd), &p, DEF_CMD_TIMEOUT))
       return false;
 
     bool success = (ntohx(reply.ok) == 1);
@@ -558,7 +571,7 @@ namespace usrp2 {
     cmd.op.decim = htonl(decimation_factor);
     
     pending_reply p(cmd.op.rid, &reply, sizeof(reply));
-    if (!transmit_cmd(&cmd, sizeof(cmd), &p, DEF_CMD_TIMEOUT))
+    if (!transmit_cmd_and_wait(&cmd, sizeof(cmd), &p, DEF_CMD_TIMEOUT))
       return false;
 
     bool success = (ntohx(reply.ok) == 1);
@@ -578,7 +591,7 @@ namespace usrp2 {
     cmd.op.scale_iq = htonl(((scale_i & 0xffff) << 16) | (scale_q & 0xffff));
     
     pending_reply p(cmd.op.rid, &reply, sizeof(reply));
-    if (!transmit_cmd(&cmd, sizeof(cmd), &p, DEF_CMD_TIMEOUT))
+    if (!transmit_cmd_and_wait(&cmd, sizeof(cmd), &p, DEF_CMD_TIMEOUT))
       return false;
 
     bool success = (ntohx(reply.ok) == 1);
@@ -600,35 +613,40 @@ namespace usrp2 {
       return false;
     }
 
-    if (d_channel_rings[channel]) {
-      std::cerr << "usrp2: channel " << channel
-               << " already streaming" << std::endl;
-      return false;
-    }
-
-    d_channel_rings[channel] = ring_sptr(new ring(d_eth_buf->max_frames()));
-
-    if (items_per_frame == 0)
-      items_per_frame = U2_MAX_SAMPLES;                // minimize overhead
-    
-    op_start_rx_streaming_cmd cmd;
-    op_generic_t reply;
-
-    memset(&cmd, 0, sizeof(cmd));
-    init_etf_hdrs(&cmd.h, d_addr, 0, CONTROL_CHAN, -1);
-    cmd.op.opcode = OP_START_RX_STREAMING;
-    cmd.op.len = sizeof(cmd.op);
-    cmd.op.rid = d_next_rid++;
-    cmd.op.items_per_frame = htonl(items_per_frame);
-    cmd.eop.opcode = OP_EOP;
-    cmd.eop.len = sizeof(cmd.eop);
+    {
+      omni_mutex_lock l(d_channel_rings_mutex);
+      if (d_channel_rings[channel]) {
+       std::cerr << "usrp2: channel " << channel
+                 << " already streaming" << std::endl;
+       return false;
+      }
+      
+      if (items_per_frame == 0)
+       items_per_frame = U2_MAX_SAMPLES;               // minimize overhead
+      
+      op_start_rx_streaming_cmd cmd;
+      op_generic_t reply;
+
+      memset(&cmd, 0, sizeof(cmd));
+      init_etf_hdrs(&cmd.h, d_addr, 0, CONTROL_CHAN, -1);
+      cmd.op.opcode = OP_START_RX_STREAMING;
+      cmd.op.len = sizeof(cmd.op);
+      cmd.op.rid = d_next_rid++;
+      cmd.op.items_per_frame = htonl(items_per_frame);
+      cmd.eop.opcode = OP_EOP;
+      cmd.eop.len = sizeof(cmd.eop);
     
-    pending_reply p(cmd.op.rid, &reply, sizeof(reply));
-    if (!transmit_cmd(&cmd, sizeof(cmd), &p, DEF_CMD_TIMEOUT))
-      return false;
+      bool success = false;
+      pending_reply p(cmd.op.rid, &reply, sizeof(reply));
+      success = transmit_cmd_and_wait(&cmd, sizeof(cmd), &p, DEF_CMD_TIMEOUT);
+      success = success && (ntohx(reply.ok) == 1);
+      
+      if (success)
+       d_channel_rings[channel] = ring_sptr(new ring(d_eth_buf->max_frames()));
 
-    bool success = (ntohx(reply.ok) == 1);
-    return success;
+      //fprintf(stderr, "usrp2::start_rx_streaming: success = %d\n", success);
+      return success;
+    }
   }
   
   bool
@@ -646,36 +664,29 @@ namespace usrp2 {
       return false;
     }
 
-#if 0 // don't be overzealous.    
-    if (!d_channel_rings[channel]) {
-      std::cerr << "usrp2: channel " << channel
-               << " not streaming" << std::endl;
-      return false;
-    }
-#endif
-
     op_stop_rx_cmd cmd;
     op_generic_t reply;
 
-    memset(&cmd, 0, sizeof(cmd));
-    init_etf_hdrs(&cmd.h, d_addr, 0, CONTROL_CHAN, -1);
-    cmd.op.opcode = OP_STOP_RX;
-    cmd.op.len = sizeof(cmd.op);
-    cmd.op.rid = d_next_rid++;
-    cmd.eop.opcode = OP_EOP;
-    cmd.eop.len = sizeof(cmd.eop);
-    
-    pending_reply p(cmd.op.rid, &reply, sizeof(reply));
-    if (!transmit_cmd(&cmd, sizeof(cmd), &p, DEF_CMD_TIMEOUT))
-      return false;
+    {
+      omni_mutex_lock l(d_channel_rings_mutex);
 
-    bool success = (ntohx(reply.ok) == 1);
-    if (success)
+      memset(&cmd, 0, sizeof(cmd));
+      init_etf_hdrs(&cmd.h, d_addr, 0, CONTROL_CHAN, -1);
+      cmd.op.opcode = OP_STOP_RX;
+      cmd.op.len = sizeof(cmd.op);
+      cmd.op.rid = d_next_rid++;
+      cmd.eop.opcode = OP_EOP;
+      cmd.eop.len = sizeof(cmd.eop);
+    
+      bool success = false;
+      pending_reply p(cmd.op.rid, &reply, sizeof(reply));
+      success = transmit_cmd_and_wait(&cmd, sizeof(cmd), &p, 10 * DEF_CMD_TIMEOUT);
+      success = success && (ntohx(reply.ok) == 1);
       d_channel_rings[channel].reset();
-
-    return success;
+      //fprintf(stderr, "usrp2::stop_rx_streaming:  success = %d\n", success);
+      return success;
+    }
   }
-  
 
   bool
   usrp2::impl::rx_samples(unsigned int channel, rx_sample_handler *handler)
@@ -740,7 +751,7 @@ namespace usrp2 {
     cmd.op.gain = htons(u2_double_to_fxpt_gain(gain));
     
     pending_reply p(cmd.op.rid, &reply, sizeof(reply));
-    if (!transmit_cmd(&cmd, sizeof(cmd), &p, DEF_CMD_TIMEOUT))
+    if (!transmit_cmd_and_wait(&cmd, sizeof(cmd), &p, DEF_CMD_TIMEOUT))
       return false;
 
     bool success = (ntohx(reply.ok) == 1);
@@ -767,7 +778,7 @@ namespace usrp2 {
     cmd.eop.len = sizeof(cmd.eop);
     
     pending_reply p(cmd.op.rid, &reply, sizeof(reply));
-    if (!transmit_cmd(&cmd, sizeof(cmd), &p, DEF_CMD_TIMEOUT))
+    if (!transmit_cmd_and_wait(&cmd, sizeof(cmd), &p, DEF_CMD_TIMEOUT))
       return false;
 
     bool success = (ntohx(reply.ok) == 1);
@@ -787,7 +798,7 @@ namespace usrp2 {
     cmd.op.freq_lo = htonl(u2_fxpt_freq_lo(fxpt));
     
     pending_reply p(cmd.op.rid, &reply, sizeof(reply));
-    if (!transmit_cmd(&cmd, sizeof(cmd), &p, DEF_CMD_TIMEOUT))
+    if (!transmit_cmd_and_wait(&cmd, sizeof(cmd), &p, DEF_CMD_TIMEOUT))
       return false;
 
     bool success = (ntohx(reply.ok) == 1);
@@ -824,7 +835,7 @@ namespace usrp2 {
     cmd.op.interp = htonl(interpolation_factor);
     
     pending_reply p(cmd.op.rid, &reply, sizeof(reply));
-    if (!transmit_cmd(&cmd, sizeof(cmd), &p, DEF_CMD_TIMEOUT))
+    if (!transmit_cmd_and_wait(&cmd, sizeof(cmd), &p, DEF_CMD_TIMEOUT))
       return false;
 
     bool success = (ntohx(reply.ok) == 1);
@@ -875,7 +886,7 @@ namespace usrp2 {
     cmd.op.scale_iq = htonl(((scale_i & 0xffff) << 16) | (scale_q & 0xffff));
     
     pending_reply p(cmd.op.rid, &reply, sizeof(reply));
-    if (!transmit_cmd(&cmd, sizeof(cmd), &p, DEF_CMD_TIMEOUT))
+    if (!transmit_cmd_and_wait(&cmd, sizeof(cmd), &p, DEF_CMD_TIMEOUT))
       return false;
 
     bool success = (ntohx(reply.ok) == 1);
@@ -925,10 +936,8 @@ namespace usrp2 {
     if (nitems == 0)
       return true;
 
-    // FIXME there's the possibility that we send fewer than 9 items in a frame.
-    // That would end up glitching the transmitter, since the ethernet will pad to
-    // 64-bytes total (9 items).  We really need some part of the stack to
-    // carry the real length (thdr?).
+    // FIXME can't deal with nitems < U2_MIN_SAMPLES (will be fixed in VRT)
+    // FIXME need to check the MTU instead of assuming 1500 bytes
 
     // fragment as necessary then fire away
 
@@ -958,7 +967,12 @@ namespace usrp2 {
 
       init_etf_hdrs(&hdrs, d_addr, flags, channel, timestamp);
 
-      size_t i = std::min((size_t) U2_MAX_SAMPLES, nitems - n);
+      // Avoid short packet by splitting last two packets if reqd
+      size_t i;
+      if ((nitems - n) > U2_MAX_SAMPLES && (nitems - n) < (U2_MAX_SAMPLES + U2_MIN_SAMPLES))
+       i = (nitems - n) / 2;
+      else
+       i = std::min((size_t) U2_MAX_SAMPLES, nitems - n);
 
       eth_iovec iov[2];
       iov[0].iov_base = &hdrs;
@@ -1000,7 +1014,7 @@ namespace usrp2 {
     cmd.eop.len = sizeof(cmd.eop);
     
     pending_reply p(cmd.op.rid, &reply, sizeof(reply));
-    if (!transmit_cmd(&cmd, sizeof(cmd), &p, DEF_CMD_TIMEOUT))
+    if (!transmit_cmd_and_wait(&cmd, sizeof(cmd), &p, DEF_CMD_TIMEOUT))
       return false;
 
     return ntohx(reply.ok) == 1;
@@ -1059,7 +1073,7 @@ namespace usrp2 {
       return false;
 
     pending_reply p(cmd.op.rid, &reply, sizeof(reply));
-    if (!transmit_cmd(&cmd, sizeof(cmd), &p, 4*DEF_CMD_TIMEOUT))
+    if (!transmit_cmd_and_wait(&cmd, sizeof(cmd), &p, 4*DEF_CMD_TIMEOUT))
       return false;
 
     bool success = (ntohx(reply.ok) == 1);
@@ -1098,7 +1112,7 @@ namespace usrp2 {
     cmd.eop.len = sizeof(cmd.eop);
     
     pending_reply p(cmd.op.rid, &reply, sizeof(reply));
-    if (!transmit_cmd(&cmd, sizeof(cmd), &p, DEF_CMD_TIMEOUT))
+    if (!transmit_cmd_and_wait(&cmd, sizeof(cmd), &p, DEF_CMD_TIMEOUT))
       return false;
 
     bool success = (ntohx(reply.ok) == 1);
@@ -1125,7 +1139,7 @@ namespace usrp2 {
     cmd.eop.len = sizeof(cmd.eop);
     
     pending_reply p(cmd.op.rid, &reply, sizeof(reply));
-    if (!transmit_cmd(&cmd, sizeof(cmd), &p, DEF_CMD_TIMEOUT))
+    if (!transmit_cmd_and_wait(&cmd, sizeof(cmd), &p, DEF_CMD_TIMEOUT))
       return false;
 
     return ntohx(reply.ok) == 1;
@@ -1147,7 +1161,7 @@ namespace usrp2 {
     cmd.eop.len = sizeof(cmd.eop);
     
     pending_reply p(cmd.op.rid, &reply, sizeof(reply));
-    if (!transmit_cmd(&cmd, sizeof(cmd), &p, DEF_CMD_TIMEOUT))
+    if (!transmit_cmd_and_wait(&cmd, sizeof(cmd), &p, DEF_CMD_TIMEOUT))
       return false;
 
     return ntohx(reply.ok) == 1;
@@ -1187,7 +1201,7 @@ namespace usrp2 {
 
     reply = (op_generic_t *)malloc(rlen+bytes);
     pending_reply p(cmd.op.rid, reply, rlen+bytes);
-    if (transmit_cmd(&cmd, sizeof(cmd), &p, DEF_CMD_TIMEOUT)) {
+    if (transmit_cmd_and_wait(&cmd, sizeof(cmd), &p, DEF_CMD_TIMEOUT)) {
       uint32_t nwords = (reply->len-rlen)/sizeof(uint32_t);
       uint32_t *data = (uint32_t *)(reply+rlen/wlen);
       for (unsigned int i = 0; i < nwords; i++)
@@ -1254,7 +1268,7 @@ namespace usrp2 {
     bool ok = false;
     op_generic_t reply;
     pending_reply p(cmd->op.rid, &reply, sizeof(reply));
-    if (transmit_cmd(cmd, l, &p, DEF_CMD_TIMEOUT))
+    if (transmit_cmd_and_wait(cmd, l, &p, DEF_CMD_TIMEOUT))
       ok = (ntohx(reply.ok) == 1);
 
     free(cmd);
@@ -1276,7 +1290,164 @@ namespace usrp2 {
     cmd.eop.len = sizeof(cmd.eop);
     
     pending_reply p(cmd.op.rid, &reply, sizeof(reply));
-    if (!transmit_cmd(&cmd, sizeof(cmd), &p, DEF_CMD_TIMEOUT))
+    if (!transmit_cmd_and_wait(&cmd, sizeof(cmd), &p, DEF_CMD_TIMEOUT))
+      return false;
+
+    bool success = (ntohx(reply.ok) == 1);
+    return success;
+  }
+
+  bool usrp2::impl::set_gpio_ddr(int bank, uint16_t value, uint16_t mask)
+  {
+    if (bank != GPIO_TX_BANK && bank != GPIO_RX_BANK) {
+      fprintf(stderr, "set_gpio_ddr: bank must be one of GPIO_RX_BANK or GPIO_TX_BANK\n");
+      return false;
+    }
+
+    op_gpio_cmd cmd;
+    op_generic_t reply;
+
+    memset(&cmd, 0, sizeof(cmd));
+    init_etf_hdrs(&cmd.h, d_addr, 0, CONTROL_CHAN, -1);
+    cmd.op.opcode = OP_GPIO_SET_DDR;
+    cmd.op.len = sizeof(cmd.op);
+    cmd.op.rid = d_next_rid++;
+    cmd.op.bank = static_cast<uint8_t>(bank);
+    cmd.op.value = htons(value);
+    cmd.op.mask = htons(mask);
+    cmd.eop.opcode = OP_EOP;
+    cmd.eop.len = sizeof(cmd.eop);
+    
+    pending_reply p(cmd.op.rid, &reply, sizeof(reply));
+    if (!transmit_cmd_and_wait(&cmd, sizeof(cmd), &p, DEF_CMD_TIMEOUT))
+      return false;
+
+    bool success = (ntohx(reply.ok) == 1);
+    return success;
+  }
+
+  bool usrp2::impl::set_gpio_sels(int bank, std::string sels)
+  {
+    if (bank != GPIO_TX_BANK && bank != GPIO_RX_BANK) {
+      fprintf(stderr, "set_gpio_ddr: bank must be one of GPIO_RX_BANK or GPIO_TX_BANK\n");
+      return false;
+    }
+
+    if (sels.size() != 16) {
+      fprintf(stderr, "set_gpio_sels: sels must be exactly 16 bytes\n");
+      return false;
+    }
+
+    op_gpio_set_sels_cmd cmd;
+    op_generic_t reply;
+
+    memset(&cmd, 0, sizeof(cmd));
+    init_etf_hdrs(&cmd.h, d_addr, 0, CONTROL_CHAN, -1);
+    cmd.op.opcode = OP_GPIO_SET_SELS;
+    cmd.op.len = sizeof(cmd.op);
+    cmd.op.rid = d_next_rid++;
+    cmd.op.bank = static_cast<uint8_t>(bank);
+    memcpy(&cmd.op.sels, sels.c_str(), 16);
+    cmd.eop.opcode = OP_EOP;
+    cmd.eop.len = sizeof(cmd.eop);
+    
+    pending_reply p(cmd.op.rid, &reply, sizeof(reply));
+    if (!transmit_cmd_and_wait(&cmd, sizeof(cmd), &p, DEF_CMD_TIMEOUT))
+      return false;
+
+    bool success = (ntohx(reply.ok) == 1);
+    return success;
+  }
+
+  bool usrp2::impl::write_gpio(int bank, uint16_t value, uint16_t mask)
+  {
+    if (bank != GPIO_TX_BANK && bank != GPIO_RX_BANK) {
+      fprintf(stderr, "set_gpio_ddr: bank must be one of GPIO_RX_BANK or GPIO_TX_BANK\n");
+      return false;
+    }
+
+    op_gpio_cmd cmd;
+    op_generic_t reply;
+
+    memset(&cmd, 0, sizeof(cmd));
+    init_etf_hdrs(&cmd.h, d_addr, 0, CONTROL_CHAN, -1);
+    cmd.op.opcode = OP_GPIO_WRITE;
+    cmd.op.len = sizeof(cmd.op);
+    cmd.op.rid = d_next_rid++;
+    cmd.op.bank = static_cast<uint8_t>(bank);
+    cmd.op.value = htons(value);
+    cmd.op.mask = htons(mask);
+    cmd.eop.opcode = OP_EOP;
+    cmd.eop.len = sizeof(cmd.eop);
+    
+    pending_reply p(cmd.op.rid, &reply, sizeof(reply));
+    if (!transmit_cmd_and_wait(&cmd, sizeof(cmd), &p, DEF_CMD_TIMEOUT))
+      return false;
+
+    bool success = (ntohx(reply.ok) == 1);
+    return success;
+  }
+
+  bool usrp2::impl::read_gpio(int bank, uint16_t *value)
+  {
+    if (bank != GPIO_TX_BANK && bank != GPIO_RX_BANK) {
+      fprintf(stderr, "set_gpio_ddr: bank must be one of GPIO_RX_BANK or GPIO_TX_BANK\n");
+      return false;
+    }
+
+    op_gpio_cmd cmd;
+    op_gpio_read_reply_t reply;
+
+    memset(&cmd, 0, sizeof(cmd));
+    init_etf_hdrs(&cmd.h, d_addr, 0, CONTROL_CHAN, -1);
+    cmd.op.opcode = OP_GPIO_READ;
+    cmd.op.len = sizeof(cmd.op);
+    cmd.op.rid = d_next_rid++;
+    cmd.op.bank = static_cast<uint8_t>(bank);
+    cmd.op.value = 0; // not used
+    cmd.op.mask = 0;  // not used
+    cmd.eop.opcode = OP_EOP;
+    cmd.eop.len = sizeof(cmd.eop);
+    
+    pending_reply p(cmd.op.rid, &reply, sizeof(reply));
+    if (!transmit_cmd_and_wait(&cmd, sizeof(cmd), &p, DEF_CMD_TIMEOUT))
+      return false;
+
+    bool success = (ntohx(reply.ok) == 1);
+    if (success && (value != NULL))
+      *value = ntohs(reply.value);
+
+    return success;
+  }
+
+  bool usrp2::impl::enable_gpio_streaming(int bank, int enable)
+  {
+    if (bank != GPIO_RX_BANK) {
+      fprintf(stderr, "enable_gpio_streaming: only RX streaming is currently implemented\n");
+      return false;
+    }
+
+    if ((enable & ~0x03) != 0) {
+      fprintf(stderr, "enable_gpio_streaming: invalid enable format\n");
+      return false;
+    }
+
+    op_gpio_cmd cmd;
+    op_generic_t reply;
+
+    memset(&cmd, 0, sizeof(cmd));
+    init_etf_hdrs(&cmd.h, d_addr, 0, CONTROL_CHAN, -1);
+    cmd.op.opcode = OP_GPIO_STREAM;
+    cmd.op.len = sizeof(cmd.op);
+    cmd.op.rid = d_next_rid++;
+    cmd.op.bank = static_cast<uint8_t>(bank);
+    cmd.op.value = htons((uint16_t)enable);
+    cmd.op.mask = 0;  // not used
+    cmd.eop.opcode = OP_EOP;
+    cmd.eop.len = sizeof(cmd.eop);
+    
+    pending_reply p(cmd.op.rid, &reply, sizeof(reply));
+    if (!transmit_cmd_and_wait(&cmd, sizeof(cmd), &p, DEF_CMD_TIMEOUT))
       return false;
 
     bool success = (ntohx(reply.ok) == 1);