replica_set_monitor.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/shared_ptr.hpp>
#include <string>
#include <set>
 
#include "mongo/base/disallow_copying.h"
#include "mongo/base/string_data.h"
#include "mongo/stdx/functional.h"
#include "mongo/util/net/hostandport.h"
 
namespace mongo {
    class BSONObj;
    class ReplicaSetMonitor;
    class TagSet;
    struct ReadPreferenceSetting;
    typedef boost::shared_ptr<ReplicaSetMonitor> ReplicaSetMonitorPtr;
 
    class ReplicaSetMonitor {
        MONGO_DISALLOW_COPYING(ReplicaSetMonitor);
    public:
        class Refresher;
 
        typedef stdx::function<void(const std::string& setName,
                                     const std::string& newConnectionString)>
                ConfigChangeHook;
 
        ReplicaSetMonitor(StringData name, const std::set<HostAndPort>& seeds);
 
        HostAndPort getHostOrRefresh(const ReadPreferenceSetting& criteria);
 
        HostAndPort getMasterOrUassert();
 
        Refresher startOrContinueRefresh();
 
        void failedHost(const HostAndPort& host);
 
        bool isPrimary(const HostAndPort& host) const;
 
        bool isHostUp(const HostAndPort& host) const;
 
        int getConsecutiveFailedScans() const;
 
        std::string getName() const;
 
        std::string getServerAddress() const;
 
        bool contains(const HostAndPort& server) const;
 
        void appendInfo(BSONObjBuilder& b) const;
 
        static void createIfNeeded(const std::string& name , const std::set<HostAndPort>& servers);
 
        static ReplicaSetMonitorPtr get(const std::string& name, bool createFromSeed = false);
 
        static std::set<std::string> getAllTrackedSets();
 
        static void remove(const std::string& name, bool clearSeedCache = false);
 
        static void setConfigChangeHook(ConfigChangeHook hook);
 
        static Status initialize();
 
        static Status shutdown(int gracePeriodMillis = 0);
 
        static int maxConsecutiveFailedChecks;
 
        //
        // internal types (defined in replica_set_monitor_internal.h)
        //
 
        struct IsMasterReply;
        struct ScanState;
        struct SetState;
        typedef boost::shared_ptr<ScanState> ScanStatePtr;
        typedef boost::shared_ptr<SetState> SetStatePtr;
 
        //
        // FOR TESTING ONLY
        //
 
        explicit ReplicaSetMonitor(const SetStatePtr& initialState)
            : _state(initialState)
        {}
 
        static bool useDeterministicHostSelection;
 
    private:
        const SetStatePtr _state; // never NULL
    };
 
    class ReplicaSetMonitor::Refresher {
    public:
        HostAndPort refreshUntilMatches(const ReadPreferenceSetting& criteria) {
            return _refreshUntilMatches(&criteria);
        };
 
        void refreshAll() { _refreshUntilMatches(NULL); }
 
        //
        // Remaining methods are only for testing and internal use.
        // Callers are responsible for holding SetState::mutex before calling any of these methods.
        //
 
        explicit Refresher(const SetStatePtr& setState);
 
        struct NextStep {
            enum StepKind {
                CONTACT_HOST, 
                WAIT, 
                DONE, 
            };
 
            explicit NextStep(StepKind step, const HostAndPort& host=HostAndPort())
                : step(step)
                , host(host)
            {}
 
            StepKind step;
            HostAndPort host;
        };
 
        NextStep getNextStep();
 
        void receivedIsMaster(const HostAndPort& from, int64_t latencyMicros, const BSONObj& reply);
 
        void failedHost(const HostAndPort& host);
 
        bool startedNewScan() const { return _startedNewScan; }
 
        static ScanStatePtr startNewScan(const SetState* set);
 
    private:
 
        bool receivedIsMasterFromMaster(const IsMasterReply& reply);
 
        void receivedIsMasterBeforeFoundMaster(const IsMasterReply& reply);
 
        HostAndPort _refreshUntilMatches(const ReadPreferenceSetting* criteria);
 
        // Both pointers are never NULL
        SetStatePtr _set;
        ScanStatePtr _scan; // May differ from _set->currentScan if a new scan has started.
        bool _startedNewScan;
    };
}