Improve encoder eviction algorithm and datastructure use

Summary:
The QPACK encoder needs to track what entries in its table are possibly in use by the peer to prevent evicting them too early.  The existing code keeps a reference count for every header in the table, and only evicts when the ref count is 0.  When sending a block referencing tht table, the encoder keeps a list of the absolute table indexes that block is referencing, so it can do the accounting.

This is expensive in both space and time, and unecessary.

We only need to know the minimum value in the table that is referenced, since eviction is in FIFO order.  So we get rid of the (relatively large) refcount table and just keep a minInUseIndex.

The encoder only needs to track the min index for each outstanding block.  The trick is computing the new min efficiently after a header acknowledgement or reset.  Since std::heap functions do not provide a function for removal of an arbitrary element, QPACKEncoder keeps a vector of mins, and the min-min.  Insert (encoding) is O(1).  Removing on ack is O(n), where n is very commonly *at most* 100 and frequently much-much smaller, and the constant factor is vector traversal.  The removal and recomputation of the min is handled in one pass.

Reviewed By: jbeshay

Differential Revision: D56500544

fbshipit-source-id: 476024c43dd3d11d86d60ef76b9773052cad5ece

Author: afrind

proxygen/lib/http/codec/compress/QPACKEncoder.cpp

@@ -84,16 +84,19 @@ std::unique_ptr<folly::IOBuf> QPACKEncoder::completeEncode(
     streamBuffer->prependChain(std::move(streamBlock));
   }
 
