diff options
Diffstat (limited to 'client')
38 files changed, 0 insertions, 10270 deletions
diff --git a/client/clientOnly.cpp b/client/clientOnly.cpp deleted file mode 100644 index 11890c84082..00000000000 --- a/client/clientOnly.cpp +++ /dev/null @@ -1,79 +0,0 @@ -// clientOnly.cpp - -/* Copyright 2009 10gen Inc. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -#include "pch.h" -#include "../client/dbclient.h" -#include "../db/cmdline.h" -#include "../s/shard.h" - -namespace mongo { - - CmdLine cmdLine; - - const char * curNs = "in client mode"; - - bool dbexitCalled = false; - - void exitCleanly( ExitCode code ) { - dbexit( code ); - } - - void dbexit( ExitCode returnCode, const char *whyMsg , bool tryToGetLock ) { - dbexitCalled = true; - out() << "dbexit called" << endl; - if ( whyMsg ) - out() << " b/c " << whyMsg << endl; - out() << "exiting" << endl; - ::exit( returnCode ); - } - - bool inShutdown() { - return dbexitCalled; - } - - void setupSignals() { - // maybe should do SIGPIPE here, not sure - } - - string getDbContext() { - return "in client only mode"; - } - - bool haveLocalShardingInfo( const string& ns ) { - return false; - } - - DBClientBase * createDirectClient() { - uassert( 10256 , "no createDirectClient in clientOnly" , 0 ); - return 0; - } - - void Shard::getAllShards( vector<Shard>& all ) { - assert(0); - } - - bool Shard::isAShardNode( const string& ident ) { - assert(0); - return false; - } - - string prettyHostName() { - assert(0); - return ""; - } - -} diff --git a/client/connpool.cpp b/client/connpool.cpp deleted file mode 100644 index ca3713dbe82..00000000000 --- a/client/connpool.cpp +++ /dev/null @@ -1,459 +0,0 @@ -/* connpool.cpp -*/ - -/* Copyright 2009 10gen Inc. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -// _ todo: reconnect? - -#include "pch.h" -#include "connpool.h" -#include "../db/commands.h" -#include "syncclusterconnection.h" -#include "../s/shard.h" - -namespace mongo { - - // ------ PoolForHost ------ - - PoolForHost::~PoolForHost() { - while ( ! _pool.empty() ) { - StoredConnection sc = _pool.top(); - delete sc.conn; - _pool.pop(); - } - } - - void PoolForHost::done( DBConnectionPool * pool, DBClientBase * c ) { - if ( _pool.size() >= _maxPerHost ) { - pool->onDestroy( c ); - delete c; - } - else { - _pool.push(c); - } - } - - DBClientBase * PoolForHost::get( DBConnectionPool * pool , double socketTimeout ) { - - time_t now = time(0); - - while ( ! _pool.empty() ) { - StoredConnection sc = _pool.top(); - _pool.pop(); - - if ( ! sc.ok( now ) ) { - pool->onDestroy( sc.conn ); - delete sc.conn; - continue; - } - - assert( sc.conn->getSoTimeout() == socketTimeout ); - - return sc.conn; - - } - - return NULL; - } - - void PoolForHost::flush() { - vector<StoredConnection> all; - while ( ! _pool.empty() ) { - StoredConnection c = _pool.top(); - _pool.pop(); - all.push_back( c ); - bool res; - c.conn->isMaster( res ); - } - - for ( vector<StoredConnection>::iterator i=all.begin(); i != all.end(); ++i ) { - _pool.push( *i ); - } - } - - void PoolForHost::getStaleConnections( vector<DBClientBase*>& stale ) { - time_t now = time(0); - - vector<StoredConnection> all; - while ( ! _pool.empty() ) { - StoredConnection c = _pool.top(); - _pool.pop(); - - if ( c.ok( now ) ) - all.push_back( c ); - else - stale.push_back( c.conn ); - } - - for ( size_t i=0; i<all.size(); i++ ) { - _pool.push( all[i] ); - } - } - - - PoolForHost::StoredConnection::StoredConnection( DBClientBase * c ) { - conn = c; - when = time(0); - } - - bool PoolForHost::StoredConnection::ok( time_t now ) { - // if connection has been idle for 30 minutes, kill it - return ( now - when ) < 1800; - } - - void PoolForHost::createdOne( DBClientBase * base) { - if ( _created == 0 ) - _type = base->type(); - _created++; - } - - unsigned PoolForHost::_maxPerHost = 50; - - // ------ DBConnectionPool ------ - - DBConnectionPool pool; - - DBConnectionPool::DBConnectionPool() - : _mutex("DBConnectionPool") , - _name( "dbconnectionpool" ) , - _hooks( new list<DBConnectionHook*>() ) { - } - - DBClientBase* DBConnectionPool::_get(const string& ident , double socketTimeout ) { - assert( ! inShutdown() ); - scoped_lock L(_mutex); - PoolForHost& p = _pools[PoolKey(ident,socketTimeout)]; - return p.get( this , socketTimeout ); - } - - DBClientBase* DBConnectionPool::_finishCreate( const string& host , double socketTimeout , DBClientBase* conn ) { - { - scoped_lock L(_mutex); - PoolForHost& p = _pools[PoolKey(host,socketTimeout)]; - p.createdOne( conn ); - } - - try { - onCreate( conn ); - onHandedOut( conn ); - } - catch ( std::exception& e ) { - delete conn; - throw; - } - - return conn; - } - - DBClientBase* DBConnectionPool::get(const ConnectionString& url, double socketTimeout) { - DBClientBase * c = _get( url.toString() , socketTimeout ); - if ( c ) { - try { - onHandedOut( c ); - } - catch ( std::exception& e ) { - delete c; - throw; - } - return c; - } - - string errmsg; - c = url.connect( errmsg, socketTimeout ); - uassert( 13328 , _name + ": connect failed " + url.toString() + " : " + errmsg , c ); - - return _finishCreate( url.toString() , socketTimeout , c ); - } - - DBClientBase* DBConnectionPool::get(const string& host, double socketTimeout) { - DBClientBase * c = _get( host , socketTimeout ); - if ( c ) { - try { - onHandedOut( c ); - } - catch ( std::exception& e ) { - delete c; - throw; - } - return c; - } - - string errmsg; - ConnectionString cs = ConnectionString::parse( host , errmsg ); - uassert( 13071 , (string)"invalid hostname [" + host + "]" + errmsg , cs.isValid() ); - - c = cs.connect( errmsg, socketTimeout ); - if ( ! c ) - throw SocketException( SocketException::CONNECT_ERROR , host , 11002 , str::stream() << _name << " error: " << errmsg ); - return _finishCreate( host , socketTimeout , c ); - } - - void DBConnectionPool::release(const string& host, DBClientBase *c) { - if ( c->isFailed() ) { - onDestroy( c ); - delete c; - return; - } - scoped_lock L(_mutex); - _pools[PoolKey(host,c->getSoTimeout())].done(this,c); - } - - - DBConnectionPool::~DBConnectionPool() { - // connection closing is handled by ~PoolForHost - } - - void DBConnectionPool::flush() { - scoped_lock L(_mutex); - for ( PoolMap::iterator i = _pools.begin(); i != _pools.end(); i++ ) { - PoolForHost& p = i->second; - p.flush(); - } - } - - void DBConnectionPool::addHook( DBConnectionHook * hook ) { - _hooks->push_back( hook ); - } - - void DBConnectionPool::onCreate( DBClientBase * conn ) { - if ( _hooks->size() == 0 ) - return; - - for ( list<DBConnectionHook*>::iterator i = _hooks->begin(); i != _hooks->end(); i++ ) { - (*i)->onCreate( conn ); - } - } - - void DBConnectionPool::onHandedOut( DBClientBase * conn ) { - if ( _hooks->size() == 0 ) - return; - - for ( list<DBConnectionHook*>::iterator i = _hooks->begin(); i != _hooks->end(); i++ ) { - (*i)->onHandedOut( conn ); - } - } - - void DBConnectionPool::onDestroy( DBClientBase * conn ) { - if ( _hooks->size() == 0 ) - return; - - for ( list<DBConnectionHook*>::iterator i = _hooks->begin(); i != _hooks->end(); i++ ) { - (*i)->onDestroy( conn ); - } - } - - void DBConnectionPool::appendInfo( BSONObjBuilder& b ) { - - int avail = 0; - long long created = 0; - - - map<ConnectionString::ConnectionType,long long> createdByType; - - set<string> replicaSets; - - BSONObjBuilder bb( b.subobjStart( "hosts" ) ); - { - scoped_lock lk( _mutex ); - for ( PoolMap::iterator i=_pools.begin(); i!=_pools.end(); ++i ) { - if ( i->second.numCreated() == 0 ) - continue; - - string s = str::stream() << i->first.ident << "::" << i->first.timeout; - - BSONObjBuilder temp( bb.subobjStart( s ) ); - temp.append( "available" , i->second.numAvailable() ); - temp.appendNumber( "created" , i->second.numCreated() ); - temp.done(); - - avail += i->second.numAvailable(); - created += i->second.numCreated(); - - long long& x = createdByType[i->second.type()]; - x += i->second.numCreated(); - - { - string setName = i->first.ident; - if ( setName.find( "/" ) != string::npos ) { - setName = setName.substr( 0 , setName.find( "/" ) ); - replicaSets.insert( setName ); - } - } - } - } - bb.done(); - - - BSONObjBuilder setBuilder( b.subobjStart( "replicaSets" ) ); - for ( set<string>::iterator i=replicaSets.begin(); i!=replicaSets.end(); ++i ) { - string rs = *i; - ReplicaSetMonitorPtr m = ReplicaSetMonitor::get( rs ); - if ( ! m ) { - warning() << "no monitor for set: " << rs << endl; - continue; - } - - BSONObjBuilder temp( setBuilder.subobjStart( rs ) ); - m->appendInfo( temp ); - temp.done(); - } - setBuilder.done(); - - { - BSONObjBuilder temp( bb.subobjStart( "createdByType" ) ); - for ( map<ConnectionString::ConnectionType,long long>::iterator i=createdByType.begin(); i!=createdByType.end(); ++i ) { - temp.appendNumber( ConnectionString::typeToString( i->first ) , i->second ); - } - temp.done(); - } - - b.append( "totalAvailable" , avail ); - b.appendNumber( "totalCreated" , created ); - } - - bool DBConnectionPool::serverNameCompare::operator()( const string& a , const string& b ) const{ - const char* ap = a.c_str(); - const char* bp = b.c_str(); - - while (true){ - if (*ap == '\0' || *ap == '/'){ - if (*bp == '\0' || *bp == '/') - return false; // equal strings - else - return true; // a is shorter - } - - if (*bp == '\0' || *bp == '/') - return false; // b is shorter - - if ( *ap < *bp) - return true; - else if (*ap > *bp) - return false; - - ++ap; - ++bp; - } - assert(false); - } - - bool DBConnectionPool::poolKeyCompare::operator()( const PoolKey& a , const PoolKey& b ) const { - if (DBConnectionPool::serverNameCompare()( a.ident , b.ident )) - return true; - - if (DBConnectionPool::serverNameCompare()( b.ident , a.ident )) - return false; - - return a.timeout < b.timeout; - } - - - void DBConnectionPool::taskDoWork() { - vector<DBClientBase*> toDelete; - - { - // we need to get the connections inside the lock - // but we can actually delete them outside - scoped_lock lk( _mutex ); - for ( PoolMap::iterator i=_pools.begin(); i!=_pools.end(); ++i ) { - i->second.getStaleConnections( toDelete ); - } - } - - for ( size_t i=0; i<toDelete.size(); i++ ) { - try { - onDestroy( toDelete[i] ); - delete toDelete[i]; - } - catch ( ... ) { - // we don't care if there was a socket error - } - } - } - - // ------ ScopedDbConnection ------ - - ScopedDbConnection * ScopedDbConnection::steal() { - assert( _conn ); - ScopedDbConnection * n = new ScopedDbConnection( _host , _conn, _socketTimeout ); - _conn = 0; - return n; - } - - void ScopedDbConnection::_setSocketTimeout(){ - if( ! _conn ) return; - if( _conn->type() == ConnectionString::MASTER ) - (( DBClientConnection* ) _conn)->setSoTimeout( _socketTimeout ); - else if( _conn->type() == ConnectionString::SYNC ) - (( SyncClusterConnection* ) _conn)->setAllSoTimeouts( _socketTimeout ); - } - - ScopedDbConnection::~ScopedDbConnection() { - if ( _conn ) { - if ( ! _conn->isFailed() ) { - /* see done() comments above for why we log this line */ - log() << "scoped connection to " << _conn->getServerAddress() << " not being returned to the pool" << endl; - } - kill(); - } - } - - ScopedDbConnection::ScopedDbConnection(const Shard& shard, double socketTimeout ) - : _host( shard.getConnString() ) , _conn( pool.get(_host, socketTimeout) ), _socketTimeout( socketTimeout ) { - _setSocketTimeout(); - } - - ScopedDbConnection::ScopedDbConnection(const Shard* shard, double socketTimeout ) - : _host( shard->getConnString() ) , _conn( pool.get(_host, socketTimeout) ), _socketTimeout( socketTimeout ) { - _setSocketTimeout(); - } - - - class PoolFlushCmd : public Command { - public: - PoolFlushCmd() : Command( "connPoolSync" , false , "connpoolsync" ) {} - virtual void help( stringstream &help ) const { help<<"internal"; } - virtual LockType locktype() const { return NONE; } - virtual bool run(const string&, mongo::BSONObj&, int, std::string&, mongo::BSONObjBuilder& result, bool) { - pool.flush(); - return true; - } - virtual bool slaveOk() const { - return true; - } - - } poolFlushCmd; - - class PoolStats : public Command { - public: - PoolStats() : Command( "connPoolStats" ) {} - virtual void help( stringstream &help ) const { help<<"stats about connection pool"; } - virtual LockType locktype() const { return NONE; } - virtual bool run(const string&, mongo::BSONObj&, int, std::string&, mongo::BSONObjBuilder& result, bool) { - pool.appendInfo( result ); - result.append( "numDBClientConnection" , DBClientConnection::getNumConnections() ); - result.append( "numAScopedConnection" , AScopedConnection::getNumConnections() ); - return true; - } - virtual bool slaveOk() const { - return true; - } - - } poolStatsCmd; - - AtomicUInt AScopedConnection::_numConnections; - -} // namespace mongo diff --git a/client/connpool.h b/client/connpool.h deleted file mode 100644 index 8733abb1f90..00000000000 --- a/client/connpool.h +++ /dev/null @@ -1,291 +0,0 @@ -/** @file connpool.h */ - -/* Copyright 2009 10gen Inc. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -#pragma once - -#include <stack> -#include "dbclient.h" -#include "redef_macros.h" - -#include "../util/background.h" - -namespace mongo { - - class Shard; - class DBConnectionPool; - - /** - * not thread safe - * thread safety is handled by DBConnectionPool - */ - class PoolForHost { - public: - PoolForHost() - : _created(0) {} - - PoolForHost( const PoolForHost& other ) { - assert(other._pool.size() == 0); - _created = other._created; - assert( _created == 0 ); - } - - ~PoolForHost(); - - int numAvailable() const { return (int)_pool.size(); } - - void createdOne( DBClientBase * base ); - long long numCreated() const { return _created; } - - ConnectionString::ConnectionType type() const { assert(_created); return _type; } - - /** - * gets a connection or return NULL - */ - DBClientBase * get( DBConnectionPool * pool , double socketTimeout ); - - void done( DBConnectionPool * pool , DBClientBase * c ); - - void flush(); - - void getStaleConnections( vector<DBClientBase*>& stale ); - - static void setMaxPerHost( unsigned max ) { _maxPerHost = max; } - static unsigned getMaxPerHost() { return _maxPerHost; } - private: - - struct StoredConnection { - StoredConnection( DBClientBase * c ); - - bool ok( time_t now ); - - DBClientBase* conn; - time_t when; - }; - - std::stack<StoredConnection> _pool; - - long long _created; - ConnectionString::ConnectionType _type; - - static unsigned _maxPerHost; - }; - - class DBConnectionHook { - public: - virtual ~DBConnectionHook() {} - virtual void onCreate( DBClientBase * conn ) {} - virtual void onHandedOut( DBClientBase * conn ) {} - virtual void onDestroy( DBClientBase * conn ) {} - }; - - /** Database connection pool. - - Generally, use ScopedDbConnection and do not call these directly. - - This class, so far, is suitable for use with unauthenticated connections. - Support for authenticated connections requires some adjustements: please - request... - - Usage: - - { - ScopedDbConnection c("myserver"); - c.conn()... - } - */ - class DBConnectionPool : public PeriodicTask { - - public: - - DBConnectionPool(); - ~DBConnectionPool(); - - /** right now just controls some asserts. defaults to "dbconnectionpool" */ - void setName( const string& name ) { _name = name; } - - void onCreate( DBClientBase * conn ); - void onHandedOut( DBClientBase * conn ); - void onDestroy( DBClientBase * conn ); - - void flush(); - - DBClientBase *get(const string& host, double socketTimeout = 0); - DBClientBase *get(const ConnectionString& host, double socketTimeout = 0); - - void release(const string& host, DBClientBase *c); - - void addHook( DBConnectionHook * hook ); // we take ownership - void appendInfo( BSONObjBuilder& b ); - - /** compares server namees, but is smart about replica set names */ - struct serverNameCompare { - bool operator()( const string& a , const string& b ) const; - }; - - virtual string taskName() const { return "DBConnectionPool-cleaner"; } - virtual void taskDoWork(); - - private: - DBConnectionPool( DBConnectionPool& p ); - - DBClientBase* _get( const string& ident , double socketTimeout ); - - DBClientBase* _finishCreate( const string& ident , double socketTimeout, DBClientBase* conn ); - - struct PoolKey { - PoolKey( string i , double t ) : ident( i ) , timeout( t ) {} - string ident; - double timeout; - }; - - struct poolKeyCompare { - bool operator()( const PoolKey& a , const PoolKey& b ) const; - }; - - typedef map<PoolKey,PoolForHost,poolKeyCompare> PoolMap; // servername -> pool - - mongo::mutex _mutex; - string _name; - - PoolMap _pools; - - // pointers owned by me, right now they leak on shutdown - // _hooks itself also leaks because it creates a shutdown race condition - list<DBConnectionHook*> * _hooks; - - }; - - extern DBConnectionPool pool; - - class AScopedConnection : boost::noncopyable { - public: - AScopedConnection() { _numConnections++; } - virtual ~AScopedConnection() { _numConnections--; } - - virtual DBClientBase* get() = 0; - virtual void done() = 0; - virtual string getHost() const = 0; - - /** - * @return true iff this has a connection to the db - */ - virtual bool ok() const = 0; - - /** - * @return total number of current instances of AScopedConnection - */ - static int getNumConnections() { return _numConnections; } - - private: - static AtomicUInt _numConnections; - }; - - /** Use to get a connection from the pool. On exceptions things - clean up nicely (i.e. the socket gets closed automatically when the - scopeddbconnection goes out of scope). - */ - class ScopedDbConnection : public AScopedConnection { - public: - /** the main constructor you want to use - throws UserException if can't connect - */ - explicit ScopedDbConnection(const string& host, double socketTimeout = 0) : _host(host), _conn( pool.get(host, socketTimeout) ), _socketTimeout( socketTimeout ) { - _setSocketTimeout(); - } - - ScopedDbConnection() : _host( "" ) , _conn(0), _socketTimeout( 0 ) {} - - /* @param conn - bind to an existing connection */ - ScopedDbConnection(const string& host, DBClientBase* conn, double socketTimeout = 0 ) : _host( host ) , _conn( conn ), _socketTimeout( socketTimeout ) { - _setSocketTimeout(); - } - - /** throws UserException if can't connect */ - explicit ScopedDbConnection(const ConnectionString& url, double socketTimeout = 0 ) : _host(url.toString()), _conn( pool.get(url, socketTimeout) ), _socketTimeout( socketTimeout ) { - _setSocketTimeout(); - } - - /** throws UserException if can't connect */ - explicit ScopedDbConnection(const Shard& shard, double socketTimeout = 0 ); - explicit ScopedDbConnection(const Shard* shard, double socketTimeout = 0 ); - - ~ScopedDbConnection(); - - /** get the associated connection object */ - DBClientBase* operator->() { - uassert( 11004 , "connection was returned to the pool already" , _conn ); - return _conn; - } - - /** get the associated connection object */ - DBClientBase& conn() { - uassert( 11005 , "connection was returned to the pool already" , _conn ); - return *_conn; - } - - /** get the associated connection object */ - DBClientBase* get() { - uassert( 13102 , "connection was returned to the pool already" , _conn ); - return _conn; - } - - bool ok() const { return _conn > 0; } - - string getHost() const { return _host; } - - /** Force closure of the connection. You should call this if you leave it in - a bad state. Destructor will do this too, but it is verbose. - */ - void kill() { - delete _conn; - _conn = 0; - } - - /** Call this when you are done with the connection. - - If you do not call done() before this object goes out of scope, - we can't be sure we fully read all expected data of a reply on the socket. so - we don't try to reuse the connection in that situation. - */ - void done() { - if ( ! _conn ) - return; - - /* we could do this, but instead of assume one is using autoreconnect mode on the connection - if ( _conn->isFailed() ) - kill(); - else - */ - pool.release(_host, _conn); - _conn = 0; - } - - ScopedDbConnection * steal(); - - private: - - void _setSocketTimeout(); - - const string _host; - DBClientBase *_conn; - const double _socketTimeout; - - }; - -} // namespace mongo - -#include "undef_macros.h" diff --git a/client/constants.h b/client/constants.h deleted file mode 100644 index 54f3fd216f2..00000000000 --- a/client/constants.h +++ /dev/null @@ -1,26 +0,0 @@ -// constants.h - -#pragma once - -namespace mongo { - - /* query results include a 32 result flag word consisting of these bits */ - enum ResultFlagType { - /* returned, with zero results, when getMore is called but the cursor id - is not valid at the server. */ - ResultFlag_CursorNotFound = 1, - - /* { $err : ... } is being returned */ - ResultFlag_ErrSet = 2, - - /* Have to update config from the server, usually $err is also set */ - ResultFlag_ShardConfigStale = 4, - - /* for backward compatability: this let's us know the server supports - the QueryOption_AwaitData option. if it doesn't, a repl slave client should sleep - a little between getMore's. - */ - ResultFlag_AwaitCapable = 8 - }; - -} diff --git a/client/dbclient.cpp b/client/dbclient.cpp deleted file mode 100644 index 6b9631b09ee..00000000000 --- a/client/dbclient.cpp +++ /dev/null @@ -1,1053 +0,0 @@ -// dbclient.cpp - connect to a Mongo database as a database, from C++ - -/* Copyright 2009 10gen Inc. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -#include "pch.h" -#include "../db/pdfile.h" -#include "dbclient.h" -#include "../bson/util/builder.h" -#include "../db/jsobj.h" -#include "../db/json.h" -#include "../db/instance.h" -#include "../util/md5.hpp" -#include "../db/dbmessage.h" -#include "../db/cmdline.h" -#include "connpool.h" -#include "../s/util.h" -#include "syncclusterconnection.h" - -namespace mongo { - - void ConnectionString::_fillServers( string s ) { - - { - string::size_type idx = s.find( '/' ); - if ( idx != string::npos ) { - _setName = s.substr( 0 , idx ); - s = s.substr( idx + 1 ); - _type = SET; - } - } - - string::size_type idx; - while ( ( idx = s.find( ',' ) ) != string::npos ) { - _servers.push_back( s.substr( 0 , idx ) ); - s = s.substr( idx + 1 ); - } - _servers.push_back( s ); - - } - - void ConnectionString::_finishInit() { - stringstream ss; - if ( _type == SET ) - ss << _setName << "/"; - for ( unsigned i=0; i<_servers.size(); i++ ) { - if ( i > 0 ) - ss << ","; - ss << _servers[i].toString(); - } - _string = ss.str(); - } - - - DBClientBase* ConnectionString::connect( string& errmsg, double socketTimeout ) const { - switch ( _type ) { - case MASTER: { - DBClientConnection * c = new DBClientConnection(true); - c->setSoTimeout( socketTimeout ); - log(1) << "creating new connection to:" << _servers[0] << endl; - if ( ! c->connect( _servers[0] , errmsg ) ) { - delete c; - return 0; - } - log(1) << "connected connection!" << endl; - return c; - } - - case PAIR: - case SET: { - DBClientReplicaSet * set = new DBClientReplicaSet( _setName , _servers , socketTimeout ); - if( ! set->connect() ) { - delete set; - errmsg = "connect failed to set "; - errmsg += toString(); - return 0; - } - return set; - } - - case SYNC: { - // TODO , don't copy - list<HostAndPort> l; - for ( unsigned i=0; i<_servers.size(); i++ ) - l.push_back( _servers[i] ); - SyncClusterConnection* c = new SyncClusterConnection( l, socketTimeout ); - return c; - } - - case INVALID: - throw UserException( 13421 , "trying to connect to invalid ConnectionString" ); - break; - } - - assert( 0 ); - return 0; - } - - ConnectionString ConnectionString::parse( const string& host , string& errmsg ) { - - string::size_type i = host.find( '/' ); - if ( i != string::npos && i != 0) { - // replica set - return ConnectionString( SET , host.substr( i + 1 ) , host.substr( 0 , i ) ); - } - - int numCommas = str::count( host , ',' ); - - if( numCommas == 0 ) - return ConnectionString( HostAndPort( host ) ); - - if ( numCommas == 1 ) - return ConnectionString( PAIR , host ); - - if ( numCommas == 2 ) - return ConnectionString( SYNC , host ); - - errmsg = (string)"invalid hostname [" + host + "]"; - return ConnectionString(); // INVALID - } - - string ConnectionString::typeToString( ConnectionType type ) { - switch ( type ) { - case INVALID: - return "invalid"; - case MASTER: - return "master"; - case PAIR: - return "pair"; - case SET: - return "set"; - case SYNC: - return "sync"; - } - assert(0); - return ""; - } - - - Query& Query::where(const string &jscode, BSONObj scope) { - /* use where() before sort() and hint() and explain(), else this will assert. */ - assert( ! isComplex() ); - BSONObjBuilder b; - b.appendElements(obj); - b.appendWhere(jscode, scope); - obj = b.obj(); - return *this; - } - - void Query::makeComplex() { - if ( isComplex() ) - return; - BSONObjBuilder b; - b.append( "query", obj ); - obj = b.obj(); - } - - Query& Query::sort(const BSONObj& s) { - appendComplex( "orderby", s ); - return *this; - } - - Query& Query::hint(BSONObj keyPattern) { - appendComplex( "$hint", keyPattern ); - return *this; - } - - Query& Query::explain() { - appendComplex( "$explain", true ); - return *this; - } - - Query& Query::snapshot() { - appendComplex( "$snapshot", true ); - return *this; - } - - Query& Query::minKey( const BSONObj &val ) { - appendComplex( "$min", val ); - return *this; - } - - Query& Query::maxKey( const BSONObj &val ) { - appendComplex( "$max", val ); - return *this; - } - - bool Query::isComplex( bool * hasDollar ) const { - if ( obj.hasElement( "query" ) ) { - if ( hasDollar ) - hasDollar[0] = false; - return true; - } - - if ( obj.hasElement( "$query" ) ) { - if ( hasDollar ) - hasDollar[0] = true; - return true; - } - - return false; - } - - BSONObj Query::getFilter() const { - bool hasDollar; - if ( ! isComplex( &hasDollar ) ) - return obj; - - return obj.getObjectField( hasDollar ? "$query" : "query" ); - } - BSONObj Query::getSort() const { - if ( ! isComplex() ) - return BSONObj(); - BSONObj ret = obj.getObjectField( "orderby" ); - if (ret.isEmpty()) - ret = obj.getObjectField( "$orderby" ); - return ret; - } - BSONObj Query::getHint() const { - if ( ! isComplex() ) - return BSONObj(); - return obj.getObjectField( "$hint" ); - } - bool Query::isExplain() const { - return isComplex() && obj.getBoolField( "$explain" ); - } - - string Query::toString() const { - return obj.toString(); - } - - /* --- dbclientcommands --- */ - - bool DBClientWithCommands::isOk(const BSONObj& o) { - return o["ok"].trueValue(); - } - - bool DBClientWithCommands::isNotMasterErrorString( const BSONElement& e ) { - return e.type() == String && str::contains( e.valuestr() , "not master" ); - } - - - enum QueryOptions DBClientWithCommands::availableOptions() { - if ( !_haveCachedAvailableOptions ) { - BSONObj ret; - if ( runCommand( "admin", BSON( "availablequeryoptions" << 1 ), ret ) ) { - _cachedAvailableOptions = ( enum QueryOptions )( ret.getIntField( "options" ) ); - } - _haveCachedAvailableOptions = true; - } - return _cachedAvailableOptions; - } - - inline bool DBClientWithCommands::runCommand(const string &dbname, const BSONObj& cmd, BSONObj &info, int options) { - string ns = dbname + ".$cmd"; - info = findOne(ns, cmd, 0 , options); - return isOk(info); - } - - /* note - we build a bson obj here -- for something that is super common like getlasterror you - should have that object prebuilt as that would be faster. - */ - bool DBClientWithCommands::simpleCommand(const string &dbname, BSONObj *info, const string &command) { - BSONObj o; - if ( info == 0 ) - info = &o; - BSONObjBuilder b; - b.append(command, 1); - return runCommand(dbname, b.done(), *info); - } - - unsigned long long DBClientWithCommands::count(const string &myns, const BSONObj& query, int options, int limit, int skip ) { - NamespaceString ns(myns); - BSONObj cmd = _countCmd( myns , query , options , limit , skip ); - BSONObj res; - if( !runCommand(ns.db.c_str(), cmd, res, options) ) - uasserted(11010,string("count fails:") + res.toString()); - return res["n"].numberLong(); - } - - BSONObj DBClientWithCommands::_countCmd(const string &myns, const BSONObj& query, int options, int limit, int skip ) { - NamespaceString ns(myns); - BSONObjBuilder b; - b.append( "count" , ns.coll ); - b.append( "query" , query ); - if ( limit ) - b.append( "limit" , limit ); - if ( skip ) - b.append( "skip" , skip ); - return b.obj(); - } - - const BSONObj getlasterrorcmdobj = fromjson("{getlasterror:1}"); - - BSONObj DBClientWithCommands::getLastErrorDetailed() { - BSONObj info; - runCommand("admin", getlasterrorcmdobj, info); - return info; - } - - string DBClientWithCommands::getLastError() { - BSONObj info = getLastErrorDetailed(); - return getLastErrorString( info ); - } - - string DBClientWithCommands::getLastErrorString( const BSONObj& info ) { - BSONElement e = info["err"]; - if( e.eoo() ) return ""; - if( e.type() == Object ) return e.toString(); - return e.str(); - } - - const BSONObj getpreverrorcmdobj = fromjson("{getpreverror:1}"); - - BSONObj DBClientWithCommands::getPrevError() { - BSONObj info; - runCommand("admin", getpreverrorcmdobj, info); - return info; - } - - BSONObj getnoncecmdobj = fromjson("{getnonce:1}"); - - string DBClientWithCommands::createPasswordDigest( const string & username , const string & clearTextPassword ) { - md5digest d; - { - md5_state_t st; - md5_init(&st); - md5_append(&st, (const md5_byte_t *) username.data(), username.length()); - md5_append(&st, (const md5_byte_t *) ":mongo:", 7 ); - md5_append(&st, (const md5_byte_t *) clearTextPassword.data(), clearTextPassword.length()); - md5_finish(&st, d); - } - return digestToString( d ); - } - - bool DBClientWithCommands::auth(const string &dbname, const string &username, const string &password_text, string& errmsg, bool digestPassword) { - string password = password_text; - if( digestPassword ) - password = createPasswordDigest( username , password_text ); - - BSONObj info; - string nonce; - if( !runCommand(dbname, getnoncecmdobj, info) ) { - errmsg = "getnonce fails - connection problem?"; - return false; - } - { - BSONElement e = info.getField("nonce"); - assert( e.type() == String ); - nonce = e.valuestr(); - } - - BSONObj authCmd; - BSONObjBuilder b; - { - - b << "authenticate" << 1 << "nonce" << nonce << "user" << username; - md5digest d; - { - md5_state_t st; - md5_init(&st); - md5_append(&st, (const md5_byte_t *) nonce.c_str(), nonce.size() ); - md5_append(&st, (const md5_byte_t *) username.data(), username.length()); - md5_append(&st, (const md5_byte_t *) password.c_str(), password.size() ); - md5_finish(&st, d); - } - b << "key" << digestToString( d ); - authCmd = b.done(); - } - - if( runCommand(dbname, authCmd, info) ) - return true; - - errmsg = info.toString(); - return false; - } - - BSONObj ismastercmdobj = fromjson("{\"ismaster\":1}"); - - bool DBClientWithCommands::isMaster(bool& isMaster, BSONObj *info) { - BSONObj o; - if ( info == 0 ) - info = &o; - bool ok = runCommand("admin", ismastercmdobj, *info); - isMaster = info->getField("ismaster").trueValue(); - return ok; - } - - bool DBClientWithCommands::createCollection(const string &ns, long long size, bool capped, int max, BSONObj *info) { - assert(!capped||size); - BSONObj o; - if ( info == 0 ) info = &o; - BSONObjBuilder b; - string db = nsToDatabase(ns.c_str()); - b.append("create", ns.c_str() + db.length() + 1); - if ( size ) b.append("size", size); - if ( capped ) b.append("capped", true); - if ( max ) b.append("max", max); - return runCommand(db.c_str(), b.done(), *info); - } - - bool DBClientWithCommands::copyDatabase(const string &fromdb, const string &todb, const string &fromhost, BSONObj *info) { - BSONObj o; - if ( info == 0 ) info = &o; - BSONObjBuilder b; - b.append("copydb", 1); - b.append("fromhost", fromhost); - b.append("fromdb", fromdb); - b.append("todb", todb); - return runCommand("admin", b.done(), *info); - } - - bool DBClientWithCommands::setDbProfilingLevel(const string &dbname, ProfilingLevel level, BSONObj *info ) { - BSONObj o; - if ( info == 0 ) info = &o; - - if ( level ) { - // Create system.profile collection. If it already exists this does nothing. - // TODO: move this into the db instead of here so that all - // drivers don't have to do this. - string ns = dbname + ".system.profile"; - createCollection(ns.c_str(), 1024 * 1024, true, 0, info); - } - - BSONObjBuilder b; - b.append("profile", (int) level); - return runCommand(dbname, b.done(), *info); - } - - BSONObj getprofilingcmdobj = fromjson("{\"profile\":-1}"); - - bool DBClientWithCommands::getDbProfilingLevel(const string &dbname, ProfilingLevel& level, BSONObj *info) { - BSONObj o; - if ( info == 0 ) info = &o; - if ( runCommand(dbname, getprofilingcmdobj, *info) ) { - level = (ProfilingLevel) info->getIntField("was"); - return true; - } - return false; - } - - DBClientWithCommands::MROutput DBClientWithCommands::MRInline (BSON("inline" << 1)); - - BSONObj DBClientWithCommands::mapreduce(const string &ns, const string &jsmapf, const string &jsreducef, BSONObj query, MROutput output) { - BSONObjBuilder b; - b.append("mapreduce", nsGetCollection(ns)); - b.appendCode("map", jsmapf); - b.appendCode("reduce", jsreducef); - if( !query.isEmpty() ) - b.append("query", query); - b.append("out", output.out); - BSONObj info; - runCommand(nsGetDB(ns), b.done(), info); - return info; - } - - bool DBClientWithCommands::eval(const string &dbname, const string &jscode, BSONObj& info, BSONElement& retValue, BSONObj *args) { - BSONObjBuilder b; - b.appendCode("$eval", jscode); - if ( args ) - b.appendArray("args", *args); - bool ok = runCommand(dbname, b.done(), info); - if ( ok ) - retValue = info.getField("retval"); - return ok; - } - - bool DBClientWithCommands::eval(const string &dbname, const string &jscode) { - BSONObj info; - BSONElement retValue; - return eval(dbname, jscode, info, retValue); - } - - list<string> DBClientWithCommands::getDatabaseNames() { - BSONObj info; - uassert( 10005 , "listdatabases failed" , runCommand( "admin" , BSON( "listDatabases" << 1 ) , info ) ); - uassert( 10006 , "listDatabases.databases not array" , info["databases"].type() == Array ); - - list<string> names; - - BSONObjIterator i( info["databases"].embeddedObjectUserCheck() ); - while ( i.more() ) { - names.push_back( i.next().embeddedObjectUserCheck()["name"].valuestr() ); - } - - return names; - } - - list<string> DBClientWithCommands::getCollectionNames( const string& db ) { - list<string> names; - - string ns = db + ".system.namespaces"; - auto_ptr<DBClientCursor> c = query( ns.c_str() , BSONObj() ); - while ( c->more() ) { - string name = c->next()["name"].valuestr(); - if ( name.find( "$" ) != string::npos ) - continue; - names.push_back( name ); - } - return names; - } - - bool DBClientWithCommands::exists( const string& ns ) { - list<string> names; - - string db = nsGetDB( ns ) + ".system.namespaces"; - BSONObj q = BSON( "name" << ns ); - return count( db.c_str() , q, QueryOption_SlaveOk ) != 0; - } - - /* --- dbclientconnection --- */ - - bool DBClientConnection::auth(const string &dbname, const string &username, const string &password_text, string& errmsg, bool digestPassword) { - string password = password_text; - if( digestPassword ) - password = createPasswordDigest( username , password_text ); - - if( autoReconnect ) { - /* note we remember the auth info before we attempt to auth -- if the connection is broken, we will - then have it for the next autoreconnect attempt. - */ - pair<string,string> p = pair<string,string>(username, password); - authCache[dbname] = p; - } - - return DBClientBase::auth(dbname, username, password.c_str(), errmsg, false); - } - - /** query N objects from the database into an array. makes sense mostly when you want a small number of results. if a huge number, use - query() and iterate the cursor. - */ - void DBClientInterface::findN(vector<BSONObj>& out, const string& ns, Query query, int nToReturn, int nToSkip, const BSONObj *fieldsToReturn, int queryOptions) { - out.reserve(nToReturn); - - auto_ptr<DBClientCursor> c = - this->query(ns, query, nToReturn, nToSkip, fieldsToReturn, queryOptions); - - uassert( 10276 , str::stream() << "DBClientBase::findN: transport error: " << getServerAddress() << " query: " << query.toString(), c.get() ); - - if ( c->hasResultFlag( ResultFlag_ShardConfigStale ) ) - throw StaleConfigException( ns , "findN stale config" ); - - for( int i = 0; i < nToReturn; i++ ) { - if ( !c->more() ) - break; - out.push_back( c->nextSafe().copy() ); - } - } - - BSONObj DBClientInterface::findOne(const string &ns, const Query& query, const BSONObj *fieldsToReturn, int queryOptions) { - vector<BSONObj> v; - findN(v, ns, query, 1, 0, fieldsToReturn, queryOptions); - return v.empty() ? BSONObj() : v[0]; - } - - bool DBClientConnection::connect(const HostAndPort& server, string& errmsg) { - _server = server; - _serverString = _server.toString(); - return _connect( errmsg ); - } - - bool DBClientConnection::_connect( string& errmsg ) { - _serverString = _server.toString(); - // we keep around SockAddr for connection life -- maybe MessagingPort - // requires that? - server.reset(new SockAddr(_server.host().c_str(), _server.port())); - p.reset(new MessagingPort( _so_timeout, _logLevel )); - - if (server->getAddr() == "0.0.0.0") { - _failed = true; - return false; - } - - // if( _so_timeout == 0 ){ - // printStackTrace(); - // log() << "Connecting to server " << _serverString << " timeout " << _so_timeout << endl; - // } - if ( !p->connect(*server) ) { - stringstream ss; - ss << "couldn't connect to server " << _serverString; - errmsg = ss.str(); - _failed = true; - return false; - } - -#ifdef MONGO_SSL - if ( cmdLine.sslOnNormalPorts ) { - p->secure( sslManager() ); - } -#endif - - return true; - } - - - inline bool DBClientConnection::runCommand(const string &dbname, const BSONObj& cmd, BSONObj &info, int options) { - if ( DBClientWithCommands::runCommand( dbname , cmd , info , options ) ) - return true; - - if ( clientSet && isNotMasterErrorString( info["errmsg"] ) ) { - clientSet->isntMaster(); - // At this point, we've probably deleted *this* object, do *not* use afterward - } - - return false; - } - - - void DBClientConnection::_checkConnection() { - if ( !_failed ) - return; - if ( lastReconnectTry && time(0)-lastReconnectTry < 2 ) { - // we wait a little before reconnect attempt to avoid constant hammering. - // but we throw we don't want to try to use a connection in a bad state - throw SocketException( SocketException::FAILED_STATE , toString() ); - } - if ( !autoReconnect ) - throw SocketException( SocketException::FAILED_STATE , toString() ); - - lastReconnectTry = time(0); - log(_logLevel) << "trying reconnect to " << _serverString << endl; - string errmsg; - _failed = false; - if ( ! _connect(errmsg) ) { - _failed = true; - log(_logLevel) << "reconnect " << _serverString << " failed " << errmsg << endl; - throw SocketException( SocketException::CONNECT_ERROR , toString() ); - } - - log(_logLevel) << "reconnect " << _serverString << " ok" << endl; - for( map< string, pair<string,string> >::iterator i = authCache.begin(); i != authCache.end(); i++ ) { - const char *dbname = i->first.c_str(); - const char *username = i->second.first.c_str(); - const char *password = i->second.second.c_str(); - if( !DBClientBase::auth(dbname, username, password, errmsg, false) ) - log(_logLevel) << "reconnect: auth failed db:" << dbname << " user:" << username << ' ' << errmsg << '\n'; - } - } - - auto_ptr<DBClientCursor> DBClientBase::query(const string &ns, Query query, int nToReturn, - int nToSkip, const BSONObj *fieldsToReturn, int queryOptions , int batchSize ) { - auto_ptr<DBClientCursor> c( new DBClientCursor( this, - ns, query.obj, nToReturn, nToSkip, - fieldsToReturn, queryOptions , batchSize ) ); - if ( c->init() ) - return c; - return auto_ptr< DBClientCursor >( 0 ); - } - - auto_ptr<DBClientCursor> DBClientBase::getMore( const string &ns, long long cursorId, int nToReturn, int options ) { - auto_ptr<DBClientCursor> c( new DBClientCursor( this, ns, cursorId, nToReturn, options ) ); - if ( c->init() ) - return c; - return auto_ptr< DBClientCursor >( 0 ); - } - - struct DBClientFunConvertor { - void operator()( DBClientCursorBatchIterator &i ) { - while( i.moreInCurrentBatch() ) { - _f( i.nextSafe() ); - } - } - boost::function<void(const BSONObj &)> _f; - }; - - unsigned long long DBClientConnection::query( boost::function<void(const BSONObj&)> f, const string& ns, Query query, const BSONObj *fieldsToReturn, int queryOptions ) { - DBClientFunConvertor fun; - fun._f = f; - boost::function<void(DBClientCursorBatchIterator &)> ptr( fun ); - return DBClientConnection::query( ptr, ns, query, fieldsToReturn, queryOptions ); - } - - unsigned long long DBClientConnection::query( boost::function<void(DBClientCursorBatchIterator &)> f, const string& ns, Query query, const BSONObj *fieldsToReturn, int queryOptions ) { - // mask options - queryOptions &= (int)( QueryOption_NoCursorTimeout | QueryOption_SlaveOk ); - unsigned long long n = 0; - - bool doExhaust = ( availableOptions() & QueryOption_Exhaust ); - if ( doExhaust ) { - queryOptions |= (int)QueryOption_Exhaust; - } - auto_ptr<DBClientCursor> c( this->query(ns, query, 0, 0, fieldsToReturn, queryOptions) ); - uassert( 13386, "socket error for mapping query", c.get() ); - - if ( !doExhaust ) { - while( c->more() ) { - DBClientCursorBatchIterator i( *c ); - f( i ); - n += i.n(); - } - return n; - } - - try { - while( 1 ) { - while( c->moreInCurrentBatch() ) { - DBClientCursorBatchIterator i( *c ); - f( i ); - n += i.n(); - } - - if( c->getCursorId() == 0 ) - break; - - c->exhaustReceiveMore(); - } - } - catch(std::exception&) { - /* connection CANNOT be used anymore as more data may be on the way from the server. - we have to reconnect. - */ - _failed = true; - p->shutdown(); - throw; - } - - return n; - } - - void DBClientBase::insert( const string & ns , BSONObj obj , int flags) { - Message toSend; - - BufBuilder b; - b.appendNum( flags ); - b.appendStr( ns ); - obj.appendSelfToBufBuilder( b ); - - toSend.setData( dbInsert , b.buf() , b.len() ); - - say( toSend ); - } - - void DBClientBase::insert( const string & ns , const vector< BSONObj > &v , int flags) { - Message toSend; - - BufBuilder b; - b.appendNum( flags ); - b.appendStr( ns ); - for( vector< BSONObj >::const_iterator i = v.begin(); i != v.end(); ++i ) - i->appendSelfToBufBuilder( b ); - - toSend.setData( dbInsert, b.buf(), b.len() ); - - say( toSend ); - } - - void DBClientBase::remove( const string & ns , Query obj , bool justOne ) { - Message toSend; - - BufBuilder b; - int opts = 0; - b.appendNum( opts ); - b.appendStr( ns ); - - int flags = 0; - if ( justOne ) - flags |= RemoveOption_JustOne; - b.appendNum( flags ); - - obj.obj.appendSelfToBufBuilder( b ); - - toSend.setData( dbDelete , b.buf() , b.len() ); - - say( toSend ); - } - - void DBClientBase::update( const string & ns , Query query , BSONObj obj , bool upsert , bool multi ) { - - BufBuilder b; - b.appendNum( (int)0 ); // reserved - b.appendStr( ns ); - - int flags = 0; - if ( upsert ) flags |= UpdateOption_Upsert; - if ( multi ) flags |= UpdateOption_Multi; - b.appendNum( flags ); - - query.obj.appendSelfToBufBuilder( b ); - obj.appendSelfToBufBuilder( b ); - - Message toSend; - toSend.setData( dbUpdate , b.buf() , b.len() ); - - say( toSend ); - - - } - - - - auto_ptr<DBClientCursor> DBClientWithCommands::getIndexes( const string &ns ) { - return query( Namespace( ns.c_str() ).getSisterNS( "system.indexes" ).c_str() , BSON( "ns" << ns ) ); - } - - void DBClientWithCommands::dropIndex( const string& ns , BSONObj keys ) { - dropIndex( ns , genIndexName( keys ) ); - } - - - void DBClientWithCommands::dropIndex( const string& ns , const string& indexName ) { - BSONObj info; - if ( ! runCommand( nsToDatabase( ns.c_str() ) , - BSON( "deleteIndexes" << NamespaceString( ns ).coll << "index" << indexName ) , - info ) ) { - log(_logLevel) << "dropIndex failed: " << info << endl; - uassert( 10007 , "dropIndex failed" , 0 ); - } - resetIndexCache(); - } - - void DBClientWithCommands::dropIndexes( const string& ns ) { - BSONObj info; - uassert( 10008 , "dropIndexes failed" , runCommand( nsToDatabase( ns.c_str() ) , - BSON( "deleteIndexes" << NamespaceString( ns ).coll << "index" << "*") , - info ) ); - resetIndexCache(); - } - - void DBClientWithCommands::reIndex( const string& ns ) { - list<BSONObj> all; - auto_ptr<DBClientCursor> i = getIndexes( ns ); - while ( i->more() ) { - all.push_back( i->next().getOwned() ); - } - - dropIndexes( ns ); - - for ( list<BSONObj>::iterator i=all.begin(); i!=all.end(); i++ ) { - BSONObj o = *i; - insert( Namespace( ns.c_str() ).getSisterNS( "system.indexes" ).c_str() , o ); - } - - } - - - string DBClientWithCommands::genIndexName( const BSONObj& keys ) { - stringstream ss; - - bool first = 1; - for ( BSONObjIterator i(keys); i.more(); ) { - BSONElement f = i.next(); - - if ( first ) - first = 0; - else - ss << "_"; - - ss << f.fieldName() << "_"; - if( f.isNumber() ) - ss << f.numberInt(); - } - return ss.str(); - } - - bool DBClientWithCommands::ensureIndex( const string &ns , BSONObj keys , bool unique, const string & name , bool cache, bool background, int version ) { - BSONObjBuilder toSave; - toSave.append( "ns" , ns ); - toSave.append( "key" , keys ); - - string cacheKey(ns); - cacheKey += "--"; - - if ( name != "" ) { - toSave.append( "name" , name ); - cacheKey += name; - } - else { - string nn = genIndexName( keys ); - toSave.append( "name" , nn ); - cacheKey += nn; - } - - if( version >= 0 ) - toSave.append("v", version); - - if ( unique ) - toSave.appendBool( "unique", unique ); - - if( background ) - toSave.appendBool( "background", true ); - - if ( _seenIndexes.count( cacheKey ) ) - return 0; - - if ( cache ) - _seenIndexes.insert( cacheKey ); - - insert( Namespace( ns.c_str() ).getSisterNS( "system.indexes" ).c_str() , toSave.obj() ); - return 1; - } - - void DBClientWithCommands::resetIndexCache() { - _seenIndexes.clear(); - } - - /* -- DBClientCursor ---------------------------------------------- */ - -#ifdef _DEBUG -#define CHECK_OBJECT( o , msg ) massert( 10337 , (string)"object not valid" + (msg) , (o).isValid() ) -#else -#define CHECK_OBJECT( o , msg ) -#endif - - void assembleRequest( const string &ns, BSONObj query, int nToReturn, int nToSkip, const BSONObj *fieldsToReturn, int queryOptions, Message &toSend ) { - CHECK_OBJECT( query , "assembleRequest query" ); - // see query.h for the protocol we are using here. - BufBuilder b; - int opts = queryOptions; - b.appendNum(opts); - b.appendStr(ns); - b.appendNum(nToSkip); - b.appendNum(nToReturn); - query.appendSelfToBufBuilder(b); - if ( fieldsToReturn ) - fieldsToReturn->appendSelfToBufBuilder(b); - toSend.setData(dbQuery, b.buf(), b.len()); - } - - void DBClientConnection::say( Message &toSend, bool isRetry ) { - checkConnection(); - try { - port().say( toSend ); - } - catch( SocketException & ) { - _failed = true; - throw; - } - } - - void DBClientConnection::sayPiggyBack( Message &toSend ) { - port().piggyBack( toSend ); - } - - bool DBClientConnection::recv( Message &m ) { - return port().recv(m); - } - - bool DBClientConnection::call( Message &toSend, Message &response, bool assertOk , string * actualServer ) { - /* todo: this is very ugly messagingport::call returns an error code AND can throw - an exception. we should make it return void and just throw an exception anytime - it fails - */ - checkConnection(); - try { - if ( !port().call(toSend, response) ) { - _failed = true; - if ( assertOk ) - uasserted( 10278 , str::stream() << "dbclient error communicating with server: " << getServerAddress() ); - - return false; - } - } - catch( SocketException & ) { - _failed = true; - throw; - } - return true; - } - - BSONElement getErrField(const BSONObj& o) { - BSONElement first = o.firstElement(); - if( strcmp(first.fieldName(), "$err") == 0 ) - return first; - - // temp - will be DEV only later - /*DEV*/ - if( 1 ) { - BSONElement e = o["$err"]; - if( !e.eoo() ) { - wassert(false); - } - return e; - } - - return BSONElement(); - } - - bool hasErrField( const BSONObj& o ){ - return ! getErrField( o ).eoo(); - } - - void DBClientConnection::checkResponse( const char *data, int nReturned, bool* retry, string* host ) { - /* check for errors. the only one we really care about at - * this stage is "not master" - */ - - *retry = false; - *host = _serverString; - - if ( clientSet && nReturned ) { - assert(data); - BSONObj o(data); - if ( isNotMasterErrorString( getErrField(o) ) ) { - clientSet->isntMaster(); - } - } - } - - void DBClientConnection::killCursor( long long cursorId ) { - StackBufBuilder b; - b.appendNum( (int)0 ); // reserved - b.appendNum( (int)1 ); // number - b.appendNum( cursorId ); - - Message m; - m.setData( dbKillCursors , b.buf() , b.len() ); - - if ( _lazyKillCursor ) - sayPiggyBack( m ); - else - say(m); - } - -#ifdef MONGO_SSL - SSLManager* DBClientConnection::sslManager() { - if ( _sslManager ) - return _sslManager; - - SSLManager* s = new SSLManager(true); - _sslManager = s; - return s; - } - - SSLManager* DBClientConnection::_sslManager = 0; -#endif - - AtomicUInt DBClientConnection::_numConnections; - bool DBClientConnection::_lazyKillCursor = true; - - - bool serverAlive( const string &uri ) { - DBClientConnection c( false, 0, 20 ); // potentially the connection to server could fail while we're checking if it's alive - so use timeouts - string err; - if ( !c.connect( uri, err ) ) - return false; - if ( !c.simpleCommand( "admin", 0, "ping" ) ) - return false; - return true; - } - -} // namespace mongo diff --git a/client/dbclient.h b/client/dbclient.h deleted file mode 100644 index ea55bb474af..00000000000 --- a/client/dbclient.h +++ /dev/null @@ -1,983 +0,0 @@ -/** @file dbclient.h - - Core MongoDB C++ driver interfaces are defined here. -*/ - -/* Copyright 2009 10gen Inc. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -#pragma once - -#include "../pch.h" -#include "../util/net/message.h" -#include "../util/net/message_port.h" -#include "../db/jsobj.h" -#include "../db/json.h" -#include <stack> - -namespace mongo { - - /** the query field 'options' can have these bits set: */ - enum QueryOptions { - /** Tailable means cursor is not closed when the last data is retrieved. rather, the cursor marks - the final object's position. you can resume using the cursor later, from where it was located, - if more data were received. Set on dbQuery and dbGetMore. - - like any "latent cursor", the cursor may become invalid at some point -- for example if that - final object it references were deleted. Thus, you should be prepared to requery if you get back - ResultFlag_CursorNotFound. - */ - QueryOption_CursorTailable = 1 << 1, - - /** allow query of replica slave. normally these return an error except for namespace "local". - */ - QueryOption_SlaveOk = 1 << 2, - - // findingStart mode is used to find the first operation of interest when - // we are scanning through a repl log. For efficiency in the common case, - // where the first operation of interest is closer to the tail than the head, - // we start from the tail of the log and work backwards until we find the - // first operation of interest. Then we scan forward from that first operation, - // actually returning results to the client. During the findingStart phase, - // we release the db mutex occasionally to avoid blocking the db process for - // an extended period of time. - QueryOption_OplogReplay = 1 << 3, - - /** The server normally times out idle cursors after an inactivy period to prevent excess memory uses - Set this option to prevent that. - */ - QueryOption_NoCursorTimeout = 1 << 4, - - /** Use with QueryOption_CursorTailable. If we are at the end of the data, block for a while rather - than returning no data. After a timeout period, we do return as normal. - */ - QueryOption_AwaitData = 1 << 5, - - /** Stream the data down full blast in multiple "more" packages, on the assumption that the client - will fully read all data queried. Faster when you are pulling a lot of data and know you want to - pull it all down. Note: it is not allowed to not read all the data unless you close the connection. - - Use the query( boost::function<void(const BSONObj&)> f, ... ) version of the connection's query() - method, and it will take care of all the details for you. - */ - QueryOption_Exhaust = 1 << 6, - - /** When sharded, this means its ok to return partial results - Usually we will fail a query if all required shards aren't up - If this is set, it'll be a partial result set - */ - QueryOption_PartialResults = 1 << 7 , - - QueryOption_AllSupported = QueryOption_CursorTailable | QueryOption_SlaveOk | QueryOption_OplogReplay | QueryOption_NoCursorTimeout | QueryOption_AwaitData | QueryOption_Exhaust | QueryOption_PartialResults - - }; - - enum UpdateOptions { - /** Upsert - that is, insert the item if no matching item is found. */ - UpdateOption_Upsert = 1 << 0, - - /** Update multiple documents (if multiple documents match query expression). - (Default is update a single document and stop.) */ - UpdateOption_Multi = 1 << 1, - - /** flag from mongo saying this update went everywhere */ - UpdateOption_Broadcast = 1 << 2 - }; - - enum RemoveOptions { - /** only delete one option */ - RemoveOption_JustOne = 1 << 0, - - /** flag from mongo saying this update went everywhere */ - RemoveOption_Broadcast = 1 << 1 - }; - - - /** - * need to put in DbMesssage::ReservedOptions as well - */ - enum InsertOptions { - /** With muli-insert keep processing inserts if one fails */ - InsertOption_ContinueOnError = 1 << 0 - }; - - class DBClientBase; - - /** - * ConnectionString handles parsing different ways to connect to mongo and determining method - * samples: - * server - * server:port - * foo/server:port,server:port SET - * server,server,server SYNC - * - * tyipcal use - * string errmsg, - * ConnectionString cs = ConnectionString::parse( url , errmsg ); - * if ( ! cs.isValid() ) throw "bad: " + errmsg; - * DBClientBase * conn = cs.connect( errmsg ); - */ - class ConnectionString { - public: - enum ConnectionType { INVALID , MASTER , PAIR , SET , SYNC }; - - ConnectionString() { - _type = INVALID; - } - - ConnectionString( const HostAndPort& server ) { - _type = MASTER; - _servers.push_back( server ); - _finishInit(); - } - - ConnectionString( ConnectionType type , const string& s , const string& setName = "" ) { - _type = type; - _setName = setName; - _fillServers( s ); - - switch ( _type ) { - case MASTER: - assert( _servers.size() == 1 ); - break; - case SET: - assert( _setName.size() ); - assert( _servers.size() >= 1 ); // 1 is ok since we can derive - break; - case PAIR: - assert( _servers.size() == 2 ); - break; - default: - assert( _servers.size() > 0 ); - } - - _finishInit(); - } - - ConnectionString( const string& s , ConnectionType favoredMultipleType ) { - _type = INVALID; - - _fillServers( s ); - if ( _type != INVALID ) { - // set already - } - else if ( _servers.size() == 1 ) { - _type = MASTER; - } - else { - _type = favoredMultipleType; - assert( _type == SET || _type == SYNC ); - } - _finishInit(); - } - - bool isValid() const { return _type != INVALID; } - - string toString() const { return _string; } - - DBClientBase* connect( string& errmsg, double socketTimeout = 0 ) const; - - string getSetName() const { return _setName; } - - vector<HostAndPort> getServers() const { return _servers; } - - ConnectionType type() const { return _type; } - - static ConnectionString parse( const string& url , string& errmsg ); - - static string typeToString( ConnectionType type ); - - private: - - void _fillServers( string s ); - void _finishInit(); - - ConnectionType _type; - vector<HostAndPort> _servers; - string _string; - string _setName; - }; - - /** - * controls how much a clients cares about writes - * default is NORMAL - */ - enum WriteConcern { - W_NONE = 0 , // TODO: not every connection type fully supports this - W_NORMAL = 1 - // TODO SAFE = 2 - }; - - class BSONObj; - class ScopedDbConnection; - class DBClientCursor; - class DBClientCursorBatchIterator; - - /** Represents a Mongo query expression. Typically one uses the QUERY(...) macro to construct a Query object. - Examples: - QUERY( "age" << 33 << "school" << "UCLA" ).sort("name") - QUERY( "age" << GT << 30 << LT << 50 ) - */ - class Query { - public: - BSONObj obj; - Query() : obj(BSONObj()) { } - Query(const BSONObj& b) : obj(b) { } - Query(const string &json) : - obj(fromjson(json)) { } - Query(const char * json) : - obj(fromjson(json)) { } - - /** Add a sort (ORDER BY) criteria to the query expression. - @param sortPattern the sort order template. For example to order by name ascending, time descending: - { name : 1, ts : -1 } - i.e. - BSON( "name" << 1 << "ts" << -1 ) - or - fromjson(" name : 1, ts : -1 ") - */ - Query& sort(const BSONObj& sortPattern); - - /** Add a sort (ORDER BY) criteria to the query expression. - This version of sort() assumes you want to sort on a single field. - @param asc = 1 for ascending order - asc = -1 for descending order - */ - Query& sort(const string &field, int asc = 1) { sort( BSON( field << asc ) ); return *this; } - - /** Provide a hint to the query. - @param keyPattern Key pattern for the index to use. - Example: - hint("{ts:1}") - */ - Query& hint(BSONObj keyPattern); - Query& hint(const string &jsonKeyPatt) { return hint(fromjson(jsonKeyPatt)); } - - /** Provide min and/or max index limits for the query. - min <= x < max - */ - Query& minKey(const BSONObj &val); - /** - max is exclusive - */ - Query& maxKey(const BSONObj &val); - - /** Return explain information about execution of this query instead of the actual query results. - Normally it is easier to use the mongo shell to run db.find(...).explain(). - */ - Query& explain(); - - /** Use snapshot mode for the query. Snapshot mode assures no duplicates are returned, or objects missed, which were - present at both the start and end of the query's execution (if an object is new during the query, or deleted during - the query, it may or may not be returned, even with snapshot mode). - - Note that short query responses (less than 1MB) are always effectively snapshotted. - - Currently, snapshot mode may not be used with sorting or explicit hints. - */ - Query& snapshot(); - - /** Queries to the Mongo database support a $where parameter option which contains - a javascript function that is evaluated to see whether objects being queried match - its criteria. Use this helper to append such a function to a query object. - Your query may also contain other traditional Mongo query terms. - - @param jscode The javascript function to evaluate against each potential object - match. The function must return true for matched objects. Use the this - variable to inspect the current object. - @param scope SavedContext for the javascript object. List in a BSON object any - variables you would like defined when the jscode executes. One can think - of these as "bind variables". - - Examples: - conn.findOne("test.coll", Query("{a:3}").where("this.b == 2 || this.c == 3")); - Query badBalance = Query().where("this.debits - this.credits < 0"); - */ - Query& where(const string &jscode, BSONObj scope); - Query& where(const string &jscode) { return where(jscode, BSONObj()); } - - /** - * @return true if this query has an orderby, hint, or some other field - */ - bool isComplex( bool * hasDollar = 0 ) const; - - BSONObj getFilter() const; - BSONObj getSort() const; - BSONObj getHint() const; - bool isExplain() const; - - string toString() const; - operator string() const { return toString(); } - private: - void makeComplex(); - template< class T > - void appendComplex( const char *fieldName, const T& val ) { - makeComplex(); - BSONObjBuilder b; - b.appendElements(obj); - b.append(fieldName, val); - obj = b.obj(); - } - }; - - /** Typically one uses the QUERY(...) macro to construct a Query object. - Example: QUERY( "age" << 33 << "school" << "UCLA" ) - */ -#define QUERY(x) mongo::Query( BSON(x) ) - - /** - interface that handles communication with the db - */ - class DBConnector { - public: - virtual ~DBConnector() {} - /** actualServer is set to the actual server where they call went if there was a choice (SlaveOk) */ - virtual bool call( Message &toSend, Message &response, bool assertOk=true , string * actualServer = 0 ) = 0; - virtual void say( Message &toSend, bool isRetry = false ) = 0; - virtual void sayPiggyBack( Message &toSend ) = 0; - /* used by QueryOption_Exhaust. To use that your subclass must implement this. */ - virtual bool recv( Message& m ) { assert(false); return false; } - // In general, for lazy queries, we'll need to say, recv, then checkResponse - virtual void checkResponse( const char* data, int nReturned, bool* retry = NULL, string* targetHost = NULL ) { - if( retry ) *retry = false; if( targetHost ) *targetHost = ""; - } - virtual bool lazySupported() const = 0; - }; - - /** - The interface that any db connection should implement - */ - class DBClientInterface : boost::noncopyable { - public: - virtual auto_ptr<DBClientCursor> query(const string &ns, Query query, int nToReturn = 0, int nToSkip = 0, - const BSONObj *fieldsToReturn = 0, int queryOptions = 0 , int batchSize = 0 ) = 0; - - virtual void insert( const string &ns, BSONObj obj , int flags=0) = 0; - - virtual void insert( const string &ns, const vector< BSONObj >& v , int flags=0) = 0; - - virtual void remove( const string &ns , Query query, bool justOne = 0 ) = 0; - - virtual void update( const string &ns , Query query , BSONObj obj , bool upsert = 0 , bool multi = 0 ) = 0; - - virtual ~DBClientInterface() { } - - /** - @return a single object that matches the query. if none do, then the object is empty - @throws AssertionException - */ - virtual BSONObj findOne(const string &ns, const Query& query, const BSONObj *fieldsToReturn = 0, int queryOptions = 0); - - /** query N objects from the database into an array. makes sense mostly when you want a small number of results. if a huge number, use - query() and iterate the cursor. - */ - void findN(vector<BSONObj>& out, const string&ns, Query query, int nToReturn, int nToSkip = 0, const BSONObj *fieldsToReturn = 0, int queryOptions = 0); - - virtual string getServerAddress() const = 0; - - /** don't use this - called automatically by DBClientCursor for you */ - virtual auto_ptr<DBClientCursor> getMore( const string &ns, long long cursorId, int nToReturn = 0, int options = 0 ) = 0; - }; - - /** - DB "commands" - Basically just invocations of connection.$cmd.findOne({...}); - */ - class DBClientWithCommands : public DBClientInterface { - set<string> _seenIndexes; - public: - /** controls how chatty the client is about network errors & such. See log.h */ - int _logLevel; - - DBClientWithCommands() : _logLevel(0), _cachedAvailableOptions( (enum QueryOptions)0 ), _haveCachedAvailableOptions(false) { } - - /** helper function. run a simple command where the command expression is simply - { command : 1 } - @param info -- where to put result object. may be null if caller doesn't need that info - @param command -- command name - @return true if the command returned "ok". - */ - bool simpleCommand(const string &dbname, BSONObj *info, const string &command); - - /** Run a database command. Database commands are represented as BSON objects. Common database - commands have prebuilt helper functions -- see below. If a helper is not available you can - directly call runCommand. - - @param dbname database name. Use "admin" for global administrative commands. - @param cmd the command object to execute. For example, { ismaster : 1 } - @param info the result object the database returns. Typically has { ok : ..., errmsg : ... } fields - set. - @param options see enum QueryOptions - normally not needed to run a command - @return true if the command returned "ok". - */ - virtual bool runCommand(const string &dbname, const BSONObj& cmd, BSONObj &info, int options=0); - - /** Authorize access to a particular database. - Authentication is separate for each database on the server -- you may authenticate for any - number of databases on a single connection. - The "admin" database is special and once authenticated provides access to all databases on the - server. - @param digestPassword if password is plain text, set this to true. otherwise assumed to be pre-digested - @return true if successful - */ - virtual bool auth(const string &dbname, const string &username, const string &pwd, string& errmsg, bool digestPassword = true); - - /** count number of objects in collection ns that match the query criteria specified - throws UserAssertion if database returns an error - */ - virtual unsigned long long count(const string &ns, const BSONObj& query = BSONObj(), int options=0, int limit=0, int skip=0 ); - - string createPasswordDigest( const string &username , const string &clearTextPassword ); - - /** returns true in isMaster parm if this db is the current master - of a replica pair. - - pass in info for more details e.g.: - { "ismaster" : 1.0 , "msg" : "not paired" , "ok" : 1.0 } - - returns true if command invoked successfully. - */ - virtual bool isMaster(bool& isMaster, BSONObj *info=0); - - /** - Create a new collection in the database. Normally, collection creation is automatic. You would - use this function if you wish to specify special options on creation. - - If the collection already exists, no action occurs. - - @param ns fully qualified collection name - @param size desired initial extent size for the collection. - Must be <= 1000000000 for normal collections. - For fixed size (capped) collections, this size is the total/max size of the - collection. - @param capped if true, this is a fixed size collection (where old data rolls out). - @param max maximum number of objects if capped (optional). - - returns true if successful. - */ - bool createCollection(const string &ns, long long size = 0, bool capped = false, int max = 0, BSONObj *info = 0); - - /** Get error result from the last write operation (insert/update/delete) on this connection. - @return error message text, or empty string if no error. - */ - string getLastError(); - - /** Get error result from the last write operation (insert/update/delete) on this connection. - @return full error object. - */ - virtual BSONObj getLastErrorDetailed(); - - /** Can be called with the returned value from getLastErrorDetailed to extract an error string. - If all you need is the string, just call getLastError() instead. - */ - static string getLastErrorString( const BSONObj& res ); - - /** Return the last error which has occurred, even if not the very last operation. - - @return { err : <error message>, nPrev : <how_many_ops_back_occurred>, ok : 1 } - - result.err will be null if no error has occurred. - */ - BSONObj getPrevError(); - - /** Reset the previous error state for this connection (accessed via getLastError and - getPrevError). Useful when performing several operations at once and then checking - for an error after attempting all operations. - */ - bool resetError() { return simpleCommand("admin", 0, "reseterror"); } - - /** Delete the specified collection. */ - virtual bool dropCollection( const string &ns ) { - string db = nsGetDB( ns ); - string coll = nsGetCollection( ns ); - uassert( 10011 , "no collection name", coll.size() ); - - BSONObj info; - - bool res = runCommand( db.c_str() , BSON( "drop" << coll ) , info ); - resetIndexCache(); - return res; - } - - /** Perform a repair and compaction of the specified database. May take a long time to run. Disk space - must be available equal to the size of the database while repairing. - */ - bool repairDatabase(const string &dbname, BSONObj *info = 0) { - return simpleCommand(dbname, info, "repairDatabase"); - } - - /** Copy database from one server or name to another server or name. - - Generally, you should dropDatabase() first as otherwise the copied information will MERGE - into whatever data is already present in this database. - - For security reasons this function only works when you are authorized to access the "admin" db. However, - if you have access to said db, you can copy any database from one place to another. - TODO: this needs enhancement to be more flexible in terms of security. - - This method provides a way to "rename" a database by copying it to a new db name and - location. The copy is "repaired" and compacted. - - fromdb database name from which to copy. - todb database name to copy to. - fromhost hostname of the database (and optionally, ":port") from which to - copy the data. copies from self if "". - - returns true if successful - */ - bool copyDatabase(const string &fromdb, const string &todb, const string &fromhost = "", BSONObj *info = 0); - - /** The Mongo database provides built-in performance profiling capabilities. Uset setDbProfilingLevel() - to enable. Profiling information is then written to the system.profiling collection, which one can - then query. - */ - enum ProfilingLevel { - ProfileOff = 0, - ProfileSlow = 1, // log very slow (>100ms) operations - ProfileAll = 2 - - }; - bool setDbProfilingLevel(const string &dbname, ProfilingLevel level, BSONObj *info = 0); - bool getDbProfilingLevel(const string &dbname, ProfilingLevel& level, BSONObj *info = 0); - - - /** This implicitly converts from char*, string, and BSONObj to be an argument to mapreduce - You shouldn't need to explicitly construct this - */ - struct MROutput { - MROutput(const char* collection) : out(BSON("replace" << collection)) {} - MROutput(const string& collection) : out(BSON("replace" << collection)) {} - MROutput(const BSONObj& obj) : out(obj) {} - - BSONObj out; - }; - static MROutput MRInline; - - /** Run a map/reduce job on the server. - - See http://www.mongodb.org/display/DOCS/MapReduce - - ns namespace (db+collection name) of input data - jsmapf javascript map function code - jsreducef javascript reduce function code. - query optional query filter for the input - output either a string collection name or an object representing output type - if not specified uses inline output type - - returns a result object which contains: - { result : <collection_name>, - numObjects : <number_of_objects_scanned>, - timeMillis : <job_time>, - ok : <1_if_ok>, - [, err : <errmsg_if_error>] - } - - For example one might call: - result.getField("ok").trueValue() - on the result to check if ok. - */ - BSONObj mapreduce(const string &ns, const string &jsmapf, const string &jsreducef, BSONObj query = BSONObj(), MROutput output = MRInline); - - /** Run javascript code on the database server. - dbname database SavedContext in which the code runs. The javascript variable 'db' will be assigned - to this database when the function is invoked. - jscode source code for a javascript function. - info the command object which contains any information on the invocation result including - the return value and other information. If an error occurs running the jscode, error - information will be in info. (try "out() << info.toString()") - retValue return value from the jscode function. - args args to pass to the jscode function. when invoked, the 'args' variable will be defined - for use by the jscode. - - returns true if runs ok. - - See testDbEval() in dbclient.cpp for an example of usage. - */ - bool eval(const string &dbname, const string &jscode, BSONObj& info, BSONElement& retValue, BSONObj *args = 0); - - /** validate a collection, checking for errors and reporting back statistics. - this operation is slow and blocking. - */ - bool validate( const string &ns , bool scandata=true ) { - BSONObj cmd = BSON( "validate" << nsGetCollection( ns ) << "scandata" << scandata ); - BSONObj info; - return runCommand( nsGetDB( ns ).c_str() , cmd , info ); - } - - /* The following helpers are simply more convenient forms of eval() for certain common cases */ - - /* invocation with no return value of interest -- with or without one simple parameter */ - bool eval(const string &dbname, const string &jscode); - template< class T > - bool eval(const string &dbname, const string &jscode, T parm1) { - BSONObj info; - BSONElement retValue; - BSONObjBuilder b; - b.append("0", parm1); - BSONObj args = b.done(); - return eval(dbname, jscode, info, retValue, &args); - } - - /** eval invocation with one parm to server and one numeric field (either int or double) returned */ - template< class T, class NumType > - bool eval(const string &dbname, const string &jscode, T parm1, NumType& ret) { - BSONObj info; - BSONElement retValue; - BSONObjBuilder b; - b.append("0", parm1); - BSONObj args = b.done(); - if ( !eval(dbname, jscode, info, retValue, &args) ) - return false; - ret = (NumType) retValue.number(); - return true; - } - - /** - get a list of all the current databases - uses the { listDatabases : 1 } command. - throws on error - */ - list<string> getDatabaseNames(); - - /** - get a list of all the current collections in db - */ - list<string> getCollectionNames( const string& db ); - - bool exists( const string& ns ); - - /** Create an index if it does not already exist. - ensureIndex calls are remembered so it is safe/fast to call this function many - times in your code. - @param ns collection to be indexed - @param keys the "key pattern" for the index. e.g., { name : 1 } - @param unique if true, indicates that key uniqueness should be enforced for this index - @param name if not specified, it will be created from the keys automatically (which is recommended) - @param cache if set to false, the index cache for the connection won't remember this call - @param background build index in the background (see mongodb docs/wiki for details) - @param v index version. leave at default value. (unit tests set this parameter.) - @return whether or not sent message to db. - should be true on first call, false on subsequent unless resetIndexCache was called - */ - virtual bool ensureIndex( const string &ns , BSONObj keys , bool unique = false, const string &name = "", - bool cache = true, bool background = false, int v = -1 ); - - /** - clears the index cache, so the subsequent call to ensureIndex for any index will go to the server - */ - virtual void resetIndexCache(); - - virtual auto_ptr<DBClientCursor> getIndexes( const string &ns ); - - virtual void dropIndex( const string& ns , BSONObj keys ); - virtual void dropIndex( const string& ns , const string& indexName ); - - /** - drops all indexes for the collection - */ - virtual void dropIndexes( const string& ns ); - - virtual void reIndex( const string& ns ); - - string genIndexName( const BSONObj& keys ); - - /** Erase / drop an entire database */ - virtual bool dropDatabase(const string &dbname, BSONObj *info = 0) { - bool ret = simpleCommand(dbname, info, "dropDatabase"); - resetIndexCache(); - return ret; - } - - virtual string toString() = 0; - - /** @return the database name portion of an ns string */ - string nsGetDB( const string &ns ) { - string::size_type pos = ns.find( "." ); - if ( pos == string::npos ) - return ns; - - return ns.substr( 0 , pos ); - } - - /** @return the collection name portion of an ns string */ - string nsGetCollection( const string &ns ) { - string::size_type pos = ns.find( "." ); - if ( pos == string::npos ) - return ""; - - return ns.substr( pos + 1 ); - } - - protected: - /** if the result of a command is ok*/ - bool isOk(const BSONObj&); - - /** if the element contains a not master error */ - bool isNotMasterErrorString( const BSONElement& e ); - - BSONObj _countCmd(const string &ns, const BSONObj& query, int options, int limit, int skip ); - - enum QueryOptions availableOptions(); - - private: - enum QueryOptions _cachedAvailableOptions; - bool _haveCachedAvailableOptions; - }; - - /** - abstract class that implements the core db operations - */ - class DBClientBase : public DBClientWithCommands, public DBConnector { - protected: - WriteConcern _writeConcern; - - public: - DBClientBase() { - _writeConcern = W_NORMAL; - } - - WriteConcern getWriteConcern() const { return _writeConcern; } - void setWriteConcern( WriteConcern w ) { _writeConcern = w; } - - /** send a query to the database. - @param ns namespace to query, format is <dbname>.<collectname>[.<collectname>]* - @param query query to perform on the collection. this is a BSONObj (binary JSON) - You may format as - { query: { ... }, orderby: { ... } } - to specify a sort order. - @param nToReturn n to return (i.e., limit). 0 = unlimited - @param nToSkip start with the nth item - @param fieldsToReturn optional template of which fields to select. if unspecified, returns all fields - @param queryOptions see options enum at top of this file - - @return cursor. 0 if error (connection failure) - @throws AssertionException - */ - virtual auto_ptr<DBClientCursor> query(const string &ns, Query query, int nToReturn = 0, int nToSkip = 0, - const BSONObj *fieldsToReturn = 0, int queryOptions = 0 , int batchSize = 0 ); - - /** don't use this - called automatically by DBClientCursor for you - @param cursorId id of cursor to retrieve - @return an handle to a previously allocated cursor - @throws AssertionException - */ - virtual auto_ptr<DBClientCursor> getMore( const string &ns, long long cursorId, int nToReturn = 0, int options = 0 ); - - /** - insert an object into the database - */ - virtual void insert( const string &ns , BSONObj obj , int flags=0); - - /** - insert a vector of objects into the database - */ - virtual void insert( const string &ns, const vector< BSONObj >& v , int flags=0); - - /** - remove matching objects from the database - @param justOne if this true, then once a single match is found will stop - */ - virtual void remove( const string &ns , Query q , bool justOne = 0 ); - - /** - updates objects matching query - */ - virtual void update( const string &ns , Query query , BSONObj obj , bool upsert = false , bool multi = false ); - - virtual bool isFailed() const = 0; - - virtual void killCursor( long long cursorID ) = 0; - - virtual bool callRead( Message& toSend , Message& response ) = 0; - // virtual bool callWrite( Message& toSend , Message& response ) = 0; // TODO: add this if needed - - virtual ConnectionString::ConnectionType type() const = 0; - - virtual double getSoTimeout() const = 0; - - }; // DBClientBase - - class DBClientReplicaSet; - - class ConnectException : public UserException { - public: - ConnectException(string msg) : UserException(9000,msg) { } - }; - - /** - A basic connection to the database. - This is the main entry point for talking to a simple Mongo setup - */ - class DBClientConnection : public DBClientBase { - public: - /** - @param _autoReconnect if true, automatically reconnect on a connection failure - @param cp used by DBClientReplicaSet. You do not need to specify this parameter - @param timeout tcp timeout in seconds - this is for read/write, not connect. - Connect timeout is fixed, but short, at 5 seconds. - */ - DBClientConnection(bool _autoReconnect=false, DBClientReplicaSet* cp=0, double so_timeout=0) : - clientSet(cp), _failed(false), autoReconnect(_autoReconnect), lastReconnectTry(0), _so_timeout(so_timeout) { - _numConnections++; - } - - virtual ~DBClientConnection() { - _numConnections--; - } - - /** Connect to a Mongo database server. - - If autoReconnect is true, you can try to use the DBClientConnection even when - false was returned -- it will try to connect again. - - @param serverHostname host to connect to. can include port number ( 127.0.0.1 , 127.0.0.1:5555 ) - If you use IPv6 you must add a port number ( ::1:27017 ) - @param errmsg any relevant error message will appended to the string - @deprecated please use HostAndPort - @return false if fails to connect. - */ - virtual bool connect(const char * hostname, string& errmsg) { - // TODO: remove this method - HostAndPort t( hostname ); - return connect( t , errmsg ); - } - - /** Connect to a Mongo database server. - - If autoReconnect is true, you can try to use the DBClientConnection even when - false was returned -- it will try to connect again. - - @param server server to connect to. - @param errmsg any relevant error message will appended to the string - @return false if fails to connect. - */ - virtual bool connect(const HostAndPort& server, string& errmsg); - - /** Connect to a Mongo database server. Exception throwing version. - Throws a UserException if cannot connect. - - If autoReconnect is true, you can try to use the DBClientConnection even when - false was returned -- it will try to connect again. - - @param serverHostname host to connect to. can include port number ( 127.0.0.1 , 127.0.0.1:5555 ) - */ - void connect(const string& serverHostname) { - string errmsg; - if( !connect(HostAndPort(serverHostname), errmsg) ) - throw ConnectException(string("can't connect ") + errmsg); - } - - virtual bool auth(const string &dbname, const string &username, const string &pwd, string& errmsg, bool digestPassword = true); - - virtual auto_ptr<DBClientCursor> query(const string &ns, Query query=Query(), int nToReturn = 0, int nToSkip = 0, - const BSONObj *fieldsToReturn = 0, int queryOptions = 0 , int batchSize = 0 ) { - checkConnection(); - return DBClientBase::query( ns, query, nToReturn, nToSkip, fieldsToReturn, queryOptions , batchSize ); - } - - /** Uses QueryOption_Exhaust - Exhaust mode sends back all data queries as fast as possible, with no back-and-for for OP_GETMORE. If you are certain - you will exhaust the query, it could be useful. - - Use DBClientCursorBatchIterator version if you want to do items in large blocks, perhaps to avoid granular locking and such. - */ - unsigned long long query( boost::function<void(const BSONObj&)> f, const string& ns, Query query, const BSONObj *fieldsToReturn = 0, int queryOptions = 0); - unsigned long long query( boost::function<void(DBClientCursorBatchIterator&)> f, const string& ns, Query query, const BSONObj *fieldsToReturn = 0, int queryOptions = 0); - - virtual bool runCommand(const string &dbname, const BSONObj& cmd, BSONObj &info, int options=0); - - /** - @return true if this connection is currently in a failed state. When autoreconnect is on, - a connection will transition back to an ok state after reconnecting. - */ - bool isFailed() const { return _failed; } - - MessagingPort& port() { assert(p); return *p; } - - string toStringLong() const { - stringstream ss; - ss << _serverString; - if ( _failed ) ss << " failed"; - return ss.str(); - } - - /** Returns the address of the server */ - string toString() { return _serverString; } - - string getServerAddress() const { return _serverString; } - - virtual void killCursor( long long cursorID ); - virtual bool callRead( Message& toSend , Message& response ) { return call( toSend , response ); } - virtual void say( Message &toSend, bool isRetry = false ); - virtual bool recv( Message& m ); - virtual void checkResponse( const char *data, int nReturned, bool* retry = NULL, string* host = NULL ); - virtual bool call( Message &toSend, Message &response, bool assertOk = true , string * actualServer = 0 ); - virtual ConnectionString::ConnectionType type() const { return ConnectionString::MASTER; } - void setSoTimeout(double to) { _so_timeout = to; } - double getSoTimeout() const { return _so_timeout; } - - virtual bool lazySupported() const { return true; } - - static int getNumConnections() { - return _numConnections; - } - - static void setLazyKillCursor( bool lazy ) { _lazyKillCursor = lazy; } - static bool getLazyKillCursor() { return _lazyKillCursor; } - - protected: - friend class SyncClusterConnection; - virtual void sayPiggyBack( Message &toSend ); - - DBClientReplicaSet *clientSet; - boost::scoped_ptr<MessagingPort> p; - boost::scoped_ptr<SockAddr> server; - bool _failed; - const bool autoReconnect; - time_t lastReconnectTry; - HostAndPort _server; // remember for reconnects - string _serverString; - void _checkConnection(); - - // throws SocketException if in failed state and not reconnecting or if waiting to reconnect - void checkConnection() { if( _failed ) _checkConnection(); } - - map< string, pair<string,string> > authCache; - double _so_timeout; - bool _connect( string& errmsg ); - - static AtomicUInt _numConnections; - static bool _lazyKillCursor; // lazy means we piggy back kill cursors on next op - -#ifdef MONGO_SSL - static SSLManager* sslManager(); - static SSLManager* _sslManager; -#endif - }; - - /** pings server to check if it's up - */ - bool serverAlive( const string &uri ); - - DBClientBase * createDirectClient(); - - BSONElement getErrField( const BSONObj& result ); - bool hasErrField( const BSONObj& result ); - -} // namespace mongo - -#include "dbclientcursor.h" -#include "dbclient_rs.h" -#include "undef_macros.h" diff --git a/client/dbclient_rs.cpp b/client/dbclient_rs.cpp deleted file mode 100644 index be66bb451c4..00000000000 --- a/client/dbclient_rs.cpp +++ /dev/null @@ -1,1097 +0,0 @@ -// dbclient.cpp - connect to a Mongo database as a database, from C++ - -/* Copyright 2009 10gen Inc. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -#include "pch.h" -#include "dbclient.h" -#include "../bson/util/builder.h" -#include "../db/jsobj.h" -#include "../db/json.h" -#include "../db/dbmessage.h" -#include "connpool.h" -#include "dbclient_rs.h" -#include "../util/background.h" - -namespace mongo { - - // -------------------------------- - // ----- ReplicaSetMonitor --------- - // -------------------------------- - - // global background job responsible for checking every X amount of time - class ReplicaSetMonitorWatcher : public BackgroundJob { - public: - ReplicaSetMonitorWatcher() : _safego("ReplicaSetMonitorWatcher::_safego") , _started(false) {} - - virtual string name() const { return "ReplicaSetMonitorWatcher"; } - - void safeGo() { - // check outside of lock for speed - if ( _started ) - return; - - scoped_lock lk( _safego ); - if ( _started ) - return; - _started = true; - - go(); - } - protected: - void run() { - log() << "starting" << endl; - while ( ! inShutdown() ) { - sleepsecs( 10 ); - try { - ReplicaSetMonitor::checkAll( true ); - } - catch ( std::exception& e ) { - error() << "check failed: " << e.what() << endl; - } - catch ( ... ) { - error() << "unkown error" << endl; - } - } - } - - mongo::mutex _safego; - bool _started; - - } replicaSetMonitorWatcher; - - string seedString( const vector<HostAndPort>& servers ){ - string seedStr; - for ( unsigned i = 0; i < servers.size(); i++ ){ - seedStr += servers[i].toString(); - if( i < servers.size() - 1 ) seedStr += ","; - } - - return seedStr; - } - - ReplicaSetMonitor::ReplicaSetMonitor( const string& name , const vector<HostAndPort>& servers ) - : _lock( "ReplicaSetMonitor instance" ) , _checkConnectionLock( "ReplicaSetMonitor check connection lock" ), _name( name ) , _master(-1), _nextSlave(0) { - - uassert( 13642 , "need at least 1 node for a replica set" , servers.size() > 0 ); - - if ( _name.size() == 0 ) { - warning() << "replica set name empty, first node: " << servers[0] << endl; - } - - log() << "starting new replica set monitor for replica set " << _name << " with seed of " << seedString( servers ) << endl; - - string errmsg; - for ( unsigned i = 0; i < servers.size(); i++ ) { - - // Don't check servers we have already - if( _find_inlock( servers[i] ) >= 0 ) continue; - - auto_ptr<DBClientConnection> conn( new DBClientConnection( true , 0, 5.0 ) ); - try{ - if( ! conn->connect( servers[i] , errmsg ) ){ - throw DBException( errmsg, 15928 ); - } - log() << "successfully connected to seed " << servers[i] << " for replica set " << this->_name << endl; - } - catch( DBException& e ){ - log() << "error connecting to seed " << servers[i] << causedBy( e ) << endl; - // skip seeds that don't work - continue; - } - - string maybePrimary; - _checkConnection( conn.get(), maybePrimary, false, -1 ); - } - - // Check everything to get the first data - _check( true ); - - log() << "replica set monitor for replica set " << _name << " started, address is " << getServerAddress() << endl; - - } - - ReplicaSetMonitor::~ReplicaSetMonitor() { - _nodes.clear(); - _master = -1; - } - - ReplicaSetMonitorPtr ReplicaSetMonitor::get( const string& name , const vector<HostAndPort>& servers ) { - scoped_lock lk( _setsLock ); - ReplicaSetMonitorPtr& m = _sets[name]; - if ( ! m ) - m.reset( new ReplicaSetMonitor( name , servers ) ); - - replicaSetMonitorWatcher.safeGo(); - - return m; - } - - ReplicaSetMonitorPtr ReplicaSetMonitor::get( const string& name ) { - scoped_lock lk( _setsLock ); - map<string,ReplicaSetMonitorPtr>::const_iterator i = _sets.find( name ); - if ( i == _sets.end() ) - return ReplicaSetMonitorPtr(); - return i->second; - } - - - void ReplicaSetMonitor::checkAll( bool checkAllSecondaries ) { - set<string> seen; - - while ( true ) { - ReplicaSetMonitorPtr m; - { - scoped_lock lk( _setsLock ); - for ( map<string,ReplicaSetMonitorPtr>::iterator i=_sets.begin(); i!=_sets.end(); ++i ) { - string name = i->first; - if ( seen.count( name ) ) - continue; - LOG(1) << "checking replica set: " << name << endl; - seen.insert( name ); - m = i->second; - break; - } - } - - if ( ! m ) - break; - - m->check( checkAllSecondaries ); - } - - - } - - void ReplicaSetMonitor::setConfigChangeHook( ConfigChangeHook hook ) { - massert( 13610 , "ConfigChangeHook already specified" , _hook == 0 ); - _hook = hook; - } - - string ReplicaSetMonitor::getServerAddress() const { - scoped_lock lk( _lock ); - return _getServerAddress_inlock(); - } - - string ReplicaSetMonitor::_getServerAddress_inlock() const { - StringBuilder ss; - if ( _name.size() ) - ss << _name << "/"; - - for ( unsigned i=0; i<_nodes.size(); i++ ) { - if ( i > 0 ) - ss << ","; - ss << _nodes[i].addr.toString(); - } - - return ss.str(); - } - - bool ReplicaSetMonitor::contains( const string& server ) const { - scoped_lock lk( _lock ); - for ( unsigned i=0; i<_nodes.size(); i++ ) { - if ( _nodes[i].addr == server ) - return true; - } - return false; - } - - - void ReplicaSetMonitor::notifyFailure( const HostAndPort& server ) { - scoped_lock lk( _lock ); - - if ( _master >= 0 && _master < (int)_nodes.size() ) { - if ( server == _nodes[_master].addr ) { - _nodes[_master].ok = false; - _master = -1; - } - } - } - - - - HostAndPort ReplicaSetMonitor::getMaster() { - { - scoped_lock lk( _lock ); - assert(_master < static_cast<int>(_nodes.size())); - if ( _master >= 0 && _nodes[_master].ok ) - return _nodes[_master].addr; - } - - _check( false ); - - scoped_lock lk( _lock ); - uassert( 10009 , str::stream() << "ReplicaSetMonitor no master found for set: " << _name , _master >= 0 ); - assert(_master < static_cast<int>(_nodes.size())); - return _nodes[_master].addr; - } - - HostAndPort ReplicaSetMonitor::getSlave( const HostAndPort& prev ) { - // make sure its valid - - bool wasFound = false; - - // This is always true, since checked in port() - assert( prev.port() >= 0 ); - if( prev.host().size() ){ - scoped_lock lk( _lock ); - for ( unsigned i=0; i<_nodes.size(); i++ ) { - if ( prev != _nodes[i].addr ) - continue; - - wasFound = true; - - if ( _nodes[i].okForSecondaryQueries() ) - return prev; - - break; - } - } - - if( prev.host().size() ){ - if( wasFound ){ LOG(1) << "slave '" << prev << "' is no longer ok to use" << endl; } - else{ LOG(1) << "slave '" << prev << "' was not found in the replica set" << endl; } - } - else LOG(1) << "slave '" << prev << "' is not initialized or invalid" << endl; - - return getSlave(); - } - - HostAndPort ReplicaSetMonitor::getSlave() { - LOG(2) << "dbclient_rs getSlave " << getServerAddress() << endl; - - scoped_lock lk( _lock ); - - for ( unsigned ii = 0; ii < _nodes.size(); ii++ ) { - _nextSlave = ( _nextSlave + 1 ) % _nodes.size(); - if ( _nextSlave != _master ) { - if ( _nodes[ _nextSlave ].okForSecondaryQueries() ) - return _nodes[ _nextSlave ].addr; - LOG(2) << "dbclient_rs getSlave not selecting " << _nodes[_nextSlave] << ", not currently okForSecondaryQueries" << endl; - } - } - - if( _master >= 0 ) { - assert( static_cast<unsigned>(_master) < _nodes.size() ); - LOG(2) << "dbclient_rs getSlave no member in secondary state found, returning primary " << _nodes[ _master ] << endl; - return _nodes[_master].addr; - } - - LOG(2) << "dbclient_rs getSlave no suitable member found, returning first node " << _nodes[ 0 ] << endl; - assert( _nodes.size() > 0 ); - return _nodes[0].addr; - } - - /** - * notify the monitor that server has failed - */ - void ReplicaSetMonitor::notifySlaveFailure( const HostAndPort& server ) { - int x = _find( server ); - if ( x >= 0 ) { - scoped_lock lk( _lock ); - _nodes[x].ok = false; - } - } - - void ReplicaSetMonitor::_checkStatus( const string& hostAddr ) { - BSONObj status; - - /* replSetGetStatus requires admin auth so use a connection from the pool, - * which are authenticated with the keyFile credentials. - */ - ScopedDbConnection authenticatedConn( hostAddr ); - - if ( !authenticatedConn->runCommand( "admin", BSON( "replSetGetStatus" << 1 ), status )) { - LOG(1) << "dbclient_rs replSetGetStatus failed" << endl; - authenticatedConn.done(); // connection worked properly, but we got an error from server - return; - } - - // Make sure we return when finished - authenticatedConn.done(); - - if( !status.hasField("members") ) { - log() << "dbclient_rs error expected members field in replSetGetStatus result" << endl; - return; - } - if( status["members"].type() != Array) { - log() << "dbclient_rs error expected members field in replSetGetStatus result to be an array" << endl; - return; - } - - BSONObjIterator hi(status["members"].Obj()); - while (hi.more()) { - BSONObj member = hi.next().Obj(); - string host = member["name"].String(); - - int m = -1; - if ((m = _find(host)) < 0) { - continue; - } - - double state = member["state"].Number(); - if (member["health"].Number() == 1 && (state == 1 || state == 2)) { - scoped_lock lk( _lock ); - _nodes[m].ok = true; - } - else { - scoped_lock lk( _lock ); - _nodes[m].ok = false; - } - } - } - - NodeDiff ReplicaSetMonitor::_getHostDiff_inlock( const BSONObj& hostList ){ - - NodeDiff diff; - set<int> nodesFound; - - int index = 0; - BSONObjIterator hi( hostList ); - while( hi.more() ){ - - string toCheck = hi.next().String(); - int nodeIndex = _find_inlock( toCheck ); - - // Node-to-add - if( nodeIndex < 0 ) diff.first.insert( toCheck ); - else nodesFound.insert( nodeIndex ); - - index++; - } - - for( size_t i = 0; i < _nodes.size(); i++ ){ - if( nodesFound.find( static_cast<int>(i) ) == nodesFound.end() ) diff.second.insert( static_cast<int>(i) ); - } - - return diff; - } - - bool ReplicaSetMonitor::_shouldChangeHosts( const BSONObj& hostList, bool inlock ){ - - int origHosts = 0; - if( ! inlock ){ - scoped_lock lk( _lock ); - origHosts = _nodes.size(); - } - else origHosts = _nodes.size(); - int numHosts = 0; - bool changed = false; - - BSONObjIterator hi(hostList); - while ( hi.more() ) { - string toCheck = hi.next().String(); - - numHosts++; - int index = 0; - if( ! inlock ) index = _find( toCheck ); - else index = _find_inlock( toCheck ); - - if ( index >= 0 ) continue; - - changed = true; - break; - } - - return changed || origHosts != numHosts; - - } - - void ReplicaSetMonitor::_checkHosts( const BSONObj& hostList, bool& changed ) { - - // Fast path, still requires intermittent locking - if( ! _shouldChangeHosts( hostList, false ) ){ - changed = false; - return; - } - - // Slow path, double-checked though - scoped_lock lk( _lock ); - - // Our host list may have changed while waiting for another thread in the meantime, - // so double-check here - // TODO: Do we really need this much protection, this should be pretty rare and not - // triggered from lots of threads, duping old behavior for safety - if( ! _shouldChangeHosts( hostList, true ) ){ - changed = false; - return; - } - - // LogLevel can be pretty low, since replica set reconfiguration should be pretty rare and - // we want to record our changes - log() << "changing hosts to " << hostList << " from " << _getServerAddress_inlock() << endl; - - NodeDiff diff = _getHostDiff_inlock( hostList ); - set<string> added = diff.first; - set<int> removed = diff.second; - - assert( added.size() > 0 || removed.size() > 0 ); - changed = true; - - // Delete from the end so we don't invalidate as we delete, delete indices are ascending - for( set<int>::reverse_iterator i = removed.rbegin(), end = removed.rend(); i != end; ++i ){ - - log() << "erasing host " << _nodes[ *i ] << " from replica set " << this->_name << endl; - _nodes.erase( _nodes.begin() + *i ); - } - - // Add new nodes - for( set<string>::iterator i = added.begin(), end = added.end(); i != end; ++i ){ - - log() << "trying to add new host " << *i << " to replica set " << this->_name << endl; - - // Connect to new node - HostAndPort h( *i ); - DBClientConnection * newConn = new DBClientConnection( true, 0, 5.0 ); - - string errmsg; - try{ - if( ! newConn->connect( h , errmsg ) ){ - throw DBException( errmsg, 15927 ); - } - log() << "successfully connected to new host " << *i << " in replica set " << this->_name << endl; - } - catch( DBException& e ){ - warning() << "cannot connect to new host " << *i << " to replica set " << this->_name << causedBy( e ) << endl; - } - - _nodes.push_back( Node( h , newConn ) ); - } - - // Invalidate the cached _master index since the _nodes structure has - // already been modified. - _master = -1; - } - - - bool ReplicaSetMonitor::_checkConnection( DBClientConnection* conn, - string& maybePrimary, bool verbose, int nodesOffset ) { - - assert( conn ); - - scoped_lock lk( _checkConnectionLock ); - bool isMaster = false; - bool changed = false; - bool errorOccured = false; - - if ( nodesOffset >= 0 ){ - scoped_lock lk( _lock ); - if ( !_checkConnMatch_inlock( conn, nodesOffset )) { - /* Another thread modified _nodes -> invariant broken. - * This also implies that another thread just passed - * through here and refreshed _nodes. So no need to do - * duplicate work. - */ - return false; - } - } - - try { - Timer t; - BSONObj o; - conn->isMaster( isMaster, &o ); - - if ( o["setName"].type() != String || o["setName"].String() != _name ) { - warning() << "node: " << conn->getServerAddress() - << " isn't a part of set: " << _name - << " ismaster: " << o << endl; - - if ( nodesOffset >= 0 ) { - scoped_lock lk( _lock ); - _nodes[nodesOffset].ok = false; - } - - return false; - } - - if ( nodesOffset >= 0 ) { - scoped_lock lk( _lock ); - - _nodes[nodesOffset].pingTimeMillis = t.millis(); - _nodes[nodesOffset].hidden = o["hidden"].trueValue(); - _nodes[nodesOffset].secondary = o["secondary"].trueValue(); - _nodes[nodesOffset].ismaster = o["ismaster"].trueValue(); - - _nodes[nodesOffset].lastIsMaster = o.copy(); - } - - log( ! verbose ) << "ReplicaSetMonitor::_checkConnection: " << conn->toString() - << ' ' << o << endl; - - // add other nodes - BSONArrayBuilder b; - if ( o["hosts"].type() == Array ) { - if ( o["primary"].type() == String ) - maybePrimary = o["primary"].String(); - - BSONObjIterator it( o["hosts"].Obj() ); - while( it.more() ) b.append( it.next() ); - } - - if (o.hasField("passives") && o["passives"].type() == Array) { - BSONObjIterator it( o["passives"].Obj() ); - while( it.more() ) b.append( it.next() ); - } - - _checkHosts( b.arr(), changed); - _checkStatus( conn->getServerAddress() ); - - } - catch ( std::exception& e ) { - log( ! verbose ) << "ReplicaSetMonitor::_checkConnection: caught exception " - << conn->toString() << ' ' << e.what() << endl; - - errorOccured = true; - } - - if ( errorOccured && nodesOffset >= 0 ) { - scoped_lock lk( _lock ); - - if (_checkConnMatch_inlock(conn, nodesOffset)) { - // Make sure _checkHosts didn't modify the _nodes structure - _nodes[nodesOffset].ok = false; - } - } - - if ( changed && _hook ) - _hook( this ); - - return isMaster; - } - - void ReplicaSetMonitor::_check( bool checkAllSecondaries ) { - LOG(1) << "_check : " << getServerAddress() << endl; - - int newMaster = -1; - shared_ptr<DBClientConnection> nodeConn; - - for ( int retry = 0; retry < 2; retry++ ) { - bool triedQuickCheck = false; - - if ( !checkAllSecondaries ) { - scoped_lock lk( _lock ); - assert(_master < static_cast<int>(_nodes.size())); - if ( _master >= 0 ) { - /* Nothing else to do since another thread already - * found the _master - */ - return; - } - } - - for ( unsigned i = 0; /* should not check while outside of lock! */ ; i++ ) { - { - scoped_lock lk( _lock ); - if ( i >= _nodes.size() ) break; - nodeConn = _nodes[i].conn; - } - - string maybePrimary; - if ( _checkConnection( nodeConn.get(), maybePrimary, retry, i ) ) { - scoped_lock lk( _lock ); - if ( _checkConnMatch_inlock( nodeConn.get(), i )) { - _master = i; - newMaster = i; - - if ( !checkAllSecondaries ) - return; - } - else { - /* - * Somebody modified _nodes and most likely set the new - * _master, so try again. - */ - break; - } - } - - - if ( ! triedQuickCheck && ! maybePrimary.empty() ) { - int probablePrimaryIdx = -1; - shared_ptr<DBClientConnection> probablePrimaryConn; - - { - scoped_lock lk( _lock ); - probablePrimaryIdx = _find_inlock( maybePrimary ); - probablePrimaryConn = _nodes[probablePrimaryIdx].conn; - } - - if ( probablePrimaryIdx >= 0 ) { - triedQuickCheck = true; - - string dummy; - if ( _checkConnection( probablePrimaryConn.get(), dummy, - false, probablePrimaryIdx ) ) { - - scoped_lock lk( _lock ); - - if ( _checkConnMatch_inlock( probablePrimaryConn.get(), - probablePrimaryIdx )) { - - _master = probablePrimaryIdx; - newMaster = probablePrimaryIdx; - - if ( ! checkAllSecondaries ) - return; - } - else { - /* - * Somebody modified _nodes and most likely set the - * new _master, so try again. - */ - break; - } - } - } - } - } - - if ( newMaster >= 0 ) - return; - - sleepsecs( 1 ); - } - } - - void ReplicaSetMonitor::check( bool checkAllSecondaries ) { - shared_ptr<DBClientConnection> masterConn; - - { - scoped_lock lk( _lock ); - - // first see if the current master is fine - if ( _master >= 0 ) { - assert(_master < static_cast<int>(_nodes.size())); - masterConn = _nodes[_master].conn; - } - } - - if ( masterConn.get() != NULL ) { - string temp; - - if ( _checkConnection( masterConn.get(), temp, false, _master )) { - if ( ! checkAllSecondaries ) { - // current master is fine, so we're done - return; - } - } - } - - // we either have no master, or the current is dead - _check( checkAllSecondaries ); - } - - int ReplicaSetMonitor::_find( const string& server ) const { - scoped_lock lk( _lock ); - return _find_inlock( server ); - } - - int ReplicaSetMonitor::_find_inlock( const string& server ) const { - const size_t size = _nodes.size(); - - for ( unsigned i = 0; i < size; i++ ) { - if ( _nodes[i].addr == server ) { - return i; - } - } - - return -1; - } - - void ReplicaSetMonitor::appendInfo( BSONObjBuilder& b ) const { - scoped_lock lk( _lock ); - BSONArrayBuilder hosts( b.subarrayStart( "hosts" ) ); - for ( unsigned i=0; i<_nodes.size(); i++ ) { - hosts.append( BSON( "addr" << _nodes[i].addr << - // "lastIsMaster" << _nodes[i].lastIsMaster << // this is a potential race, so only used when debugging - "ok" << _nodes[i].ok << - "ismaster" << _nodes[i].ismaster << - "hidden" << _nodes[i].hidden << - "secondary" << _nodes[i].secondary << - "pingTimeMillis" << _nodes[i].pingTimeMillis ) ); - - } - hosts.done(); - - b.append( "master" , _master ); - b.append( "nextSlave" , _nextSlave ); - } - - bool ReplicaSetMonitor::_checkConnMatch_inlock( DBClientConnection* conn, - size_t nodeOffset ) const { - - return ( nodeOffset < _nodes.size() && - conn->getServerAddress() == _nodes[nodeOffset].conn->getServerAddress() ); - } - - - mongo::mutex ReplicaSetMonitor::_setsLock( "ReplicaSetMonitor" ); - map<string,ReplicaSetMonitorPtr> ReplicaSetMonitor::_sets; - ReplicaSetMonitor::ConfigChangeHook ReplicaSetMonitor::_hook; - // -------------------------------- - // ----- DBClientReplicaSet --------- - // -------------------------------- - - DBClientReplicaSet::DBClientReplicaSet( const string& name , const vector<HostAndPort>& servers, double so_timeout ) - : _monitor( ReplicaSetMonitor::get( name , servers ) ), - _so_timeout( so_timeout ) { - } - - DBClientReplicaSet::~DBClientReplicaSet() { - } - - DBClientConnection * DBClientReplicaSet::checkMaster() { - HostAndPort h = _monitor->getMaster(); - - if ( h == _masterHost && _master ) { - // a master is selected. let's just make sure connection didn't die - if ( ! _master->isFailed() ) - return _master.get(); - - _monitor->notifyFailure( _masterHost ); - } - - _masterHost = _monitor->getMaster(); - _master.reset( new DBClientConnection( true , this , _so_timeout ) ); - string errmsg; - if ( ! _master->connect( _masterHost , errmsg ) ) { - _monitor->notifyFailure( _masterHost ); - uasserted( 13639 , str::stream() << "can't connect to new replica set master [" << _masterHost.toString() << "] err: " << errmsg ); - } - _auth( _master.get() ); - return _master.get(); - } - - DBClientConnection * DBClientReplicaSet::checkSlave() { - HostAndPort h = _monitor->getSlave( _slaveHost ); - - if ( h == _slaveHost && _slave ) { - if ( ! _slave->isFailed() ) - return _slave.get(); - _monitor->notifySlaveFailure( _slaveHost ); - _slaveHost = _monitor->getSlave(); - } - else { - _slaveHost = h; - } - - _slave.reset( new DBClientConnection( true , this , _so_timeout ) ); - _slave->connect( _slaveHost ); - _auth( _slave.get() ); - return _slave.get(); - } - - - void DBClientReplicaSet::_auth( DBClientConnection * conn ) { - for ( list<AuthInfo>::iterator i=_auths.begin(); i!=_auths.end(); ++i ) { - const AuthInfo& a = *i; - string errmsg; - if ( ! conn->auth( a.dbname , a.username , a.pwd , errmsg, a.digestPassword ) ) - warning() << "cached auth failed for set: " << _monitor->getName() << " db: " << a.dbname << " user: " << a.username << endl; - - } - } - - DBClientConnection& DBClientReplicaSet::masterConn() { - return *checkMaster(); - } - - DBClientConnection& DBClientReplicaSet::slaveConn() { - return *checkSlave(); - } - - bool DBClientReplicaSet::connect() { - try { - checkMaster(); - } - catch (AssertionException&) { - if (_master && _monitor) { - _monitor->notifyFailure(_masterHost); - } - return false; - } - return true; - } - - bool DBClientReplicaSet::auth(const string &dbname, const string &username, const string &pwd, string& errmsg, bool digestPassword ) { - DBClientConnection * m = checkMaster(); - - // first make sure it actually works - if( ! m->auth(dbname, username, pwd, errmsg, digestPassword ) ) - return false; - - // now that it does, we should save so that for a new node we can auth - _auths.push_back( AuthInfo( dbname , username , pwd , digestPassword ) ); - return true; - } - - // ------------- simple functions ----------------- - - void DBClientReplicaSet::insert( const string &ns , BSONObj obj , int flags) { - checkMaster()->insert(ns, obj, flags); - } - - void DBClientReplicaSet::insert( const string &ns, const vector< BSONObj >& v , int flags) { - checkMaster()->insert(ns, v, flags); - } - - void DBClientReplicaSet::remove( const string &ns , Query obj , bool justOne ) { - checkMaster()->remove(ns, obj, justOne); - } - - void DBClientReplicaSet::update( const string &ns , Query query , BSONObj obj , bool upsert , bool multi ) { - return checkMaster()->update(ns, query, obj, upsert,multi); - } - - auto_ptr<DBClientCursor> DBClientReplicaSet::query(const string &ns, Query query, int nToReturn, int nToSkip, - const BSONObj *fieldsToReturn, int queryOptions, int batchSize) { - - if ( queryOptions & QueryOption_SlaveOk ) { - // we're ok sending to a slave - // we'll try 2 slaves before just using master - // checkSlave will try a different slave automatically after a failure - for ( int i=0; i<3; i++ ) { - try { - return checkSlaveQueryResult( checkSlave()->query(ns,query,nToReturn,nToSkip,fieldsToReturn,queryOptions,batchSize) ); - } - catch ( DBException &e ) { - LOG(1) << "can't query replica set slave " << i << " : " << _slaveHost << causedBy( e ) << endl; - } - } - } - - return checkMaster()->query(ns,query,nToReturn,nToSkip,fieldsToReturn,queryOptions,batchSize); - } - - BSONObj DBClientReplicaSet::findOne(const string &ns, const Query& query, const BSONObj *fieldsToReturn, int queryOptions) { - if ( queryOptions & QueryOption_SlaveOk ) { - // we're ok sending to a slave - // we'll try 2 slaves before just using master - // checkSlave will try a different slave automatically after a failure - for ( int i=0; i<3; i++ ) { - try { - return checkSlave()->findOne(ns,query,fieldsToReturn,queryOptions); - } - catch ( DBException &e ) { - LOG(1) << "can't findone replica set slave " << i << " : " << _slaveHost << causedBy( e ) << endl; - } - } - } - - return checkMaster()->findOne(ns,query,fieldsToReturn,queryOptions); - } - - void DBClientReplicaSet::killCursor( long long cursorID ) { - // we should neve call killCursor on a replica set conncetion - // since we don't know which server it belongs to - // can't assume master because of slave ok - // and can have a cursor survive a master change - assert(0); - } - - void DBClientReplicaSet::isntMaster() { - log() << "got not master for: " << _masterHost << endl; - _monitor->notifyFailure( _masterHost ); - _master.reset(); - } - - auto_ptr<DBClientCursor> DBClientReplicaSet::checkSlaveQueryResult( auto_ptr<DBClientCursor> result ){ - if ( result.get() == NULL ) return result; - - BSONObj error; - bool isError = result->peekError( &error ); - if( ! isError ) return result; - - // We only check for "not master or secondary" errors here - - // If the error code here ever changes, we need to change this code also - BSONElement code = error["code"]; - if( code.isNumber() && code.Int() == 13436 /* not master or secondary */ ){ - isntSecondary(); - throw DBException( str::stream() << "slave " << _slaveHost.toString() << " is no longer secondary", 14812 ); - } - - return result; - } - - void DBClientReplicaSet::isntSecondary() { - log() << "slave no longer has secondary status: " << _slaveHost << endl; - // Failover to next slave - _monitor->notifySlaveFailure( _slaveHost ); - _slave.reset(); - } - - void DBClientReplicaSet::say( Message& toSend, bool isRetry ) { - - if( ! isRetry ) - _lazyState = LazyState(); - - int lastOp = -1; - bool slaveOk = false; - - if ( ( lastOp = toSend.operation() ) == dbQuery ) { - // TODO: might be possible to do this faster by changing api - DbMessage dm( toSend ); - QueryMessage qm( dm ); - if ( ( slaveOk = ( qm.queryOptions & QueryOption_SlaveOk ) ) ) { - - for ( int i = _lazyState._retries; i < 3; i++ ) { - try { - DBClientConnection* slave = checkSlave(); - slave->say( toSend ); - - _lazyState._lastOp = lastOp; - _lazyState._slaveOk = slaveOk; - _lazyState._retries = i; - _lazyState._lastClient = slave; - return; - } - catch ( DBException &e ) { - LOG(1) << "can't callLazy replica set slave " << i << " : " << _slaveHost << causedBy( e ) << endl; - } - } - } - } - - DBClientConnection* master = checkMaster(); - master->say( toSend ); - - _lazyState._lastOp = lastOp; - _lazyState._slaveOk = slaveOk; - _lazyState._retries = 3; - _lazyState._lastClient = master; - return; - } - - bool DBClientReplicaSet::recv( Message& m ) { - - assert( _lazyState._lastClient ); - - // TODO: It would be nice if we could easily wrap a conn error as a result error - try { - return _lazyState._lastClient->recv( m ); - } - catch( DBException& e ){ - log() << "could not receive data from " << _lazyState._lastClient << causedBy( e ) << endl; - return false; - } - } - - void DBClientReplicaSet::checkResponse( const char* data, int nReturned, bool* retry, string* targetHost ){ - - // For now, do exactly as we did before, so as not to break things. In general though, we - // should fix this so checkResponse has a more consistent contract. - if( ! retry ){ - if( _lazyState._lastClient ) - return _lazyState._lastClient->checkResponse( data, nReturned ); - else - return checkMaster()->checkResponse( data, nReturned ); - } - - *retry = false; - if( targetHost && _lazyState._lastClient ) *targetHost = _lazyState._lastClient->getServerAddress(); - else if (targetHost) *targetHost = ""; - - if( ! _lazyState._lastClient ) return; - if( nReturned != 1 && nReturned != -1 ) return; - - BSONObj dataObj; - if( nReturned == 1 ) dataObj = BSONObj( data ); - - // Check if we should retry here - if( _lazyState._lastOp == dbQuery && _lazyState._slaveOk ){ - - // Check the error code for a slave not secondary error - if( nReturned == -1 || - ( hasErrField( dataObj ) && ! dataObj["code"].eoo() && dataObj["code"].Int() == 13436 ) ){ - - bool wasMaster = false; - if( _lazyState._lastClient == _slave.get() ){ - isntSecondary(); - } - else if( _lazyState._lastClient == _master.get() ){ - wasMaster = true; - isntMaster(); - } - else - warning() << "passed " << dataObj << " but last rs client " << _lazyState._lastClient->toString() << " is not master or secondary" << endl; - - if( _lazyState._retries < 3 ){ - _lazyState._retries++; - *retry = true; - } - else{ - (void)wasMaster; // silence set-but-not-used warning - // assert( wasMaster ); - // printStackTrace(); - log() << "too many retries (" << _lazyState._retries << "), could not get data from replica set" << endl; - } - } - } - } - - - bool DBClientReplicaSet::call( Message &toSend, Message &response, bool assertOk , string * actualServer ) { - const char * ns = 0; - - if ( toSend.operation() == dbQuery ) { - // TODO: might be possible to do this faster by changing api - DbMessage dm( toSend ); - QueryMessage qm( dm ); - ns = qm.ns; - - if ( qm.queryOptions & QueryOption_SlaveOk ) { - for ( int i=0; i<3; i++ ) { - try { - DBClientConnection* s = checkSlave(); - if ( actualServer ) - *actualServer = s->getServerAddress(); - return s->call( toSend , response , assertOk ); - } - catch ( DBException &e ) { - LOG(1) << "can't call replica set slave " << i << " : " << _slaveHost << causedBy( e ) << endl; - if ( actualServer ) - *actualServer = ""; - } - } - } - } - - DBClientConnection* m = checkMaster(); - if ( actualServer ) - *actualServer = m->getServerAddress(); - - if ( ! m->call( toSend , response , assertOk ) ) - return false; - - if ( ns ) { - QueryResult * res = (QueryResult*)response.singleData(); - if ( res->nReturned == 1 ) { - BSONObj x(res->data() ); - if ( str::contains( ns , "$cmd" ) ) { - if ( isNotMasterErrorString( x["errmsg"] ) ) - isntMaster(); - } - else { - if ( isNotMasterErrorString( getErrField( x ) ) ) - isntMaster(); - } - } - } - - return true; - } - -} diff --git a/client/dbclient_rs.h b/client/dbclient_rs.h deleted file mode 100644 index 318b3cf913d..00000000000 --- a/client/dbclient_rs.h +++ /dev/null @@ -1,384 +0,0 @@ -/** @file dbclient_rs.h Connect to a Replica Set, from C++ */ - -/* Copyright 2009 10gen Inc. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -#pragma once - -#include "../pch.h" -#include "dbclient.h" - -namespace mongo { - - class ReplicaSetMonitor; - typedef shared_ptr<ReplicaSetMonitor> ReplicaSetMonitorPtr; - typedef pair<set<string>,set<int> > NodeDiff; - - /** - * manages state about a replica set for client - * keeps tabs on whose master and what slaves are up - * can hand a slave to someone for SLAVE_OK - * one instace per process per replica set - * TODO: we might be able to use a regular Node * to avoid _lock - */ - class ReplicaSetMonitor { - public: - - typedef boost::function1<void,const ReplicaSetMonitor*> ConfigChangeHook; - - /** - * gets a cached Monitor per name or will create if doesn't exist - */ - static ReplicaSetMonitorPtr get( const string& name , const vector<HostAndPort>& servers ); - - /** - * gets a cached Monitor per name or will return none if it doesn't exist - */ - static ReplicaSetMonitorPtr get( const string& name ); - - - /** - * checks all sets for current master and new secondaries - * usually only called from a BackgroundJob - */ - static void checkAll( bool checkAllSecondaries ); - - /** - * this is called whenever the config of any repclia set changes - * currently only 1 globally - * asserts if one already exists - * ownership passes to ReplicaSetMonitor and the hook will actually never be deleted - */ - static void setConfigChangeHook( ConfigChangeHook hook ); - - ~ReplicaSetMonitor(); - - /** @return HostAndPort or throws an exception */ - HostAndPort getMaster(); - - /** - * notify the monitor that server has faild - */ - void notifyFailure( const HostAndPort& server ); - - /** @return prev if its still ok, and if not returns a random slave that is ok for reads */ - HostAndPort getSlave( const HostAndPort& prev ); - - /** @return a random slave that is ok for reads */ - HostAndPort getSlave(); - - - /** - * notify the monitor that server has faild - */ - void notifySlaveFailure( const HostAndPort& server ); - - /** - * checks for current master and new secondaries - */ - void check( bool checkAllSecondaries ); - - string getName() const { return _name; } - - string getServerAddress() const; - - bool contains( const string& server ) const; - - void appendInfo( BSONObjBuilder& b ) const; - - private: - /** - * This populates a list of hosts from the list of seeds (discarding the - * seed list). - * @param name set name - * @param servers seeds - */ - ReplicaSetMonitor( const string& name , const vector<HostAndPort>& servers ); - - /** - * Checks all connections from the host list and sets the current - * master. - * - * @param checkAllSecondaries if set to false, stop immediately when - * the master is found or when _master is not -1. - */ - void _check( bool checkAllSecondaries ); - - /** - * Use replSetGetStatus command to make sure hosts in host list are up - * and readable. Sets Node::ok appropriately. - */ - void _checkStatus( const string& hostAddr ); - - /** - * Add array of hosts to host list. Doesn't do anything if hosts are - * already in host list. - * @param hostList the list of hosts to add - * @param changed if new hosts were added - */ - void _checkHosts(const BSONObj& hostList, bool& changed); - - /** - * Updates host list. - * Invariant: if nodesOffset is >= 0, _nodes[nodesOffset].conn should be - * equal to conn. - * - * @param conn the connection to check - * @param maybePrimary OUT - * @param verbose - * @param nodesOffset - offset into _nodes array, -1 for not in it - * - * @return true if the connection is good or false if invariant - * is broken - */ - bool _checkConnection( DBClientConnection* conn, string& maybePrimary, - bool verbose, int nodesOffset ); - - string _getServerAddress_inlock() const; - - NodeDiff _getHostDiff_inlock( const BSONObj& hostList ); - bool _shouldChangeHosts( const BSONObj& hostList, bool inlock ); - - /** - * @return the index to _nodes corresponding to the server address. - */ - int _find( const string& server ) const ; - int _find_inlock( const string& server ) const ; - - /** - * Checks whether the given connection matches the connection stored in _nodes. - * Mainly used for sanity checking to confirm that nodeOffset still - * refers to the right connection after releasing and reacquiring - * a mutex. - */ - bool _checkConnMatch_inlock( DBClientConnection* conn, size_t nodeOffset ) const; - - // protects _nodes and indices pointing to it (_master & _nextSlave) - mutable mongo::mutex _lock; - - /** - * "Synchronizes" the _checkConnection method. Should ideally be one mutex per - * connection object being used. The purpose of this lock is to make sure that - * the reply from the connection the lock holder got is the actual response - * to what it sent. - * - * Deadlock WARNING: never acquire this while holding _lock - */ - mutable mongo::mutex _checkConnectionLock; - - string _name; - struct Node { - Node( const HostAndPort& a , DBClientConnection* c ) - : addr( a ) , conn(c) , ok( c != NULL ), - ismaster(false), secondary( false ) , hidden( false ) , pingTimeMillis(0) { - } - - bool okForSecondaryQueries() const { - return ok && secondary && ! hidden; - } - - BSONObj toBSON() const { - return BSON( "addr" << addr.toString() << - "isMaster" << ismaster << - "secondary" << secondary << - "hidden" << hidden << - "ok" << ok ); - } - - string toString() const { - return toBSON().toString(); - } - - HostAndPort addr; - shared_ptr<DBClientConnection> conn; - - // if this node is in a failure state - // used for slave routing - // this is too simple, should make it better - bool ok; - - // as reported by ismaster - BSONObj lastIsMaster; - - bool ismaster; - bool secondary; - bool hidden; - - int pingTimeMillis; - - }; - - /** - * Host list. - */ - vector<Node> _nodes; - - int _master; // which node is the current master. -1 means no master is known - int _nextSlave; // which node is the current slave - - static mongo::mutex _setsLock; // protects _sets - static map<string,ReplicaSetMonitorPtr> _sets; // set name to Monitor - - static ConfigChangeHook _hook; - }; - - /** Use this class to connect to a replica set of servers. The class will manage - checking for which server in a replica set is master, and do failover automatically. - - This can also be used to connect to replica pairs since pairs are a subset of sets - - On a failover situation, expect at least one operation to return an error (throw - an exception) before the failover is complete. Operations are not retried. - */ - class DBClientReplicaSet : public DBClientBase { - - public: - /** Call connect() after constructing. autoReconnect is always on for DBClientReplicaSet connections. */ - DBClientReplicaSet( const string& name , const vector<HostAndPort>& servers, double so_timeout=0 ); - virtual ~DBClientReplicaSet(); - - /** Returns false if nomember of the set were reachable, or neither is - * master, although, - * when false returned, you can still try to use this connection object, it will - * try reconnects. - */ - bool connect(); - - /** Authorize. Authorizes all nodes as needed - */ - virtual bool auth(const string &dbname, const string &username, const string &pwd, string& errmsg, bool digestPassword = true ); - - // ----------- simple functions -------------- - - /** throws userassertion "no master found" */ - virtual auto_ptr<DBClientCursor> query(const string &ns, Query query, int nToReturn = 0, int nToSkip = 0, - const BSONObj *fieldsToReturn = 0, int queryOptions = 0 , int batchSize = 0 ); - - /** throws userassertion "no master found" */ - virtual BSONObj findOne(const string &ns, const Query& query, const BSONObj *fieldsToReturn = 0, int queryOptions = 0); - - virtual void insert( const string &ns , BSONObj obj , int flags=0); - - /** insert multiple objects. Note that single object insert is asynchronous, so this version - is only nominally faster and not worth a special effort to try to use. */ - virtual void insert( const string &ns, const vector< BSONObj >& v , int flags=0); - - virtual void remove( const string &ns , Query obj , bool justOne = 0 ); - - virtual void update( const string &ns , Query query , BSONObj obj , bool upsert = 0 , bool multi = 0 ); - - virtual void killCursor( long long cursorID ); - - // ---- access raw connections ---- - - DBClientConnection& masterConn(); - DBClientConnection& slaveConn(); - - // ---- callback pieces ------- - - virtual void say( Message &toSend, bool isRetry = false ); - virtual bool recv( Message &toRecv ); - virtual void checkResponse( const char* data, int nReturned, bool* retry = NULL, string* targetHost = NULL ); - - /* this is the callback from our underlying connections to notify us that we got a "not master" error. - */ - void isntMaster(); - - /* this is used to indicate we got a "not master or secondary" error from a secondary. - */ - void isntSecondary(); - - // ----- status ------ - - virtual bool isFailed() const { return ! _master || _master->isFailed(); } - - // ----- informational ---- - - double getSoTimeout() const { return _so_timeout; } - - string toString() { return getServerAddress(); } - - string getServerAddress() const { return _monitor->getServerAddress(); } - - virtual ConnectionString::ConnectionType type() const { return ConnectionString::SET; } - virtual bool lazySupported() const { return true; } - - // ---- low level ------ - - virtual bool call( Message &toSend, Message &response, bool assertOk=true , string * actualServer = 0 ); - virtual bool callRead( Message& toSend , Message& response ) { return checkMaster()->callRead( toSend , response ); } - - - protected: - virtual void sayPiggyBack( Message &toSend ) { checkMaster()->say( toSend ); } - - private: - - // Used to simplify slave-handling logic on errors - auto_ptr<DBClientCursor> checkSlaveQueryResult( auto_ptr<DBClientCursor> result ); - - DBClientConnection * checkMaster(); - DBClientConnection * checkSlave(); - - void _auth( DBClientConnection * conn ); - - ReplicaSetMonitorPtr _monitor; - - HostAndPort _masterHost; - scoped_ptr<DBClientConnection> _master; - - HostAndPort _slaveHost; - scoped_ptr<DBClientConnection> _slave; - - double _so_timeout; - - /** - * for storing authentication info - * fields are exactly for DBClientConnection::auth - */ - struct AuthInfo { - AuthInfo( string d , string u , string p , bool di ) - : dbname( d ) , username( u ) , pwd( p ) , digestPassword( di ) {} - string dbname; - string username; - string pwd; - bool digestPassword; - }; - - // we need to store so that when we connect to a new node on failure - // we can re-auth - // this could be a security issue, as the password is stored in memory - // not sure if/how we should handle - list<AuthInfo> _auths; - - protected: - - /** - * for storing (non-threadsafe) information between lazy calls - */ - class LazyState { - public: - LazyState() : _lastClient( NULL ), _lastOp( -1 ), _slaveOk( false ), _retries( 0 ) {} - DBClientConnection* _lastClient; - int _lastOp; - bool _slaveOk; - int _retries; - - } _lazyState; - - }; - - -} diff --git a/client/dbclientcursor.cpp b/client/dbclientcursor.cpp deleted file mode 100644 index 9da45e5c32d..00000000000 --- a/client/dbclientcursor.cpp +++ /dev/null @@ -1,319 +0,0 @@ -// dbclient.cpp - connect to a Mongo database as a database, from C++ - -/* Copyright 2009 10gen Inc. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -#include "pch.h" -#include "dbclient.h" -#include "../db/dbmessage.h" -#include "../db/cmdline.h" -#include "connpool.h" -#include "../s/shard.h" - -namespace mongo { - - void assembleRequest( const string &ns, BSONObj query, int nToReturn, int nToSkip, const BSONObj *fieldsToReturn, int queryOptions, Message &toSend ); - - int DBClientCursor::nextBatchSize() { - - if ( nToReturn == 0 ) - return batchSize; - - if ( batchSize == 0 ) - return nToReturn; - - return batchSize < nToReturn ? batchSize : nToReturn; - } - - void DBClientCursor::_assembleInit( Message& toSend ) { - if ( !cursorId ) { - assembleRequest( ns, query, nextBatchSize() , nToSkip, fieldsToReturn, opts, toSend ); - } - else { - BufBuilder b; - b.appendNum( opts ); - b.appendStr( ns ); - b.appendNum( nToReturn ); - b.appendNum( cursorId ); - toSend.setData( dbGetMore, b.buf(), b.len() ); - } - } - - bool DBClientCursor::init() { - Message toSend; - _assembleInit( toSend ); - - if ( !_client->call( toSend, *b.m, false ) ) { - // log msg temp? - log() << "DBClientCursor::init call() failed" << endl; - return false; - } - if ( b.m->empty() ) { - // log msg temp? - log() << "DBClientCursor::init message from call() was empty" << endl; - return false; - } - dataReceived(); - return true; - } - - void DBClientCursor::initLazy( bool isRetry ) { - verify( 15875 , _client->lazySupported() ); - Message toSend; - _assembleInit( toSend ); - _client->say( toSend, isRetry ); - } - - bool DBClientCursor::initLazyFinish( bool& retry ) { - - bool recvd = _client->recv( *b.m ); - - // If we get a bad response, return false - if ( ! recvd || b.m->empty() ) { - - if( !recvd ) - log() << "DBClientCursor::init lazy say() failed" << endl; - if( b.m->empty() ) - log() << "DBClientCursor::init message from say() was empty" << endl; - - _client->checkResponse( NULL, -1, &retry, &_lazyHost ); - - return false; - - } - - dataReceived( retry, _lazyHost ); - return ! retry; - } - - void DBClientCursor::requestMore() { - assert( cursorId && b.pos == b.nReturned ); - - if (haveLimit) { - nToReturn -= b.nReturned; - assert(nToReturn > 0); - } - BufBuilder b; - b.appendNum(opts); - b.appendStr(ns); - b.appendNum(nextBatchSize()); - b.appendNum(cursorId); - - Message toSend; - toSend.setData(dbGetMore, b.buf(), b.len()); - auto_ptr<Message> response(new Message()); - - if ( _client ) { - _client->call( toSend, *response ); - this->b.m = response; - dataReceived(); - } - else { - assert( _scopedHost.size() ); - ScopedDbConnection conn( _scopedHost ); - conn->call( toSend , *response ); - _client = conn.get(); - this->b.m = response; - dataReceived(); - _client = 0; - conn.done(); - } - } - - /** with QueryOption_Exhaust, the server just blasts data at us (marked at end with cursorid==0). */ - void DBClientCursor::exhaustReceiveMore() { - assert( cursorId && b.pos == b.nReturned ); - assert( !haveLimit ); - auto_ptr<Message> response(new Message()); - assert( _client ); - if ( _client->recv(*response) ) { - b.m = response; - dataReceived(); - } - } - - void DBClientCursor::dataReceived( bool& retry, string& host ) { - - QueryResult *qr = (QueryResult *) b.m->singleData(); - resultFlags = qr->resultFlags(); - - if ( qr->resultFlags() & ResultFlag_ErrSet ) { - wasError = true; - } - - if ( qr->resultFlags() & ResultFlag_CursorNotFound ) { - // cursor id no longer valid at the server. - assert( qr->cursorId == 0 ); - cursorId = 0; // 0 indicates no longer valid (dead) - if ( ! ( opts & QueryOption_CursorTailable ) ) - throw UserException( 13127 , "getMore: cursor didn't exist on server, possible restart or timeout?" ); - } - - if ( cursorId == 0 || ! ( opts & QueryOption_CursorTailable ) ) { - // only set initially: we don't want to kill it on end of data - // if it's a tailable cursor - cursorId = qr->cursorId; - } - - b.nReturned = qr->nReturned; - b.pos = 0; - b.data = qr->data(); - - _client->checkResponse( b.data, b.nReturned, &retry, &host ); // watches for "not master" - - /* this assert would fire the way we currently work: - assert( nReturned || cursorId == 0 ); - */ - } - - /** If true, safe to call next(). Requests more from server if necessary. */ - bool DBClientCursor::more() { - _assertIfNull(); - - if ( !_putBack.empty() ) - return true; - - if (haveLimit && b.pos >= nToReturn) - return false; - - if ( b.pos < b.nReturned ) - return true; - - if ( cursorId == 0 ) - return false; - - requestMore(); - return b.pos < b.nReturned; - } - - BSONObj DBClientCursor::next() { - DEV _assertIfNull(); - if ( !_putBack.empty() ) { - BSONObj ret = _putBack.top(); - _putBack.pop(); - return ret; - } - - uassert(13422, "DBClientCursor next() called but more() is false", b.pos < b.nReturned); - - b.pos++; - BSONObj o(b.data); - b.data += o.objsize(); - /* todo would be good to make data null at end of batch for safety */ - return o; - } - - void DBClientCursor::peek(vector<BSONObj>& v, int atMost) { - int m = atMost; - - /* - for( stack<BSONObj>::iterator i = _putBack.begin(); i != _putBack.end(); i++ ) { - if( m == 0 ) - return; - v.push_back(*i); - m--; - n++; - } - */ - - int p = b.pos; - const char *d = b.data; - while( m && p < b.nReturned ) { - BSONObj o(d); - d += o.objsize(); - p++; - m--; - v.push_back(o); - } - } - - bool DBClientCursor::peekError(BSONObj* error){ - if( ! wasError ) return false; - - vector<BSONObj> v; - peek(v, 1); - - assert( v.size() == 1 ); - assert( hasErrField( v[0] ) ); - - if( error ) *error = v[0].getOwned(); - return true; - } - - void DBClientCursor::attach( AScopedConnection * conn ) { - assert( _scopedHost.size() == 0 ); - assert( conn ); - assert( conn->get() ); - - if ( conn->get()->type() == ConnectionString::SET || - conn->get()->type() == ConnectionString::SYNC ) { - if( _lazyHost.size() > 0 ) - _scopedHost = _lazyHost; - else if( _client ) - _scopedHost = _client->getServerAddress(); - else - massert(14821, "No client or lazy client specified, cannot store multi-host connection.", false); - } - else { - _scopedHost = conn->getHost(); - } - - conn->done(); - _client = 0; - _lazyHost = ""; - } - - DBClientCursor::~DBClientCursor() { - if (!this) - return; - - DESTRUCTOR_GUARD ( - - if ( cursorId && _ownCursor && ! inShutdown() ) { - BufBuilder b; - b.appendNum( (int)0 ); // reserved - b.appendNum( (int)1 ); // number - b.appendNum( cursorId ); - - Message m; - m.setData( dbKillCursors , b.buf() , b.len() ); - - if ( _client ) { - - // Kill the cursor the same way the connection itself would. Usually, non-lazily - if( DBClientConnection::getLazyKillCursor() ) - _client->sayPiggyBack( m ); - else - _client->say( m ); - - } - else { - assert( _scopedHost.size() ); - ScopedDbConnection conn( _scopedHost ); - - if( DBClientConnection::getLazyKillCursor() ) - conn->sayPiggyBack( m ); - else - conn->say( m ); - - conn.done(); - } - } - - ); - } - - -} // namespace mongo diff --git a/client/dbclientcursor.h b/client/dbclientcursor.h deleted file mode 100644 index 977bd30561e..00000000000 --- a/client/dbclientcursor.h +++ /dev/null @@ -1,239 +0,0 @@ -// file dbclientcursor.h - -/* Copyright 2009 10gen Inc. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -#pragma once - -#include "../pch.h" -#include "../util/net/message.h" -#include "../db/jsobj.h" -#include "../db/json.h" -#include <stack> - -namespace mongo { - - class AScopedConnection; - - /** for mock purposes only -- do not create variants of DBClientCursor, nor hang code here */ - class DBClientCursorInterface { - public: - virtual ~DBClientCursorInterface() {} - - virtual bool more() = 0; - virtual BSONObj next() = 0; - - // TODO bring more of the DBClientCursor interface to here - - protected: - DBClientCursorInterface() {} - }; - - /** Queries return a cursor object */ - class DBClientCursor : public DBClientCursorInterface { - public: - /** If true, safe to call next(). Requests more from server if necessary. */ - bool more(); - - /** If true, there is more in our local buffers to be fetched via next(). Returns - false when a getMore request back to server would be required. You can use this - if you want to exhaust whatever data has been fetched to the client already but - then perhaps stop. - */ - int objsLeftInBatch() const { _assertIfNull(); return _putBack.size() + b.nReturned - b.pos; } - bool moreInCurrentBatch() { return objsLeftInBatch() > 0; } - - /** next - @return next object in the result cursor. - on an error at the remote server, you will get back: - { $err: <string> } - if you do not want to handle that yourself, call nextSafe(). - */ - BSONObj next(); - - /** - restore an object previously returned by next() to the cursor - */ - void putBack( const BSONObj &o ) { _putBack.push( o.getOwned() ); } - - /** throws AssertionException if get back { $err : ... } */ - BSONObj nextSafe() { - BSONObj o = next(); - if( strcmp(o.firstElementFieldName(), "$err") == 0 ) { - string s = "nextSafe(): " + o.toString(); - if( logLevel >= 5 ) - log() << s << endl; - uasserted(13106, s); - } - return o; - } - - /** peek ahead at items buffered for future next() calls. - never requests new data from the server. so peek only effective - with what is already buffered. - WARNING: no support for _putBack yet! - */ - void peek(vector<BSONObj>&, int atMost); - - /** - * peek ahead and see if an error occurred, and get the error if so. - */ - bool peekError(BSONObj* error = NULL); - - /** - iterate the rest of the cursor and return the number if items - */ - int itcount() { - int c = 0; - while ( more() ) { - next(); - c++; - } - return c; - } - - /** cursor no longer valid -- use with tailable cursors. - note you should only rely on this once more() returns false; - 'dead' may be preset yet some data still queued and locally - available from the dbclientcursor. - */ - bool isDead() const { return !this || cursorId == 0; } - - bool tailable() const { return (opts & QueryOption_CursorTailable) != 0; } - - /** see ResultFlagType (constants.h) for flag values - mostly these flags are for internal purposes - - ResultFlag_ErrSet is the possible exception to that - */ - bool hasResultFlag( int flag ) { - _assertIfNull(); - return (resultFlags & flag) != 0; - } - - DBClientCursor( DBClientBase* client, const string &_ns, BSONObj _query, int _nToReturn, - int _nToSkip, const BSONObj *_fieldsToReturn, int queryOptions , int bs ) : - _client(client), - ns(_ns), - query(_query), - nToReturn(_nToReturn), - haveLimit( _nToReturn > 0 && !(queryOptions & QueryOption_CursorTailable)), - nToSkip(_nToSkip), - fieldsToReturn(_fieldsToReturn), - opts(queryOptions), - batchSize(bs==1?2:bs), - cursorId(), - _ownCursor( true ), - wasError( false ) { - } - - DBClientCursor( DBClientBase* client, const string &_ns, long long _cursorId, int _nToReturn, int options ) : - _client(client), - ns(_ns), - nToReturn( _nToReturn ), - haveLimit( _nToReturn > 0 && !(options & QueryOption_CursorTailable)), - opts( options ), - cursorId(_cursorId), - _ownCursor( true ) { - } - - virtual ~DBClientCursor(); - - long long getCursorId() const { return cursorId; } - - /** by default we "own" the cursor and will send the server a KillCursor - message when ~DBClientCursor() is called. This function overrides that. - */ - void decouple() { _ownCursor = false; } - - void attach( AScopedConnection * conn ); - - /** - * actually does the query - */ - bool init(); - - void initLazy( bool isRetry = false ); - bool initLazyFinish( bool& retry ); - - class Batch : boost::noncopyable { - friend class DBClientCursor; - auto_ptr<Message> m; - int nReturned; - int pos; - const char *data; - public: - Batch() : m( new Message() ), nReturned(), pos(), data() { } - }; - - private: - friend class DBClientBase; - friend class DBClientConnection; - - int nextBatchSize(); - - Batch b; - DBClientBase* _client; - string ns; - BSONObj query; - int nToReturn; - bool haveLimit; - int nToSkip; - const BSONObj *fieldsToReturn; - int opts; - int batchSize; - stack< BSONObj > _putBack; - int resultFlags; - long long cursorId; - bool _ownCursor; // see decouple() - string _scopedHost; - string _lazyHost; - bool wasError; - - void dataReceived() { bool retry; string lazyHost; dataReceived( retry, lazyHost ); } - void dataReceived( bool& retry, string& lazyHost ); - void requestMore(); - void exhaustReceiveMore(); // for exhaust - - // Don't call from a virtual function - void _assertIfNull() const { uassert(13348, "connection died", this); } - - // non-copyable , non-assignable - DBClientCursor( const DBClientCursor& ); - DBClientCursor& operator=( const DBClientCursor& ); - - // init pieces - void _assembleInit( Message& toSend ); - }; - - /** iterate over objects in current batch only - will not cause a network call - */ - class DBClientCursorBatchIterator { - public: - DBClientCursorBatchIterator( DBClientCursor &c ) : _c( c ), _n() {} - bool moreInCurrentBatch() { return _c.moreInCurrentBatch(); } - BSONObj nextSafe() { - massert( 13383, "BatchIterator empty", moreInCurrentBatch() ); - ++_n; - return _c.nextSafe(); - } - int n() const { return _n; } - private: - DBClientCursor &_c; - int _n; - }; - -} // namespace mongo - -#include "undef_macros.h" diff --git a/client/dbclientmockcursor.h b/client/dbclientmockcursor.h deleted file mode 100644 index 8d85ff5ad2e..00000000000 --- a/client/dbclientmockcursor.h +++ /dev/null @@ -1,40 +0,0 @@ -//@file dbclientmockcursor.h - -/* Copyright 2010 10gen Inc. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -#pragma once - -#include "dbclientcursor.h" - -namespace mongo { - - class DBClientMockCursor : public DBClientCursorInterface { - public: - DBClientMockCursor( const BSONArray& mockCollection ) : _iter( mockCollection ) {} - virtual ~DBClientMockCursor() {} - - bool more() { return _iter.more(); } - BSONObj next() { return _iter.next().Obj(); } - - private: - BSONObjIterator _iter; - - // non-copyable , non-assignable - DBClientMockCursor( const DBClientMockCursor& ); - DBClientMockCursor& operator=( const DBClientMockCursor& ); - }; - -} // namespace mongo diff --git a/client/distlock.cpp b/client/distlock.cpp deleted file mode 100644 index d31e304aba4..00000000000 --- a/client/distlock.cpp +++ /dev/null @@ -1,967 +0,0 @@ -// @file distlock.h - -/* Copyright 2009 10gen Inc. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -#include "pch.h" -#include "dbclient.h" -#include "distlock.h" - -namespace mongo { - - LabeledLevel DistributedLock::logLvl( 1 ); - DistributedLock::LastPings DistributedLock::lastPings; - - ThreadLocalValue<string> distLockIds(""); - - /* ================== - * Module initialization - */ - - boost::once_flag _init = BOOST_ONCE_INIT; - static string* _cachedProcessString = NULL; - - static void initModule() { - // cache process string - stringstream ss; - ss << getHostName() << ":" << cmdLine.port << ":" << time(0) << ":" << rand(); - _cachedProcessString = new string( ss.str() ); - } - - /* =================== */ - - string getDistLockProcess() { - boost::call_once( initModule, _init ); - assert( _cachedProcessString ); - return *_cachedProcessString; - } - - string getDistLockId() { - string s = distLockIds.get(); - if ( s.empty() ) { - stringstream ss; - ss << getDistLockProcess() << ":" << getThreadName() << ":" << rand(); - s = ss.str(); - distLockIds.set( s ); - } - return s; - } - - - class DistributedLockPinger { - public: - - DistributedLockPinger() - : _mutex( "DistributedLockPinger" ) { - } - - void _distLockPingThread( ConnectionString addr, string process, unsigned long long sleepTime ) { - - setThreadName( "LockPinger" ); - - string pingId = pingThreadId( addr, process ); - - log( DistributedLock::logLvl - 1 ) << "creating distributed lock ping thread for " << addr - << " and process " << process - << " (sleeping for " << sleepTime << "ms)" << endl; - - static int loops = 0; - while( ! inShutdown() && ! shouldKill( addr, process ) ) { - - log( DistributedLock::logLvl + 2 ) << "distributed lock pinger '" << pingId << "' about to ping." << endl; - - Date_t pingTime; - - try { - ScopedDbConnection conn( addr, 30.0 ); - - pingTime = jsTime(); - - // refresh the entry corresponding to this process in the lockpings collection - conn->update( DistributedLock::lockPingNS , - BSON( "_id" << process ) , - BSON( "$set" << BSON( "ping" << pingTime ) ) , - true ); - - string err = conn->getLastError(); - if ( ! err.empty() ) { - warning() << "pinging failed for distributed lock pinger '" << pingId << "'." - << causedBy( err ) << endl; - conn.done(); - - // Sleep for normal ping time - sleepmillis(sleepTime); - continue; - } - - // remove really old entries from the lockpings collection if they're not holding a lock - // (this may happen if an instance of a process was taken down and no new instance came up to - // replace it for a quite a while) - // if the lock is taken, the take-over mechanism should handle the situation - auto_ptr<DBClientCursor> c = conn->query( DistributedLock::locksNS , BSONObj() ); - // TODO: Would be good to make clear whether query throws or returns empty on errors - uassert( 16060, str::stream() << "cannot query locks collection on config server " << conn.getHost(), c.get() ); - - set<string> pids; - while ( c->more() ) { - BSONObj lock = c->next(); - if ( ! lock["process"].eoo() ) { - pids.insert( lock["process"].valuestrsafe() ); - } - } - - Date_t fourDays = pingTime - ( 4 * 86400 * 1000 ); // 4 days - conn->remove( DistributedLock::lockPingNS , BSON( "_id" << BSON( "$nin" << pids ) << "ping" << LT << fourDays ) ); - err = conn->getLastError(); - if ( ! err.empty() ) { - warning() << "ping cleanup for distributed lock pinger '" << pingId << " failed." - << causedBy( err ) << endl; - conn.done(); - - // Sleep for normal ping time - sleepmillis(sleepTime); - continue; - } - - // create index so remove is fast even with a lot of servers - if ( loops++ == 0 ) { - conn->ensureIndex( DistributedLock::lockPingNS , BSON( "ping" << 1 ) ); - } - - log( DistributedLock::logLvl - ( loops % 10 == 0 ? 1 : 0 ) ) << "cluster " << addr << " pinged successfully at " << pingTime - << " by distributed lock pinger '" << pingId - << "', sleeping for " << sleepTime << "ms" << endl; - - // Remove old locks, if possible - // Make sure no one else is adding to this list at the same time - scoped_lock lk( _mutex ); - - int numOldLocks = _oldLockOIDs.size(); - if( numOldLocks > 0 ) - log( DistributedLock::logLvl - 1 ) << "trying to delete " << _oldLockOIDs.size() << " old lock entries for process " << process << endl; - - bool removed = false; - for( list<OID>::iterator i = _oldLockOIDs.begin(); i != _oldLockOIDs.end(); - i = ( removed ? _oldLockOIDs.erase( i ) : ++i ) ) { - removed = false; - try { - // Got OID from lock with id, so we don't need to specify id again - conn->update( DistributedLock::locksNS , - BSON( "ts" << *i ), - BSON( "$set" << BSON( "state" << 0 ) ) ); - - // Either the update went through or it didn't, either way we're done trying to - // unlock - log( DistributedLock::logLvl - 1 ) << "handled late remove of old distributed lock with ts " << *i << endl; - removed = true; - } - catch( UpdateNotTheSame& ) { - log( DistributedLock::logLvl - 1 ) << "partially removed old distributed lock with ts " << *i << endl; - removed = true; - } - catch ( std::exception& e) { - warning() << "could not remove old distributed lock with ts " << *i - << causedBy( e ) << endl; - } - - } - - if( numOldLocks > 0 && _oldLockOIDs.size() > 0 ){ - log( DistributedLock::logLvl - 1 ) << "not all old lock entries could be removed for process " << process << endl; - } - - conn.done(); - - } - catch ( std::exception& e ) { - warning() << "distributed lock pinger '" << pingId << "' detected an exception while pinging." - << causedBy( e ) << endl; - } - - sleepmillis(sleepTime); - } - - warning() << "removing distributed lock ping thread '" << pingId << "'" << endl; - - - if( shouldKill( addr, process ) ) - finishKill( addr, process ); - - } - - void distLockPingThread( ConnectionString addr, long long clockSkew, string processId, unsigned long long sleepTime ) { - try { - jsTimeVirtualThreadSkew( clockSkew ); - _distLockPingThread( addr, processId, sleepTime ); - } - catch ( std::exception& e ) { - error() << "unexpected error while running distributed lock pinger for " << addr << ", process " << processId << causedBy( e ) << endl; - } - catch ( ... ) { - error() << "unknown error while running distributed lock pinger for " << addr << ", process " << processId << endl; - } - } - - string pingThreadId( const ConnectionString& conn, const string& processId ) { - return conn.toString() + "/" + processId; - } - - string got( DistributedLock& lock, unsigned long long sleepTime ) { - - // Make sure we don't start multiple threads for a process id - scoped_lock lk( _mutex ); - - const ConnectionString& conn = lock.getRemoteConnection(); - const string& processId = lock.getProcessId(); - string s = pingThreadId( conn, processId ); - - // Ignore if we already have a pinging thread for this process. - if ( _seen.count( s ) > 0 ) return s; - - // Check our clock skew - try { - if( lock.isRemoteTimeSkewed() ) { - throw LockException( str::stream() << "clock skew of the cluster " << conn.toString() << " is too far out of bounds to allow distributed locking." , 13650 ); - } - } - catch( LockException& e) { - throw LockException( str::stream() << "error checking clock skew of cluster " << conn.toString() << causedBy( e ) , 13651); - } - - boost::thread t( boost::bind( &DistributedLockPinger::distLockPingThread, this, conn, getJSTimeVirtualThreadSkew(), processId, sleepTime) ); - - _seen.insert( s ); - - return s; - } - - void addUnlockOID( const OID& oid ) { - // Modifying the lock from some other thread - scoped_lock lk( _mutex ); - _oldLockOIDs.push_back( oid ); - } - - bool willUnlockOID( const OID& oid ) { - scoped_lock lk( _mutex ); - return find( _oldLockOIDs.begin(), _oldLockOIDs.end(), oid ) != _oldLockOIDs.end(); - } - - void kill( const ConnectionString& conn, const string& processId ) { - // Make sure we're in a consistent state before other threads can see us - scoped_lock lk( _mutex ); - - string pingId = pingThreadId( conn, processId ); - - assert( _seen.count( pingId ) > 0 ); - _kill.insert( pingId ); - - } - - bool shouldKill( const ConnectionString& conn, const string& processId ) { - return _kill.count( pingThreadId( conn, processId ) ) > 0; - } - - void finishKill( const ConnectionString& conn, const string& processId ) { - // Make sure we're in a consistent state before other threads can see us - scoped_lock lk( _mutex ); - - string pingId = pingThreadId( conn, processId ); - - _kill.erase( pingId ); - _seen.erase( pingId ); - - } - - set<string> _kill; - set<string> _seen; - mongo::mutex _mutex; - list<OID> _oldLockOIDs; - - } distLockPinger; - - - const string DistributedLock::lockPingNS = "config.lockpings"; - const string DistributedLock::locksNS = "config.locks"; - - /** - * Create a new distributed lock, potentially with a custom sleep and takeover time. If a custom sleep time is - * specified (time between pings) - */ - DistributedLock::DistributedLock( const ConnectionString& conn , const string& name , unsigned long long lockTimeout, bool asProcess ) - : _conn(conn) , _name(name) , _id( BSON( "_id" << name ) ), _processId( asProcess ? getDistLockId() : getDistLockProcess() ), - _lockTimeout( lockTimeout == 0 ? LOCK_TIMEOUT : lockTimeout ), _maxClockSkew( _lockTimeout / LOCK_SKEW_FACTOR ), _maxNetSkew( _maxClockSkew ), _lockPing( _maxClockSkew ), - _mutex( "DistributedLock" ) - { - log( logLvl ) << "created new distributed lock for " << name << " on " << conn - << " ( lock timeout : " << _lockTimeout - << ", ping interval : " << _lockPing << ", process : " << asProcess << " )" - << endl; - - - } - - DistributedLock::PingData DistributedLock::LastPings::getLastPing( const ConnectionString& conn, const string& lockName ){ - scoped_lock lock( _mutex ); - return _lastPings[ std::pair< string, string >( conn.toString(), lockName ) ]; - } - - void DistributedLock::LastPings::setLastPing( const ConnectionString& conn, const string& lockName, const PingData& pd ){ - scoped_lock lock( _mutex ); - _lastPings[ std::pair< string, string >( conn.toString(), lockName ) ] = pd; - } - - Date_t DistributedLock::getRemoteTime() { - return DistributedLock::remoteTime( _conn, _maxNetSkew ); - } - - bool DistributedLock::isRemoteTimeSkewed() { - return !DistributedLock::checkSkew( _conn, NUM_LOCK_SKEW_CHECKS, _maxClockSkew, _maxNetSkew ); - } - - const ConnectionString& DistributedLock::getRemoteConnection() { - return _conn; - } - - const string& DistributedLock::getProcessId() { - return _processId; - } - - /** - * Returns the remote time as reported by the cluster or server. The maximum difference between the reported time - * and the actual time on the remote server (at the completion of the function) is the maxNetSkew - */ - Date_t DistributedLock::remoteTime( const ConnectionString& cluster, unsigned long long maxNetSkew ) { - - ConnectionString server( *cluster.getServers().begin() ); - ScopedDbConnection conn( server ); - - BSONObj result; - long long delay; - - try { - Date_t then = jsTime(); - bool success = conn->runCommand( string("admin"), BSON( "serverStatus" << 1 ), result ); - delay = jsTime() - then; - - if( !success ) - throw TimeNotFoundException( str::stream() << "could not get status from server " - << server.toString() << " in cluster " << cluster.toString() - << " to check time", 13647 ); - - // Make sure that our delay is not more than 2x our maximum network skew, since this is the max our remote - // time value can be off by if we assume a response in the middle of the delay. - if( delay > (long long) (maxNetSkew * 2) ) - throw TimeNotFoundException( str::stream() << "server " << server.toString() - << " in cluster " << cluster.toString() - << " did not respond within max network delay of " - << maxNetSkew << "ms", 13648 ); - } - catch(...) { - conn.done(); - throw; - } - - conn.done(); - - return result["localTime"].Date() - (delay / 2); - - } - - bool DistributedLock::checkSkew( const ConnectionString& cluster, unsigned skewChecks, unsigned long long maxClockSkew, unsigned long long maxNetSkew ) { - - vector<HostAndPort> servers = cluster.getServers(); - - if(servers.size() < 1) return true; - - vector<long long> avgSkews; - - for(unsigned i = 0; i < skewChecks; i++) { - - // Find the average skew for each server - unsigned s = 0; - for(vector<HostAndPort>::iterator si = servers.begin(); si != servers.end(); ++si,s++) { - - if(i == 0) avgSkews.push_back(0); - - // Could check if this is self, but shouldn't matter since local network connection should be fast. - ConnectionString server( *si ); - - vector<long long> skew; - - BSONObj result; - - Date_t remote = remoteTime( server, maxNetSkew ); - Date_t local = jsTime(); - - // Remote time can be delayed by at most MAX_NET_SKEW - - // Skew is how much time we'd have to add to local to get to remote - avgSkews[s] += (long long) (remote - local); - - log( logLvl + 1 ) << "skew from remote server " << server << " found: " << (long long) (remote - local) << endl; - - } - } - - // Analyze skews - - long long serverMaxSkew = 0; - long long serverMinSkew = 0; - - for(unsigned s = 0; s < avgSkews.size(); s++) { - - long long avgSkew = (avgSkews[s] /= skewChecks); - - // Keep track of max and min skews - if(s == 0) { - serverMaxSkew = avgSkew; - serverMinSkew = avgSkew; - } - else { - if(avgSkew > serverMaxSkew) - serverMaxSkew = avgSkew; - if(avgSkew < serverMinSkew) - serverMinSkew = avgSkew; - } - - } - - long long totalSkew = serverMaxSkew - serverMinSkew; - - // Make sure our max skew is not more than our pre-set limit - if(totalSkew > (long long) maxClockSkew) { - log( logLvl + 1 ) << "total clock skew of " << totalSkew << "ms for servers " << cluster << " is out of " << maxClockSkew << "ms bounds." << endl; - return false; - } - - log( logLvl + 1 ) << "total clock skew of " << totalSkew << "ms for servers " << cluster << " is in " << maxClockSkew << "ms bounds." << endl; - return true; - } - - // For use in testing, ping thread should run indefinitely in practice. - bool DistributedLock::killPinger( DistributedLock& lock ) { - if( lock._threadId == "") return false; - - distLockPinger.kill( lock._conn, lock._processId ); - return true; - } - - // Semantics of this method are basically that if the lock cannot be acquired, returns false, can be retried. - // If the lock should not be tried again (some unexpected error) a LockException is thrown. - // If we are only trying to re-enter a currently held lock, reenter should be true. - // Note: reenter doesn't actually make this lock re-entrant in the normal sense, since it can still only - // be unlocked once, instead it is used to verify that the lock is already held. - bool DistributedLock::lock_try( const string& why , bool reenter, BSONObj * other ) { - - // TODO: Start pinging only when we actually get the lock? - // If we don't have a thread pinger, make sure we shouldn't have one - if( _threadId == "" ){ - scoped_lock lk( _mutex ); - _threadId = distLockPinger.got( *this, _lockPing ); - } - - // This should always be true, if not, we are using the lock incorrectly. - assert( _name != "" ); - - log( logLvl ) << "trying to acquire new distributed lock for " << _name << " on " << _conn - << " ( lock timeout : " << _lockTimeout - << ", ping interval : " << _lockPing << ", process : " << _processId << " )" - << endl; - - // write to dummy if 'other' is null - BSONObj dummyOther; - if ( other == NULL ) - other = &dummyOther; - - ScopedDbConnection conn( _conn ); - - BSONObjBuilder queryBuilder; - queryBuilder.appendElements( _id ); - queryBuilder.append( "state" , 0 ); - - { - // make sure its there so we can use simple update logic below - BSONObj o = conn->findOne( locksNS , _id ).getOwned(); - - // Case 1: No locks - if ( o.isEmpty() ) { - try { - log( logLvl ) << "inserting initial doc in " << locksNS << " for lock " << _name << endl; - conn->insert( locksNS , BSON( "_id" << _name << "state" << 0 << "who" << "" ) ); - } - catch ( UserException& e ) { - warning() << "could not insert initial doc for distributed lock " << _name << causedBy( e ) << endl; - } - } - - // Case 2: A set lock that we might be able to force - else if ( o["state"].numberInt() > 0 ) { - - string lockName = o["_id"].String() + string("/") + o["process"].String(); - - bool canReenter = reenter && o["process"].String() == _processId && ! distLockPinger.willUnlockOID( o["ts"].OID() ) && o["state"].numberInt() == 2; - if( reenter && ! canReenter ) { - log( logLvl - 1 ) << "not re-entering distributed lock " << lockName; - if( o["process"].String() != _processId ) log( logLvl - 1 ) << ", different process " << _processId << endl; - else if( o["state"].numberInt() == 2 ) log( logLvl - 1 ) << ", state not finalized" << endl; - else log( logLvl - 1 ) << ", ts " << o["ts"].OID() << " scheduled for late unlock" << endl; - - // reset since we've been bounced by a previous lock not being where we thought it was, - // and should go through full forcing process if required. - // (in theory we should never see a ping here if used correctly) - *other = o; other->getOwned(); conn.done(); resetLastPing(); - return false; - } - - BSONObj lastPing = conn->findOne( lockPingNS , o["process"].wrap( "_id" ) ); - if ( lastPing.isEmpty() ) { - log( logLvl ) << "empty ping found for process in lock '" << lockName << "'" << endl; - // TODO: Using 0 as a "no time found" value Will fail if dates roll over, but then, so will a lot. - lastPing = BSON( "_id" << o["process"].String() << "ping" << (Date_t) 0 ); - } - - unsigned long long elapsed = 0; - unsigned long long takeover = _lockTimeout; - PingData _lastPingCheck = getLastPing(); - - log( logLvl ) << "checking last ping for lock '" << lockName << "'" << " against process " << _lastPingCheck.get<0>() << " and ping " << _lastPingCheck.get<1>() << endl; - - try { - - Date_t remote = remoteTime( _conn ); - - // Timeout the elapsed time using comparisons of remote clock - // For non-finalized locks, timeout 15 minutes since last seen (ts) - // For finalized locks, timeout 15 minutes since last ping - bool recPingChange = o["state"].numberInt() == 2 && ( _lastPingCheck.get<0>() != lastPing["_id"].String() || _lastPingCheck.get<1>() != lastPing["ping"].Date() ); - bool recTSChange = _lastPingCheck.get<3>() != o["ts"].OID(); - - if( recPingChange || recTSChange ) { - // If the ping has changed since we last checked, mark the current date and time - setLastPing( PingData( lastPing["_id"].String().c_str(), lastPing["ping"].Date(), remote, o["ts"].OID() ) ); - } - else { - - // GOTCHA! Due to network issues, it is possible that the current time - // is less than the remote time. We *have* to check this here, otherwise - // we overflow and our lock breaks. - if(_lastPingCheck.get<2>() >= remote) - elapsed = 0; - else - elapsed = remote - _lastPingCheck.get<2>(); - } - } - catch( LockException& e ) { - - // Remote server cannot be found / is not responsive - warning() << "Could not get remote time from " << _conn << causedBy( e ); - // If our config server is having issues, forget all the pings until we can see it again - resetLastPing(); - - } - - if ( elapsed <= takeover && ! canReenter ) { - log( logLvl ) << "could not force lock '" << lockName << "' because elapsed time " << elapsed << " <= takeover time " << takeover << endl; - *other = o; other->getOwned(); conn.done(); - return false; - } - else if( elapsed > takeover && canReenter ) { - log( logLvl - 1 ) << "not re-entering distributed lock " << lockName << "' because elapsed time " << elapsed << " > takeover time " << takeover << endl; - *other = o; other->getOwned(); conn.done(); - return false; - } - - log( logLvl - 1 ) << ( canReenter ? "re-entering" : "forcing" ) << " lock '" << lockName << "' because " - << ( canReenter ? "re-entering is allowed, " : "" ) - << "elapsed time " << elapsed << " > takeover time " << takeover << endl; - - if( elapsed > takeover ) { - - // Lock may forced, reset our timer if succeeds or fails - // Ensures that another timeout must happen if something borks up here, and resets our pristine - // ping state if acquired. - resetLastPing(); - - try { - - // Check the clock skew again. If we check this before we get a lock - // and after the lock times out, we can be pretty sure the time is - // increasing at the same rate on all servers and therefore our - // timeout is accurate - uassert( 14023, str::stream() << "remote time in cluster " << _conn.toString() << " is now skewed, cannot force lock.", !isRemoteTimeSkewed() ); - - // Make sure we break the lock with the correct "ts" (OID) value, otherwise - // we can overwrite a new lock inserted in the meantime. - conn->update( locksNS , BSON( "_id" << _id["_id"].String() << "state" << o["state"].numberInt() << "ts" << o["ts"] ), - BSON( "$set" << BSON( "state" << 0 ) ) ); - - BSONObj err = conn->getLastErrorDetailed(); - string errMsg = DBClientWithCommands::getLastErrorString(err); - - // TODO: Clean up all the extra code to exit this method, probably with a refactor - if ( !errMsg.empty() || !err["n"].type() || err["n"].numberInt() < 1 ) { - ( errMsg.empty() ? log( logLvl - 1 ) : warning() ) << "Could not force lock '" << lockName << "' " - << ( !errMsg.empty() ? causedBy(errMsg) : string("(another force won)") ) << endl; - *other = o; other->getOwned(); conn.done(); - return false; - } - - } - catch( UpdateNotTheSame& ) { - // Ok to continue since we know we forced at least one lock document, and all lock docs - // are required for a lock to be held. - warning() << "lock forcing " << lockName << " inconsistent" << endl; - } - catch( std::exception& e ) { - conn.done(); - throw LockException( str::stream() << "exception forcing distributed lock " - << lockName << causedBy( e ), 13660); - } - - } - else { - - assert( canReenter ); - - // Lock may be re-entered, reset our timer if succeeds or fails - // Not strictly necessary, but helpful for small timeouts where thread scheduling is significant. - // This ensures that two attempts are still required for a force if not acquired, and resets our - // state if we are acquired. - resetLastPing(); - - // Test that the lock is held by trying to update the finalized state of the lock to the same state - // if it does not update or does not update on all servers, we can't re-enter. - try { - - // Test the lock with the correct "ts" (OID) value - conn->update( locksNS , BSON( "_id" << _id["_id"].String() << "state" << 2 << "ts" << o["ts"] ), - BSON( "$set" << BSON( "state" << 2 ) ) ); - - BSONObj err = conn->getLastErrorDetailed(); - string errMsg = DBClientWithCommands::getLastErrorString(err); - - // TODO: Clean up all the extra code to exit this method, probably with a refactor - if ( ! errMsg.empty() || ! err["n"].type() || err["n"].numberInt() < 1 ) { - ( errMsg.empty() ? log( logLvl - 1 ) : warning() ) << "Could not re-enter lock '" << lockName << "' " - << ( !errMsg.empty() ? causedBy(errMsg) : string("(not sure lock is held)") ) - << " gle: " << err - << endl; - *other = o; other->getOwned(); conn.done(); - return false; - } - - } - catch( UpdateNotTheSame& ) { - // NOT ok to continue since our lock isn't held by all servers, so isn't valid. - warning() << "inconsistent state re-entering lock, lock " << lockName << " not held" << endl; - *other = o; other->getOwned(); conn.done(); - return false; - } - catch( std::exception& e ) { - conn.done(); - throw LockException( str::stream() << "exception re-entering distributed lock " - << lockName << causedBy( e ), 13660); - } - - log( logLvl - 1 ) << "re-entered distributed lock '" << lockName << "'" << endl; - *other = o; other->getOwned(); conn.done(); - return true; - - } - - log( logLvl - 1 ) << "lock '" << lockName << "' successfully forced" << endl; - - // We don't need the ts value in the query, since we will only ever replace locks with state=0. - } - // Case 3: We have an expired lock - else if ( o["ts"].type() ) { - queryBuilder.append( o["ts"] ); - } - } - - // Always reset our ping if we're trying to get a lock, since getting a lock implies the lock state is open - // and no locks need to be forced. If anything goes wrong, we don't want to remember an old lock. - resetLastPing(); - - bool gotLock = false; - BSONObj currLock; - - BSONObj lockDetails = BSON( "state" << 1 << "who" << getDistLockId() << "process" << _processId << - "when" << jsTime() << "why" << why << "ts" << OID::gen() ); - BSONObj whatIWant = BSON( "$set" << lockDetails ); - - BSONObj query = queryBuilder.obj(); - - string lockName = _name + string("/") + _processId; - - try { - - // Main codepath to acquire lock - - log( logLvl ) << "about to acquire distributed lock '" << lockName << ":\n" - << lockDetails.jsonString(Strict, true) << "\n" - << query.jsonString(Strict, true) << endl; - - conn->update( locksNS , query , whatIWant ); - - BSONObj err = conn->getLastErrorDetailed(); - string errMsg = DBClientWithCommands::getLastErrorString(err); - - currLock = conn->findOne( locksNS , _id ); - - if ( !errMsg.empty() || !err["n"].type() || err["n"].numberInt() < 1 ) { - ( errMsg.empty() ? log( logLvl - 1 ) : warning() ) << "could not acquire lock '" << lockName << "' " - << ( !errMsg.empty() ? causedBy( errMsg ) : string("(another update won)") ) << endl; - *other = currLock; - other->getOwned(); - gotLock = false; - } - else { - gotLock = true; - } - - } - catch ( UpdateNotTheSame& up ) { - - // this means our update got through on some, but not others - warning() << "distributed lock '" << lockName << " did not propagate properly." << causedBy( up ) << endl; - - // Overall protection derives from: - // All unlocking updates use the ts value when setting state to 0 - // This ensures that during locking, we can override all smaller ts locks with - // our own safe ts value and not be unlocked afterward. - for ( unsigned i = 0; i < up.size(); i++ ) { - - ScopedDbConnection indDB( up[i].first ); - BSONObj indUpdate; - - try { - - indUpdate = indDB->findOne( locksNS , _id ); - - // If we override this lock in any way, grab and protect it. - // We assume/ensure that if a process does not have all lock documents, it is no longer - // holding the lock. - // Note - finalized locks may compete too, but we know they've won already if competing - // in this round. Cleanup of crashes during finalizing may take a few tries. - if( indUpdate["ts"] < lockDetails["ts"] || indUpdate["state"].numberInt() == 0 ) { - - BSONObj grabQuery = BSON( "_id" << _id["_id"].String() << "ts" << indUpdate["ts"].OID() ); - - // Change ts so we won't be forced, state so we won't be relocked - BSONObj grabChanges = BSON( "ts" << lockDetails["ts"].OID() << "state" << 1 ); - - // Either our update will succeed, and we'll grab the lock, or it will fail b/c some other - // process grabbed the lock (which will change the ts), but the lock will be set until forcing - indDB->update( locksNS, grabQuery, BSON( "$set" << grabChanges ) ); - - indUpdate = indDB->findOne( locksNS, _id ); - - // Our lock should now be set until forcing. - assert( indUpdate["state"].numberInt() == 1 ); - - } - // else our lock is the same, in which case we're safe, or it's a bigger lock, - // in which case we won't need to protect anything since we won't have the lock. - - } - catch( std::exception& e ) { - conn.done(); - throw LockException( str::stream() << "distributed lock " << lockName - << " had errors communicating with individual server " - << up[1].first << causedBy( e ), 13661 ); - } - - assert( !indUpdate.isEmpty() ); - - // Find max TS value - if ( currLock.isEmpty() || currLock["ts"] < indUpdate["ts"] ) { - currLock = indUpdate.getOwned(); - } - - indDB.done(); - - } - - // Locks on all servers are now set and safe until forcing - - if ( currLock["ts"] == lockDetails["ts"] ) { - log( logLvl - 1 ) << "lock update won, completing lock propagation for '" << lockName << "'" << endl; - gotLock = true; - } - else { - log( logLvl - 1 ) << "lock update lost, lock '" << lockName << "' not propagated." << endl; - - // Register the lock for deletion, to speed up failover - // Not strictly necessary, but helpful - distLockPinger.addUnlockOID( lockDetails["ts"].OID() ); - - gotLock = false; - } - } - catch( std::exception& e ) { - conn.done(); - throw LockException( str::stream() << "exception creating distributed lock " - << lockName << causedBy( e ), 13663 ); - } - - // Complete lock propagation - if( gotLock ) { - - // This is now safe, since we know that no new locks will be placed on top of the ones we've checked for at - // least 15 minutes. Sets the state = 2, so that future clients can determine that the lock is truly set. - // The invariant for rollbacks is that we will never force locks with state = 2 and active pings, since that - // indicates the lock is active, but this means the process creating/destroying them must explicitly poll - // when something goes wrong. - try { - - BSONObjBuilder finalLockDetails; - BSONObjIterator bi( lockDetails ); - while( bi.more() ) { - BSONElement el = bi.next(); - if( (string) ( el.fieldName() ) == "state" ) - finalLockDetails.append( "state", 2 ); - else finalLockDetails.append( el ); - } - - conn->update( locksNS , _id , BSON( "$set" << finalLockDetails.obj() ) ); - - BSONObj err = conn->getLastErrorDetailed(); - string errMsg = DBClientWithCommands::getLastErrorString(err); - - currLock = conn->findOne( locksNS , _id ); - - if ( !errMsg.empty() || !err["n"].type() || err["n"].numberInt() < 1 ) { - warning() << "could not finalize winning lock " << lockName - << ( !errMsg.empty() ? causedBy( errMsg ) : " (did not update lock) " ) << endl; - gotLock = false; - } - else { - // SUCCESS! - gotLock = true; - } - - } - catch( std::exception& e ) { - conn.done(); - - // Register the bad final lock for deletion, in case it exists - distLockPinger.addUnlockOID( lockDetails["ts"].OID() ); - - throw LockException( str::stream() << "exception finalizing winning lock" - << causedBy( e ), 13662 ); - } - - } - - *other = currLock; - other->getOwned(); - - // Log our lock results - if(gotLock) - log( logLvl - 1 ) << "distributed lock '" << lockName << "' acquired, ts : " << currLock["ts"].OID() << endl; - else - log( logLvl - 1 ) << "distributed lock '" << lockName << "' was not acquired." << endl; - - conn.done(); - - return gotLock; - } - - // Unlock now takes an optional pointer to the lock, so you can be specific about which - // particular lock you want to unlock. This is required when the config server is down, - // and so cannot tell you what lock ts you should try later. - void DistributedLock::unlock( BSONObj* oldLockPtr ) { - - assert( _name != "" ); - - string lockName = _name + string("/") + _processId; - - const int maxAttempts = 3; - int attempted = 0; - - BSONObj oldLock; - if( oldLockPtr ) oldLock = *oldLockPtr; - - while ( ++attempted <= maxAttempts ) { - - ScopedDbConnection conn( _conn ); - - try { - - if( oldLock.isEmpty() ) - oldLock = conn->findOne( locksNS, _id ); - - if( oldLock["state"].eoo() || oldLock["state"].numberInt() != 2 || oldLock["ts"].eoo() ) { - warning() << "cannot unlock invalid distributed lock " << oldLock << endl; - conn.done(); - break; - } - - // Use ts when updating lock, so that new locks can be sure they won't get trampled. - conn->update( locksNS , - BSON( "_id" << _id["_id"].String() << "ts" << oldLock["ts"].OID() ), - BSON( "$set" << BSON( "state" << 0 ) ) ); - - // Check that the lock was actually unlocked... if not, try again - BSONObj err = conn->getLastErrorDetailed(); - string errMsg = DBClientWithCommands::getLastErrorString(err); - - if ( !errMsg.empty() || !err["n"].type() || err["n"].numberInt() < 1 ){ - warning() << "distributed lock unlock update failed, retrying " - << ( errMsg.empty() ? causedBy( "( update not registered )" ) : causedBy( errMsg ) ) << endl; - conn.done(); - continue; - } - - log( logLvl - 1 ) << "distributed lock '" << lockName << "' unlocked. " << endl; - conn.done(); - return; - } - catch( UpdateNotTheSame& ) { - log( logLvl - 1 ) << "distributed lock '" << lockName << "' unlocked (messily). " << endl; - conn.done(); - break; - } - catch ( std::exception& e) { - warning() << "distributed lock '" << lockName << "' failed unlock attempt." - << causedBy( e ) << endl; - - conn.done(); - // TODO: If our lock timeout is small, sleeping this long may be unsafe. - if( attempted != maxAttempts) sleepsecs(1 << attempted); - } - } - - if( attempted > maxAttempts && ! oldLock.isEmpty() && ! oldLock["ts"].eoo() ) { - - log( logLvl - 1 ) << "could not unlock distributed lock with ts " << oldLock["ts"].OID() - << ", will attempt again later" << endl; - - // We couldn't unlock the lock at all, so try again later in the pinging thread... - distLockPinger.addUnlockOID( oldLock["ts"].OID() ); - } - else if( attempted > maxAttempts ) { - warning() << "could not unlock untracked distributed lock, a manual force may be required" << endl; - } - - warning() << "distributed lock '" << lockName << "' couldn't consummate unlock request. " - << "lock may be taken over after " << ( _lockTimeout / (60 * 1000) ) - << " minutes timeout." << endl; - } - - - -} diff --git a/client/distlock.h b/client/distlock.h deleted file mode 100644 index 106a5d00001..00000000000 --- a/client/distlock.h +++ /dev/null @@ -1,244 +0,0 @@ -// distlock.h - -/* Copyright 2009 10gen Inc. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -#pragma once - -#include "../pch.h" -#include "dbclient.h" -#include "connpool.h" -#include "redef_macros.h" -#include "syncclusterconnection.h" - -#define LOCK_TIMEOUT (15 * 60 * 1000) -#define LOCK_SKEW_FACTOR (30) -#define LOCK_PING (LOCK_TIMEOUT / LOCK_SKEW_FACTOR) -#define MAX_LOCK_NET_SKEW (LOCK_TIMEOUT / LOCK_SKEW_FACTOR) -#define MAX_LOCK_CLOCK_SKEW (LOCK_TIMEOUT / LOCK_SKEW_FACTOR) -#define NUM_LOCK_SKEW_CHECKS (3) - -// The maximum clock skew we need to handle between config servers is -// 2 * MAX_LOCK_NET_SKEW + MAX_LOCK_CLOCK_SKEW. - -// Net effect of *this* clock being slow is effectively a multiplier on the max net skew -// and a linear increase or decrease of the max clock skew. - -namespace mongo { - - /** - * Exception class to encapsulate exceptions while managing distributed locks - */ - class LockException : public DBException { - public: - LockException( const char * msg , int code ) : DBException( msg, code ) {} - LockException( const string& msg, int code ) : DBException( msg, code ) {} - virtual ~LockException() throw() { } - }; - - /** - * Indicates an error in retrieving time values from remote servers. - */ - class TimeNotFoundException : public LockException { - public: - TimeNotFoundException( const char * msg , int code ) : LockException( msg, code ) {} - TimeNotFoundException( const string& msg, int code ) : LockException( msg, code ) {} - virtual ~TimeNotFoundException() throw() { } - }; - - /** - * The distributed lock is a configdb backed way of synchronizing system-wide tasks. A task must be identified by a - * unique name across the system (e.g., "balancer"). A lock is taken by writing a document in the configdb's locks - * collection with that name. - * - * To be maintained, each taken lock needs to be revalidaded ("pinged") within a pre-established amount of time. This - * class does this maintenance automatically once a DistributedLock object was constructed. - */ - class DistributedLock { - public: - - static LabeledLevel logLvl; - - typedef boost::tuple<string, Date_t, Date_t, OID> PingData; - - class LastPings { - public: - LastPings() : _mutex( "DistributedLock::LastPings" ) {} - ~LastPings(){} - - PingData getLastPing( const ConnectionString& conn, const string& lockName ); - void setLastPing( const ConnectionString& conn, const string& lockName, const PingData& pd ); - - mongo::mutex _mutex; - map< std::pair<string, string>, PingData > _lastPings; - }; - - static LastPings lastPings; - - /** - * The constructor does not connect to the configdb yet and constructing does not mean the lock was acquired. - * Construction does trigger a lock "pinging" mechanism, though. - * - * @param conn address of config(s) server(s) - * @param name identifier for the lock - * @param lockTimeout how long can the log go "unpinged" before a new attempt to lock steals it (in minutes). - * @param lockPing how long to wait between lock pings - * @param legacy use legacy logic - * - */ - DistributedLock( const ConnectionString& conn , const string& name , unsigned long long lockTimeout = 0, bool asProcess = false ); - ~DistributedLock(){}; - - /** - * Attempts to acquire 'this' lock, checking if it could or should be stolen from the previous holder. Please - * consider using the dist_lock_try construct to acquire this lock in an exception safe way. - * - * @param why human readable description of why the lock is being taken (used to log) - * @param whether this is a lock re-entry or a new lock - * @param other configdb's lock document that is currently holding the lock, if lock is taken, or our own lock - * details if not - * @return true if it managed to grab the lock - */ - bool lock_try( const string& why , bool reenter = false, BSONObj * other = 0 ); - - /** - * Releases a previously taken lock. - */ - void unlock( BSONObj* oldLockPtr = NULL ); - - Date_t getRemoteTime(); - - bool isRemoteTimeSkewed(); - - const string& getProcessId(); - - const ConnectionString& getRemoteConnection(); - - /** - * Check the skew between a cluster of servers - */ - static bool checkSkew( const ConnectionString& cluster, unsigned skewChecks = NUM_LOCK_SKEW_CHECKS, unsigned long long maxClockSkew = MAX_LOCK_CLOCK_SKEW, unsigned long long maxNetSkew = MAX_LOCK_NET_SKEW ); - - /** - * Get the remote time from a server or cluster - */ - static Date_t remoteTime( const ConnectionString& cluster, unsigned long long maxNetSkew = MAX_LOCK_NET_SKEW ); - - static bool killPinger( DistributedLock& lock ); - - /** - * Namespace for lock pings - */ - static const string lockPingNS; - - /** - * Namespace for locks - */ - static const string locksNS; - - const ConnectionString _conn; - const string _name; - const BSONObj _id; - const string _processId; - - // Timeout for lock, usually LOCK_TIMEOUT - const unsigned long long _lockTimeout; - const unsigned long long _maxClockSkew; - const unsigned long long _maxNetSkew; - const unsigned long long _lockPing; - - private: - - void resetLastPing(){ lastPings.setLastPing( _conn, _name, PingData() ); } - void setLastPing( const PingData& pd ){ lastPings.setLastPing( _conn, _name, pd ); } - PingData getLastPing(){ return lastPings.getLastPing( _conn, _name ); } - - // May or may not exist, depending on startup - mongo::mutex _mutex; - string _threadId; - - }; - - class dist_lock_try { - public: - - dist_lock_try() : _lock(NULL), _got(false) {} - - dist_lock_try( const dist_lock_try& that ) : _lock(that._lock), _got(that._got), _other(that._other) { - _other.getOwned(); - - // Make sure the lock ownership passes to this object, - // so we only unlock once. - ((dist_lock_try&) that)._got = false; - ((dist_lock_try&) that)._lock = NULL; - ((dist_lock_try&) that)._other = BSONObj(); - } - - // Needed so we can handle lock exceptions in context of lock try. - dist_lock_try& operator=( const dist_lock_try& that ){ - - if( this == &that ) return *this; - - _lock = that._lock; - _got = that._got; - _other = that._other; - _other.getOwned(); - _why = that._why; - - // Make sure the lock ownership passes to this object, - // so we only unlock once. - ((dist_lock_try&) that)._got = false; - ((dist_lock_try&) that)._lock = NULL; - ((dist_lock_try&) that)._other = BSONObj(); - - return *this; - } - - dist_lock_try( DistributedLock * lock , string why ) - : _lock(lock), _why(why) { - _got = _lock->lock_try( why , false , &_other ); - } - - ~dist_lock_try() { - if ( _got ) { - assert( ! _other.isEmpty() ); - _lock->unlock( &_other ); - } - } - - bool reestablish(){ - return retry(); - } - - bool retry() { - assert( _lock ); - assert( _got ); - assert( ! _other.isEmpty() ); - - return _got = _lock->lock_try( _why , true, &_other ); - } - - bool got() const { return _got; } - BSONObj other() const { return _other; } - - private: - DistributedLock * _lock; - bool _got; - BSONObj _other; - string _why; - }; - -} - diff --git a/client/distlock_test.cpp b/client/distlock_test.cpp deleted file mode 100644 index 5f37e6b3e9b..00000000000 --- a/client/distlock_test.cpp +++ /dev/null @@ -1,445 +0,0 @@ -// distlock_test.h - -/* Copyright 2009 10gen Inc. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -#include <iostream> -#include "../pch.h" -#include "dbclient.h" -#include "distlock.h" -#include "../db/commands.h" -#include "../util/bson_util.h" - -// Modify some config options for the RNG, since they cause MSVC to fail -#include <boost/config.hpp> - -#if defined(BOOST_MSVC) && defined(BOOST_NO_MEMBER_TEMPLATE_FRIENDS) -#undef BOOST_NO_MEMBER_TEMPLATE_FRIENDS -#define BOOST_RNG_HACK -#endif - -// Well, sort-of cross-platform RNG -#include <boost/random/mersenne_twister.hpp> - -#ifdef BOOST_RNG_HACK -#define BOOST_NO_MEMBER_TEMPLATE_FRIENDS -#undef BOOST_RNG_HACK -#endif - - -#include <boost/random/uniform_int.hpp> -#include <boost/random/variate_generator.hpp> - - -// TODO: Make a method in BSONObj if useful, don't modify for now -#define string_field(obj, name, def) ( obj.hasField(name) ? obj[name].String() : def ) -#define number_field(obj, name, def) ( obj.hasField(name) ? obj[name].Number() : def ) - -namespace mongo { - - class TestDistLockWithSync: public Command { - public: - TestDistLockWithSync() : - Command("_testDistLockWithSyncCluster") { - } - virtual void help(stringstream& help) const { - help << "should not be calling this directly" << endl; - } - - virtual bool slaveOk() const { - return false; - } - virtual bool adminOnly() const { - return true; - } - virtual LockType locktype() const { - return NONE; - } - - static void runThread() { - while (keepGoing) { - if (current->lock_try( "test" )) { - count++; - int before = count; - sleepmillis(3); - int after = count; - - if (after != before) { - error() << " before: " << before << " after: " << after - << endl; - } - - current->unlock(); - } - } - } - - bool run(const string&, BSONObj& cmdObj, int, string& errmsg, - BSONObjBuilder& result, bool) { - Timer t; - DistributedLock lk(ConnectionString(cmdObj["host"].String(), - ConnectionString::SYNC), "testdistlockwithsync", 0, 0); - current = &lk; - count = 0; - gotit = 0; - errors = 0; - keepGoing = true; - - vector<shared_ptr<boost::thread> > l; - for (int i = 0; i < 4; i++) { - l.push_back( - shared_ptr<boost::thread> (new boost::thread(runThread))); - } - - int secs = 10; - if (cmdObj["secs"].isNumber()) - secs = cmdObj["secs"].numberInt(); - sleepsecs(secs); - keepGoing = false; - - for (unsigned i = 0; i < l.size(); i++) - l[i]->join(); - - current = 0; - - result.append("count", count); - result.append("gotit", gotit); - result.append("errors", errors); - result.append("timeMS", t.millis()); - - return errors == 0; - } - - // variables for test - static DistributedLock * current; - static int gotit; - static int errors; - static AtomicUInt count; - - static bool keepGoing; - - } testDistLockWithSyncCmd; - - DistributedLock * TestDistLockWithSync::current; - AtomicUInt TestDistLockWithSync::count; - int TestDistLockWithSync::gotit; - int TestDistLockWithSync::errors; - bool TestDistLockWithSync::keepGoing; - - - - class TestDistLockWithSkew: public Command { - public: - - static const int logLvl = 1; - - TestDistLockWithSkew() : - Command("_testDistLockWithSkew") { - } - virtual void help(stringstream& help) const { - help << "should not be calling this directly" << endl; - } - - virtual bool slaveOk() const { - return false; - } - virtual bool adminOnly() const { - return true; - } - virtual LockType locktype() const { - return NONE; - } - - void runThread(ConnectionString& hostConn, unsigned threadId, unsigned seed, - BSONObj& cmdObj, BSONObjBuilder& result) { - - stringstream ss; - ss << "thread-" << threadId; - setThreadName(ss.str().c_str()); - - // Lock name - string lockName = string_field(cmdObj, "lockName", this->name + "_lock"); - - // Range of clock skew in diff threads - int skewRange = (int) number_field(cmdObj, "skewRange", 1); - - // How long to wait with the lock - int threadWait = (int) number_field(cmdObj, "threadWait", 30); - if(threadWait <= 0) threadWait = 1; - - // Max amount of time (ms) a thread waits before checking the lock again - int threadSleep = (int) number_field(cmdObj, "threadSleep", 30); - if(threadSleep <= 0) threadSleep = 1; - - // How long until the lock is forced in ms, only compared locally - unsigned long long takeoverMS = (unsigned long long) number_field(cmdObj, "takeoverMS", 0); - - // Whether or not we should hang some threads - int hangThreads = (int) number_field(cmdObj, "hangThreads", 0); - - - boost::mt19937 gen((boost::mt19937::result_type) seed); - - boost::variate_generator<boost::mt19937&, boost::uniform_int<> > randomSkew(gen, boost::uniform_int<>(0, skewRange)); - boost::variate_generator<boost::mt19937&, boost::uniform_int<> > randomWait(gen, boost::uniform_int<>(1, threadWait)); - boost::variate_generator<boost::mt19937&, boost::uniform_int<> > randomSleep(gen, boost::uniform_int<>(1, threadSleep)); - boost::variate_generator<boost::mt19937&, boost::uniform_int<> > randomNewLock(gen, boost::uniform_int<>(0, 3)); - - - int skew = 0; - if (!lock.get()) { - - // Pick a skew, but the first two threads skew the whole range - if(threadId == 0) - skew = -skewRange / 2; - else if(threadId == 1) - skew = skewRange / 2; - else skew = randomSkew() - (skewRange / 2); - - // Skew this thread - jsTimeVirtualThreadSkew( skew ); - - log() << "Initializing lock with skew of " << skew << " for thread " << threadId << endl; - - lock.reset(new DistributedLock(hostConn, lockName, takeoverMS, true )); - - log() << "Skewed time " << jsTime() << " for thread " << threadId << endl - << " max wait (with lock: " << threadWait << ", after lock: " << threadSleep << ")" << endl - << " takeover in " << takeoverMS << "(ms remote)" << endl; - - } - - DistributedLock* myLock = lock.get(); - - bool errors = false; - BSONObj lockObj; - while (keepGoing) { - try { - - if (myLock->lock_try("Testing distributed lock with skew.", false, &lockObj )) { - - log() << "**** Locked for thread " << threadId << " with ts " << lockObj["ts"] << endl; - - if( count % 2 == 1 && ! myLock->lock_try( "Testing lock re-entry.", true ) ) { - errors = true; - log() << "**** !Could not re-enter lock already held" << endl; - break; - } - - if( count % 3 == 1 && myLock->lock_try( "Testing lock non-re-entry.", false ) ) { - errors = true; - log() << "**** !Invalid lock re-entry" << endl; - break; - } - - count++; - int before = count; - int sleep = randomWait(); - sleepmillis(sleep); - int after = count; - - if(after != before) { - errors = true; - log() << "**** !Bad increment while sleeping with lock for: " << sleep << "ms" << endl; - break; - } - - // Unlock only half the time... - if(hangThreads == 0 || threadId % hangThreads != 0) { - log() << "**** Unlocking for thread " << threadId << " with ts " << lockObj["ts"] << endl; - myLock->unlock( &lockObj ); - } - else { - log() << "**** Not unlocking for thread " << threadId << endl; - assert( DistributedLock::killPinger( *myLock ) ); - // We're simulating a crashed process... - break; - } - } - - } - catch( LockException& e ) { - log() << "*** !Could not try distributed lock." << causedBy( e ) << endl; - break; - } - - // Create a new lock 1/3 of the time - if( randomNewLock() > 1 ){ - lock.reset(new DistributedLock( hostConn, lockName, takeoverMS, true )); - myLock = lock.get(); - } - - sleepmillis(randomSleep()); - } - - result << "errors" << errors - << "skew" << skew - << "takeover" << (long long) takeoverMS - << "localTimeout" << (takeoverMS > 0); - - } - - void test(ConnectionString& hostConn, string& lockName, unsigned seed) { - return; - } - - bool run(const string&, BSONObj& cmdObj, int, string& errmsg, - BSONObjBuilder& result, bool) { - - Timer t; - - ConnectionString hostConn(cmdObj["host"].String(), - ConnectionString::SYNC); - - unsigned seed = (unsigned) number_field(cmdObj, "seed", 0); - int numThreads = (int) number_field(cmdObj, "numThreads", 4); - int wait = (int) number_field(cmdObj, "wait", 10000); - - log() << "Starting " << this->name << " with -" << endl - << " seed: " << seed << endl - << " numThreads: " << numThreads << endl - << " total wait: " << wait << endl << endl; - - // Skew host clocks if needed - try { - skewClocks( hostConn, cmdObj ); - } - catch( DBException e ) { - errmsg = str::stream() << "Clocks could not be skewed." << causedBy( e ); - return false; - } - - count = 0; - keepGoing = true; - - vector<shared_ptr<boost::thread> > threads; - vector<shared_ptr<BSONObjBuilder> > results; - for (int i = 0; i < numThreads; i++) { - results.push_back(shared_ptr<BSONObjBuilder> (new BSONObjBuilder())); - threads.push_back(shared_ptr<boost::thread> (new boost::thread( - boost::bind(&TestDistLockWithSkew::runThread, this, - hostConn, (unsigned) i, seed + i, boost::ref(cmdObj), - boost::ref(*(results[i].get())))))); - } - - sleepsecs(wait / 1000); - keepGoing = false; - - bool errors = false; - for (unsigned i = 0; i < threads.size(); i++) { - threads[i]->join(); - errors = errors || results[i].get()->obj()["errors"].Bool(); - } - - result.append("count", count); - result.append("errors", errors); - result.append("timeMS", t.millis()); - - return !errors; - - } - - /** - * Skews the clocks of a remote cluster by a particular amount, specified by - * the "skewHosts" element in a BSONObj. - */ - static void skewClocks( ConnectionString& cluster, BSONObj& cmdObj ) { - - vector<long long> skew; - if(cmdObj.hasField("skewHosts")) { - bsonArrToNumVector<long long>(cmdObj["skewHosts"], skew); - } - else { - log( logLvl ) << "No host clocks to skew." << endl; - return; - } - - log( logLvl ) << "Skewing clocks of hosts " << cluster << endl; - - unsigned s = 0; - for(vector<long long>::iterator i = skew.begin(); i != skew.end(); ++i,s++) { - - ConnectionString server( cluster.getServers()[s] ); - ScopedDbConnection conn( server ); - - BSONObj result; - try { - bool success = conn->runCommand( string("admin"), BSON( "_skewClockCommand" << 1 << "skew" << *i ), result ); - - uassert(13678, str::stream() << "Could not communicate with server " << server.toString() << " in cluster " << cluster.toString() << " to change skew by " << *i, success ); - - log( logLvl + 1 ) << " Skewed host " << server << " clock by " << *i << endl; - } - catch(...) { - conn.done(); - throw; - } - - conn.done(); - - } - - } - - // variables for test - thread_specific_ptr<DistributedLock> lock; - AtomicUInt count; - bool keepGoing; - - } testDistLockWithSkewCmd; - - - /** - * Utility command to virtually skew the clock of a mongo server a particular amount. - * This skews the clock globally, per-thread skew is also possible. - */ - class SkewClockCommand: public Command { - public: - SkewClockCommand() : - Command("_skewClockCommand") { - } - virtual void help(stringstream& help) const { - help << "should not be calling this directly" << endl; - } - - virtual bool slaveOk() const { - return false; - } - virtual bool adminOnly() const { - return true; - } - virtual LockType locktype() const { - return NONE; - } - - bool run(const string&, BSONObj& cmdObj, int, string& errmsg, - BSONObjBuilder& result, bool) { - - long long skew = (long long) number_field(cmdObj, "skew", 0); - - log() << "Adjusting jsTime() clock skew to " << skew << endl; - - jsTimeVirtualSkew( skew ); - - log() << "JSTime adjusted, now is " << jsTime() << endl; - - return true; - - } - - } testSkewClockCommand; - -} - diff --git a/client/examples/authTest.cpp b/client/examples/authTest.cpp deleted file mode 100644 index 71cdd390cff..00000000000 --- a/client/examples/authTest.cpp +++ /dev/null @@ -1,54 +0,0 @@ -// authTest.cpp - -/* Copyright 2009 10gen Inc. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -#include <iostream> - -#include "client/dbclient.h" - -using namespace mongo; - -int main( int argc, const char **argv ) { - - const char *port = "27017"; - if ( argc != 1 ) { - if ( argc != 3 ) - throw -12; - port = argv[ 2 ]; - } - - DBClientConnection conn; - string errmsg; - if ( ! conn.connect( string( "127.0.0.1:" ) + port , errmsg ) ) { - cout << "couldn't connect : " << errmsg << endl; - throw -11; - } - - { - // clean up old data from any previous tests - conn.remove( "test.system.users" , BSONObj() ); - } - - conn.insert( "test.system.users" , BSON( "user" << "eliot" << "pwd" << conn.createPasswordDigest( "eliot" , "bar" ) ) ); - - errmsg.clear(); - bool ok = conn.auth( "test" , "eliot" , "bar" , errmsg ); - if ( ! ok ) - cout << errmsg << endl; - MONGO_assert( ok ); - - MONGO_assert( ! conn.auth( "test" , "eliot" , "bars" , errmsg ) ); -} diff --git a/client/examples/clientTest.cpp b/client/examples/clientTest.cpp deleted file mode 100644 index aaea6bd1bdf..00000000000 --- a/client/examples/clientTest.cpp +++ /dev/null @@ -1,279 +0,0 @@ -// clientTest.cpp - -/* Copyright 2009 10gen Inc. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -/** - * a simple test for the c++ driver - */ - -// this header should be first to ensure that it includes cleanly in any context -#include "client/dbclient.h" - -#include <iostream> - -#ifndef assert -# define assert(x) MONGO_assert(x) -#endif - -using namespace std; -using namespace mongo; - -int main( int argc, const char **argv ) { - - const char *port = "27017"; - if ( argc != 1 ) { - if ( argc != 3 ) - throw -12; - port = argv[ 2 ]; - } - - DBClientConnection conn; - string errmsg; - if ( ! conn.connect( string( "127.0.0.1:" ) + port , errmsg ) ) { - cout << "couldn't connect : " << errmsg << endl; - throw -11; - } - - const char * ns = "test.test1"; - - conn.dropCollection(ns); - - // clean up old data from any previous tests - conn.remove( ns, BSONObj() ); - assert( conn.findOne( ns , BSONObj() ).isEmpty() ); - - // test insert - conn.insert( ns ,BSON( "name" << "eliot" << "num" << 1 ) ); - assert( ! conn.findOne( ns , BSONObj() ).isEmpty() ); - - // test remove - conn.remove( ns, BSONObj() ); - assert( conn.findOne( ns , BSONObj() ).isEmpty() ); - - - // insert, findOne testing - conn.insert( ns , BSON( "name" << "eliot" << "num" << 1 ) ); - { - BSONObj res = conn.findOne( ns , BSONObj() ); - assert( strstr( res.getStringField( "name" ) , "eliot" ) ); - assert( ! strstr( res.getStringField( "name2" ) , "eliot" ) ); - assert( 1 == res.getIntField( "num" ) ); - } - - - // cursor - conn.insert( ns ,BSON( "name" << "sara" << "num" << 2 ) ); - { - auto_ptr<DBClientCursor> cursor = conn.query( ns , BSONObj() ); - int count = 0; - while ( cursor->more() ) { - count++; - BSONObj obj = cursor->next(); - } - assert( count == 2 ); - } - - { - auto_ptr<DBClientCursor> cursor = conn.query( ns , BSON( "num" << 1 ) ); - int count = 0; - while ( cursor->more() ) { - count++; - BSONObj obj = cursor->next(); - } - assert( count == 1 ); - } - - { - auto_ptr<DBClientCursor> cursor = conn.query( ns , BSON( "num" << 3 ) ); - int count = 0; - while ( cursor->more() ) { - count++; - BSONObj obj = cursor->next(); - } - assert( count == 0 ); - } - - // update - { - BSONObj res = conn.findOne( ns , BSONObjBuilder().append( "name" , "eliot" ).obj() ); - assert( ! strstr( res.getStringField( "name2" ) , "eliot" ) ); - - BSONObj after = BSONObjBuilder().appendElements( res ).append( "name2" , "h" ).obj(); - - conn.update( ns , BSONObjBuilder().append( "name" , "eliot2" ).obj() , after ); - res = conn.findOne( ns , BSONObjBuilder().append( "name" , "eliot" ).obj() ); - assert( ! strstr( res.getStringField( "name2" ) , "eliot" ) ); - assert( conn.findOne( ns , BSONObjBuilder().append( "name" , "eliot2" ).obj() ).isEmpty() ); - - conn.update( ns , BSONObjBuilder().append( "name" , "eliot" ).obj() , after ); - res = conn.findOne( ns , BSONObjBuilder().append( "name" , "eliot" ).obj() ); - assert( strstr( res.getStringField( "name" ) , "eliot" ) ); - assert( strstr( res.getStringField( "name2" ) , "h" ) ); - assert( conn.findOne( ns , BSONObjBuilder().append( "name" , "eliot2" ).obj() ).isEmpty() ); - - // upsert - conn.update( ns , BSONObjBuilder().append( "name" , "eliot2" ).obj() , after , 1 ); - assert( ! conn.findOne( ns , BSONObjBuilder().append( "name" , "eliot" ).obj() ).isEmpty() ); - - } - - { - // ensure index - assert( conn.ensureIndex( ns , BSON( "name" << 1 ) ) ); - assert( ! conn.ensureIndex( ns , BSON( "name" << 1 ) ) ); - } - - { - // hint related tests - assert( conn.findOne(ns, "{}")["name"].str() == "sara" ); - - assert( conn.findOne(ns, "{ name : 'eliot' }")["name"].str() == "eliot" ); - assert( conn.getLastError() == "" ); - - // nonexistent index test - bool asserted = false; - try { - conn.findOne(ns, Query("{name:\"eliot\"}").hint("{foo:1}")); - } - catch ( ... ) { - asserted = true; - } - assert( asserted ); - - //existing index - assert( conn.findOne(ns, Query("{name:'eliot'}").hint("{name:1}")).hasElement("name") ); - - // run validate - assert( conn.validate( ns ) ); - } - - { - // timestamp test - - const char * tsns = "test.tstest1"; - conn.dropCollection( tsns ); - - { - mongo::BSONObjBuilder b; - b.appendTimestamp( "ts" ); - conn.insert( tsns , b.obj() ); - } - - mongo::BSONObj out = conn.findOne( tsns , mongo::BSONObj() ); - Date_t oldTime = out["ts"].timestampTime(); - unsigned int oldInc = out["ts"].timestampInc(); - - { - mongo::BSONObjBuilder b1; - b1.append( out["_id"] ); - - mongo::BSONObjBuilder b2; - b2.append( out["_id"] ); - b2.appendTimestamp( "ts" ); - - conn.update( tsns , b1.obj() , b2.obj() ); - } - - BSONObj found = conn.findOne( tsns , mongo::BSONObj() ); - cout << "old: " << out << "\nnew: " << found << endl; - assert( ( oldTime < found["ts"].timestampTime() ) || - ( oldTime == found["ts"].timestampTime() && oldInc < found["ts"].timestampInc() ) ); - - } - - { - // check that killcursors doesn't affect last error - assert( conn.getLastError().empty() ); - - BufBuilder b; - b.appendNum( (int)0 ); // reserved - b.appendNum( (int)-1 ); // invalid # of cursors triggers exception - b.appendNum( (int)-1 ); // bogus cursor id - - Message m; - m.setData( dbKillCursors, b.buf(), b.len() ); - - // say() is protected in DBClientConnection, so get superclass - static_cast< DBConnector* >( &conn )->say( m ); - - assert( conn.getLastError().empty() ); - } - - { - list<string> l = conn.getDatabaseNames(); - for ( list<string>::iterator i = l.begin(); i != l.end(); i++ ) { - cout << "db name : " << *i << endl; - } - - l = conn.getCollectionNames( "test" ); - for ( list<string>::iterator i = l.begin(); i != l.end(); i++ ) { - cout << "coll name : " << *i << endl; - } - } - - { - //Map Reduce (this mostly just tests that it compiles with all output types) - const string ns = "test.mr"; - conn.insert(ns, BSON("a" << 1)); - conn.insert(ns, BSON("a" << 1)); - - const char* map = "function() { emit(this.a, 1); }"; - const char* reduce = "function(key, values) { return Array.sum(values); }"; - - const string outcoll = ns + ".out"; - - BSONObj out; - out = conn.mapreduce(ns, map, reduce, BSONObj()); // default to inline - //MONGO_PRINT(out); - out = conn.mapreduce(ns, map, reduce, BSONObj(), outcoll); - //MONGO_PRINT(out); - out = conn.mapreduce(ns, map, reduce, BSONObj(), outcoll.c_str()); - //MONGO_PRINT(out); - out = conn.mapreduce(ns, map, reduce, BSONObj(), BSON("reduce" << outcoll)); - //MONGO_PRINT(out); - } - - { - // test timeouts - - DBClientConnection conn( true , 0 , 2 ); - if ( ! conn.connect( string( "127.0.0.1:" ) + port , errmsg ) ) { - cout << "couldn't connect : " << errmsg << endl; - throw -11; - } - conn.insert( "test.totest" , BSON( "x" << 1 ) ); - BSONObj res; - - bool gotError = false; - assert( conn.eval( "test" , "return db.totest.findOne().x" , res ) ); - try { - conn.eval( "test" , "sleep(5000); return db.totest.findOne().x" , res ); - } - catch ( std::exception& e ) { - gotError = true; - log() << e.what() << endl; - } - assert( gotError ); - // sleep so the server isn't locked anymore - sleepsecs( 4 ); - - assert( conn.eval( "test" , "return db.totest.findOne().x" , res ) ); - - - } - - cout << "client test finished!" << endl; -} diff --git a/client/examples/first.cpp b/client/examples/first.cpp deleted file mode 100644 index ab5efb325f5..00000000000 --- a/client/examples/first.cpp +++ /dev/null @@ -1,86 +0,0 @@ -// first.cpp - -/* Copyright 2009 10gen Inc. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -/** - * this is a good first example of how to use mongo from c++ - */ - -#include <iostream> - -#include "client/dbclient.h" - -using namespace std; - -void insert( mongo::DBClientConnection & conn , const char * name , int num ) { - mongo::BSONObjBuilder obj; - obj.append( "name" , name ); - obj.append( "num" , num ); - conn.insert( "test.people" , obj.obj() ); -} - -int main( int argc, const char **argv ) { - - const char *port = "27017"; - if ( argc != 1 ) { - if ( argc != 3 ) - throw -12; - port = argv[ 2 ]; - } - - mongo::DBClientConnection conn; - string errmsg; - if ( ! conn.connect( string( "127.0.0.1:" ) + port , errmsg ) ) { - cout << "couldn't connect : " << errmsg << endl; - throw -11; - } - - { - // clean up old data from any previous tests - mongo::BSONObjBuilder query; - conn.remove( "test.people" , query.obj() ); - } - - insert( conn , "eliot" , 15 ); - insert( conn , "sara" , 23 ); - - { - mongo::BSONObjBuilder query; - auto_ptr<mongo::DBClientCursor> cursor = conn.query( "test.people" , query.obj() ); - cout << "using cursor" << endl; - while ( cursor->more() ) { - mongo::BSONObj obj = cursor->next(); - cout << "\t" << obj.jsonString() << endl; - } - - } - - { - mongo::BSONObjBuilder query; - query.append( "name" , "eliot" ); - mongo::BSONObj res = conn.findOne( "test.people" , query.obj() ); - cout << res.isEmpty() << "\t" << res.jsonString() << endl; - } - - { - mongo::BSONObjBuilder query; - query.append( "name" , "asd" ); - mongo::BSONObj res = conn.findOne( "test.people" , query.obj() ); - cout << res.isEmpty() << "\t" << res.jsonString() << endl; - } - - -} diff --git a/client/examples/httpClientTest.cpp b/client/examples/httpClientTest.cpp deleted file mode 100644 index 4055d4492d5..00000000000 --- a/client/examples/httpClientTest.cpp +++ /dev/null @@ -1,58 +0,0 @@ -// httpClientTest.cpp - -/* Copyright 2009 10gen Inc. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -#include <iostream> - -#include "client/dbclient.h" -#include "util/net/httpclient.h" - -using namespace mongo; - -void play( string url ) { - cout << "[" << url << "]" << endl; - - HttpClient c; - HttpClient::Result r; - MONGO_assert( c.get( url , &r ) == 200 ); - - HttpClient::Headers h = r.getHeaders(); - MONGO_assert( h["Content-Type"].find( "text/html" ) == 0 ); - - cout << "\tHeaders" << endl; - for ( HttpClient::Headers::iterator i = h.begin() ; i != h.end(); ++i ) { - cout << "\t\t" << i->first << "\t" << i->second << endl; - } - -} - -int main( int argc, const char **argv ) { - - int port = 27017; - if ( argc != 1 ) { - if ( argc != 3 ) - throw -12; - port = atoi( argv[ 2 ] ); - } - port += 1000; - - play( str::stream() << "http://localhost:" << port << "/" ); - -#ifdef MONGO_SSL - play( "https://www.10gen.com/" ); -#endif - -} diff --git a/client/examples/insert_demo.cpp b/client/examples/insert_demo.cpp deleted file mode 100644 index 14ac79ee1a0..00000000000 --- a/client/examples/insert_demo.cpp +++ /dev/null @@ -1,47 +0,0 @@ -/* - C++ client program which inserts documents in a MongoDB database. - - How to build and run: - - Using mongo_client_lib.cpp: - g++ -I .. -I ../.. insert_demo.cpp ../mongo_client_lib.cpp -lboost_thread-mt -lboost_filesystem - ./a.out -*/ - -#include <iostream> -#include "dbclient.h" // the mongo c++ driver - -using namespace std; -using namespace mongo; -using namespace bson; - -int main() { - try { - cout << "connecting to localhost..." << endl; - DBClientConnection c; - c.connect("localhost"); - cout << "connected ok" << endl; - - bo o = BSON( "hello" << "world" ); - - cout << "inserting..." << endl; - - time_t start = time(0); - for( unsigned i = 0; i < 1000000; i++ ) { - c.insert("test.foo", o); - } - - // wait until all operations applied - cout << "getlasterror returns: \"" << c.getLastError() << '"' << endl; - - time_t done = time(0); - time_t dt = done-start; - cout << dt << " seconds " << 1000000/dt << " per second" << endl; - } - catch(DBException& e) { - cout << "caught DBException " << e.toString() << endl; - return 1; - } - - return 0; -} diff --git a/client/examples/rs.cpp b/client/examples/rs.cpp deleted file mode 100644 index 3307d87b56b..00000000000 --- a/client/examples/rs.cpp +++ /dev/null @@ -1,118 +0,0 @@ -// rs.cpp - -/* Copyright 2009 10gen Inc. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -/** - * example of using replica sets from c++ - */ - -#include "client/dbclient.h" -#include <iostream> -#include <vector> - -using namespace mongo; -using namespace std; - -void workerThread( string collName , bool print , DBClientReplicaSet * conn ) { - - while ( true ) { - try { - conn->update( collName , BSONObj() , BSON( "$inc" << BSON( "x" << 1 ) ) , true ); - - BSONObj x = conn->findOne( collName , BSONObj() ); - - if ( print ) { - cout << x << endl; - } - - BSONObj a = conn->slaveConn().findOne( collName , BSONObj() , 0 , QueryOption_SlaveOk ); - BSONObj b = conn->findOne( collName , BSONObj() , 0 , QueryOption_SlaveOk ); - - if ( print ) { - cout << "\t A " << a << endl; - cout << "\t B " << b << endl; - } - } - catch ( std::exception& e ) { - cout << "ERROR: " << e.what() << endl; - } - sleepmillis( 10 ); - } -} - -int main( int argc , const char ** argv ) { - - unsigned nThreads = 1; - bool print = false; - bool testTimeout = false; - - for ( int i=1; i<argc; i++ ) { - if ( mongoutils::str::equals( "--threads" , argv[i] ) ) { - nThreads = atoi( argv[++i] ); - } - else if ( mongoutils::str::equals( "--print" , argv[i] ) ) { - print = true; - } - // Run a special mode to demonstrate the DBClientReplicaSet so_timeout option. - else if ( mongoutils::str::equals( "--testTimeout" , argv[i] ) ) { - testTimeout = true; - } - else { - cerr << "unknown option: " << argv[i] << endl; - return 1; - } - - } - - string errmsg; - ConnectionString cs = ConnectionString::parse( "foo/127.0.0.1" , errmsg ); - if ( ! cs.isValid() ) { - cout << "error parsing url: " << errmsg << endl; - return 1; - } - - DBClientReplicaSet * conn = dynamic_cast<DBClientReplicaSet*>(cs.connect( errmsg, testTimeout ? 10 : 0 )); - if ( ! conn ) { - cout << "error connecting: " << errmsg << endl; - return 2; - } - - string collName = "test.rs1"; - - conn->dropCollection( collName ); - - if ( testTimeout ) { - conn->insert( collName, BSONObj() ); - try { - conn->count( collName, BSON( "$where" << "sleep(40000)" ) ); - } catch( DBException& ) { - return 0; - } - cout << "expected socket exception" << endl; - return 1; - } - - vector<boost::shared_ptr<boost::thread> > threads; - for ( unsigned i=0; i<nThreads; i++ ) { - string errmsg; - threads.push_back( boost::shared_ptr<boost::thread>( new boost::thread( boost::bind( workerThread , collName , print , (DBClientReplicaSet*)cs.connect(errmsg) ) ) ) ); - } - - for ( unsigned i=0; i<threads.size(); i++ ) { - threads[i]->join(); - } - -} diff --git a/client/examples/second.cpp b/client/examples/second.cpp deleted file mode 100644 index 6cc2111580f..00000000000 --- a/client/examples/second.cpp +++ /dev/null @@ -1,56 +0,0 @@ -// second.cpp - -/* Copyright 2009 10gen Inc. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -#include <iostream> - -#include "client/dbclient.h" - -using namespace std; -using namespace mongo; - -int main( int argc, const char **argv ) { - - const char *port = "27017"; - if ( argc != 1 ) { - if ( argc != 3 ) - throw -12; - port = argv[ 2 ]; - } - - DBClientConnection conn; - string errmsg; - if ( ! conn.connect( string( "127.0.0.1:" ) + port , errmsg ) ) { - cout << "couldn't connect : " << errmsg << endl; - throw -11; - } - - const char * ns = "test.second"; - - conn.remove( ns , BSONObj() ); - - conn.insert( ns , BSON( "name" << "eliot" << "num" << 17 ) ); - conn.insert( ns , BSON( "name" << "sara" << "num" << 24 ) ); - - auto_ptr<DBClientCursor> cursor = conn.query( ns , BSONObj() ); - cout << "using cursor" << endl; - while ( cursor->more() ) { - BSONObj obj = cursor->next(); - cout << "\t" << obj.jsonString() << endl; - } - - conn.ensureIndex( ns , BSON( "name" << 1 << "num" << -1 ) ); -} diff --git a/client/examples/simple_client_demo.vcxproj b/client/examples/simple_client_demo.vcxproj deleted file mode 100755 index 4658a42900f..00000000000 --- a/client/examples/simple_client_demo.vcxproj +++ /dev/null @@ -1,92 +0,0 @@ -<?xml version="1.0" encoding="utf-8"?>
-<Project DefaultTargets="Build" ToolsVersion="4.0" xmlns="http://schemas.microsoft.com/developer/msbuild/2003">
- <ItemGroup Label="ProjectConfigurations">
- <ProjectConfiguration Include="Debug|Win32">
- <Configuration>Debug</Configuration>
- <Platform>Win32</Platform>
- </ProjectConfiguration>
- <ProjectConfiguration Include="Release|Win32">
- <Configuration>Release</Configuration>
- <Platform>Win32</Platform>
- </ProjectConfiguration>
- </ItemGroup>
- <PropertyGroup Label="Globals">
- <ProjectGuid>{89C30BC3-2874-4F2C-B4DA-EB04E9782236}</ProjectGuid>
- <Keyword>Win32Proj</Keyword>
- <RootNamespace>simple_client_demo</RootNamespace>
- </PropertyGroup>
- <Import Project="$(VCTargetsPath)\Microsoft.Cpp.Default.props" />
- <PropertyGroup Condition="'$(Configuration)|$(Platform)'=='Debug|Win32'" Label="Configuration">
- <ConfigurationType>Application</ConfigurationType>
- <UseDebugLibraries>true</UseDebugLibraries>
- <CharacterSet>Unicode</CharacterSet>
- </PropertyGroup>
- <PropertyGroup Condition="'$(Configuration)|$(Platform)'=='Release|Win32'" Label="Configuration">
- <ConfigurationType>Application</ConfigurationType>
- <UseDebugLibraries>false</UseDebugLibraries>
- <WholeProgramOptimization>true</WholeProgramOptimization>
- <CharacterSet>Unicode</CharacterSet>
- </PropertyGroup>
- <Import Project="$(VCTargetsPath)\Microsoft.Cpp.props" />
- <ImportGroup Label="ExtensionSettings">
- </ImportGroup>
- <ImportGroup Label="PropertySheets" Condition="'$(Configuration)|$(Platform)'=='Debug|Win32'">
- <Import Project="$(UserRootDir)\Microsoft.Cpp.$(Platform).user.props" Condition="exists('$(UserRootDir)\Microsoft.Cpp.$(Platform).user.props')" Label="LocalAppDataPlatform" />
- </ImportGroup>
- <ImportGroup Label="PropertySheets" Condition="'$(Configuration)|$(Platform)'=='Release|Win32'">
- <Import Project="$(UserRootDir)\Microsoft.Cpp.$(Platform).user.props" Condition="exists('$(UserRootDir)\Microsoft.Cpp.$(Platform).user.props')" Label="LocalAppDataPlatform" />
- </ImportGroup>
- <PropertyGroup Label="UserMacros" />
- <PropertyGroup Condition="'$(Configuration)|$(Platform)'=='Debug|Win32'">
- <LinkIncremental>true</LinkIncremental>
- <IncludePath>..\..;..\..\pcre-7.4;$(IncludePath)</IncludePath>
- <LibraryPath>\boost\lib\vs2010_32;$(LibraryPath)</LibraryPath>
- </PropertyGroup>
- <PropertyGroup Condition="'$(Configuration)|$(Platform)'=='Release|Win32'">
- <LinkIncremental>false</LinkIncremental>
- <IncludePath>..\..;..\..\pcre-7.4;$(IncludePath)</IncludePath>
- <LibraryPath>\boost\lib\vs2010_32;$(LibraryPath)</LibraryPath>
- </PropertyGroup>
- <ItemDefinitionGroup Condition="'$(Configuration)|$(Platform)'=='Debug|Win32'">
- <ClCompile>
- <PrecompiledHeader>
- </PrecompiledHeader>
- <WarningLevel>Level3</WarningLevel>
- <Optimization>Disabled</Optimization>
- <PreprocessorDefinitions> _CRT_SECURE_NO_WARNINGS;_DEBUG;_CONSOLE;%(PreprocessorDefinitions)</PreprocessorDefinitions>
- <AdditionalIncludeDirectories>c:\boost;\boost</AdditionalIncludeDirectories>
- </ClCompile>
- <Link>
- <SubSystem>Console</SubSystem>
- <GenerateDebugInformation>true</GenerateDebugInformation>
- <AdditionalDependencies>ws2_32.lib;kernel32.lib;user32.lib;gdi32.lib;winspool.lib;comdlg32.lib;advapi32.lib;shell32.lib;ole32.lib;oleaut32.lib;uuid.lib;odbc32.lib;odbccp32.lib;%(AdditionalDependencies)</AdditionalDependencies>
- </Link>
- </ItemDefinitionGroup>
- <ItemDefinitionGroup Condition="'$(Configuration)|$(Platform)'=='Release|Win32'">
- <ClCompile>
- <WarningLevel>Level3</WarningLevel>
- <PrecompiledHeader>
- </PrecompiledHeader>
- <Optimization>MaxSpeed</Optimization>
- <FunctionLevelLinking>true</FunctionLevelLinking>
- <IntrinsicFunctions>true</IntrinsicFunctions>
- <PreprocessorDefinitions> _CRT_SECURE_NO_WARNINGS;NDEBUG;_CONSOLE;%(PreprocessorDefinitions)</PreprocessorDefinitions>
- <RuntimeLibrary>MultiThreaded</RuntimeLibrary>
- <AdditionalIncludeDirectories>c:\boost;\boost</AdditionalIncludeDirectories>
- </ClCompile>
- <Link>
- <SubSystem>Console</SubSystem>
- <GenerateDebugInformation>true</GenerateDebugInformation>
- <EnableCOMDATFolding>true</EnableCOMDATFolding>
- <OptimizeReferences>true</OptimizeReferences>
- <AdditionalDependencies>ws2_32.lib;kernel32.lib;user32.lib;gdi32.lib;winspool.lib;comdlg32.lib;advapi32.lib;shell32.lib;ole32.lib;oleaut32.lib;uuid.lib;odbc32.lib;odbccp32.lib;%(AdditionalDependencies)</AdditionalDependencies>
- </Link>
- </ItemDefinitionGroup>
- <ItemGroup>
- <ClCompile Include="..\mongo_client_lib.cpp" />
- <ClCompile Include="..\simple_client_demo.cpp" />
- </ItemGroup>
- <Import Project="$(VCTargetsPath)\Microsoft.Cpp.targets" />
- <ImportGroup Label="ExtensionTargets">
- </ImportGroup>
-</Project>
\ No newline at end of file diff --git a/client/examples/simple_client_demo.vcxproj.filters b/client/examples/simple_client_demo.vcxproj.filters deleted file mode 100755 index d6580c32279..00000000000 --- a/client/examples/simple_client_demo.vcxproj.filters +++ /dev/null @@ -1,21 +0,0 @@ -<?xml version="1.0" encoding="utf-8"?>
-<Project ToolsVersion="4.0" xmlns="http://schemas.microsoft.com/developer/msbuild/2003">
- <ItemGroup>
- <Filter Include="Source Files">
- <UniqueIdentifier>{4FC737F1-C7A5-4376-A066-2A32D752A2FF}</UniqueIdentifier>
- <Extensions>cpp;c;cc;cxx;def;odl;idl;hpj;bat;asm;asmx</Extensions>
- </Filter>
- <Filter Include="Header Files">
- <UniqueIdentifier>{93995380-89BD-4b04-88EB-625FBE52EBFB}</UniqueIdentifier>
- <Extensions>h;hpp;hxx;hm;inl;inc;xsd</Extensions>
- </Filter>
- </ItemGroup>
- <ItemGroup>
- <ClCompile Include="..\simple_client_demo.cpp">
- <Filter>Source Files</Filter>
- </ClCompile>
- <ClCompile Include="..\mongo_client_lib.cpp">
- <Filter>Source Files</Filter>
- </ClCompile>
- </ItemGroup>
-</Project>
\ No newline at end of file diff --git a/client/examples/tail.cpp b/client/examples/tail.cpp deleted file mode 100644 index 90e62d279c1..00000000000 --- a/client/examples/tail.cpp +++ /dev/null @@ -1,46 +0,0 @@ -// tail.cpp - -/* Copyright 2009 10gen Inc. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -/* example of using a tailable cursor */ - -#include "../../client/dbclient.h" -#include "../../util/goodies.h" - -using namespace mongo; - -void tail(DBClientBase& conn, const char *ns) { - BSONElement lastId = minKey.firstElement(); - Query query = Query(); - - auto_ptr<DBClientCursor> c = - conn.query(ns, query, 0, 0, 0, QueryOption_CursorTailable); - - while( 1 ) { - if( !c->more() ) { - if( c->isDead() ) { - break; // we need to requery - } - - // all data (so far) exhausted, wait for more - sleepsecs(1); - continue; - } - BSONObj o = c->next(); - lastId = o["_id"]; - cout << o.toString() << endl; - } -} diff --git a/client/examples/tutorial.cpp b/client/examples/tutorial.cpp deleted file mode 100644 index 3cdf3593cd8..00000000000 --- a/client/examples/tutorial.cpp +++ /dev/null @@ -1,67 +0,0 @@ -//tutorial.cpp - -/* Copyright 2009 10gen Inc. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -#include <iostream> -#include "../../client/dbclient.h" - -// g++ tutorial.cpp -lmongoclient -lboost_thread -lboost_filesystem -o tutorial - -using namespace mongo; - -void printIfAge(DBClientConnection& c, int age) { - auto_ptr<DBClientCursor> cursor = c.query("tutorial.persons", QUERY( "age" << age ).sort("name") ); - while( cursor->more() ) { - BSONObj p = cursor->next(); - cout << p.getStringField("name") << endl; - } -} - -void run() { - DBClientConnection c; - c.connect("localhost"); //"192.168.58.1"); - cout << "connected ok" << endl; - BSONObj p = BSON( "name" << "Joe" << "age" << 33 ); - c.insert("tutorial.persons", p); - p = BSON( "name" << "Jane" << "age" << 40 ); - c.insert("tutorial.persons", p); - p = BSON( "name" << "Abe" << "age" << 33 ); - c.insert("tutorial.persons", p); - p = BSON( "name" << "Samantha" << "age" << 21 << "city" << "Los Angeles" << "state" << "CA" ); - c.insert("tutorial.persons", p); - - c.ensureIndex("tutorial.persons", fromjson("{age:1}")); - - cout << "count:" << c.count("tutorial.persons") << endl; - - auto_ptr<DBClientCursor> cursor = c.query("tutorial.persons", BSONObj()); - while( cursor->more() ) { - cout << cursor->next().toString() << endl; - } - - cout << "\nprintifage:\n"; - printIfAge(c, 33); -} - -int main() { - try { - run(); - } - catch( DBException &e ) { - cout << "caught " << e.what() << endl; - } - return 0; -} diff --git a/client/examples/whereExample.cpp b/client/examples/whereExample.cpp deleted file mode 100644 index 12b68d7add3..00000000000 --- a/client/examples/whereExample.cpp +++ /dev/null @@ -1,69 +0,0 @@ -// @file whereExample.cpp -// @see http://www.mongodb.org/display/DOCS/Server-side+Code+Execution - -/* Copyright 2009 10gen Inc. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -#include <iostream> - -#include "client/dbclient.h" - -using namespace std; -using namespace mongo; - -int main( int argc, const char **argv ) { - - const char *port = "27017"; - if ( argc != 1 ) { - if ( argc != 3 ) - throw -12; - port = argv[ 2 ]; - } - - DBClientConnection conn; - string errmsg; - if ( ! conn.connect( string( "127.0.0.1:" ) + port , errmsg ) ) { - cout << "couldn't connect : " << errmsg << endl; - throw -11; - } - - const char * ns = "test.where"; - - conn.remove( ns , BSONObj() ); - - conn.insert( ns , BSON( "name" << "eliot" << "num" << 17 ) ); - conn.insert( ns , BSON( "name" << "sara" << "num" << 24 ) ); - - auto_ptr<DBClientCursor> cursor = conn.query( ns , BSONObj() ); - - while ( cursor->more() ) { - BSONObj obj = cursor->next(); - cout << "\t" << obj.jsonString() << endl; - } - - cout << "now using $where" << endl; - - Query q = Query("{}").where("this.name == name" , BSON( "name" << "sara" )); - - cursor = conn.query( ns , q ); - - int num = 0; - while ( cursor->more() ) { - BSONObj obj = cursor->next(); - cout << "\t" << obj.jsonString() << endl; - num++; - } - MONGO_assert( num == 1 ); -} diff --git a/client/gridfs.cpp b/client/gridfs.cpp deleted file mode 100644 index 233724ae8e5..00000000000 --- a/client/gridfs.cpp +++ /dev/null @@ -1,245 +0,0 @@ -// gridfs.cpp - -/* Copyright 2009 10gen - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -#include "pch.h" -#include <fcntl.h> -#include <utility> - -#include "gridfs.h" -#include <boost/smart_ptr.hpp> - -#if defined(_WIN32) -#include <io.h> -#endif - -#ifndef MIN -#define MIN(a,b) ( (a) < (b) ? (a) : (b) ) -#endif - -namespace mongo { - - const unsigned DEFAULT_CHUNK_SIZE = 256 * 1024; - - GridFSChunk::GridFSChunk( BSONObj o ) { - _data = o; - } - - GridFSChunk::GridFSChunk( BSONObj fileObject , int chunkNumber , const char * data , int len ) { - BSONObjBuilder b; - b.appendAs( fileObject["_id"] , "files_id" ); - b.append( "n" , chunkNumber ); - b.appendBinData( "data" , len, BinDataGeneral, data ); - _data = b.obj(); - } - - - GridFS::GridFS( DBClientBase& client , const string& dbName , const string& prefix ) : _client( client ) , _dbName( dbName ) , _prefix( prefix ) { - _filesNS = dbName + "." + prefix + ".files"; - _chunksNS = dbName + "." + prefix + ".chunks"; - _chunkSize = DEFAULT_CHUNK_SIZE; - - client.ensureIndex( _filesNS , BSON( "filename" << 1 ) ); - client.ensureIndex( _chunksNS , BSON( "files_id" << 1 << "n" << 1 ) ); - } - - GridFS::~GridFS() { - - } - - void GridFS::setChunkSize(unsigned int size) { - massert( 13296 , "invalid chunk size is specified", (size == 0)); - _chunkSize = size; - } - - BSONObj GridFS::storeFile( const char* data , size_t length , const string& remoteName , const string& contentType) { - char const * const end = data + length; - - OID id; - id.init(); - BSONObj idObj = BSON("_id" << id); - - int chunkNumber = 0; - while (data < end) { - int chunkLen = MIN(_chunkSize, (unsigned)(end-data)); - GridFSChunk c(idObj, chunkNumber, data, chunkLen); - _client.insert( _chunksNS.c_str() , c._data ); - - chunkNumber++; - data += chunkLen; - } - - return insertFile(remoteName, id, length, contentType); - } - - - BSONObj GridFS::storeFile( const string& fileName , const string& remoteName , const string& contentType) { - uassert( 10012 , "file doesn't exist" , fileName == "-" || boost::filesystem::exists( fileName ) ); - - FILE* fd; - if (fileName == "-") - fd = stdin; - else - fd = fopen( fileName.c_str() , "rb" ); - uassert( 10013 , "error opening file", fd); - - OID id; - id.init(); - BSONObj idObj = BSON("_id" << id); - - int chunkNumber = 0; - gridfs_offset length = 0; - while (!feof(fd)) { - //boost::scoped_array<char>buf (new char[_chunkSize+1]); - char * buf = new char[_chunkSize+1]; - char* bufPos = buf;//.get(); - unsigned int chunkLen = 0; // how much in the chunk now - while(chunkLen != _chunkSize && !feof(fd)) { - int readLen = fread(bufPos, 1, _chunkSize - chunkLen, fd); - chunkLen += readLen; - bufPos += readLen; - - assert(chunkLen <= _chunkSize); - } - - GridFSChunk c(idObj, chunkNumber, buf, chunkLen); - _client.insert( _chunksNS.c_str() , c._data ); - - length += chunkLen; - chunkNumber++; - delete[] buf; - } - - if (fd != stdin) - fclose( fd ); - - return insertFile((remoteName.empty() ? fileName : remoteName), id, length, contentType); - } - - BSONObj GridFS::insertFile(const string& name, const OID& id, gridfs_offset length, const string& contentType) { - - BSONObj res; - if ( ! _client.runCommand( _dbName.c_str() , BSON( "filemd5" << id << "root" << _prefix ) , res ) ) - throw UserException( 9008 , "filemd5 failed" ); - - BSONObjBuilder file; - file << "_id" << id - << "filename" << name - << "chunkSize" << _chunkSize - << "uploadDate" << DATENOW - << "md5" << res["md5"] - ; - - if (length < 1024*1024*1024) { // 2^30 - file << "length" << (int) length; - } - else { - file << "length" << (long long) length; - } - - if (!contentType.empty()) - file << "contentType" << contentType; - - BSONObj ret = file.obj(); - _client.insert(_filesNS.c_str(), ret); - - return ret; - } - - void GridFS::removeFile( const string& fileName ) { - auto_ptr<DBClientCursor> files = _client.query( _filesNS , BSON( "filename" << fileName ) ); - while (files->more()) { - BSONObj file = files->next(); - BSONElement id = file["_id"]; - _client.remove( _filesNS.c_str() , BSON( "_id" << id ) ); - _client.remove( _chunksNS.c_str() , BSON( "files_id" << id ) ); - } - } - - GridFile::GridFile( GridFS * grid , BSONObj obj ) { - _grid = grid; - _obj = obj; - } - - GridFile GridFS::findFile( const string& fileName ) { - return findFile( BSON( "filename" << fileName ) ); - }; - - GridFile GridFS::findFile( BSONObj query ) { - query = BSON("query" << query << "orderby" << BSON("uploadDate" << -1)); - return GridFile( this , _client.findOne( _filesNS.c_str() , query ) ); - } - - auto_ptr<DBClientCursor> GridFS::list() { - return _client.query( _filesNS.c_str() , BSONObj() ); - } - - auto_ptr<DBClientCursor> GridFS::list( BSONObj o ) { - return _client.query( _filesNS.c_str() , o ); - } - - BSONObj GridFile::getMetadata() { - BSONElement meta_element = _obj["metadata"]; - if( meta_element.eoo() ) { - return BSONObj(); - } - - return meta_element.embeddedObject(); - } - - GridFSChunk GridFile::getChunk( int n ) { - _exists(); - BSONObjBuilder b; - b.appendAs( _obj["_id"] , "files_id" ); - b.append( "n" , n ); - - BSONObj o = _grid->_client.findOne( _grid->_chunksNS.c_str() , b.obj() ); - uassert( 10014 , "chunk is empty!" , ! o.isEmpty() ); - return GridFSChunk(o); - } - - gridfs_offset GridFile::write( ostream & out ) { - _exists(); - - const int num = getNumChunks(); - - for ( int i=0; i<num; i++ ) { - GridFSChunk c = getChunk( i ); - - int len; - const char * data = c.data( len ); - out.write( data , len ); - } - - return getContentLength(); - } - - gridfs_offset GridFile::write( const string& where ) { - if (where == "-") { - return write( cout ); - } - else { - ofstream out(where.c_str() , ios::out | ios::binary ); - uassert(13325, "couldn't open file: " + where, out.is_open() ); - return write( out ); - } - } - - void GridFile::_exists() { - uassert( 10015 , "doesn't exists" , exists() ); - } - -} diff --git a/client/gridfs.h b/client/gridfs.h deleted file mode 100644 index b52cf75117a..00000000000 --- a/client/gridfs.h +++ /dev/null @@ -1,205 +0,0 @@ -/** @file gridfs.h */ - -/* Copyright 2009 10gen Inc. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -#pragma once - -#include "dbclient.h" -#include "redef_macros.h" - -namespace mongo { - - typedef unsigned long long gridfs_offset; - - class GridFS; - class GridFile; - - class GridFSChunk { - public: - GridFSChunk( BSONObj data ); - GridFSChunk( BSONObj fileId , int chunkNumber , const char * data , int len ); - - int len() { - int len; - _data["data"].binDataClean( len ); - return len; - } - - const char * data( int & len ) { - return _data["data"].binDataClean( len ); - } - - private: - BSONObj _data; - friend class GridFS; - }; - - - /** - GridFS is for storing large file-style objects in MongoDB. - @see http://www.mongodb.org/display/DOCS/GridFS+Specification - */ - class GridFS { - public: - /** - * @param client - db connection - * @param dbName - root database name - * @param prefix - if you want your data somewhere besides <dbname>.fs - */ - GridFS( DBClientBase& client , const string& dbName , const string& prefix="fs" ); - ~GridFS(); - - /** - * @param - */ - void setChunkSize(unsigned int size); - - /** - * puts the file reference by fileName into the db - * @param fileName local filename relative to process - * @param remoteName optional filename to use for file stored in GridFS - * (default is to use fileName parameter) - * @param contentType optional MIME type for this object. - * (default is to omit) - * @return the file object - */ - BSONObj storeFile( const string& fileName , const string& remoteName="" , const string& contentType=""); - - /** - * puts the file represented by data into the db - * @param data pointer to buffer to store in GridFS - * @param length length of buffer - * @param remoteName optional filename to use for file stored in GridFS - * (default is to use fileName parameter) - * @param contentType optional MIME type for this object. - * (default is to omit) - * @return the file object - */ - BSONObj storeFile( const char* data , size_t length , const string& remoteName , const string& contentType=""); - - /** - * removes file referenced by fileName from the db - * @param fileName filename (in GridFS) of the file to remove - * @return the file object - */ - void removeFile( const string& fileName ); - - /** - * returns a file object matching the query - */ - GridFile findFile( BSONObj query ); - - /** - * equiv to findFile( { filename : filename } ) - */ - GridFile findFile( const string& fileName ); - - /** - * convenience method to get all the files - */ - auto_ptr<DBClientCursor> list(); - - /** - * convenience method to get all the files with a filter - */ - auto_ptr<DBClientCursor> list( BSONObj query ); - - private: - DBClientBase& _client; - string _dbName; - string _prefix; - string _filesNS; - string _chunksNS; - unsigned int _chunkSize; - - // insert fileobject. All chunks must be in DB. - BSONObj insertFile(const string& name, const OID& id, gridfs_offset length, const string& contentType); - - friend class GridFile; - }; - - /** - wrapper for a file stored in the Mongo database - */ - class GridFile { - public: - /** - * @return whether or not this file exists - * findFile will always return a GriFile, so need to check this - */ - bool exists() { - return ! _obj.isEmpty(); - } - - string getFilename() { - return _obj["filename"].str(); - } - - int getChunkSize() { - return (int)(_obj["chunkSize"].number()); - } - - gridfs_offset getContentLength() { - return (gridfs_offset)(_obj["length"].number()); - } - - string getContentType() { - return _obj["contentType"].valuestr(); - } - - Date_t getUploadDate() { - return _obj["uploadDate"].date(); - } - - string getMD5() { - return _obj["md5"].str(); - } - - BSONElement getFileField( const string& name ) { - return _obj[name]; - } - - BSONObj getMetadata(); - - int getNumChunks() { - return (int) ceil( (double)getContentLength() / (double)getChunkSize() ); - } - - GridFSChunk getChunk( int n ); - - /** - write the file to the output stream - */ - gridfs_offset write( ostream & out ); - - /** - write the file to this filename - */ - gridfs_offset write( const string& where ); - - private: - GridFile( GridFS * grid , BSONObj obj ); - - void _exists(); - - GridFS * _grid; - BSONObj _obj; - - friend class GridFS; - }; -} - -#include "undef_macros.h" diff --git a/client/model.cpp b/client/model.cpp deleted file mode 100644 index bd10a3c5528..00000000000 --- a/client/model.cpp +++ /dev/null @@ -1,138 +0,0 @@ -// model.cpp - -/* Copyright 2009 10gen - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -#include "pch.h" -#include "model.h" -#include "connpool.h" - -namespace mongo { - - bool Model::load(BSONObj& query) { - ScopedDbConnection conn( modelServer() ); - - BSONObj b = conn->findOne(getNS(), query); - conn.done(); - - if ( b.isEmpty() ) - return false; - - unserialize(b); - _id = b["_id"].wrap().getOwned(); - return true; - } - - void Model::remove( bool safe ) { - uassert( 10016 , "_id isn't set - needed for remove()" , _id["_id"].type() ); - - ScopedDbConnection conn( modelServer() ); - conn->remove( getNS() , _id ); - - string errmsg = ""; - if ( safe ) - errmsg = conn->getLastError(); - - conn.done(); - - if ( safe && errmsg.size() ) - throw UserException( 9002 , (string)"error on Model::remove: " + errmsg ); - } - - void Model::save( bool safe ) { - ScopedDbConnection conn( modelServer() ); - - BSONObjBuilder b; - serialize( b ); - - BSONElement myId; - { - BSONObjIterator i = b.iterator(); - while ( i.more() ) { - BSONElement e = i.next(); - if ( strcmp( e.fieldName() , "_id" ) == 0 ) { - myId = e; - break; - } - } - } - - if ( myId.type() ) { - if ( _id.isEmpty() ) { - _id = myId.wrap(); - } - else if ( myId.woCompare( _id.firstElement() ) ) { - stringstream ss; - ss << "_id from serialize and stored differ: "; - ss << '[' << myId << "] != "; - ss << '[' << _id.firstElement() << ']'; - throw UserException( 13121 , ss.str() ); - } - } - - if ( _id.isEmpty() ) { - OID oid; - oid.init(); - b.appendOID( "_id" , &oid ); - - BSONObj o = b.obj(); - conn->insert( getNS() , o ); - _id = o["_id"].wrap().getOwned(); - - log(4) << "inserted new model " << getNS() << " " << o << endl; - } - else { - if ( myId.eoo() ) { - myId = _id["_id"]; - b.append( myId ); - } - - assert( ! myId.eoo() ); - - BSONObjBuilder qb; - qb.append( myId ); - - BSONObj q = qb.obj(); - BSONObj o = b.obj(); - - log(4) << "updated model" << getNS() << " " << q << " " << o << endl; - - conn->update( getNS() , q , o , true ); - - } - - string errmsg = ""; - if ( safe ) - errmsg = conn->getLastError(); - - conn.done(); - - if ( safe && errmsg.size() ) - throw UserException( 9003 , (string)"error on Model::save: " + errmsg ); - } - - BSONObj Model::toObject() { - BSONObjBuilder b; - serialize( b ); - return b.obj(); - } - - void Model::append( const char * name , BSONObjBuilder& b ) { - BSONObjBuilder bb( b.subobjStart( name ) ); - serialize( bb ); - bb.done(); - } - -} // namespace mongo diff --git a/client/model.h b/client/model.h deleted file mode 100644 index 7dd31434f49..00000000000 --- a/client/model.h +++ /dev/null @@ -1,62 +0,0 @@ -/** @file model.h */ - -/* Copyright 2009 10gen - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -#pragma once - -#include "dbclient.h" -#include "redef_macros.h" - -namespace mongo { - - /** Model is a base class for defining objects which are serializable to the Mongo - database via the database driver. - - Definition - Your serializable class should inherit from Model and implement the abstract methods - below. - - Loading - To load, first construct an (empty) object. Then call load(). Do not load an object - more than once. - */ - class Model { - public: - Model() { } - virtual ~Model() { } - - virtual const char * getNS() = 0; - virtual void serialize(BSONObjBuilder& to) = 0; - virtual void unserialize(const BSONObj& from) = 0; - virtual BSONObj toObject(); - virtual void append( const char * name , BSONObjBuilder& b ); - - virtual string modelServer() = 0; - - /** Load a single object. - @return true if successful. - */ - virtual bool load(BSONObj& query); - virtual void save( bool safe=false ); - virtual void remove( bool safe=false ); - - protected: - BSONObj _id; - }; - -} // namespace mongo - -#include "undef_macros.h" diff --git a/client/mongo_client_lib.cpp b/client/mongo_client_lib.cpp deleted file mode 100644 index 8100d719e81..00000000000 --- a/client/mongo_client_lib.cpp +++ /dev/null @@ -1,89 +0,0 @@ -/* @file client_lib.cpp - - MongoDB C++ Driver - - Normally one includes dbclient.h, and links against libmongoclient.a, when connecting to MongoDB - from C++. However, if you have a situation where the pre-built library does not work, you can use - this file instead to build all the necessary symbols. To do so, include mongo_client_lib.cpp in your - project. - - GCC - --- - For example, to build and run simple_client_demo.cpp with GCC and run it: - - g++ -I .. simple_client_demo.cpp mongo_client_lib.cpp -lboost_thread-mt -lboost_filesystem - ./a.out - - Visual Studio (2010 tested) - --------------------------- - First, see client/examples/simple_client_demo.vcxproj. - - Be sure to include your boost include directory in your project as an Additional Include Directory. - - Define _CRT_SECURE_NO_WARNINGS to avoid warnings on use of strncpy and such by the MongoDB client code. - - Include the boost libraries directory. - - Linker.Input.Additional Dependencies - add ws2_32.lib for the Winsock library. -*/ - -/* Copyright 2009 10gen Inc. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -#if defined(_WIN32) -// C4800 forcing value to bool 'true' or 'false' (performance warning) -#pragma warning( disable : 4800 ) -#endif - -#include "../util/md5main.cpp" - -#define MONGO_EXPOSE_MACROS -#include "../pch.h" - -#include "../util/assert_util.cpp" -#include "../util/net/message.cpp" -#include "../util/util.cpp" -#include "../util/background.cpp" -#include "../util/base64.cpp" -#include "../util/net/sock.cpp" -#include "../util/log.cpp" -#include "../util/password.cpp" -#include "../util/net/message_port.cpp" - -#include "../util/concurrency/thread_pool.cpp" -#include "../util/concurrency/vars.cpp" -#include "../util/concurrency/task.cpp" -#include "../util/concurrency/spin_lock.cpp" - -#include "connpool.cpp" -#include "syncclusterconnection.cpp" -#include "dbclient.cpp" -#include "clientOnly.cpp" -#include "gridfs.cpp" -#include "dbclientcursor.cpp" - -#include "../util/text.cpp" -#include "dbclient_rs.cpp" -#include "../bson/oid.cpp" - -#include "../db/lasterror.cpp" -#include "../db/json.cpp" -#include "../db/jsobj.cpp" -//#include "../db/common.cpp" -#include "../db/nonce.cpp" -#include "../db/commands.cpp" - -#include "../pch.cpp" - -extern "C" { -#include "../util/md5.c" -} - diff --git a/client/parallel.cpp b/client/parallel.cpp deleted file mode 100644 index df2f3fcfc5b..00000000000 --- a/client/parallel.cpp +++ /dev/null @@ -1,804 +0,0 @@ -// parallel.cpp -/* - * Copyright 2010 10gen Inc. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - - -#include "pch.h" -#include "parallel.h" -#include "connpool.h" -#include "../db/queryutil.h" -#include "../db/dbmessage.h" -#include "../s/util.h" -#include "../s/shard.h" - -namespace mongo { - - // -------- ClusteredCursor ----------- - - ClusteredCursor::ClusteredCursor( QueryMessage& q ) { - _ns = q.ns; - _query = q.query.copy(); - _options = q.queryOptions; - _fields = q.fields.copy(); - _batchSize = q.ntoreturn; - if ( _batchSize == 1 ) - _batchSize = 2; - - _done = false; - _didInit = false; - } - - ClusteredCursor::ClusteredCursor( const string& ns , const BSONObj& q , int options , const BSONObj& fields ) { - _ns = ns; - _query = q.getOwned(); - _options = options; - _fields = fields.getOwned(); - _batchSize = 0; - - _done = false; - _didInit = false; - } - - ClusteredCursor::~ClusteredCursor() { - _done = true; // just in case - } - - void ClusteredCursor::init() { - if ( _didInit ) - return; - _didInit = true; - _init(); - } - - void ClusteredCursor::_checkCursor( DBClientCursor * cursor ) { - assert( cursor ); - - if ( cursor->hasResultFlag( ResultFlag_ShardConfigStale ) ) { - throw StaleConfigException( _ns , "ClusteredCursor::_checkCursor" ); - } - - if ( cursor->hasResultFlag( ResultFlag_ErrSet ) ) { - BSONObj o = cursor->next(); - throw UserException( o["code"].numberInt() , o["$err"].String() ); - } - } - - auto_ptr<DBClientCursor> ClusteredCursor::query( const string& server , int num , BSONObj extra , int skipLeft , bool lazy ) { - uassert( 10017 , "cursor already done" , ! _done ); - assert( _didInit ); - - BSONObj q = _query; - if ( ! extra.isEmpty() ) { - q = concatQuery( q , extra ); - } - - try { - ShardConnection conn( server , _ns ); - - if ( conn.setVersion() ) { - conn.done(); - throw StaleConfigException( _ns , "ClusteredCursor::query" , true ); - } - - LOG(5) << "ClusteredCursor::query (" << type() << ") server:" << server - << " ns:" << _ns << " query:" << q << " num:" << num - << " _fields:" << _fields << " options: " << _options << endl; - - auto_ptr<DBClientCursor> cursor = - conn->query( _ns , q , num , 0 , ( _fields.isEmpty() ? 0 : &_fields ) , _options , _batchSize == 0 ? 0 : _batchSize + skipLeft ); - - if ( ! cursor.get() && _options & QueryOption_PartialResults ) { - _done = true; - conn.done(); - return cursor; - } - - massert( 13633 , str::stream() << "error querying server: " << server , cursor.get() ); - - cursor->attach( &conn ); // this calls done on conn - assert( ! conn.ok() ); - _checkCursor( cursor.get() ); - return cursor; - } - catch ( SocketException& e ) { - if ( ! ( _options & QueryOption_PartialResults ) ) - throw e; - _done = true; - return auto_ptr<DBClientCursor>(); - } - } - - BSONObj ClusteredCursor::explain( const string& server , BSONObj extra ) { - BSONObj q = _query; - if ( ! extra.isEmpty() ) { - q = concatQuery( q , extra ); - } - - BSONObj o; - - ShardConnection conn( server , _ns ); - auto_ptr<DBClientCursor> cursor = conn->query( _ns , Query( q ).explain() , abs( _batchSize ) * -1 , 0 , _fields.isEmpty() ? 0 : &_fields ); - if ( cursor.get() && cursor->more() ) - o = cursor->next().getOwned(); - conn.done(); - return o; - } - - BSONObj ClusteredCursor::concatQuery( const BSONObj& query , const BSONObj& extraFilter ) { - if ( ! query.hasField( "query" ) ) - return _concatFilter( query , extraFilter ); - - BSONObjBuilder b; - BSONObjIterator i( query ); - while ( i.more() ) { - BSONElement e = i.next(); - - if ( strcmp( e.fieldName() , "query" ) ) { - b.append( e ); - continue; - } - - b.append( "query" , _concatFilter( e.embeddedObjectUserCheck() , extraFilter ) ); - } - return b.obj(); - } - - BSONObj ClusteredCursor::_concatFilter( const BSONObj& filter , const BSONObj& extra ) { - BSONObjBuilder b; - b.appendElements( filter ); - b.appendElements( extra ); - return b.obj(); - // TODO: should do some simplification here if possibl ideally - } - - BSONObj ClusteredCursor::explain() { - // Note: by default we filter out allPlans and oldPlan in the shell's - // explain() function. If you add any recursive structures, make sure to - // edit the JS to make sure everything gets filtered. - - BSONObjBuilder b; - b.append( "clusteredType" , type() ); - - long long millis = 0; - double numExplains = 0; - - map<string,long long> counters; - - map<string,list<BSONObj> > out; - { - _explain( out ); - - BSONObjBuilder x( b.subobjStart( "shards" ) ); - for ( map<string,list<BSONObj> >::iterator i=out.begin(); i!=out.end(); ++i ) { - string shard = i->first; - list<BSONObj> l = i->second; - BSONArrayBuilder y( x.subarrayStart( shard ) ); - for ( list<BSONObj>::iterator j=l.begin(); j!=l.end(); ++j ) { - BSONObj temp = *j; - y.append( temp ); - - BSONObjIterator k( temp ); - while ( k.more() ) { - BSONElement z = k.next(); - if ( z.fieldName()[0] != 'n' ) - continue; - long long& c = counters[z.fieldName()]; - c += z.numberLong(); - } - - millis += temp["millis"].numberLong(); - numExplains++; - } - y.done(); - } - x.done(); - } - - for ( map<string,long long>::iterator i=counters.begin(); i!=counters.end(); ++i ) - b.appendNumber( i->first , i->second ); - - b.appendNumber( "millisTotal" , millis ); - b.append( "millisAvg" , (int)((double)millis / numExplains ) ); - b.append( "numQueries" , (int)numExplains ); - b.append( "numShards" , (int)out.size() ); - - return b.obj(); - } - - // -------- FilteringClientCursor ----------- - FilteringClientCursor::FilteringClientCursor( const BSONObj filter ) - : _matcher( filter ) , _done( true ) { - } - - FilteringClientCursor::FilteringClientCursor( auto_ptr<DBClientCursor> cursor , const BSONObj filter ) - : _matcher( filter ) , _cursor( cursor ) , _done( cursor.get() == 0 ) { - } - - FilteringClientCursor::FilteringClientCursor( DBClientCursor* cursor , const BSONObj filter ) - : _matcher( filter ) , _cursor( cursor ) , _done( cursor == 0 ) { - } - - - FilteringClientCursor::~FilteringClientCursor() { - } - - void FilteringClientCursor::reset( auto_ptr<DBClientCursor> cursor ) { - _cursor = cursor; - _next = BSONObj(); - _done = _cursor.get() == 0; - } - - void FilteringClientCursor::reset( DBClientCursor* cursor ) { - _cursor.reset( cursor ); - _next = BSONObj(); - _done = cursor == 0; - } - - - bool FilteringClientCursor::more() { - if ( ! _next.isEmpty() ) - return true; - - if ( _done ) - return false; - - _advance(); - return ! _next.isEmpty(); - } - - BSONObj FilteringClientCursor::next() { - assert( ! _next.isEmpty() ); - assert( ! _done ); - - BSONObj ret = _next; - _next = BSONObj(); - _advance(); - return ret; - } - - BSONObj FilteringClientCursor::peek() { - if ( _next.isEmpty() ) - _advance(); - return _next; - } - - void FilteringClientCursor::_advance() { - assert( _next.isEmpty() ); - if ( ! _cursor.get() || _done ) - return; - - while ( _cursor->more() ) { - _next = _cursor->next(); - if ( _matcher.matches( _next ) ) { - if ( ! _cursor->moreInCurrentBatch() ) - _next = _next.getOwned(); - return; - } - _next = BSONObj(); - } - _done = true; - } - - // -------- SerialServerClusteredCursor ----------- - - SerialServerClusteredCursor::SerialServerClusteredCursor( const set<ServerAndQuery>& servers , QueryMessage& q , int sortOrder) : ClusteredCursor( q ) { - for ( set<ServerAndQuery>::const_iterator i = servers.begin(); i!=servers.end(); i++ ) - _servers.push_back( *i ); - - if ( sortOrder > 0 ) - sort( _servers.begin() , _servers.end() ); - else if ( sortOrder < 0 ) - sort( _servers.rbegin() , _servers.rend() ); - - _serverIndex = 0; - - _needToSkip = q.ntoskip; - } - - bool SerialServerClusteredCursor::more() { - - // TODO: optimize this by sending on first query and then back counting - // tricky in case where 1st server doesn't have any after - // need it to send n skipped - while ( _needToSkip > 0 && _current.more() ) { - _current.next(); - _needToSkip--; - } - - if ( _current.more() ) - return true; - - if ( _serverIndex >= _servers.size() ) { - return false; - } - - ServerAndQuery& sq = _servers[_serverIndex++]; - - _current.reset( query( sq._server , 0 , sq._extra ) ); - return more(); - } - - BSONObj SerialServerClusteredCursor::next() { - uassert( 10018 , "no more items" , more() ); - return _current.next(); - } - - void SerialServerClusteredCursor::_explain( map< string,list<BSONObj> >& out ) { - for ( unsigned i=0; i<_servers.size(); i++ ) { - ServerAndQuery& sq = _servers[i]; - list<BSONObj> & l = out[sq._server]; - l.push_back( explain( sq._server , sq._extra ) ); - } - } - - // -------- ParallelSortClusteredCursor ----------- - - ParallelSortClusteredCursor::ParallelSortClusteredCursor( const set<ServerAndQuery>& servers , QueryMessage& q , - const BSONObj& sortKey ) - : ClusteredCursor( q ) , _servers( servers ) { - _sortKey = sortKey.getOwned(); - _needToSkip = q.ntoskip; - _finishCons(); - } - - ParallelSortClusteredCursor::ParallelSortClusteredCursor( const set<ServerAndQuery>& servers , const string& ns , - const Query& q , - int options , const BSONObj& fields ) - : ClusteredCursor( ns , q.obj , options , fields ) , _servers( servers ) { - _sortKey = q.getSort().copy(); - _needToSkip = 0; - _finishCons(); - } - - void ParallelSortClusteredCursor::_finishCons() { - _numServers = _servers.size(); - _lastFrom = 0; - _cursors = 0; - - if ( ! _sortKey.isEmpty() && ! _fields.isEmpty() ) { - // we need to make sure the sort key is in the projection - - set<string> sortKeyFields; - _sortKey.getFieldNames(sortKeyFields); - - BSONObjBuilder b; - bool isNegative = false; - { - BSONObjIterator i( _fields ); - while ( i.more() ) { - BSONElement e = i.next(); - b.append( e ); - - string fieldName = e.fieldName(); - - // exact field - bool found = sortKeyFields.erase(fieldName); - - // subfields - set<string>::const_iterator begin = sortKeyFields.lower_bound(fieldName + ".\x00"); - set<string>::const_iterator end = sortKeyFields.lower_bound(fieldName + ".\xFF"); - sortKeyFields.erase(begin, end); - - if ( ! e.trueValue() ) { - uassert( 13431 , "have to have sort key in projection and removing it" , !found && begin == end ); - } - else if (!e.isABSONObj()) { - isNegative = true; - } - } - } - - if (isNegative) { - for (set<string>::const_iterator it(sortKeyFields.begin()), end(sortKeyFields.end()); it != end; ++it) { - b.append(*it, 1); - } - } - - _fields = b.obj(); - } - } - - // TODO: Merge with futures API? We do a lot of error checking here that would be useful elsewhere. - void ParallelSortClusteredCursor::_init() { - - // log() << "Starting parallel search..." << endl; - - // make sure we're not already initialized - assert( ! _cursors ); - _cursors = new FilteringClientCursor[_numServers]; - - bool returnPartial = ( _options & QueryOption_PartialResults ); - - vector<ServerAndQuery> queries( _servers.begin(), _servers.end() ); - set<int> retryQueries; - int finishedQueries = 0; - - vector< shared_ptr<ShardConnection> > conns; - vector<string> servers; - - // Since we may get all sorts of errors, record them all as they come and throw them later if necessary - vector<string> staleConfigExs; - vector<string> socketExs; - vector<string> otherExs; - bool allConfigStale = false; - - int retries = -1; - - // Loop through all the queries until we've finished or gotten a socket exception on all of them - // We break early for non-socket exceptions, and socket exceptions if we aren't returning partial results - do { - retries++; - - bool firstPass = retryQueries.size() == 0; - - if( ! firstPass ){ - log() << "retrying " << ( returnPartial ? "(partial) " : "" ) << "parallel connection to "; - for( set<int>::iterator it = retryQueries.begin(); it != retryQueries.end(); ++it ){ - log() << queries[*it]._server << ", "; - } - log() << finishedQueries << " finished queries." << endl; - } - - size_t num = 0; - for ( vector<ServerAndQuery>::iterator it = queries.begin(); it != queries.end(); ++it ) { - size_t i = num++; - - const ServerAndQuery& sq = *it; - - // If we're not retrying this cursor on later passes, continue - if( ! firstPass && retryQueries.find( i ) == retryQueries.end() ) continue; - - // log() << "Querying " << _query << " from " << _ns << " for " << sq._server << endl; - - BSONObj q = _query; - if ( ! sq._extra.isEmpty() ) { - q = concatQuery( q , sq._extra ); - } - - string errLoc = " @ " + sq._server; - - if( firstPass ){ - - // This may be the first time connecting to this shard, if so we can get an error here - try { - conns.push_back( shared_ptr<ShardConnection>( new ShardConnection( sq._server , _ns ) ) ); - } - catch( std::exception& e ){ - socketExs.push_back( e.what() + errLoc ); - if( ! returnPartial ){ - num--; - break; - } - conns.push_back( shared_ptr<ShardConnection>() ); - continue; - } - - servers.push_back( sq._server ); - } - - if ( conns[i]->setVersion() ) { - conns[i]->done(); - staleConfigExs.push_back( (string)"stale config detected for " + StaleConfigException( _ns , "ParallelCursor::_init" , true ).what() + errLoc ); - break; - } - - LOG(5) << "ParallelSortClusteredCursor::init server:" << sq._server << " ns:" << _ns - << " query:" << q << " _fields:" << _fields << " options: " << _options << endl; - - if( ! _cursors[i].raw() ) - _cursors[i].reset( new DBClientCursor( conns[i]->get() , _ns , q , - 0 , // nToReturn - 0 , // nToSkip - _fields.isEmpty() ? 0 : &_fields , // fieldsToReturn - _options , - // NtoReturn is weird. - // If zero, it means use default size, so we do that for all cursors - // If positive, it's the batch size (we don't want this cursor limiting results), tha - // done at a higher level - // If negative, it's the batch size, but we don't create a cursor - so we don't want - // to create a child cursor either. - // Either way, if non-zero, we want to pull back the batch size + the skip amount as - // quickly as possible. Potentially, for a cursor on a single shard or if we keep be - // chunks, we can actually add the skip value into the cursor and/or make some assump - // return value size ( (batch size + skip amount) / num_servers ). - _batchSize == 0 ? 0 : - ( _batchSize > 0 ? _batchSize + _needToSkip : - _batchSize - _needToSkip ) // batchSize - ) ); - - try{ - _cursors[i].raw()->initLazy( ! firstPass ); - } - catch( SocketException& e ){ - socketExs.push_back( e.what() + errLoc ); - _cursors[i].reset( NULL ); - conns[i]->done(); - if( ! returnPartial ) break; - } - catch( std::exception& e){ - otherExs.push_back( e.what() + errLoc ); - _cursors[i].reset( NULL ); - conns[i]->done(); - break; - } - - } - - // Go through all the potentially started cursors and finish initializing them or log any errors and - // potentially retry - // TODO: Better error classification would make this easier, errors are indicated in all sorts of ways - // here that we need to trap. - for ( size_t i = 0; i < num; i++ ) { - - // log() << "Finishing query for " << cons[i].get()->getHost() << endl; - string errLoc = " @ " + queries[i]._server; - - if( ! _cursors[i].raw() || ( ! firstPass && retryQueries.find( i ) == retryQueries.end() ) ){ - if( conns[i] ) conns[i].get()->done(); - continue; - } - - assert( conns[i] ); - retryQueries.erase( i ); - - bool retry = false; - - try { - - if( ! _cursors[i].raw()->initLazyFinish( retry ) ) { - - warning() << "invalid result from " << conns[i]->getHost() << ( retry ? ", retrying" : "" ) << endl; - _cursors[i].reset( NULL ); - - if( ! retry ){ - socketExs.push_back( str::stream() << "error querying server: " << servers[i] ); - conns[i]->done(); - } - else { - retryQueries.insert( i ); - } - - continue; - } - } - catch ( MsgAssertionException& e ){ - socketExs.push_back( e.what() + errLoc ); - _cursors[i].reset( NULL ); - conns[i]->done(); - continue; - } - catch ( SocketException& e ) { - socketExs.push_back( e.what() + errLoc ); - _cursors[i].reset( NULL ); - conns[i]->done(); - continue; - } - catch( std::exception& e ){ - otherExs.push_back( e.what() + errLoc ); - _cursors[i].reset( NULL ); - conns[i]->done(); - continue; - } - - try { - _cursors[i].raw()->attach( conns[i].get() ); // this calls done on conn - _checkCursor( _cursors[i].raw() ); - - finishedQueries++; - } - catch ( StaleConfigException& e ){ - - // Our stored configuration data is actually stale, we need to reload it - // when we throw our exception - allConfigStale = true; - - staleConfigExs.push_back( (string)"stale config detected for " + e.what() + errLoc ); - _cursors[i].reset( NULL ); - conns[i]->done(); - continue; - } - catch( std::exception& e ){ - otherExs.push_back( e.what() + errLoc ); - _cursors[i].reset( NULL ); - conns[i]->done(); - continue; - } - } - - // Don't exceed our max retries, should not happen - assert( retries < 5 ); - } - while( retryQueries.size() > 0 /* something to retry */ && - ( socketExs.size() == 0 || returnPartial ) /* no conn issues */ && - staleConfigExs.size() == 0 /* no config issues */ && - otherExs.size() == 0 /* no other issues */); - - // Assert that our conns are all closed! - for( vector< shared_ptr<ShardConnection> >::iterator i = conns.begin(); i < conns.end(); ++i ){ - assert( ! (*i) || ! (*i)->ok() ); - } - - // Handle errors we got during initialization. - // If we're returning partial results, we can ignore socketExs, but nothing else - // Log a warning in any case, so we don't lose these messages - bool throwException = ( socketExs.size() > 0 && ! returnPartial ) || staleConfigExs.size() > 0 || otherExs.size() > 0; - - if( socketExs.size() > 0 || staleConfigExs.size() > 0 || otherExs.size() > 0 ) { - - vector<string> errMsgs; - - errMsgs.insert( errMsgs.end(), staleConfigExs.begin(), staleConfigExs.end() ); - errMsgs.insert( errMsgs.end(), otherExs.begin(), otherExs.end() ); - errMsgs.insert( errMsgs.end(), socketExs.begin(), socketExs.end() ); - - stringstream errMsg; - errMsg << "could not initialize cursor across all shards because : "; - for( vector<string>::iterator i = errMsgs.begin(); i != errMsgs.end(); i++ ){ - if( i != errMsgs.begin() ) errMsg << " :: and :: "; - errMsg << *i; - } - - if( throwException && staleConfigExs.size() > 0 ) - throw StaleConfigException( _ns , errMsg.str() , ! allConfigStale ); - else if( throwException ) - throw DBException( errMsg.str(), 14827 ); - else - warning() << errMsg.str() << endl; - } - - if( retries > 0 ) - log() << "successfully finished parallel query after " << retries << " retries" << endl; - - } - - ParallelSortClusteredCursor::~ParallelSortClusteredCursor() { - delete [] _cursors; - _cursors = 0; - } - - bool ParallelSortClusteredCursor::more() { - - if ( _needToSkip > 0 ) { - int n = _needToSkip; - _needToSkip = 0; - - while ( n > 0 && more() ) { - BSONObj x = next(); - n--; - } - - _needToSkip = n; - } - - for ( int i=0; i<_numServers; i++ ) { - if ( _cursors[i].more() ) - return true; - } - return false; - } - - BSONObj ParallelSortClusteredCursor::next() { - BSONObj best = BSONObj(); - int bestFrom = -1; - - for( int j = 0; j < _numServers; j++ ){ - - // Iterate _numServers times, starting one past the last server we used. - // This means we actually start at server #1, not #0, but shouldn't matter - - int i = ( j + _lastFrom + 1 ) % _numServers; - - if ( ! _cursors[i].more() ) - continue; - - BSONObj me = _cursors[i].peek(); - - if ( best.isEmpty() ) { - best = me; - bestFrom = i; - if( _sortKey.isEmpty() ) break; - continue; - } - - int comp = best.woSortOrder( me , _sortKey , true ); - if ( comp < 0 ) - continue; - - best = me; - bestFrom = i; - } - - _lastFrom = bestFrom; - - uassert( 10019 , "no more elements" , ! best.isEmpty() ); - _cursors[bestFrom].next(); - - return best; - } - - void ParallelSortClusteredCursor::_explain( map< string,list<BSONObj> >& out ) { - for ( set<ServerAndQuery>::iterator i=_servers.begin(); i!=_servers.end(); ++i ) { - const ServerAndQuery& sq = *i; - list<BSONObj> & l = out[sq._server]; - l.push_back( explain( sq._server , sq._extra ) ); - } - - } - - // ----------------- - // ---- Future ----- - // ----------------- - - Future::CommandResult::CommandResult( const string& server , const string& db , const BSONObj& cmd , int options , DBClientBase * conn ) - :_server(server) ,_db(db) , _options(options), _cmd(cmd) ,_conn(conn) ,_done(false) - { - try { - if ( ! _conn ){ - _connHolder.reset( new ScopedDbConnection( _server ) ); - _conn = _connHolder->get(); - } - - if ( _conn->lazySupported() ) { - _cursor.reset( new DBClientCursor(_conn, _db + ".$cmd", _cmd, -1/*limit*/, 0, NULL, _options, 0)); - _cursor->initLazy(); - } - else { - _done = true; // we set _done first because even if there is an error we're done - _ok = _conn->runCommand( db , cmd , _res , options ); - } - } - catch ( std::exception& e ) { - error() << "Future::spawnComand (part 1) exception: " << e.what() << endl; - _ok = false; - _done = true; - } - } - - bool Future::CommandResult::join() { - if (_done) - return _ok; - - try { - // TODO: Allow retries? - bool retry = false; - bool finished = _cursor->initLazyFinish( retry ); - - // Shouldn't need to communicate with server any more - if ( _connHolder ) - _connHolder->done(); - - uassert(14812, str::stream() << "Error running command on server: " << _server, finished); - massert(14813, "Command returned nothing", _cursor->more()); - - _res = _cursor->nextSafe(); - _ok = _res["ok"].trueValue(); - - } - catch ( std::exception& e ) { - error() << "Future::spawnComand (part 2) exception: " << e.what() << endl; - _ok = false; - } - - _done = true; - return _ok; - } - - shared_ptr<Future::CommandResult> Future::spawnCommand( const string& server , const string& db , const BSONObj& cmd , int options , DBClientBase * conn ) { - shared_ptr<Future::CommandResult> res (new Future::CommandResult( server , db , cmd , options , conn )); - return res; - } - -} diff --git a/client/parallel.h b/client/parallel.h deleted file mode 100644 index cbf64826596..00000000000 --- a/client/parallel.h +++ /dev/null @@ -1,315 +0,0 @@ -// parallel.h - -/* Copyright 2009 10gen Inc. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -/** - tools for working in parallel/sharded/clustered environment - */ - -#include "../pch.h" -#include "dbclient.h" -#include "redef_macros.h" -#include "../db/dbmessage.h" -#include "../db/matcher.h" -#include "../util/concurrency/mvar.h" - -namespace mongo { - - /** - * holder for a server address and a query to run - */ - class ServerAndQuery { - public: - ServerAndQuery( const string& server , BSONObj extra = BSONObj() , BSONObj orderObject = BSONObj() ) : - _server( server ) , _extra( extra.getOwned() ) , _orderObject( orderObject.getOwned() ) { - } - - bool operator<( const ServerAndQuery& other ) const { - if ( ! _orderObject.isEmpty() ) - return _orderObject.woCompare( other._orderObject ) < 0; - - if ( _server < other._server ) - return true; - if ( other._server > _server ) - return false; - return _extra.woCompare( other._extra ) < 0; - } - - string toString() const { - StringBuilder ss; - ss << "server:" << _server << " _extra:" << _extra.toString() << " _orderObject:" << _orderObject.toString(); - return ss.str(); - } - - operator string() const { - return toString(); - } - - string _server; - BSONObj _extra; - BSONObj _orderObject; - }; - - /** - * this is a cursor that works over a set of servers - * can be used in serial/paralellel as controlled by sub classes - */ - class ClusteredCursor { - public: - ClusteredCursor( QueryMessage& q ); - ClusteredCursor( const string& ns , const BSONObj& q , int options=0 , const BSONObj& fields=BSONObj() ); - virtual ~ClusteredCursor(); - - /** call before using */ - void init(); - - virtual bool more() = 0; - virtual BSONObj next() = 0; - - static BSONObj concatQuery( const BSONObj& query , const BSONObj& extraFilter ); - - virtual string type() const = 0; - - virtual BSONObj explain(); - - protected: - - virtual void _init() = 0; - - auto_ptr<DBClientCursor> query( const string& server , int num = 0 , BSONObj extraFilter = BSONObj() , int skipLeft = 0 , bool lazy=false ); - BSONObj explain( const string& server , BSONObj extraFilter = BSONObj() ); - - /** - * checks the cursor for any errors - * will throw an exceptionif an error is encountered - */ - void _checkCursor( DBClientCursor * cursor ); - - static BSONObj _concatFilter( const BSONObj& filter , const BSONObj& extraFilter ); - - virtual void _explain( map< string,list<BSONObj> >& out ) = 0; - - string _ns; - BSONObj _query; - int _options; - BSONObj _fields; - int _batchSize; - - bool _didInit; - - bool _done; - }; - - - class FilteringClientCursor { - public: - FilteringClientCursor( const BSONObj filter = BSONObj() ); - FilteringClientCursor( DBClientCursor* cursor , const BSONObj filter = BSONObj() ); - FilteringClientCursor( auto_ptr<DBClientCursor> cursor , const BSONObj filter = BSONObj() ); - ~FilteringClientCursor(); - - void reset( auto_ptr<DBClientCursor> cursor ); - void reset( DBClientCursor* cursor ); - - bool more(); - BSONObj next(); - - BSONObj peek(); - - DBClientCursor* raw() { return _cursor.get(); } - - private: - void _advance(); - - Matcher _matcher; - auto_ptr<DBClientCursor> _cursor; - - BSONObj _next; - bool _done; - }; - - - class Servers { - public: - Servers() { - } - - void add( const ServerAndQuery& s ) { - add( s._server , s._extra ); - } - - void add( const string& server , const BSONObj& filter ) { - vector<BSONObj>& mine = _filters[server]; - mine.push_back( filter.getOwned() ); - } - - // TOOO: pick a less horrible name - class View { - View( const Servers* s ) { - for ( map<string, vector<BSONObj> >::const_iterator i=s->_filters.begin(); i!=s->_filters.end(); ++i ) { - _servers.push_back( i->first ); - _filters.push_back( i->second ); - } - } - public: - int size() const { - return _servers.size(); - } - - string getServer( int n ) const { - return _servers[n]; - } - - vector<BSONObj> getFilter( int n ) const { - return _filters[ n ]; - } - - private: - vector<string> _servers; - vector< vector<BSONObj> > _filters; - - friend class Servers; - }; - - View view() const { - return View( this ); - } - - - private: - map<string, vector<BSONObj> > _filters; - - friend class View; - }; - - - /** - * runs a query in serial across any number of servers - * returns all results from 1 server, then the next, etc... - */ - class SerialServerClusteredCursor : public ClusteredCursor { - public: - SerialServerClusteredCursor( const set<ServerAndQuery>& servers , QueryMessage& q , int sortOrder=0); - virtual bool more(); - virtual BSONObj next(); - virtual string type() const { return "SerialServer"; } - - protected: - virtual void _explain( map< string,list<BSONObj> >& out ); - - void _init() {} - - vector<ServerAndQuery> _servers; - unsigned _serverIndex; - - FilteringClientCursor _current; - - int _needToSkip; - }; - - - /** - * runs a query in parellel across N servers - * sots - */ - class ParallelSortClusteredCursor : public ClusteredCursor { - public: - ParallelSortClusteredCursor( const set<ServerAndQuery>& servers , QueryMessage& q , const BSONObj& sortKey ); - ParallelSortClusteredCursor( const set<ServerAndQuery>& servers , const string& ns , - const Query& q , int options=0, const BSONObj& fields=BSONObj() ); - virtual ~ParallelSortClusteredCursor(); - virtual bool more(); - virtual BSONObj next(); - virtual string type() const { return "ParallelSort"; } - protected: - void _finishCons(); - void _init(); - - virtual void _explain( map< string,list<BSONObj> >& out ); - - int _numServers; - int _lastFrom; - set<ServerAndQuery> _servers; - BSONObj _sortKey; - - FilteringClientCursor * _cursors; - int _needToSkip; - }; - - /** - * tools for doing asynchronous operations - * right now uses underlying sync network ops and uses another thread - * should be changed to use non-blocking io - */ - class Future { - public: - class CommandResult { - public: - - string getServer() const { return _server; } - - bool isDone() const { return _done; } - - bool ok() const { - assert( _done ); - return _ok; - } - - BSONObj result() const { - assert( _done ); - return _res; - } - - /** - blocks until command is done - returns ok() - */ - bool join(); - - private: - - CommandResult( const string& server , const string& db , const BSONObj& cmd , int options , DBClientBase * conn ); - - string _server; - string _db; - int _options; - BSONObj _cmd; - DBClientBase * _conn; - scoped_ptr<ScopedDbConnection> _connHolder; // used if not provided a connection - - scoped_ptr<DBClientCursor> _cursor; - - BSONObj _res; - bool _ok; - bool _done; - - friend class Future; - }; - - - /** - * @param server server name - * @param db db name - * @param cmd cmd to exec - * @param conn optional connection to use. will use standard pooled if non-specified - */ - static shared_ptr<CommandResult> spawnCommand( const string& server , const string& db , const BSONObj& cmd , int options , DBClientBase * conn = 0 ); - }; - - -} - -#include "undef_macros.h" diff --git a/client/redef_macros.h b/client/redef_macros.h deleted file mode 100644 index 5a39561afd5..00000000000 --- a/client/redef_macros.h +++ /dev/null @@ -1,102 +0,0 @@ -/** @file redef_macros.h macros for mongo internals - - @see undef_macros.h undefines these after use to minimize name pollution. -*/ - -/* Copyright 2009 10gen Inc. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -// If you define a new global un-prefixed macro, please add it here and in undef_macros - -#define MONGO_MACROS_PUSHED 1 - -// util/allocator.h -#pragma push_macro("malloc") -#undef malloc -#define malloc MONGO_malloc -#pragma push_macro("realloc") -#undef realloc -#define realloc MONGO_realloc - -// util/assert_util.h -#pragma push_macro("assert") -#undef assert -#define assert MONGO_assert -#pragma push_macro("verify") -#undef verify -#define verify MONGO_verify -#pragma push_macro("dassert") -#undef dassert -#define dassert MONGO_dassert -#pragma push_macro("wassert") -#undef wassert -#define wassert MONGO_wassert -#pragma push_macro("massert") -#undef massert -#define massert MONGO_massert -#pragma push_macro("uassert") -#undef uassert -#define uassert MONGO_uassert -#define BOOST_CHECK_EXCEPTION MONGO_BOOST_CHECK_EXCEPTION -#pragma push_macro("DESTRUCTOR_GUARD") -#undef DESTRUCTOR_GUARD -#define DESTRUCTOR_GUARD MONGO_DESTRUCTOR_GUARD - -// util/goodies.h -#pragma push_macro("PRINT") -#undef PRINT -#define PRINT MONGO_PRINT -#pragma push_macro("PRINTFL") -#undef PRINTFL -#define PRINTFL MONGO_PRINTFL -#pragma push_macro("asctime") -#undef asctime -#define asctime MONGO_asctime -#pragma push_macro("gmtime") -#undef gmtime -#define gmtime MONGO_gmtime -#pragma push_macro("localtime") -#undef localtime -#define localtime MONGO_localtime -#pragma push_macro("ctime") -#undef ctime -#define ctime MONGO_ctime - -// util/debug_util.h -#pragma push_macro("DEV") -#undef DEV -#define DEV MONGO_DEV -#pragma push_macro("DEBUGGING") -#undef DEBUGGING -#define DEBUGGING MONGO_DEBUGGING -#pragma push_macro("SOMETIMES") -#undef SOMETIMES -#define SOMETIMES MONGO_SOMETIMES -#pragma push_macro("OCCASIONALLY") -#undef OCCASIONALLY -#define OCCASIONALLY MONGO_OCCASIONALLY -#pragma push_macro("RARELY") -#undef RARELY -#define RARELY MONGO_RARELY -#pragma push_macro("ONCE") -#undef ONCE -#define ONCE MONGO_ONCE - -// util/log.h -#pragma push_macro("LOG") -#undef LOG -#define LOG MONGO_LOG - - diff --git a/client/simple_client_demo.cpp b/client/simple_client_demo.cpp deleted file mode 100644 index f4278dd4e54..00000000000 --- a/client/simple_client_demo.cpp +++ /dev/null @@ -1,54 +0,0 @@ -/* simple_client_demo.cpp - - See also : http://www.mongodb.org/pages/viewpage.action?pageId=133415 - - How to build and run: - - (1) Using the mongoclient: - g++ simple_client_demo.cpp -lmongoclient -lboost_thread-mt -lboost_filesystem -lboost_program_options - ./a.out - - (2) using client_lib.cpp: - g++ -I .. simple_client_demo.cpp mongo_client_lib.cpp -lboost_thread-mt -lboost_filesystem - ./a.out -*/ - -#include <iostream> -#include "dbclient.h" // the mongo c++ driver - -using namespace std; -using namespace mongo; -using namespace bson; - -int main() { - try { - cout << "connecting to localhost..." << endl; - DBClientConnection c; - c.connect("localhost"); - cout << "connected ok" << endl; - unsigned long long count = c.count("test.foo"); - cout << "count of exiting documents in collection test.foo : " << count << endl; - - bo o = BSON( "hello" << "world" ); - c.insert("test.foo", o); - - string e = c.getLastError(); - if( !e.empty() ) { - cout << "insert #1 failed: " << e << endl; - } - - // make an index with a unique key constraint - c.ensureIndex("test.foo", BSON("hello"<<1), /*unique*/true); - - c.insert("test.foo", o); // will cause a dup key error on "hello" field - cout << "we expect a dup key error here:" << endl; - cout << " " << c.getLastErrorDetailed().toString() << endl; - } - catch(DBException& e) { - cout << "caught DBException " << e.toString() << endl; - return 1; - } - - return 0; -} - diff --git a/client/syncclusterconnection.cpp b/client/syncclusterconnection.cpp deleted file mode 100644 index 34633d15fa8..00000000000 --- a/client/syncclusterconnection.cpp +++ /dev/null @@ -1,407 +0,0 @@ -// syncclusterconnection.cpp -/* - * Copyright 2010 10gen Inc. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - - -#include "pch.h" -#include "syncclusterconnection.h" -#include "../db/dbmessage.h" - -// error codes 8000-8009 - -namespace mongo { - - SyncClusterConnection::SyncClusterConnection( const list<HostAndPort> & L, double socketTimeout) : _mutex("SyncClusterConnection"), _socketTimeout( socketTimeout ) { - { - stringstream s; - int n=0; - for( list<HostAndPort>::const_iterator i = L.begin(); i != L.end(); i++ ) { - if( ++n > 1 ) s << ','; - s << i->toString(); - } - _address = s.str(); - } - for( list<HostAndPort>::const_iterator i = L.begin(); i != L.end(); i++ ) - _connect( i->toString() ); - } - - SyncClusterConnection::SyncClusterConnection( string commaSeperated, double socketTimeout) : _mutex("SyncClusterConnection"), _socketTimeout( socketTimeout ) { - _address = commaSeperated; - string::size_type idx; - while ( ( idx = commaSeperated.find( ',' ) ) != string::npos ) { - string h = commaSeperated.substr( 0 , idx ); - commaSeperated = commaSeperated.substr( idx + 1 ); - _connect( h ); - } - _connect( commaSeperated ); - uassert( 8004 , "SyncClusterConnection needs 3 servers" , _conns.size() == 3 ); - } - - SyncClusterConnection::SyncClusterConnection( string a , string b , string c, double socketTimeout) : _mutex("SyncClusterConnection"), _socketTimeout( socketTimeout ) { - _address = a + "," + b + "," + c; - // connect to all even if not working - _connect( a ); - _connect( b ); - _connect( c ); - } - - SyncClusterConnection::SyncClusterConnection( SyncClusterConnection& prev, double socketTimeout) : _mutex("SyncClusterConnection"), _socketTimeout( socketTimeout ) { - assert(0); - } - - SyncClusterConnection::~SyncClusterConnection() { - for ( size_t i=0; i<_conns.size(); i++ ) - delete _conns[i]; - _conns.clear(); - } - - bool SyncClusterConnection::prepare( string& errmsg ) { - _lastErrors.clear(); - return fsync( errmsg ); - } - - bool SyncClusterConnection::fsync( string& errmsg ) { - bool ok = true; - errmsg = ""; - for ( size_t i=0; i<_conns.size(); i++ ) { - BSONObj res; - try { - if ( _conns[i]->simpleCommand( "admin" , &res , "fsync" ) ) - continue; - } - catch ( DBException& e ) { - errmsg += e.toString(); - } - catch ( std::exception& e ) { - errmsg += e.what(); - } - catch ( ... ) { - } - ok = false; - errmsg += " " + _conns[i]->toString() + ":" + res.toString(); - } - return ok; - } - - void SyncClusterConnection::_checkLast() { - _lastErrors.clear(); - vector<string> errors; - - for ( size_t i=0; i<_conns.size(); i++ ) { - BSONObj res; - string err; - try { - if ( ! _conns[i]->runCommand( "admin" , BSON( "getlasterror" << 1 << "fsync" << 1 ) , res ) ) - err = "cmd failed: "; - } - catch ( std::exception& e ) { - err += e.what(); - } - catch ( ... ) { - err += "unknown failure"; - } - _lastErrors.push_back( res.getOwned() ); - errors.push_back( err ); - } - - assert( _lastErrors.size() == errors.size() && _lastErrors.size() == _conns.size() ); - - stringstream err; - bool ok = true; - - for ( size_t i = 0; i<_conns.size(); i++ ) { - BSONObj res = _lastErrors[i]; - if ( res["ok"].trueValue() && (res["fsyncFiles"].numberInt() > 0 || res.hasElement("waited"))) - continue; - ok = false; - err << _conns[i]->toString() << ": " << res << " " << errors[i]; - } - - if ( ok ) - return; - throw UserException( 8001 , (string)"SyncClusterConnection write op failed: " + err.str() ); - } - - BSONObj SyncClusterConnection::getLastErrorDetailed() { - if ( _lastErrors.size() ) - return _lastErrors[0]; - return DBClientBase::getLastErrorDetailed(); - } - - void SyncClusterConnection::_connect( string host ) { - log() << "SyncClusterConnection connecting to [" << host << "]" << endl; - DBClientConnection * c = new DBClientConnection( true ); - c->setSoTimeout( _socketTimeout ); - string errmsg; - if ( ! c->connect( host , errmsg ) ) - log() << "SyncClusterConnection connect fail to: " << host << " errmsg: " << errmsg << endl; - _connAddresses.push_back( host ); - _conns.push_back( c ); - } - - bool SyncClusterConnection::callRead( Message& toSend , Message& response ) { - // TODO: need to save state of which one to go back to somehow... - return _conns[0]->callRead( toSend , response ); - } - - BSONObj SyncClusterConnection::findOne(const string &ns, const Query& query, const BSONObj *fieldsToReturn, int queryOptions) { - - if ( ns.find( ".$cmd" ) != string::npos ) { - string cmdName = query.obj.firstElementFieldName(); - - int lockType = _lockType( cmdName ); - - if ( lockType > 0 ) { // write $cmd - string errmsg; - if ( ! prepare( errmsg ) ) - throw UserException( 13104 , (string)"SyncClusterConnection::findOne prepare failed: " + errmsg ); - - vector<BSONObj> all; - for ( size_t i=0; i<_conns.size(); i++ ) { - all.push_back( _conns[i]->findOne( ns , query , 0 , queryOptions ).getOwned() ); - } - - _checkLast(); - - for ( size_t i=0; i<all.size(); i++ ) { - BSONObj temp = all[i]; - if ( isOk( temp ) ) - continue; - stringstream ss; - ss << "write $cmd failed on a node: " << temp.jsonString(); - ss << " " << _conns[i]->toString(); - ss << " ns: " << ns; - ss << " cmd: " << query.toString(); - throw UserException( 13105 , ss.str() ); - } - - return all[0]; - } - } - - return DBClientBase::findOne( ns , query , fieldsToReturn , queryOptions ); - } - - bool SyncClusterConnection::auth(const string &dbname, const string &username, const string &password_text, string& errmsg, bool digestPassword) { - for (vector<DBClientConnection*>::iterator it = _conns.begin(); it < _conns.end(); it++) { - massert( 15848, "sync cluster of sync clusters?", (*it)->type() != ConnectionString::SYNC); - - if (!(*it)->auth(dbname, username, password_text, errmsg, digestPassword)) { - return false; - } - } - return true; - } - - auto_ptr<DBClientCursor> SyncClusterConnection::query(const string &ns, Query query, int nToReturn, int nToSkip, - const BSONObj *fieldsToReturn, int queryOptions, int batchSize ) { - _lastErrors.clear(); - if ( ns.find( ".$cmd" ) != string::npos ) { - string cmdName = query.obj.firstElementFieldName(); - int lockType = _lockType( cmdName ); - uassert( 13054 , (string)"write $cmd not supported in SyncClusterConnection::query for:" + cmdName , lockType <= 0 ); - } - - return _queryOnActive( ns , query , nToReturn , nToSkip , fieldsToReturn , queryOptions , batchSize ); - } - - bool SyncClusterConnection::_commandOnActive(const string &dbname, const BSONObj& cmd, BSONObj &info, int options ) { - auto_ptr<DBClientCursor> cursor = _queryOnActive( dbname + ".$cmd" , cmd , 1 , 0 , 0 , options , 0 ); - if ( cursor->more() ) - info = cursor->next().copy(); - else - info = BSONObj(); - return isOk( info ); - } - - auto_ptr<DBClientCursor> SyncClusterConnection::_queryOnActive(const string &ns, Query query, int nToReturn, int nToSkip, - const BSONObj *fieldsToReturn, int queryOptions, int batchSize ) { - - for ( size_t i=0; i<_conns.size(); i++ ) { - try { - auto_ptr<DBClientCursor> cursor = - _conns[i]->query( ns , query , nToReturn , nToSkip , fieldsToReturn , queryOptions , batchSize ); - if ( cursor.get() ) - return cursor; - log() << "query failed to: " << _conns[i]->toString() << " no data" << endl; - } - catch ( ... ) { - log() << "query failed to: " << _conns[i]->toString() << " exception" << endl; - } - } - throw UserException( 8002 , "all servers down!" ); - } - - auto_ptr<DBClientCursor> SyncClusterConnection::getMore( const string &ns, long long cursorId, int nToReturn, int options ) { - uassert( 10022 , "SyncClusterConnection::getMore not supported yet" , 0); - auto_ptr<DBClientCursor> c; - return c; - } - - void SyncClusterConnection::insert( const string &ns, BSONObj obj , int flags) { - - uassert( 13119 , (string)"SyncClusterConnection::insert obj has to have an _id: " + obj.jsonString() , - ns.find( ".system.indexes" ) != string::npos || obj["_id"].type() ); - - string errmsg; - if ( ! prepare( errmsg ) ) - throw UserException( 8003 , (string)"SyncClusterConnection::insert prepare failed: " + errmsg ); - - for ( size_t i=0; i<_conns.size(); i++ ) { - _conns[i]->insert( ns , obj , flags); - } - - _checkLast(); - } - - void SyncClusterConnection::insert( const string &ns, const vector< BSONObj >& v , int flags) { - uassert( 10023 , "SyncClusterConnection bulk insert not implemented" , 0); - } - - void SyncClusterConnection::remove( const string &ns , Query query, bool justOne ) { - string errmsg; - if ( ! prepare( errmsg ) ) - throw UserException( 8020 , (string)"SyncClusterConnection::remove prepare failed: " + errmsg ); - - for ( size_t i=0; i<_conns.size(); i++ ) { - _conns[i]->remove( ns , query , justOne ); - } - - _checkLast(); - } - - void SyncClusterConnection::update( const string &ns , Query query , BSONObj obj , bool upsert , bool multi ) { - - if ( upsert ) { - uassert( 13120 , "SyncClusterConnection::update upsert query needs _id" , query.obj["_id"].type() ); - } - - if ( _writeConcern ) { - string errmsg; - if ( ! prepare( errmsg ) ) - throw UserException( 8005 , (string)"SyncClusterConnection::udpate prepare failed: " + errmsg ); - } - - for ( size_t i = 0; i < _conns.size(); i++ ) { - try { - _conns[i]->update( ns , query , obj , upsert , multi ); - } - catch ( std::exception& e ) { - if ( _writeConcern ) - throw e; - } - } - - if ( _writeConcern ) { - _checkLast(); - assert( _lastErrors.size() > 1 ); - - int a = _lastErrors[0]["n"].numberInt(); - for ( unsigned i=1; i<_lastErrors.size(); i++ ) { - int b = _lastErrors[i]["n"].numberInt(); - if ( a == b ) - continue; - - throw UpdateNotTheSame( 8017 , - str::stream() - << "update not consistent " - << " ns: " << ns - << " query: " << query.toString() - << " update: " << obj - << " gle1: " << _lastErrors[0] - << " gle2: " << _lastErrors[i] , - _connAddresses , _lastErrors ); - } - } - } - - string SyncClusterConnection::_toString() const { - stringstream ss; - ss << "SyncClusterConnection [" << _address << "]"; - return ss.str(); - } - - bool SyncClusterConnection::call( Message &toSend, Message &response, bool assertOk , string * actualServer ) { - uassert( 8006 , "SyncClusterConnection::call can only be used directly for dbQuery" , - toSend.operation() == dbQuery ); - - DbMessage d( toSend ); - uassert( 8007 , "SyncClusterConnection::call can't handle $cmd" , strstr( d.getns(), "$cmd" ) == 0 ); - - for ( size_t i=0; i<_conns.size(); i++ ) { - try { - bool ok = _conns[i]->call( toSend , response , assertOk ); - if ( ok ) { - if ( actualServer ) - *actualServer = _connAddresses[i]; - return ok; - } - log() << "call failed to: " << _conns[i]->toString() << " no data" << endl; - } - catch ( ... ) { - log() << "call failed to: " << _conns[i]->toString() << " exception" << endl; - } - } - throw UserException( 8008 , "all servers down!" ); - } - - void SyncClusterConnection::say( Message &toSend, bool isRetry ) { - string errmsg; - if ( ! prepare( errmsg ) ) - throw UserException( 13397 , (string)"SyncClusterConnection::say prepare failed: " + errmsg ); - - for ( size_t i=0; i<_conns.size(); i++ ) { - _conns[i]->say( toSend ); - } - - _checkLast(); - } - - void SyncClusterConnection::sayPiggyBack( Message &toSend ) { - assert(0); - } - - int SyncClusterConnection::_lockType( const string& name ) { - { - scoped_lock lk(_mutex); - map<string,int>::iterator i = _lockTypes.find( name ); - if ( i != _lockTypes.end() ) - return i->second; - } - - BSONObj info; - uassert( 13053 , str::stream() << "help failed: " << info , _commandOnActive( "admin" , BSON( name << "1" << "help" << 1 ) , info ) ); - - int lockType = info["lockType"].numberInt(); - - scoped_lock lk(_mutex); - _lockTypes[name] = lockType; - return lockType; - } - - void SyncClusterConnection::killCursor( long long cursorID ) { - // should never need to do this - assert(0); - } - - void SyncClusterConnection::setAllSoTimeouts( double socketTimeout ){ - _socketTimeout = socketTimeout; - for ( size_t i=0; i<_conns.size(); i++ ) - - if( _conns[i] ) _conns[i]->setSoTimeout( socketTimeout ); - } - -} diff --git a/client/syncclusterconnection.h b/client/syncclusterconnection.h deleted file mode 100644 index 68dd338a408..00000000000 --- a/client/syncclusterconnection.h +++ /dev/null @@ -1,147 +0,0 @@ -// @file syncclusterconnection.h - -/* - * Copyright 2010 10gen Inc. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -#pragma once - -#include "../pch.h" -#include "dbclient.h" -#include "redef_macros.h" - -namespace mongo { - - /** - * This is a connection to a cluster of servers that operate as one - * for super high durability. - * - * Write operations are two-phase. First, all nodes are asked to fsync. If successful - * everywhere, the write is sent everywhere and then followed by an fsync. There is no - * rollback if a problem occurs during the second phase. Naturally, with all these fsyncs, - * these operations will be quite slow -- use sparingly. - * - * Read operations are sent to a single random node. - * - * The class checks if a command is read or write style, and sends to a single - * node if a read lock command and to all in two phases with a write style command. - */ - class SyncClusterConnection : public DBClientBase { - public: - /** - * @param commaSeparated should be 3 hosts comma separated - */ - SyncClusterConnection( const list<HostAndPort> &, double socketTimeout = 0); - SyncClusterConnection( string commaSeparated, double socketTimeout = 0); - SyncClusterConnection( string a , string b , string c, double socketTimeout = 0 ); - ~SyncClusterConnection(); - - /** - * @return true if all servers are up and ready for writes - */ - bool prepare( string& errmsg ); - - /** - * runs fsync on all servers - */ - bool fsync( string& errmsg ); - - // --- from DBClientInterface - - virtual BSONObj findOne(const string &ns, const Query& query, const BSONObj *fieldsToReturn, int queryOptions); - - virtual auto_ptr<DBClientCursor> query(const string &ns, Query query, int nToReturn, int nToSkip, - const BSONObj *fieldsToReturn, int queryOptions, int batchSize ); - - virtual auto_ptr<DBClientCursor> getMore( const string &ns, long long cursorId, int nToReturn, int options ); - - virtual void insert( const string &ns, BSONObj obj, int flags=0); - - virtual void insert( const string &ns, const vector< BSONObj >& v, int flags=0); - - virtual void remove( const string &ns , Query query, bool justOne ); - - virtual void update( const string &ns , Query query , BSONObj obj , bool upsert , bool multi ); - - virtual bool call( Message &toSend, Message &response, bool assertOk , string * actualServer ); - virtual void say( Message &toSend, bool isRetry = false ); - virtual void sayPiggyBack( Message &toSend ); - - virtual void killCursor( long long cursorID ); - - virtual string getServerAddress() const { return _address; } - virtual bool isFailed() const { return false; } - virtual string toString() { return _toString(); } - - virtual BSONObj getLastErrorDetailed(); - - virtual bool callRead( Message& toSend , Message& response ); - - virtual ConnectionString::ConnectionType type() const { return ConnectionString::SYNC; } - - void setAllSoTimeouts( double socketTimeout ); - double getSoTimeout() const { return _socketTimeout; } - - virtual bool auth(const string &dbname, const string &username, const string &password_text, string& errmsg, bool digestPassword); - - virtual bool lazySupported() const { return false; } - private: - SyncClusterConnection( SyncClusterConnection& prev, double socketTimeout = 0 ); - string _toString() const; - bool _commandOnActive(const string &dbname, const BSONObj& cmd, BSONObj &info, int options=0); - auto_ptr<DBClientCursor> _queryOnActive(const string &ns, Query query, int nToReturn, int nToSkip, - const BSONObj *fieldsToReturn, int queryOptions, int batchSize ); - int _lockType( const string& name ); - void _checkLast(); - void _connect( string host ); - - string _address; - vector<string> _connAddresses; - vector<DBClientConnection*> _conns; - map<string,int> _lockTypes; - mongo::mutex _mutex; - - vector<BSONObj> _lastErrors; - - double _socketTimeout; - }; - - class UpdateNotTheSame : public UserException { - public: - UpdateNotTheSame( int code , const string& msg , const vector<string>& addrs , const vector<BSONObj>& lastErrors ) - : UserException( code , msg ) , _addrs( addrs ) , _lastErrors( lastErrors ) { - assert( _addrs.size() == _lastErrors.size() ); - } - - virtual ~UpdateNotTheSame() throw() { - } - - unsigned size() const { - return _addrs.size(); - } - - pair<string,BSONObj> operator[](unsigned i) const { - return make_pair( _addrs[i] , _lastErrors[i] ); - } - - private: - - vector<string> _addrs; - vector<BSONObj> _lastErrors; - }; - -}; - -#include "undef_macros.h" diff --git a/client/undef_macros.h b/client/undef_macros.h deleted file mode 100644 index c880f063771..00000000000 --- a/client/undef_macros.h +++ /dev/null @@ -1,83 +0,0 @@ -/** @file undef_macros.h remove mongo implementation macros after using */ - -/* Copyright 2009 10gen Inc. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -// If you define a new global un-prefixed macro, please add it here and in redef_macros - -// #pragma once // this file is intended to be processed multiple times - -#if !defined (MONGO_EXPOSE_MACROS) - -#ifdef MONGO_MACROS_PUSHED - -// util/allocator.h -#undef malloc -#pragma pop_macro("malloc") -#undef realloc -#pragma pop_macro("realloc") - -// util/assert_util.h -#undef assert -#pragma pop_macro("assert") -#undef dassert -#pragma pop_macro("dassert") -#undef wassert -#pragma pop_macro("wassert") -#undef massert -#pragma pop_macro("massert") -#undef uassert -#pragma pop_macro("uassert") -#undef BOOST_CHECK_EXCEPTION -#undef verify -#pragma pop_macro("verify") -#undef DESTRUCTOR_GUARD -#pragma pop_macro("DESTRUCTOR_GUARD") - -// util/goodies.h -#undef PRINT -#pragma pop_macro("PRINT") -#undef PRINTFL -#pragma pop_macro("PRINTFL") -#undef asctime -#pragma pop_macro("asctime") -#undef gmtime -#pragma pop_macro("gmtime") -#undef localtime -#pragma pop_macro("localtime") -#undef ctime -#pragma pop_macro("ctime") - -// util/debug_util.h -#undef DEV -#pragma pop_macro("DEV") -#undef DEBUGGING -#pragma pop_macro("DEBUGGING") -#undef SOMETIMES -#pragma pop_macro("SOMETIMES") -#undef OCCASIONALLY -#pragma pop_macro("OCCASIONALLY") -#undef RARELY -#pragma pop_macro("RARELY") -#undef ONCE -#pragma pop_macro("ONCE") - -// util/log.h -#undef LOG -#pragma pop_macro("LOG") - -#undef MONGO_MACROS_PUSHED -#endif -#endif |