mirrors/dragonfly
1
0
Fork 0
mirror of https://github.com/dragonflydb/dragonfly.git synced 2025-05-10 18:05:44 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions 5

fix: remove empty hop for non-expiring transactions (#1605)

Browse source 
Signed-off-by: Vladislav Oleshko <vlad@dragonflydb.io>
This commit is contained in:
Vladislav 2023-08-03 16:26:58 +03:00 • committed by GitHub
parent 3c345c1226
commit da17a39410
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
2 changed files with 23 additions and 29 deletions
Download patch file Download diff file Expand all files Collapse all files

45
src/server/transaction.cc
View file

@ -926,24 +926,20 @@ void Transaction::RunQuickie(EngineShard* shard) {
// runs in coordinator thread.
// Marks the transaction as expired and removes it from the waiting queue.
void Transaction::UnwatchBlocking(bool should_expire, WaitKeysProvider wcb) {
DVLOG(1) << "UnwatchBlocking " << DebugId() << " expire: " << should_expire;
void Transaction::ExpireBlocking(WaitKeysProvider wcb) {
DCHECK(!IsGlobal());
DVLOG(1) << "ExpireBlocking " << DebugId();
run_count_.store(unique_shard_cnt_, memory_order_release);
auto expire_cb = [this, &wcb, should_expire] {
auto expire_cb = [this, &wcb] {
EngineShard* es = EngineShard::tlocal();
ArgSlice wkeys = wcb(this, es);
UnwatchShardCb(wkeys, should_expire, es);
ExpireShardCb(wcb(this, es), es);
};
IterateActiveShards([&expire_cb](PerShardData& sd, auto i) { shard_set->Add(i, expire_cb); });
// Wait for all callbacks to conclude.
WaitForShardCallbacks();
DVLOG(1) << "UnwatchBlocking finished " << DebugId();
DVLOG(1) << "ExpireBlocking finished " << DebugId();
}
string_view Transaction::Name() const {
@ -1141,9 +1137,10 @@ bool Transaction::WaitOnWatch(const time_point& tp, WaitKeysProvider wkeys_provi
--stats->num_blocked_clients;
bool is_expired = (coordinator_state_ & COORD_CANCELLED) || status == cv_status::timeout;
UnwatchBlocking(is_expired, wkeys_provider);
coordinator_state_ &= ~COORD_BLOCKED;
if (is_expired)
ExpireBlocking(wkeys_provider);
coordinator_state_ &= ~COORD_BLOCKED;
return !is_expired;
}
@ -1165,23 +1162,19 @@ OpStatus Transaction::WatchInShard(ArgSlice keys, EngineShard* shard) {
return OpStatus::OK;
}
void Transaction::UnwatchShardCb(ArgSlice wkeys, bool should_expire, EngineShard* shard) {
if (should_expire) {
auto lock_args = GetLockArgs(shard->shard_id());
shard->db_slice().Release(Mode(), lock_args);
void Transaction::ExpireShardCb(ArgSlice wkeys, EngineShard* shard) {
auto lock_args = GetLockArgs(shard->shard_id());
shard->db_slice().Release(Mode(), lock_args);
unsigned sd_idx = SidToId(shard->shard_id());
auto& sd = shard_data_[sd_idx];
sd.local_mask |= EXPIRED_Q;
sd.local_mask &= ~KEYLOCK_ACQUIRED;
shard->blocking_controller()->FinalizeWatched(wkeys, this);
DCHECK(!shard->blocking_controller()->awakened_transactions().contains(this));
}
unsigned sd_idx = SidToId(shard->shard_id());
auto& sd = shard_data_[sd_idx];
sd.local_mask |= EXPIRED_Q;
sd.local_mask &= ~KEYLOCK_ACQUIRED;
// Need to see why I decided to call this.
// My guess - probably to trigger the run of stalled transactions in case
// this shard concurrently awoke this transaction and stalled the processing
// of TxQueue.
shard->blocking_controller()->FinalizeWatched(wkeys, this);
DCHECK(!shard->blocking_controller()->awakened_transactions().contains(this));
// Resume processing of transaction queue
shard->PollExecution("unwatchcb", nullptr);
CHECK_GE(DecreaseRunCnt(), 1u);