-  // curOutstanding_.references could be empty, if the block encodes only static
-  // headers and/or literals.  If so we don't track anything.
-  if (!curOutstanding_.references.empty()) {
+  // curOutstanding_.minInUseIndex could be max, if the block encodes only
+  // static headers and/or literals.  If so we don't track anything.
+  if (curOutstanding_.minInUseIndex != std::numeric_limits<uint32_t>::max()) {
+    outstandingMins_.push_back(curOutstanding_.minInUseIndex);
     if (curOutstanding_.vulnerable) {
       DCHECK(allowVulnerable());
       numVulnerable_++;
     }
     numOutstandingBlocks_++;
     outstanding_[streamId].emplace_back(std::move(curOutstanding_));
     curOutstanding_.vulnerable = false;
+    curOutstanding_.minInUseIndex = std::numeric_limits<uint32_t>::max();
+    curOutstanding_.maxInUseIndex = 0;
   }
 
   controlBuffer_.setWriteBuf(nullptr);
@@ -275,14 +278,17 @@ void QPACKEncoder::trackReference(uint32_t absoluteIndex,
   CHECK_NE(absoluteIndex, 0);
   if (absoluteIndex > requiredInsertCount) {
     requiredInsertCount = absoluteIndex;
+    curOutstanding_.maxInUseIndex = requiredInsertCount;
     if (table_.isVulnerable(absoluteIndex)) {
       curOutstanding_.vulnerable = true;
     }
   }
-  auto res = curOutstanding_.references.insert(absoluteIndex);
-  if (res.second) {
-    VLOG(5) << "Bumping refcount for absoluteIndex=" << absoluteIndex;
-    table_.addRef(absoluteIndex);
+  if (absoluteIndex < curOutstanding_.minInUseIndex) {
+    curOutstanding_.minInUseIndex = absoluteIndex;
+    minOutstandingMin_ =
+        std::min(minOutstandingMin_, curOutstanding_.minInUseIndex);
+    // Update table min here to prevent new encodes from evicting reference
+    table_.setMinInUseIndex(minOutstandingMin_);
   }
 }
 
@@ -437,44 +443,60 @@ HPACK::DecodeError QPACKEncoder::onHeaderAck(uint64_t streamId, bool all) {
   VLOG(5) << ((all) ? "onCancelStream" : "onHeaderAck")
           << " streamId=" << streamId;
   if (all) {
-    // Happens when a stream is reset
+    // Happens when a stream is reset (should be rare)
     for (auto& block : it->second) {
-      for (auto i : block.references) {
-        VLOG(5) << "Decrementing refcount for absoluteIndex=" << i;
-        table_.subRef(i);
-      }
       if (block.vulnerable) {
         numVulnerable_--;
       }
+      removeFromMinOutstanding(block.minInUseIndex);
     }
     numOutstandingBlocks_ -= it->second.size();
     it->second.clear();
   } else {
     auto block = std::move(it->second.front());
     numOutstandingBlocks_--;
     it->second.pop_front();
-    // a different stream, sub all the references
-    for (auto i : block.references) {
-      VLOG(5) << "Decrementing refcount for absoluteIndex=" << i;
-      table_.subRef(i);
-    }
     if (block.vulnerable) {
       numVulnerable_--;
     }
-    // requiredInsertCount is implicitly acknowledged
-    if (!block.references.empty()) {
-      auto requiredInsertCount = *block.references.rbegin();
-      VLOG(5) << "Implicitly acknowledging requiredInsertCount="
-              << requiredInsertCount;
-      table_.setAcknowledgedInsertCount(requiredInsertCount);
-    }
+    CHECK_NE(block.minInUseIndex, std::numeric_limits<uint32_t>::max());
+    removeFromMinOutstanding(block.minInUseIndex);
+    // Up through maxInUseIndex is implicitly acknowledged
+    VLOG(5) << "Implicitly acknowledging requiredInsertCount="
+            << block.maxInUseIndex;
+    table_.setAcknowledgedInsertCount(block.maxInUseIndex);
   }
   if (it->second.empty()) {
     outstanding_.erase(it);
   }
+  VLOG(6) << "New min use index=" << minOutstandingMin_;
+  table_.setMinInUseIndex(minOutstandingMin_);
   return HPACK::DecodeError::NONE;
 }
 
+void QPACKEncoder::removeFromMinOutstanding(uint32_t valToRemove) {
+  CHECK(!outstandingMins_.empty());
+  VLOG(10) << "mins remove val=" << valToRemove;
+  bool recomputeMin = (valToRemove == minOutstandingMin_);
+  uint32_t newMin = std::numeric_limits<uint32_t>::max();
+  size_t i = 0;
+  for (; i < outstandingMins_.size(); i++) {
+    auto value = outstandingMins_.at(i);
+    if (value == valToRemove) {
+      outstandingMins_.at(i) = outstandingMins_.back();
+      outstandingMins_.pop_back();
+      break;
+    }
+    newMin = std::min(newMin, value);
+  }
+  if (recomputeMin) {
+    for (; i < outstandingMins_.size(); i++) {
+      newMin = std::min(newMin, outstandingMins_.at(i));
+    }
+    minOutstandingMin_ = newMin;
+  }
+}
+
 void QPACKEncoder::setMaxNumOutstandingBlocks(uint32_t value) {
   maxNumOutstandingBlocks_ = value;
 }

proxygen/lib/http/codec/compress/QPACKEncoder.h

@@ -170,14 +170,18 @@ class QPACKEncoder
     return maxEncoderStreamBytes_ >= 0;
   }
 
+  void removeFromMinOutstanding(uint32_t val);
+
   HPACKEncodeBuffer controlBuffer_;
-  using BlockReferences = std::set<uint32_t>;
   struct OutstandingBlock {
-    BlockReferences references;
+    uint32_t minInUseIndex{std::numeric_limits<uint32_t>::max()};
+    uint32_t maxInUseIndex{0};
     bool vulnerable{false};
   };
   // Map streamID -> list of table index references for each outstanding block;
   std::unordered_map<uint64_t, std::list<OutstandingBlock>> outstanding_;
+  std::vector<uint32_t> outstandingMins_;
+  uint32_t minOutstandingMin_{std::numeric_limits<uint32_t>::max()};
   OutstandingBlock curOutstanding_;
   uint32_t maxDepends_{0};
   uint32_t maxVulnerable_{HPACK::kDefaultBlocking};

proxygen/lib/http/codec/compress/QPACKHeaderTable.cpp

@@ -34,8 +34,8 @@ namespace proxygen {
 QPACKHeaderTable::QPACKHeaderTable(uint32_t capacityVal, bool trackReferences)
     : HeaderTable(capacityVal) {
   if (trackReferences) {
-    refCount_ = std::make_unique<std::vector<uint16_t>>(table_.size(), 0);
     minFree_ = getMinFree(capacityVal);
+    trackReferences_ = true;
   } else {
     minFree_ = 0;
     disableNamesIndex();
@@ -49,13 +49,10 @@ bool QPACKHeaderTable::add(HPACKHeader header) {
     return false;
   }
 
-  DVLOG(6) << "Adding header=" << header;
+  DVLOG(6) << "Adding header=" << header << " absIndex=" << insertCount_ + 1;
   if (!HeaderTable::add(std::move(header))) {
     return false;
   }
-  if (refCount_) {
-    (*refCount_)[head_] = 0;
-  }
   DCHECK_EQ(internalToAbsolute(head_), insertCount_);
   // Increase minUsable_ until the free space + drainedBytes is >= minFree.
   // For HPACK, minFree is 0 and this is a no-op.
@@ -74,7 +71,7 @@ bool QPACKHeaderTable::setCapacity(uint32_t capacity) {
   if (!HeaderTable::setCapacity(capacity)) {
     return false;
   }
-  if (refCount_) {
+  if (trackReferences_) {
     minFree_ = getMinFree(capacity);
   } // else minFree is always 0
   return true;
@@ -135,8 +132,9 @@ const HPACKHeader& QPACKHeaderTable::getHeader(uint32_t index,
 
 uint32_t QPACKHeaderTable::removeLast() {
   auto idx = tail();
-  if (refCount_) {
-    CHECK_EQ((*refCount_)[idx], 0) << "Removed header with nonzero references";
+  if (trackReferences_) {
+    CHECK_LT(internalToAbsolute(idx), minInUseIndex_)
+        << "Removed in use header";
   }
   auto removedBytes = HeaderTable::removeLast();
   // Only non-zero when minUsable_ > insertCount_ - size_.
@@ -165,24 +163,6 @@ void QPACKHeaderTable::increaseTableLengthTo(uint32_t newLength) {
   }
 }
 
-void QPACKHeaderTable::resizeTable(uint32_t newLength) {
-  HeaderTable::resizeTable(newLength);
-  if (refCount_) {
-    refCount_->resize(newLength);
-  }
-}
-
-void QPACKHeaderTable::updateResizedTable(uint32_t oldTail,
-                                          uint32_t oldLength,
-                                          uint32_t newLength) {
-  HeaderTable::updateResizedTable(oldTail, oldLength, newLength);
-  if (refCount_) {
-    std::move_backward(refCount_->begin() + oldTail,
-                       refCount_->begin() + oldLength,
-                       refCount_->begin() + newLength);
-  }
-}
-
 uint32_t QPACKHeaderTable::evict(uint32_t needed, uint32_t desiredCapacity) {
   if (bytes_ + needed < desiredCapacity ||
       !canEvict(bytes_ + needed - desiredCapacity)) {
@@ -192,21 +172,22 @@ uint32_t QPACKHeaderTable::evict(uint32_t needed, uint32_t desiredCapacity) {
 }
 
 bool QPACKHeaderTable::canEvict(uint32_t needed) {
-  if (size_ == 0 || !refCount_) {
+  if (size_ == 0 || !trackReferences_) {
     return needed <= capacity_;
   }
   uint32_t freeable = 0;
   uint32_t i = tail();
   uint32_t nChecked = 0;
   while (nChecked++ < size() && freeable < needed &&
-         ((*refCount_)[i] == 0) && // don't evict referenced or unacked headers
+         internalToAbsolute(i) < minInUseIndex_ && // don't evict referenced or
+                                                   // unacked headers
          internalToAbsolute(i) <= ackedInsertCount_) {
     freeable += table_[i].bytes();
     i = next(i);
   }
   if (freeable < needed) {
     DVLOG(5) << "header=" << table_[i].name << ":" << table_[i].value
-             << " blocked eviction, recount=" << (*refCount_)[i];
+             << " blocked eviction, minInUseIndex_=" << minInUseIndex_;
     return false;
   }
   return true;
@@ -253,20 +234,6 @@ std::pair<bool, uint32_t> QPACKHeaderTable::maybeDuplicate(
   return {false, absIndex};
 }
 
-void QPACKHeaderTable::addRef(uint32_t absIndex) {
-  // refCount is 16 bits.  It should really never get this big in practice,
-  // unless a decoder is not sending HEADER_ACK in a timely way.
-  CHECK(refCount_);
-  (*refCount_)[absoluteToInternal(absIndex)]++;
-}
-
-void QPACKHeaderTable::subRef(uint32_t absIndex) {
-  CHECK(refCount_);
-  uint32_t index = absoluteToInternal(absIndex);
-  CHECK_GT((*refCount_)[index], 0);
-  (*refCount_)[index]--;
-}
-
 // Converts an array index in [0..table_.size() - 1] to an absolute
 // external index
 uint32_t QPACKHeaderTable::internalToAbsolute(uint32_t internalIndex) const {

proxygen/lib/http/codec/compress/QPACKHeaderTable.h

@@ -161,16 +161,10 @@ class QPACKHeaderTable : public HeaderTable {
     return insertCount_ - absIndex + 1;
   }
 
-  /**
-   * Add a reference for the given index.  Entries with non-zero references
-   * cannot be evicted
-   */
-  void addRef(uint32_t absIndex);
-
-  /**
-   * Subtract a reference for the given index
-   */
-  void subRef(uint32_t absIndex);
+  void setMinInUseIndex(
+      uint32_t minInUseIndex = std::numeric_limits<uint32_t>::max()) {
+    minInUseIndex_ = minInUseIndex;
+  }
 
  private:
   /*
@@ -186,11 +180,6 @@ class QPACKHeaderTable : public HeaderTable {
    */
   void increaseTableLengthTo(uint32_t newLength) override;
 
-  void resizeTable(uint32_t newLength) override;
-
-  void updateResizedTable(uint32_t oldTail,
-                          uint32_t oldLength,
-                          uint32_t newLength) override;
   /**
    * Removes one header entry from the beginning of the header table.
    */
@@ -218,7 +207,8 @@ class QPACKHeaderTable : public HeaderTable {
   uint32_t minUsable_{1};
   uint32_t ackedInsertCount_{0};
   uint32_t minFree_{0};
-  std::unique_ptr<std::vector<uint16_t>> refCount_;
+  bool trackReferences_{false};
+  uint32_t minInUseIndex_{std::numeric_limits<uint32_t>::max()};
 };
 
 } // namespace proxygen

proxygen/lib/http/codec/compress/test/QPACKContextTests.cpp

@@ -484,6 +484,27 @@ TEST(QPACKContextTests, TestAcks) {
   EXPECT_EQ(encoder.onInsertCountIncrement(0), HPACK::DecodeError::INVALID_ACK);
 }
 
+TEST(QPACKContextTests, TestEncodeBlocksSelfEviction) {
+  // The references already made in an encode prevent eviction even before
+  // finishing the encode.
+  QPACKEncoder encoder(false, 192); // min free 48
+  QPACKDecoder decoder(192);
+
+  vector<HPACKHeader> req;
+  req.emplace_back("aaaa", "xxxxxxxxxxxx"); // 48
+  auto result = encoder.encode(req, 0, 1);
+  verifyDecode(decoder, std::move(result), req);
+  EXPECT_EQ(encoder.onHeaderAck(1, false), HPACK::DecodeError::NONE);
+
+  req.emplace_back("bbbb", "xxxxxxxxxxxx");  // 48
+  req.emplace_back("cccc", "xxxxxxxxxxxxx"); // 49, drains A
+  req.emplace_back("dddd", "xxxxxxxxxxxxx"); // 48, not enough space
+  result = encoder.encode(req, 0, 1);
+  EXPECT_FALSE(stringInOutput(result.stream.get(), "aaaa"));
+  EXPECT_TRUE(stringInOutput(result.stream.get(), "dddd"));
+  verifyDecode(decoder, std::move(result), req);
+}
+
 TEST(QPACKContextTests, TestImplicitAcks) {
   QPACKEncoder encoder(false, 1024);
   QPACKDecoder decoder(1024);

proxygen/lib/http/codec/compress/test/QPACKHeaderTableTests.cpp

@@ -50,19 +50,12 @@ TEST_F(QPACKHeaderTableTests, Eviction) {
   for (auto i = 0; i < max; i++) {
     EXPECT_TRUE(table_.add(accept.copy()));
   }
-  for (auto i = 1; i <= max; i++) {
-    table_.addRef(i);
-  }
+  table_.setMinInUseIndex(1);
   table_.setAcknowledgedInsertCount(max);
   EXPECT_FALSE(table_.canIndex(accept.name, accept.value));
   EXPECT_FALSE(table_.add(accept.copy()));
-  table_.subRef(1);
-  EXPECT_TRUE(table_.canIndex(accept.name, accept.value));
-  EXPECT_TRUE(table_.add(accept.copy()));
 
-  table_.subRef(3);
-  EXPECT_FALSE(table_.canIndex(accept.name, accept.value));
-  table_.subRef(2);
+  table_.setMinInUseIndex(std::numeric_limits<uint32_t>::max());
   EXPECT_TRUE(table_.canIndex(accept.name, accept.value));
 }
 
@@ -82,11 +75,11 @@ TEST_F(QPACKHeaderTableTests, BadEviction) {
 
   // Ack all headers but mark the first as in use
   table_.setAcknowledgedInsertCount(max);
-  table_.addRef(1);
+  table_.setMinInUseIndex(1);
   EXPECT_FALSE(table_.setCapacity(capacity / 2));
 
   // Clear all refs
-  table_.subRef(1);
+  table_.setMinInUseIndex(std::numeric_limits<uint32_t>::max());
   EXPECT_TRUE(table_.setCapacity(capacity / 2));
   EXPECT_EQ(table_.size(), max / 2);
 }
@@ -167,7 +160,7 @@ TEST_F(QPACKHeaderTableTests, Duplication) {
 
   // Hold a ref to oldest entry, prevents eviction
   auto oldestAbsolute = table_.getInsertCount() - table_.size() + 1;
-  table_.addRef(oldestAbsolute);
+  table_.setMinInUseIndex(oldestAbsolute);
 
   // Table should be full
   EXPECT_FALSE(table_.canIndex(accept.name, accept.value));
@@ -186,7 +179,7 @@ TEST_F(QPACKHeaderTableTests, CanEvictWithRoom) {
   table_.setAcknowledgedInsertCount(table_.getInsertCount());
   // abs index = 1 is evictable, but index = 2 is referenced, so we can
   // insert up to (320 - 8 * 39) + 39 = 47
-  table_.addRef(2);
+  table_.setMinInUseIndex(2);
   EXPECT_TRUE(table_.canIndex(fortySevenBytes.name, fortySevenBytes.value));
   EXPECT_TRUE(table_.add(fortySevenBytes.copy()));
 }