Skip to content

CBL-8580: Use AtomicRetained in Replicator::Subreplicators #2517

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jianminzhao merged 1 commit into from
Jul 1, 2026
+23 −16
Merged

CBL-8580: Use AtomicRetained in Replicator::Subreplicators #2517

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
33 changes: 20 additions & 13 deletions Replicator/Replicator.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -252,9 +252,13 @@ namespace litecore::repl {

// Called after the checkpoint is established.
void Replicator::startReplicating(CollectionIndex coll) {
if ( _options->push(coll) > kC4Passive ) _subRepls[coll].pusher->start();
if ( _options->pull(coll) > kC4Passive )
_subRepls[coll].puller->start(_subRepls[coll].checkpointer->remoteMinSequence());
if ( _options->push(coll) > kC4Passive ) {
if ( auto pusher = _subRepls[coll].pusher.get() ) pusher->start();
}
if ( _options->pull(coll) > kC4Passive ) {
if ( auto puller = _subRepls[coll].puller.get() )
puller->start(_subRepls[coll].checkpointer->remoteMinSequence());
}
}

pair<int, websocket::Headers> Replicator::httpResponse() const {
Expand Down Expand Up @@ -379,9 +383,9 @@ namespace litecore::repl {

CollectionIndex coll = task->collectionIndex();
if ( coll != kNotCollectionIndex ) {
if ( task == _subRepls[coll].pusher ) {
if ( auto pusher = _subRepls[coll].pusher.get(); pusher == task ) {
updatePushStatus(coll, taskStatus);
} else if ( task == _subRepls[coll].puller ) {
} else if ( auto puller = _subRepls[coll].puller.get(); puller == task ) {
updatePullStatus(coll, taskStatus);
}
}
Expand Down Expand Up @@ -656,8 +660,8 @@ namespace litecore::repl {
// Clear connection() and notify the other agents to do the same:
_connectionClosed();
for ( auto& sub : _subRepls ) {
if ( sub.pusher ) sub.pusher->connectionClosed();
if ( sub.puller ) sub.puller->connectionClosed();
if ( auto pusher = sub.pusher.get() ) pusher->connectionClosed();
if ( auto puller = sub.puller.get() ) puller->connectionClosed();
}

if ( status.isNormal() && closedByPeer && _options->isActive() ) {
Expand Down Expand Up @@ -718,9 +722,10 @@ namespace litecore::repl {
cLogInfo(coll, "No local checkpoint '%.*s'", SPLAT(sub.checkpointer->initialCheckpointID()));
// If pulling into an empty db with no checkpoint, it's safe to skip deleted
// revisions as an optimization.
if ( _options->pull(coll) > kC4Passive && sub.puller
if ( auto puller = sub.puller.get();
puller && _options->pull(coll) > kC4Passive
&& _db->useCollection(sub.collectionSpec)->getLastSequence() == 0_seq )
sub.puller->setSkipDeleted();
puller->setSkipDeleted();
}
return true;
} catch ( ... ) {
Expand Down Expand Up @@ -767,8 +772,9 @@ namespace litecore::repl {

if ( !refresh && sub.hadLocalCheckpoint ) {
// Compare checkpoints, reset if mismatched:
bool valid = sub.checkpointer->validateWith(remoteCheckpoint);
if ( !valid && sub.pusher ) sub.pusher->checkpointIsInvalid();
if ( !sub.checkpointer->validateWith(remoteCheckpoint) ) {
if ( auto pusher = sub.pusher.get() ) pusher->checkpointIsInvalid();
}

if ( !refresh ) {
// Now we have the checkpoints! Time to start replicating:
Expand Down Expand Up @@ -898,8 +904,9 @@ namespace litecore::repl {

if ( _subRepls[i].hadLocalCheckpoint ) {
// Compare checkpoints, reset if mismatched:
bool valid = _subRepls[i].checkpointer->validateWith(remoteCheckpoints[i]);
if ( !valid && _subRepls[i].pusher ) _subRepls[i].pusher->checkpointIsInvalid();
if ( !_subRepls[i].checkpointer->validateWith(remoteCheckpoints[i]) ) {
if ( auto pusher = _subRepls[i].pusher.get() ) pusher->checkpointIsInvalid();
}
}
// Now we have the checkpoints! Time to start replicating:
startReplicating(i);
Expand Down
6 changes: 3 additions & 3 deletions Replicator/Replicator.hh
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,14 +11,14 @@
//

#pragma once
#include "AtomicRetained.hh"
#include "Worker.hh"
#include "Checkpointer.hh"
#include "BLIPConnection.hh"
#include "Batcher.hh"
#include "Stopwatch.hh"
#include "c4DatabaseTypes.h"
#include <access_lock.hh>
#include <array>
#include <optional>
#include <utility>

Expand Down Expand Up @@ -232,8 +232,8 @@ namespace litecore::repl {
// Member variables:

struct SubReplicator {
Retained<Pusher> pusher;
Retained<Puller> puller;
AtomicRetained<Pusher> pusher;
AtomicRetained<Puller> puller;
Status pushStatus; // Current status of Pusher
Status pullStatus; // Current status of Puller
unique_ptr<Checkpointer> checkpointer; // Object that manages checkpoints
Expand Down
Loading