Session.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 Hyperspace_Session_h
#define Hyperspace_Session_h

#include <Hyperspace/ClientKeepaliveHandler.h>
#include <Hyperspace/DirEntry.h>
#include <Hyperspace/DirEntryAttr.h>
#include <Hyperspace/HandleCallback.h>
#include <Hyperspace/HsCommandInterpreter.h>
#include <Hyperspace/LockSequencer.h>
#include <Hyperspace/Protocol.h>

#include <AsyncComm/Comm.h>
#include <AsyncComm/CommBuf.h>
#include <AsyncComm/ConnectionManager.h>
#include <AsyncComm/DispatchHandler.h>

#include <Common/DynamicBuffer.h>
#include <Common/Properties.h>
#include <Common/Status.h>
#include <Common/String.h>
#include <Common/Timer.h>

#include <chrono>
#include <condition_variable>
#include <memory>
#include <mutex>
#include <unordered_map>
#include <vector>

namespace Hypertable {
  class SleepWakeNotifier;
}

namespace Hyperspace {

  using namespace std;
  class HsCommandInterpreter;
  enum {
    OPEN_FLAG_READ           = 0x00001,
    OPEN_FLAG_WRITE          = 0x00002,
    OPEN_FLAG_LOCK           = 0x00004,
    OPEN_FLAG_CREATE         = 0x00008,
    OPEN_FLAG_EXCL           = 0x00010,
    OPEN_FLAG_TEMP           = 0x00020,
    OPEN_FLAG_LOCK_SHARED    = 0x00044,
    OPEN_FLAG_LOCK_EXCLUSIVE = 0x00084
  };

  class SessionCallback {
  public:
    SessionCallback() : m_id(0){}
    virtual ~SessionCallback() { return; }
    virtual void safe() = 0;
    virtual void expired() = 0;
    virtual void jeopardy() = 0;
    virtual void disconnected() = 0;
    virtual void reconnected() = 0;
    void set_id(uint32_t id) {m_id = id;}
    uint32_t get_id() {return m_id;}
  private:
    uint32_t m_id;
  };

  class Session {

  public:

    enum State {
      STATE_EXPIRED,
      STATE_JEOPARDY,
      STATE_SAFE,
      STATE_DISCONNECTED
    };

    enum Locate {
      LOCATE_MASTER=1,
      LOCATE_REPLICAS
    };

    Session(Comm *comm, PropertiesPtr &props);

    virtual ~Session();

    void add_callback(SessionCallback *cb);

    bool remove_callback(SessionCallback *cb);

    uint64_t open(const std::string &name, uint32_t flags,
                  HandleCallbackPtr &callback, Timer *timer = 0);

    uint64_t open(const std::string &name, uint32_t flags, Timer *timer = 0);

    uint64_t create(const std::string &name, uint32_t flags,
                    HandleCallbackPtr &callback,
                    const std::vector<Attribute> &init_attrs, Timer *timer=0);

    /*
    int cancel(uint64_t handle);
    int poison(uint64_t handle);
    */

    void close(uint64_t handle, Timer *timer=0);

    void close_nowait(uint64_t handle);

    void mkdir(const std::string &name, Timer *timer=0);

    void mkdir(const std::string &name, const std::vector<Attribute> &init_attrs, Timer *timer=0);

    void mkdirs(const std::string &name, Timer *timer=0);

    void mkdirs(const std::string &name, const std::vector<Attribute> &init_attrs, Timer *timer=0);

    void attr_set(uint64_t handle, const std::string &attr,
                  const void *value, size_t value_len, Timer *timer=0);

    void attr_set(uint64_t handle, const std::vector<Attribute> &attrs,
                  Timer *timer=0);

    void attr_set(const std::string &name, const std::string &attr,
                  const void *value, size_t value_len, Timer *timer=0);

    void attr_set(const std::string &name, uint32_t oflags, const std::string &attr,
                  const void *value, size_t value_len, Timer *timer=0);

    void attr_set(const std::string &name, uint32_t oflags,
                  const std::vector<Attribute> &attrs, Timer *timer=0);

    uint64_t attr_incr(uint64_t handle, const std::string &attr, Timer *timer=0);

    uint64_t attr_incr(const std::string &name, const std::string &attr, Timer *timer=0);

    void attr_list(uint64_t handle, vector<String> &anames, Timer *timer=0);

    bool attr_exists(uint64_t handle, const std::string& attr, Timer *timer=0);
    bool attr_exists(const std::string &name, const std::string& attr, Timer *timer=0);

    void attr_get(uint64_t handle, const std::string &attr,
                  DynamicBuffer &value, Timer *timer=0);

