FastUDPSrc and friends: Don't send the same packet over & over.

Send clones. Linked lists don't work well with the same packet over & over. I hate these abortions of elements. Performance hackers suck.
tbarbette · Feb 26, 2010 · f380887 · f380887
1 parent 5f309ac
commit f380887
Show file tree

Hide file tree

Showing 8 changed files with 59 additions and 59 deletions.
diff --git a/elements/linuxmodule/fasttcpflows.cc b/elements/linuxmodule/fasttcpflows.cc
@@ -84,8 +84,10 @@ FastTCPFlows::change_ports(int flow)
 {
   unsigned short sport = (click_random() >> 2) % 0xFFFF;
   unsigned short dport = (click_random() >> 2) % 0xFFFF;
+  WritablePacket *q = _flows[flow].syn_packet->uniqueify(); // better not fail
+  _flows[flow].syn_packet = q;
   click_ip *ip =
-    reinterpret_cast<click_ip *>(_flows[flow].syn_packet->data()+14);
+    reinterpret_cast<click_ip *>(q->data()+14);
   click_tcp *tcp = reinterpret_cast<click_tcp *>(ip + 1);
   tcp->th_sport = sport;
   tcp->th_dport = dport;
@@ -95,7 +97,9 @@ FastTCPFlows::change_ports(int flow)
   tcp->th_sum = csum_tcpudp_magic
     (_sipaddr.s_addr, _dipaddr.s_addr, len, IP_PROTO_TCP, csum);
 
-  ip = reinterpret_cast<click_ip *>(_flows[flow].data_packet->data()+14);
+  q = _flows[flow].data_packet->uniqueify(); // better not fail
+  _flows[flow].data_packet = q;
+  ip = reinterpret_cast<click_ip *>(q->data()+14);
   tcp = reinterpret_cast<click_tcp *>(ip + 1);
   tcp->th_sport = sport;
   tcp->th_dport = dport;
@@ -105,7 +109,9 @@ FastTCPFlows::change_ports(int flow)
   tcp->th_sum = csum_tcpudp_magic
     (_sipaddr.s_addr, _dipaddr.s_addr, len, IP_PROTO_TCP, csum);
 
-  ip = reinterpret_cast<click_ip *>(_flows[flow].fin_packet->data()+14);
+  q = _flows[flow].fin_packet->uniqueify(); // better not fail
+  _flows[flow].fin_packet = q;
+  ip = reinterpret_cast<click_ip *>(q->data()+14);
   tcp = reinterpret_cast<click_tcp *>(ip + 1);
   tcp->th_sport = sport;
   tcp->th_dport = dport;
@@ -123,8 +129,7 @@ FastTCPFlows::get_packet()
     for (unsigned i=0; i<_nflows; i++) {
       if (_flows[i].flow_count != _flowsize) {
 	_flows[i].flow_count = _flowsize;
-        atomic_inc(&(_flows[i].fin_packet->skb())->users);
-        return reinterpret_cast<Packet *>(_flows[i].fin_packet->skb());
+	return _flows[i].fin_packet->clone();
       }
     }
     _sent_all_fins = true;
@@ -138,16 +143,11 @@ FastTCPFlows::get_packet()
     }
     _flows[flow].flow_count++;
     if (_flows[flow].flow_count == 1) {
-      atomic_inc(&(_flows[flow].syn_packet->skb())->users);
-      return reinterpret_cast<Packet *>(_flows[flow].syn_packet->skb());
-    }
-    else if (_flows[flow].flow_count == _flowsize) {
-      atomic_inc(&(_flows[flow].fin_packet->skb())->users);
-      return reinterpret_cast<Packet *>(_flows[flow].fin_packet->skb());
-    }
-    else {
-      atomic_inc(&(_flows[flow].data_packet->skb())->users);
-      return reinterpret_cast<Packet *>(_flows[flow].data_packet->skb());
+      return _flows[flow].syn_packet->clone();
+    } else if (_flows[flow].flow_count == _flowsize) {
+      return _flows[flow].fin_packet->clone();
+    } else {
+      return _flows[flow].data_packet->clone();
     }
   }
 }
@@ -165,10 +165,11 @@ FastTCPFlows::initialize(ErrorHandler *)
     unsigned short dport = (click_random() >> 2) % 0xFFFF;
 
     // SYN packet
-    _flows[i].syn_packet = Packet::make(_len);
+    WritablePacket *q = Packet::make(_len);
+    _flows[i].syn_packet = q;
     memcpy(_flows[i].syn_packet->data(), &_ethh, 14);
     click_ip *ip =
-      reinterpret_cast<click_ip *>(_flows[i].syn_packet->data()+14);
+      reinterpret_cast<click_ip *>(q->data()+14);
     click_tcp *tcp = reinterpret_cast<click_tcp *>(ip + 1);
     // set up IP header
     ip->ip_v = 4;
@@ -201,9 +202,10 @@ FastTCPFlows::initialize(ErrorHandler *)
 				    len, IP_PROTO_TCP, csum);
 
     // DATA packet with PUSH and ACK
