Parallel dir entry sync options

Author: zwoop

doc/admin-guide/files/records.yaml.en.rst

@@ -2573,6 +2573,23 @@ Cache Control
    :units: millisecond
 
    How long to wait between each write cycle when syncing the cache directory to disk.
+.. ts:cv:: CONFIG proxy.config.cache.dir.sync_parallel_tasks INT 1
+   :reloadable:
+
+   Number of parallel tasks to use for directory syncing. Each task syncs
+   directories for a different physical drive on ET_TASK threads.
+
+   ======= ==================================================================
+   Value   Description
+   ======= ==================================================================
+   ``-1``  Unlimited - one task per drive (maximum parallelism)
+   ``1``   Sequential - one task for all drives (default, safe)
+   ``N``   Parallel - up to N tasks (drives) sync concurrently
+   ======= ==================================================================
+
+   Default is ``1`` (sequential). Set to ``-1`` for maximum parallelism on
+   high-end NVMe arrays, or to ``4-8`` for balanced performance on multi-drive
+   systems.
 
 .. ts:cv:: CONFIG proxy.config.cache.limits.http.max_alts INT 5
 

doc/developer-guide/cache-architecture/architecture.en.rst

@@ -963,6 +963,26 @@ own entry in the volume directory which are computationally chained (each
 table is accumulated in the earliest ``Doc`` which has the offsets of the first
 byte for each fragment.
 
+Directory Synchronization
+-------------------------
+
+The in-memory stripe directories are periodically synchronized to disk to ensure
+cache metadata persistence across restarts. This is handled by ``CacheSync``
+instances scheduled on ET_TASK threads.
+
+By default, directory syncing is sequential (one task handles all stripes). The
+:ts:cv:`proxy.config.cache.dir.sync_parallel_tasks` configuration allows parallel
+syncing across multiple drives:
+
+* ``1`` - Sequential (default): single task syncs all stripes
+* ``N`` - Parallel: up to N tasks sync concurrently
+* ``<0`` - Maximum: one task per physical drive
+
+Stripes are distributed across sync tasks based on their physical drives, ensuring
+that I/O operations for different drives can proceed in parallel on separate ET_TASK
+threads. Each ``CacheSync`` instance maintains a list of stripe indices to sync and
+processes them sequentially within its task.
+
 .. _evacuation-mechanics:
 
 Evacuation Mechanics

src/iocore/cache/Cache.cc

@@ -64,6 +64,7 @@ int     cache_config_log_alternate_eviction        = 0;
 int     cache_config_dir_sync_frequency            = 60;
 int     cache_config_dir_sync_delay                = 500;
 int     cache_config_dir_sync_max_write            = (2 * 1024 * 1024);
+int     cache_config_dir_sync_parallel_tasks       = 1;
 int     cache_config_permit_pinning                = 0;
 int     cache_config_select_alternate              = 1;
 int     cache_config_max_doc_size                  = 0;
@@ -90,7 +91,6 @@ Cache                                  *theCache = nullptr;
 std::vector<std::unique_ptr<CacheDisk>> gdisks;
 int                                     gndisks                      = 0;
 Cache                                  *caches[NUM_CACHE_FRAG_TYPES] = {nullptr};
-CacheSync                              *cacheDirSync                 = nullptr;
 Store                                   theCacheStore;
 StripeSM                              **gstripes  = nullptr;
 std::atomic<int>                        gnstripes = 0;
@@ -884,6 +884,9 @@ ink_cache_init(ts::ModuleVersion v)
 
   cacheProcessor.wait_for_cache = RecGetRecordInt("proxy.config.http.wait_for_cache").value_or(0);
 
+  RecEstablishStaticConfigInt32(cache_config_dir_sync_parallel_tasks, "proxy.config.cache.dir.sync_parallel_tasks");
+  Dbg(dbg_ctl_cache_init, "proxy.config.cache.dir.sync_parallel_tasks = %d", cache_config_dir_sync_parallel_tasks);
+
   RecEstablishStaticConfigInt32(cache_config_persist_bad_disks, "proxy.config.cache.persist_bad_disks");
   Dbg(dbg_ctl_cache_init, "proxy.config.cache.persist_bad_disks = %d", cache_config_persist_bad_disks);
   if (cache_config_persist_bad_disks) {

src/iocore/cache/CacheDir.cc

@@ -30,6 +30,11 @@
 
 #include "tscore/hugepages.h"
 #include "tscore/Random.h"
+#include "iocore/eventsystem/Tasks.h"
+
+#include <vector>
+#include <map>
+#include <atomic>
 
 #ifdef LOOP_CHECK_MODE
 #define DIR_LOOP_THRESHOLD 1000
@@ -849,14 +854,51 @@ dir_lookaside_remove(const CacheKey *key, StripeSM *stripe)
   return;
 }
 
-// Cache Sync
-//
+// Cache Dir Sync
+
+static std::vector<std::unique_ptr<CacheSync>> cache_syncs;
 
 void
 dir_sync_init()
 {
-  cacheDirSync          = new CacheSync;
-  cacheDirSync->trigger = eventProcessor.schedule_in(cacheDirSync, HRTIME_SECONDS(cache_config_dir_sync_frequency));
+  static std::atomic<bool> initialized{false};
+  ink_release_assert(!initialized.exchange(true)); // Should only be called once
+
+  std::map<CacheDisk *, std::vector<int>> drive_stripe_map;
+
+  // Group stripes by disk
+  for (int i = 0; i < gnstripes; i++) {
+    drive_stripe_map[gstripes[i]->disk].push_back(i);
+  }
+
+  // Any negative value means "all drives" (maximum parallelism)
+  int num_tasks = (cache_config_dir_sync_parallel_tasks < 0) ?
+                    drive_stripe_map.size() :
+                    std::min(drive_stripe_map.size(), static_cast<size_t>(std::max(1, cache_config_dir_sync_parallel_tasks)));
+
+  cache_syncs.resize(num_tasks);
+  for (int i = 0; i < num_tasks; i++) {
+    cache_syncs[i] = std::make_unique<CacheSync>();
+  }
+
+  int task_idx = 0;
+
+  for (auto &[disk, indices] : drive_stripe_map) {
+    int target_task = task_idx % num_tasks;
+
+    Dbg(dbg_ctl_cache_dir_sync, "Disk %s: %zu stripe(s) assigned to task %d", disk->path, indices.size(), target_task);
+    for (int stripe_idx : indices) {
+      cache_syncs[target_task]->stripe_indices.push_back(stripe_idx);
+      gstripes[stripe_idx]->cache_sync = cache_syncs[target_task].get();
+    }
+    task_idx++;
+  }
+
+  for (int i = 0; i < num_tasks; i++) {
+    cache_syncs[i]->current_index = 0;
+    cache_syncs[i]->trigger =
+      eventProcessor.schedule_in(cache_syncs[i].get(), HRTIME_SECONDS(cache_config_dir_sync_frequency), ET_TASK);
+  }
 }
 
 void
@@ -909,6 +951,16 @@ void
 sync_cache_dir_on_shutdown()
 {
   Dbg(dbg_ctl_cache_dir_sync, "sync started");
+
+  // Cancel any active async sync tasks first
+  for (auto &sync : cache_syncs) {
+    if (sync && sync->trigger) {
+      sync->trigger->cancel_action();
+      sync->trigger = nullptr;
+    }
+  }
+  cache_syncs.clear();
+
   EThread *t = reinterpret_cast<EThread *>(0xdeadbeef);
   for (int i = 0; i < gnstripes; i++) {
     gstripes[i]->shutdown(t);
@@ -917,34 +969,36 @@ sync_cache_dir_on_shutdown()
 }
 
 int
-CacheSync::mainEvent(int event, Event *e)
+CacheSync::mainEvent(int event, Event * /* e ATS_UNUSED */)
 {
   if (trigger) {
     trigger->cancel_action();
     trigger = nullptr;
   }
 
 Lrestart:
-  if (stripe_index >= gnstripes) {
-    stripe_index = 0;
+  if (current_index >= static_cast<int>(stripe_indices.size())) {
+    current_index = 0;
+#if FREE_BUF_BETWEEN_CYCLES
+    // Free buffer between sync cycles to avoid holding large amounts of memory
+    // ToDo: If we decide needed, we should add a RecordsConfig here.
     if (buf) {
       if (buf_huge) {
         ats_free_hugepage(buf, buflen);
       } else {
         ats_free(buf);
       }
-      buflen   = 0;
       buf      = nullptr;
+      buflen   = 0;
       buf_huge = false;
     }
-    Dbg(dbg_ctl_cache_dir_sync, "sync done");
-    if (event == EVENT_INTERVAL) {
-      trigger = e->ethread->schedule_in(this, HRTIME_SECONDS(cache_config_dir_sync_frequency));
-    } else {
-      trigger = eventProcessor.schedule_in(this, HRTIME_SECONDS(cache_config_dir_sync_frequency));
-    }
+#endif
+    Dbg(dbg_ctl_cache_dir_sync, "sync cycle done");
+    trigger = eventProcessor.schedule_in(this, HRTIME_SECONDS(cache_config_dir_sync_frequency), ET_TASK);
     return EVENT_CONT;
   }
+  stripe_index = stripe_indices[current_index];
+  current_index++;
 
   StripeSM *stripe = gstripes[stripe_index]; // must be named "vol" to make STAT macros work.
 
@@ -1059,9 +1113,7 @@ CacheSync::mainEvent(int event, Event *e)
     return EVENT_CONT;
   }
 Ldone:
-  // done
   writepos = 0;
-  ++stripe_index;
   goto Lrestart;
 }
 

src/iocore/cache/P_CacheDir.h

@@ -28,6 +28,7 @@
 #include "iocore/eventsystem/Continuation.h"
 #include "iocore/aio/AIO.h"
 #include "tscore/Version.h"
+#include "tscore/hugepages.h"
 
 #include <cstdint>
 #include <ctime>
@@ -39,6 +40,7 @@ struct CacheVC;
 class CacheEvacuateDocVC;
 
 // #define LOOP_CHECK_MODE 1
+// #define FREE_BUF_BETWEEN_CYCLES 1
 
 /*
   Directory layout
@@ -246,10 +248,29 @@ struct CacheSync : public Continuation {
   AIOCallback io;
   Event      *trigger    = nullptr;
   ink_hrtime  start_time = 0;
-  int         mainEvent(int event, Event *e);
-  void        aio_write(int fd, char *b, int n, off_t o);
+
+  std::vector<int> stripe_indices;
+  int              current_index{0};
+
+  int  mainEvent(int event, Event *e);
+  void aio_write(int fd, char *b, int n, off_t o);
 
   CacheSync() : Continuation(new_ProxyMutex()) { SET_HANDLER(&CacheSync::mainEvent); }
+
+  ~CacheSync()
+  {
+    if (trigger) {
+      trigger->cancel_action();
+      trigger = nullptr;
+    }
+    if (buf) {
+      if (buf_huge) {
+        ats_free_hugepage(buf, buflen);
+      } else {
+        ats_free(buf);
+      }
+    }
+  }
 };
 
 struct StripteHeaderFooter {

src/iocore/cache/P_CacheInternal.h

@@ -96,6 +96,7 @@ extern CacheStatsBlock cache_rsb;
 extern int cache_config_dir_sync_frequency;
 extern int cache_config_dir_sync_delay;
 extern int cache_config_dir_sync_max_write;
+extern int cache_config_dir_sync_parallel_tasks;
 extern int cache_config_http_max_alts;
 extern int cache_config_log_alternate_eviction;
 extern int cache_config_permit_pinning;
@@ -140,7 +141,6 @@ struct CacheRemoveCont : public Continuation {
 // Global Data
 extern ClassAllocator<CacheVC>            cacheVConnectionAllocator;
 extern ClassAllocator<CacheEvacuateDocVC> cacheEvacuateDocVConnectionAllocator;
-extern CacheSync                         *cacheDirSync;
 // Function Prototypes
 int                 cache_write(CacheVC *, CacheHTTPInfoVector *);
 int                 get_alternate_index(CacheHTTPInfoVector *cache_vector, CacheKey key);

src/iocore/cache/StripeSM.cc

@@ -709,9 +709,9 @@ StripeSM::aggWriteDone(int event, Event *e)
 {
   cancel_trigger();
 
-  // ensure we have the cacheDirSync lock if we intend to call it later
+  // ensure we have the cache_sync lock if we intend to call it later
   // retaking the current mutex recursively is a NOOP
-  CACHE_TRY_LOCK(lock, dir_sync_waiting ? cacheDirSync->mutex : mutex, mutex->thread_holding);
+  CACHE_TRY_LOCK(lock, dir_sync_waiting ? cache_sync->mutex : mutex, mutex->thread_holding);
   if (!lock.is_locked()) {
     eventProcessor.schedule_in(this, HRTIME_MSECONDS(cache_config_mutex_retry_delay));
     return EVENT_CONT;
@@ -759,7 +759,7 @@ StripeSM::aggWriteDone(int event, Event *e)
   }
   if (dir_sync_waiting) {
     dir_sync_waiting = false;
-    cacheDirSync->handleEvent(EVENT_IMMEDIATE, nullptr);
+    cache_sync->handleEvent(EVENT_IMMEDIATE, nullptr);
   }
   if (this->_write_buffer.get_pending_writers().head || sync.head) {
     return aggWrite(event, e);

src/iocore/cache/StripeSM.h

@@ -85,13 +85,14 @@ class StripeSM : public Continuation, public Stripe
 
   StripeInitInfo *init_info = nullptr;
 
-  Cache   *cache                = nullptr;
-  uint32_t last_sync_serial     = 0;
-  uint32_t last_write_serial    = 0;
-  bool     recover_wrapped      = false;
-  bool     dir_sync_waiting     = false;
-  bool     dir_sync_in_progress = false;
-  bool     writing_end_marker   = false;
+  Cache     *cache                = nullptr;
+  uint32_t   last_sync_serial     = 0;
+  uint32_t   last_write_serial    = 0;
+  bool       recover_wrapped      = false;
+  bool       dir_sync_waiting     = false;
+  bool       dir_sync_in_progress = false;
+  bool       writing_end_marker   = false;
+  CacheSync *cache_sync           = nullptr;
 
   CacheKey          first_fragment_key;
   int64_t           first_fragment_offset = 0;

src/records/RecordsConfig.cc

@@ -853,6 +853,8 @@ static constexpr RecordElement RecordsConfig[] =
   ,
   {RECT_CONFIG, "proxy.config.cache.dir.sync_max_write", RECD_INT, "2097152", RECU_DYNAMIC, RR_NULL, RECC_NULL, nullptr, RECA_NULL}
   ,
+  {RECT_CONFIG, "proxy.config.cache.dir.sync_parallel_tasks", RECD_INT, "1", RECU_DYNAMIC, RR_NULL, RECC_NULL, nullptr, RECA_NULL}
+  ,
   {RECT_CONFIG, "proxy.config.cache.hostdb.disable_reverse_lookup", RECD_INT, "0", RECU_DYNAMIC, RR_NULL, RECC_NULL, nullptr, RECA_NULL}
   ,
   {RECT_CONFIG, "proxy.config.cache.select_alternate", RECD_INT, "1", RECU_DYNAMIC, RR_NULL, RECC_NULL, nullptr, RECA_NULL}