sharding w version numbers checkpoint

5 files changed, 111 insertions, 15 deletions

diff --git a/s/commands.cpp b/s/commands.cpp
index e92073ec50c..48ef355a4d7 100644
--- a/s/commands.cpp
+++ b/s/commands.cpp
@@ -389,6 +389,12 @@ namespace mongo {
                 
                 // update config db
                 s.setServer( to );
+                
+                // need to increment version # for old server
+                Shard * randomShardOnOldServer = info->findShardOnServer( from );
+                if ( randomShardOnOldServer )
+                    randomShardOnOldServer->_markModified();
+
                 info->save();
 
                 // delete old data
diff --git a/s/shard.cpp b/s/shard.cpp
index a9f9b67af45..9e293c360b1 100644
--- a/s/shard.cpp
+++ b/s/shard.cpp
@@ -122,6 +122,26 @@ namespace mongo {
         _lastmod = 0;
     }
 
+    void Shard::save( bool check ){
+        cout << "HERE: " << _id << endl;
+        bool reload = ! _lastmod;
+        Model::save( check );
+        cout << "\t" << _id << endl;
+        if ( reload ){
+            // need to do this so that we get the new _lastMod and therefore version number
+
+            massert( "_id has to be filled in already" , ! _id.isEmpty() );
+            
+            string b = toString();
+            BSONObj q = _id.copy();
+            massert( "how could load fail?" , load( q ) );
+            cout << "before: " << q << "\t" << b << endl;
+            cout << "after : " << _id << "\t" << toString() << endl;
+            massert( "shard reload changed content!" , b == toString() );
+            massert( "id changed!" , q["_id"] == _id["_id"] );
+        }
+    }
+
     string Shard::toString() const {
         stringstream ss;
         ss << "shard  ns:" << _ns << " server: " << _server << " min: " << _min << " max: " << _max;
@@ -140,6 +160,7 @@ namespace mongo {
             BSONObj d = cursor->next();
             s->unserialize( d );
             _shards.push_back( s );
+            s->_id = d["_id"].wrap().getOwned();
         }
         conn.done();
         
@@ -150,7 +171,7 @@ namespace mongo {
             s->_max = _key.globalMax();
             s->_server = config->getPrimary();
             s->_markModified();
-
+            
             _shards.push_back( s );
 
             log() << "no shards for:" << ns << " so creating first: " << s->toString() << endl;
@@ -178,6 +199,17 @@ namespace mongo {
         throw UserException( "couldn't find a shard which should be impossible" );
     }
 
+    Shard* ShardManager::findShardOnServer( const string& server ) const {
+
+        for ( vector<Shard*>::const_iterator i=_shards.begin(); i!=_shards.end(); i++ ){
+            Shard* s = *i;
+            if ( s->getServer() == server )
+                return s;
+        }
+
+        return 0;
+    }
+
     int ShardManager::getShardsForQuery( vector<Shard*>& shards , const BSONObj& query ){
         int added = 0;
         
@@ -192,22 +224,25 @@ namespace mongo {
     }
     
     void ShardManager::save(){
+        ServerShardVersion a = getVersion();
+        
         for ( vector<Shard*>::const_iterator i=_shards.begin(); i!=_shards.end(); i++ ){
             Shard* s = *i;
             if ( ! s->_modified )
                 continue;
             s->save( true );
         }
-    }
 
+        massert( "how did version get smalled" , getVersion() >= a );
+    }
+    
     ServerShardVersion ShardManager::getVersion( const string& server ) const{
         // TODO: cache or something?
         
         ServerShardVersion max = 0;
-        cout << "getVersion for: " << server << endl;
+
         for ( vector<Shard*>::const_iterator i=_shards.begin(); i!=_shards.end(); i++ ){
             Shard* s = *i;
-            cout << "\t" << s->getServer() << endl;
             if ( s->getServer() != server )
                 continue;
             
@@ -218,6 +253,18 @@ namespace mongo {
         return max;
     }
 
+    ServerShardVersion ShardManager::getVersion() const{
+        ServerShardVersion max = 0;
+
+        for ( vector<Shard*>::const_iterator i=_shards.begin(); i!=_shards.end(); i++ ){
+            Shard* s = *i;
+            if ( s->_lastmod > max )
+                max = s->_lastmod;
+        }        
+
+        return max;
+    }
+
     string ShardManager::toString() const {
         stringstream ss;
         ss << "ShardManager: " << _ns << " key:" << _key.toString() << "\n";
diff --git a/s/shard.h b/s/shard.h
index 3cceca75d9d..482203188d1 100644
--- a/s/shard.h
+++ b/s/shard.h
@@ -43,6 +43,8 @@ namespace mongo {
      */    
     class Shard : public Model , boost::noncopyable {
     public:
+
+        Shard( ShardManager * info );
         
         BSONObj& getMin(){
             return _min;
@@ -77,13 +79,12 @@ namespace mongo {
         virtual void unserialize(const BSONObj& from);
         virtual string modelServer();
 
-    public:
-        Shard( ShardManager * info );
+        virtual void save( bool check=false );
+
+        void _markModified();
         
     private:
         
-        void _markModified();
-
         ShardManager * _manager;
         
         string _ns;
@@ -120,8 +121,11 @@ namespace mongo {
         bool hasShardKey( const BSONObj& obj );
 
         Shard& findShard( const BSONObj& obj );
+        Shard* findShardOnServer( const string& server ) const;
+        
         ShardKeyPattern& getShardKey(){  return _key; }
         
+
         /**
          * @return number of shards added to the vector
          */
@@ -133,13 +137,16 @@ namespace mongo {
         operator string() const { return toString(); }
 
         ServerShardVersion getVersion( const string& server ) const;
+        ServerShardVersion getVersion() const;
     private:
         DBConfig * _config;
         string _ns;
         ShardKeyPattern _key;
         
         vector<Shard*> _shards;
-        
+
+        mutex _lock;
+
         friend class Shard;
     };
 
diff --git a/s/strategy.cpp b/s/strategy.cpp
index e84695c108b..5650962d2fa 100644
--- a/s/strategy.cpp
+++ b/s/strategy.cpp
@@ -50,7 +50,7 @@ namespace mongo {
         dbcon.done();
     }
 
-    void checkShardVersion( DBClientBase& conn , const string& ns ){
+    void checkShardVersion( DBClientBase& conn , const string& ns , bool authoritative ){
         // TODO: cache, optimize, etc...
         
         DBConfig * conf = grid.getDBConfig( ns );
@@ -60,12 +60,43 @@ namespace mongo {
         if ( ! conf->sharded( ns ) )
             return;
         
-        ServerShardVersion version = conf->getShardManager( ns )->getVersion( conn.getServerAddress() );
-        cout << "got version: " << version << " for : " << ns << endl;
         
+        ShardManager * manager = conf->getShardManager( ns , authoritative );
         
-        // grid->getVersion( conn.server() , ns );
-        // check
+        ServerShardVersion version = manager->getVersion( conn.getServerAddress() );
+
+        BSONObj result;
+        if ( setShardVersion( conn , ns , version , authoritative , result ) ){
+            return;
+        }
+
+        log(1) << "       setShardVersion failed!\n" << result << endl;
+
+        if ( result.getBoolField( "need_authoritative" ) )
+            massert( "need_authoritative set but in authoritative mode already" , ! authoritative );
+        
+        if ( ! authoritative ){
+            checkShardVersion( conn , ns , 1 );
+            return;
+        }
+        
+        log(1) << "     setShardVersion failed: " << result << endl;
+        massert( "setShardVersion failed!" , 0 );
     }
+    
+    bool setShardVersion( DBClientBase & conn , const string& ns , ServerShardVersion version , bool authoritative , BSONObj& result ){
 
+        BSONObjBuilder cmdBuilder;
+        cmdBuilder.append( "setShardVersion" , ns.c_str() );
+        cmdBuilder.append( "configdb" , configServer.modelServer() );
+        cmdBuilder.appendTimestamp( "version" , version );
+        if ( authoritative )
+            cmdBuilder.appendBool( "authoritative" , 1 );
+        BSONObj cmd = cmdBuilder.obj();
+        
+        log(1) << "    setShardVersion  " << conn.getServerAddress() << "  " << ns << "  " << cmd << endl;
+        
+        return conn.runCommand( "admin" , cmd , result );
+    }
+    
 }
diff --git a/s/strategy.h b/s/strategy.h
index f0e71909bf5..6ab244c0427 100644
--- a/s/strategy.h
+++ b/s/strategy.h
@@ -2,6 +2,9 @@
 
 #pragma once
 
+#include "stdafx.h"
+#include "shard.h"
+
 namespace mongo {
     
     class Strategy {
@@ -20,8 +23,10 @@ namespace mongo {
         
     };
 
-    void checkShardVersion( DBClientBase & conn , const string& ns );
+    void checkShardVersion( DBClientBase & conn , const string& ns , bool authoritative = false );
     
+    bool setShardVersion( DBClientBase & conn , const string& ns , ServerShardVersion version , bool authoritative , BSONObj& result );
+
     extern Strategy * SINGLE;
     extern Strategy * SHARDED;
 }