| /* |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance |
| * with the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, |
| * software distributed under the License is distributed on an |
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| * KIND, either express or implied. See the License for the |
| * specific language governing permissions and limitations |
| * under the License. |
| */ |
| |
| #pragma once |
| |
| #include "pegasus_write_service.h" |
| #include "pegasus_server_impl.h" |
| #include "logging_utils.h" |
| |
| #include "base/pegasus_key_schema.h" |
| #include "meta_store.h" |
| #include "rocksdb_wrapper.h" |
| |
| #include <dsn/utility/filesystem.h> |
| #include <dsn/utility/string_conv.h> |
| #include <gtest/gtest_prod.h> |
| #include <dsn/utility/defer.h> |
| |
| namespace pegasus { |
| namespace server { |
| |
| /// internal error codes used for fail injection |
| static constexpr int FAIL_DB_WRITE_BATCH_PUT = -101; |
| static constexpr int FAIL_DB_WRITE_BATCH_DELETE = -102; |
| static constexpr int FAIL_DB_WRITE = -103; |
| static constexpr int FAIL_DB_GET = -104; |
| |
| struct db_get_context |
| { |
| // value read from DB. |
| std::string raw_value; |
| |
| // is the record found in DB. |
| bool found{false}; |
| |
| // the expiration time encoded in raw_value. |
| uint32_t expire_ts{0}; |
| |
| // is the record expired. |
| bool expired{false}; |
| }; |
| |
| inline int get_cluster_id_if_exists() |
| { |
| // cluster_id is 0 if not configured, which means it will accept writes |
| // from any cluster as long as the timestamp is larger. |
| static auto cluster_id_res = |
| dsn::replication::get_duplication_cluster_id(dsn::replication::get_current_cluster_name()); |
| static uint64_t cluster_id = cluster_id_res.is_ok() ? cluster_id_res.get_value() : 0; |
| return cluster_id; |
| } |
| |
| inline dsn::error_code get_external_files_path(const std::string &bulk_load_dir, |
| const dsn::replication::bulk_load_metadata &metadata, |
| /*out*/ std::vector<std::string> &files_path) |
| { |
| for (const auto &f_meta : metadata.files) { |
| const std::string &file_name = |
| dsn::utils::filesystem::path_combine(bulk_load_dir, f_meta.name); |
| if (dsn::utils::filesystem::verify_file(file_name, f_meta.md5, f_meta.size)) { |
| files_path.emplace_back(file_name); |
| } else { |
| break; |
| } |
| } |
| return files_path.size() == metadata.files.size() ? dsn::ERR_OK : dsn::ERR_WRONG_CHECKSUM; |
| } |
| |
| class pegasus_write_service::impl : public dsn::replication::replica_base |
| { |
| public: |
| explicit impl(pegasus_server_impl *server) |
| : replica_base(server), |
| _primary_address(server->_primary_address), |
| _pegasus_data_version(server->_pegasus_data_version), |
| _pfc_recent_expire_count(server->_pfc_recent_expire_count) |
| { |
| _rocksdb_wrapper = dsn::make_unique<rocksdb_wrapper>(server); |
| } |
| |
| int empty_put(int64_t decree) |
| { |
| int err = |
| _rocksdb_wrapper->write_batch_put(decree, dsn::string_view(), dsn::string_view(), 0); |
| auto cleanup = dsn::defer([this]() { _rocksdb_wrapper->clear_up_write_batch(); }); |
| if (err) { |
| return err; |
| } |
| |
| err = _rocksdb_wrapper->write(decree); |
| return err; |
| } |
| |
| int multi_put(const db_write_context &ctx, |
| const dsn::apps::multi_put_request &update, |
| dsn::apps::update_response &resp) |
| { |
| int64_t decree = ctx.decree; |
| resp.app_id = get_gpid().get_app_id(); |
| resp.partition_index = get_gpid().get_partition_index(); |
| resp.decree = decree; |
| resp.server = _primary_address; |
| |
| if (update.kvs.empty()) { |
| derror_replica("invalid argument for multi_put: decree = {}, error = {}", |
| decree, |
| "request.kvs is empty"); |
| resp.error = rocksdb::Status::kInvalidArgument; |
| // we should write empty record to update rocksdb's last flushed decree |
| return empty_put(decree); |
| } |
| |
| auto cleanup = dsn::defer([this]() { _rocksdb_wrapper->clear_up_write_batch(); }); |
| for (auto &kv : update.kvs) { |
| resp.error = _rocksdb_wrapper->write_batch_put_ctx( |
| ctx, |
| composite_raw_key(update.hash_key, kv.key), |
| kv.value, |
| static_cast<uint32_t>(update.expire_ts_seconds)); |
| if (resp.error) { |
| return resp.error; |
| } |
| } |
| |
| resp.error = _rocksdb_wrapper->write(decree); |
| return resp.error; |
| } |
| |
| int multi_remove(int64_t decree, |
| const dsn::apps::multi_remove_request &update, |
| dsn::apps::multi_remove_response &resp) |
| { |
| resp.app_id = get_gpid().get_app_id(); |
| resp.partition_index = get_gpid().get_partition_index(); |
| resp.decree = decree; |
| resp.server = _primary_address; |
| |
| if (update.sort_keys.empty()) { |
| derror_replica("invalid argument for multi_remove: decree = {}, error = {}", |
| decree, |
| "request.sort_keys is empty"); |
| resp.error = rocksdb::Status::kInvalidArgument; |
| // we should write empty record to update rocksdb's last flushed decree |
| return empty_put(decree); |
| } |
| |
| auto cleanup = dsn::defer([this]() { _rocksdb_wrapper->clear_up_write_batch(); }); |
| for (auto &sort_key : update.sort_keys) { |
| resp.error = _rocksdb_wrapper->write_batch_delete( |
| decree, composite_raw_key(update.hash_key, sort_key)); |
| if (resp.error) { |
| return resp.error; |
| } |
| } |
| |
| resp.error = _rocksdb_wrapper->write(decree); |
| if (resp.error == 0) { |
| resp.count = update.sort_keys.size(); |
| } |
| return resp.error; |
| } |
| |
| int incr(int64_t decree, const dsn::apps::incr_request &update, dsn::apps::incr_response &resp) |
| { |
| resp.app_id = get_gpid().get_app_id(); |
| resp.partition_index = get_gpid().get_partition_index(); |
| resp.decree = decree; |
| resp.server = _primary_address; |
| |
| dsn::string_view raw_key(update.key.data(), update.key.length()); |
| int64_t new_value = 0; |
| uint32_t new_expire_ts = 0; |
| db_get_context get_ctx; |
| int err = _rocksdb_wrapper->get(raw_key, &get_ctx); |
| if (err != 0) { |
| resp.error = err; |
| return err; |
| } |
| if (!get_ctx.found) { |
| // old value is not found, set to 0 before increment |
| new_value = update.increment; |
| new_expire_ts = update.expire_ts_seconds > 0 ? update.expire_ts_seconds : 0; |
| } else if (get_ctx.expired) { |
| // ttl timeout, set to 0 before increment |
| new_value = update.increment; |
| new_expire_ts = update.expire_ts_seconds > 0 ? update.expire_ts_seconds : 0; |
| } else { |
| ::dsn::blob old_value; |
| pegasus_extract_user_data( |
| _pegasus_data_version, std::move(get_ctx.raw_value), old_value); |
| if (old_value.length() == 0) { |
| // empty old value, set to 0 before increment |
| new_value = update.increment; |
| } else { |
| int64_t old_value_int; |
| if (!dsn::buf2int64(old_value, old_value_int)) { |
| // invalid old value |
| derror_replica("incr failed: decree = {}, error = " |
| "old value \"{}\" is not an integer or out of range", |
| decree, |
| utils::c_escape_string(old_value)); |
| resp.error = rocksdb::Status::kInvalidArgument; |
| // we should write empty record to update rocksdb's last flushed decree |
| return empty_put(decree); |
| } |
| new_value = old_value_int + update.increment; |
| if ((update.increment > 0 && new_value < old_value_int) || |
| (update.increment < 0 && new_value > old_value_int)) { |
| // new value is out of range, return old value by 'new_value' |
| derror_replica("incr failed: decree = {}, error = " |
| "new value is out of range, old_value = {}, increment = {}", |
| decree, |
| old_value_int, |
| update.increment); |
| resp.error = rocksdb::Status::kInvalidArgument; |
| resp.new_value = old_value_int; |
| // we should write empty record to update rocksdb's last flushed decree |
| return empty_put(decree); |
| } |
| } |
| // set new ttl |
| if (update.expire_ts_seconds == 0) { |
| new_expire_ts = get_ctx.expire_ts; |
| } else if (update.expire_ts_seconds < 0) { |
| new_expire_ts = 0; |
| } else { // update.expire_ts_seconds > 0 |
| new_expire_ts = update.expire_ts_seconds; |
| } |
| } |
| |
| auto cleanup = dsn::defer([this]() { _rocksdb_wrapper->clear_up_write_batch(); }); |
| resp.error = _rocksdb_wrapper->write_batch_put( |
| decree, update.key, std::to_string(new_value), new_expire_ts); |
| if (resp.error) { |
| return resp.error; |
| } |
| |
| resp.error = _rocksdb_wrapper->write(decree); |
| if (resp.error == 0) { |
| resp.new_value = new_value; |
| } |
| return resp.error; |
| } |
| |
| int check_and_set(int64_t decree, |
| const dsn::apps::check_and_set_request &update, |
| dsn::apps::check_and_set_response &resp) |
| { |
| resp.app_id = get_gpid().get_app_id(); |
| resp.partition_index = get_gpid().get_partition_index(); |
| resp.decree = decree; |
| resp.server = _primary_address; |
| |
| if (!is_check_type_supported(update.check_type)) { |
| derror_replica("invalid argument for check_and_set: decree = {}, error = {}", |
| decree, |
| "check type {} not supported", |
| update.check_type); |
| resp.error = rocksdb::Status::kInvalidArgument; |
| // we should write empty record to update rocksdb's last flushed decree |
| return empty_put(decree); |
| } |
| |
| ::dsn::blob check_key; |
| pegasus_generate_key(check_key, update.hash_key, update.check_sort_key); |
| |
| db_get_context get_context; |
| dsn::string_view check_raw_key(check_key.data(), check_key.length()); |
| int err = _rocksdb_wrapper->get(check_raw_key, &get_context); |
| if (err != 0) { |
| // read check value failed |
| derror_rocksdb("Error to GetCheckValue for CheckAndSet decree: {}, hash_key: {}, " |
| "check_sort_key: {}", |
| decree, |
| utils::c_escape_string(update.hash_key), |
| utils::c_escape_string(update.check_sort_key)); |
| resp.error = err; |
| return resp.error; |
| } |
| |
| ::dsn::blob check_value; |
| bool value_exist = !get_context.expired && get_context.found; |
| if (value_exist) { |
| pegasus_extract_user_data( |
| _pegasus_data_version, std::move(get_context.raw_value), check_value); |
| } |
| |
| if (update.return_check_value) { |
| resp.check_value_returned = true; |
| if (value_exist) { |
| resp.check_value_exist = true; |
| resp.check_value = check_value; |
| } |
| } |
| |
| bool invalid_argument = false; |
| bool passed = validate_check(decree, |
| update.check_type, |
| update.check_operand, |
| value_exist, |
| check_value, |
| invalid_argument); |
| |
| if (passed) { |
| // check passed, write new value |
| ::dsn::blob set_key; |
| if (update.set_diff_sort_key) { |
| pegasus_generate_key(set_key, update.hash_key, update.set_sort_key); |
| } else { |
| set_key = check_key; |
| } |
| resp.error = _rocksdb_wrapper->write_batch_put( |
| decree, |
| set_key, |
| update.set_value, |
| static_cast<uint32_t>(update.set_expire_ts_seconds)); |
| } else { |
| // check not passed, write empty record to update rocksdb's last flushed decree |
| resp.error = _rocksdb_wrapper->write_batch_put( |
| decree, dsn::string_view(), dsn::string_view(), 0); |
| } |
| |
| auto cleanup = dsn::defer([this]() { _rocksdb_wrapper->clear_up_write_batch(); }); |
| if (resp.error) { |
| return resp.error; |
| } |
| |
| resp.error = _rocksdb_wrapper->write(decree); |
| if (resp.error) { |
| return resp.error; |
| } |
| |
| if (!passed) { |
| // check not passed, return proper error code to user |
| resp.error = |
| invalid_argument ? rocksdb::Status::kInvalidArgument : rocksdb::Status::kTryAgain; |
| } |
| |
| return 0; |
| } |
| |
| int check_and_mutate(int64_t decree, |
| const dsn::apps::check_and_mutate_request &update, |
| dsn::apps::check_and_mutate_response &resp) |
| { |
| resp.app_id = get_gpid().get_app_id(); |
| resp.partition_index = get_gpid().get_partition_index(); |
| resp.decree = decree; |
| resp.server = _primary_address; |
| |
| if (update.mutate_list.empty()) { |
| derror_replica("invalid argument for check_and_mutate: decree = {}, error = {}", |
| decree, |
| "mutate list is empty"); |
| resp.error = rocksdb::Status::kInvalidArgument; |
| // we should write empty record to update rocksdb's last flushed decree |
| return empty_put(decree); |
| } |
| |
| for (int i = 0; i < update.mutate_list.size(); ++i) { |
| auto &mu = update.mutate_list[i]; |
| if (mu.operation != ::dsn::apps::mutate_operation::MO_PUT && |
| mu.operation != ::dsn::apps::mutate_operation::MO_DELETE) { |
| derror_replica("invalid argument for check_and_mutate: decree = {}, error = " |
| "mutation[{}] uses invalid operation {}", |
| decree, |
| i, |
| mu.operation); |
| resp.error = rocksdb::Status::kInvalidArgument; |
| // we should write empty record to update rocksdb's last flushed decree |
| return empty_put(decree); |
| } |
| } |
| |
| if (!is_check_type_supported(update.check_type)) { |
| derror_replica("invalid argument for check_and_mutate: decree = {}, error = {}", |
| decree, |
| "check type {} not supported", |
| update.check_type); |
| resp.error = rocksdb::Status::kInvalidArgument; |
| // we should write empty record to update rocksdb's last flushed decree |
| return empty_put(decree); |
| } |
| |
| ::dsn::blob check_key; |
| pegasus_generate_key(check_key, update.hash_key, update.check_sort_key); |
| |
| db_get_context get_context; |
| dsn::string_view check_raw_key(check_key.data(), check_key.length()); |
| int err = _rocksdb_wrapper->get(check_raw_key, &get_context); |
| if (err != 0) { |
| // read check value failed |
| derror_rocksdb("Error to GetCheckValue for CheckAndMutate decree: {}, hash_key: {}, " |
| "check_sort_key: {}", |
| decree, |
| utils::c_escape_string(update.hash_key), |
| utils::c_escape_string(update.check_sort_key)); |
| resp.error = err; |
| return resp.error; |
| } |
| |
| ::dsn::blob check_value; |
| bool value_exist = !get_context.expired && get_context.found; |
| if (value_exist) { |
| pegasus_extract_user_data( |
| _pegasus_data_version, std::move(get_context.raw_value), check_value); |
| } |
| |
| if (update.return_check_value) { |
| resp.check_value_returned = true; |
| if (value_exist) { |
| resp.check_value_exist = true; |
| resp.check_value = check_value; |
| } |
| } |
| |
| bool invalid_argument = false; |
| bool passed = validate_check(decree, |
| update.check_type, |
| update.check_operand, |
| value_exist, |
| check_value, |
| invalid_argument); |
| |
| if (passed) { |
| for (auto &m : update.mutate_list) { |
| ::dsn::blob key; |
| pegasus_generate_key(key, update.hash_key, m.sort_key); |
| if (m.operation == ::dsn::apps::mutate_operation::MO_PUT) { |
| resp.error = _rocksdb_wrapper->write_batch_put( |
| decree, key, m.value, static_cast<uint32_t>(m.set_expire_ts_seconds)); |
| } else { |
| dassert_f(m.operation == ::dsn::apps::mutate_operation::MO_DELETE, |
| "m.operation = %d", |
| m.operation); |
| resp.error = _rocksdb_wrapper->write_batch_delete(decree, key); |
| } |
| |
| // in case of failure, cancel mutations |
| if (resp.error) |
| break; |
| } |
| } else { |
| // check not passed, write empty record to update rocksdb's last flushed decree |
| resp.error = _rocksdb_wrapper->write_batch_put( |
| decree, dsn::string_view(), dsn::string_view(), 0); |
| } |
| |
| auto cleanup = dsn::defer([this]() { _rocksdb_wrapper->clear_up_write_batch(); }); |
| if (resp.error) { |
| return resp.error; |
| } |
| |
| resp.error = _rocksdb_wrapper->write(decree); |
| if (resp.error) { |
| return resp.error; |
| } |
| |
| if (!passed) { |
| // check not passed, return proper error code to user |
| resp.error = |
| invalid_argument ? rocksdb::Status::kInvalidArgument : rocksdb::Status::kTryAgain; |
| } |
| return 0; |
| } |
| |
| // \return ERR_WRONG_CHECKSUM: verify files failed |
| // \return ERR_INGESTION_FAILED: rocksdb ingestion failed |
| // \return ERR_OK: rocksdb ingestion succeed |
| dsn::error_code ingestion_files(const int64_t decree, |
| const std::string &bulk_load_dir, |
| const dsn::replication::bulk_load_metadata &metadata) |
| { |
| // verify external files before ingestion |
| std::vector<std::string> sst_file_list; |
| dsn::error_code err = get_external_files_path(bulk_load_dir, metadata, sst_file_list); |
| if (err != dsn::ERR_OK) { |
| return err; |
| } |
| |
| // ingest external files |
| if (dsn_unlikely(_rocksdb_wrapper->ingestion_files(decree, sst_file_list) != 0)) { |
| return dsn::ERR_INGESTION_FAILED; |
| } |
| return dsn::ERR_OK; |
| } |
| |
| /// For batch write. |
| |
| int batch_put(const db_write_context &ctx, |
| const dsn::apps::update_request &update, |
| dsn::apps::update_response &resp) |
| { |
| resp.error = _rocksdb_wrapper->write_batch_put_ctx( |
| ctx, update.key, update.value, static_cast<uint32_t>(update.expire_ts_seconds)); |
| _update_responses.emplace_back(&resp); |
| return resp.error; |
| } |
| |
| int batch_remove(int64_t decree, const dsn::blob &key, dsn::apps::update_response &resp) |
| { |
| resp.error = _rocksdb_wrapper->write_batch_delete(decree, key); |
| _update_responses.emplace_back(&resp); |
| return resp.error; |
| } |
| |
| int batch_commit(int64_t decree) |
| { |
| int err = _rocksdb_wrapper->write(decree); |
| clear_up_batch_states(decree, err); |
| return err; |
| } |
| |
| void batch_abort(int64_t decree, int err) { clear_up_batch_states(decree, err); } |
| |
| void set_default_ttl(uint32_t ttl) { _rocksdb_wrapper->set_default_ttl(ttl); } |
| |
| private: |
| void clear_up_batch_states(int64_t decree, int err) |
| { |
| if (!_update_responses.empty()) { |
| dsn::apps::update_response resp; |
| resp.error = err; |
| resp.app_id = get_gpid().get_app_id(); |
| resp.partition_index = get_gpid().get_partition_index(); |
| resp.decree = decree; |
| resp.server = _primary_address; |
| for (dsn::apps::update_response *uresp : _update_responses) { |
| *uresp = resp; |
| } |
| _update_responses.clear(); |
| } |
| |
| _rocksdb_wrapper->clear_up_write_batch(); |
| } |
| |
| static dsn::blob composite_raw_key(dsn::string_view hash_key, dsn::string_view sort_key) |
| { |
| dsn::blob raw_key; |
| pegasus_generate_key(raw_key, hash_key, sort_key); |
| return raw_key; |
| } |
| |
| // return true if the check type is supported |
| static bool is_check_type_supported(::dsn::apps::cas_check_type::type check_type) |
| { |
| return check_type >= ::dsn::apps::cas_check_type::CT_NO_CHECK && |
| check_type <= ::dsn::apps::cas_check_type::CT_VALUE_INT_GREATER; |
| } |
| |
| // return true if check passed. |
| // for int compare, if check operand or value are not valid integer, then return false, |
| // and set out param `invalid_argument' to false. |
| bool validate_check(int64_t decree, |
| ::dsn::apps::cas_check_type::type check_type, |
| const ::dsn::blob &check_operand, |
| bool value_exist, |
| const ::dsn::blob &value, |
| bool &invalid_argument) |
| { |
| invalid_argument = false; |
| switch (check_type) { |
| case ::dsn::apps::cas_check_type::CT_NO_CHECK: |
| return true; |
| case ::dsn::apps::cas_check_type::CT_VALUE_NOT_EXIST: |
| return !value_exist; |
| case ::dsn::apps::cas_check_type::CT_VALUE_NOT_EXIST_OR_EMPTY: |
| return !value_exist || value.length() == 0; |
| case ::dsn::apps::cas_check_type::CT_VALUE_EXIST: |
| return value_exist; |
| case ::dsn::apps::cas_check_type::CT_VALUE_NOT_EMPTY: |
| return value_exist && value.length() != 0; |
| case ::dsn::apps::cas_check_type::CT_VALUE_MATCH_ANYWHERE: |
| case ::dsn::apps::cas_check_type::CT_VALUE_MATCH_PREFIX: |
| case ::dsn::apps::cas_check_type::CT_VALUE_MATCH_POSTFIX: { |
| if (!value_exist) |
| return false; |
| if (check_operand.length() == 0) |
| return true; |
| if (value.length() < check_operand.length()) |
| return false; |
| if (check_type == ::dsn::apps::cas_check_type::CT_VALUE_MATCH_ANYWHERE) { |
| return dsn::string_view(value).find(check_operand) != dsn::string_view::npos; |
| } else if (check_type == ::dsn::apps::cas_check_type::CT_VALUE_MATCH_PREFIX) { |
| return ::memcmp(value.data(), check_operand.data(), check_operand.length()) == 0; |
| } else { // check_type == ::dsn::apps::cas_check_type::CT_VALUE_MATCH_POSTFIX |
| return ::memcmp(value.data() + value.length() - check_operand.length(), |
| check_operand.data(), |
| check_operand.length()) == 0; |
| } |
| } |
| case ::dsn::apps::cas_check_type::CT_VALUE_BYTES_LESS: |
| case ::dsn::apps::cas_check_type::CT_VALUE_BYTES_LESS_OR_EQUAL: |
| case ::dsn::apps::cas_check_type::CT_VALUE_BYTES_EQUAL: |
| case ::dsn::apps::cas_check_type::CT_VALUE_BYTES_GREATER_OR_EQUAL: |
| case ::dsn::apps::cas_check_type::CT_VALUE_BYTES_GREATER: { |
| if (!value_exist) |
| return false; |
| int c = dsn::string_view(value).compare(dsn::string_view(check_operand)); |
| if (c < 0) { |
| return check_type <= ::dsn::apps::cas_check_type::CT_VALUE_BYTES_LESS_OR_EQUAL; |
| } else if (c == 0) { |
| return check_type >= ::dsn::apps::cas_check_type::CT_VALUE_BYTES_LESS_OR_EQUAL && |
| check_type <= ::dsn::apps::cas_check_type::CT_VALUE_BYTES_GREATER_OR_EQUAL; |
| } else { // c > 0 |
| return check_type >= ::dsn::apps::cas_check_type::CT_VALUE_BYTES_GREATER_OR_EQUAL; |
| } |
| } |
| case ::dsn::apps::cas_check_type::CT_VALUE_INT_LESS: |
| case ::dsn::apps::cas_check_type::CT_VALUE_INT_LESS_OR_EQUAL: |
| case ::dsn::apps::cas_check_type::CT_VALUE_INT_EQUAL: |
| case ::dsn::apps::cas_check_type::CT_VALUE_INT_GREATER_OR_EQUAL: |
| case ::dsn::apps::cas_check_type::CT_VALUE_INT_GREATER: { |
| if (!value_exist) |
| return false; |
| int64_t check_value_int; |
| if (!dsn::buf2int64(value, check_value_int)) { |
| // invalid check value |
| derror_replica("check failed: decree = {}, error = " |
| "check value \"{}\" is not an integer or out of range", |
| decree, |
| utils::c_escape_string(value)); |
| invalid_argument = true; |
| return false; |
| } |
| int64_t check_operand_int; |
| if (!dsn::buf2int64(check_operand, check_operand_int)) { |
| // invalid check operand |
| derror_replica("check failed: decree = {}, error = " |
| "check operand \"{}\" is not an integer or out of range", |
| decree, |
| utils::c_escape_string(check_operand)); |
| invalid_argument = true; |
| return false; |
| } |
| if (check_value_int < check_operand_int) { |
| return check_type <= ::dsn::apps::cas_check_type::CT_VALUE_INT_LESS_OR_EQUAL; |
| } else if (check_value_int == check_operand_int) { |
| return check_type >= ::dsn::apps::cas_check_type::CT_VALUE_INT_LESS_OR_EQUAL && |
| check_type <= ::dsn::apps::cas_check_type::CT_VALUE_INT_GREATER_OR_EQUAL; |
| } else { // check_value_int > check_operand_int |
| return check_type >= ::dsn::apps::cas_check_type::CT_VALUE_INT_GREATER_OR_EQUAL; |
| } |
| } |
| default: |
| dassert(false, "unsupported check type: %d", check_type); |
| } |
| return false; |
| } |
| |
| private: |
| friend class pegasus_write_service_test; |
| friend class pegasus_server_write_test; |
| friend class pegasus_write_service_impl_test; |
| friend class rocksdb_wrapper_test; |
| FRIEND_TEST(pegasus_write_service_impl_test, put_verify_timetag); |
| FRIEND_TEST(pegasus_write_service_impl_test, verify_timetag_compatible_with_version_0); |
| |
| const std::string _primary_address; |
| const uint32_t _pegasus_data_version; |
| |
| ::dsn::perf_counter_wrapper &_pfc_recent_expire_count; |
| |
| std::unique_ptr<rocksdb_wrapper> _rocksdb_wrapper; |
| |
| // for setting update_response.error after committed. |
| std::vector<dsn::apps::update_response *> _update_responses; |
| }; |
| |
| } // namespace server |
| } // namespace pegasus |
