SERVER-13256 Rename chunk_diff-inl.cpp to a header to reflect its usage

1 files changed, 267 insertions, 0 deletions

diff --git a/src/mongo/s/chunk_diff-inl.h b/src/mongo/s/chunk_diff-inl.h
new file mode 100644
index 00000000000..0f88f2517b7
--- /dev/null
+++ b/src/mongo/s/chunk_diff-inl.h
@@ -0,0 +1,267 @@
+// @file chunk_diff_impl.hpp
+
+/**
+*    Copyright (C) 2008 10gen Inc.
+*
+*    This program is free software: you can redistribute it and/or  modify
+*    it under the terms of the GNU Affero General Public License, version 3,
+*    as published by the Free Software Foundation.
+*
+*    This program is distributed in the hope that it will be useful,
+*    but WITHOUT ANY WARRANTY; without even the implied warranty of
+*    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+*    GNU Affero General Public License for more details.
+*
+*    You should have received a copy of the GNU Affero General Public License
+*    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+*
+*    As a special exception, the copyright holders give permission to link the
+*    code of portions of this program with the OpenSSL library under certain
+*    conditions as described in each individual source file and distribute
+*    linked combinations including the program with the OpenSSL library. You
+*    must comply with the GNU Affero General Public License in all respects
+*    for all of the code used other than as permitted herein. If you modify
+*    file(s) with this exception, you may extend this exception to your
+*    version of the file(s), but you are not obligated to do so. If you do not
+*    wish to do so, delete this exception statement from your version. If you
+*    delete this exception statement from all source files in the program,
+*    then also delete it in the license file.
+*/
+
+#pragma once
+
+#include "mongo/logger/log_severity.h"
+#include "mongo/logger/logger.h"
+#include "mongo/logger/logstream_builder.h"
+#include "mongo/s/chunk_version.h"
+#include "mongo/s/type_chunk.h"
+#include "mongo/util/concurrency/thread_name.h"
+
+namespace mongo {
+
+    template < class ValType, class ShardType >
+    bool ConfigDiffTracker<ValType,ShardType>::
+        isOverlapping( const BSONObj& min, const BSONObj& max )
+    {
+        RangeOverlap overlap = overlappingRange( min, max );
+
+        return overlap.first != overlap.second;
+    }
+
+    template < class ValType, class ShardType >
+    void ConfigDiffTracker<ValType,ShardType>::
+        removeOverlapping( const BSONObj& min, const BSONObj& max )
+    {
+        verifyAttached();
+
+        RangeOverlap overlap = overlappingRange( min, max );
+
+        _currMap->erase( overlap.first, overlap.second );
+    }
+
+    template < class ValType, class ShardType >
+    typename ConfigDiffTracker<ValType,ShardType>::RangeOverlap ConfigDiffTracker<ValType,ShardType>::
+        overlappingRange( const BSONObj& min, const BSONObj& max )
+    {
+        verifyAttached();
+
+        typename RangeMap::iterator low;
+        typename RangeMap::iterator high;
+
+        if( isMinKeyIndexed() ){
+            // Returns the first chunk with a min key that is >= min - implies the
+            // previous chunk cannot overlap min
+            low = _currMap->lower_bound( min );
+            // Returns the first chunk with a min key that is >= max - implies the
+            // chunk does not overlap max
+            high = _currMap->lower_bound( max );
+        }
+        else{
+            // Returns the first chunk with a max key that is > min - implies that
+            // the chunk overlaps min
+            low = _currMap->upper_bound( min );
+            // Returns the first chunk with a max key that is > max - implies that
+            // the next chunk cannot not overlap max
+            high = _currMap->upper_bound( max );
+        }
+
+        return RangeOverlap( low, high );
+    }
+
+    template<class ValType, class ShardType>
+    int ConfigDiffTracker<ValType, ShardType>::calculateConfigDiff(const std::string& config) {
+        verifyAttached();
+
+        // Get the diff query required
+        Query diffQuery = configDiffQuery();
+
+        ScopedDbConnection conn(config, 30.0);
+
+        try {
+
+            // Open a cursor for the diff chunks
+            std::auto_ptr<DBClientCursor> cursor = conn->query(
+                    ChunkType::ConfigNS, diffQuery, 0, 0, 0, 0, ( DEBUG_BUILD ? 2 : 1000000 ) );
+            verify( cursor.get() );
+
+            int diff = calculateConfigDiff( *cursor.get() );
+
+            conn.done();
+
+            return diff;
+        }
+        catch( DBException& e ){
+            // Should only happen on connection errors
+            e.addContext( str::stream() << "could not calculate config difference for ns " << _ns << " on " << config );
+            throw;
+        }
+    }
+
+    template < class ValType, class ShardType >
+    int ConfigDiffTracker<ValType,ShardType>::
+        calculateConfigDiff( DBClientCursorInterface& diffCursor )
+    {
+        verifyAttached();
+
+        // Apply the chunk changes to the ranges and versions
+
+        //
+        // Overall idea here is to work in two steps :
+        // 1. For all the new chunks we find, increment the maximum version per-shard and
+        //    per-collection, and remove any conflicting chunks from the ranges
+        // 2. For all the new chunks we're interested in (all of them for mongos, just chunks on the
+        //    shard for mongod) add them to the ranges
+        //
+
+        std::vector<BSONObj> newTracked;
+        // Store epoch now so it doesn't change when we change max
+        OID currEpoch = _maxVersion->epoch();
+
+        _validDiffs = 0;
+        while( diffCursor.more() ){
+
+            BSONObj diffChunkDoc = diffCursor.next();
+
+            ChunkVersion chunkVersion = ChunkVersion::fromBSON(diffChunkDoc, ChunkType::DEPRECATED_lastmod());
+
+            if( diffChunkDoc[ChunkType::min()].type() != Object ||
+                diffChunkDoc[ChunkType::max()].type() != Object ||
+                diffChunkDoc[ChunkType::shard()].type() != String )
+            {
+                using namespace logger;
+                LogstreamBuilder(globalLogDomain(), getThreadName(), LogSeverity::Warning(),
+                                 LogComponent::kSharding)
+                          << "got invalid chunk document " << diffChunkDoc
+                          << " when trying to load differing chunks" << std::endl;
+                continue;
+            }
+
+            if( ! chunkVersion.isSet() || ! chunkVersion.hasEqualEpoch( currEpoch ) ){
+
+                using namespace logger;
+                LogstreamBuilder(globalLogDomain(), getThreadName(), LogSeverity::Warning(),
+                                 LogComponent::kSharding)
+                          << "got invalid chunk version " << chunkVersion << " in document " << diffChunkDoc
+                          << " when trying to load differing chunks at version "
+                          << ChunkVersion( _maxVersion->majorVersion(),
+                                           _maxVersion->minorVersion(),
+                                           currEpoch ) << std::endl;
+
+                // Don't keep loading, since we know we'll be broken here
+                return -1;
+            }
+
+            _validDiffs++;
+
+            // Get max changed version and chunk version
+            if( chunkVersion > *_maxVersion ) *_maxVersion = chunkVersion;
+
+            // Chunk version changes
+            ShardType shard = shardFor( diffChunkDoc[ChunkType::shard()].String() );
+            typename std::map<ShardType, ChunkVersion>::iterator shardVersionIt = _maxShardVersions->find( shard );
+            if( shardVersionIt == _maxShardVersions->end() || shardVersionIt->second < chunkVersion ){
+                (*_maxShardVersions)[ shard ] = chunkVersion;
+            }
+
+            // See if we need to remove any chunks we are currently tracking b/c of this chunk's changes
+            removeOverlapping(diffChunkDoc[ChunkType::min()].Obj(),
+                              diffChunkDoc[ChunkType::max()].Obj());
+
+            // Figure out which of the new chunks we need to track
+            // Important - we need to actually own this doc, in case the cursor decides to getMore or unbuffer
+            if( isTracked( diffChunkDoc ) ) newTracked.push_back( diffChunkDoc.getOwned() );
+        }
+
+        using namespace logger;
+        if (globalLogDomain()->shouldLog(LogComponent::kSharding, LogSeverity::Debug(3))) {
+            LogstreamBuilder(globalLogDomain(), getThreadName(), LogSeverity::Debug(3),
+                             LogComponent::kSharding)
+                << "found " << _validDiffs
+                << " new chunks for collection " << _ns
+                << " (tracking " << newTracked.size()
+                << "), new version is " << *_maxVersion
+                << std::endl;
+        }
+
+        for( std::vector<BSONObj>::iterator it = newTracked.begin(); it != newTracked.end(); it++ ){
+
+            BSONObj chunkDoc = *it;
+
+            // Important - we need to make sure we actually own the min and max here
+            BSONObj min = chunkDoc[ChunkType::min()].Obj().getOwned();
+            BSONObj max = chunkDoc[ChunkType::max()].Obj().getOwned();
+
+            // Invariant enforced by sharding
+            // It's possible to read inconsistent state b/c of getMore() and yielding, so we want
+            // to detect as early as possible.
+            // TODO: This checks for overlap, we also should check for holes here iff we're tracking
+            // all chunks
+            if( isOverlapping( min, max ) ) return -1;
+
+            _currMap->insert( rangeFor( chunkDoc, min, max ) );
+        }
+
+        return _validDiffs;
+    }
+
+    template<class ValType, class ShardType>
+    Query ConfigDiffTracker<ValType, ShardType>::configDiffQuery() const {
+
+        verifyAttached();
+
+        //
+        // Basic idea behind the query is to find all the chunks $gte the current max version.
+        // Currently, any splits and merges will increment the current max version.
+        //
+
+        BSONObjBuilder queryB;
+        queryB.append(ChunkType::ns(), _ns);
+        BSONObjBuilder tsBuilder(queryB.subobjStart(ChunkType::DEPRECATED_lastmod()));
+        tsBuilder.appendTimestamp( "$gte", _maxVersion->toLong() );
+        tsBuilder.done();
+        BSONObj query = queryB.obj();
+
+        //
+        // NOTE: IT IS IMPORTANT FOR CONSISTENCY THAT WE SORT BY ASC VERSION, TO HANDLE
+        // CURSOR YIELDING BETWEEN CHUNKS BEING MIGRATED.
+        //
+        // This ensures that changes to chunk version (which will always be higher) will always
+        // come *after* our current position in the chunk cursor.
+        //
+
+        Query queryObj(query);
+        queryObj.sort(BSON( "lastmod" << 1 ));
+
+        using namespace logger;
+        if (globalLogDomain()->shouldLog(LogComponent::kSharding, LogSeverity::Debug(2))) {
+            LogstreamBuilder(globalLogDomain(), getThreadName(), LogSeverity::Debug(2),
+                             LogComponent::kSharding)
+                << "major version query from " << *_maxVersion << " and over "
+                << _maxShardVersions->size() << " shards is " << queryObj << std::endl;
+        }
+
+        return queryObj;
+    }
+
+} // namespace mongo
+