Update helio to pull in epoll socket fix (#5974)

* Update helio to pull in epoll socket fix
* server/core: remove queue task index

Queue task index added in
#5716 has since been
removed from helio in romange/helio#476

The call sites are adjusted to previous version in this changeset.

Signed-off-by: Abhijat Malviya <[email protected]>

Author: abhijat

helio

@@ -39,7 +39,7 @@ class TaskQueue {
     util::detail::ResultMover<ResultType> mover;
 
     ++blocked_submitters_;
-    Add([&mover, f = std::forward<F>(f), done](unsigned) mutable {
+    Add([&mover, f = std::forward<F>(f), done]() mutable {
       mover.Apply(f);
       done.Notify();
     });

src/core/task_queue.h

@@ -972,7 +972,7 @@ void DebugCmd::PopulateRangeFiber(uint64_t from, uint64_t num_of_keys,
     ++index;
 
     if (shard_batch.sz == 32) {
-      ess.Add(sid, [this, index, options, shard_batch](unsigned) {
+      ess.Add(sid, [this, index, options, shard_batch]() {
         DoPopulateBatch(options, shard_batch);
         if (index % 50 == 0) {
           ThisFiber::Yield();

src/server/debugcmd.cc

@@ -99,7 +99,7 @@ class EngineShardSet {
   template <typename U> void AwaitRunningOnShardQueue(U&& func) {
     util::fb2::BlockingCounter bc(size_);
     for (size_t i = 0; i < size_; ++i) {
-      Add(i, [&func, bc](unsigned) mutable {
+      Add(i, [&func, bc]() mutable {
         func(EngineShard::tlocal());
         bc->Dec();
       });

src/server/engine_shard_set.h

@@ -256,7 +256,7 @@ bool MultiCommandSquasher::ExecuteSquashed(facade::RedisReplyBuilder* rb) {
   atomic_uint64_t max_sched_cycles{0}, max_exec_cycles{0};
   base::SpinLock lock;
   uint64_t fiber_running_cycles{0}, proactor_running_cycles{0};
-  uint32_t max_sched_thread_id{0}, max_sched_seq_num{0}, max_task_index{0};
+  uint32_t max_sched_thread_id{0}, max_sched_seq_num{0};
   vector<string> past_fibers;
 
   // Atomic transactions (that have all keys locked) perform hops and run squashed commands via
@@ -273,7 +273,7 @@ bool MultiCommandSquasher::ExecuteSquashed(facade::RedisReplyBuilder* rb) {
 
     // Saves work in case logging is disable (i.e. log_squash_threshold_cached is high).
     const uint64_t min_threshold_cycles = CycleClock::FromUsec(log_squash_threshold_cached / 5);
-    auto cb = [&, bc, rb](unsigned task_index) mutable {
+    auto cb = [&, bc, rb]() mutable {
       uint64_t sched_time = CycleClock::Now() - start;
 
       // Update max_sched_cycles in lock-free fashion, to avoid contention
@@ -291,7 +291,6 @@ bool MultiCommandSquasher::ExecuteSquashed(facade::RedisReplyBuilder* rb) {
             max_sched_thread_id = ProactorBase::me()->GetPoolIndex();
             max_sched_seq_num = fb2::GetFiberRunSeq();
             past_fibers = fb2::GetPastFiberNames();
-            max_task_index = task_index;
           }
           break;
         }
@@ -356,10 +355,10 @@ bool MultiCommandSquasher::ExecuteSquashed(facade::RedisReplyBuilder* rb) {
 
     LOG_EVERY_T(INFO, 0.1)
         << "Squashed " << order_.size() << " commands. "
-        << "Total/Fanout/MaxSchedTime/ThreadCbTime/ThreadId/FiberCbTime/FiberSeq/TaskSeq/"
+        << "Total/Fanout/MaxSchedTime/ThreadCbTime/ThreadId/FiberCbTime/FiberSeq/"
         << "MaxExecTime: " << total_usec << "/" << num_shards_ << "/" << max_sched_usec << "/"
         << proactor_running_usec << "/" << max_sched_thread_id << "/" << fiber_running_usec << "/"
-        << "/" << max_sched_seq_num << "/" << max_task_index << "/" << max_exec_usec
+        << "/" << max_sched_seq_num << "/" << max_exec_usec
         << "\n past fibers: " << absl::StrJoin(past_fibers, ", ")
         << "\ncoordinator thread running time: "
         << CycleClock::ToUsec(ProactorBase::me()->GetCurrentBusyCycles());

src/server/multi_command_squasher.cc

@@ -2107,7 +2107,7 @@ error_code RdbLoader::Load(io::Source* src) {
         FlushShardAsync(i);
 
         // Active database if not existed before.
-        shard_set->Add(i, [dbid](unsigned) { GetCurrentDbSlice().ActivateDb(dbid); });
+        shard_set->Add(i, [dbid] { GetCurrentDbSlice().ActivateDb(dbid); });
       }
 
       cur_db_index_ = dbid;
@@ -2202,7 +2202,7 @@ void RdbLoader::FinishLoad(absl::Time start_time, size_t* keys_loaded) {
     FlushShardAsync(i);
 
     // Send sentinel callbacks to ensure that all previous messages have been processed.
-    shard_set->Add(i, [bc](unsigned) mutable { bc->Dec(); });
+    shard_set->Add(i, [bc]() mutable { bc->Dec(); });
   }
   bc->Wait();  // wait for sentinels to report.
   // Decrement local one if it exists
@@ -2523,7 +2523,7 @@ void RdbLoader::FlushShardAsync(ShardId sid) {
   if (out_buf.empty())
     return;
 
-  auto cb = [indx = this->cur_db_index_, this, ib = std::move(out_buf)](unsigned) {
+  auto cb = [indx = this->cur_db_index_, this, ib = std::move(out_buf)] {
     auto& db_slice = GetCurrentDbSlice();
 
     // Before we start loading, increment LoadInProgress.
@@ -2793,7 +2793,7 @@ bool RdbLoader::ShouldDiscardKey(std::string_view key, const ObjSettings& settin
 void RdbLoader::LoadScriptFromAux(string&& body) {
   ServerState* ss = ServerState::tlocal();
   auto interpreter = ss->BorrowInterpreter();
-  absl::Cleanup clean = [ss, interpreter]() { ss->ReturnInterpreter(interpreter); };
+  absl::Cleanup clean = [ss, interpreter] { ss->ReturnInterpreter(interpreter); };
 
   if (script_mgr_) {
     auto res = script_mgr_->Insert(body, interpreter);

src/server/rdb_load.cc

@@ -778,7 +778,7 @@ void Transaction::ScheduleInternal() {
         shard_set->Add(unique_shard_id_, &Transaction::ScheduleBatchInShard);
       }
     } else {
-      auto cb = [&schedule_ctx](unsigned) {
+      auto cb = [&schedule_ctx] {
         if (!schedule_ctx.trans->ScheduleInShard(EngineShard::tlocal(),
                                                  schedule_ctx.optimistic_execution)) {
           schedule_ctx.fail_cnt.fetch_add(1, memory_order_relaxed);
@@ -824,8 +824,7 @@ void Transaction::ScheduleInternal() {
     // See https://github.com/dragonflydb/dragonfly/issues/150 for more info.
     if (should_poll_execution.load(memory_order_relaxed)) {
       IterateActiveShards([](const auto& sd, auto i) {
-        shard_set->Add(
-            i, [](unsigned) { EngineShard::tlocal()->PollExecution("cancel_cleanup", nullptr); });
+        shard_set->Add(i, [] { EngineShard::tlocal()->PollExecution("cancel_cleanup", nullptr); });
       });
     }
     InitTxTime();  // update time for next scheduling attempt
@@ -855,7 +854,7 @@ void Transaction::UnlockMulti() {
   DCHECK_EQ(shard_data_.size(), shard_set->size());
   for (ShardId i = 0; i < shard_data_.size(); ++i) {
     vector<LockFp> fps = std::move(sharded_keys[i]);
-    shard_set->Add(i, [this, fps = std::move(fps)](unsigned) {
+    shard_set->Add(i, [this, fps = std::move(fps)] {
       this->UnlockMultiShardCb(fps, EngineShard::tlocal());
       intrusive_ptr_release(this);
     });
@@ -941,7 +940,7 @@ void Transaction::DispatchHop() {
 
   use_count_.fetch_add(run_cnt, memory_order_relaxed);  // for each pointer from poll_cb
 
-  auto poll_cb = [this](unsigned) {
+  auto poll_cb = [this] {
     CHECK(namespace_ != nullptr);
     EngineShard::tlocal()->PollExecution("exec_cb", this);
     DVLOG(3) << "ptr_release " << DebugId();
@@ -1015,7 +1014,7 @@ void Transaction::ExpireBlocking(WaitKeys wkeys) {
   DVLOG(1) << "ExpireBlocking " << DebugId();
   run_barrier_.Start(unique_shard_cnt_);
 
-  auto expire_cb = [this, &wkeys](unsigned) {
+  auto expire_cb = [this, &wkeys] {
     EngineShard* es = EngineShard::tlocal();
     if (wkeys) {
       IndexSlice is(0, 1);
@@ -1186,7 +1185,7 @@ bool Transaction::ScheduleInShard(EngineShard* shard, bool execute_optimistic) {
   return true;
 }
 
-void Transaction::ScheduleBatchInShard(unsigned) {
+void Transaction::ScheduleBatchInShard() {
   EngineShard* shard = EngineShard::tlocal();
   auto& stats = shard->stats();
   stats.tx_batch_schedule_calls_total++;

src/server/transaction.cc

@@ -515,7 +515,7 @@ class Transaction {
   bool ScheduleInShard(EngineShard* shard, bool execute_optimistic);
 
   // Optimized extension of ScheduleInShard. Pulls several transactions queued for scheduling.
-  static void ScheduleBatchInShard(unsigned);
+  static void ScheduleBatchInShard();
 
   // Set ARMED flags, start run barrier and submit poll tasks. Doesn't wait for the run barrier
   void DispatchHop();