TableMutatorAsync.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_Lib_TableMutatorAsync_h
#define Hypertable_Lib_TableMutatorAsync_h

#include "Cells.h"
#include "ClientObject.h"
#include "KeySpec.h"
#include "Table.h"
#include "TableMutatorAsyncScatterBuffer.h"
#include "RangeLocator.h"
#include "Schema.h"
#include "TableIdentifier.h"

#include <AsyncComm/ConnectionManager.h>

#include <Common/Properties.h>
#include <Common/StringExt.h>
#include <Common/Timer.h>

#include <condition_variable>
#include <iostream>
#include <mutex>

namespace Hypertable {

  class TableMutatorAsync;
  typedef std::shared_ptr<TableMutatorAsync> TableMutatorAsyncPtr;

  class IndexMutatorCallback;
  typedef std::shared_ptr<IndexMutatorCallback> IndexMutatorCallbackPtr;

  class TableMutator;

  class TableMutatorAsync : public ClientObject {

  public:

    TableMutatorAsync(PropertiesPtr &props, Comm *comm,
                      ApplicationQueueInterfacePtr &app_queue, Table *table,
                      RangeLocatorPtr &range_locator, uint32_t timeout_ms, ResultCallback *cb,
                      uint32_t flags = 0, bool explicit_block_only = false);

    TableMutatorAsync(std::mutex &mutex, std::condition_variable &cond, PropertiesPtr &props, Comm *comm,
                      ApplicationQueueInterfacePtr &app_queue, Table *table,
                      RangeLocatorPtr &range_locator, uint32_t timeout_ms, ResultCallback *cb,
                      uint32_t flags = 0, bool explicit_block_only = false,
              TableMutator *mutator = 0);

    virtual ~TableMutatorAsync();

    virtual uint64_t memory_used() { return m_memory_used; }

    uint64_t get_resend_count() { return m_resends; }

    void set(const KeySpec &key, const void *value, uint32_t value_len);

    void set(const KeySpec &key, const char *value) {
      if (value)
        set(key, value, strlen(value));
      else
        set(key, 0, 0);
    }

    void set(const KeySpec &key, const std::string &value) {
      set(key, value.data(), value.length());
    }

    void set_delete(const KeySpec &key);

    void set_cells(const Cells &cells) {
      set_cells(cells.begin(), cells.end());
    }

    void set_cells(Cells::const_iterator start, Cells::const_iterator end);

    void flush(bool sync=true);

    void buffer_finish(uint32_t id, int error, bool retry);
    void cancel();
    bool is_cancelled();
    void get_unsynced_rangeservers(std::vector<CommAddress> &unsynced);
    TableMutatorAsyncScatterBufferPtr get_outstanding_buffer(size_t id);
    bool retry(uint32_t timeout_ms);
    void update_outstanding(TableMutatorAsyncScatterBufferPtr &buffer);
    void get_failed_mutations(FailedMutations &failed_mutations) {
      std::lock_guard<std::mutex> lock(m_member_mutex);
      failed_mutations = m_failed_mutations;
    }
    bool has_outstanding() {
      std::lock_guard<std::mutex> lock(m_mutex);
      return !m_outstanding_buffers.empty();
    }
    bool has_outstanding_unlocked() {
      return !m_outstanding_buffers.empty();
    }
    bool needs_flush();

    SchemaPtr schema() { std::lock_guard<std::mutex> lock(m_mutex); return m_schema; }

  protected:
    void wait_for_completion();

  private:
    friend class TableMutator;
    void flush_with_tablequeue(TableMutator *mutator, bool sync=true);

    void initialize(PropertiesPtr &props);

    void initialize_indices(PropertiesPtr &props);

    friend class IndexMutatorCallback;
    void update_without_index(const Cell &cell);

    void update_without_index(const Key &full_key, const ColumnFamilySpec *cf, const Cell &cell);

    void update_without_index(const Key &full_key, const ColumnFamilySpec *cf, const void *value,
            size_t value_len);

    enum Operation {
      SET = 1,
      SET_CELLS,
      SET_DELETE,
      FLUSH
    };

    void do_sync();

    void to_full_key(const void *row, const char *cf, const void *cq,
                     int64_t ts, int64_t rev, uint8_t flag, Key &full_key,
                     ColumnFamilySpec **pcf = 0);

    void to_full_key(const KeySpec &key, Key &full_key,
                     ColumnFamilySpec **cf = 0) {
      to_full_key(key.row, key.column_family, key.column_qualifier,
                  key.timestamp, key.revision, key.flag, full_key, cf);
    }

    void to_full_key(const Cell &cell, Key &full_key,
                     ColumnFamilySpec **cf = 0) {
      to_full_key(cell.row_key, cell.column_family, cell.column_qualifier,
                  cell.timestamp, cell.revision, cell.flag, full_key, cf);
    }

    void update_unsynced_rangeservers(const CommAddressSet &unsynced);

    void handle_send_exceptions(const String& info);

    bool mutated() {
      std::lock_guard<std::mutex> lock(m_member_mutex);
      return m_mutated;
    }

    bool key_uses_index(Key &key);

    void update_with_index(Key &key, const ColumnFamilySpec *cf, const void *value,
                           uint32_t value_len);

    typedef std::map<uint32_t, TableMutatorAsyncScatterBufferPtr> ScatterBufferAsyncMap;

    const static uint32_t ms_max_sync_retries = 5;

    PropertiesPtr m_props;
    Comm *m_comm {};
    ApplicationQueueInterfacePtr  m_app_queue;
    Table *m_table {};
    SchemaPtr m_schema;     // needs mutex
    RangeLocatorPtr m_range_locator;
    TableIdentifierManaged m_table_identifier;    // needs mutex
    uint64_t m_memory_used {};  // protected by buffer_mutex
    uint64_t m_max_memory {};
    ScatterBufferAsyncMap  m_outstanding_buffers;  // protected by buffer mutex
    TableMutatorAsyncScatterBufferPtr m_current_buffer; // needs mutex
    uint64_t m_resends {};  // needs mutex
    uint32_t m_timeout_ms {};
    ResultCallback *m_cb {};
    uint32_t m_flags {};
    CommAddressSet m_unsynced_rangeservers;  // needs mutex
    std::mutex m_buffer_mutex;
    std::mutex &m_mutex;
    std::mutex m_member_mutex;
    std::condition_variable m_buffer_cond;
    std::condition_variable &m_cond;
    uint32_t m_next_buffer_id {}; // needs mutex
    FailedMutations m_failed_mutations;
    TableMutatorAsyncPtr m_index_mutator;
    TableMutatorAsyncPtr m_qualifier_index_mutator;
    IndexMutatorCallbackPtr m_imc;
    TableMutator *m_mutator {};
    bool m_explicit_block_only {};
    bool m_cancelled {};
    bool m_mutated {};      // needs mutex
    bool m_use_index {};

  };

}

#endif // Hypertable_Lib_TableMutatorAsync_h