    void attr_get(const std::string &name, const std::string &attr,
                  DynamicBuffer &value, Timer *timer=0);

    void attr_get(const std::string &name, const std::string &attr,
                  bool& attr_exists, DynamicBuffer &value, Timer *timer=0);

    void attrs_get(uint64_t handle, const std::vector<std::string> &attrs,
                  std::vector<DynamicBufferPtr> &values, Timer *timer=0);

    void attrs_get(const std::string &name, const std::vector<std::string> &attrs,
                  std::vector<DynamicBufferPtr> &values, Timer *timer=0);

    void attr_del(uint64_t handle, const std::string &name, Timer *timer=0);

    bool exists(const std::string &name, Timer *timer=0);

    void unlink(const std::string &name, Timer *timer=0);

    void readdir(uint64_t handle, std::vector<DirEntry> &listing,
                 Timer *timer=0);

    void readdir_attr(uint64_t handle, const std::string &attr, bool include_sub_entries,
                      std::vector<DirEntryAttr> &listing, Timer *timer=0);

    void readdir_attr(const std::string &name, const std::string &attr, bool include_sub_entries,
                      std::vector<DirEntryAttr> &listing, Timer *timer=0);

    void readpath_attr(uint64_t handle, const std::string &attr,
                       std::vector<DirEntryAttr> &listing, Timer *timer=0);

    void readpath_attr(const std::string &name, const std::string &attr,
                       std::vector<DirEntryAttr> &listing, Timer *timer=0);

    void lock(uint64_t handle, LockMode mode,
              LockSequencer *sequencerp, Timer *timer=0);

    void try_lock(uint64_t handle, LockMode mode, LockStatus *statusp,
                  LockSequencer *sequencerp, Timer *timer=0);

    void release(uint64_t handle, Timer *timer=0);

    void get_sequencer(uint64_t handle, LockSequencer *sequencerp,
                       Timer *timer=0);

    void check_sequencer(LockSequencer &sequencer, Timer *timer=0);


    String locate(int type);

    int status(Status &status, Timer *timer=0);

    bool wait_for_connection(uint32_t max_wait_ms);

    bool wait_for_connection(Timer &timer);

    void set_verbose_flag(bool verbose) { m_verbose = verbose; }

    int state_transition(int state);

    int get_state();

    bool expired();

     HsCommandInterpreterPtr create_hs_interpreter();

    void advance_expire_time(std::chrono::steady_clock::time_point now) {
      std::lock_guard<std::mutex> lock(m_mutex);
      m_expire_time = now + std::chrono::milliseconds(m_lease_interval);
    }

    void update_master_addr(const String &host);

    void handle_sleep();

    void handle_wakeup();

    void shutdown(Timer *timer=0);

  private:

    typedef std::unordered_map<uint64_t, SessionCallback *> CallbackMap;

    void mkdir(const std::string &name, bool create_intermediate, const std::vector<Attribute> *init_attrs, Timer *timer);
    void decode_listing(Hypertable::EventPtr& event_ptr, std::vector<DirEntryAttr> &listing);
    void decode_value(Hypertable::EventPtr& event_ptr, DynamicBuffer &value);
    void decode_values(Hypertable::EventPtr& event_ptr, std::vector<DynamicBufferPtr> &values);
    bool wait_for_safe();
    int send_message(CommBufPtr &, DispatchHandler *, Timer *timer);
    void normalize_name(const std::string &name, std::string &normal);
    uint64_t open(ClientHandleStatePtr &, CommBufPtr &, Timer *timer);

    std::mutex m_mutex;
    std::condition_variable m_cond;
    Comm                      *m_comm;
    PropertiesPtr             m_cfg;
    bool                      m_verbose;
    bool                      m_silent;
    bool                      m_reconnect;
    uint16_t                  m_hyperspace_port;
    int                       m_state;
    uint32_t                  m_grace_period;
    uint32_t                  m_lease_interval;
    uint32_t                  m_timeout_ms;
    std::chrono::steady_clock::time_point m_expire_time;
    InetAddr                  m_master_addr;
    ClientKeepaliveHandlerPtr m_keepalive_handler_ptr;
    CallbackMap               m_callbacks;
    uint64_t                  m_last_callback_id;
    std::mutex m_callback_mutex;
    vector<String>            m_hyperspace_replicas;
    String                    m_hyperspace_master;

    SleepWakeNotifier *m_sleep_wake_notifier;
  };

  typedef std::shared_ptr<Session> SessionPtr;

  void close_handle(SessionPtr hyperspace, uint64_t handle);
  void close_handle_ptr(SessionPtr hyperspace, uint64_t *handlep);

} // namespace Hyperspace

#endif // Hyperspace_Session_h