-    _flows[i].data_packet = Packet::make(_len);
-    memcpy(_flows[i].data_packet->data(), &_ethh, 14);
-    ip = reinterpret_cast<click_ip *>(_flows[i].data_packet->data()+14);
+    q = Packet::make(_len);
+    _flows[i].data_packet = q;
+    memcpy(q->data(), &_ethh, 14);
+    ip = reinterpret_cast<click_ip *>(q->data()+14);
     tcp = reinterpret_cast<click_tcp *>(ip + 1);
     // set up IP header
     ip->ip_v = 4;
@@ -236,9 +238,10 @@ FastTCPFlows::initialize(ErrorHandler *)
 				    len, IP_PROTO_TCP, csum);
 
     // FIN packet
-    _flows[i].fin_packet = Packet::make(_len);
-    memcpy(_flows[i].fin_packet->data(), &_ethh, 14);
-    ip = reinterpret_cast<click_ip *>(_flows[i].fin_packet->data()+14);
+    q = Packet::make(_len);
+    _flows[i].fin_packet = q;
+    memcpy(q->data(), &_ethh, 14);
+    ip = reinterpret_cast<click_ip *>(q->data()+14);
     tcp = reinterpret_cast<click_tcp *>(ip + 1);
     // set up IP header
     ip->ip_v = 4;

diff --git a/elements/linuxmodule/fasttcpflows.hh b/elements/linuxmodule/fasttcpflows.hh
@@ -73,9 +73,9 @@ class FastTCPFlows : public Element {
   click_jiffies_t _last;
 
   struct flow_t {
-    WritablePacket *syn_packet;
-    WritablePacket *fin_packet;
-    WritablePacket *data_packet;
+    Packet *syn_packet;
+    Packet *fin_packet;
+    Packet *data_packet;
     int flow_count;
   };
   flow_t *_flows;

diff --git a/elements/linuxmodule/fastudpflows.cc b/elements/linuxmodule/fastudpflows.cc
@@ -79,7 +79,9 @@ FastUDPFlows::configure(Vector<String> &conf, ErrorHandler *errh)
 void
 FastUDPFlows::change_ports(int flow)
 {
-  click_ip *ip = reinterpret_cast<click_ip *>(_flows[flow].packet->data()+14);
+  WritablePacket *q = _flows[flow].packet->uniqueify(); // better not fail
+  _flows[flow].packet = q;
+  click_ip *ip = reinterpret_cast<click_ip *>(q->data()+14);
   click_udp *udp = reinterpret_cast<click_udp *>(ip + 1);
 
   udp->uh_sport = (click_random() >> 2) % 0xFFFF;
@@ -110,8 +112,7 @@ FastUDPFlows::get_packet()
     _flows[flow].flow_count = 0;
   }
   _flows[flow].flow_count++;
-  atomic_inc(&(_flows[flow].skb)->users);
-  return reinterpret_cast<Packet *>(_flows[flow].skb);
+  return _flows[flow].packet->clone();
 }
 
 
@@ -122,9 +123,10 @@ FastUDPFlows::initialize(ErrorHandler *)
   _flows = new flow_t[_nflows];
 
   for (int i=0; i<_nflows; i++) {
-    _flows[i].packet = Packet::make(_len);
-    memcpy(_flows[i].packet->data(), &_ethh, 14);
-    click_ip *ip = reinterpret_cast<click_ip *>(_flows[i].packet->data()+14);
+    WritablePacket *q = Packet::make(_len);
+    _flows[i].packet = q;
+    memcpy(q->data(), &_ethh, 14);
+    click_ip *ip = reinterpret_cast<click_ip *>(q->data()+14);
     click_udp *udp = reinterpret_cast<click_udp *>(ip + 1);
 
     // set up IP header
@@ -155,7 +157,6 @@ FastUDPFlows::initialize(ErrorHandler *)
 				      len, IP_PROTO_UDP, csum);
     } else
       udp->uh_sum = 0;
-    _flows[i].skb = _flows[i].packet->skb();
     _flows[i].flow_count = 0;
   }
   _last_flow = 0;

