summaryrefslogtreecommitdiff
path: root/db/repl/rs_sync.cpp
diff options
context:
space:
mode:
Diffstat (limited to 'db/repl/rs_sync.cpp')
-rw-r--r--db/repl/rs_sync.cpp648
1 files changed, 0 insertions, 648 deletions
diff --git a/db/repl/rs_sync.cpp b/db/repl/rs_sync.cpp
deleted file mode 100644
index c86dbbbf2d6..00000000000
--- a/db/repl/rs_sync.cpp
+++ /dev/null
@@ -1,648 +0,0 @@
-/**
-* Copyright (C) 2008 10gen Inc.
-*
-* This program is free software: you can redistribute it and/or modify
-* it under the terms of the GNU Affero General Public License, version 3,
-* as published by the Free Software Foundation.
-*
-* This program is distributed in the hope that it will be useful,
-* but WITHOUT ANY WARRANTY; without even the implied warranty of
-* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
-* GNU Affero General Public License for more details.
-*
-* You should have received a copy of the GNU Affero General Public License
-* along with this program. If not, see <http://www.gnu.org/licenses/>.
-*/
-
-#include "pch.h"
-#include "../client.h"
-#include "../../client/dbclient.h"
-#include "rs.h"
-#include "../repl.h"
-#include "connections.h"
-
-namespace mongo {
-
- using namespace bson;
- extern unsigned replSetForceInitialSyncFailure;
-
- void NOINLINE_DECL blank(const BSONObj& o) {
- if( *o.getStringField("op") != 'n' ) {
- log() << "replSet skipping bad op in oplog: " << o.toString() << rsLog;
- }
- }
-
- /* apply the log op that is in param o
- @return bool failedUpdate
- */
- bool ReplSetImpl::syncApply(const BSONObj &o) {
- const char *ns = o.getStringField("ns");
- if ( *ns == '.' || *ns == 0 ) {
- blank(o);
- return false;
- }
-
- Client::Context ctx(ns);
- ctx.getClient()->curop()->reset();
- return applyOperation_inlock(o);
- }
-
- /* initial oplog application, during initial sync, after cloning.
- @return false on failure.
- this method returns an error and doesn't throw exceptions (i think).
- */
- bool ReplSetImpl::initialSyncOplogApplication(
- const Member *source,
- OpTime applyGTE,
- OpTime minValid) {
- if( source == 0 ) return false;
-
- const string hn = source->h().toString();
- OplogReader r;
-
- try {
- if( !r.connect(hn) ) {
- log() << "replSet initial sync error can't connect to " << hn << " to read " << rsoplog << rsLog;
- return false;
- }
-
- r.tailingQueryGTE( rsoplog, applyGTE );
- if ( !r.haveCursor() ) {
- log() << "replSet initial sync oplog query error" << rsLog;
- return false;
- }
-
- {
- if( !r.more() ) {
- sethbmsg("replSet initial sync error reading remote oplog");
- log() << "replSet initial sync error remote oplog (" << rsoplog << ") on host " << hn << " is empty?" << rsLog;
- return false;
- }
- bo op = r.next();
- OpTime t = op["ts"]._opTime();
- r.putBack(op);
-
- if( op.firstElementFieldName() == string("$err") ) {
- log() << "replSet initial sync error querying " << rsoplog << " on " << hn << " : " << op.toString() << rsLog;
- return false;
- }
-
- uassert( 13508 , str::stream() << "no 'ts' in first op in oplog: " << op , !t.isNull() );
- if( t > applyGTE ) {
- sethbmsg(str::stream() << "error " << hn << " oplog wrapped during initial sync");
- log() << "replSet initial sync expected first optime of " << applyGTE << rsLog;
- log() << "replSet initial sync but received a first optime of " << t << " from " << hn << rsLog;
- return false;
- }
-
- sethbmsg(str::stream() << "initial oplog application from " << hn << " starting at "
- << t.toStringPretty() << " to " << minValid.toStringPretty());
- }
- }
- catch(DBException& e) {
- log() << "replSet initial sync failing: " << e.toString() << rsLog;
- return false;
- }
-
- /* we lock outside the loop to avoid the overhead of locking on every operation. */
- writelock lk("");
-
- // todo : use exhaust
- OpTime ts;
- time_t start = time(0);
- unsigned long long n = 0;
- while( 1 ) {
- try {
- if( !r.more() )
- break;
- BSONObj o = r.nextSafe(); /* note we might get "not master" at some point */
- ts = o["ts"]._opTime();
-
- {
- if( (source->state() != MemberState::RS_PRIMARY &&
- source->state() != MemberState::RS_SECONDARY) ||
- replSetForceInitialSyncFailure ) {
-
- int f = replSetForceInitialSyncFailure;
- if( f > 0 ) {
- replSetForceInitialSyncFailure = f-1;
- log() << "replSet test code invoked, replSetForceInitialSyncFailure" << rsLog;
- throw DBException("forced error",0);
- }
- log() << "replSet we are now primary" << rsLog;
- throw DBException("primary changed",0);
- }
-
- if( ts >= applyGTE ) { // optimes before we started copying need not be applied.
- bool failedUpdate = syncApply(o);
- if( failedUpdate && shouldRetry(o, hn)) {
- failedUpdate = syncApply(o);
- uassert(15915, "replSet update still fails after adding missing object", !failedUpdate);
- }
- }
- _logOpObjRS(o); /* with repl sets we write the ops to our oplog too */
- }
-
- if ( ++n % 1000 == 0 ) {
- time_t now = time(0);
- if (now - start > 10) {
- // simple progress metering
- log() << "replSet initialSyncOplogApplication applied " << n << " operations, synced to "
- << ts.toStringPretty() << rsLog;
- start = now;
- }
- }
-
- if ( ts > minValid ) {
- break;
- }
-
- getDur().commitIfNeeded();
- }
- catch (DBException& e) {
- // skip duplicate key exceptions
- if( e.getCode() == 11000 || e.getCode() == 11001 || e.getCode() == 12582) {
- continue;
- }
-
- // handle cursor not found (just requery)
- if( e.getCode() == 13127 ) {
- r.resetCursor();
- r.tailingQueryGTE(rsoplog, ts);
- if( r.haveCursor() ) {
- continue;
- }
- }
-
- // TODO: handle server restart
-
- if( ts <= minValid ) {
- // didn't make it far enough
- log() << "replSet initial sync failing, error applying oplog " << e.toString() << rsLog;
- return false;
- }
-
- // otherwise, whatever
- break;
- }
- }
- return true;
- }
-
- /* should be in RECOVERING state on arrival here.
- readlocks
- @return true if transitioned to SECONDARY
- */
- bool ReplSetImpl::tryToGoLiveAsASecondary(OpTime& /*out*/ minvalid) {
- bool golive = false;
-
- {
- lock lk( this );
-
- if (_maintenanceMode > 0) {
- // we're not actually going live
- return true;
- }
- }
-
- {
- readlock lk("local.replset.minvalid");
- BSONObj mv;
- if( Helpers::getSingleton("local.replset.minvalid", mv) ) {
- minvalid = mv["ts"]._opTime();
- if( minvalid <= lastOpTimeWritten ) {
- golive=true;
- }
- }
- else
- golive = true; /* must have been the original member */
- }
- if( golive ) {
- sethbmsg("");
- changeState(MemberState::RS_SECONDARY);
- }
- return golive;
- }
-
- bool ReplSetImpl::_isStale(OplogReader& r, const string& hn) {
- BSONObj remoteOldestOp = r.findOne(rsoplog, Query());
- OpTime ts = remoteOldestOp["ts"]._opTime();
- DEV log() << "replSet remoteOldestOp: " << ts.toStringLong() << rsLog;
- else LOG(3) << "replSet remoteOldestOp: " << ts.toStringLong() << rsLog;
- DEV {
- log() << "replSet lastOpTimeWritten: " << lastOpTimeWritten.toStringLong() << rsLog;
- log() << "replSet our state: " << state().toString() << rsLog;
- }
- if( lastOpTimeWritten >= ts ) {
- return false;
- }
-
- // we're stale
- log() << "replSet error RS102 too stale to catch up, at least from " << hn << rsLog;
- log() << "replSet our last optime : " << lastOpTimeWritten.toStringLong() << rsLog;
- log() << "replSet oldest at " << hn << " : " << ts.toStringLong() << rsLog;
- log() << "replSet See http://www.mongodb.org/display/DOCS/Resyncing+a+Very+Stale+Replica+Set+Member" << rsLog;
-
- // reset minvalid so that we can't become primary prematurely
- {
- writelock lk("local.replset.minvalid");
- Helpers::putSingleton("local.replset.minvalid", remoteOldestOp);
- }
-
- sethbmsg("error RS102 too stale to catch up");
- changeState(MemberState::RS_RECOVERING);
- sleepsecs(120);
- return true;
- }
-
- /**
- * Tries to connect the oplog reader to a potential sync source. If
- * successful, it checks that we are not stale compared to this source.
- *
- * @param r reader to populate
- * @param hn hostname to try
- *
- * @return if both checks pass, it returns true, otherwise false.
- */
- bool ReplSetImpl::_getOplogReader(OplogReader& r, string& hn) {
- assert(r.conn() == 0);
-
- if( !r.connect(hn) ) {
- LOG(2) << "replSet can't connect to " << hn << " to read operations" << rsLog;
- r.resetConnection();
- return false;
- }
- if( _isStale(r, hn)) {
- r.resetConnection();
- return false;
- }
- return true;
- }
-
- /* tail an oplog. ok to return, will be re-called. */
- void ReplSetImpl::syncTail() {
- // todo : locking vis a vis the mgr...
- OplogReader r;
- string hn;
- const Member *target = 0;
-
- // if we cannot reach the master but someone else is more up-to-date
- // than we are, sync from them.
- target = getMemberToSyncTo();
- if (target != 0) {
- hn = target->h().toString();
- if (!_getOplogReader(r, hn)) {
- // we might be stale wrt the primary, but could still sync from
- // a secondary
- target = 0;
- }
- }
-
- // no server found
- if (target == 0) {
- // if there is no one to sync from
- OpTime minvalid;
- tryToGoLiveAsASecondary(minvalid);
- return;
- }
-
- r.tailingQueryGTE(rsoplog, lastOpTimeWritten);
- // if target cut connections between connecting and querying (for
- // example, because it stepped down) we might not have a cursor
- if ( !r.haveCursor() ) {
- return;
- }
-
- uassert(1000, "replSet source for syncing doesn't seem to be await capable -- is it an older version of mongodb?", r.awaitCapable() );
-
- {
- if( !r.more() ) {
- /* maybe we are ahead and need to roll back? */
- try {
- bo theirLastOp = r.getLastOp(rsoplog);
- if( theirLastOp.isEmpty() ) {
- log() << "replSet error empty query result from " << hn << " oplog" << rsLog;
- sleepsecs(2);
- return;
- }
- OpTime theirTS = theirLastOp["ts"]._opTime();
- if( theirTS < lastOpTimeWritten ) {
- log() << "replSet we are ahead of the primary, will try to roll back" << rsLog;
- syncRollback(r);
- return;
- }
- /* we're not ahead? maybe our new query got fresher data. best to come back and try again */
- log() << "replSet syncTail condition 1" << rsLog;
- sleepsecs(1);
- }
- catch(DBException& e) {
- log() << "replSet error querying " << hn << ' ' << e.toString() << rsLog;
- sleepsecs(2);
- }
- return;
- }
-
- BSONObj o = r.nextSafe();
- OpTime ts = o["ts"]._opTime();
- long long h = o["h"].numberLong();
- if( ts != lastOpTimeWritten || h != lastH ) {
- log() << "replSet our last op time written: " << lastOpTimeWritten.toStringPretty() << rsLog;
- log() << "replset source's GTE: " << ts.toStringPretty() << rsLog;
- syncRollback(r);
- return;
- }
- }
-
- /* we have now checked if we need to rollback and we either don't have to or did it. */
- {
- OpTime minvalid;
- tryToGoLiveAsASecondary(minvalid);
- }
-
- while( 1 ) {
- while( 1 ) {
- if( !r.moreInCurrentBatch() ) {
- /* we need to occasionally check some things. between
- batches is probably a good time. */
-
- /* perhaps we should check this earlier? but not before the rollback checks. */
- if( state().recovering() ) {
- /* can we go to RS_SECONDARY state? we can if not too old and if minvalid achieved */
- OpTime minvalid;
- bool golive = ReplSetImpl::tryToGoLiveAsASecondary(minvalid);
- if( golive ) {
- ;
- }
- else {
- sethbmsg(str::stream() << "still syncing, not yet to minValid optime" << minvalid.toString());
- }
-
- /* todo: too stale capability */
- }
-
- if( !target->hbinfo().hbstate.readable() ) {
- return;
- }
- }
- if( !r.more() )
- break;
- {
- BSONObj o = r.nextSafe(); /* note we might get "not master" at some point */
-
- int sd = myConfig().slaveDelay;
- // ignore slaveDelay if the box is still initializing. once
- // it becomes secondary we can worry about it.
- if( sd && box.getState().secondary() ) {
- const OpTime ts = o["ts"]._opTime();
- long long a = ts.getSecs();
- long long b = time(0);
- long long lag = b - a;
- long long sleeptime = sd - lag;
- if( sleeptime > 0 ) {
- uassert(12000, "rs slaveDelay differential too big check clocks and systems", sleeptime < 0x40000000);
- if( sleeptime < 60 ) {
- sleepsecs((int) sleeptime);
- }
- else {
- log() << "replSet slavedelay sleep long time: " << sleeptime << rsLog;
- // sleep(hours) would prevent reconfigs from taking effect & such!
- long long waitUntil = b + sleeptime;
- while( 1 ) {
- sleepsecs(6);
- if( time(0) >= waitUntil )
- break;
-
- if( !target->hbinfo().hbstate.readable() ) {
- break;
- }
-
- if( myConfig().slaveDelay != sd ) // reconf
- break;
- }
- }
- }
-
- }
-
- try {
- writelock lk("");
-
- /* if we have become primary, we dont' want to apply things from elsewhere
- anymore. assumePrimary is in the db lock so we are safe as long as
- we check after we locked above. */
- if( box.getState().primary() ) {
- log(0) << "replSet stopping syncTail we are now primary" << rsLog;
- return;
- }
-
- syncApply(o);
- _logOpObjRS(o); // with repl sets we write the ops to our oplog too
- }
- catch (DBException& e) {
- sethbmsg(str::stream() << "syncTail: " << e.toString() << ", syncing: " << o);
- sleepsecs(30);
- return;
- }
- }
- }
- r.tailCheck();
- if( !r.haveCursor() ) {
- LOG(1) << "replSet end syncTail pass with " << hn << rsLog;
- // TODO : reuse our connection to the primary.
- return;
- }
-
- if( !target->hbinfo().hbstate.readable() ) {
- return;
- }
- // looping back is ok because this is a tailable cursor
- }
- }
-
- void ReplSetImpl::_syncThread() {
- StateBox::SP sp = box.get();
- if( sp.state.primary() ) {
- sleepsecs(1);
- return;
- }
- if( _blockSync || sp.state.fatal() || sp.state.startup() ) {
- sleepsecs(5);
- return;
- }
-
- /* do we have anything at all? */
- if( lastOpTimeWritten.isNull() ) {
- syncDoInitialSync();
- return; // _syncThread will be recalled, starts from top again in case sync failed.
- }
-
- /* we have some data. continue tailing. */
- syncTail();
- }
-
- void ReplSetImpl::syncThread() {
- while( 1 ) {
- // After a reconfig, we may not be in the replica set anymore, so
- // check that we are in the set (and not an arbiter) before
- // trying to sync with other replicas.
- if( ! _self ) {
- log() << "replSet warning did not detect own host and port, not syncing, config: " << theReplSet->config() << rsLog;
- return;
- }
- if( myConfig().arbiterOnly ) {
- return;
- }
-
- try {
- _syncThread();
- }
- catch(DBException& e) {
- sethbmsg(str::stream() << "syncThread: " << e.toString());
- sleepsecs(10);
- }
- catch(...) {
- sethbmsg("unexpected exception in syncThread()");
- // TODO : SET NOT SECONDARY here?
- sleepsecs(60);
- }
- sleepsecs(1);
-
- /* normally msgCheckNewState gets called periodically, but in a single node repl set there
- are no heartbeat threads, so we do it here to be sure. this is relevant if the singleton
- member has done a stepDown() and needs to come back up.
- */
- OCCASIONALLY {
- mgr->send( boost::bind(&Manager::msgCheckNewState, theReplSet->mgr) );
- }
- }
- }
-
- void startSyncThread() {
- static int n;
- if( n != 0 ) {
- log() << "replSet ERROR : more than one sync thread?" << rsLog;
- assert( n == 0 );
- }
- n++;
-
- Client::initThread("rsSync");
- cc().iAmSyncThread(); // for isSyncThread() (which is used not used much, is used in secondary create index code
- replLocalAuth();
- theReplSet->syncThread();
- cc().shutdown();
- }
-
- void GhostSync::starting() {
- Client::initThread("rsGhostSync");
- replLocalAuth();
- }
-
- void ReplSetImpl::blockSync(bool block) {
- _blockSync = block;
- if (_blockSync) {
- // syncing is how we get into SECONDARY state, so we'll be stuck in
- // RECOVERING until we unblock
- changeState(MemberState::RS_RECOVERING);
- }
- }
-
- void GhostSync::associateSlave(const BSONObj& id, const int memberId) {
- const OID rid = id["_id"].OID();
- rwlock lk( _lock , true );
- GhostSlave &slave = _ghostCache[rid];
- if (slave.init) {
- LOG(1) << "tracking " << slave.slave->h().toString() << " as " << rid << rsLog;
- return;
- }
-
- slave.slave = (Member*)rs->findById(memberId);
- if (slave.slave != 0) {
- slave.init = true;
- }
- else {
- log() << "replset couldn't find a slave with id " << memberId
- << ", not tracking " << rid << rsLog;
- }
- }
-
- void GhostSync::updateSlave(const mongo::OID& rid, const OpTime& last) {
- rwlock lk( _lock , false );
- MAP::iterator i = _ghostCache.find( rid );
- if ( i == _ghostCache.end() ) {
- OCCASIONALLY warning() << "couldn't update slave " << rid << " no entry" << rsLog;
- return;
- }
-
- GhostSlave& slave = i->second;
- if (!slave.init) {
- OCCASIONALLY log() << "couldn't update slave " << rid << " not init" << rsLog;
- return;
- }
-
- ((ReplSetConfig::MemberCfg)slave.slave->config()).updateGroups(last);
- }
-
- void GhostSync::percolate(const BSONObj& id, const OpTime& last) {
- const OID rid = id["_id"].OID();
- GhostSlave* slave;
- {
- rwlock lk( _lock , false );
-
- MAP::iterator i = _ghostCache.find( rid );
- if ( i == _ghostCache.end() ) {
- OCCASIONALLY log() << "couldn't percolate slave " << rid << " no entry" << rsLog;
- return;
- }
-
- slave = &(i->second);
- if (!slave->init) {
- OCCASIONALLY log() << "couldn't percolate slave " << rid << " not init" << rsLog;
- return;
- }
- }
-
- assert(slave->slave);
-
- const Member *target = rs->_currentSyncTarget;
- if (!target || rs->box.getState().primary()
- // we are currently syncing from someone who's syncing from us
- // the target might end up with a new Member, but s.slave never
- // changes so we'll compare the names
- || target == slave->slave || target->fullName() == slave->slave->fullName()) {
- LOG(1) << "replica set ghost target no good" << endl;
- return;
- }
-
- try {
- if (!slave->reader.haveCursor()) {
- if (!slave->reader.connect(id, slave->slave->id(), target->fullName())) {
- // error message logged in OplogReader::connect
- return;
- }
- slave->reader.ghostQueryGTE(rsoplog, last);
- }
-
- LOG(1) << "replSet last: " << slave->last.toString() << " to " << last.toString() << rsLog;
- if (slave->last > last) {
- return;
- }
-
- while (slave->last <= last) {
- if (!slave->reader.more()) {
- // we'll be back
- return;
- }
-
- BSONObj o = slave->reader.nextSafe();
- slave->last = o["ts"]._opTime();
- }
- LOG(2) << "now last is " << slave->last.toString() << rsLog;
- }
- catch (DBException& e) {
- // we'll be back
- LOG(2) << "replSet ghost sync error: " << e.what() << " for "
- << slave->slave->fullName() << rsLog;
- slave->reader.resetConnection();
- }
- }
-}
generated by cgit v1.2.3 (git 2.39.1) at 2026-09-11 14:34:32 +0000