net: queue packets at the L3 protocol level

If an L3 packet receiver is not able to register itself as a packet receiver
after processing a packet, or if it is simply not dispatched quickly enough,
then we will drop packets.

Add a queue at the protocol layer to buffer those packets.

net/net.cc

@@ -30,6 +30,11 @@ void packet::linearize(size_t at_frag, size_t desired_size) {
     _deleter = make_deleter(std::move(_deleter), [buf = std::move(new_frag)] {});
 }
 
+l3_protocol::l3_protocol(interface* netif, uint16_t proto_num)
+    : _netif(netif), _proto_num(proto_num) {
+    _netif->register_l3(proto_num, 100);
+}
+
 future<packet, ethernet_address> l3_protocol::receive() {
     return _netif->receive(_proto_num);
 };
@@ -39,25 +44,37 @@ future<> l3_protocol::send(ethernet_address to, packet p) {
 }
 
 future<packet, ethernet_address> interface::receive(uint16_t proto_num) {
-    auto& pr = _proto_map[proto_num] = promise<packet, ethernet_address>();
-    return pr.get_future();
+    auto i = _proto_map.find(proto_num);
+    assert(i != _proto_map.end());
+    auto& q = i->second;
+    return q.not_empty().then([&q] {
+        auto x = q.pop();
+        return make_ready_future<packet, ethernet_address>(std::move(std::get<0>(x)), std::move(std::get<1>(x)));
+    });
 }
 
 interface::interface(std::unique_ptr<device> dev)
     : _dev(std::move(dev)), _hw_address(_dev->hw_address()) {
 }
 
+void interface::register_l3(uint16_t proto_num, size_t queue_length) {
+    _proto_map.emplace(std::piecewise_construct,
+                std::make_tuple(proto_num),
+                std::make_tuple(queue_length));
+}
+
 void interface::run() {
     _dev->receive().then([this] (packet p) {
         auto eh = p.get_header<eth_hdr>(0);
         if (eh) {
             ntoh(*eh);
             auto i = _proto_map.find(eh->eth_proto);
-            if (i != _proto_map.end()) {
+            if (i != _proto_map.end() && !i->second.full()) {
                 auto from = eh->src_mac;
                 p.trim_front(sizeof(*eh));
-                i->second.set_value(std::move(p), from);
-                _proto_map.erase(i);
+                i->second.push(std::make_tuple(std::move(p), from));
+            } else {
+                print("dropping packet: no handler for protcol 0x%x\n", eh->eth_proto);
             }
         }
         run();

net/net.hh

@@ -7,6 +7,7 @@
 
 #include "core/reactor.hh"
 #include "core/deleter.hh"
+#include "core/queue.hh"
 #include "ethernet.hh"
 #include <unordered_map>
 
@@ -118,7 +119,7 @@ class l3_protocol {
     interface* _netif;
     uint16_t _proto_num;
 public:
-    explicit l3_protocol(interface* netif, uint16_t proto_num) : _netif(netif), _proto_num(proto_num) {}
+    explicit l3_protocol(interface* netif, uint16_t proto_num);
     future<> send(ethernet_address to, packet p);
     future<packet, ethernet_address> receive();
 private:
@@ -128,7 +129,7 @@ private:
 
 class interface {
     std::unique_ptr<device> _dev;
-    std::unordered_map<uint16_t, promise<packet, ethernet_address>> _proto_map;
+    std::unordered_map<uint16_t, queue<std::tuple<packet, ethernet_address>>> _proto_map;
     ethernet_address _hw_address;
 private:
     future<packet, ethernet_address> receive(uint16_t proto_num);
@@ -137,6 +138,7 @@ public:
     explicit interface(std::unique_ptr<device> dev);
     ethernet_address hw_address() { return _hw_address; }
     void run();
+    void register_l3(uint16_t proto_num, size_t queue_length = 100);
     friend class l3_protocol;
 };