Eine aufbereitete Darstellung der Quelle

 
     
 
 
Anforderungen  |   Konzepte  |   Entwurf  |   Entwicklung  |   Qualitätssicherung  |   Lebenszyklus  |   Steuerung
 
 
 
 

Benutzer

Quelle  rdb_sst_info.h   Sprache: C

 

/*
   Copyright (c) 2016, Facebook, Inc.

   This program 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 2 of the License.

   This program 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 02111-1301 USA */


#pragma once

/* C++ standard header files */
#include <atomic>
#include <condition_variable>
#include <mutex>
#include <queue>
#include <stack>
#include <string>
#include <thread>
#include <utility>
#include <vector>

/* RocksDB header files */
#include "rocksdb/db.h"
#include "rocksdb/sst_file_writer.h"

/* MyRocks header files */
#include "./rdb_utils.h"

namespace myrocks {

class Rdb_sst_file_ordered {
 private:
  class Rdb_sst_file {
   private:
    Rdb_sst_file(const Rdb_sst_file &p) = delete;
    Rdb_sst_file &operator=(const Rdb_sst_file &p) = delete;

    rocksdb::DB *const m_db;
    rocksdb::ColumnFamilyHandle *const m_cf;
    const rocksdb::DBOptions &m_db_options;
    rocksdb::SstFileWriter *m_sst_file_writer;
    const std::string m_name;
    const bool m_tracing;
    const rocksdb::Comparator *m_comparator;

    std::string generateKey(const std::string &key);

   public:
    Rdb_sst_file(rocksdb::DB *const db, rocksdb::ColumnFamilyHandle *const cf,
                 const rocksdb::DBOptions &db_options, const std::string &name,
                 const bool tracing);
    ~Rdb_sst_file();

    rocksdb::Status open();
    rocksdb::Status put(const rocksdb::Slice &key, const rocksdb::Slice &value);
    rocksdb::Status commit();

    inline const std::string get_name() const { return m_name; }
    inline int compare(rocksdb::Slice key1, rocksdb::Slice key2) {
      return m_comparator->Compare(key1, key2);
    }
  };

  class Rdb_sst_stack {
   private:
    char *m_buffer;
    size_t m_buffer_size;
    size_t m_offset;
    std::stack<std::tuple<size_t, size_t, size_t>> m_stack;

   public:
    explicit Rdb_sst_stack(size_t max_size)
        : m_buffer(nullptr), m_buffer_size(max_size) {}
    ~Rdb_sst_stack() { delete[] m_buffer; }

    void reset() { m_offset = 0; }
    bool empty() { return m_stack.empty(); }
    void push(const rocksdb::Slice &key, const rocksdb::Slice &value);
    std::pair<rocksdb::Slice, rocksdb::Slice> top();
    void pop() { m_stack.pop(); }
    size_t size() { return m_stack.size(); }
  };

  bool m_use_stack;
  bool m_first;
  std::string m_first_key;
  std::string m_first_value;
  Rdb_sst_stack m_stack;
  Rdb_sst_file m_file;

  rocksdb::Status apply_first();

 public:
  Rdb_sst_file_ordered(rocksdb::DB *const db,
                       rocksdb::ColumnFamilyHandle *const cf,
                       const rocksdb::DBOptions &db_options,
                       const std::string &name, const bool tracing,
                       size_t max_size);

  inline rocksdb::Status open() { return m_file.open(); }
  rocksdb::Status put(const rocksdb::Slice &key, const rocksdb::Slice &value);
  rocksdb::Status commit();
  inline const std::string get_name() const { return m_file.get_name(); }
};

class Rdb_sst_info {
 private:
  Rdb_sst_info(const Rdb_sst_info &p) = delete;
  Rdb_sst_info &operator=(const Rdb_sst_info &p) = delete;

  rocksdb::DB *const m_db;
  rocksdb::ColumnFamilyHandle *const m_cf;
  const rocksdb::DBOptions &m_db_options;
  uint64_t m_curr_size;
  uint64_t m_max_size;
  uint32_t m_sst_count;
  std::atomic<int> m_background_error;
  bool m_done;
  std::string m_prefix;
  static std::atomic<uint64_t> m_prefix_counter;
  static std::string m_suffix;
  mysql_mutex_t m_commit_mutex;
  Rdb_sst_file_ordered *m_sst_file;

  // List of committed SST files - we'll ingest them later in one single batch
  std::vector<std::string> m_committed_files;

  const bool m_tracing;
  bool m_print_client_error;

  int open_new_sst_file();
  void close_curr_sst_file();
  void commit_sst_file(Rdb_sst_file_ordered *sst_file);

  void set_error_msg(const std::string &sst_file_name,
                     const rocksdb::Status &s);

