transport-rxqueue.hpp

Go to the documentation of this file.

#ifndef NDNPH_FACE_TRANSPORT_RXQUEUE_HPP
#define NDNPH_FACE_TRANSPORT_RXQUEUE_HPP
 
#include "../port/queue/port.hpp"
#include "transport.hpp"
 
#ifndef NDNPH_TRANSPORT_RXQUEUELEN
#define NDNPH_TRANSPORT_RXQUEUELEN 8
#endif
 
namespace ndnph {
namespace transport {
 
struct RxQueueItem {
  Region* region = nullptr;
  uint64_t endpointId = 0;
  uint8_t* pkt = nullptr;
  ssize_t pktLen = -1;
};
 
class RxQueueMixin : public virtual Transport {
protected:
  template<typename F>
  void initAllocBuffers(const F& makeRegion) {
    for (size_t i = 0; i < NDNPH_TRANSPORT_RXQUEUELEN; ++i) {
      RxQueueItem item;
      item.region = makeRegion();
      if (item.region == nullptr || !m_allocQ.push(item)) {
        break;
      }
    }
  }
 
  class RxContext {
  public:
    explicit RxContext(RxQueueMixin& transport)
      : m_transport(transport) {
      bool ok = false;
      std::tie(m_item, ok) = transport.m_allocQ.pop();
      if (ok) {
        Region& region = *m_item.region;
        region.reset();
        m_bufLen = region.available();
        m_item.pkt = region.alloc(m_bufLen);
        m_item.pktLen = -1;
      }
    }
 
    ~RxContext() {
      if (m_item.region == nullptr) {
        return;
      }
      bool ok = false;
      if (m_item.pktLen < 0) {
        ok = m_transport.m_allocQ.push(m_item);
      } else {
        ok = m_transport.m_rxQ.push(m_item);
      }
      NDNPH_ASSERT(ok);
    }
 
    operator bool() const {
      return m_item.pkt != nullptr;
    }
 
    uint8_t* buf() {
      return m_item.pkt;
    }
 
    size_t bufLen() {
      return m_bufLen;
    }
 
    void operator()(size_t pktLen, uint64_t endpointId = 0) {
      m_item.pktLen = pktLen;
      m_item.endpointId = endpointId;
    }
 
  private:
    RxQueueMixin& m_transport;
    RxQueueItem m_item;
    size_t m_bufLen;
  };
 
  RxContext receiving() {
    return RxContext(*this);
  }
 
  void loopRxQueue() {
    while (true) {
      RxQueueItem item;
      bool ok = false;
      std::tie(item, ok) = m_rxQ.pop();
      if (!ok) {
        break;
      }
      invokeRxCallback(item.pkt, item.pktLen, item.endpointId);
      m_allocQ.push(item);
    }
  }
 
private:
  port::SafeQueue<RxQueueItem, NDNPH_TRANSPORT_RXQUEUELEN> m_allocQ;
  port::SafeQueue<RxQueueItem, NDNPH_TRANSPORT_RXQUEUELEN> m_rxQ;
};
 
class DynamicRxQueueMixin : public RxQueueMixin {
public:
  static constexpr size_t DEFAULT_BUFLEN = 1500;
 
protected:
  explicit DynamicRxQueueMixin(size_t bufLen = DEFAULT_BUFLEN)
    : m_region(sizeofSubRegions(bufLen, NDNPH_TRANSPORT_RXQUEUELEN)) {
    this->initAllocBuffers([this, bufLen] { return makeSubRegion(m_region, bufLen); });
  }
 
private:
  DynamicRegion m_region;
};
 
} // namespace transport
} // namespace ndnph
 
#endif // NDNPH_FACE_TRANSPORT_RXQUEUE_HPP