Operation.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; 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 Hypertable_Master_Operation_h
#define Hypertable_Master_Operation_h

#include "Context.h"
#include "MetaLogEntityTypes.h"

#include <Hypertable/Lib/MetaLogEntity.h>

#include <AsyncComm/Clock.h>
#include <AsyncComm/Event.h>

#include <Common/ScopeGuard.h>

#include <ctime>
#include <set>

namespace Hypertable {


  namespace OperationState {

    enum {
      INITIAL = 0,
      COMPLETE = 1,
      UNUSED = 2,
      STARTED = 3,
      ASSIGN_ID = 4,
      ASSIGN_LOCATION = 5,
      ASSIGN_METADATA_RANGES = 6,
      LOAD_RANGE = 7,
      LOAD_ROOT_METADATA_RANGE = 8,
      LOAD_SECOND_METADATA_RANGE = 9,
      WRITE_METADATA = 10,
      CREATE_RS_METRICS = 11,
      VALIDATE_SCHEMA = 12,
      SCAN_METADATA = 13,
      ISSUE_REQUESTS = 14,
      UPDATE_HYPERSPACE = 15,
      ACKNOWLEDGE = 16,
      FINALIZE = 17,
      CREATE_INDEX = 18,
      CREATE_QUALIFIER_INDEX = 19,
      PREPARE = 20,
      COMMIT = 21,
      PHANTOM_LOAD = 22,
      REPLAY_FRAGMENTS = 23,
      CREATE_INDICES = 24,
      DROP_INDICES = 25,
      SUSPEND_TABLE_MAINTENANCE = 26,
      RESUME_TABLE_MAINTENANCE = 27,
      DROP_VALUE_INDEX = 28,
      DROP_QUALIFIER_INDEX = 29,
      RENAME_VALUE_INDEX = 30,
      RENAME_QUALIFIER_INDEX = 31
    };
    const char *get_text(int state);
  }

  namespace Dependency {
    extern const char *INIT;
    extern const char *SERVERS;
    extern const char *ROOT;
    extern const char *METADATA;
    extern const char *SYSTEM;
    extern const char *USER;
    extern const char *RECOVER_SERVER;
    extern const char *RECOVERY_BLOCKER;
    extern const char *RECOVERY;
  }

  typedef std::set<String> DependencySet;

  class Operation : public MetaLog::Entity {
  public:

    Operation(ContextPtr &context, int32_t type);

    Operation(ContextPtr &context, EventPtr &event, int32_t type);

    Operation(ContextPtr &context, const MetaLog::EntityHeader &header_);

    virtual ~Operation() { }

    virtual void execute() = 0;

    const String name() override = 0;

    virtual const String label() = 0;

    virtual const String graphviz_label() { return label(); }

    virtual bool exclusive() { return false; }

    virtual size_t encoded_result_length() const;

    virtual void encode_result(uint8_t **bufp) const;

    virtual void decode_result(const uint8_t **bufp, size_t *remainp);

    virtual uint8_t encoding_version_state() const = 0;

    virtual size_t encoded_length_state() const = 0;

    virtual void encode_state(uint8_t **bufp) const = 0;

    virtual void decode_state(uint8_t version, const uint8_t **bufp,
                              size_t *remainp) = 0;

    virtual void decode_state_old(uint8_t version, const uint8_t **bufp,
                                  size_t *remainp) = 0;

    virtual void display_state(std::ostream &os) = 0;

    void decode(const uint8_t **bufp, size_t *remainp,
                uint16_t definition_version) override;

    void display(std::ostream &os) override;

    int64_t id() { return header.id; }

    ClockT::time_point expiration_time() { std::lock_guard<std::mutex> lock(m_mutex); return m_expiration_time; }

    void set_remove_approval_mask(uint16_t mask) {
      std::lock_guard<std::mutex> lock(m_mutex);
      m_remove_approval_mask = mask;
    }

    uint16_t get_remove_approval_mask() {
      std::lock_guard<std::mutex> lock(m_mutex);
      return m_remove_approval_mask;
    }

    bool remove_approval_add(uint16_t approval) {
      std::lock_guard<std::mutex> lock(m_mutex);
      m_remove_approvals |= approval;
      return m_remove_approvals == m_remove_approval_mask;
    }

