RecordTimestamp: Enable the use of callbacks instead of marking packets

on-path

elements/analysis/recordtimestamp.cc

@@ -22,6 +22,11 @@
 
 #include <click/args.hh>
 #include <click/error.hh>
+#include <click/hashtable.hh>
+
+#if HAVE_DPDK
+#include "../elements/userlevel/todpdkdevice.hh"
+#endif
 
 CLICK_DECLS
 
@@ -32,13 +37,43 @@ RecordTimestamp::RecordTimestamp() :
 RecordTimestamp::~RecordTimestamp() {
 }
 
+
+inline uint16_t
+RecordTimestamp::calc_latency(uint16_t port __rte_unused, uint16_t qidx,
+                struct rte_mbuf **pkts, uint16_t nb_pkts, void *ptr)
+{
+    RecordTimestamp* rt = ((RecordTimestamp*)ptr);
+    for (unsigned i = 0; i < nb_pkts; i++) {
+
+        unsigned char *data = rte_pktmbuf_mtod(pkts[i], unsigned char *);
+        uint64_t n = (*(reinterpret_cast<const uint64_t *>(data + rt->_offset)));
+        if (rt->_sample > 1) {
+            if (n % rt->_sample == 0)
+                n = n / rt->_sample;
+            else
+                continue;
+        }
+
+    //click_chatter("Record qidx%d %p{element} -> %d",qidx, rt, n);
+        rt->_timestamps[n] = TimestampT::now_steady();
+    }
+    return nb_pkts;
+
+}
+
 int RecordTimestamp::configure(Vector<String> &conf, ErrorHandler *errh) {
     uint32_t n = 0;
     Element *e = NULL;
+#if HAVE_DPDK
+    ToDPDKDevice* _tx_dev;
+    int _tx_dev_id = 0;
+#endif
     if (Args(conf, this, errh)
             .read("COUNTER", e)
             .read("N", n)
-            .read("OFFSET", _offset)
+            .read("TXDEV", ElementCastArg("ToDPDKDevice"), _tx_dev)
+            .read("TXDEV_QID", _tx_dev_id)
+            .read_or_set("OFFSET", _offset, -1)
             .read_or_set("DYNAMIC", _dynamic, false)
             .read_or_set("NET_ORDER", _net_order, false)
             .read_or_set("SAMPLE", _sample, false)
@@ -56,6 +91,26 @@ int RecordTimestamp::configure(Vector<String> &conf, ErrorHandler *errh) {
     if (_np) {
         _net_order = _np->has_net_order();
     }
+#if HAVE_DPDK
+    if (_tx_dev) {
+        if (_dynamic || _offset < 0)
+            return errh->error("TXDEV is only compatible with OFFSET given and non-dynamic mode");
+
+        if (_net_order)
+            return errh->error("TXDEV is only compatible with non-net order");
+
+        _timestamps.resize(_timestamps.size() == 0? _timestamps.capacity():_timestamps.size() * 2, Timestamp::uninitialized_t());
+                    DPDKDevice::all_initialized.post(new Router::FctFuture([this,_tx_dev,_tx_dev_id](ErrorHandler* errh) {
+            if (rte_eth_add_tx_callback(_tx_dev->port_id(), _tx_dev_id, calc_latency, this) == 0) {
+                return errh->error("Port %d/%d Callback could not be set %d: %s", _tx_dev->port_id(), _tx_dev_id, rte_errno, rte_strerror(rte_errno));
+            }
+            return 0;
+            },this));
+    } else
+#endif
+
+    if (ninputs() != 1 or noutputs() !=1)
+        return errh->error("You need to pass either TXDEV or use the element in path.");
 
     return 0;
 }
@@ -71,8 +126,7 @@ RecordTimestamp::rmaction(Packet *p) {
             else
                 return;
         }
-        assert(i < ULLONG_MAX);
-        while (i >= (unsigned)_timestamps.size()) {
+        while (unlikely(i >= (unsigned)_timestamps.size())) {
             if (!_dynamic && i >= (unsigned)_timestamps.capacity()) {
                 click_chatter("Fatal error: DYNAMIC is not set and record timestamp reserved capacity is too small. Use N to augment the capacity.");
                 assert(false);
@@ -99,6 +153,8 @@ void RecordTimestamp::push_batch(int, PacketBatch *batch) {
 }
 #endif
 
+
 CLICK_ENDDECLS
+
 ELEMENT_REQUIRES(userlevel)
 EXPORT_ELEMENT(RecordTimestamp)

elements/analysis/recordtimestamp.hh

@@ -5,10 +5,15 @@
 #include <click/batchelement.hh>
 #include <click/timestamp.hh>
 #include <click/tsctimestamp.hh>
+#include <click/hashtable.hh>
 #include "numberpacket.hh"
 
 CLICK_DECLS
 
+#if HAVE_DPDK
+struct rte_mbuf;
+class ToDPDKDevice;
+#endif
 class NumberPacket;
 #define TimestampT TSCTimestamp
 #define TimestampUnread TSCTimestamp(1)
@@ -70,7 +75,7 @@ public:
     ~RecordTimestamp() CLICK_COLD;
 
     const char *class_name() const override { return "RecordTimestamp"; }
-    const char *port_count() const override { return PORTS_1_1; }
+    const char *port_count() const override { return "0-1/="; }
     const char *processing() const override { return PUSH; }
     const char *flow_code() const override { return "x/x"; }
 
@@ -93,11 +98,16 @@ public:
                      NumberPacket::read_number_of_packet(p, offset, net_order);
     }
 
+#if HAVE_DPDK
+    inline static uint16_t calc_latency(uint16_t port, uint16_t qidx,
+                struct rte_mbuf **pkts, uint16_t nb_pkts, void *ptr);
+#endif
 private:
     int _offset;
     bool _dynamic;
     bool _net_order;
     uint32_t _sample;
+    ToDPDKDevice* _tx_dev;
     Vector<TimestampT> _timestamps;
     NumberPacket *_np;
 };

elements/userlevel/todpdkdevice.hh

@@ -149,6 +149,10 @@ public:
 #endif
     void push(int port, Packet *p);
 
+
+    uint16_t port_id() {
+        return _dev->port_id;
+    }
 protected:
 
     inline void warn_congestion();

include/click/dpdkdevice.hh

@@ -37,6 +37,7 @@
 #include <click/args.hh>
 #include <click/etheraddress.hh>
 #include <click/timer.hh>
+#include <click/router.hh>
 
 #if RTE_VERSION < RTE_VERSION_NUM(19,8,0,0)
 #define rte_ipv4_hdr ipv4_hdr
@@ -232,6 +233,8 @@ public:
 
     static int get_port_numa_node(portid_t port_id);
 
+    static Router::ChildrenFuture all_initialized;
+
 #if HAVE_FLOW_API
     int set_mode(
         String mode, int num_pools, Vector<int> vf_vlan,

lib/dpdkdevice.cc

@@ -1283,7 +1283,7 @@ int DPDKDevice::initialize(ErrorHandler *errh)
         }
     }
 #endif
-
+    all_initialized.solve_initialize(errh);
     return 0;
 }
 
@@ -1579,6 +1579,7 @@ HashTable<portid_t, DPDKDevice*> DPDKDevice::_devs;
 struct rte_mempool** DPDKDevice::_pktmbuf_pools = 0;
 unsigned DPDKDevice::_nr_pktmbuf_pools = 0;
 bool DPDKDevice::no_more_buffer_msg_printed = false;
+Router::ChildrenFuture DPDKDevice::all_initialized;
 
 CLICK_ENDDECLS