test/brpc_ch_bounded_load_balancer_unittest.cpp - brpc - 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 <cmath>
 #include <cstdlib>
 #include <cstring>
 #include <map>
 #include <sstream>
 #include <vector>
 #include <gflags/gflags.h>
 #include <gtest/gtest.h>
 #include "butil/macros.h"
 #include "brpc/socket.h"
 #include "brpc/excluded_servers.h"
 #include "brpc/policy/consistent_hashing_load_balancer.h"
 #include "brpc/policy/hasher.h"

 namespace {

 brpc::ServerId CreateServer(const char* addr, const char* tag = "") {
     butil::EndPoint point;
     EXPECT_EQ(0, str2endpoint(addr, &point));
     brpc::ServerId id(8888);
     brpc::SocketOptions options;
     options.remote_side = point;
     EXPECT_EQ(0, brpc::Socket::Create(options, &id.id));
     id.tag = tag;
     return id;
 }

 void DestroyServers(const std::vector<brpc::ServerId>& ids) {
     for (size_t i = 0; i < ids.size(); ++i) {
         brpc::Socket::SetFailed(ids[i].id);
     }
 }

 brpc::LoadBalancer::SelectIn MakeInput(uint64_t code,
                                        bool changable_weights = true) {
     brpc::LoadBalancer::SelectIn in = { 0, changable_weights, true, code, nullptr };
     return in;
 }

 int64_t TotalInflightOf(brpc::LoadBalancer* lb) {
     std::ostringstream os;
     brpc::DescribeOptions opt;
     opt.verbose = true;
     lb->Describe(os, opt);
     const std::string desc = os.str();
     const size_t pos = desc.find("total_inflight=");
     EXPECT_NE(std::string::npos, pos) << desc;
     return strtoll(desc.c_str() + pos + strlen("total_inflight="), nullptr, 10);
 }

 class CHBoundedLoadTest : public testing::Test {};

 TEST_F(CHBoundedLoadTest, load_factor_validation) {
     brpc::policy::ConsistentHashingBoundedLoadBalancer lb(
         brpc::policy::CONS_HASH_LB_MURMUR3);
     ASSERT_EQ(nullptr, lb.New("load_factor=1.0"));
     ASSERT_EQ(nullptr, lb.New("load_factor=0.5"));
     ASSERT_EQ(nullptr, lb.New("load_factor=abc"));
     brpc::LoadBalancer* valid = lb.New("load_factor=1.5");
     ASSERT_TRUE(valid != nullptr);
     valid->Destroy();

     ASSERT_EQ("", GFLAGS_NAMESPACE::SetCommandLineOption(
         "chash_bounded_load_factor", "0.9"));
     ASSERT_EQ("", GFLAGS_NAMESPACE::SetCommandLineOption(
         "chash_bounded_load_factor", "1.0"));
     ASSERT_NE("", GFLAGS_NAMESPACE::SetCommandLineOption(
         "chash_bounded_load_factor", "1.25"));
 }

 TEST_F(CHBoundedLoadTest, replicas_parameter) {
     brpc::policy::ConsistentHashingLoadBalancer classic_lb(
         brpc::policy::CONS_HASH_LB_MURMUR3);
     brpc::policy::ConsistentHashingBoundedLoadBalancer bounded_lb(
         brpc::policy::CONS_HASH_LB_MURMUR3);
     const brpc::LoadBalancer* lbs[] = { &classic_lb, &bounded_lb };
     for (size_t i = 0; i < arraysize(lbs); ++i) {
         ASSERT_EQ(nullptr, lbs[i]->New("replicas=abc"));
         brpc::LoadBalancer* lb = lbs[i]->New("replicas=300");
         ASSERT_TRUE(lb != nullptr);
         std::ostringstream os;
         brpc::DescribeOptions opt;
         opt.verbose = true;
         lb->Describe(os, opt);
         ASSERT_NE(std::string::npos,
                   os.str().find("replica per host: 300")) << os.str();
         lb->Destroy();
     }
     brpc::LoadBalancer* lb = bounded_lb.New("replicas=200 load_factor=2");
     ASSERT_TRUE(lb != nullptr);
     lb->Destroy();
 }

 TEST_F(CHBoundedLoadTest, hot_key_is_capped) {
     const size_t N = 8;
     const size_t K = 200;
     const double FACTOR = 1.25;
     std::vector<brpc::ServerId> ids;
     brpc::policy::ConsistentHashingBoundedLoadBalancer lb(
         brpc::policy::CONS_HASH_LB_MURMUR3);
     brpc::policy::ConsistentHashingLoadBalancer classic_lb(
         brpc::policy::CONS_HASH_LB_MURMUR3);
     for (size_t i = 0; i < N; ++i) {
         char addr[32];
         snprintf(addr, sizeof(addr), "192.168.1.%d:8080", (int)i);
         ids.push_back(CreateServer(addr));
     }
     ASSERT_EQ(N, lb.AddServersInBatch(ids));
     ASSERT_EQ(N, classic_lb.AddServersInBatch(ids));

     const std::string hot_key = "hot_key";
     brpc::LoadBalancer::SelectIn in =
         MakeInput(brpc::policy::MurmurHash32(hot_key.data(), hot_key.size()));

     // Classic CH sends every request for the key to one server.
     std::map<brpc::SocketId, size_t> classic_counts;
     for (size_t i = 0; i < K; ++i) {
         brpc::SocketUniquePtr ptr;
         brpc::LoadBalancer::SelectOut out(&ptr);
         ASSERT_EQ(0, classic_lb.SelectServer(in, &out));
         ++classic_counts[ptr->id()];
     }
     ASSERT_EQ(1UL, classic_counts.size());
     const brpc::SocketId primary = classic_counts.begin()->first;

     // Bounded-load CH spreads the same hot key once the primary server hits
     // its capacity: no server exceeds ceil(FACTOR * K / N) outstanding
     // requests(the cap of the last selection, when total load is largest).
     std::map<brpc::SocketId, size_t> counts;
     for (size_t i = 0; i < K; ++i) {
         brpc::SocketUniquePtr ptr;
         brpc::LoadBalancer::SelectOut out(&ptr);
         ASSERT_EQ(0, lb.SelectServer(in, &out));
         ASSERT_TRUE(out.need_feedback);
         ++counts[ptr->id()];
     }
     const size_t cap = (size_t)std::ceil(FACTOR * K / N);
     size_t max_count = 0;
     for (std::map<brpc::SocketId, size_t>::iterator it = counts.begin();
          it != counts.end(); ++it) {
         max_count = std::max(max_count, it->second);
     }
     ASSERT_LE(max_count, cap);
     ASSERT_GT(counts.size(), 1UL);
     // The first selection matches classic CH: locality is unchanged until
     // the primary server is at capacity.
     ASSERT_GT(counts[primary], 0UL);
     ASSERT_EQ((int64_t)K, TotalInflightOf(&lb));

     DestroyServers(ids);
 }

 TEST_F(CHBoundedLoadTest, overflow_walks_to_ring_successor) {
     const size_t N = 8;
     const size_t K = 200;
     std::vector<brpc::ServerId> ids;
     brpc::policy::ConsistentHashingBoundedLoadBalancer lb(
         brpc::policy::CONS_HASH_LB_MURMUR3);
     brpc::policy::ConsistentHashingLoadBalancer classic_lb(
         brpc::policy::CONS_HASH_LB_MURMUR3);
     std::map<brpc::SocketId, brpc::ServerId> id_map;
     for (size_t i = 0; i < N; ++i) {
         char addr[32];
         snprintf(addr, sizeof(addr), "192.168.1.%d:8080", (int)i);
         ids.push_back(CreateServer(addr));
         id_map[ids.back().id] = ids.back();
     }
     ASSERT_EQ(N, lb.AddServersInBatch(ids));
     ASSERT_EQ(N, classic_lb.AddServersInBatch(ids));

     const std::string hot_key = "another_hot_key";
     brpc::LoadBalancer::SelectIn in =
         MakeInput(brpc::policy::MurmurHash32(hot_key.data(), hot_key.size()));

     // First-use order of servers under overflow.
     std::vector<brpc::SocketId> bounded_order;
     std::map<brpc::SocketId, size_t> seen;
     for (size_t i = 0; i < K; ++i) {
         brpc::SocketUniquePtr ptr;
         brpc::LoadBalancer::SelectOut out(&ptr);
         ASSERT_EQ(0, lb.SelectServer(in, &out));
         if (++seen[ptr->id()] == 1) {
             bounded_order.push_back(ptr->id());
         }
     }
     ASSERT_GT(bounded_order.size(), 1UL);

     // Expected overflow targets are the ring successors: what classic CH
     // picks as servers are removed one by one.
     for (size_t i = 0; i < bounded_order.size(); ++i) {
         brpc::SocketUniquePtr ptr;
         brpc::LoadBalancer::SelectOut out(&ptr);
         ASSERT_EQ(0, classic_lb.SelectServer(in, &out));
         ASSERT_EQ(bounded_order[i], ptr->id()) << "i=" << i;
         ASSERT_TRUE(classic_lb.RemoveServer(id_map[ptr->id()]));
     }

     DestroyServers(ids);
 }

 TEST_F(CHBoundedLoadTest, feedback_decrements_load) {
     const size_t N = 3;
     std::vector<brpc::ServerId> ids;
     brpc::policy::ConsistentHashingBoundedLoadBalancer lb(
         brpc::policy::CONS_HASH_LB_MURMUR3);
     for (size_t i = 0; i < N; ++i) {
         char addr[32];
         snprintf(addr, sizeof(addr), "192.168.1.%d:8080", (int)i);
         ids.push_back(CreateServer(addr));
     }
     ASSERT_EQ(N, lb.AddServersInBatch(ids));

     const std::string key = "some_key";
     brpc::LoadBalancer::SelectIn in =
         MakeInput(brpc::policy::MurmurHash32(key.data(), key.size()));

     // With capacity ceil(1.25 * 1 / 3) = 1 an idle ring always routes the
     // key to its primary server, so select+feedback staying on one server
     // for many rounds proves the counters return to zero every round.
     brpc::SocketId primary = 0;
     for (size_t i = 0; i < 100; ++i) {
         brpc::SocketUniquePtr ptr;
         brpc::LoadBalancer::SelectOut out(&ptr);
         ASSERT_EQ(0, lb.SelectServer(in, &out));
         ASSERT_TRUE(out.need_feedback);
         if (i == 0) {
             primary = ptr->id();
         } else {
             ASSERT_EQ(primary, ptr->id()) << "i=" << i;
         }
         const brpc::LoadBalancer::CallInfo info = { 0, ptr->id(), 0, nullptr };
         lb.Feedback(info);
     }
     ASSERT_EQ(0, TotalInflightOf(&lb));

     // Saturate the primary without feedback: the key overflows...
     std::vector<brpc::SocketId> outstanding;
     bool overflowed = false;
     for (size_t i = 0; i < 20; ++i) {
         brpc::SocketUniquePtr ptr;
         brpc::LoadBalancer::SelectOut out(&ptr);
         ASSERT_EQ(0, lb.SelectServer(in, &out));
         outstanding.push_back(ptr->id());
         overflowed |= (ptr->id() != primary);
     }
     ASSERT_TRUE(overflowed);
     // ...and returns to the primary once the load drains.
     for (size_t i = 0; i < outstanding.size(); ++i) {
         const brpc::LoadBalancer::CallInfo info = { 0, outstanding[i], 0, nullptr };
         lb.Feedback(info);
     }
     ASSERT_EQ(0, TotalInflightOf(&lb));
     brpc::SocketUniquePtr ptr;
     brpc::LoadBalancer::SelectOut out(&ptr);
     ASSERT_EQ(0, lb.SelectServer(in, &out));
     ASSERT_EQ(primary, ptr->id());
     const brpc::LoadBalancer::CallInfo info = { 0, ptr->id(), 0, nullptr };
     lb.Feedback(info);

     DestroyServers(ids);
 }

 TEST_F(CHBoundedLoadTest, feedback_after_server_removed_keeps_total_consistent) {
     const size_t N = 3;
     std::vector<brpc::ServerId> ids;
     brpc::policy::ConsistentHashingBoundedLoadBalancer lb(
         brpc::policy::CONS_HASH_LB_MURMUR3);
     for (size_t i = 0; i < N; ++i) {
         char addr[32];
         snprintf(addr, sizeof(addr), "192.168.1.%d:8080", (int)i);
         ids.push_back(CreateServer(addr));
     }
     ASSERT_EQ(N, lb.AddServersInBatch(ids));

     const std::string key = "some_key";
     brpc::LoadBalancer::SelectIn in =
         MakeInput(brpc::policy::MurmurHash32(key.data(), key.size()));
     brpc::SocketUniquePtr ptr;
     brpc::LoadBalancer::SelectOut out(&ptr);
     ASSERT_EQ(0, lb.SelectServer(in, &out));
     ASSERT_EQ(1, TotalInflightOf(&lb));

     brpc::ServerId selected;
     for (size_t i = 0; i < ids.size(); ++i) {
         if (ids[i].id == ptr->id()) {
             selected = ids[i];
         }
     }
     ASSERT_TRUE(lb.RemoveServer(selected));
     const brpc::LoadBalancer::CallInfo info = { 0, ptr->id(), 0, nullptr };
     lb.Feedback(info);
     ASSERT_EQ(0, TotalInflightOf(&lb));
     ASSERT_TRUE(lb.AddServer(selected));

     DestroyServers(ids);
 }

 TEST_F(CHBoundedLoadTest, no_accounting_without_changable_weights) {
     const size_t N = 4;
     std::vector<brpc::ServerId> ids;
     brpc::policy::ConsistentHashingBoundedLoadBalancer lb(
         brpc::policy::CONS_HASH_LB_MURMUR3);
     for (size_t i = 0; i < N; ++i) {
         char addr[32];
         snprintf(addr, sizeof(addr), "192.168.1.%d:8080", (int)i);
         ids.push_back(CreateServer(addr));
     }
     ASSERT_EQ(N, lb.AddServersInBatch(ids));

     const std::string key = "some_key";
     brpc::LoadBalancer::SelectIn in = MakeInput(
         brpc::policy::MurmurHash32(key.data(), key.size()), false);
     brpc::SocketId first = 0;
     for (size_t i = 0; i < 50; ++i) {
         brpc::SocketUniquePtr ptr;
         brpc::LoadBalancer::SelectOut out(&ptr);
         ASSERT_EQ(0, lb.SelectServer(in, &out));
         ASSERT_FALSE(out.need_feedback);
         if (i == 0) {
             first = ptr->id();
         } else {
             ASSERT_EQ(first, ptr->id());
         }
     }
     ASSERT_EQ(0, TotalInflightOf(&lb));

     DestroyServers(ids);
 }

 } // namespace
	// 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 <cmath>
	#include <cstdlib>
	#include <cstring>
	#include <map>
	#include <sstream>
	#include <vector>
	#include <gflags/gflags.h>
	#include <gtest/gtest.h>
	#include "butil/macros.h"
	#include "brpc/socket.h"
	#include "brpc/excluded_servers.h"
	#include "brpc/policy/consistent_hashing_load_balancer.h"
	#include "brpc/policy/hasher.h"

	namespace {

	brpc::ServerId CreateServer(const char* addr, const char* tag = "") {
	butil::EndPoint point;
	EXPECT_EQ(0, str2endpoint(addr, &point));
	brpc::ServerId id(8888);
	brpc::SocketOptions options;
	options.remote_side = point;
	EXPECT_EQ(0, brpc::Socket::Create(options, &id.id));
	id.tag = tag;
	return id;
	}

	void DestroyServers(const std::vector<brpc::ServerId>& ids) {
	for (size_t i = 0; i < ids.size(); ++i) {
	brpc::Socket::SetFailed(ids[i].id);
	}
	}

	brpc::LoadBalancer::SelectIn MakeInput(uint64_t code,
	bool changable_weights = true) {
	brpc::LoadBalancer::SelectIn in = { 0, changable_weights, true, code, nullptr };
	return in;
	}

	int64_t TotalInflightOf(brpc::LoadBalancer* lb) {
	std::ostringstream os;
	brpc::DescribeOptions opt;
	opt.verbose = true;
	lb->Describe(os, opt);
	const std::string desc = os.str();
	const size_t pos = desc.find("total_inflight=");
	EXPECT_NE(std::string::npos, pos) << desc;
	return strtoll(desc.c_str() + pos + strlen("total_inflight="), nullptr, 10);
	}

	class CHBoundedLoadTest : public testing::Test {};

	TEST_F(CHBoundedLoadTest, load_factor_validation) {
	brpc::policy::ConsistentHashingBoundedLoadBalancer lb(
	brpc::policy::CONS_HASH_LB_MURMUR3);
	ASSERT_EQ(nullptr, lb.New("load_factor=1.0"));
	ASSERT_EQ(nullptr, lb.New("load_factor=0.5"));
	ASSERT_EQ(nullptr, lb.New("load_factor=abc"));
	brpc::LoadBalancer* valid = lb.New("load_factor=1.5");
	ASSERT_TRUE(valid != nullptr);
	valid->Destroy();

	ASSERT_EQ("", GFLAGS_NAMESPACE::SetCommandLineOption(
	"chash_bounded_load_factor", "0.9"));
	ASSERT_EQ("", GFLAGS_NAMESPACE::SetCommandLineOption(
	"chash_bounded_load_factor", "1.0"));
	ASSERT_NE("", GFLAGS_NAMESPACE::SetCommandLineOption(
	"chash_bounded_load_factor", "1.25"));
	}

	TEST_F(CHBoundedLoadTest, replicas_parameter) {
	brpc::policy::ConsistentHashingLoadBalancer classic_lb(
	brpc::policy::CONS_HASH_LB_MURMUR3);
	brpc::policy::ConsistentHashingBoundedLoadBalancer bounded_lb(
	brpc::policy::CONS_HASH_LB_MURMUR3);
	const brpc::LoadBalancer* lbs[] = { &classic_lb, &bounded_lb };
	for (size_t i = 0; i < arraysize(lbs); ++i) {
	ASSERT_EQ(nullptr, lbs[i]->New("replicas=abc"));
	brpc::LoadBalancer* lb = lbs[i]->New("replicas=300");
	ASSERT_TRUE(lb != nullptr);
	std::ostringstream os;
	brpc::DescribeOptions opt;
	opt.verbose = true;
	lb->Describe(os, opt);
	ASSERT_NE(std::string::npos,
	os.str().find("replica per host: 300")) << os.str();
	lb->Destroy();
	}
	brpc::LoadBalancer* lb = bounded_lb.New("replicas=200 load_factor=2");
	ASSERT_TRUE(lb != nullptr);
	lb->Destroy();
	}

	TEST_F(CHBoundedLoadTest, hot_key_is_capped) {
	const size_t N = 8;
	const size_t K = 200;
	const double FACTOR = 1.25;
	std::vector<brpc::ServerId> ids;
	brpc::policy::ConsistentHashingBoundedLoadBalancer lb(
	brpc::policy::CONS_HASH_LB_MURMUR3);
	brpc::policy::ConsistentHashingLoadBalancer classic_lb(
	brpc::policy::CONS_HASH_LB_MURMUR3);
	for (size_t i = 0; i < N; ++i) {
	char addr[32];
	snprintf(addr, sizeof(addr), "192.168.1.%d:8080", (int)i);
	ids.push_back(CreateServer(addr));
	}
	ASSERT_EQ(N, lb.AddServersInBatch(ids));
	ASSERT_EQ(N, classic_lb.AddServersInBatch(ids));

	const std::string hot_key = "hot_key";
	brpc::LoadBalancer::SelectIn in =
	MakeInput(brpc::policy::MurmurHash32(hot_key.data(), hot_key.size()));

	// Classic CH sends every request for the key to one server.
	std::map<brpc::SocketId, size_t> classic_counts;
	for (size_t i = 0; i < K; ++i) {
	brpc::SocketUniquePtr ptr;
	brpc::LoadBalancer::SelectOut out(&ptr);
	ASSERT_EQ(0, classic_lb.SelectServer(in, &out));
	++classic_counts[ptr->id()];
	}
	ASSERT_EQ(1UL, classic_counts.size());
	const brpc::SocketId primary = classic_counts.begin()->first;

	// Bounded-load CH spreads the same hot key once the primary server hits
	// its capacity: no server exceeds ceil(FACTOR * K / N) outstanding
	// requests(the cap of the last selection, when total load is largest).
	std::map<brpc::SocketId, size_t> counts;
	for (size_t i = 0; i < K; ++i) {
	brpc::SocketUniquePtr ptr;
	brpc::LoadBalancer::SelectOut out(&ptr);
	ASSERT_EQ(0, lb.SelectServer(in, &out));
	ASSERT_TRUE(out.need_feedback);
	++counts[ptr->id()];
	}
	const size_t cap = (size_t)std::ceil(FACTOR * K / N);
	size_t max_count = 0;
	for (std::map<brpc::SocketId, size_t>::iterator it = counts.begin();
	it != counts.end(); ++it) {
	max_count = std::max(max_count, it->second);
	}
	ASSERT_LE(max_count, cap);
	ASSERT_GT(counts.size(), 1UL);
	// The first selection matches classic CH: locality is unchanged until
	// the primary server is at capacity.
	ASSERT_GT(counts[primary], 0UL);
	ASSERT_EQ((int64_t)K, TotalInflightOf(&lb));

	DestroyServers(ids);
	}

	TEST_F(CHBoundedLoadTest, overflow_walks_to_ring_successor) {
	const size_t N = 8;
	const size_t K = 200;
	std::vector<brpc::ServerId> ids;
	brpc::policy::ConsistentHashingBoundedLoadBalancer lb(
	brpc::policy::CONS_HASH_LB_MURMUR3);
	brpc::policy::ConsistentHashingLoadBalancer classic_lb(
	brpc::policy::CONS_HASH_LB_MURMUR3);
	std::map<brpc::SocketId, brpc::ServerId> id_map;
	for (size_t i = 0; i < N; ++i) {
	char addr[32];
	snprintf(addr, sizeof(addr), "192.168.1.%d:8080", (int)i);
	ids.push_back(CreateServer(addr));
	id_map[ids.back().id] = ids.back();
	}
	ASSERT_EQ(N, lb.AddServersInBatch(ids));
	ASSERT_EQ(N, classic_lb.AddServersInBatch(ids));

	const std::string hot_key = "another_hot_key";
	brpc::LoadBalancer::SelectIn in =
	MakeInput(brpc::policy::MurmurHash32(hot_key.data(), hot_key.size()));

	// First-use order of servers under overflow.
	std::vector<brpc::SocketId> bounded_order;
	std::map<brpc::SocketId, size_t> seen;
	for (size_t i = 0; i < K; ++i) {
	brpc::SocketUniquePtr ptr;
	brpc::LoadBalancer::SelectOut out(&ptr);
	ASSERT_EQ(0, lb.SelectServer(in, &out));
	if (++seen[ptr->id()] == 1) {
	bounded_order.push_back(ptr->id());
	}
	}
	ASSERT_GT(bounded_order.size(), 1UL);

	// Expected overflow targets are the ring successors: what classic CH
	// picks as servers are removed one by one.
	for (size_t i = 0; i < bounded_order.size(); ++i) {
	brpc::SocketUniquePtr ptr;
	brpc::LoadBalancer::SelectOut out(&ptr);
	ASSERT_EQ(0, classic_lb.SelectServer(in, &out));
	ASSERT_EQ(bounded_order[i], ptr->id()) << "i=" << i;
	ASSERT_TRUE(classic_lb.RemoveServer(id_map[ptr->id()]));
	}

	DestroyServers(ids);
	}

	TEST_F(CHBoundedLoadTest, feedback_decrements_load) {
	const size_t N = 3;
	std::vector<brpc::ServerId> ids;
	brpc::policy::ConsistentHashingBoundedLoadBalancer lb(
	brpc::policy::CONS_HASH_LB_MURMUR3);
	for (size_t i = 0; i < N; ++i) {
	char addr[32];
	snprintf(addr, sizeof(addr), "192.168.1.%d:8080", (int)i);
	ids.push_back(CreateServer(addr));
	}
	ASSERT_EQ(N, lb.AddServersInBatch(ids));

	const std::string key = "some_key";
	brpc::LoadBalancer::SelectIn in =
	MakeInput(brpc::policy::MurmurHash32(key.data(), key.size()));

	// With capacity ceil(1.25 * 1 / 3) = 1 an idle ring always routes the
	// key to its primary server, so select+feedback staying on one server
	// for many rounds proves the counters return to zero every round.
	brpc::SocketId primary = 0;
	for (size_t i = 0; i < 100; ++i) {
	brpc::SocketUniquePtr ptr;
	brpc::LoadBalancer::SelectOut out(&ptr);
	ASSERT_EQ(0, lb.SelectServer(in, &out));
	ASSERT_TRUE(out.need_feedback);
	if (i == 0) {
	primary = ptr->id();
	} else {
	ASSERT_EQ(primary, ptr->id()) << "i=" << i;
	}
	const brpc::LoadBalancer::CallInfo info = { 0, ptr->id(), 0, nullptr };
	lb.Feedback(info);
	}
	ASSERT_EQ(0, TotalInflightOf(&lb));

	// Saturate the primary without feedback: the key overflows...
	std::vector<brpc::SocketId> outstanding;
	bool overflowed = false;
	for (size_t i = 0; i < 20; ++i) {
	brpc::SocketUniquePtr ptr;
	brpc::LoadBalancer::SelectOut out(&ptr);
	ASSERT_EQ(0, lb.SelectServer(in, &out));
	outstanding.push_back(ptr->id());
	overflowed \|= (ptr->id() != primary);
	}
	ASSERT_TRUE(overflowed);
	// ...and returns to the primary once the load drains.
	for (size_t i = 0; i < outstanding.size(); ++i) {
	const brpc::LoadBalancer::CallInfo info = { 0, outstanding[i], 0, nullptr };
	lb.Feedback(info);
	}
	ASSERT_EQ(0, TotalInflightOf(&lb));
	brpc::SocketUniquePtr ptr;
	brpc::LoadBalancer::SelectOut out(&ptr);
	ASSERT_EQ(0, lb.SelectServer(in, &out));
	ASSERT_EQ(primary, ptr->id());
	const brpc::LoadBalancer::CallInfo info = { 0, ptr->id(), 0, nullptr };
	lb.Feedback(info);

	DestroyServers(ids);
	}

	TEST_F(CHBoundedLoadTest, feedback_after_server_removed_keeps_total_consistent) {
	const size_t N = 3;
	std::vector<brpc::ServerId> ids;
	brpc::policy::ConsistentHashingBoundedLoadBalancer lb(
	brpc::policy::CONS_HASH_LB_MURMUR3);
	for (size_t i = 0; i < N; ++i) {
	char addr[32];
	snprintf(addr, sizeof(addr), "192.168.1.%d:8080", (int)i);
	ids.push_back(CreateServer(addr));
	}
	ASSERT_EQ(N, lb.AddServersInBatch(ids));

	const std::string key = "some_key";
	brpc::LoadBalancer::SelectIn in =
	MakeInput(brpc::policy::MurmurHash32(key.data(), key.size()));
	brpc::SocketUniquePtr ptr;
	brpc::LoadBalancer::SelectOut out(&ptr);
	ASSERT_EQ(0, lb.SelectServer(in, &out));
	ASSERT_EQ(1, TotalInflightOf(&lb));

	brpc::ServerId selected;
	for (size_t i = 0; i < ids.size(); ++i) {
	if (ids[i].id == ptr->id()) {
	selected = ids[i];
	}
	}
	ASSERT_TRUE(lb.RemoveServer(selected));
	const brpc::LoadBalancer::CallInfo info = { 0, ptr->id(), 0, nullptr };
	lb.Feedback(info);
	ASSERT_EQ(0, TotalInflightOf(&lb));
	ASSERT_TRUE(lb.AddServer(selected));

	DestroyServers(ids);
	}

	TEST_F(CHBoundedLoadTest, no_accounting_without_changable_weights) {
	const size_t N = 4;
	std::vector<brpc::ServerId> ids;
	brpc::policy::ConsistentHashingBoundedLoadBalancer lb(
	brpc::policy::CONS_HASH_LB_MURMUR3);
	for (size_t i = 0; i < N; ++i) {
	char addr[32];
	snprintf(addr, sizeof(addr), "192.168.1.%d:8080", (int)i);
	ids.push_back(CreateServer(addr));
	}
	ASSERT_EQ(N, lb.AddServersInBatch(ids));

	const std::string key = "some_key";
	brpc::LoadBalancer::SelectIn in = MakeInput(
	brpc::policy::MurmurHash32(key.data(), key.size()), false);
	brpc::SocketId first = 0;
	for (size_t i = 0; i < 50; ++i) {
	brpc::SocketUniquePtr ptr;
	brpc::LoadBalancer::SelectOut out(&ptr);
	ASSERT_EQ(0, lb.SelectServer(in, &out));
	ASSERT_FALSE(out.need_feedback);
	if (i == 0) {
	first = ptr->id();
	} else {
	ASSERT_EQ(first, ptr->id());
	}
	}
	ASSERT_EQ(0, TotalInflightOf(&lb));

	DestroyServers(ids);
	}

	} // namespace