chore: update helio together with new mimalloc version (#3040)

Signed-off-by: Roman Gershman <roman@dragonflydb.io>

helio

@@ -1 +1 @@
-Subproject commit 075fed73d7be99618731b727b36de41732b5223e
+Subproject commit ab82d99f58b7b1ad8d0808061d9082972085576f

src/core/compact_object_test.cc

@@ -106,9 +106,6 @@ class CompactObjectTest : public ::testing::Test {
 };
 
 TEST_F(CompactObjectTest, WastedMemoryDetection) {
-  GTEST_SKIP() << "TODO: this test is unreliable and must be revisited";
-  mi_option_set(mi_option_decommit_delay, 0);
-
   size_t allocated = 0, commited = 0, wasted = 0;
   // By setting the threshold to high value we are expecting
   // To find locations where we have wasted memory
@@ -181,15 +178,12 @@ TEST_F(CompactObjectTest, WastedMemoryDetection) {
 }
 
 TEST_F(CompactObjectTest, WastedMemoryDontCount) {
-  GTEST_SKIP() << "TODO: this test is unreliable and must be revisited";
-
   // The commited memory per blocks are:
   // 64bit => 4K
   // 128bit => 8k
   // 256 => 16k
   // and so on, which mean every n * sizeof(ptr) ^ 2 == 2^11*2*(n-1) (where n starts with 1)
   constexpr std::size_t kExpectedFor256MemWasted = 0x4000;  // memory block 256
-  mi_option_set(mi_option_decommit_delay, 0);
   auto* myheap = mi_heap_get_backing();
 
   size_t allocated = 0, commited = 0, wasted = 0;

src/core/segment_allocator.cc

@@ -3,22 +3,23 @@
 //
 #include "core/segment_allocator.h"
 
-#include <mimalloc-types.h>
+#include <mimalloc/types.h>
 
 #include "base/logging.h"
 
-constexpr size_t kSegmentSize = MI_SEGMENT_SIZE;
-
-// mimalloc uses 32MiB segments and we might need change this code if it changes.
-static_assert(kSegmentSize == 1 << 25);
+constexpr size_t kSegmentShift = MI_SEGMENT_SHIFT;
 
 namespace dfly {
 
 SegmentAllocator::SegmentAllocator(mi_heap_t* heap) : heap_(heap) {
+  // mimalloc uses 4MiB segments and we might need change this code if it changes.
+  constexpr size_t kSegLogSpan = 32 - kSegmentIdBits + 3;
+  static_assert(kSegLogSpan == kSegmentShift);
+  static_assert((~kSegmentAlignMask) == (MI_SEGMENT_SIZE - 1));
 }
 
 void SegmentAllocator::ValidateMapSize() {
-  if (address_table_.size() > 1u << 12) {
+  if (address_table_.size() > (1u << kSegmentIdBits)) {
     // This can happen if we restrict dragonfly to small number of threads on high-memory machine,
     // for example.
     LOG(WARNING) << "address_table_ map is growing too large: " << address_table_.size();

src/core/segment_allocator.h

@@ -17,14 +17,16 @@ namespace dfly {
 
 /**
  * @brief Tightly coupled with mi_malloc 2.x implementation.
- *        Fetches 8MB segment pointers from the allocated pointers.
+ *        Fetches 4MB segment pointers from the allocated pointers.
  *        Provides own indexing of small pointers to real address space using the segment ptrs/
  */
 
 class SegmentAllocator {
-  static constexpr uint32_t kSegmentIdBits = 12;
-  static constexpr uint32_t kSegmentIdMask = (1 << kSegmentIdBits) - 1;
-  static constexpr uint64_t kSegmentAlignMask = ~((1 << 23) - 1);
+  static constexpr uint32_t kSegmentIdBits = 13;
+  static constexpr uint32_t kSegmentIdMask = (1u << kSegmentIdBits) - 1;
+
+  // Segment range that we cover within a single segment.
+  static constexpr uint64_t kSegmentAlignMask = ~((1ULL << (32 - kSegmentIdBits + 3)) - 1);
 
  public:
   using Ptr = uint32_t;

src/facade/dragonfly_connection.cc

@@ -93,7 +93,7 @@ bool MatchHttp11Line(string_view line) {
          absl::EndsWith(line, "HTTP/1.1");
 }
 
-void UpdateIoBufCapacity(const base::IoBuf& io_buf, ConnectionStats* stats,
+void UpdateIoBufCapacity(const io::IoBuf& io_buf, ConnectionStats* stats,
                          absl::FunctionRef<void()> f) {
   const size_t prev_capacity = io_buf.Capacity();
   f();

src/facade/dragonfly_connection.h

@@ -14,11 +14,11 @@
 #include <utility>
 #include <variant>
 
-#include "base/io_buf.h"
 #include "facade/acl_commands_def.h"
 #include "facade/facade_types.h"
 #include "facade/memcache_parser.h"
 #include "facade/resp_expr.h"
+#include "io/io_buf.h"
 #include "util/connection.h"
 #include "util/fibers/fibers.h"
 #include "util/http/http_handler.h"
@@ -279,7 +279,7 @@ class Connection : public util::Connection {
 
   struct MemoryUsage {
     size_t mem = 0;
-    base::IoBuf::MemoryUsage buf_mem;
+    io::IoBuf::MemoryUsage buf_mem;
   };
   MemoryUsage GetMemoryUsage() const;
 
@@ -398,7 +398,7 @@ class Connection : public util::Connection {
 
   size_t pending_pipeline_cmd_cnt_ = 0;  // how many queued async commands in dispatch_q
 
-  base::IoBuf io_buf_;  // used in io loop and parsers
+  io::IoBuf io_buf_;  // used in io loop and parsers
   std::unique_ptr<RedisParser> redis_parser_;
   std::unique_ptr<MemcacheParser> memcache_parser_;
 

src/server/dfly_bench.cc

@@ -13,9 +13,9 @@
 #include "absl/time/time.h"
 #include "base/histogram.h"
 #include "base/init.h"
-#include "base/io_buf.h"
 #include "base/zipf_gen.h"
 #include "facade/redis_parser.h"
+#include "io/io_buf.h"
 #include "util/fibers/dns_resolve.h"
 #include "util/fibers/pool.h"
 #include "util/fibers/uring_socket.h"
@@ -268,7 +268,7 @@ void Driver::Run(uint32_t num_reqs, uint64_t cycle_ns, base::Histogram* dest) {
 
 void Driver::ReceiveFb(base::Histogram* dest) {
   facade::RedisParser parser{1 << 16, false};
-  base::IoBuf io_buf{512};
+  io::IoBuf io_buf{512};
   unsigned num_resp = 0;
   while (true) {
     auto buf = io_buf.AppendBuffer();

src/server/dfly_main.cc

@@ -709,7 +709,7 @@ Usage: dragonfly [FLAGS]
   // export MIMALLOC_VERBOSE=1 to see the options before the override.
   mi_option_enable(mi_option_show_errors);
   mi_option_set(mi_option_max_warnings, 0);
-  mi_option_set(mi_option_decommit_delay, 1);
+  mi_option_enable(mi_option_purge_decommits);
 
   fb2::SetDefaultStackResource(&fb2::std_malloc_resource, kFiberDefaultStackSize);
 

src/server/dragonfly_test.cc

@@ -23,6 +23,7 @@ extern "C" {
 #include "server/test_utils.h"
 
 ABSL_DECLARE_FLAG(float, mem_defrag_threshold);
+ABSL_DECLARE_FLAG(float, mem_defrag_waste_threshold);
 ABSL_DECLARE_FLAG(uint32_t, mem_defrag_check_sec_interval);
 ABSL_DECLARE_FLAG(std::vector<std::string>, rename_command);
 ABSL_DECLARE_FLAG(double, oom_deny_ratio);
@@ -645,6 +646,8 @@ TEST_F(DefragDflyEngineTest, TestDefragOption) {
   // mem_defrag_threshold is based on RSS statistic, but we don't count it in the test
   absl::SetFlag(&FLAGS_mem_defrag_threshold, 0.0);
   absl::SetFlag(&FLAGS_mem_defrag_check_sec_interval, 0);
+  absl::SetFlag(&FLAGS_mem_defrag_waste_threshold, 0.1);
+
   //  Fill data into dragonfly and then check if we have
   //  any location in memory to defrag. See issue #448 for details about this.
   constexpr size_t kMaxMemoryForTest = 1'100'000;
@@ -689,13 +692,13 @@ TEST_F(DefragDflyEngineTest, TestDefragOption) {
 
   ArgSlice delete_cmd(keys);
   r = CheckedInt(delete_cmd);
-  LOG(WARNING) << "finish deleting memory entries " << r;
+  LOG(INFO) << "finish deleting memory entries " << r;
   // the first element in this is the command del so size is one less
   ASSERT_EQ(r, keys2delete.size() - 1);
   // At this point we need to see whether we did running the task and whether the task did something
   shard_set->pool()->AwaitFiberOnAll([&](unsigned index, ProactorBase* base) {
     EngineShard* shard = EngineShard::tlocal();
-    ASSERT_FALSE(shard == nullptr);  // we only have one and its should not be empty!
+    ASSERT_TRUE(shard != nullptr);  // we only have one and its should not be empty!
     // a "busy wait" to ensure that memory defragmentations was successful:
     // the task ran and did it work
     auto stats = shard->stats();

src/server/io_utils.cc

@@ -20,7 +20,7 @@ io::Result<size_t> BufferedStreamerBase::WriteSome(const iovec* vec, uint32_t le
       write_len += vec->iov_len;
     }
     if (write_len < max_buffered_mem_) {
-      producer_buf_ = base::IoBuf{max_buffered_mem_};
+      producer_buf_ = io::IoBuf{max_buffered_mem_};
     }
   }
 

src/server/io_utils.h

@@ -2,8 +2,8 @@
 // See LICENSE for licensing terms.
 //
 
-#include "base/io_buf.h"
 #include "io/io.h"
+#include "io/io_buf.h"
 #include "server/common.h"
 
 namespace dfly {
@@ -72,7 +72,7 @@ class BufferedStreamerBase : public io::Sink {
   unsigned max_buffered_cnt_;  // Max buffered entries before stall
   unsigned max_buffered_mem_;  // Max buffered mem before stall
 
-  base::IoBuf producer_buf_, consumer_buf_;  // Two buffers that are swapped in turns.
+  io::IoBuf producer_buf_, consumer_buf_;  // Two buffers that are swapped in turns.
 };
 
 }  // namespace dfly

src/server/journal/serializer.cc

@@ -6,10 +6,10 @@
 
 #include <system_error>
 
-#include "base/io_buf.h"
 #include "base/logging.h"
 #include "glog/logging.h"
 #include "io/io.h"
+#include "io/io_buf.h"
 #include "server/common.h"
 #include "server/error.h"
 #include "server/journal/types.h"

src/server/journal/serializer.h

@@ -7,8 +7,8 @@
 #include <optional>
 #include <string>
 
-#include "base/io_buf.h"
 #include "io/io.h"
+#include "io/io_buf.h"
 #include "server/common.h"
 #include "server/journal/types.h"
 

src/server/malloc_stats.cc

@@ -4,7 +4,7 @@
 
 // clang-format off
 #include <mimalloc.h>
-#include <mimalloc-types.h>
+#include <mimalloc/types.h>
 
 // clang-format on
 

src/server/memory_cmd.cc

@@ -12,13 +12,13 @@
 
 #include <mimalloc.h>
 
-#include "base/io_buf.h"
 #include "base/logging.h"
 #include "core/allocation_tracker.h"
 #include "facade/cmd_arg_parser.h"
 #include "facade/dragonfly_connection.h"
 #include "facade/dragonfly_listener.h"
 #include "facade/error.h"
+#include "io/io_buf.h"
 #include "server/engine_shard_set.h"
 #include "server/main_service.h"
 #include "server/server_family.h"
@@ -172,11 +172,11 @@ struct ConnectionMemoryUsage {
   size_t connection_count = 0;
   size_t connection_size = 0;
   size_t pipelined_bytes = 0;
-  base::IoBuf::MemoryUsage connections_memory;
+  io::IoBuf::MemoryUsage connections_memory;
 
   size_t replication_connection_count = 0;
   size_t replication_connection_size = 0;
-  base::IoBuf::MemoryUsage replication_memory;
+  io::IoBuf::MemoryUsage replication_memory;
 };
 
 ConnectionMemoryUsage GetConnectionMemoryUsage(ServerFamily* server) {

src/server/protocol_client.h

@@ -8,9 +8,9 @@
 #include <queue>
 #include <variant>
 
-#include "base/io_buf.h"
 #include "facade/facade_types.h"
 #include "facade/redis_parser.h"
+#include "io/io_buf.h"
 #include "server/common.h"
 #include "server/journal/types.h"
 #include "server/version.h"

src/server/rdb_load.cc

@@ -233,10 +233,10 @@ class DecompressImpl {
   }
   virtual ~DecompressImpl() {
   }
-  virtual io::Result<base::IoBuf*> Decompress(std::string_view str) = 0;
+  virtual io::Result<io::IoBuf*> Decompress(std::string_view str) = 0;
 
  protected:
-  base::IoBuf uncompressed_mem_buf_;
+  io::IoBuf uncompressed_mem_buf_;
 };
 
 class ZstdDecompress : public DecompressImpl {
@@ -248,13 +248,13 @@ class ZstdDecompress : public DecompressImpl {
     ZSTD_freeDCtx(dctx_);
   }
 
-  io::Result<base::IoBuf*> Decompress(std::string_view str);
+  io::Result<io::IoBuf*> Decompress(std::string_view str);
 
  private:
   ZSTD_DCtx* dctx_;
 };
 
-io::Result<base::IoBuf*> ZstdDecompress::Decompress(std::string_view str) {
+io::Result<io::IoBuf*> ZstdDecompress::Decompress(std::string_view str) {
   // Prepare membuf memory to uncompressed string.
   auto uncomp_size = ZSTD_getFrameContentSize(str.data(), str.size());
   if (uncomp_size == ZSTD_CONTENTSIZE_UNKNOWN) {

src/server/rdb_load.h

@@ -9,10 +9,10 @@ extern "C" {
 #include "redis/rdb.h"
 }
 
-#include "base/io_buf.h"
 #include "base/mpsc_intrusive_queue.h"
 #include "base/pod_array.h"
 #include "io/io.h"
+#include "io/io_buf.h"
 #include "server/common.h"
 #include "server/journal/serializer.h"
 

src/server/rdb_save.h

@@ -11,9 +11,9 @@ extern "C" {
 
 #include <optional>
 
-#include "base/io_buf.h"
 #include "base/pod_array.h"
 #include "io/io.h"
+#include "io/io_buf.h"
 #include "server/common.h"
 #include "server/journal/serializer.h"
 #include "server/journal/types.h"
@@ -181,7 +181,7 @@ class SerializerBase {
   std::error_code SaveLzfBlob(const ::io::Bytes& src, size_t uncompressed_len);
 
   CompressionMode compression_mode_;
-  base::IoBuf mem_buf_;
+  io::IoBuf mem_buf_;
   std::unique_ptr<CompressorImpl> compressor_impl_;
 
   static constexpr size_t kMinStrSizeToCompress = 256;

src/server/replica.h

@@ -9,9 +9,9 @@
 #include <queue>
 #include <variant>
 
-#include "base/io_buf.h"
 #include "facade/facade_types.h"
 #include "facade/redis_parser.h"
+#include "io/io_buf.h"
 #include "server/cluster/cluster_defs.h"
 #include "server/common.h"
 #include "server/journal/tx_executor.h"

src/server/tiering/disk_storage.cc

@@ -7,8 +7,8 @@
 #include <system_error>
 
 #include "base/flags.h"
-#include "base/io_buf.h"
 #include "base/logging.h"
+#include "io/io_buf.h"
 #include "server/error.h"
 #include "server/tiering/common.h"
 #include "util/fibers/uring_proactor.h"