Commit 9e63a755 authored by Marc Celani's avatar Marc Celani Committed by Facebook Github Bot

DigestBuilder for buffering writes to TDigest

Summary: TDigest is designed to process a stream of data in batches. DigestBuilder is a fast, thread safe approach for buffering those writes and building a digest.

Reviewed By: davidtgoldblatt

Differential Revision: D7582745

fbshipit-source-id: 528ad9b21685746887b3a602d433662221fb875a
parent 51946874
......@@ -631,6 +631,7 @@ if (BUILD_TESTS)
TEST openssl_hash_test SOURCES OpenSSLHashTest.cpp
DIRECTORY stats/test/
TEST digest_builder_test SOURCES DigestBuilderTest.cpp
TEST histogram_test SOURCES HistogramTest.cpp
TEST sliding_window_test SOURCES SlidingWindowTest.cpp
TEST tdigest_test SOURCES TDigestTest.cpp
......
......@@ -439,6 +439,8 @@ nobase_follyinclude_HEADERS = \
ssl/detail/OpenSSLThreading.h \
ssl/detail/SSLSessionImpl.h \
stats/detail/Bucket.h \
stats/detail/DigestBuilder-defs.h \
stats/detail/DigestBuilder.h \
stats/detail/SlidingWindow-defs.h \
stats/detail/SlidingWindow.h \
stats/BucketedTimeSeries-defs.h \
......
/*
* Copyright 2018-present Facebook, Inc.
*
* Licensed 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 <folly/stats/detail/DigestBuilder.h>
#include <algorithm>
#include <folly/concurrency/CacheLocality.h>
namespace folly {
namespace detail {
template <typename DigestT>
DigestBuilder<DigestT>::DigestBuilder(size_t bufferSize, size_t digestSize)
: nextPos_(0),
digestSize_(digestSize),
cpuLocalBuffersReady_(false),
buffer_(bufferSize) {}
template <typename DigestT>
DigestT DigestBuilder<DigestT>::buildSyncFree() const {
std::vector<double> values;
std::vector<DigestT> digests;
auto numElems =
std::min(nextPos_.load(std::memory_order_relaxed), buffer_.size());
values.insert(values.end(), buffer_.begin(), buffer_.begin() + numElems);
if (cpuLocalBuffersReady_.load(std::memory_order_relaxed)) {
for (const auto& cpuLocalBuffer : cpuLocalBuffers_) {
if (cpuLocalBuffer.digest) {
digests.push_back(*cpuLocalBuffer.digest);
}
values.insert(
values.end(),
cpuLocalBuffer.buffer.begin(),
cpuLocalBuffer.buffer.end());
}
}
std::sort(values.begin(), values.end());
DigestT digest(digestSize_);
digests.push_back(digest.merge(values));
return DigestT::merge(digests);
}
template <typename DigestT>
void DigestBuilder<DigestT>::append(double value) {
auto pos = nextPos_.load(std::memory_order_relaxed);
if (pos < buffer_.size()) {
pos = nextPos_.fetch_add(1, std::memory_order_relaxed);
if (pos < buffer_.size()) {
buffer_[pos] = value;
if (pos == buffer_.size() - 1) {
// The shared buffer is full. From here on out, appends will go to a
// cpu local.
auto& cl = CacheLocality::system();
cpuLocalBuffers_.resize(cl.numCachesByLevel[0]);
cpuLocalBuffersReady_.store(true, std::memory_order_release);
}
return;
}
}
while (!cpuLocalBuffersReady_.load(std::memory_order_acquire)) {
}
auto which = AccessSpreader<>::current(cpuLocalBuffers_.size());
auto& cpuLocalBuf = cpuLocalBuffers_[which];
SpinLockGuard g(cpuLocalBuf.mutex);
cpuLocalBuf.buffer.push_back(value);
if (cpuLocalBuf.buffer.size() > buffer_.size()) {
std::sort(cpuLocalBuf.buffer.begin(), cpuLocalBuf.buffer.end());
if (!cpuLocalBuf.digest) {
cpuLocalBuf.digest = std::make_unique<DigestT>(digestSize_);
}
*cpuLocalBuf.digest = cpuLocalBuf.digest->merge(cpuLocalBuf.buffer);
cpuLocalBuf.buffer.clear();
}
}
} // namespace detail
} // namespace folly
/*
* Copyright 2018-present Facebook, Inc.
*
* Licensed 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 <memory>
#include <folly/SpinLock.h>
namespace folly {
namespace detail {
/*
* Stat digests, such as TDigest, can be expensive to merge. It is faster to
* buffer writes and merge them in larger chunks. DigestBuilder buffers writes
* to improve performance.
*
* The first bufferSize values are stored in a shared buffer. For cold stats,
* this shared buffer minimizes memory usage at reasonable cpu cost. Warm stats
* will fill the shared buffer, and begin to spill to cpu local buffers. Hot
* stats will merge the cpu local buffers into cpu-local digests.
*/
template <typename DigestT>
class DigestBuilder {
public:
explicit DigestBuilder(size_t bufferSize, size_t digestSize);
/*
* Builds a DigestT from the buffer in a sync free manner. It is the
* responsibility of the caller to synchronize with all appenders.
*/
DigestT buildSyncFree() const;
/*
* Adds a value to the buffer.
*/
void append(double value);
private:
struct alignas(hardware_destructive_interference_size) CpuLocalBuffer {
public:
mutable SpinLock mutex;
std::vector<double> buffer;
std::unique_ptr<DigestT> digest;
};
std::atomic<size_t> nextPos_;
std::vector<CpuLocalBuffer> cpuLocalBuffers_;
size_t digestSize_;
std::atomic<bool> cpuLocalBuffersReady_;
std::vector<double> buffer_;
};
} // namespace detail
} // namespace folly
/*
* Copyright 2018-present Facebook, Inc.
*
* Licensed 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 <folly/stats/detail/DigestBuilder-defs.h>
#include <chrono>
#include <condition_variable>
#include <thread>
#include <folly/Benchmark.h>
#include <folly/Range.h>
#include <folly/portability/GFlags.h>
DEFINE_int32(digest_merge_time_ns, 13000, "Time to merge into the digest");
using namespace folly;
using namespace folly::detail;
class FreeDigest {
public:
explicit FreeDigest(size_t) {}
FreeDigest merge(Range<const double*>) const {
auto start = std::chrono::steady_clock::now();
auto finish = start + std::chrono::nanoseconds{FLAGS_digest_merge_time_ns};
while (std::chrono::steady_clock::now() < finish) {
}
return FreeDigest(100);
}
};
unsigned int append(unsigned int iters, size_t bufSize, size_t nThreads) {
iters = 1000000;
auto buffer = std::make_shared<DigestBuilder<FreeDigest>>(bufSize, 100);
std::atomic<size_t> numDone{0};
std::mutex m;
size_t numWaiters = 0;
std::condition_variable cv;
std::vector<std::thread> threads;
threads.reserve(nThreads);
BENCHMARK_SUSPEND {
for (size_t i = 0; i < nThreads; ++i) {
threads.emplace_back([&]() {
{
std::unique_lock<std::mutex> g(m);
++numWaiters;
cv.wait(g);
}
for (size_t iter = 0; iter < iters; ++iter) {
buffer->append(iter);
}
++numDone;
});
}
while (true) {
{
std::unique_lock<std::mutex> g(m);
if (numWaiters < nThreads) {
continue;
}
}
cv.notify_all();
break;
}
}
while (numDone < nThreads) {
}
BENCHMARK_SUSPEND {
for (auto& thread : threads) {
thread.join();
}
}
return iters;
}
BENCHMARK_NAMED_PARAM_MULTI(append, 1000x1, 1000, 1)
BENCHMARK_RELATIVE_NAMED_PARAM_MULTI(append, 1000x2, 1000, 2)
BENCHMARK_RELATIVE_NAMED_PARAM_MULTI(append, 1000x4, 1000, 4)
BENCHMARK_RELATIVE_NAMED_PARAM_MULTI(append, 1000x8, 1000, 8)
BENCHMARK_RELATIVE_NAMED_PARAM_MULTI(append, 1000x16, 1000, 16)
BENCHMARK_RELATIVE_NAMED_PARAM_MULTI(append, 1000x32, 1000, 32)
BENCHMARK_DRAW_LINE();
BENCHMARK_NAMED_PARAM_MULTI(append, 10000x1, 10000, 1)
BENCHMARK_RELATIVE_NAMED_PARAM_MULTI(append, 10000x2, 10000, 2)
BENCHMARK_RELATIVE_NAMED_PARAM_MULTI(append, 10000x4, 10000, 4)
BENCHMARK_RELATIVE_NAMED_PARAM_MULTI(append, 10000x8, 10000, 8)
BENCHMARK_RELATIVE_NAMED_PARAM_MULTI(append, 10000x16, 10000, 16)
BENCHMARK_RELATIVE_NAMED_PARAM_MULTI(append, 10000x32, 10000, 32)
/*
* ./digest_buffer_benchmark
* ============================================================================
* folly/stats/test/DigestBuilderBenchmark.cpp relative time/iter iters/s
* ============================================================================
* append(1000x1) 51.79ns 19.31M
* append(1000x2) 97.04% 53.37ns 18.74M
* append(1000x4) 95.68% 54.12ns 18.48M
* append(1000x8) 91.95% 56.32ns 17.76M
* append(1000x16) 62.12% 83.36ns 12.00M
* append(1000x32) 38.12% 135.85ns 7.36M
* ----------------------------------------------------------------------------
* append(10000x1) 46.34ns 21.58M
* append(10000x2) 97.91% 47.33ns 21.13M
* append(10000x4) 95.27% 48.64ns 20.56M
* append(10000x8) 91.39% 50.70ns 19.72M
* append(10000x16) 55.26% 83.85ns 11.93M
* append(10000x32) 35.57% 130.25ns 7.68M
* ============================================================================
*/
int main(int argc, char* argv[]) {
gflags::ParseCommandLineFlags(&argc, &argv, true);
folly::runBenchmarks();
return 0;
}
/*
* Copyright 2018-present Facebook, Inc.
*
* Licensed 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 <folly/stats/detail/DigestBuilder-defs.h>
#include <chrono>
#include <random>
#include <thread>
#include <folly/Range.h>
#include <folly/portability/GTest.h>
using namespace folly;
using namespace folly::detail;
class SimpleDigest {
public:
explicit SimpleDigest(size_t sz) : sz_(sz) {}
SimpleDigest merge(Range<const double*> r) const {
EXPECT_EQ(1000, r.size());
for (size_t i = 0; i < 1000; ++i) {
EXPECT_GE(1000, r[i]);
}
return *this;
}
static SimpleDigest merge(Range<const SimpleDigest*> r) {
EXPECT_EQ(1, r.size());
return *r.begin();
}
int64_t getSize() const {
return sz_;
}
private:
int64_t sz_;
};
TEST(DigestBuilder, Basic) {
DigestBuilder<SimpleDigest> builder(1000, 100);
std::vector<std::thread> threads;
for (int i = 0; i < 10; ++i) {
threads.push_back(std::thread([i, &builder]() {
for (int j = 0; j < 100; ++j) {
builder.append(i * 100 + j);
}
}));
}
for (auto& thread : threads) {
thread.join();
}
EXPECT_EQ(100, builder.buildSyncFree().getSize());
}
......@@ -4,12 +4,16 @@ CPPFLAGS = -I$(top_srcdir)/test/gtest/googletest/include
ldadd = $(top_builddir)/test/libfollytestmain.la
TESTS = \
digest_builder_test \
histogram_test \
sliding_window_test \
tdigest_test
check_PROGRAMS = $(TESTS)
digest_builder_test_SOURCES = DigestBuilderTest.cpp
digest_builder_test_LDADD = $(ldadd)
histogram_test_SOURCES = HistogramTest.cpp
histogram_test_LDADD = $(ldadd)
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment