/** * 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 . */ /* clientcursor.cpp ClientCursor is a wrapper that represents a cursorid from our database application's perspective. Cursor -- and its derived classes -- are our internal cursors. */ #include "pch.h" #include "query.h" #include "introspect.h" #include #include "db.h" #include "commands.h" #include "repl_block.h" namespace mongo { typedef multimap CCByLoc; CCById ClientCursor::clientCursorsById; boost::recursive_mutex ClientCursor::ccmutex; long long ClientCursor::numberTimedOut = 0; void aboutToDeleteForSharding( const Database* db , const DiskLoc& dl ); // from s/d_logic.h /*static*/ void ClientCursor::assertNoCursors() { recursive_scoped_lock lock(ccmutex); if( clientCursorsById.size() ) { log() << "ERROR clientcursors exist but should not at this point" << endl; ClientCursor *cc = clientCursorsById.begin()->second; log() << "first one: " << cc->cursorid << ' ' << cc->ns << endl; clientCursorsById.clear(); assert(false); } } void ClientCursor::setLastLoc_inlock(DiskLoc L) { assert( pos != -2 ); // defensive - see ~ClientCursor if ( L == _lastLoc ) return; CCByLoc& bl = byLoc(); if ( !_lastLoc.isNull() ) { CCByLoc::iterator i = kv_find(bl, _lastLoc, this); if ( i != bl.end() ) bl.erase(i); } if ( !L.isNull() ) bl.insert( make_pair(L, this) ); _lastLoc = L; } /* ------------------------------------------- */ /* must call this when a btree node is updated */ //void removedKey(const DiskLoc& btreeLoc, int keyPos) { //} /* todo: this implementation is incomplete. we use it as a prefix for dropDatabase, which works fine as the prefix will end with '.'. however, when used with drop and dropIndexes, this could take out cursors that belong to something else -- if you drop "foo", currently, this will kill cursors for "foobar". */ void ClientCursor::invalidate(const char *nsPrefix) { vector toDelete; int len = strlen(nsPrefix); assert( len > 0 && strchr(nsPrefix, '.') ); { //cout << "\nTEMP invalidate " << nsPrefix << endl; recursive_scoped_lock lock(ccmutex); Database *db = cc().database(); assert(db); assert( str::startsWith(nsPrefix, db->name) ); for( CCById::iterator i = clientCursorsById.begin(); i != clientCursorsById.end(); ++i ) { ClientCursor *cc = i->second; if( cc->_db != db ) continue; if ( strncmp(nsPrefix, cc->ns.c_str(), len) == 0 ) { toDelete.push_back(i->second); } } /* note : we can't iterate byloc because clientcursors may exist with a loc of null in which case they are not in the map. perhaps they should not exist though in the future? something to change??? CCByLoc& bl = db->ccByLoc; for ( CCByLoc::iterator i = bl.begin(); i != bl.end(); ++i ) { ClientCursor *cc = i->second; if ( strncmp(nsPrefix, cc->ns.c_str(), len) == 0 ) { assert( cc->_db == db ); toDelete.push_back(i->second); } }*/ for ( vector::iterator i = toDelete.begin(); i != toDelete.end(); ++i ) delete (*i); /*cout << "TEMP after invalidate " << endl; for( auto i = clientCursorsById.begin(); i != clientCursorsById.end(); ++i ) { cout << " " << i->second->ns << endl; } cout << "TEMP after invalidate done" << endl;*/ } } bool ClientCursor::shouldTimeout( unsigned millis ){ _idleAgeMillis += millis; return _idleAgeMillis > 600000 && _pinValue == 0; } /* called every 4 seconds. millis is amount of idle time passed since the last call -- could be zero */ void ClientCursor::idleTimeReport(unsigned millis) { readlock lk(""); recursive_scoped_lock lock(ccmutex); for ( CCById::iterator i = clientCursorsById.begin(); i != clientCursorsById.end(); ) { CCById::iterator j = i; i++; if( j->second->shouldTimeout( millis ) ){ numberTimedOut++; log(1) << "killing old cursor " << j->second->cursorid << ' ' << j->second->ns << " idle:" << j->second->idleTime() << "ms\n"; delete j->second; } } } /* must call when a btree bucket going away. note this is potentially slow */ void ClientCursor::informAboutToDeleteBucket(const DiskLoc& b) { recursive_scoped_lock lock(ccmutex); Database *db = cc().database(); CCByLoc& bl = db->ccByLoc; RARELY if ( bl.size() > 70 ) { log() << "perf warning: byLoc.size=" << bl.size() << " in aboutToDeleteBucket\n"; } for ( CCByLoc::iterator i = bl.begin(); i != bl.end(); i++ ) i->second->c->aboutToDeleteBucket(b); } void aboutToDeleteBucket(const DiskLoc& b) { ClientCursor::informAboutToDeleteBucket(b); } /* must call this on a delete so we clean up the cursors. */ void ClientCursor::aboutToDelete(const DiskLoc& dl) { recursive_scoped_lock lock(ccmutex); Database *db = cc().database(); assert(db); aboutToDeleteForSharding( db , dl ); CCByLoc& bl = db->ccByLoc; CCByLoc::iterator j = bl.lower_bound(dl); CCByLoc::iterator stop = bl.upper_bound(dl); if ( j == stop ) return; vector toAdvance; while ( 1 ) { toAdvance.push_back(j->second); DEV assert( j->first == dl ); ++j; if ( j == stop ) break; } wassert( toAdvance.size() < 5000 ); for ( vector::iterator i = toAdvance.begin(); i != toAdvance.end(); ++i ){ ClientCursor* cc = *i; wassert(cc->_db == db); if ( cc->_doingDeletes ) continue; Cursor *c = cc->c.get(); if ( c->capped() ){ delete cc; continue; } c->checkLocation(); DiskLoc tmp1 = c->refLoc(); if ( tmp1 != dl ) { /* this might indicate a failure to call ClientCursor::updateLocation() */ problem() << "warning: cursor loc " << tmp1 << " does not match byLoc position " << dl << " !" << endl; } c->advance(); if ( c->eof() ) { // advanced to end // leave ClientCursor in place so next getMore doesn't fail // still need to mark new location though cc->updateLocation(); } else { wassert( c->refLoc() != dl ); cc->updateLocation(); } } } void aboutToDelete(const DiskLoc& dl) { ClientCursor::aboutToDelete(dl); } ClientCursor::~ClientCursor() { assert( pos != -2 ); { recursive_scoped_lock lock(ccmutex); setLastLoc_inlock( DiskLoc() ); // removes us from bylocation multimap clientCursorsById.erase(cursorid); // defensive: (CursorId&) cursorid = -1; pos = -2; } } /* call when cursor's location changes so that we can update the cursorsbylocation map. if you are locked and internally iterating, only need to call when you are ready to "unlock". */ void ClientCursor::updateLocation() { assert( cursorid ); _idleAgeMillis = 0; DiskLoc cl = c->refLoc(); if ( lastLoc() == cl ) { //log() << "info: lastloc==curloc " << ns << '\n'; } else { recursive_scoped_lock lock(ccmutex); setLastLoc_inlock(cl); } // may be necessary for MultiCursor even when cl hasn't changed c->noteLocation(); } int ClientCursor::yieldSuggest() { int writers = 0; int readers = 0; int micros = Client::recommendedYieldMicros( &writers , &readers ); if ( micros > 0 && writers == 0 && dbMutex.getState() <= 0 ){ // we have a read lock, and only reads are coming on, so why bother unlocking micros = 0; } return micros; } bool ClientCursor::yieldSometimes(){ if ( ! _yieldSometimesTracker.ping() ) return true; int micros = yieldSuggest(); return ( micros > 0 ) ? yield( micros ) : true; } void ClientCursor::staticYield( int micros ) { { dbtempreleasecond unlock; if ( unlock.unlocked() ){ if ( micros == -1 ) micros = Client::recommendedYieldMicros(); if ( micros > 0 ) sleepmicros( micros ); } else { log( LL_WARNING ) << "ClientCursor::yield can't unlock b/c of recursive lock" << endl; } } } bool ClientCursor::prepareToYield( YieldData &data ) { if ( ! c->supportYields() ) return false; // need to store in case 'this' gets deleted data._id = cursorid; data._doingDeletes = _doingDeletes; _doingDeletes = false; updateLocation(); { /* a quick test that our temprelease is safe. todo: make a YieldingCursor class and then make the following code part of a unit test. */ const int test = 0; static bool inEmpty = false; if( test && !inEmpty ) { inEmpty = true; log() << "TEST: manipulate collection during cc:yield" << endl; if( test == 1 ) Helpers::emptyCollection(ns.c_str()); else if( test == 2 ) { BSONObjBuilder b; string m; dropCollection(ns.c_str(), m, b); } else { dropDatabase(ns.c_str()); } } } return true; } bool ClientCursor::recoverFromYield( const YieldData &data ) { ClientCursor *cc = ClientCursor::find( data._id , false ); if ( cc == 0 ){ // id was deleted return false; } cc->_doingDeletes = data._doingDeletes; cc->c->checkLocation(); return true; } bool ClientCursor::yield( int micros ) { if ( ! c->supportYields() ) return true; YieldData data; prepareToYield( data ); staticYield( micros ); return ClientCursor::recoverFromYield( data ); } int ctmLast = 0; // so we don't have to do find() which is a little slow very often. long long ClientCursor::allocCursorId_inlock() { if( 0 ) { static long long z; ++z; cout << "TEMP alloccursorid " << z << endl; return z; } long long x; int ctm = (int) curTimeMillis(); while ( 1 ) { x = (((long long)rand()) << 32); x = x | ctm | 0x80000000; // OR to make sure not zero if ( ctm != ctmLast || ClientCursor::find_inlock(x, false) == 0 ) break; } ctmLast = ctm; //DEV tlog() << " alloccursorid " << x << endl; return x; } void ClientCursor::storeOpForSlave( DiskLoc last ){ if ( ! ( _queryOptions & QueryOption_OplogReplay )) return; if ( last.isNull() ) return; BSONElement e = last.obj()["ts"]; if ( e.type() == Date || e.type() == Timestamp ) _slaveReadTill = e._opTime(); } void ClientCursor::updateSlaveLocation( CurOp& curop ){ if ( _slaveReadTill.isNull() ) return; mongo::updateSlaveLocation( curop , ns.c_str() , _slaveReadTill ); } void ClientCursor::appendStats( BSONObjBuilder& result ){ recursive_scoped_lock lock(ccmutex); result.appendNumber("totalOpen", clientCursorsById.size() ); result.appendNumber("clientCursors_size", (int) numCursors()); result.appendNumber("timedOut" , numberTimedOut); } // QUESTION: Restrict to the namespace from which this command was issued? // Alternatively, make this command admin-only? class CmdCursorInfo : public Command { public: CmdCursorInfo() : Command( "cursorInfo", true ) {} virtual bool slaveOk() const { return true; } virtual void help( stringstream& help ) const { help << " example: { cursorInfo : 1 }"; } virtual LockType locktype() const { return NONE; } bool run(const string& dbname, BSONObj& jsobj, string& errmsg, BSONObjBuilder& result, bool fromRepl ){ ClientCursor::appendStats( result ); return true; } } cmdCursorInfo; void ClientCursorMonitor::run(){ Client::initThread("clientcursormon"); Client& client = cc(); unsigned old = curTimeMillis(); while ( ! inShutdown() ){ unsigned now = curTimeMillis(); ClientCursor::idleTimeReport( now - old ); old = now; sleepsecs(4); } client.shutdown(); } void ClientCursor::find( const string& ns , set& all ){ recursive_scoped_lock lock(ccmutex); for ( CCById::iterator i=clientCursorsById.begin(); i!=clientCursorsById.end(); ++i ){ if ( i->second->ns == ns ) all.insert( i->first ); } } ClientCursorMonitor clientCursorMonitor; } // namespace mongo