Merge pull request #1228 from fwsGonzo/dev

Bufferstore chaining to support on-demand buffer count

Author: fwsGonzo

api/net/buffer_store.hpp

@@ -35,7 +35,11 @@ namespace net
    **/
   class BufferStore {
   public:
-    using buffer_t = uint8_t*;
+    struct buffer_t
+    {
+      BufferStore* bufstore;
+      uint8_t*     addr;
+    };
 
     BufferStore() = delete;
     BufferStore(size_t num, size_t bufsize);
@@ -45,41 +49,43 @@ namespace net
     void release(void*);
 
     /** Get size of a buffer **/
-    inline size_t bufsize() const noexcept
+    size_t bufsize() const noexcept
     { return bufsize_; }
 
-    inline size_t poolsize() const noexcept
+    size_t poolsize() const noexcept
     { return poolsize_; }
 
     /** Check if a buffer belongs here */
-    inline bool is_from_pool(buffer_t addr) const noexcept
+    bool is_from_pool(uint8_t* addr) const noexcept
     { return addr >= pool_begin() and addr < pool_end(); }
 
     /** Check if an address is the start of a buffer */
-    inline bool is_buffer(buffer_t addr) const noexcept
+    bool is_buffer(uint8_t* addr) const noexcept
     { return (addr - pool_) % bufsize_ == 0; }
 
-    inline size_t available() const noexcept
+    size_t available() const noexcept
     { return available_.size(); }
 
     /** move this bufferstore to the current CPU **/
     void move_to_this_cpu() noexcept;
 
   private:
-    buffer_t pool_begin() const noexcept {
+    uint8_t* pool_begin() const noexcept {
       return pool_;
     }
-    buffer_t pool_end() const noexcept {
+    uint8_t* pool_end() const noexcept {
       return pool_begin() + poolsize_;
     }
-    size_t buffer_id(buffer_t addr) const {
-      return (addr - pool_) / bufsize_;
-    }
+
+    BufferStore* get_next_bufstore();
+    inline buffer_t get_buffer_directly() noexcept;
+    inline void     release_directly(uint8_t*);
 
     size_t               poolsize_;
     size_t               bufsize_;
-    buffer_t             pool_;
-    std::vector<buffer_t> available_;
+    uint8_t*             pool_;
+    std::vector<uint8_t*> available_;
+    BufferStore*         next_;
     int                  cpu;
     static bool          smp_enabled_;
 #ifndef INCLUDEOS_SINGLE_THREADED

src/drivers/virtionet.cpp

@@ -44,7 +44,7 @@ void VirtioNet::get_config() {
 VirtioNet::VirtioNet(hw::PCI_Device& d)
   : Virtio(d),
     Link(Link_protocol{{this, &VirtioNet::transmit}, mac()},
-        std::max(256u, queue_size(0) * 2), 2048 /* half-page buffers */),
+        256u, 2048 /* 256x half-page buffers */),
     packets_rx_{Statman::get().create(Stat::UINT64, device_name() + ".packets_rx").get_uint64()},
     packets_tx_{Statman::get().create(Stat::UINT64, device_name() + ".packets_tx").get_uint64()}
 {
@@ -119,7 +119,7 @@ VirtioNet::VirtioNet(hw::PCI_Device& d)
        rx_q.size() / 2, bufstore().bufsize());
 
   for (int i = 0; i < rx_q.size() / 2; i++)
-      add_receive_buffer(bufstore().get_buffer());
+      add_receive_buffer(bufstore().get_buffer().addr);
 
   // Step 4 - If there are many queues, we should negotiate the number.
   // Set config length, based on whether there are multiple queues
@@ -196,7 +196,7 @@ void VirtioNet::msix_recv_handler()
 
     dequeued_rx++;
     // Requeue a new buffer
-    add_receive_buffer(bufstore().get_buffer());
+    add_receive_buffer(bufstore().get_buffer().addr);
 
     // Stat increase packets received
     packets_rx_++;
@@ -270,13 +270,14 @@ VirtioNet::recv_packet(uint8_t* data, uint16_t size)
 net::Packet_ptr
 VirtioNet::create_packet(int link_offset)
 {
-  auto* ptr = (net::Packet*) bufstore().get_buffer();
+  auto buffer = bufstore().get_buffer();
+  auto* ptr = (net::Packet*) buffer.addr;
 
   new (ptr) net::Packet(
         sizeof(virtio_net_hdr) + link_offset, 
         0, 
         sizeof(virtio_net_hdr) + packet_len(), 
-        &bufstore());
+        buffer.bufstore);
 
   return net::Packet_ptr(ptr);
 }

src/drivers/vmxnet3.cpp

@@ -340,7 +340,7 @@ void vmxnet3::refill(rxring_state& rxq)
         (rxq.producers & vmxnet3::NUM_RX_DESC) ? 0 : VMXNET3_RXF_GEN;
 
     // get a pointer to packet data
-    auto* pkt_data = bufstore().get_buffer();
+    auto* pkt_data = bufstore().get_buffer().addr;
     rxq.buffers[i] = &pkt_data[sizeof(net::Packet)];
 
     // assign rx descriptor
@@ -369,8 +369,13 @@ vmxnet3::recv_packet(uint8_t* data, uint16_t size)
 net::Packet_ptr
 vmxnet3::create_packet(int link_offset)
 {
-  auto* ptr = (net::Packet*) bufstore().get_buffer();
-  new (ptr) net::Packet(frame_offset_device() + link_offset, 0 , frame_offset_device() + packet_len(), &bufstore());
+  auto buffer = bufstore().get_buffer();
+  auto* ptr = (net::Packet*) buffer.addr;
+  new (ptr) net::Packet(
+        frame_offset_device() + link_offset, 
+        0, 
+        frame_offset_device() + packet_len(), 
+        buffer.bufstore);
   return net::Packet_ptr(ptr);
 }
 

src/kernel/irq_manager.cpp

@@ -33,7 +33,11 @@ static std::array<IRQ_manager, SMP_MAX_CORES> managers;
 
 IRQ_manager& IRQ_manager::get(int cpuid)
 {
+#ifndef INCLUDEOS_SINGLE_THREADED
   return managers.at(cpuid);
+#else
+  return managers[0];
+#endif
 }
 IRQ_manager& IRQ_manager::get()
 {

src/net/buffer_store.cpp

@@ -17,8 +17,6 @@
 
 //#define DEBUG
 //#undef  NO_DEBUG
-#include <debug>
-#include <cassert>
 #if !defined(__MACH__)
 #include <malloc.h>
 #else
@@ -27,26 +25,33 @@ extern void *memalign(size_t, size_t);
 #include <net/buffer_store.hpp>
 #include <kernel/syscalls.hpp>
 #include <common>
+#include <debug>
+#include <info>
+#include <cassert>
 #include <smp>
 #define PAGE_SIZE     0x1000
 
+#define ENABLE_BUFFERSTORE_CHAIN
+#define BS_CHAIN_ALLOC_PACKETS   2048
+
 namespace net {
 
   bool BufferStore::smp_enabled_ = false;
 
   BufferStore::BufferStore(size_t num, size_t bufsize) :
     poolsize_  {num * bufsize},
-    bufsize_   {bufsize}
+    bufsize_   {bufsize},
+    next_(nullptr)
   {
     assert(num != 0);
     assert(bufsize != 0);
     const size_t DATA_SIZE  = poolsize_;
 
-    this->pool_ = (buffer_t) memalign(PAGE_SIZE, DATA_SIZE);
+    this->pool_ = (uint8_t*) memalign(PAGE_SIZE, DATA_SIZE);
     assert(this->pool_);
 
     available_.reserve(num);
-    for (buffer_t b = pool_end()-bufsize; b >= pool_begin(); b -= bufsize) {
+    for (uint8_t* b = pool_end()-bufsize; b >= pool_begin(); b -= bufsize) {
         available_.push_back(b);
     }
     assert(available() == num);
@@ -61,9 +66,33 @@ namespace net {
   }
 
   BufferStore::~BufferStore() {
+    delete this->next_;
     free(this->pool_);
   }
 
+  BufferStore* BufferStore::get_next_bufstore()
+  {
+#ifdef ENABLE_BUFFERSTORE_CHAIN
+    BufferStore* parent = this;
+    while (parent->next_ != nullptr) {
+        parent = parent->next_;
+        if (parent->available() != 0) return parent;
+    }
+    INFO("BufferStore", "Allocating %u new packets", BS_CHAIN_ALLOC_PACKETS);
+    parent->next_ = new BufferStore(BS_CHAIN_ALLOC_PACKETS, bufsize());
+    return parent->next_;
+#else
+    return nullptr;
+#endif
+  }
+
+  BufferStore::buffer_t BufferStore::get_buffer_directly() noexcept
+  {
+    auto addr = available_.back();
+    available_.pop_back();
+    return { this, addr };
+  }
+
   BufferStore::buffer_t BufferStore::get_buffer()
   {
 #ifndef INCLUDEOS_SINGLE_THREADED
@@ -78,7 +107,11 @@ namespace net {
 #ifndef INCLUDEOS_SINGLE_THREADED
       if (is_locked) unlock(plock);
 #endif
-      panic("<BufferStore> Storage pool full! Don't know how to increase pool size yet.\n");
+#ifdef ENABLE_BUFFERSTORE_CHAIN
+      return get_next_bufstore()->get_buffer_directly();
+#else
+      panic("<BufferStore> Buffer pool empty! Not configured to increase pool size.\n");
+#endif
     }
 
     auto addr = available_.back();
@@ -87,12 +120,12 @@ namespace net {
 #ifndef INCLUDEOS_SINGLE_THREADED
     if (is_locked) unlock(plock);
 #endif
-    return addr;
+    return { this, addr };
   }
 
   void BufferStore::release(void* addr)
   {
-    buffer_t buff = (buffer_t) addr;
+    auto* buff = (uint8_t*) addr;
     debug("Release %p -> ", buff);
 
 #ifndef INCLUDEOS_SINGLE_THREADED
@@ -111,6 +144,18 @@ namespace net {
       debug("released\n");
       return;
     }
+#ifdef ENABLE_BUFFERSTORE_CHAIN
+    // try to release buffer on linked bufferstore
+    BufferStore* ptr = next_;
+    while (ptr != nullptr) {
+      if (ptr->is_from_pool(buff)) {
+        debug("released on other bufferstore\n");
+        ptr->release_directly(buff);
+        return;
+      }
+      ptr = ptr->next_;
+    }
+#endif
     // buffer not owned by bufferstore, so just delete it?
     debug("deleted\n");
     delete[] buff;
@@ -119,6 +164,11 @@ namespace net {
 #endif
   }
 
+  void BufferStore::release_directly(uint8_t* buffer)
+  {
+    available_.push_back(buffer);
+  }
+
   void BufferStore::move_to_this_cpu() noexcept
   {
     this->cpu = SMP::cpu_id();

test/CMakeLists.txt

@@ -8,7 +8,7 @@ if ("${ARCH}" STREQUAL "")
   set (ARCH "ARCH_X86")
 endif("${ARCH}" STREQUAL "")
 
-add_definitions(-D${ARCH})
+add_definitions(-D${ARCH} -DINCLUDEOS_SINGLE_THREADED)
 
 set(CMAKE_CXX_STANDARD 14)
 set(CMAKE_C_FLAGS "-m32 -g -O0 -std=c11 -Wall -Wextra")

test/net/integration/bufstore/service.cpp

@@ -30,9 +30,10 @@ BufferStore bufstore_{ BUFFER_CNT,  MTU };
 
 auto create_packet(BufferStore& bufstore) {
   // get buffer (as packet + data)
-  auto* ptr = (Packet*) bufstore.get_buffer();
+  auto buffer = bufstore.get_buffer();
   // place packet at front of buffer
-  new (ptr) Packet(0, 0, MTU, &bufstore);
+  auto* ptr = (Packet*) buffer.addr;
+  new (ptr) Packet(0, 0, MTU, buffer.bufstore);
   // regular shared_ptr that calls delete on Packet
   return std::unique_ptr<Packet>(ptr);
 }
@@ -61,28 +62,24 @@ void Service::start(const std::string&)
   packet = nullptr;
   CHECKSERT(bufstore_.available() == BUFFER_CNT, "Bufcount is now %i", BUFFER_CNT);
 
-  INFO("Test 2","Create and chain packets, release one-by-one");
+  INFO("Test 2", "Create and chain %u packets, release one-by-one", BUFFER_CNT*10);
 
   // Reinitialize the first packet
   packet = create_packet(bufstore_);
   CHECKSERT(bufstore_.available() == BUFFER_CNT - 1, "Bufcount is now %i", BUFFER_CNT - 1);
 
   // Chain
-  for (int i = 0; i < chain_size - 1; i++){
+  for (int i = 0; i < 10*BUFFER_CNT - 1; i++){
     auto chained_packet = create_packet(bufstore_);
     packet->chain(std::move(chained_packet));
-    CHECKSERT(bufstore_.available() == BUFFER_CNT - i - 2, "Bufcount is now %i", BUFFER_CNT - i - 2);
   }
 
   INFO("Test 2","Releasing packet-chain one-by-one");
 
   // Release one-by-one
   auto tail = std::move(packet);
   size_t i = 0;
-  while(tail && i < BUFFER_CNT - 1) {
-    CHECKSERT(bufstore_.available() == i,
-              "Bufcount is now %i == %i", i,
-              bufstore_.available());
+  while(tail && i < 10*BUFFER_CNT - 1) {
     tail = tail->detach_tail();
     i++;
   }

test/net/unit/packets.cpp

@@ -46,8 +46,9 @@ CASE("Create empty packet")
 
 Packet_ptr create_packet() noexcept
 {
-  auto* ptr = (net::Packet*) bufstore.get_buffer();
-  new (ptr) net::Packet(DRIVER_OFFSET, 0, DRIVER_OFFSET + PACKET_CAPA, &bufstore);
+  auto buffer = bufstore.get_buffer();
+  auto* ptr = (net::Packet*) buffer.addr;
+  new (ptr) net::Packet(DRIVER_OFFSET, 0, DRIVER_OFFSET + PACKET_CAPA, buffer.bufstore);
   return net::Packet_ptr(ptr);
 }
 

test/skipped_tests.json

@@ -3,8 +3,6 @@
     "reason": "Serial input from vmrunner to vm does not work yet"},
   {"name": "net/integration/transmit",
    "reason": "The test suffers from UDP packet loss which is expected."},
-  {"name": "net/integration/router",
-   "reason": "The test suffers uses too many buffers at present."},
   {"name": "posix/integration/ucontext",
    "reason": "The test shows that ucontext is unrealiable, as is."}
 ]