Compare commits
14 Commits
f0c774de5a
...
6a9acb20dc
Author | SHA1 | Date | |
---|---|---|---|
6a9acb20dc | |||
29c0a98f01 | |||
b0670976ff | |||
317325aeb8 | |||
13ad145811 | |||
12a373324e | |||
45a444eb29 | |||
fe75412dfe | |||
85136c9af2 | |||
5187f32346 | |||
3a62c389ef | |||
48ed06a255 | |||
7d5d52afe3 | |||
bb9892cc84 |
@ -15,6 +15,7 @@
|
||||
* OS X support is temporarily disabled
|
||||
* \#19 \[bug\] Rename file is broken for files that are existing
|
||||
* \#23 \[bug\] Incorrect file size displayed while upload is pending
|
||||
* \#24 RocksDB implementations should be transactional
|
||||
|
||||
### Changes from v2.0.1-rc
|
||||
|
||||
|
@ -47,11 +47,11 @@ public:
|
||||
};
|
||||
|
||||
public:
|
||||
[[nodiscard]] virtual auto add_resume(resume_entry entry) -> bool = 0;
|
||||
[[nodiscard]] virtual auto add_resume(const resume_entry &entry) -> bool = 0;
|
||||
|
||||
[[nodiscard]] virtual auto add_upload(upload_entry entry) -> bool = 0;
|
||||
[[nodiscard]] virtual auto add_upload(const upload_entry &entry) -> bool = 0;
|
||||
|
||||
[[nodiscard]] virtual auto add_upload_active(upload_active_entry entry)
|
||||
[[nodiscard]] virtual auto add_upload_active(const upload_active_entry &entry)
|
||||
-> bool = 0;
|
||||
|
||||
virtual void clear() = 0;
|
||||
|
@ -41,7 +41,7 @@ private:
|
||||
const app_config &cfg_;
|
||||
|
||||
private:
|
||||
std::unique_ptr<rocksdb::DB> db_;
|
||||
std::unique_ptr<rocksdb::TransactionDB> db_{nullptr};
|
||||
std::atomic<std::uint64_t> id_{0U};
|
||||
rocksdb::ColumnFamilyHandle *resume_family_{};
|
||||
rocksdb::ColumnFamilyHandle *upload_active_family_{};
|
||||
@ -57,12 +57,23 @@ private:
|
||||
perform_action(std::string_view function_name,
|
||||
std::function<rocksdb::Status()> action) -> bool;
|
||||
|
||||
[[nodiscard]] auto perform_action(
|
||||
std::string_view function_name,
|
||||
std::function<rocksdb::Status(rocksdb::Transaction *txn)> action) -> bool;
|
||||
|
||||
[[nodiscard]] auto remove_resume(const std::string &api_path,
|
||||
rocksdb::Transaction *txn)
|
||||
-> rocksdb::Status;
|
||||
|
||||
[[nodiscard]] auto add_resume(const resume_entry &entry,
|
||||
rocksdb::Transaction *txn) -> rocksdb::Status;
|
||||
|
||||
public:
|
||||
[[nodiscard]] auto add_resume(resume_entry entry) -> bool override;
|
||||
[[nodiscard]] auto add_resume(const resume_entry &entry) -> bool override;
|
||||
|
||||
[[nodiscard]] auto add_upload(upload_entry entry) -> bool override;
|
||||
[[nodiscard]] auto add_upload(const upload_entry &entry) -> bool override;
|
||||
|
||||
[[nodiscard]] auto add_upload_active(upload_active_entry entry)
|
||||
[[nodiscard]] auto add_upload_active(const upload_active_entry &entry)
|
||||
-> bool override;
|
||||
|
||||
void clear() override;
|
||||
|
@ -40,7 +40,9 @@ public:
|
||||
|
||||
private:
|
||||
const app_config &cfg_;
|
||||
std::unique_ptr<rocksdb::DB> db_{nullptr};
|
||||
|
||||
private:
|
||||
std::unique_ptr<rocksdb::TransactionDB> db_{nullptr};
|
||||
rocksdb::ColumnFamilyHandle *default_family_{};
|
||||
rocksdb::ColumnFamilyHandle *pinned_family_{};
|
||||
rocksdb::ColumnFamilyHandle *size_family_{};
|
||||
@ -59,8 +61,21 @@ private:
|
||||
perform_action(std::string_view function_name,
|
||||
std::function<rocksdb::Status()> action) -> api_error;
|
||||
|
||||
[[nodiscard]] auto perform_action(
|
||||
std::string_view function_name,
|
||||
std::function<rocksdb::Status(rocksdb::Transaction *txn)> action)
|
||||
-> api_error;
|
||||
|
||||
[[nodiscard]] auto remove_api_path(const std::string &api_path,
|
||||
const std::string &source_path,
|
||||
rocksdb::Transaction *txn)
|
||||
-> rocksdb::Status;
|
||||
|
||||
[[nodiscard]] auto update_item_meta(const std::string &api_path,
|
||||
json json_data) -> api_error;
|
||||
json json_data,
|
||||
rocksdb::Transaction *base_txn = nullptr,
|
||||
rocksdb::Status *status = nullptr)
|
||||
-> api_error;
|
||||
|
||||
public:
|
||||
void clear() override;
|
||||
|
@ -42,11 +42,11 @@ private:
|
||||
utils::db::sqlite::db3_t db_;
|
||||
|
||||
public:
|
||||
[[nodiscard]] auto add_resume(resume_entry entry) -> bool override;
|
||||
[[nodiscard]] auto add_resume(const resume_entry &entry) -> bool override;
|
||||
|
||||
[[nodiscard]] auto add_upload(upload_entry entry) -> bool override;
|
||||
[[nodiscard]] auto add_upload(const upload_entry &entry) -> bool override;
|
||||
|
||||
[[nodiscard]] auto add_upload_active(upload_active_entry entry)
|
||||
[[nodiscard]] auto add_upload_active(const upload_active_entry &entry)
|
||||
-> bool override;
|
||||
|
||||
void clear() override;
|
||||
|
@ -35,7 +35,7 @@ public:
|
||||
public:
|
||||
explicit encrypt_provider(app_config &config);
|
||||
|
||||
~encrypt_provider() override = default;
|
||||
~encrypt_provider() override;
|
||||
|
||||
public:
|
||||
encrypt_provider(const encrypt_provider &) = delete;
|
||||
@ -45,9 +45,10 @@ public:
|
||||
|
||||
private:
|
||||
struct reader_info final {
|
||||
std::chrono::system_clock::time_point last_access_time =
|
||||
std::chrono::system_clock::now();
|
||||
std::unique_ptr<utils::encryption::encrypting_reader> reader{};
|
||||
std::chrono::system_clock::time_point last_access_time{
|
||||
std::chrono::system_clock::now(),
|
||||
};
|
||||
std::unique_ptr<utils::encryption::encrypting_reader> reader;
|
||||
std::mutex reader_mtx;
|
||||
};
|
||||
|
||||
@ -56,10 +57,9 @@ private:
|
||||
utils::db::sqlite::db3_t db_;
|
||||
|
||||
private:
|
||||
i_file_manager *fm_ = nullptr;
|
||||
std::unordered_map<std::string, std::shared_ptr<reader_info>>
|
||||
reader_lookup_{};
|
||||
std::recursive_mutex reader_lookup_mtx_{};
|
||||
i_file_manager *fm_{nullptr};
|
||||
std::unordered_map<std::string, std::shared_ptr<reader_info>> reader_lookup_;
|
||||
std::recursive_mutex reader_lookup_mtx_;
|
||||
|
||||
private:
|
||||
static auto create_api_file(const std::string &api_path, bool directory,
|
||||
@ -83,62 +83,68 @@ public:
|
||||
[[nodiscard]] auto create_directory(const std::string &api_path,
|
||||
api_meta_map &meta) -> api_error override;
|
||||
|
||||
[[nodiscard]] auto create_directory_clone_source_meta(
|
||||
const std::string & /*source_api_path*/,
|
||||
const std::string & /*api_path*/) -> api_error override {
|
||||
[[nodiscard]] auto
|
||||
create_directory_clone_source_meta(const std::string & /*source_api_path*/,
|
||||
const std::string & /*api_path*/)
|
||||
-> api_error override {
|
||||
return api_error::not_implemented;
|
||||
}
|
||||
|
||||
[[nodiscard]] auto create_file(const std::string & /*api_path*/,
|
||||
api_meta_map & /*meta*/)
|
||||
-> api_error override {
|
||||
return api_error::not_implemented;
|
||||
}
|
||||
|
||||
[[nodiscard]] auto
|
||||
create_file(const std::string & /*api_path*/,
|
||||
api_meta_map & /*meta*/) -> api_error override {
|
||||
return api_error::not_implemented;
|
||||
}
|
||||
|
||||
[[nodiscard]] auto get_api_path_from_source(
|
||||
const std::string & /*source_path*/,
|
||||
std::string & /*api_path*/) const -> api_error override;
|
||||
get_api_path_from_source(const std::string & /*source_path*/,
|
||||
std::string & /*api_path*/) const
|
||||
-> api_error override;
|
||||
|
||||
[[nodiscard]] auto get_directory_item_count(const std::string &api_path) const
|
||||
-> std::uint64_t override;
|
||||
|
||||
[[nodiscard]] auto
|
||||
get_directory_items(const std::string &api_path,
|
||||
directory_item_list &list) const -> api_error override;
|
||||
[[nodiscard]] auto get_directory_items(const std::string &api_path,
|
||||
directory_item_list &list) const
|
||||
-> api_error override;
|
||||
|
||||
[[nodiscard]] auto get_file(const std::string &api_path,
|
||||
api_file &file) const -> api_error override;
|
||||
[[nodiscard]] auto get_file(const std::string &api_path, api_file &file) const
|
||||
-> api_error override;
|
||||
|
||||
[[nodiscard]] auto get_file_list(api_file_list &list,
|
||||
std::string &marker) const
|
||||
-> api_error override;
|
||||
|
||||
[[nodiscard]] auto get_file_size(const std::string &api_path,
|
||||
std::uint64_t &file_size) const
|
||||
-> api_error override;
|
||||
|
||||
[[nodiscard]] auto get_filesystem_item(const std::string &api_path,
|
||||
bool directory,
|
||||
filesystem_item &fsi) const
|
||||
-> api_error override;
|
||||
|
||||
[[nodiscard]] auto get_filesystem_item_and_file(const std::string &api_path,
|
||||
api_file &file,
|
||||
filesystem_item &fsi) const
|
||||
-> api_error override;
|
||||
|
||||
[[nodiscard]] auto
|
||||
get_file_list(api_file_list &list,
|
||||
std::string &marker) const -> api_error override;
|
||||
get_filesystem_item_from_source_path(const std::string &source_path,
|
||||
filesystem_item &fsi) const
|
||||
-> api_error override;
|
||||
|
||||
[[nodiscard]] auto
|
||||
get_file_size(const std::string &api_path,
|
||||
std::uint64_t &file_size) const -> api_error override;
|
||||
[[nodiscard]] auto get_pinned_files() const
|
||||
-> std::vector<std::string> override;
|
||||
|
||||
[[nodiscard]] auto
|
||||
get_filesystem_item(const std::string &api_path, bool directory,
|
||||
filesystem_item &fsi) const -> api_error override;
|
||||
[[nodiscard]] auto get_item_meta(const std::string &api_path,
|
||||
api_meta_map &meta) const
|
||||
-> api_error override;
|
||||
|
||||
[[nodiscard]] auto get_filesystem_item_and_file(
|
||||
const std::string &api_path, api_file &file,
|
||||
filesystem_item &fsi) const -> api_error override;
|
||||
|
||||
[[nodiscard]] auto get_filesystem_item_from_source_path(
|
||||
const std::string &source_path,
|
||||
filesystem_item &fsi) const -> api_error override;
|
||||
|
||||
[[nodiscard]] auto
|
||||
get_pinned_files() const -> std::vector<std::string> override;
|
||||
|
||||
[[nodiscard]] auto
|
||||
get_item_meta(const std::string &api_path,
|
||||
api_meta_map &meta) const -> api_error override;
|
||||
|
||||
[[nodiscard]] auto
|
||||
get_item_meta(const std::string &api_path, const std::string &key,
|
||||
std::string &value) const -> api_error override;
|
||||
[[nodiscard]] auto get_item_meta(const std::string &api_path,
|
||||
const std::string &key,
|
||||
std::string &value) const
|
||||
-> api_error override;
|
||||
|
||||
[[nodiscard]] auto get_total_drive_space() const -> std::uint64_t override;
|
||||
|
||||
@ -153,11 +159,11 @@ public:
|
||||
[[nodiscard]] auto is_directory(const std::string &api_path,
|
||||
bool &exists) const -> api_error override;
|
||||
|
||||
[[nodiscard]] auto is_file(const std::string &api_path,
|
||||
bool &exists) const -> api_error override;
|
||||
[[nodiscard]] auto is_file(const std::string &api_path, bool &exists) const
|
||||
-> api_error override;
|
||||
|
||||
[[nodiscard]] auto
|
||||
is_file_writeable(const std::string &api_path) const -> bool override;
|
||||
[[nodiscard]] auto is_file_writeable(const std::string &api_path) const
|
||||
-> bool override;
|
||||
|
||||
[[nodiscard]] auto is_online() const -> bool override;
|
||||
|
||||
@ -167,42 +173,44 @@ public:
|
||||
return false;
|
||||
}
|
||||
|
||||
[[nodiscard]] auto
|
||||
read_file_bytes(const std::string &api_path, std::size_t size,
|
||||
std::uint64_t offset, data_buffer &data,
|
||||
stop_type &stop_requested) -> api_error override;
|
||||
[[nodiscard]] auto read_file_bytes(const std::string &api_path,
|
||||
std::size_t size, std::uint64_t offset,
|
||||
data_buffer &data,
|
||||
stop_type &stop_requested)
|
||||
-> api_error override;
|
||||
|
||||
[[nodiscard]] auto
|
||||
remove_directory(const std::string & /*api_path*/) -> api_error override {
|
||||
[[nodiscard]] auto remove_directory(const std::string & /*api_path*/)
|
||||
-> api_error override {
|
||||
return api_error::not_implemented;
|
||||
}
|
||||
|
||||
[[nodiscard]] auto
|
||||
remove_file(const std::string & /*api_path*/) -> api_error override {
|
||||
[[nodiscard]] auto remove_file(const std::string & /*api_path*/)
|
||||
-> api_error override {
|
||||
return api_error::not_implemented;
|
||||
}
|
||||
|
||||
[[nodiscard]] auto
|
||||
remove_item_meta(const std::string & /*api_path*/,
|
||||
const std::string & /*key*/) -> api_error override {
|
||||
[[nodiscard]] auto remove_item_meta(const std::string & /*api_path*/,
|
||||
const std::string & /*key*/)
|
||||
-> api_error override {
|
||||
return api_error::success;
|
||||
}
|
||||
|
||||
[[nodiscard]] auto
|
||||
rename_file(const std::string & /*from_api_path*/,
|
||||
const std::string & /*to_api_path*/) -> api_error override {
|
||||
[[nodiscard]] auto rename_file(const std::string & /*from_api_path*/,
|
||||
const std::string & /*to_api_path*/)
|
||||
-> api_error override {
|
||||
return api_error::not_implemented;
|
||||
}
|
||||
|
||||
[[nodiscard]] auto
|
||||
set_item_meta(const std::string & /*api_path*/, const std::string & /*key*/,
|
||||
const std::string & /*value*/) -> api_error override {
|
||||
[[nodiscard]] auto set_item_meta(const std::string & /*api_path*/,
|
||||
const std::string & /*key*/,
|
||||
const std::string & /*value*/)
|
||||
-> api_error override {
|
||||
return api_error::success;
|
||||
}
|
||||
|
||||
[[nodiscard]] auto
|
||||
set_item_meta(const std::string & /*api_path*/,
|
||||
const api_meta_map & /*meta*/) -> api_error override {
|
||||
[[nodiscard]] auto set_item_meta(const std::string & /*api_path*/,
|
||||
const api_meta_map & /*meta*/)
|
||||
-> api_error override {
|
||||
return api_error::success;
|
||||
}
|
||||
|
||||
@ -211,10 +219,10 @@ public:
|
||||
|
||||
void stop() override;
|
||||
|
||||
[[nodiscard]] auto
|
||||
upload_file(const std::string & /*api_path*/,
|
||||
[[nodiscard]] auto upload_file(const std::string & /*api_path*/,
|
||||
const std::string & /*source_path*/,
|
||||
stop_type & /*stop_requested*/) -> api_error override {
|
||||
stop_type & /*stop_requested*/)
|
||||
-> api_error override {
|
||||
return api_error::not_implemented;
|
||||
}
|
||||
};
|
||||
|
@ -34,14 +34,14 @@ namespace {
|
||||
create_rocksdb(const repertory::app_config &cfg, const std::string &name,
|
||||
const std::vector<rocksdb::ColumnFamilyDescriptor> &families,
|
||||
std::vector<rocksdb::ColumnFamilyHandle *> &handles, bool clear)
|
||||
-> std::unique_ptr<rocksdb::DB> {
|
||||
-> std::unique_ptr<rocksdb::TransactionDB> {
|
||||
REPERTORY_USES_FUNCTION_NAME();
|
||||
|
||||
auto path = repertory::utils::path::combine(cfg.get_data_directory(), {name});
|
||||
if (clear &&
|
||||
not repertory::utils::file::directory{path}.remove_recursively()) {
|
||||
repertory::utils::error::raise_error(
|
||||
function_name, "failed to remove file mgr db|" + path);
|
||||
function_name, "failed to remove " + name + " db|" + path);
|
||||
}
|
||||
|
||||
rocksdb::Options options{};
|
||||
@ -50,14 +50,15 @@ create_rocksdb(const repertory::app_config &cfg, const std::string &name,
|
||||
options.db_log_dir = cfg.get_log_directory();
|
||||
options.keep_log_file_num = 10;
|
||||
|
||||
rocksdb::DB *ptr{};
|
||||
auto status = rocksdb::DB::Open(options, path, families, &handles, &ptr);
|
||||
rocksdb::TransactionDB *ptr{};
|
||||
auto status = rocksdb::TransactionDB::Open(
|
||||
options, rocksdb::TransactionDBOptions{}, path, families, &handles, &ptr);
|
||||
if (not status.ok()) {
|
||||
repertory::utils::error::raise_error(function_name, status.ToString());
|
||||
throw repertory::startup_exception(status.ToString());
|
||||
}
|
||||
|
||||
return std::unique_ptr<rocksdb::DB>(ptr);
|
||||
return std::unique_ptr<rocksdb::TransactionDB>(ptr);
|
||||
}
|
||||
} // namespace
|
||||
|
||||
@ -78,7 +79,7 @@ void rdb_file_mgr_db::create_or_open(bool clear) {
|
||||
families.emplace_back("upload", rocksdb::ColumnFamilyOptions());
|
||||
|
||||
auto handles = std::vector<rocksdb::ColumnFamilyHandle *>();
|
||||
db_ = create_rocksdb(cfg_, "mgr", families, handles, clear);
|
||||
db_ = create_rocksdb(cfg_, "file_mgr", families, handles, clear);
|
||||
|
||||
std::size_t idx{};
|
||||
resume_family_ = handles[idx++];
|
||||
@ -86,37 +87,50 @@ void rdb_file_mgr_db::create_or_open(bool clear) {
|
||||
upload_family_ = handles[idx++];
|
||||
}
|
||||
|
||||
auto rdb_file_mgr_db::add_resume(resume_entry entry) -> bool {
|
||||
auto rdb_file_mgr_db::add_resume(const resume_entry &entry) -> bool {
|
||||
REPERTORY_USES_FUNCTION_NAME();
|
||||
|
||||
return perform_action(
|
||||
function_name,
|
||||
[this, &entry](rocksdb::Transaction *txn) -> rocksdb::Status {
|
||||
return add_resume(entry, txn);
|
||||
});
|
||||
}
|
||||
|
||||
auto rdb_file_mgr_db::add_resume(const resume_entry &entry,
|
||||
rocksdb::Transaction *txn) -> rocksdb::Status {
|
||||
REPERTORY_USES_FUNCTION_NAME();
|
||||
|
||||
return perform_action(function_name, [this, &entry]() -> rocksdb::Status {
|
||||
auto data = json({
|
||||
{"chunk_size", entry.chunk_size},
|
||||
{"read_state", utils::string::from_dynamic_bitset(entry.read_state)},
|
||||
{"source_path", entry.source_path},
|
||||
});
|
||||
return db_->Put(rocksdb::WriteOptions{}, resume_family_, entry.api_path,
|
||||
data.dump());
|
||||
});
|
||||
return txn->Put(resume_family_, entry.api_path, data.dump());
|
||||
}
|
||||
|
||||
auto rdb_file_mgr_db::add_upload(upload_entry entry) -> bool {
|
||||
auto rdb_file_mgr_db::add_upload(const upload_entry &entry) -> bool {
|
||||
REPERTORY_USES_FUNCTION_NAME();
|
||||
|
||||
return perform_action(function_name, [this, &entry]() -> rocksdb::Status {
|
||||
return db_->Put(rocksdb::WriteOptions{}, upload_family_,
|
||||
utils::string::zero_pad(std::to_string(++id_), 20U) + '|' +
|
||||
entry.api_path,
|
||||
return perform_action(
|
||||
function_name,
|
||||
[this, &entry](rocksdb::Transaction *txn) -> rocksdb::Status {
|
||||
return txn->Put(upload_family_,
|
||||
utils::string::zero_pad(std::to_string(++id_), 20U) +
|
||||
'|' + entry.api_path,
|
||||
entry.source_path);
|
||||
});
|
||||
}
|
||||
|
||||
auto rdb_file_mgr_db::add_upload_active(upload_active_entry entry) -> bool {
|
||||
auto rdb_file_mgr_db::add_upload_active(const upload_active_entry &entry)
|
||||
-> bool {
|
||||
REPERTORY_USES_FUNCTION_NAME();
|
||||
|
||||
return perform_action(function_name, [this, &entry]() -> rocksdb::Status {
|
||||
return db_->Put(rocksdb::WriteOptions{}, upload_active_family_,
|
||||
entry.api_path, entry.source_path);
|
||||
return perform_action(
|
||||
function_name,
|
||||
[this, &entry](rocksdb::Transaction *txn) -> rocksdb::Status {
|
||||
return txn->Put(upload_active_family_, entry.api_path,
|
||||
entry.source_path);
|
||||
});
|
||||
}
|
||||
|
||||
@ -215,14 +229,56 @@ auto rdb_file_mgr_db::perform_action(std::string_view function_name,
|
||||
return false;
|
||||
}
|
||||
|
||||
auto rdb_file_mgr_db::perform_action(
|
||||
std::string_view function_name,
|
||||
std::function<rocksdb::Status(rocksdb::Transaction *txn)> action) -> bool {
|
||||
std::unique_ptr<rocksdb::Transaction> txn{
|
||||
db_->BeginTransaction(rocksdb::WriteOptions{},
|
||||
rocksdb::TransactionOptions{}),
|
||||
};
|
||||
|
||||
try {
|
||||
auto res = action(txn.get());
|
||||
if (res.ok()) {
|
||||
auto commit_res = txn->Commit();
|
||||
if (commit_res.ok()) {
|
||||
return true;
|
||||
}
|
||||
|
||||
utils::error::raise_error(function_name,
|
||||
"rocksdb commit failed|" + res.ToString());
|
||||
return false;
|
||||
}
|
||||
|
||||
utils::error::raise_error(function_name,
|
||||
"rocksdb action failed|" + res.ToString());
|
||||
} catch (const std::exception &ex) {
|
||||
utils::error::raise_error(function_name, ex,
|
||||
"failed to handle rocksdb action");
|
||||
}
|
||||
|
||||
auto rollback_res = txn->Rollback();
|
||||
utils::error::raise_error(function_name, "rocksdb rollback failed|" +
|
||||
rollback_res.ToString());
|
||||
return false;
|
||||
}
|
||||
|
||||
auto rdb_file_mgr_db::remove_resume(const std::string &api_path) -> bool {
|
||||
REPERTORY_USES_FUNCTION_NAME();
|
||||
|
||||
return perform_action(function_name, [this, &api_path]() -> rocksdb::Status {
|
||||
return db_->Delete(rocksdb::WriteOptions{}, resume_family_, api_path);
|
||||
return perform_action(
|
||||
function_name,
|
||||
[this, &api_path](rocksdb::Transaction *txn) -> rocksdb::Status {
|
||||
return remove_resume(api_path, txn);
|
||||
});
|
||||
}
|
||||
|
||||
auto rdb_file_mgr_db::remove_resume(const std::string &api_path,
|
||||
rocksdb::Transaction *txn)
|
||||
-> rocksdb::Status {
|
||||
return txn->Delete(resume_family_, api_path);
|
||||
}
|
||||
|
||||
auto rdb_file_mgr_db::remove_upload(const std::string &api_path) -> bool {
|
||||
REPERTORY_USES_FUNCTION_NAME();
|
||||
|
||||
@ -235,8 +291,10 @@ auto rdb_file_mgr_db::remove_upload(const std::string &api_path) -> bool {
|
||||
continue;
|
||||
}
|
||||
|
||||
return perform_action(function_name, [this, &iter]() -> rocksdb::Status {
|
||||
return db_->Delete(rocksdb::WriteOptions{}, upload_family_, iter->key());
|
||||
return perform_action(
|
||||
function_name,
|
||||
[this, &iter](rocksdb::Transaction *txn) -> rocksdb::Status {
|
||||
return txn->Delete(upload_family_, iter->key());
|
||||
});
|
||||
}
|
||||
|
||||
@ -247,9 +305,10 @@ auto rdb_file_mgr_db::remove_upload_active(const std::string &api_path)
|
||||
-> bool {
|
||||
REPERTORY_USES_FUNCTION_NAME();
|
||||
|
||||
return perform_action(function_name, [this, &api_path]() -> rocksdb::Status {
|
||||
return db_->Delete(rocksdb::WriteOptions{}, upload_active_family_,
|
||||
api_path);
|
||||
return perform_action(
|
||||
function_name,
|
||||
[this, &api_path](rocksdb::Transaction *txn) -> rocksdb::Status {
|
||||
return txn->Delete(upload_active_family_, api_path);
|
||||
});
|
||||
}
|
||||
|
||||
@ -279,10 +338,15 @@ auto rdb_file_mgr_db::rename_resume(const std::string &from_api_path,
|
||||
data.at("source_path").get<std::string>(),
|
||||
};
|
||||
|
||||
if (not remove_resume(from_api_path)) {
|
||||
return false;
|
||||
return perform_action(function_name,
|
||||
[this, &entry, &from_api_path](
|
||||
rocksdb::Transaction *txn) -> rocksdb::Status {
|
||||
auto txn_res = remove_resume(from_api_path, txn);
|
||||
if (not txn_res.ok()) {
|
||||
return txn_res;
|
||||
}
|
||||
|
||||
return add_resume(entry);
|
||||
return add_resume(entry, txn);
|
||||
});
|
||||
}
|
||||
} // namespace repertory
|
||||
|
@ -33,14 +33,14 @@ namespace {
|
||||
create_rocksdb(const repertory::app_config &cfg, const std::string &name,
|
||||
const std::vector<rocksdb::ColumnFamilyDescriptor> &families,
|
||||
std::vector<rocksdb::ColumnFamilyHandle *> &handles, bool clear)
|
||||
-> std::unique_ptr<rocksdb::DB> {
|
||||
-> std::unique_ptr<rocksdb::TransactionDB> {
|
||||
REPERTORY_USES_FUNCTION_NAME();
|
||||
|
||||
auto path = repertory::utils::path::combine(cfg.get_data_directory(), {name});
|
||||
if (clear &&
|
||||
not repertory::utils::file::directory{path}.remove_recursively()) {
|
||||
repertory::utils::error::raise_error(function_name,
|
||||
"failed to remove meta db|" + path);
|
||||
repertory::utils::error::raise_error(
|
||||
function_name, "failed to remove " + name + " db|" + path);
|
||||
}
|
||||
|
||||
rocksdb::Options options{};
|
||||
@ -49,14 +49,15 @@ create_rocksdb(const repertory::app_config &cfg, const std::string &name,
|
||||
options.db_log_dir = cfg.get_log_directory();
|
||||
options.keep_log_file_num = 10;
|
||||
|
||||
rocksdb::DB *ptr{};
|
||||
auto status = rocksdb::DB::Open(options, path, families, &handles, &ptr);
|
||||
rocksdb::TransactionDB *ptr{};
|
||||
auto status = rocksdb::TransactionDB::Open(
|
||||
options, rocksdb::TransactionDBOptions{}, path, families, &handles, &ptr);
|
||||
if (not status.ok()) {
|
||||
repertory::utils::error::raise_error(function_name, status.ToString());
|
||||
throw repertory::startup_exception(status.ToString());
|
||||
}
|
||||
|
||||
return std::unique_ptr<rocksdb::DB>(ptr);
|
||||
return std::unique_ptr<rocksdb::TransactionDB>(ptr);
|
||||
}
|
||||
} // namespace
|
||||
|
||||
@ -92,7 +93,7 @@ void rdb_meta_db::clear() { create_or_open(true); }
|
||||
auto rdb_meta_db::create_iterator(rocksdb::ColumnFamilyHandle *family) const
|
||||
-> std::shared_ptr<rocksdb::Iterator> {
|
||||
return std::shared_ptr<rocksdb::Iterator>(
|
||||
db_->NewIterator(rocksdb::ReadOptions(), family));
|
||||
db_->NewIterator(rocksdb::ReadOptions{}, family));
|
||||
}
|
||||
|
||||
auto rdb_meta_db::get_api_path(const std::string &source_path,
|
||||
@ -104,7 +105,7 @@ auto rdb_meta_db::get_api_path(const std::string &source_path,
|
||||
}
|
||||
|
||||
return perform_action(function_name, [&]() -> rocksdb::Status {
|
||||
return db_->Get(rocksdb::ReadOptions(), source_family_, source_path,
|
||||
return db_->Get(rocksdb::ReadOptions{}, source_family_, source_path,
|
||||
&api_path);
|
||||
});
|
||||
}
|
||||
@ -129,7 +130,7 @@ auto rdb_meta_db::get_item_meta_json(const std::string &api_path,
|
||||
{
|
||||
std::string value;
|
||||
auto res = perform_action(function_name, [&]() -> rocksdb::Status {
|
||||
return db_->Get(rocksdb::ReadOptions(), default_family_, api_path,
|
||||
return db_->Get(rocksdb::ReadOptions{}, default_family_, api_path,
|
||||
&value);
|
||||
});
|
||||
if (res != api_error::success) {
|
||||
@ -144,7 +145,7 @@ auto rdb_meta_db::get_item_meta_json(const std::string &api_path,
|
||||
{
|
||||
std::string value;
|
||||
auto res = perform_action(function_name, [&]() -> rocksdb::Status {
|
||||
return db_->Get(rocksdb::ReadOptions(), pinned_family_, api_path,
|
||||
return db_->Get(rocksdb::ReadOptions{}, pinned_family_, api_path,
|
||||
&value);
|
||||
});
|
||||
if (res != api_error::success) {
|
||||
@ -158,7 +159,7 @@ auto rdb_meta_db::get_item_meta_json(const std::string &api_path,
|
||||
{
|
||||
std::string value;
|
||||
auto res = perform_action(function_name, [&]() -> rocksdb::Status {
|
||||
return db_->Get(rocksdb::ReadOptions(), size_family_, api_path, &value);
|
||||
return db_->Get(rocksdb::ReadOptions{}, size_family_, api_path, &value);
|
||||
});
|
||||
if (res != api_error::success) {
|
||||
return res;
|
||||
@ -199,13 +200,13 @@ auto rdb_meta_db::get_item_meta(const std::string &api_path,
|
||||
|
||||
if (key == META_PINNED) {
|
||||
return perform_action(function_name, [&]() -> rocksdb::Status {
|
||||
return db_->Get(rocksdb::ReadOptions(), pinned_family_, api_path, &value);
|
||||
return db_->Get(rocksdb::ReadOptions{}, pinned_family_, api_path, &value);
|
||||
});
|
||||
}
|
||||
|
||||
if (key == META_SIZE) {
|
||||
return perform_action(function_name, [&]() -> rocksdb::Status {
|
||||
return db_->Get(rocksdb::ReadOptions(), size_family_, api_path, &value);
|
||||
return db_->Get(rocksdb::ReadOptions{}, size_family_, api_path, &value);
|
||||
});
|
||||
}
|
||||
|
||||
@ -271,21 +272,84 @@ auto rdb_meta_db::perform_action(std::string_view function_name,
|
||||
return res.IsNotFound() ? api_error::item_not_found : api_error::error;
|
||||
}
|
||||
|
||||
auto rdb_meta_db::perform_action(
|
||||
std::string_view function_name,
|
||||
std::function<rocksdb::Status(rocksdb::Transaction *txn)> action)
|
||||
-> api_error {
|
||||
std::unique_ptr<rocksdb::Transaction> txn{
|
||||
db_->BeginTransaction(rocksdb::WriteOptions{},
|
||||
rocksdb::TransactionOptions{}),
|
||||
};
|
||||
|
||||
try {
|
||||
auto res = action(txn.get());
|
||||
if (res.ok()) {
|
||||
auto commit_res = txn->Commit();
|
||||
if (commit_res.ok()) {
|
||||
return api_error::success;
|
||||
}
|
||||
|
||||
utils::error::raise_error(function_name,
|
||||
"rocksdb commit failed|" + res.ToString());
|
||||
return api_error::error;
|
||||
}
|
||||
|
||||
utils::error::raise_error(function_name,
|
||||
"rocksdb action failed|" + res.ToString());
|
||||
} catch (const std::exception &ex) {
|
||||
utils::error::raise_error(function_name, ex,
|
||||
"failed to handle rocksdb action");
|
||||
}
|
||||
|
||||
auto rollback_res = txn->Rollback();
|
||||
utils::error::raise_error(function_name, "rocksdb rollback failed|" +
|
||||
rollback_res.ToString());
|
||||
return api_error::error;
|
||||
}
|
||||
|
||||
void rdb_meta_db::remove_api_path(const std::string &api_path) {
|
||||
REPERTORY_USES_FUNCTION_NAME();
|
||||
|
||||
std::string source_path;
|
||||
[[maybe_unused]] auto res = get_item_meta(api_path, META_SOURCE, source_path);
|
||||
|
||||
res = perform_action(
|
||||
function_name, [this, &api_path, &source_path]() -> rocksdb::Status {
|
||||
db_->Delete(rocksdb::WriteOptions(), pinned_family_, api_path);
|
||||
db_->Delete(rocksdb::WriteOptions(), size_family_, api_path);
|
||||
if (not source_path.empty()) {
|
||||
db_->Delete(rocksdb::WriteOptions(), source_family_, source_path);
|
||||
auto res = get_item_meta(api_path, META_SOURCE, source_path);
|
||||
if (res != api_error::success) {
|
||||
utils::error::raise_api_path_error(function_name, api_path, res,
|
||||
"failed to get source path");
|
||||
}
|
||||
return db_->Delete(rocksdb::WriteOptions(), default_family_, api_path);
|
||||
|
||||
res = perform_action(function_name,
|
||||
[this, &api_path, &source_path](
|
||||
rocksdb::Transaction *txn) -> rocksdb::Status {
|
||||
return remove_api_path(api_path, source_path, txn);
|
||||
});
|
||||
if (res != api_error::success) {
|
||||
utils::error::raise_api_path_error(function_name, api_path, res,
|
||||
"failed to remove api path");
|
||||
}
|
||||
}
|
||||
|
||||
auto rdb_meta_db::remove_api_path(const std::string &api_path,
|
||||
const std::string &source_path,
|
||||
rocksdb::Transaction *txn)
|
||||
-> rocksdb::Status {
|
||||
auto txn_res = txn->Delete(pinned_family_, api_path);
|
||||
if (not txn_res.ok()) {
|
||||
return txn_res;
|
||||
}
|
||||
|
||||
txn_res = txn->Delete(size_family_, api_path);
|
||||
if (not txn_res.ok()) {
|
||||
return txn_res;
|
||||
}
|
||||
|
||||
if (not source_path.empty()) {
|
||||
txn_res = txn->Delete(source_family_, source_path);
|
||||
if (not txn_res.ok()) {
|
||||
return txn_res;
|
||||
}
|
||||
}
|
||||
|
||||
return txn->Delete(default_family_, api_path);
|
||||
}
|
||||
|
||||
auto rdb_meta_db::remove_item_meta(const std::string &api_path,
|
||||
@ -309,14 +373,27 @@ auto rdb_meta_db::remove_item_meta(const std::string &api_path,
|
||||
auto rdb_meta_db::rename_item_meta(const std::string &from_api_path,
|
||||
const std::string &to_api_path)
|
||||
-> api_error {
|
||||
REPERTORY_USES_FUNCTION_NAME();
|
||||
|
||||
json json_data;
|
||||
auto res = get_item_meta_json(from_api_path, json_data);
|
||||
if (res != api_error::success) {
|
||||
return res;
|
||||
}
|
||||
|
||||
remove_api_path(from_api_path);
|
||||
return update_item_meta(to_api_path, json_data);
|
||||
return perform_action(
|
||||
function_name, [&](rocksdb::Transaction *txn) -> rocksdb::Status {
|
||||
auto txn_res = remove_api_path(
|
||||
from_api_path, json_data[META_SOURCE].get<std::string>(), txn);
|
||||
if (not txn_res.ok()) {
|
||||
return txn_res;
|
||||
}
|
||||
|
||||
rocksdb::Status status;
|
||||
[[maybe_unused]] auto api_res =
|
||||
update_item_meta(to_api_path, json_data, txn, &status);
|
||||
return status;
|
||||
});
|
||||
}
|
||||
|
||||
auto rdb_meta_db::set_item_meta(const std::string &api_path,
|
||||
@ -325,14 +402,16 @@ auto rdb_meta_db::set_item_meta(const std::string &api_path,
|
||||
REPERTORY_USES_FUNCTION_NAME();
|
||||
|
||||
if (key == META_PINNED) {
|
||||
return perform_action(function_name, [&]() -> rocksdb::Status {
|
||||
return db_->Put(rocksdb::WriteOptions(), pinned_family_, api_path, value);
|
||||
return perform_action(function_name,
|
||||
[&](rocksdb::Transaction *txn) -> rocksdb::Status {
|
||||
return txn->Put(pinned_family_, api_path, value);
|
||||
});
|
||||
}
|
||||
|
||||
if (key == META_SIZE) {
|
||||
return perform_action(function_name, [&]() -> rocksdb::Status {
|
||||
return db_->Put(rocksdb::WriteOptions(), size_family_, api_path, value);
|
||||
return perform_action(function_name,
|
||||
[&](rocksdb::Transaction *txn) -> rocksdb::Status {
|
||||
return txn->Put(size_family_, api_path, value);
|
||||
});
|
||||
}
|
||||
|
||||
@ -362,8 +441,9 @@ auto rdb_meta_db::set_item_meta(const std::string &api_path,
|
||||
return update_item_meta(api_path, json_data);
|
||||
}
|
||||
|
||||
auto rdb_meta_db::update_item_meta(const std::string &api_path, json json_data)
|
||||
-> api_error {
|
||||
auto rdb_meta_db::update_item_meta(const std::string &api_path, json json_data,
|
||||
rocksdb::Transaction *base_txn,
|
||||
rocksdb::Status *status) -> api_error {
|
||||
REPERTORY_USES_FUNCTION_NAME();
|
||||
|
||||
try {
|
||||
@ -393,51 +473,67 @@ auto rdb_meta_db::update_item_meta(const std::string &api_path, json json_data)
|
||||
json_data[META_SIZE] = std::to_string(size);
|
||||
json_data[META_SOURCE] = source_path;
|
||||
|
||||
if (not directory) {
|
||||
auto should_del_source{false};
|
||||
std::string orig_source_path;
|
||||
if (not directory) {
|
||||
auto res = get_item_meta(api_path, META_SOURCE, orig_source_path);
|
||||
if (res != api_error::success && res != api_error::item_not_found) {
|
||||
return res;
|
||||
}
|
||||
|
||||
if (not orig_source_path.empty() && orig_source_path != source_path) {
|
||||
res = perform_action(function_name, [&]() -> rocksdb::Status {
|
||||
return db_->Delete(rocksdb::WriteOptions(), source_family_,
|
||||
orig_source_path);
|
||||
});
|
||||
if (res != api_error::success && res != api_error::item_not_found) {
|
||||
return res;
|
||||
}
|
||||
}
|
||||
should_del_source =
|
||||
not orig_source_path.empty() && orig_source_path != source_path;
|
||||
}
|
||||
|
||||
json_data.erase(META_PINNED);
|
||||
json_data.erase(META_SIZE);
|
||||
|
||||
return perform_action(function_name, [&]() -> rocksdb::Status {
|
||||
auto res = db_->Put(rocksdb::WriteOptions(), pinned_family_, api_path,
|
||||
utils::string::from_bool(pinned));
|
||||
const auto set_status = [&status](rocksdb::Status res) -> rocksdb::Status {
|
||||
if (status != nullptr) {
|
||||
*status = res;
|
||||
}
|
||||
|
||||
return res;
|
||||
};
|
||||
|
||||
const auto do_transaction =
|
||||
[&](rocksdb::Transaction *txn) -> rocksdb::Status {
|
||||
if (should_del_source) {
|
||||
auto res = set_status(txn->Delete(source_family_, orig_source_path));
|
||||
if (not res.ok()) {
|
||||
return res;
|
||||
}
|
||||
}
|
||||
|
||||
auto res = set_status(
|
||||
txn->Put(pinned_family_, api_path, utils::string::from_bool(pinned)));
|
||||
if (not res.ok()) {
|
||||
return res;
|
||||
}
|
||||
|
||||
res = db_->Put(rocksdb::WriteOptions(), size_family_, api_path,
|
||||
std::to_string(size));
|
||||
res = set_status(txn->Put(size_family_, api_path, std::to_string(size)));
|
||||
if (not res.ok()) {
|
||||
return res;
|
||||
}
|
||||
|
||||
if (not source_path.empty()) {
|
||||
res = db_->Put(rocksdb::WriteOptions(), source_family_, source_path,
|
||||
api_path);
|
||||
res = set_status(txn->Put(source_family_, source_path, api_path));
|
||||
if (not res.ok()) {
|
||||
return res;
|
||||
}
|
||||
}
|
||||
|
||||
return db_->Put(rocksdb::WriteOptions(), default_family_, api_path,
|
||||
json_data.dump());
|
||||
});
|
||||
return set_status(txn->Put(default_family_, api_path, json_data.dump()));
|
||||
};
|
||||
|
||||
if (base_txn == nullptr) {
|
||||
return perform_action(function_name, do_transaction);
|
||||
}
|
||||
|
||||
auto res = set_status(do_transaction(base_txn));
|
||||
if (res.ok()) {
|
||||
return api_error::success;
|
||||
}
|
||||
} catch (const std::exception &e) {
|
||||
utils::error::raise_api_path_error(function_name, api_path, e,
|
||||
"failed to update item meta");
|
||||
|
@ -82,7 +82,7 @@ sqlite_file_mgr_db::sqlite_file_mgr_db(const app_config &cfg) {
|
||||
|
||||
sqlite_file_mgr_db::~sqlite_file_mgr_db() { db_.reset(); }
|
||||
|
||||
auto sqlite_file_mgr_db::add_resume(resume_entry entry) -> bool {
|
||||
auto sqlite_file_mgr_db::add_resume(const resume_entry &entry) -> bool {
|
||||
return utils::db::sqlite::db_insert{*db_, resume_table}
|
||||
.or_replace()
|
||||
.column_value("api_path", entry.api_path)
|
||||
@ -94,7 +94,7 @@ auto sqlite_file_mgr_db::add_resume(resume_entry entry) -> bool {
|
||||
.ok();
|
||||
}
|
||||
|
||||
auto sqlite_file_mgr_db::add_upload(upload_entry entry) -> bool {
|
||||
auto sqlite_file_mgr_db::add_upload(const upload_entry &entry) -> bool {
|
||||
return utils::db::sqlite::db_insert{*db_, upload_table}
|
||||
.or_replace()
|
||||
.column_value("api_path", entry.api_path)
|
||||
@ -103,7 +103,8 @@ auto sqlite_file_mgr_db::add_upload(upload_entry entry) -> bool {
|
||||
.ok();
|
||||
}
|
||||
|
||||
auto sqlite_file_mgr_db::add_upload_active(upload_active_entry entry) -> bool {
|
||||
auto sqlite_file_mgr_db::add_upload_active(const upload_active_entry &entry)
|
||||
-> bool {
|
||||
return utils::db::sqlite::db_insert{*db_, upload_active_table}
|
||||
.or_replace()
|
||||
.column_value("api_path", entry.api_path)
|
||||
|
@ -474,15 +474,18 @@ auto open_file::close() -> bool {
|
||||
}
|
||||
}
|
||||
|
||||
if (err != api_error::success || read_state_.all()) {
|
||||
mgr_.remove_resume(get_api_path(), get_source_path());
|
||||
}
|
||||
|
||||
if (err == api_error::success) {
|
||||
return true;
|
||||
}
|
||||
|
||||
mgr_.remove_resume(get_api_path(), get_source_path());
|
||||
if (not utils::file::file(fsi_.source_path).remove()) {
|
||||
utils::error::raise_api_path_error(
|
||||
function_name, get_api_path(), fsi_.source_path,
|
||||
utils::get_last_error_code(), "failed to delete file");
|
||||
utils::get_last_error_code(), "failed to delete source file");
|
||||
}
|
||||
|
||||
auto parent = utils::path::get_parent_path(fsi_.source_path);
|
||||
@ -493,7 +496,7 @@ auto open_file::close() -> bool {
|
||||
if (res != api_error::success) {
|
||||
utils::error::raise_api_path_error(function_name, get_api_path(),
|
||||
fsi_.source_path, res,
|
||||
"failed to set file meta");
|
||||
"failed to set new source path");
|
||||
}
|
||||
|
||||
return true;
|
||||
|
Reference in New Issue
Block a user