    bool removal_approved();

    void record_state(std::vector<MetaLog::EntityPtr> &additional);

    void record_state() {
      std::vector<MetaLog::EntityPtr> additional;
      record_state(additional);
    }

    void complete_error(int error, const String &msg, std::vector<MetaLog::EntityPtr> &additional);

    void complete_error(int error, const String &msg, MetaLog::EntityPtr additional=MetaLog::EntityPtr());

    void complete_error(Exception &e, std::vector<MetaLog::EntityPtr> &additional) {
      complete_error(e.code(), e.what(), additional);
    }

    void complete_error(Exception &e, MetaLog::EntityPtr additional=MetaLog::EntityPtr()) {
      complete_error(e.code(), e.what(), additional);
    }

    void complete_ok(std::vector<MetaLog::EntityPtr> &additional);
    void complete_ok(MetaLog::EntityPtr additional=MetaLog::EntityPtr());

    virtual int64_t hash_code() const { return m_hash_code; }

    virtual void exclusivities(DependencySet &exclusivities);
    virtual void dependencies(DependencySet &dependencies);
    virtual void obstructions(DependencySet &obstructions);

    void add_exclusivity(const String &exclusivity) { m_exclusivities.insert(exclusivity); }
    void add_dependency(const String &dependency) { m_dependencies.insert(dependency); }
    void add_obstruction(const String &obstruction) { m_obstructions.insert(obstruction); }
    void add_obstruction_permanent(const String &obstruction) {
      m_obstructions_permanent.insert(obstruction);
    }

    void fetch_sub_operations(std::vector<std::shared_ptr<Operation> > &sub_ops);

    void pre_run();
    void post_run();
    int32_t get_state() { std::lock_guard<std::mutex> lock(m_mutex); return m_state; }
    void set_state(int32_t state) { std::lock_guard<std::mutex> lock(m_mutex); m_state = state; }
    virtual bool is_perpetual() { return false; }
    bool block();
    bool unblock();
    bool is_blocked() { std::lock_guard<std::mutex> lock(m_mutex); return m_blocked; }
    bool is_complete() { std::lock_guard<std::mutex> lock(m_mutex); return m_state == OperationState::COMPLETE; }

    int32_t get_original_type() { return m_original_type; }
    void set_original_type(int32_t original_type) { m_original_type = original_type; }

    void set_ephemeral() {
      std::lock_guard<std::mutex> lock(m_mutex);
      m_ephemeral = true;
    }

    bool ephemeral() {
      std::lock_guard<std::mutex> lock(m_mutex);
      return m_ephemeral;
    }

    int32_t get_error() {
      std::lock_guard<std::mutex> lock(m_mutex);
      HT_ASSERT(m_state == OperationState::COMPLETE);
      return m_error;
    }

    String get_error_msg() {
      std::lock_guard<std::mutex> lock(m_mutex);
      HT_ASSERT(m_state == OperationState::COMPLETE);
      return m_error_msg;
    }

  protected:

    uint8_t encoding_version() const override;

    size_t encoded_length_internal() const override;

    void encode_internal(uint8_t **bufp) const override;

    void decode_internal(uint8_t version, const uint8_t **bufp,
                         size_t *remainp) override;

    void decode_old(const uint8_t **bufp, size_t *remainp,
                    uint16_t definition_version);

    bool validate_subops();

    void stage_subop(std::shared_ptr<Operation> operation);

    ContextPtr m_context;

    EventPtr m_event;

    int32_t m_state {OperationState::INITIAL};

    uint16_t m_remove_approvals {};

    uint16_t m_remove_approval_mask {};

    int32_t m_original_type {};

    int32_t m_error {};

    String m_error_msg;

    bool m_unblock_on_exit {};

    bool m_blocked {};

    bool m_ephemeral {};

    ClockT::time_point m_expiration_time;

    int64_t m_hash_code;

    DependencySet m_exclusivities;

    DependencySet m_dependencies;

    DependencySet m_obstructions;

    DependencySet m_obstructions_permanent;

    std::vector<int64_t> m_sub_ops;

  };

  typedef std::shared_ptr<Operation> OperationPtr;


} // namespace Hypertable

#endif // Hypertable_Master_Operation_h