Future.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 Hypertable. If not, see <http://www.gnu.org/licenses/>
 */

#ifndef Hypertable_Lib_Future_h
#define Hypertable_Lib_Future_h

#include "ResultCallback.h"
#include "Result.h"

#include <list>
#include <map>
#include <memory>

namespace Hypertable {

  using namespace std;

  class Future : public ResultCallback {
  public:

    Future(size_t capacity=0) : m_capacity(capacity) { }

    virtual ~Future() { cancel(); }

    bool get(ResultPtr &result);

    bool get(ResultPtr &result, uint32_t timeout_ms, bool &timed_out);

    void cancel();

    void register_scanner(TableScannerAsync *scanner);

    void deregister_scanner(TableScannerAsync *scanner);

    void register_mutator(TableMutatorAsync *scanner);

    void deregister_mutator(TableMutatorAsync *scanner);

    bool is_full() {
      std::lock_guard<std::mutex> lock(m_outstanding_mutex);
      return !has_remaining_capacity();
    }

    bool is_empty() {
      std::lock_guard<std::mutex> lock(m_outstanding_mutex);
      return _is_empty();
    }

    bool is_cancelled() {
      std::lock_guard<std::mutex> lock(m_outstanding_mutex);
      return _is_cancelled();
    }

    bool has_outstanding() {
      return !is_done();
    }

  private:
    friend class TableScannerAsync;
    friend class TableMutator;
    typedef list<ResultPtr> ResultQueue;

    void scan_ok(TableScannerAsync *scanner, ScanCellsPtr &cells);
    void scan_error(TableScannerAsync *scanner, int error, const std::string &error_msg,
                    bool eos);
    void update_ok(TableMutatorAsync *mutator);
    void update_error(TableMutatorAsync *mutator, int error, FailedMutations &failures);

    size_t memory_used() {
      return m_memory_used;
    }

    bool has_remaining_capacity() {
      if (!m_capacity)
        // unbounded buffer
        return true;
      else
        return m_memory_used < m_capacity;
    }

    bool _is_empty() { return m_queue.empty(); }
    bool _is_cancelled() const {
      return m_cancelled;
    }

    bool _is_done() {
      return m_outstanding == 0;
    }

    void enqueue(ResultPtr &result);

    ResultQueue m_queue;
    size_t m_capacity {};
    size_t m_memory_used {};
    bool m_cancelled {};
    typedef map<uint64_t, TableScannerAsync *> ScannerMap;
    typedef map<uint64_t, TableMutatorAsync *> MutatorMap;
    ScannerMap m_scanner_map;
    MutatorMap m_mutator_map;
  };
  typedef std::shared_ptr<Future> FuturePtr;
}

#endif // Hypertable_Lib_Future_h