src/server/test/pegasus_write_service_test.cpp - incubator-pegasus - Git at Google

 /*
  * 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.
  */

 #include <dsn/utility/fail_point.h>
 #include "base/pegasus_key_schema.h"
 #include "pegasus_server_test_base.h"
 #include "server/pegasus_server_write.h"
 #include "server/pegasus_write_service_impl.h"
 #include "message_utils.h"

 namespace pegasus {
 namespace server {

 class pegasus_write_service_test : public pegasus_server_test_base
 {
 protected:
     pegasus_write_service *_write_svc{nullptr};
     std::unique_ptr<pegasus_server_write> _server_write;

 public:
     pegasus_write_service_test() = default;

     void SetUp() override
     {
         start();
         _server_write = dsn::make_unique<pegasus_server_write>(_server.get(), true);
         _write_svc = _server_write->_write_svc.get();
     }

     void test_multi_put()
     {
         dsn::fail::setup();

         dsn::apps::multi_put_request request;
         dsn::apps::update_response response;

         int64_t decree = 10;
         std::string hash_key = "hash_key";

         // alarm for empty request
         request.hash_key = dsn::blob(hash_key.data(), 0, hash_key.size());
         auto ctx = db_write_context::create(decree, 1000);
         int err = _write_svc->multi_put(ctx, request, response);
         ASSERT_EQ(err, 0);
         verify_response(response, rocksdb::Status::kInvalidArgument, decree);

         constexpr int kv_num = 100;
         std::string sort_key[kv_num];
         std::string value[kv_num];

         for (int i = 0; i < 100; i++) {
             sort_key[i] = "sort_key_" + std::to_string(i);
             value[i] = "value_" + std::to_string(i);
         }

         for (int i = 0; i < 100; i++) {
             request.kvs.emplace_back();
             request.kvs.back().key.assign(sort_key[i].data(), 0, sort_key[i].size());
             request.kvs.back().value.assign(value[i].data(), 0, value[i].size());
         }

         {
             dsn::fail::cfg("db_write_batch_put", "100%1*return()");
             err = _write_svc->multi_put(ctx, request, response);
             ASSERT_EQ(err, FAIL_DB_WRITE_BATCH_PUT);
             verify_response(response, err, decree);
         }

         {
             dsn::fail::cfg("db_write", "100%1*return()");
             err = _write_svc->multi_put(ctx, request, response);
             ASSERT_EQ(err, FAIL_DB_WRITE);
             verify_response(response, err, decree);
         }

         { // success
             err = _write_svc->multi_put(ctx, request, response);
             ASSERT_EQ(err, 0);
             verify_response(response, 0, decree);
         }

         dsn::fail::teardown();
     }

     void test_multi_remove()
     {
         dsn::fail::setup();

         dsn::apps::multi_remove_request request;
         dsn::apps::multi_remove_response response;

         int64_t decree = 10;
         std::string hash_key = "hash_key";

         // alarm for empty request
         request.hash_key = dsn::blob(hash_key.data(), 0, hash_key.size());
         int err = _write_svc->multi_remove(decree, request, response);
         ASSERT_EQ(err, 0);
         verify_response(response, rocksdb::Status::kInvalidArgument, decree);

         constexpr int kv_num = 100;
         std::string sort_key[kv_num];

         for (int i = 0; i < kv_num; i++) {
             sort_key[i] = "sort_key_" + std::to_string(i);
         }

         for (int i = 0; i < kv_num; i++) {
             request.sort_keys.emplace_back();
             request.sort_keys.back().assign(sort_key[i].data(), 0, sort_key[i].size());
         }

         {
             dsn::fail::cfg("db_write_batch_delete", "100%1*return()");
             err = _write_svc->multi_remove(decree, request, response);
             ASSERT_EQ(err, FAIL_DB_WRITE_BATCH_DELETE);
             verify_response(response, err, decree);
         }

         {
             dsn::fail::cfg("db_write", "100%1*return()");
             err = _write_svc->multi_remove(decree, request, response);
             ASSERT_EQ(err, FAIL_DB_WRITE);
             verify_response(response, err, decree);
         }

         { // success
             err = _write_svc->multi_remove(decree, request, response);
             ASSERT_EQ(err, 0);
             verify_response(response, 0, decree);
         }

         dsn::fail::teardown();
     }

     void test_batched_writes()
     {
         int64_t decree = 10;
         std::string hash_key = "hash_key";

         auto ctx = db_write_context::create(decree, 1000);

         constexpr int kv_num = 100;
         dsn::blob key[kv_num];
         std::string value[kv_num];

         for (int i = 0; i < kv_num; i++) {
             std::string sort_key = "sort_key_" + std::to_string(i);
             pegasus::pegasus_generate_key(key[i], hash_key, sort_key);

             value[i] = "value_" + std::to_string(i);
         }

         // It's dangerous to use std::vector<> here, since the address
         // of response may be changed due to capacity increase.
         std::array<dsn::apps::update_response, kv_num> responses;
         {
             _write_svc->batch_prepare(decree);
             for (int i = 0; i < kv_num; i++) {
                 dsn::apps::update_request req;
                 req.key = key[i];
                 _write_svc->batch_put(ctx, req, responses[i]);
             }
             for (int i = 0; i < kv_num; i++) {
                 _write_svc->batch_remove(decree, key[i], responses[i]);
             }
             _write_svc->batch_commit(decree);
         }

         for (const dsn::apps::update_response &resp : responses) {
             verify_response(resp, 0, decree);
         }
     }

     template <typename TResponse>
     void verify_response(const TResponse &response, int err, int64_t decree)
     {
         ASSERT_EQ(response.error, err);
         ASSERT_EQ(response.app_id, _gpid.get_app_id());
         ASSERT_EQ(response.partition_index, _gpid.get_partition_index());
         ASSERT_EQ(response.decree, decree);
         ASSERT_EQ(response.server, _write_svc->_impl->_primary_address);
         ASSERT_EQ(_write_svc->_impl->_rocksdb_wrapper->_write_batch->Count(), 0);
         ASSERT_EQ(_write_svc->_impl->_update_responses.size(), 0);
     }
 };

 TEST_F(pegasus_write_service_test, multi_put) { test_multi_put(); }

 TEST_F(pegasus_write_service_test, multi_remove) { test_multi_remove(); }

 TEST_F(pegasus_write_service_test, batched_writes) { test_batched_writes(); }

 TEST_F(pegasus_write_service_test, duplicate_not_batched)
 {
     std::string hash_key = "hash_key";
     constexpr int kv_num = 100;
     std::string sort_key[kv_num];
     std::string value[kv_num];

     for (int i = 0; i < 100; i++) {
         sort_key[i] = "sort_key_" + std::to_string(i);
         value[i] = "value_" + std::to_string(i);
     }

     dsn::apps::duplicate_request duplicate;
     duplicate.timestamp = 1000;
     duplicate.cluster_id = 2;
     dsn::apps::duplicate_response resp;

     {
         dsn::apps::multi_put_request mput;
         for (int i = 0; i < 100; i++) {
             mput.kvs.emplace_back();
             mput.kvs.back().key.assign(sort_key[i].data(), 0, sort_key[i].size());
             mput.kvs.back().value.assign(value[i].data(), 0, value[i].size());
         }
         dsn::message_ptr mput_msg = pegasus::create_multi_put_request(mput);

         duplicate.task_code = dsn::apps::RPC_RRDB_RRDB_MULTI_PUT;
         duplicate.raw_message = dsn::move_message_to_blob(mput_msg.get());

         _write_svc->duplicate(1, duplicate, resp);
         ASSERT_EQ(resp.error, 0);
     }

     {
         dsn::apps::multi_remove_request mremove;
         for (int i = 0; i < 100; i++) {
             mremove.sort_keys.emplace_back();
             mremove.sort_keys.back().assign(sort_key[i].data(), 0, sort_key[i].size());
         }
         dsn::message_ptr mremove_msg = pegasus::create_multi_remove_request(mremove);

         duplicate.task_code = dsn::apps::RPC_RRDB_RRDB_MULTI_REMOVE;
         duplicate.raw_message = dsn::move_message_to_blob(mremove_msg.get());

         _write_svc->duplicate(1, duplicate, resp);
         ASSERT_EQ(resp.error, 0);
     }
 }

 TEST_F(pegasus_write_service_test, duplicate_batched)
 {
     std::string hash_key = "hash_key";
     constexpr int kv_num = 100;
     std::string sort_key[kv_num];
     std::string value[kv_num];

     for (int i = 0; i < 100; i++) {
         sort_key[i] = "sort_key_" + std::to_string(i);
         value[i] = "value_" + std::to_string(i);
     }

     {
         dsn::apps::duplicate_request duplicate;
         duplicate.timestamp = 1000;
         duplicate.cluster_id = 2;
         dsn::apps::duplicate_response resp;

         for (int i = 0; i < kv_num; i++) {
             dsn::apps::update_request request;
             pegasus::pegasus_generate_key(request.key, hash_key, sort_key[i]);
             request.value.assign(value[i].data(), 0, value[i].size());

             dsn::message_ptr msg_ptr = pegasus::create_put_request(request);
             duplicate.raw_message = dsn::move_message_to_blob(msg_ptr.get());
             duplicate.task_code = dsn::apps::RPC_RRDB_RRDB_PUT;
             _write_svc->duplicate(1, duplicate, resp);
             ASSERT_EQ(resp.error, 0);
         }
     }
 }

 TEST_F(pegasus_write_service_test, illegal_duplicate_request)
 {
     std::string hash_key = "hash_key";
     std::string sort_key = "sort_key";
     std::string value = "value";

     // cluster=13 is from nowhere
     dsn::apps::duplicate_request duplicate;
     duplicate.cluster_id = 13;
     duplicate.timestamp = 10;
     dsn::apps::duplicate_response resp;

     dsn::apps::update_request request;
     pegasus::pegasus_generate_key(request.key, hash_key, sort_key);
     request.value.assign(value.data(), 0, value.size());

     dsn::message_ptr msg_ptr = pegasus::create_put_request(request); // auto release memory
     duplicate.raw_message = dsn::move_message_to_blob(msg_ptr.get());
     duplicate.task_code = dsn::apps::RPC_RRDB_RRDB_PUT;
     _write_svc->duplicate(1, duplicate, resp);
     ASSERT_EQ(resp.error, rocksdb::Status::kInvalidArgument);
 }

 } // namespace server
 } // namespace pegasus
	/*
	* 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.
	*/

	#include <dsn/utility/fail_point.h>
	#include "base/pegasus_key_schema.h"
	#include "pegasus_server_test_base.h"
	#include "server/pegasus_server_write.h"
	#include "server/pegasus_write_service_impl.h"
	#include "message_utils.h"

	namespace pegasus {
	namespace server {

	class pegasus_write_service_test : public pegasus_server_test_base
	{
	protected:
	pegasus_write_service *_write_svc{nullptr};
	std::unique_ptr<pegasus_server_write> _server_write;

	public:
	pegasus_write_service_test() = default;

	void SetUp() override
	{
	start();
	_server_write = dsn::make_unique<pegasus_server_write>(_server.get(), true);
	_write_svc = _server_write->_write_svc.get();
	}

	void test_multi_put()
	{
	dsn::fail::setup();

	dsn::apps::multi_put_request request;
	dsn::apps::update_response response;

	int64_t decree = 10;
	std::string hash_key = "hash_key";

	// alarm for empty request
	request.hash_key = dsn::blob(hash_key.data(), 0, hash_key.size());
	auto ctx = db_write_context::create(decree, 1000);
	int err = _write_svc->multi_put(ctx, request, response);
	ASSERT_EQ(err, 0);
	verify_response(response, rocksdb::Status::kInvalidArgument, decree);

	constexpr int kv_num = 100;
	std::string sort_key[kv_num];
	std::string value[kv_num];

	for (int i = 0; i < 100; i++) {
	sort_key[i] = "sort_key_" + std::to_string(i);
	value[i] = "value_" + std::to_string(i);
	}

	for (int i = 0; i < 100; i++) {
	request.kvs.emplace_back();
	request.kvs.back().key.assign(sort_key[i].data(), 0, sort_key[i].size());
	request.kvs.back().value.assign(value[i].data(), 0, value[i].size());
	}

	{
	dsn::fail::cfg("db_write_batch_put", "100%1*return()");
	err = _write_svc->multi_put(ctx, request, response);
	ASSERT_EQ(err, FAIL_DB_WRITE_BATCH_PUT);
	verify_response(response, err, decree);
	}

	{
	dsn::fail::cfg("db_write", "100%1*return()");
	err = _write_svc->multi_put(ctx, request, response);
	ASSERT_EQ(err, FAIL_DB_WRITE);
	verify_response(response, err, decree);
	}

	{ // success
	err = _write_svc->multi_put(ctx, request, response);
	ASSERT_EQ(err, 0);
	verify_response(response, 0, decree);
	}

	dsn::fail::teardown();
	}

	void test_multi_remove()
	{
	dsn::fail::setup();

	dsn::apps::multi_remove_request request;
	dsn::apps::multi_remove_response response;

	int64_t decree = 10;
	std::string hash_key = "hash_key";

	// alarm for empty request
	request.hash_key = dsn::blob(hash_key.data(), 0, hash_key.size());
	int err = _write_svc->multi_remove(decree, request, response);
	ASSERT_EQ(err, 0);
	verify_response(response, rocksdb::Status::kInvalidArgument, decree);

	constexpr int kv_num = 100;
	std::string sort_key[kv_num];

	for (int i = 0; i < kv_num; i++) {
	sort_key[i] = "sort_key_" + std::to_string(i);
	}

	for (int i = 0; i < kv_num; i++) {
	request.sort_keys.emplace_back();
	request.sort_keys.back().assign(sort_key[i].data(), 0, sort_key[i].size());
	}

	{
	dsn::fail::cfg("db_write_batch_delete", "100%1*return()");
	err = _write_svc->multi_remove(decree, request, response);
	ASSERT_EQ(err, FAIL_DB_WRITE_BATCH_DELETE);
	verify_response(response, err, decree);
	}

	{
	dsn::fail::cfg("db_write", "100%1*return()");
	err = _write_svc->multi_remove(decree, request, response);
	ASSERT_EQ(err, FAIL_DB_WRITE);
	verify_response(response, err, decree);
	}

	{ // success
	err = _write_svc->multi_remove(decree, request, response);
	ASSERT_EQ(err, 0);
	verify_response(response, 0, decree);
	}

	dsn::fail::teardown();
	}

	void test_batched_writes()
	{
	int64_t decree = 10;
	std::string hash_key = "hash_key";

	auto ctx = db_write_context::create(decree, 1000);

	constexpr int kv_num = 100;
	dsn::blob key[kv_num];
	std::string value[kv_num];

	for (int i = 0; i < kv_num; i++) {
	std::string sort_key = "sort_key_" + std::to_string(i);
	pegasus::pegasus_generate_key(key[i], hash_key, sort_key);

	value[i] = "value_" + std::to_string(i);
	}

	// It's dangerous to use std::vector<> here, since the address
	// of response may be changed due to capacity increase.
	std::array<dsn::apps::update_response, kv_num> responses;
	{
	_write_svc->batch_prepare(decree);
	for (int i = 0; i < kv_num; i++) {
	dsn::apps::update_request req;
	req.key = key[i];
	_write_svc->batch_put(ctx, req, responses[i]);
	}
	for (int i = 0; i < kv_num; i++) {
	_write_svc->batch_remove(decree, key[i], responses[i]);
	}
	_write_svc->batch_commit(decree);
	}

	for (const dsn::apps::update_response &resp : responses) {
	verify_response(resp, 0, decree);
	}
	}

	template <typename TResponse>
	void verify_response(const TResponse &response, int err, int64_t decree)
	{
	ASSERT_EQ(response.error, err);
	ASSERT_EQ(response.app_id, _gpid.get_app_id());
	ASSERT_EQ(response.partition_index, _gpid.get_partition_index());
	ASSERT_EQ(response.decree, decree);
	ASSERT_EQ(response.server, _write_svc->_impl->_primary_address);
	ASSERT_EQ(_write_svc->_impl->_rocksdb_wrapper->_write_batch->Count(), 0);
	ASSERT_EQ(_write_svc->_impl->_update_responses.size(), 0);
	}
	};

	TEST_F(pegasus_write_service_test, multi_put) { test_multi_put(); }

	TEST_F(pegasus_write_service_test, multi_remove) { test_multi_remove(); }

	TEST_F(pegasus_write_service_test, batched_writes) { test_batched_writes(); }

	TEST_F(pegasus_write_service_test, duplicate_not_batched)
	{
	std::string hash_key = "hash_key";
	constexpr int kv_num = 100;
	std::string sort_key[kv_num];
	std::string value[kv_num];

	for (int i = 0; i < 100; i++) {
	sort_key[i] = "sort_key_" + std::to_string(i);
	value[i] = "value_" + std::to_string(i);
	}

	dsn::apps::duplicate_request duplicate;
	duplicate.timestamp = 1000;
	duplicate.cluster_id = 2;
	dsn::apps::duplicate_response resp;

	{
	dsn::apps::multi_put_request mput;
	for (int i = 0; i < 100; i++) {
	mput.kvs.emplace_back();
	mput.kvs.back().key.assign(sort_key[i].data(), 0, sort_key[i].size());
	mput.kvs.back().value.assign(value[i].data(), 0, value[i].size());
	}
	dsn::message_ptr mput_msg = pegasus::create_multi_put_request(mput);

	duplicate.task_code = dsn::apps::RPC_RRDB_RRDB_MULTI_PUT;
	duplicate.raw_message = dsn::move_message_to_blob(mput_msg.get());

	_write_svc->duplicate(1, duplicate, resp);
	ASSERT_EQ(resp.error, 0);
	}

	{
	dsn::apps::multi_remove_request mremove;
	for (int i = 0; i < 100; i++) {
	mremove.sort_keys.emplace_back();
	mremove.sort_keys.back().assign(sort_key[i].data(), 0, sort_key[i].size());
	}
	dsn::message_ptr mremove_msg = pegasus::create_multi_remove_request(mremove);

	duplicate.task_code = dsn::apps::RPC_RRDB_RRDB_MULTI_REMOVE;
	duplicate.raw_message = dsn::move_message_to_blob(mremove_msg.get());

	_write_svc->duplicate(1, duplicate, resp);
	ASSERT_EQ(resp.error, 0);
	}
	}

	TEST_F(pegasus_write_service_test, duplicate_batched)
	{
	std::string hash_key = "hash_key";
	constexpr int kv_num = 100;
	std::string sort_key[kv_num];
	std::string value[kv_num];

	for (int i = 0; i < 100; i++) {
	sort_key[i] = "sort_key_" + std::to_string(i);
	value[i] = "value_" + std::to_string(i);
	}

	{
	dsn::apps::duplicate_request duplicate;
	duplicate.timestamp = 1000;
	duplicate.cluster_id = 2;
	dsn::apps::duplicate_response resp;

	for (int i = 0; i < kv_num; i++) {
	dsn::apps::update_request request;
	pegasus::pegasus_generate_key(request.key, hash_key, sort_key[i]);
	request.value.assign(value[i].data(), 0, value[i].size());

	dsn::message_ptr msg_ptr = pegasus::create_put_request(request);
	duplicate.raw_message = dsn::move_message_to_blob(msg_ptr.get());
	duplicate.task_code = dsn::apps::RPC_RRDB_RRDB_PUT;
	_write_svc->duplicate(1, duplicate, resp);
	ASSERT_EQ(resp.error, 0);
	}
	}
	}

	TEST_F(pegasus_write_service_test, illegal_duplicate_request)
	{
	std::string hash_key = "hash_key";
	std::string sort_key = "sort_key";
	std::string value = "value";

	// cluster=13 is from nowhere
	dsn::apps::duplicate_request duplicate;
	duplicate.cluster_id = 13;
	duplicate.timestamp = 10;
	dsn::apps::duplicate_response resp;

	dsn::apps::update_request request;
	pegasus::pegasus_generate_key(request.key, hash_key, sort_key);
	request.value.assign(value.data(), 0, value.size());

	dsn::message_ptr msg_ptr = pegasus::create_put_request(request); // auto release memory
	duplicate.raw_message = dsn::move_message_to_blob(msg_ptr.get());
	duplicate.task_code = dsn::apps::RPC_RRDB_RRDB_PUT;
	_write_svc->duplicate(1, duplicate, resp);
	ASSERT_EQ(resp.error, rocksdb::Status::kInvalidArgument);
	}

	} // namespace server
	} // namespace pegasus