replica_set_monitor_internal.h

/*    Copyright 2014 MongoDB 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 <boost/thread/condition_variable.hpp>
#include <deque>
#include <set>
#include <string>
#include <vector>

#include "mongo/base/disallow_copying.h"
#include "mongo/client/dbclient_rs.h" // for TagSet and ReadPreferenceSettings
#include "mongo/client/replica_set_monitor.h"
#include "mongo/db/jsobj.h"
#include "mongo/platform/cstdint.h"
#include "mongo/platform/random.h"
#include "mongo/platform/unordered_map.h"
#include "mongo/util/net/hostandport.h"

namespace mongo {
    // Internal connection cache for isMaster. Used to amortize the cost of establishing connections
    // to a remote host. Not really a pool because we only keep one connection per host.
    class ConnectionCache {
    public:
        // Uses a cached connection (may need to reconnect) to execute a remote isMaster call
        // on the given host. Returns microseconds taken for remote call to complete. Throws.
        uint64_t timedIsMaster(const HostAndPort& host, BSONObj* out);

    private:
        typedef unordered_map<HostAndPort, boost::shared_ptr<DBClientConnection> >
            ConnectionMap;

        // Guard that creates a connection if needed, or removes it from
        // the cache. This is so multiple isMaster calls can be in flight in the same
        // set simultaneously.
        class ConnectionGuard {
        public:
            // Lifetime of guard must be subset of lifetime of host
            ConnectionGuard(ConnectionCache& cache,
                            const HostAndPort& host)
                : _cache(cache)
                , _host(host)
            {
                // this could throw.
                _conn = _cache.getConnectionTo(_host);
            }

            ~ConnectionGuard() {
                _cache.returnConnection(_host, _conn);
            }

            boost::shared_ptr<DBClientConnection> _conn;
            // Reference to owner.
            ConnectionCache& _cache;
            const HostAndPort& _host;
        };

        // If we don't have a connection to this host cached,
        // or we have one that is dead, we reconnect and return
        // If we have one, we remove it from the cache and return it
        boost::shared_ptr<DBClientConnection> getConnectionTo(const HostAndPort& host);

        void returnConnection(const HostAndPort& host, boost::shared_ptr<DBClientConnection> conn);

        inline bool connectionOk(const boost::shared_ptr<DBClientConnection>& conn) {
            // checks are ordered from cheap to expensive
            return conn && !conn->isFailed() && conn->isStillConnected();
        }

        ConnectionMap _cacheStorage;
        boost::mutex _cacheLock;
    };

    struct ReplicaSetMonitor::IsMasterReply {
        IsMasterReply() : ok(false) {}
        IsMasterReply(const HostAndPort& host, int64_t latencyMicros, const BSONObj& reply)
                : ok(false)
                , host(host)
                , latencyMicros(latencyMicros) {
            parse(reply);
        }

        void parse(const BSONObj& obj);

        bool ok; // if false, ignore all other fields
        BSONObj raw; // Always owned. Other fields are allowed to be a view into this.
        std::string setName;
        bool isMaster;
        bool secondary;
        bool hidden;
        HostAndPort primary; // empty if not present
        std::set<HostAndPort> normalHosts; // both "hosts" and "passives"
        BSONObj tags;

        // remaining fields aren't in isMaster reply, but are known to caller.
        HostAndPort host;
        int64_t latencyMicros; // ignored if negative
    };

    struct ReplicaSetMonitor::SetState {
        MONGO_DISALLOW_COPYING(SetState);
    public:

        // A single node in the replicaSet
        struct Node {
            explicit Node(const HostAndPort& host)
                    : host(host)
                    , latencyMicros(unknownLatency) {
                markFailed();
            }

            void markFailed() {
                isUp = false;
                isMaster = false;
            }

            bool matches(const ReadPreference& pref) const;

            bool matches(const BSONObj& tag) const;

            void update(const IsMasterReply& reply);

            // Intentionally chosen to compare worse than all known latencies.
            static const int64_t unknownLatency; // = numeric_limits<int64_t>::max()

            HostAndPort host;
            bool isUp;
            bool isMaster; // implies isUp
            int64_t latencyMicros; // unknownLatency if unknown
            BSONObj tags; // owned
        };
        typedef std::vector<Node> Nodes;

        SetState(StringData name, const std::set<HostAndPort>& seedNodes);

        HostAndPort getMatchingHost(const ReadPreferenceSetting& criteria) const;

        Node* findNode(const HostAndPort& host);

        Node* findOrCreateNode(const HostAndPort& host);

        void updateNodeIfInNodes(const IsMasterReply& reply);

        std::string getServerAddress() const;

        void checkInvariants() const;

        static ConfigChangeHook configChangeHook;

        boost::mutex mutex; // must hold this to access any other member or method (except name).

        // If Refresher::getNextStep returns WAIT, you should wait on the condition_variable,
        // releasing mutex. It will be notified when either getNextStep will return something other
        // than WAIT, or a new host is available for consideration by getMatchingHost. Essentially,
        // this will be hit whenever the _refreshUntilMatches loop has the potential to make
        // progress.
        // TODO consider splitting cv into two: one for when looking for a master, one for all other
        // cases.
        boost::condition_variable cv;

        const std::string name; // safe to read outside lock since it is const
        int consecutiveFailedScans;
        std::set<HostAndPort> seedNodes; // updated whenever a master reports set membership changes
        HostAndPort lastSeenMaster; // empty if we have never seen a master. can be same as current
        Nodes nodes; // maintained sorted and unique by host
        ScanStatePtr currentScan; // NULL if no scan in progress
        int64_t latencyThresholdMicros;
        mutable PseudoRandom rand; // only used for host selection to balance load
        mutable int roundRobin; // used when useDeterministicHostSelection is true
        mutable ConnectionCache connectionCache;
    };

    struct ReplicaSetMonitor::ScanState {
        MONGO_DISALLOW_COPYING(ScanState);
    public:
        ScanState() : foundUpMaster(false), foundAnyUpNodes(false) {}

        template <typename Container>
        void enqueAllUntriedHosts(const Container& container, PseudoRandom& rand);

        // Access to fields is guarded by associated SetState's mutex.
        bool foundUpMaster;
        bool foundAnyUpNodes;
        std::deque<HostAndPort> hostsToScan; // Work queue.
        std::set<HostAndPort> possibleNodes; // Nodes reported by non-primary hosts.
        std::set<HostAndPort> waitingFor; // Hosts we have dispatched but haven't replied yet.
        std::set<HostAndPort> triedHosts; // Hosts that have been returned from getNextStep.

        // All responses go here until we find a master.
        typedef std::vector<IsMasterReply> UnconfirmedReplies;
        UnconfirmedReplies unconfirmedReplies;
    };
}