ConnectionManager.h

Go to the documentation of this file.

/* -*- c++ -*-
 * Copyright (C) 2007-2015 Hypertable, Inc.
 *
 * This file is part of Hypertable.
 *
 * Hypertable is free software; you can redistribute it and/or
 * modify it under the terms of the GNU General Public License
 * as published by the Free Software Foundation; either version 3
 * of the License, or any later version.
 *
 * Hypertable 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 General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
 * 02110-1301, USA.
 */

#ifndef AsyncComm_ConnectionManager_h
#define AsyncComm_ConnectionManager_h

#include <AsyncComm/Comm.h>
#include <AsyncComm/CommAddress.h>
#include <AsyncComm/ConnectionInitializer.h>
#include <AsyncComm/DispatchHandler.h>

#include <Common/SockAddrMap.h>
#include <Common/Timer.h>

#include <chrono>
#include <condition_variable>
#include <mutex>
#include <queue>
#include <string>
#include <thread>
#include <unordered_map>

namespace Hypertable {

  class Event;

  class ConnectionManager : public DispatchHandler {

    enum class State {
      DISCONNECTED = 0,
      CONNECTED,
      READY,
      DECOMMISSIONED
    };

    class ConnectionState {
    public:
      CommAddress addr;
      CommAddress local_addr;
      InetAddr inet_addr;
      uint32_t timeout_ms;
      DispatchHandlerPtr handler;
      ConnectionInitializerPtr initializer;
      State state {};
      std::mutex mutex;
      std::condition_variable cond;
      std::chrono::steady_clock::time_point next_retry;
      std::string service_name;
    };
    typedef std::shared_ptr<ConnectionState> ConnectionStatePtr;

    struct LtConnectionState {
      bool operator()(const ConnectionStatePtr &cs1,
                      const ConnectionStatePtr &cs2) const {
        return std::chrono::operator>(cs1->next_retry, cs2->next_retry);
      }
    };

    class SharedImpl {
    public:

      ~SharedImpl() {
        shutdown = true;
        retry_cond.notify_one();
        if (thread.joinable())
          thread.join();
      }

      Comm *comm;
      std::mutex mutex;
      std::condition_variable retry_cond;
      std::thread thread;
      SockAddrMap<ConnectionStatePtr> conn_map;
      std::unordered_map<String, ConnectionStatePtr> conn_map_proxy;
      std::priority_queue<ConnectionStatePtr, std::vector<ConnectionStatePtr>,
          LtConnectionState> retry_queue;
      bool quiet_mode;
      bool shutdown;
    };

    typedef std::shared_ptr<SharedImpl> SharedImplPtr;

  public:

    ConnectionManager(Comm *comm = 0) {
      m_impl = std::make_shared<SharedImpl>();
      m_impl->comm = comm ? comm : Comm::instance();
      m_impl->quiet_mode = false;
      m_impl->shutdown = false;
    }

    ConnectionManager(const ConnectionManager &cm) = delete;

    virtual ~ConnectionManager() { }

    void add(const CommAddress &addr, uint32_t timeout_ms,
             const char *service_name);

    void add(const CommAddress &addr, uint32_t timeout_ms,
             const char *service_name, DispatchHandlerPtr &handler);

    void add_with_initializer(const CommAddress &addr, uint32_t timeout_ms,
                              const char *service_name,
                              DispatchHandlerPtr &handler,
                              ConnectionInitializerPtr &initializer);

    void add(const CommAddress &addr, const CommAddress &local_addr,
             uint32_t timeout_ms, const char *service_name);

    void add(const CommAddress &addr, const CommAddress &local_addr,
             uint32_t timeout_ms, const char *service_name,
             DispatchHandlerPtr &handler);

    int remove(const CommAddress &addr);

    bool wait_for_connection(const CommAddress &addr, uint32_t max_wait_ms);

    bool wait_for_connection(const CommAddress &addr, Timer &timer);

    Comm *get_comm() { return m_impl->comm; }

    void set_quiet_mode(bool mode) { m_impl->quiet_mode = mode; }

    virtual void handle(EventPtr &event);

    void connect_retry_loop();

  private:

    void add_internal(const CommAddress &addr, const CommAddress &local_addr,
                      uint32_t timeout_ms, const char *service_name,
                      DispatchHandlerPtr &handler,
                      ConnectionInitializerPtr &initializer);

    bool wait_for_connection(ConnectionStatePtr &conn_state, Timer &timer);

    void send_connect_request(ConnectionStatePtr &conn_state);

    void send_initialization_request(ConnectionStatePtr &conn_state);

    void schedule_retry(ConnectionStatePtr &conn_state, const std::string &message);

    SharedImplPtr m_impl;

  };
  typedef std::shared_ptr<ConnectionManager> ConnectionManagerPtr;

}

#endif // AsyncComm_ConnectionManager_h