diff --git a/elements/linuxmodule/fastudpflows.hh b/elements/linuxmodule/fastudpflows.hh
@@ -70,9 +70,8 @@ class FastUDPFlows : public Element {
   click_jiffies_t _last;
 
   struct flow_t {
-    WritablePacket *packet;
-    struct sk_buff *skb;
-    int flow_count;
+      Packet *packet;
+      int flow_count;
   };
   flow_t *_flows;
   void change_ports(int);

diff --git a/elements/linuxmodule/fastudpsrc.cc b/elements/linuxmodule/fastudpsrc.cc
@@ -80,7 +80,9 @@ FastUDPSource::configure(Vector<String> &conf, ErrorHandler *errh)
 void
 FastUDPSource::incr_ports()
 {
-  click_ip *ip = reinterpret_cast<click_ip *>(_packet->data()+14);
+  WritablePacket *q = _packet->uniqueify(); // better not fail
+  _packet = q;
+  click_ip *ip = reinterpret_cast<click_ip *>(q->data()+14);
   click_udp *udp = reinterpret_cast<click_udp *>(ip + 1);
   _incr++;
   udp->uh_sport = htons(_sport+_incr);
@@ -100,9 +102,10 @@ FastUDPSource::initialize(ErrorHandler *)
 {
   _count = 0;
   _incr = 0;
-  _packet = Packet::make(_len);
-  memcpy(_packet->data(), &_ethh, 14);
-  click_ip *ip = reinterpret_cast<click_ip *>(_packet->data()+14);
+  WritablePacket *q = Packet::make(_len);
+  _packet = q;
+  memcpy(q->data(), &_ethh, 14);
+  click_ip *ip = reinterpret_cast<click_ip *>(q->data()+14);
   click_udp *udp = reinterpret_cast<click_udp *>(ip + 1);
 
   // set up IP header
@@ -134,7 +137,6 @@ FastUDPSource::initialize(ErrorHandler *)
   } else
     udp->uh_sum = 0;
 
-  _skb = _packet->skb();
   return 0;
 }
 
@@ -157,16 +159,13 @@ FastUDPSource::pull(int)
   if(_rate_limited){
     if (_rate.need_update(Timestamp::now())) {
       _rate.update();
-      atomic_inc(&_skb->users);
-      p = reinterpret_cast<Packet *>(_skb);
+      p = _packet->clone();
     }
   } else {
-    atomic_inc(&_skb->users);
-    p = reinterpret_cast<Packet *>(_skb);
+    p = _packet->clone();
   }
 
   if(p) {
-    assert(atomic_read(&_skb->users) > 1);
     _count++;
     if(_count == 1)
       _first = click_jiffies();

diff --git a/elements/linuxmodule/fastudpsrc.hh b/elements/linuxmodule/fastudpsrc.hh
@@ -69,8 +69,7 @@ class FastUDPSource : public Element {
   unsigned short _incr;
   unsigned int _interval;
   bool _cksum;
-  WritablePacket *_packet;
-  struct sk_buff *_skb;
+  Packet *_packet;
   click_jiffies_t _first;
   click_jiffies_t _last;
 

diff --git a/elements/linuxmodule/fastudpsrcip6.cc b/elements/linuxmodule/fastudpsrcip6.cc
@@ -80,7 +80,9 @@ FastUDPSourceIP6::configure(Vector<String> &conf, ErrorHandler *errh)
 void
 FastUDPSourceIP6::incr_ports()
 {
-  click_ip6 *ip6 = reinterpret_cast<click_ip6 *>(_packet->data()+14);
+  WritablePacket *q = _packet->uniqueify(); // better not fail
+  _packet = q;
+  click_ip6 *ip6 = reinterpret_cast<click_ip6 *>(q->data()+14);
   click_udp *udp = reinterpret_cast<click_udp *>(ip6 + 1);
   _incr++;
   udp->uh_sport = htons(_sport+_incr);
@@ -102,9 +104,10 @@ FastUDPSourceIP6::initialize(ErrorHandler *)
 {
   _count = 0;
   _incr = 0;
-  _packet = Packet::make(_len);
-  memcpy(_packet->data(), &_ethh, 14);
-  click_ip6 *ip6 = reinterpret_cast<click_ip6 *>(_packet->data()+14);
+  WritablePacket *q = Packet::make(_len);
+  _packet = q;
+  memcpy(q->data(), &_ethh, 14);
+  click_ip6 *ip6 = reinterpret_cast<click_ip6 *>(q->data()+14);
   click_udp *udp = reinterpret_cast<click_udp *>(ip6 + 1);
 
   // set up IP6 header
@@ -134,7 +137,6 @@ FastUDPSourceIP6::initialize(ErrorHandler *)
   } else
     udp->uh_sum = 0;
 
-  _skb = _packet->skb();
   return 0;
 }
 
@@ -157,12 +159,10 @@ FastUDPSourceIP6::pull(int)
   if(_rate_limited){
     if (_rate.need_update(Timestamp::now())) {
       _rate.update();
-      atomic_inc(&_skb->users);
-      p = reinterpret_cast<Packet *>(_skb);
+      p = _packet->clone();
     }
   } else {
-    atomic_inc(&_skb->users);
-    p = reinterpret_cast<Packet *>(_skb);
+    p = _packet->clone();
   }
 
   if(p) {

diff --git a/elements/linuxmodule/fastudpsrcip6.hh b/elements/linuxmodule/fastudpsrcip6.hh
@@ -70,8 +70,7 @@ class FastUDPSourceIP6 : public Element {
   unsigned short _incr;
   unsigned int _interval;
   bool _cksum;
-  WritablePacket *_packet;
-  struct sk_buff *_skb;
+  Packet *_packet;
   click_jiffies_t _first;
   click_jiffies_t _last;