 public:
  Rdb_sst_info(rocksdb::DB *const db, const std::string &tablename,
               const std::string &indexname,
               rocksdb::ColumnFamilyHandle *const cf,
               const rocksdb::DBOptions &db_options, const bool tracing);
  ~Rdb_sst_info();

  /*
    This is the unit of work returned from Rdb_sst_info::finish and represents
    a group of SST to be ingested atomically with other Rdb_sst_commit_info.
    This is always local to the bulk loading complete operation so no locking
    is required
   */

  class Rdb_sst_commit_info {
   public:
    Rdb_sst_commit_info() : m_committed(true), m_cf(nullptr) {}

    Rdb_sst_commit_info(Rdb_sst_commit_info &&rhs) noexcept
        : m_committed(rhs.m_committed),
          m_cf(rhs.m_cf),
          m_committed_files(std::move(rhs.m_committed_files)) {
      rhs.m_committed = true;
      rhs.m_cf = nullptr;
    }

    Rdb_sst_commit_info &operator=(Rdb_sst_commit_info &&rhs) noexcept {
      reset();

      m_cf = rhs.m_cf;
      m_committed_files = std::move(rhs.m_committed_files);
      m_committed = rhs.m_committed;

      rhs.m_committed = true;
      rhs.m_cf = nullptr;

      return *this;
    }

    Rdb_sst_commit_info(const Rdb_sst_commit_info &) = delete;
    Rdb_sst_commit_info &operator=(const Rdb_sst_commit_info &) ;= delete;

    ~Rdb_sst_commit_info() { reset(); }

    void reset() {
      if (!m_committed) {
        for (auto sst_file : m_committed_files) {
          // In case something went wrong attempt to delete the temporary file.
          // If everything went fine that file will have been renamed and this
          // function call will fail.
          std::remove(sst_file.c_str());
        }
      }
      m_committed_files.clear();
      m_cf = nullptr;
      m_committed = true;
    }

    bool has_work() const {
      return m_cf != nullptr && m_committed_files.size() > 0;
    }

    void init(rocksdb::ColumnFamilyHandle *cf,
              std::vector<std::string> &&files) {
      DBUG_ASSERT(m_cf == nullptr && m_committed_files.size() == 0 &&
                  m_committed);
      m_cf = cf;
      m_committed_files = std::move(files);
      m_committed = false;
    }

    rocksdb::ColumnFamilyHandle *get_cf() const { return m_cf; }

    const std::vector<std::string> &get_committed_files() const {
      return m_committed_files;
    }

    void commit() { m_committed = true; }

   private:
    bool m_committed;
    rocksdb::ColumnFamilyHandle *m_cf;
    std::vector<std::string> m_committed_files;
  };

  int put(const rocksdb::Slice &key, const rocksdb::Slice &value);
  int finish(Rdb_sst_commit_info *commit_info, bool print_client_error = true);

  bool is_done() const { return m_done; }

  bool have_background_error() { return m_background_error != 0; }

  int get_and_reset_background_error() {
    int ret = m_background_error;
    while (!m_background_error.compare_exchange_weak(ret, HA_EXIT_SUCCESS)) {
      // Do nothing
    }

    return ret;
  }

  void set_background_error(int code) {
    int expected = HA_EXIT_SUCCESS;
    // Only assign 'code' into the error if it is already 0, otherwise ignore it
    m_background_error.compare_exchange_strong(expected, code);
  }

  /** Return the list of committed files later to be ingested **/
  const std::vector<std::string> &get_committed_files() {
    return m_committed_files;
  }

  rocksdb::ColumnFamilyHandle *get_cf() const { return m_cf; }

  static void init(const rocksdb::DB *const db);

  static void report_error_msg(const rocksdb::Status &s,
                               const char *sst_file_name);
};

}  // namespace myrocks

Messung V0.5 in Prozent
C=96 H=98 G=96

¤ Dauer der Verarbeitung: 0.9 Sekunden  (vorverarbeitet am  2026-10-08) ¤

*© Formatika GbR, Deutschland






Wurzel

Suchen

PVS Prover

Isabelle Prover

NIST Cobol Testsuite

Cephes Mathematical Library

Vienna Development Method

Haftungshinweis

Die Informationen auf dieser Webseite wurden nach bestem Wissen sorgfältig zusammengestellt. Es wird jedoch weder Vollständigkeit, noch Richtigkeit, noch Qualität der bereit gestellten Informationen zugesichert.

Bemerkung:

Die farbliche Syntaxdarstellung und die Messung sind noch experimentell.






                                                                                                                                                                                                                                                                                                                                                                                                     


Neuigkeiten

     Aktuelles
     Motto des Tages

Open Source Software

     Quellcodebibliothek
     Eigene Quellcodes
     Fremde Quellcodes
     Suchen

Jenseits des Üblichen ....
    

Besucherstatistik

Besucherstatistik

Statistik
#Sources=1126864
#Domains=1897691