wbtiger
/
foundationdb
mirror of https://github.com/apple/foundationdb.git
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Various ALP fixes

This commit is contained in:
Lukas Joswiak 2021-05-20 11:16:31 -07:00
parent c2f39bc8ef
commit ca79b8eaab
12 changed files with 146 additions and 44 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

27
fdbclient/ActorLineageProfiler.cpp
View File

@ -63,6 +63,8 @@ class Packer : public msgpack::packer<msgpack::sbuffer> {
std::string,
std::string_view,
std::vector<std::any>,
std::vector<std::string>,
std::vector<std::string_view>,
std::map<std::string, std::any>,
std::map<std::string_view, std::any>,
std::vector<std::map<std::string_view, std::any>>>::populate(visitorMap);
@ -196,6 +198,10 @@ std::shared_ptr<Sample> SampleCollectorT::collect() {
void SampleCollection_t::refresh() {
auto sample = _collector->collect();
// TODO: Should only call ingest when deleting from memory
if (sample.get() != 0) {
config->ingest(sample);
}
auto min = std::min(sample->time - windowSize, sample->time);
{
Lock _{ mutex };
@ -212,7 +218,6 @@ void SampleCollection_t::refresh() {
oldest = data.front()->time;
}
}
//config->ingest(sample);
}
std::vector<std::shared_ptr<Sample>> SampleCollection_t::get(double from /*= 0.0*/,
@ -268,11 +273,11 @@ struct ProfilerImpl {
};
ActorLineageProfilerT::ActorLineageProfilerT() : impl(new ProfilerImpl()) {
collection->collector()->addGetter(WaitState::Network,
std::bind(&ActorLineageSet::copy, std::ref(g_network->getActorLineageSet())));
collection->collector()->addGetter(
WaitState::Disk,
std::bind(&ActorLineageSet::copy, std::ref(IAsyncFileSystem::filesystem()->getActorLineageSet())));
// collection->collector()->addGetter(WaitState::Network,
// std::bind(&ActorLineageSet::copy, std::ref(g_network->getActorLineageSet())));
// collection->collector()->addGetter(
// WaitState::Disk,
// std::bind(&ActorLineageSet::copy, std::ref(IAsyncFileSystem::filesystem()->getActorLineageSet())));
collection->collector()->addGetter(WaitState::Running, []() {
auto res = currentLineageThreadSafe.get();
if (res.isValid()) {
@ -316,7 +321,7 @@ void ProfilerConfigT::reset(std::map<std::string, std::string> const& config) {
err.description = format("Unexpected option %s", kv.first.c_str());
throw err;
}
if (kv.first == "collector") {
if (kv.first == "ingestor") {
std::string val = kv.second;
std::for_each(val.begin(), val.end(), [](auto c) { return std::tolower(c); });
if (val == "none") {
@ -324,12 +329,12 @@ void ProfilerConfigT::reset(std::map<std::string, std::string> const& config) {
} else if (val == "fluentd") {
useFluentD = true;
} else {
err.description = format("Unsupported collector: %s", val.c_str());
err.description = format("Unsupported ingestor: %s", val.c_str());
throw err;
}
} else if (kv.first == "collector_endpoint") {
} else if (kv.first == "ingestor_endpoint") {
endpoint = kv.second;
} else if (kv.first == "collector_protocol") {
} else if (kv.first == "ingestor_protocol") {
auto val = kv.second;
std::for_each(val.begin(), val.end(), [](auto c) { return std::tolower(c); });
if (val == "tcp") {
@ -358,7 +363,7 @@ void ProfilerConfigT::reset(std::map<std::string, std::string> const& config) {
throw err;
}
setBackend(std::make_shared<FluentDIngestor>(
useTCP ? FluentDIngestor::Protocol::TCP : FluentDIngestor::Protocol::TCP, address));
useTCP ? FluentDIngestor::Protocol::TCP : FluentDIngestor::Protocol::UDP, address));
}
}

1
fdbclient/ActorLineageProfiler.h
View File

@ -108,6 +108,7 @@ private: // construction
void setBackend(std::shared_ptr<SampleIngestor> ingestor) { this->ingestor = ingestor; }
public:
void ingest(std::shared_ptr<Sample> sample) { ingestor->ingest(sample); }
void reset(std::map<std::string, std::string> const& config);
std::map<std::string, std::string> getConfig() const;
};

2
fdbclient/CMakeLists.txt
View File

@ -76,6 +76,8 @@ set(FDBCLIENT_SRCS
StorageServerInterface.h
Subspace.cpp
Subspace.h
StackLineage.h
StackLineage.cpp
SystemData.cpp
SystemData.h
TagThrottle.actor.cpp

18
fdbclient/FluentDSampleIngestor.cpp
View File

@ -52,6 +52,7 @@ class SampleSender : public std::enable_shared_from_this<SampleSender<Protocol,
Socket& socket;
Callback callback;
Iter iter, end;
std::shared_ptr<Sample> sample_; // to keep from being deallocated
struct Buf {
const char* data;
@ -72,20 +73,25 @@ class SampleSender : public std::enable_shared_from_this<SampleSender<Protocol,
}
void send(boost::asio::ip::tcp::socket& socket, std::shared_ptr<Buf> const& buf) {
// auto self = this->shared_from_this();
boost::asio::async_write(
socket,
boost::asio::const_buffer(buf->data, buf->size),
[buf, self = this->shared_from_this()](auto const& ec, size_t) { self->sendCompletionHandler(ec); });
[buf, this](auto const& ec, size_t) {
this->sendCompletionHandler(ec);
});
}
void send(boost::asio::ip::udp::socket& socket, std::shared_ptr<Buf> const& buf) {
socket.async_send(
boost::asio::const_buffer(buf->data, buf->size),
[buf, self = this->shared_from_this()](auto const& ec, size_t) { self->sendCompletionHandler(ec); });
// [buf, self = this->shared_from_this()](auto const& ec, size_t) { self->sendCompletionHandler(ec); });
[buf, this](auto const& ec, size_t) { this->sendCompletionHandler(ec); });
}
void sendNext() {
if (iter == end) {
callback(boost::system::error_code());
return;
}
// 1. calculate size of buffer
unsigned size = 1; // 1 for fixmap identifier byte
@ -118,7 +124,11 @@ class SampleSender : public std::enable_shared_from_this<SampleSender<Protocol,
public:
SampleSender(Socket& socket, Callback const& callback, std::shared_ptr<Sample> const& sample)
: socket(socket), callback(callback), iter(sample->data.begin()), end(sample->data.end()) {
: socket(socket),
callback(callback),
sample_(sample),
iter(sample->data.begin()),
end(sample->data.end()) {
sendNext();
}
};
@ -197,7 +207,7 @@ struct FluentDIngestorImpl {
Protocol protocol;
NetworkAddress endpoint;
boost::asio::io_context& io_context;
std::unique_ptr<FluentDSocket> socket;
std::shared_ptr<FluentDSocket> socket;
boost::asio::steady_timer retryTimer;
FluentDIngestorImpl(Protocol protocol, NetworkAddress const& endpoint)
: protocol(protocol), endpoint(endpoint), io_context(ActorLineageProfiler::instance().context()),

45
fdbclient/SpecialKeySpace.actor.cpp
View File

@ -1941,11 +1941,11 @@ void parse(StringRef& val, double& d) {
}
void parse(StringRef& val, WaitState& w) {
if (val == LiteralStringRef("disk")) {
if (val == LiteralStringRef("disk") || val == LiteralStringRef("Disk")) {
w = WaitState::Disk;
} else if (val == LiteralStringRef("network")) {
} else if (val == LiteralStringRef("network") || val == LiteralStringRef("Network")) {
w = WaitState::Network;
} else if (val == LiteralStringRef("running")) {
} else if (val == LiteralStringRef("running") || val == LiteralStringRef("Running")) {
w = WaitState::Running;
} else {
throw std::range_error("failed to parse run state");
@ -2088,20 +2088,20 @@ ACTOR static Future<RangeResult> actorLineageGetRangeActor(ReadYourWritesTransac
time_t dt = 0;
int seq = -1;
for (const auto& sample : reply.samples) {
for (const auto& [waitState, data] : sample.data) {
time_t datetime = (time_t)sample.time;
seq = dt == datetime ? seq + 1 : 0;
dt = datetime;
time_t datetime = (time_t)sample.time;
char buf[50];
struct tm* tm;
tm = localtime(&datetime);
size_t size = strftime(buf, 50, "%FT%T%z", tm);
std::string date(buf, size);
seq = dt == datetime ? seq + 1 : 0;
dt = datetime;
for (const auto& [waitState, data] : sample.data) {
if (seq < seqStart) { continue; }
else if (seq >= seqEnd) { break; }
char buf[50];
struct tm* tm;
tm = localtime(&datetime);
size_t size = strftime(buf, 50, "%FT%T%z", tm);
std::string date(buf, size);
std::ostringstream streamKey;
if (SpecialKeySpace::getActorLineageApiCommandRange("state").contains(kr)) {
streamKey << SpecialKeySpace::getActorLineageApiCommandPrefix("state").toString() << host.toString()
@ -2109,7 +2109,6 @@ ACTOR static Future<RangeResult> actorLineageGetRangeActor(ReadYourWritesTransac
} else if (SpecialKeySpace::getActorLineageApiCommandRange("time").contains(kr)) {
streamKey << SpecialKeySpace::getActorLineageApiCommandPrefix("time").toString() << host.toString()
<< "/" << date << "/" << to_string(waitState);
;
} else {
ASSERT(false);
}
@ -2123,6 +2122,21 @@ ACTOR static Future<RangeResult> actorLineageGetRangeActor(ReadYourWritesTransac
result.push_back_deep(result.arena(), KeyValueRef(streamKey.str(), stream.str()));
}
if (sample.data.size() == 0) {
std::ostringstream streamKey;
if (SpecialKeySpace::getActorLineageApiCommandRange("state").contains(kr)) {
streamKey << SpecialKeySpace::getActorLineageApiCommandPrefix("state").toString() << host.toString()
<< "/Running/" << date;
} else if (SpecialKeySpace::getActorLineageApiCommandRange("time").contains(kr)) {
streamKey << SpecialKeySpace::getActorLineageApiCommandPrefix("time").toString() << host.toString()
<< "/" << date << "/Running";
} else {
ASSERT(false);
}
streamKey << "/" << seq;
result.push_back_deep(result.arena(), KeyValueRef(streamKey.str(), "{}"_sr));
}
}
return result;
@ -2150,7 +2164,7 @@ Future<RangeResult> ActorProfilerConf::getRange(ReadYourWritesTransaction* ryw,
break;
} else if (p.first > begin) {
KeyValueRef kv;
kv.key = StringRef(res.arena(), p.first);
kv.key = StringRef(res.arena(), p.first).withPrefix(kr.begin, res.arena());
kv.value = StringRef(res.arena(), p.second);
res.push_back(res.arena(), kv);
}
@ -2160,6 +2174,7 @@ Future<RangeResult> ActorProfilerConf::getRange(ReadYourWritesTransaction* ryw,
void ActorProfilerConf::set(ReadYourWritesTransaction* ryw, const KeyRef& key, const ValueRef& value) {
config[key.removePrefix(range.begin).toString()] = value.toString();
ryw->getSpecialKeySpaceWriteMap().insert(key, std::make_pair(true, Optional<Value>(value)));
didWrite = true;
}

29
fdbclient/StackLineage.cpp Normal file
View File

@ -0,0 +1,29 @@
/*
* StackLineage.cpp
*
* This source file is part of the FoundationDB open source project
*
* Copyright 2013-2021 Apple Inc. and the FoundationDB project authors
*
* 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 "fdbclient/StackLineage.h"
std::vector<StringRef> getActorStackTrace() {
return currentLineage->stack(&StackLineage::actorName);
}
namespace {
StackLineageCollector stackLineageCollector;
}

41
fdbclient/StackLineage.h Normal file
View File

@ -0,0 +1,41 @@
/*
* StackLineage.h
*
* This source file is part of the FoundationDB open source project
*
* Copyright 2013-2021 Apple Inc. and the FoundationDB project authors
*
* 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 <string_view>
#include "flow/flow.h"
#include "fdbclient/ActorLineageProfiler.h"
extern std::vector<StringRef> getActorStackTrace();
struct StackLineageCollector : IALPCollector<StackLineage> {
StackLineageCollector() : IALPCollector() {}
std::optional<std::any> collect(ActorLineage* lineage) override {
auto vec = lineage->stack(&StackLineage::actorName);
std::vector<std::string_view> res;
for (const auto& str : vec) {
res.push_back(std::string_view(reinterpret_cast<const char*>(str.begin()), str.size()));
}
return res;
}
};

1
fdbserver/SigStack.cpp
View File

@ -1,4 +1,5 @@
#include "flow/flow.h"
#include "fdbclient/StackLineage.h"
#include <csignal>
#include <iostream>
#include <string_view>

2
flow/Net2.actor.cpp
View File

@ -1435,6 +1435,8 @@ void Net2::run() {
checkForSlowTask(tscBegin, timestampCounter(), taskEnd - taskBegin, TaskPriority::RunCycleFunction);
}
currentLineage = Reference<ActorLineage>();
currentLineageThreadSafe.replace(Reference<ActorLineage>());
double sleepTime = 0;
bool b = ready.empty();
if (b) {

6
flow/actorcompiler/ActorCompiler.cs
View File

@ -452,7 +452,8 @@ namespace actorcompiler
fullClassName,
string.Join(", ", actor.parameters.Select(p => p.name).ToArray()));
writer.WriteLine("\trestore_lineage _;");
if (actor.IsCancellable())
writer.WriteLine("\trestore_lineage _;");
if (actor.returnType != null)
writer.WriteLine("\treturn Future<{1}>({0});", newActor, actor.returnType);
else
@ -1287,7 +1288,8 @@ namespace actorcompiler
constructor.WriteLine("{");
constructor.Indent(+1);
ProbeEnter(constructor, actor.name);
constructor.WriteLine("currentLineage->modify(&StackLineage::actorName) = LiteralStringRef(\"{0}\");", actor.name);
if (actor.IsCancellable())
constructor.WriteLine("currentLineage->modify(&StackLineage::actorName) = LiteralStringRef(\"{0}\");", actor.name);
constructor.WriteLine("this->{0};", body.call());
ProbeExit(constructor, actor.name);
WriteFunction(writer, constructor, constructor.BodyText);

4
flow/flow.cpp
View File

@ -43,10 +43,6 @@ using namespace std::literals;
const std::string_view StackLineage::name = "StackLineage"sv;
std::vector<StringRef> getActorStackTrace() {
return currentLineage->stack(&StackLineage::actorName);
}
#if (defined(__linux__) || defined(__FreeBSD__)) && defined(__AVX__) && !defined(MEMORY_SANITIZER)
// For benchmarking; need a version of rte_memcpy that doesn't live in the same compilation unit as the test.
void* rte_memcpy_noinline(void* __restrict __dest, const void* __restrict __src, size_t __n) {

14
flow/flow.h
View File

@ -448,7 +448,7 @@ struct LineageProperties : LineagePropertiesBase {
}
};
struct ActorLineage : ThreadSafeReferenceCounted<ActorLineage>, public FastAllocated<ActorLineage> {
struct ActorLineage : ThreadSafeReferenceCounted<ActorLineage> {
friend class LocalLineage;
private:
@ -516,6 +516,11 @@ public:
extern thread_local Reference<ActorLineage> currentLineage;
extern WriteOnlyVariable<ActorLineage, unsigned> currentLineageThreadSafe;
struct StackLineage : LineageProperties<StackLineage> {
static const std::string_view name;
StringRef actorName;
};
// This class can be used in order to modify all lineage properties
// of actors created within a (non-actor) scope
struct LocalLineage {
@ -541,13 +546,6 @@ struct restore_lineage {
}
};
struct StackLineage : LineageProperties<StackLineage> {
static const std::string_view name;
StringRef actorName;
};
extern std::vector<StringRef> getActorStackTrace();
// SAV is short for Single Assignment Variable: It can be assigned for only once!
template <class T>
struct SAV : private Callback<T>, FastAllocated<SAV<T>> {