storage: last-ref unmap and defer munmap to the disk thread

src/Makefile.am

@@ -170,6 +170,8 @@ libtorrent_other_la_SOURCES = \
 	utils/diffie_hellman.h \
 	utils/fd_close_queue.cc \
 	utils/fd_close_queue.h \
+	utils/memory_unmap_queue.cc \
+	utils/memory_unmap_queue.h \
 	utils/functional.h \
 	utils/instrumentation.cc \
 	utils/instrumentation.h \

src/data/chunk_list.cc

@@ -162,9 +162,9 @@ ChunkList::get(size_type index, get_flags flags) {
   return ChunkHandle(node, flags & get_writable, flags & get_blocking);
 }
 
-// The chunks in 'm_queue' have been modified and need to be synced
-// when appropriate. Hopefully keeping the chunks mmap'ed for a while
-// will allow us to schedule writes at more resonable intervals.
+// Writable and read handles share the same last-ref unmap path: drop the
+// hold and munmap when references hit zero. Do not keep a sticky writable
+// ref on m_queue just to leave the chunk mmap'd for a later sync pass.
 
 void
 ChunkList::release(ChunkHandle* handle, release_flags flags) {
@@ -185,31 +185,16 @@ ChunkList::release(ChunkHandle* handle, release_flags flags) {
     handle->object()->dec_blocking();
   }
 
-  if (handle->is_writable()) {
-
-    if (handle->object()->writable() == 1) {
-      if (is_queued(handle->object()))
-        throw internal_error("ChunkList::release(...) tried to queue an already queued chunk.");
-
-      // Only add those that have a modification time set?
-      //
-      // Only chunks that are not already in the queue will execute
-      // this branch.
-      m_queue.push_back(handle->object());
-
-      runtime::memory_manager()->account_sync_queue(m_chunk_size);
-
-    } else {
-      handle->object()->dec_rw();
-    }
+  if (handle->is_writable())
+    handle->object()->dec_rw();
+  else
+    handle->object()->dec_references();
 
-  } else {
-    if (handle->object()->dec_references() == 0) {
-      if (is_queued(handle->object()))
-        throw internal_error("ChunkList::release(...) tried to unmap a queued chunk.");
+  if (handle->object()->references() == 0) {
+    if (is_queued(handle->object()))
+      throw internal_error("ChunkList::release(...) tried to unmap a queued chunk.");
 
-      clear_chunk(handle->object(), flags);
-    }
+    clear_chunk(handle->object(), flags);
   }
 
   handle->clear();
@@ -220,7 +205,24 @@ ChunkList::clear_chunk(ChunkListNode* node, release_flags flags) {
   if (!node->is_valid())
     throw internal_error("ChunkList::clear_chunk(...) !node->is_valid().");
 
-  delete node->chunk();
+  // Drop VA ownership here and queue MS_ASYNC + munmap on ChunkManager's
+  // MemoryUnmapQueue worker so MemoryChunk stays a simple type.
+  Chunk* chunk = node->chunk();
+
+  for (auto& part : *chunk) {
+    MemoryChunk& mc = part.chunk();
+
+    if (!mc.is_valid())
+      continue;
+
+    void*  ptr    = mc.ptr();
+    size_t length = static_cast<size_t>(mc.end() - mc.ptr());
+
+    mc.clear();
+    m_manager->queue_munmap(ptr, length);
+  }
+
+  delete chunk;
   node->set_chunk(NULL);
 
   m_manager->deallocate(m_chunk_size, (flags & release_dont_log) ? ChunkManager::allocate_dont_log : 0);

src/data/chunk_manager.cc

@@ -9,12 +9,21 @@
 #include "torrent/exceptions.h"
 #include "torrent/runtime/memory_manager.h"
 #include "utils/instrumentation.h"
+#include "utils/memory_unmap_queue.h"
 
 namespace torrent {
 
-ChunkManager::ChunkManager() = default;
+ChunkManager::ChunkManager() {
+  m_munmap_queue = std::make_unique<utils::MemoryUnmapQueue>();
+}
+
 ChunkManager::~ChunkManager() = default;
 
+void
+ChunkManager::queue_munmap(void* ptr, size_t length) {
+  m_munmap_queue->queue(ptr, length);
+}
+
 void
 ChunkManager::insert(ChunkList* chunkList) {
   chunkList->set_manager(this);

src/data/chunk_manager.h

@@ -1,6 +1,7 @@
 #ifndef LIBTORRENT_CHUNK_MANAGER_H
 #define LIBTORRENT_CHUNK_MANAGER_H
 
+#include <memory>
 #include <vector>
 #include <torrent/common.h>
 
@@ -9,6 +10,12 @@ namespace torrent {
 // TODO: Currently all chunk lists are inserted, despite the download
 // not being open/active.
 
+namespace utils {
+
+class MemoryUnmapQueue;
+
+} // namespace utils
+
 class LIBTORRENT_EXPORT ChunkManager : private std::vector<ChunkList*> {
 public:
   using base_type = std::vector<ChunkList*>;
@@ -46,6 +53,9 @@ class LIBTORRENT_EXPORT ChunkManager : private std::vector<ChunkList*> {
   bool                allocate(uint32_t size, int flags = 0);
   void                deallocate(uint32_t size, int flags = 0);
 
+  // Drop VA ownership; worker thread does MS_ASYNC + munmap.
+  void                queue_munmap(void* ptr, size_t length);
+
   // TODO: Add as a subscription.
   void                try_free_memory(uint64_t size);
 
@@ -59,6 +69,8 @@ class LIBTORRENT_EXPORT ChunkManager : private std::vector<ChunkList*> {
 
   std::chrono::seconds m_last_try_free_memory{};
   size_type            m_last_freed_index{};
+
+  std::unique_ptr<utils::MemoryUnmapQueue> m_munmap_queue;
 };
 
 } // namespace torrent

src/data/chunk_part.cc

@@ -11,7 +11,10 @@ void
 ChunkPart::clear() {
   switch (m_mapped) {
   case MAPPED_MMAP:
-    m_chunk.unmap();
+    // ChunkList::clear_chunk may already have queued the mapping and cleared
+    // the MemoryChunk; only munmap when this path still owns the VA.
+    if (m_chunk.is_valid())
+      m_chunk.unmap();
     break;
 
   default:

src/utils/memory_unmap_queue.cc

@@ -0,0 +1,82 @@
+#include "config.h"
+
+#include "utils/memory_unmap_queue.h"
+
+#include <sys/mman.h>
+
+#include "torrent/exceptions.h"
+
+namespace torrent::utils {
+
+MemoryUnmapQueue::MemoryUnmapQueue() {
+  m_worker = std::async(std::launch::async, [this]() {
+      bool is_running = true;
+
+      while (is_running) {
+        m_wakeup_worker.wait(false, std::memory_order_acquire);
+
+        queue_type queue;
+
+        {
+          std::lock_guard<std::mutex> guard(m_mutex);
+
+          if (m_should_shutdown) {
+            if (m_queue.empty())
+              return;
+
+            is_running = false;
+          }
+
+          if (m_queue.empty())
+            throw internal_error("MemoryUnmapQueue worker thread woke up but queue is empty.");
+
+          queue.swap(m_queue);
+
+          m_wakeup_worker.store(false, std::memory_order_release);
+        }
+
+        // MS_ASYNC starts writeback of dirty MAP_SHARED pages as regions are
+        // retired so the kernel does not accumulate multi-GB dirty sets.
+        // Failures are ignored — munmap still proceeds; durability is not
+        // required here.
+        for (auto& region : queue) {
+          ::msync(region.first, region.second, MS_ASYNC);
+          ::munmap(region.first, region.second);
+        }
+      }
+    });
+}
+
+MemoryUnmapQueue::~MemoryUnmapQueue() {
+  {
+    std::lock_guard<std::mutex> guard(m_mutex);
+    m_should_shutdown = true;
+  }
+
+  m_wakeup_worker.store(true, std::memory_order_release);
+  m_wakeup_worker.notify_all();
+
+  m_worker.wait();
+}
+
+void
+MemoryUnmapQueue::queue(void* ptr, size_t length) {
+  if (ptr == nullptr || length == 0)
+    return;
+
+  {
+    std::lock_guard<std::mutex> guard(m_mutex);
+
+    if (!m_queue.empty()) {
+      m_queue.emplace_back(ptr, length);
+      return;
+    }
+
+    m_queue.emplace_back(ptr, length);
+  }
+
+  m_wakeup_worker.store(true, std::memory_order_release);
+  m_wakeup_worker.notify_all();
+}
+
+} // namespace torrent::utils

src/utils/memory_unmap_queue.h

@@ -0,0 +1,44 @@
+#ifndef LIBTORRENT_UTILS_MEMORY_UNMAP_QUEUE_H
+#define LIBTORRENT_UTILS_MEMORY_UNMAP_QUEUE_H
+
+#include <future>
+#include <mutex>
+#include <utility>
+#include <vector>
+#include <torrent/common.h>
+
+namespace torrent::utils {
+
+// Persistent std::async worker that issues MS_ASYNC + munmap off the caller
+// (and off ThreadDisk). Same shape as FdCloseQueue.
+
+class MemoryUnmapQueue {
+public:
+  MemoryUnmapQueue();
+  ~MemoryUnmapQueue();
+
+  void                queue(void* ptr, size_t length);
+
+private:
+  MemoryUnmapQueue(const MemoryUnmapQueue&) = delete;
+  MemoryUnmapQueue& operator=(const MemoryUnmapQueue&) = delete;
+
+  using queue_type = std::vector<std::pair<void*, size_t>>;
+
+  std::future<void>   m_worker;
+
+  align_cacheline
+
+  std::mutex          m_mutex;
+  queue_type          m_queue;
+
+  bool                m_should_shutdown{};
+
+  align_cacheline
+
+  std::atomic<bool>   m_wakeup_worker{};
+};
+
+} // namespace torrent::utils
+
+#endif