path: root/src/mongo/client/syncclusterconnection.h

// @file syncclusterconnection.h

/*
 *    Copyright 2010 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/bson/bsonelement.h"
#include "mongo/bson/bsonobj.h"
#include "mongo/client/dbclientinterface.h"
#include "mongo/stdx/mutex.h"
#include "mongo/util/concurrency/mutex.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:
    using DBClientBase::query;
    using DBClientBase::update;
    using DBClientBase::remove;

    class QueryHandler;

    /**
     * @param commaSeparated should be 3 hosts comma separated
     */
    SyncClusterConnection(const std::list<HostAndPort>&, double socketTimeout = 0);
    SyncClusterConnection(std::string commaSeparated, double socketTimeout = 0);
    SyncClusterConnection(const std::string& a,
                          const std::string& b,
                          const std::string& c,
                          double socketTimeout = 0);
    ~SyncClusterConnection();

    /**
     * @return true if all servers are up and ready for writes
     */
    bool prepare(std::string& errmsg);

    // --- from DBClientInterface

    virtual BSONObj findOne(const std::string& ns,
                            const Query& query,
                            const BSONObj* fieldsToReturn,
                            int queryOptions);

    virtual std::unique_ptr<DBClientCursor> query(const std::string& ns,
                                                  Query query,
                                                  int nToReturn,
                                                  int nToSkip,
                                                  const BSONObj* fieldsToReturn,
                                                  int queryOptions,
                                                  int batchSize);

    virtual std::unique_ptr<DBClientCursor> getMore(const std::string& ns,
                                                    long long cursorId,
                                                    int nToReturn,
                                                    int options);

    virtual void insert(const std::string& ns, BSONObj obj, int flags = 0);

    virtual void insert(const std::string& ns, const std::vector<BSONObj>& v, int flags = 0);

    virtual void remove(const std::string& ns, Query query, int flags);

    virtual void update(const std::string& ns, Query query, BSONObj obj, int flags);

    virtual bool call(Message& toSend, Message& response, bool assertOk, std::string* actualServer);
    virtual void say(Message& toSend, bool isRetry = false, std::string* actualServer = 0);
    virtual void sayPiggyBack(Message& toSend);

    virtual void killCursor(long long cursorID);

    virtual std::string getServerAddress() const {
        return _address;
    }
    virtual bool isFailed() const {
        return false;
    }
    virtual bool isStillConnected();
    virtual std::string toString() const {
        return _toString();
    }

    virtual BSONObj getLastErrorDetailed(
        const std::string& db, bool fsync = false, bool j = false, int w = 0, int wtimeout = 0);
    virtual BSONObj getLastErrorDetailed(bool fsync = false,
                                         bool j = false,
                                         int w = 0,
                                         int wtimeout = 0);

    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 lazySupported() const {
        return false;
    }

    // This override of runCommand intentionally calls findOne to construct a legacy
    // OP_QUERY command. The reason for this is that delicate logic for targeting/locking
    // config servers is in SyncClusterConnection::findOne, and refactoring that logic
    // is both risky and of dubious value as we move to config server replica sets (CSRS).
    bool runCommand(const std::string& dbname,
                    const BSONObj& cmd,
                    BSONObj& info,
                    int options) final;

    void setRequestMetadataWriter(rpc::RequestMetadataWriter writer) final;
    void setReplyMetadataReader(rpc::ReplyMetadataReader reader) final;

    /**
     * Allow custom query processing through an external (e.g. mongos-only) service.
     *
     * Takes ownership of attached handler.
     */
    void attachQueryHandler(QueryHandler* handler);

protected:
    virtual void _auth(const BSONObj& params);

private:
    SyncClusterConnection(SyncClusterConnection& prev, double socketTimeout = 0);
    std::string _toString() const;
    bool _commandOnActive(const std::string& dbname,
                          const BSONObj& cmd,
                          BSONObj& info,
                          int options = 0);
    std::unique_ptr<DBClientCursor> _queryOnActive(const std::string& ns,
                                                   Query query,
                                                   int nToReturn,
                                                   int nToSkip,
                                                   const BSONObj* fieldsToReturn,
                                                   int queryOptions,
                                                   int batchSize);
    int _lockType(const std::string& name);
    void _checkLast();
    void _connect(const std::string& host);

    std::string _address;
    std::vector<std::string> _connAddresses;
    std::vector<DBClientConnection*> _conns;

    std::vector<BSONObj> _lastErrors;

    // Optionally attached by user
    std::unique_ptr<QueryHandler> _customQueryHandler;

    stdx::mutex _mutex;
    std::map<std::string, int> _lockTypes;
    // End mutex

    double _socketTimeout;
};

/**
 * Interface for custom query processing for the SCC.
 * Allows plugging different host query behaviors for different types of queries.
 */
class SyncClusterConnection::QueryHandler {
public:
    virtual ~QueryHandler(){};

    /**
     * Returns true if the query can be processed using this handler.
     */
    virtual bool canHandleQuery(const std::string& ns, Query query) = 0;

    /**
     * Returns a cursor on one of the hosts with the desired results for the query.
     * May throw or return an empty unique_ptr on failure.
     */
    virtual std::unique_ptr<DBClientCursor> handleQuery(const std::vector<std::string>& hosts,
                                                        const std::string& ns,
                                                        Query query,
                                                        int nToReturn,
                                                        int nToSkip,
                                                        const BSONObj* fieldsToReturn,
                                                        int queryOptions,
                                                        int batchSize) = 0;
};

class UpdateNotTheSame : public UserException {
public:
    UpdateNotTheSame(int code,
                     const std::string& msg,
                     const std::vector<std::string>& addrs,
                     const std::vector<BSONObj>& lastErrors)
        : UserException(code, msg), _addrs(addrs), _lastErrors(lastErrors) {
        verify(_addrs.size() == _lastErrors.size());
    }

    virtual ~UpdateNotTheSame() throw() {}

    unsigned size() const {
        return _addrs.size();
    }

    std::pair<std::string, BSONObj> operator[](unsigned i) const {
        return std::make_pair(_addrs[i], _lastErrors[i]);
    }

private:
    std::vector<std::string> _addrs;
    std::vector<BSONObj> _lastErrors;
};
};