chore: reduce frequency of clock invocations (#5946)

Partially addresses #5942 as now squashing transactions read clock only once.

Moreover for most transactions we called clock twice - in the c'tor and during scheduling.
This change removes the scheduling override unless transaction is repeated.

Signed-off-by: Roman Gershman <[email protected]>

Author: romange

src/server/generic_family.cc

@@ -960,7 +960,10 @@ OpResult<uint64_t> OpTtl(Transaction* t, EngineShard* shard, string_view key) {
   auto opExpireTimeResult = OpExpireTime(t, shard, key);
 
   if (opExpireTimeResult) {
-    int64_t ttl_ms = opExpireTimeResult.value() - t->GetDbContext().time_now_ms;
+    auto now = t->GetDbContext().time_now_ms;
+    DCHECK_GT(now, 0u);
+
+    int64_t ttl_ms = opExpireTimeResult.value() - now;
     DCHECK_GT(ttl_ms, 0);  // Otherwise FindReadOnly would return null.
     return ttl_ms;
   } else {
@@ -1898,6 +1901,7 @@ void GenericFamily::Time(CmdArgList args, const CommandContext& cmd_cntx) {
   } else {
     now_usec = absl::GetCurrentTimeNanos() / 1000;
   }
+  DCHECK_GT(now_usec, 0u);
 
   auto* rb = static_cast<RedisReplyBuilder*>(cmd_cntx.rb);
   rb->StartArray(2);

src/server/server_family.cc

@@ -3886,7 +3886,7 @@ void ServerFamily::Role(CmdArgList args, const CommandContext& cmd_cntx) {
     rb->StartArray(4 + cluster_replicas_.size() * 3);
     rb->SendBulkString(GetFlag(FLAGS_info_replication_valkey_compatible) ? "slave" : "replica");
 
-    auto send_replica_info = [rb](Replica::Summary rinfo) {
+    auto send_replica_info = [rb](const Replica::Summary& rinfo) {
       rb->SendBulkString(rinfo.host);
       rb->SendBulkString(absl::StrCat(rinfo.port));
       if (rinfo.full_sync_done) {
@@ -3908,7 +3908,7 @@ void ServerFamily::Role(CmdArgList args, const CommandContext& cmd_cntx) {
 }
 
 void ServerFamily::Script(CmdArgList args, const CommandContext& cmd_cntx) {
-  script_mgr_->Run(std::move(args), cmd_cntx.tx, cmd_cntx.rb, cmd_cntx.conn_cntx);
+  script_mgr_->Run(args, cmd_cntx.tx, cmd_cntx.rb, cmd_cntx.conn_cntx);
 }
 
 void ServerFamily::LastSave(CmdArgList args, const CommandContext& cmd_cntx) {

src/server/stream_family.cc

@@ -2808,8 +2808,10 @@ void StreamFamily::XClaim(CmdArgList args, const CommandContext& cmd_cntx) {
   if (!ParseXclaimOptions(args, opts, cmd_cntx.rb))
     return;
 
-  if (uint64_t now = cmd_cntx.tx->GetDbContext().time_now_ms;
-      opts.delivery_time < 0 || static_cast<uint64_t>(opts.delivery_time) > now)
+  uint64_t now = cmd_cntx.tx->GetDbContext().time_now_ms;
+  DCHECK_GT(now, 0u);
+
+  if (opts.delivery_time < 0 || static_cast<uint64_t>(opts.delivery_time) > now)
     opts.delivery_time = now;
 
   auto cb = [&](Transaction* t, EngineShard* shard) {

src/server/stream_family_test.cc

@@ -599,9 +599,11 @@ TEST_F(StreamFamilyTest, Xclaim) {
   resp = Run({"xclaim", "foo", "group", "alice", "0", "1-4", "TIME",
               absl::StrCat(TEST_current_time_ms - 500), "justid"});
   EXPECT_THAT(resp.GetString(), "1-4");
+
   // min idle time is exceeded for this entry
   resp = Run({"xclaim", "foo", "group", "bob", "600", "1-4"});
-  EXPECT_THAT(resp, ArrLen(0));
+  ASSERT_THAT(resp, ArrLen(0));
+
   resp = Run({"xclaim", "foo", "group", "bob", "400", "1-4", "justid"});
   EXPECT_THAT(resp.GetString(), "1-4");
 

src/server/transaction.cc

@@ -542,6 +542,8 @@ void Transaction::MultiUpdateWithParent(const Transaction* parent) {
   // DCHECK(multi_);
   // DCHECK(parent->multi_);  // it might not be a squasher yet, but certainly is multi
   DCHECK_EQ(multi_->role, SQUASHED_STUB);
+  DCHECK(parent->time_now_ms_);
+
   txid_ = parent->txid_;
   time_now_ms_ = parent->time_now_ms_;
   unique_slot_checker_ = parent->unique_slot_checker_;
@@ -752,8 +754,6 @@ void Transaction::ScheduleInternal() {
     if (unique_shard_cnt_ > 1)
       txid_ = op_seq.fetch_add(1, memory_order_relaxed);
 
-    InitTxTime();
-
     run_barrier_.Start(unique_shard_cnt_);
 
     if (CanRunInlined()) {
@@ -828,6 +828,7 @@ void Transaction::ScheduleInternal() {
             i, [](unsigned) { EngineShard::tlocal()->PollExecution("cancel_cleanup", nullptr); });
       });
     }
+    InitTxTime();  // update time for next scheduling attempt
   }
 
   coordinator_state_ |= COORD_SCHED;