test/bvar_reducer_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 <limits>                           //std::numeric_limits

 #include "bvar/reducer.h"

 #include "butil/time.h"
 #include "butil/macros.h"
 #include "butil/string_printf.h"
 #include "butil/string_splitter.h"

 #include <gtest/gtest.h>

 namespace {
 class ReducerTest : public testing::Test {
 protected:
     void SetUp() {}
     void TearDown() {}
 };

 TEST_F(ReducerTest, atomicity) {
     ASSERT_EQ(sizeof(int32_t), sizeof(bvar::detail::ElementContainer<int32_t>));
     ASSERT_EQ(sizeof(int64_t), sizeof(bvar::detail::ElementContainer<int64_t>));
     ASSERT_EQ(sizeof(float), sizeof(bvar::detail::ElementContainer<float>));
     ASSERT_EQ(sizeof(double), sizeof(bvar::detail::ElementContainer<double>));
 }

 TEST_F(ReducerTest, adder) {
     bvar::Adder<uint32_t> reducer1;
     ASSERT_TRUE(reducer1.valid());
     reducer1 << 2 << 4;
     ASSERT_EQ(6ul, reducer1.get_value());
 #ifdef BAIDU_INTERNAL
     boost::any v1;
     reducer1.get_value(&v1);
     ASSERT_EQ(6u, boost::any_cast<unsigned int>(v1));
 #endif

     bvar::Adder<double> reducer2;
     ASSERT_TRUE(reducer2.valid());
     reducer2 << 2.0 << 4.0;
     ASSERT_DOUBLE_EQ(6.0, reducer2.get_value());

     bvar::Adder<int> reducer3;
     ASSERT_TRUE(reducer3.valid());
     reducer3 << -9 << 1 << 0 << 3;
     ASSERT_EQ(-5, reducer3.get_value());
 }

 const size_t OPS_PER_THREAD = 500000;

 static void *thread_counter(void *arg) {
     bvar::Adder<uint64_t> *reducer = (bvar::Adder<uint64_t> *)arg;
     butil::Timer timer;
     timer.start();
     for (size_t i = 0; i < OPS_PER_THREAD; ++i) {
         (*reducer) << 2;
     }
     timer.stop();
     return (void *)(timer.n_elapsed());
 }

 void *add_atomic(void *arg) {
     butil::atomic<uint64_t> *counter = (butil::atomic<uint64_t> *)arg;
     butil::Timer timer;
     timer.start();
     for (size_t i = 0; i < OPS_PER_THREAD / 100; ++i) {
         counter->fetch_add(2, butil::memory_order_relaxed);
     }
     timer.stop();
     return (void *)(timer.n_elapsed());
 }

 static long start_perf_test_with_atomic(size_t num_thread) {
     butil::atomic<uint64_t> counter(0);
     pthread_t threads[num_thread];
     for (size_t i = 0; i < num_thread; ++i) {
         pthread_create(&threads[i], NULL, &add_atomic, (void *)&counter);
     }
     long totol_time = 0;
     for (size_t i = 0; i < num_thread; ++i) {
         void *ret;
         pthread_join(threads[i], &ret);
         totol_time += (long)ret;
     }
     long avg_time = totol_time / (OPS_PER_THREAD / 100 * num_thread);
     EXPECT_EQ(2ul * num_thread * OPS_PER_THREAD / 100, counter.load());
     return avg_time;
 }

 static long start_perf_test_with_adder(size_t num_thread) {
     bvar::Adder<uint64_t> reducer;
     EXPECT_TRUE(reducer.valid());
     pthread_t threads[num_thread];
     for (size_t i = 0; i < num_thread; ++i) {
         pthread_create(&threads[i], NULL, &thread_counter, (void *)&reducer);
     }
     long totol_time = 0;
     for (size_t i = 0; i < num_thread; ++i) {
         void *ret = NULL;
         pthread_join(threads[i], &ret);
         totol_time += (long)ret;
     }
     long avg_time = totol_time / (OPS_PER_THREAD * num_thread);
     EXPECT_EQ(2ul * num_thread * OPS_PER_THREAD, reducer.get_value());
     return avg_time;
 }

 TEST_F(ReducerTest, perf) {
     std::ostringstream oss;
     for (size_t i = 1; i <= 24; ++i) {
         oss << i << '\t' << start_perf_test_with_adder(i) << '\n';
     }
     LOG(INFO) << "Adder performance:\n" << oss.str();
     oss.str("");
     for (size_t i = 1; i <= 24; ++i) {
         oss << i << '\t' << start_perf_test_with_atomic(i) << '\n';
     }
     LOG(INFO) << "Atomic performance:\n" << oss.str();
 }

 TEST_F(ReducerTest, Min) {
     bvar::Miner<uint64_t> reducer;
     ASSERT_EQ(std::numeric_limits<uint64_t>::max(), reducer.get_value());
     reducer << 10 << 20;
     ASSERT_EQ(10ul, reducer.get_value());
     reducer << 5;
     ASSERT_EQ(5ul, reducer.get_value());
     reducer << std::numeric_limits<uint64_t>::max();
     ASSERT_EQ(5ul, reducer.get_value());
     reducer << 0;
     ASSERT_EQ(0ul, reducer.get_value());

     bvar::Miner<int> reducer2;
     ASSERT_EQ(std::numeric_limits<int>::max(), reducer2.get_value());
     reducer2 << 10 << 20;
     ASSERT_EQ(10, reducer2.get_value());
     reducer2 << -5;
     ASSERT_EQ(-5, reducer2.get_value());
     reducer2 << std::numeric_limits<int>::max();
     ASSERT_EQ(-5, reducer2.get_value());
     reducer2 << 0;
     ASSERT_EQ(-5, reducer2.get_value());
     reducer2 << std::numeric_limits<int>::min();
     ASSERT_EQ(std::numeric_limits<int>::min(), reducer2.get_value());
 }

 TEST_F(ReducerTest, max) {
     bvar::Maxer<uint64_t> reducer;
     ASSERT_EQ(std::numeric_limits<uint64_t>::min(), reducer.get_value());
     ASSERT_TRUE(reducer.valid());
     reducer << 20 << 10;
     ASSERT_EQ(20ul, reducer.get_value());
     reducer << 30;
     ASSERT_EQ(30ul, reducer.get_value());
     reducer << 0;
     ASSERT_EQ(30ul, reducer.get_value());

     bvar::Maxer<int> reducer2;
     ASSERT_EQ(std::numeric_limits<int>::min(), reducer2.get_value());
     ASSERT_TRUE(reducer2.valid());
     reducer2 << 20 << 10;
     ASSERT_EQ(20, reducer2.get_value());
     reducer2 << 30;
     ASSERT_EQ(30, reducer2.get_value());
     reducer2 << 0;
     ASSERT_EQ(30, reducer2.get_value());
     reducer2 << std::numeric_limits<int>::max();
     ASSERT_EQ(std::numeric_limits<int>::max(), reducer2.get_value());
 }

 bvar::Adder<long> g_a;

 TEST_F(ReducerTest, global) {
     ASSERT_TRUE(g_a.valid());
     g_a.get_value();
 }

 void ReducerTest_window() {
     bvar::Adder<int> c1;
     bvar::Maxer<int> c2;
     bvar::Miner<int> c3;
     bvar::Window<bvar::Adder<int> > w1(&c1, 1);
     bvar::Window<bvar::Adder<int> > w2(&c1, 2);
     bvar::Window<bvar::Adder<int> > w3(&c1, 3);
     bvar::Window<bvar::Maxer<int> > w4(&c2, 1);
     bvar::Window<bvar::Maxer<int> > w5(&c2, 2);
     bvar::Window<bvar::Maxer<int> > w6(&c2, 3);
     bvar::Window<bvar::Miner<int> > w7(&c3, 1);
     bvar::Window<bvar::Miner<int> > w8(&c3, 2);
     bvar::Window<bvar::Miner<int> > w9(&c3, 3);

 #if !BRPC_WITH_GLOG
     logging::StringSink log_str;
     logging::LogSink* old_sink = logging::SetLogSink(&log_str);
     c2.get_value();
     ASSERT_EQ(&log_str, logging::SetLogSink(old_sink));
     ASSERT_NE(std::string::npos, log_str.find(
                   "You should not call Reducer<int, bvar::detail::MaxTo<int>>"
                   "::get_value() when a Window<> is used because the operator"
                   " does not have inverse."));
 #endif
     const int N = 6000;
     int count = 0;
     int total_count = 0;
     int64_t last_time = butil::gettimeofday_us();
     for (int i = 1; i <= N; ++i) {
         c1 << 1;
         c2 << N - i;
         c3 << i;
         ++count;
         ++total_count;
         int64_t now = butil::gettimeofday_us();
         if (now - last_time >= 1000000L) {
             last_time = now;
             ASSERT_EQ(total_count, c1.get_value());
             LOG(INFO) << "c1=" << total_count
                       << " count=" << count
                       << " w1=" << w1
                       << " w2=" << w2
                       << " w3=" << w3
                       << " w4=" << w4
                       << " w5=" << w5
                       << " w6=" << w6
                       << " w7=" << w7
                       << " w8=" << w8
                       << " w9=" << w9;
             count = 0;
         } else {
             usleep(950);
         }
     }
 }

 TEST_F(ReducerTest, window) {
 #if !BRPC_WITH_GLOG
     ReducerTest_window();
     logging::StringSink log_str;
     logging::LogSink* old_sink = logging::SetLogSink(&log_str);
     sleep(1);
     ASSERT_EQ(&log_str, logging::SetLogSink(old_sink));
     if (log_str.find("Removed ") != std::string::npos) {
         ASSERT_NE(std::string::npos, log_str.find("Removed 3, sampled 0")) << log_str;
     }
 #endif
 }

 struct Foo {
     int x;
     Foo() : x(0) {}
     explicit Foo(int x2) : x(x2) {}
     void operator+=(const Foo& rhs) {
         x += rhs.x;
     }
 };

 std::ostream& operator<<(std::ostream& os, const Foo& f) {
     return os << "Foo{" << f.x << "}";
 }

 TEST_F(ReducerTest, non_primitive) {
     bvar::Adder<Foo> adder;
     adder << Foo(2) << Foo(3) << Foo(4);
     ASSERT_EQ(9, adder.get_value().x);
 }

 bool g_stop = false;
 struct StringAppenderResult {
     int count;
 };

 static void* string_appender(void* arg) {
     bvar::Adder<std::string>* cater = (bvar::Adder<std::string>*)arg;
     int count = 0;
     std::string id = butil::string_printf("%lld", (long long)pthread_self());
     std::string tmp = "a";
     for (count = 0; !count || !g_stop; ++count) {
         *cater << id << ":";
         for (char c = 'a'; c <= 'z'; ++c) {
             tmp[0] = c;
             *cater << tmp;
         }
         *cater << ".";
     }
     StringAppenderResult* res = new StringAppenderResult;
     res->count = count;
     LOG(INFO) << "Appended " << count;
     return res;
 }

 TEST_F(ReducerTest, non_primitive_mt) {
     bvar::Adder<std::string> cater;
     pthread_t th[8];
     g_stop = false;
     for (size_t i = 0; i < arraysize(th); ++i) {
         pthread_create(&th[i], NULL, string_appender, &cater);
     }
     usleep(50000);
     g_stop = true;
     butil::hash_map<pthread_t, int> appended_count;
     for (size_t i = 0; i < arraysize(th); ++i) {
         StringAppenderResult* res = NULL;
         pthread_join(th[i], (void**)&res);
         appended_count[th[i]] = res->count;
         delete res;
     }
     butil::hash_map<pthread_t, int> got_count;
     std::string res = cater.get_value();
     for (butil::StringSplitter sp(res.c_str(), '.'); sp; ++sp) {
         char* endptr = NULL;
         ++got_count[(pthread_t)strtoll(sp.field(), &endptr, 10)];
         ASSERT_EQ(27LL, sp.field() + sp.length() - endptr)
             << butil::StringPiece(sp.field(), sp.length());
         ASSERT_EQ(0, memcmp(":abcdefghijklmnopqrstuvwxyz", endptr, 27));
     }
     ASSERT_EQ(appended_count.size(), got_count.size());
     for (size_t i = 0; i < arraysize(th); ++i) {
         ASSERT_EQ(appended_count[th[i]], got_count[th[i]]);
     }
 }

 TEST_F(ReducerTest, simple_window) {
     bvar::Adder<int64_t> a;
     bvar::Window<bvar::Adder<int64_t> > w(&a, 10);
     a << 100;
     sleep(3);
     const int64_t v = w.get_value();
     ASSERT_EQ(100, v) << "v=" << v;
 }
 } // 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 <limits> //std::numeric_limits

	#include "bvar/reducer.h"

	#include "butil/time.h"
	#include "butil/macros.h"
	#include "butil/string_printf.h"
	#include "butil/string_splitter.h"

	#include <gtest/gtest.h>

	namespace {
	class ReducerTest : public testing::Test {
	protected:
	void SetUp() {}
	void TearDown() {}
	};

	TEST_F(ReducerTest, atomicity) {
	ASSERT_EQ(sizeof(int32_t), sizeof(bvar::detail::ElementContainer<int32_t>));
	ASSERT_EQ(sizeof(int64_t), sizeof(bvar::detail::ElementContainer<int64_t>));
	ASSERT_EQ(sizeof(float), sizeof(bvar::detail::ElementContainer<float>));
	ASSERT_EQ(sizeof(double), sizeof(bvar::detail::ElementContainer<double>));
	}

	TEST_F(ReducerTest, adder) {
	bvar::Adder<uint32_t> reducer1;
	ASSERT_TRUE(reducer1.valid());
	reducer1 << 2 << 4;
	ASSERT_EQ(6ul, reducer1.get_value());
	#ifdef BAIDU_INTERNAL
	boost::any v1;
	reducer1.get_value(&v1);
	ASSERT_EQ(6u, boost::any_cast<unsigned int>(v1));
	#endif

	bvar::Adder<double> reducer2;
	ASSERT_TRUE(reducer2.valid());
	reducer2 << 2.0 << 4.0;
	ASSERT_DOUBLE_EQ(6.0, reducer2.get_value());

	bvar::Adder<int> reducer3;
	ASSERT_TRUE(reducer3.valid());
	reducer3 << -9 << 1 << 0 << 3;
	ASSERT_EQ(-5, reducer3.get_value());
	}

	const size_t OPS_PER_THREAD = 500000;

	static void thread_counter(void arg) {
	bvar::Adder<uint64_t> reducer = (bvar::Adder<uint64_t> )arg;
	butil::Timer timer;
	timer.start();
	for (size_t i = 0; i < OPS_PER_THREAD; ++i) {
	(*reducer) << 2;
	}
	timer.stop();
	return (void *)(timer.n_elapsed());
	}

	void add_atomic(void arg) {
	butil::atomic<uint64_t> counter = (butil::atomic<uint64_t> )arg;
	butil::Timer timer;
	timer.start();
	for (size_t i = 0; i < OPS_PER_THREAD / 100; ++i) {
	counter->fetch_add(2, butil::memory_order_relaxed);
	}
	timer.stop();
	return (void *)(timer.n_elapsed());
	}

	static long start_perf_test_with_atomic(size_t num_thread) {
	butil::atomic<uint64_t> counter(0);
	pthread_t threads[num_thread];
	for (size_t i = 0; i < num_thread; ++i) {
	pthread_create(&threads[i], NULL, &add_atomic, (void *)&counter);
	}
	long totol_time = 0;
	for (size_t i = 0; i < num_thread; ++i) {
	void *ret;
	pthread_join(threads[i], &ret);
	totol_time += (long)ret;
	}
	long avg_time = totol_time / (OPS_PER_THREAD / 100 * num_thread);
	EXPECT_EQ(2ul * num_thread * OPS_PER_THREAD / 100, counter.load());
	return avg_time;
	}

	static long start_perf_test_with_adder(size_t num_thread) {
	bvar::Adder<uint64_t> reducer;
	EXPECT_TRUE(reducer.valid());
	pthread_t threads[num_thread];
	for (size_t i = 0; i < num_thread; ++i) {
	pthread_create(&threads[i], NULL, &thread_counter, (void *)&reducer);
	}
	long totol_time = 0;
	for (size_t i = 0; i < num_thread; ++i) {
	void *ret = NULL;
	pthread_join(threads[i], &ret);
	totol_time += (long)ret;
	}
	long avg_time = totol_time / (OPS_PER_THREAD * num_thread);
	EXPECT_EQ(2ul * num_thread * OPS_PER_THREAD, reducer.get_value());
	return avg_time;
	}

	TEST_F(ReducerTest, perf) {
	std::ostringstream oss;
	for (size_t i = 1; i <= 24; ++i) {
	oss << i << '\t' << start_perf_test_with_adder(i) << '\n';
	}
	LOG(INFO) << "Adder performance:\n" << oss.str();
	oss.str("");
	for (size_t i = 1; i <= 24; ++i) {
	oss << i << '\t' << start_perf_test_with_atomic(i) << '\n';
	}
	LOG(INFO) << "Atomic performance:\n" << oss.str();
	}

	TEST_F(ReducerTest, Min) {
	bvar::Miner<uint64_t> reducer;
	ASSERT_EQ(std::numeric_limits<uint64_t>::max(), reducer.get_value());
	reducer << 10 << 20;
	ASSERT_EQ(10ul, reducer.get_value());
	reducer << 5;
	ASSERT_EQ(5ul, reducer.get_value());
	reducer << std::numeric_limits<uint64_t>::max();
	ASSERT_EQ(5ul, reducer.get_value());
	reducer << 0;
	ASSERT_EQ(0ul, reducer.get_value());

	bvar::Miner<int> reducer2;
	ASSERT_EQ(std::numeric_limits<int>::max(), reducer2.get_value());
	reducer2 << 10 << 20;
	ASSERT_EQ(10, reducer2.get_value());
	reducer2 << -5;
	ASSERT_EQ(-5, reducer2.get_value());
	reducer2 << std::numeric_limits<int>::max();
	ASSERT_EQ(-5, reducer2.get_value());
	reducer2 << 0;
	ASSERT_EQ(-5, reducer2.get_value());
	reducer2 << std::numeric_limits<int>::min();
	ASSERT_EQ(std::numeric_limits<int>::min(), reducer2.get_value());
	}

	TEST_F(ReducerTest, max) {
	bvar::Maxer<uint64_t> reducer;
	ASSERT_EQ(std::numeric_limits<uint64_t>::min(), reducer.get_value());
	ASSERT_TRUE(reducer.valid());
	reducer << 20 << 10;
	ASSERT_EQ(20ul, reducer.get_value());
	reducer << 30;
	ASSERT_EQ(30ul, reducer.get_value());
	reducer << 0;
	ASSERT_EQ(30ul, reducer.get_value());

	bvar::Maxer<int> reducer2;
	ASSERT_EQ(std::numeric_limits<int>::min(), reducer2.get_value());
	ASSERT_TRUE(reducer2.valid());
	reducer2 << 20 << 10;
	ASSERT_EQ(20, reducer2.get_value());
	reducer2 << 30;
	ASSERT_EQ(30, reducer2.get_value());
	reducer2 << 0;
	ASSERT_EQ(30, reducer2.get_value());
	reducer2 << std::numeric_limits<int>::max();
	ASSERT_EQ(std::numeric_limits<int>::max(), reducer2.get_value());
	}

	bvar::Adder<long> g_a;

	TEST_F(ReducerTest, global) {
	ASSERT_TRUE(g_a.valid());
	g_a.get_value();
	}

	void ReducerTest_window() {
	bvar::Adder<int> c1;
	bvar::Maxer<int> c2;
	bvar::Miner<int> c3;
	bvar::Window<bvar::Adder<int> > w1(&c1, 1);
	bvar::Window<bvar::Adder<int> > w2(&c1, 2);
	bvar::Window<bvar::Adder<int> > w3(&c1, 3);
	bvar::Window<bvar::Maxer<int> > w4(&c2, 1);
	bvar::Window<bvar::Maxer<int> > w5(&c2, 2);
	bvar::Window<bvar::Maxer<int> > w6(&c2, 3);
	bvar::Window<bvar::Miner<int> > w7(&c3, 1);
	bvar::Window<bvar::Miner<int> > w8(&c3, 2);
	bvar::Window<bvar::Miner<int> > w9(&c3, 3);

	#if !BRPC_WITH_GLOG
	logging::StringSink log_str;
	logging::LogSink* old_sink = logging::SetLogSink(&log_str);
	c2.get_value();
	ASSERT_EQ(&log_str, logging::SetLogSink(old_sink));
	ASSERT_NE(std::string::npos, log_str.find(
	"You should not call Reducer<int, bvar::detail::MaxTo<int>>"
	"::get_value() when a Window<> is used because the operator"
	" does not have inverse."));
	#endif
	const int N = 6000;
	int count = 0;
	int total_count = 0;
	int64_t last_time = butil::gettimeofday_us();
	for (int i = 1; i <= N; ++i) {
	c1 << 1;
	c2 << N - i;
	c3 << i;
	++count;
	++total_count;
	int64_t now = butil::gettimeofday_us();
	if (now - last_time >= 1000000L) {
	last_time = now;
	ASSERT_EQ(total_count, c1.get_value());
	LOG(INFO) << "c1=" << total_count
	<< " count=" << count
	<< " w1=" << w1
	<< " w2=" << w2
	<< " w3=" << w3
	<< " w4=" << w4
	<< " w5=" << w5
	<< " w6=" << w6
	<< " w7=" << w7
	<< " w8=" << w8
	<< " w9=" << w9;
	count = 0;
	} else {
	usleep(950);
	}
	}
	}

	TEST_F(ReducerTest, window) {
	#if !BRPC_WITH_GLOG
	ReducerTest_window();
	logging::StringSink log_str;
	logging::LogSink* old_sink = logging::SetLogSink(&log_str);
	sleep(1);
	ASSERT_EQ(&log_str, logging::SetLogSink(old_sink));
	if (log_str.find("Removed ") != std::string::npos) {
	ASSERT_NE(std::string::npos, log_str.find("Removed 3, sampled 0")) << log_str;
	}
	#endif
	}

	struct Foo {
	int x;
	Foo() : x(0) {}
	explicit Foo(int x2) : x(x2) {}
	void operator+=(const Foo& rhs) {
	x += rhs.x;
	}
	};

	std::ostream& operator<<(std::ostream& os, const Foo& f) {
	return os << "Foo{" << f.x << "}";
	}

	TEST_F(ReducerTest, non_primitive) {
	bvar::Adder<Foo> adder;
	adder << Foo(2) << Foo(3) << Foo(4);
	ASSERT_EQ(9, adder.get_value().x);
	}

	bool g_stop = false;
	struct StringAppenderResult {
	int count;
	};

	static void* string_appender(void* arg) {
	bvar::Adder<std::string>* cater = (bvar::Adder<std::string>*)arg;
	int count = 0;
	std::string id = butil::string_printf("%lld", (long long)pthread_self());
	std::string tmp = "a";
	for (count = 0; !count \|\| !g_stop; ++count) {
	*cater << id << ":";
	for (char c = 'a'; c <= 'z'; ++c) {
	tmp[0] = c;
	*cater << tmp;
	}
	*cater << ".";
	}
	StringAppenderResult* res = new StringAppenderResult;
	res->count = count;
	LOG(INFO) << "Appended " << count;
	return res;
	}

	TEST_F(ReducerTest, non_primitive_mt) {
	bvar::Adder<std::string> cater;
	pthread_t th[8];
	g_stop = false;
	for (size_t i = 0; i < arraysize(th); ++i) {
	pthread_create(&th[i], NULL, string_appender, &cater);
	}
	usleep(50000);
	g_stop = true;
	butil::hash_map<pthread_t, int> appended_count;
	for (size_t i = 0; i < arraysize(th); ++i) {
	StringAppenderResult* res = NULL;
	pthread_join(th[i], (void**)&res);
	appended_count[th[i]] = res->count;
	delete res;
	}
	butil::hash_map<pthread_t, int> got_count;
	std::string res = cater.get_value();
	for (butil::StringSplitter sp(res.c_str(), '.'); sp; ++sp) {
	char* endptr = NULL;
	++got_count[(pthread_t)strtoll(sp.field(), &endptr, 10)];
	ASSERT_EQ(27LL, sp.field() + sp.length() - endptr)
	<< butil::StringPiece(sp.field(), sp.length());
	ASSERT_EQ(0, memcmp(":abcdefghijklmnopqrstuvwxyz", endptr, 27));
	}
	ASSERT_EQ(appended_count.size(), got_count.size());
	for (size_t i = 0; i < arraysize(th); ++i) {
	ASSERT_EQ(appended_count[th[i]], got_count[th[i]]);
	}
	}

	TEST_F(ReducerTest, simple_window) {
	bvar::Adder<int64_t> a;
	bvar::Window<bvar::Adder<int64_t> > w(&a, 10);
	a << 100;
	sleep(3);
	const int64_t v = w.get_value();
	ASSERT_EQ(100, v) << "v=" << v;
	}
	} // namespace