From ca79b8eaab4e1aa1363106971bc9934f20034c69 Mon Sep 17 00:00:00 2001 From: Lukas Joswiak Date: Thu, 20 May 2021 11:16:31 -0700 Subject: [PATCH] Various ALP fixes --- fdbclient/ActorLineageProfiler.cpp | 27 ++++++++++------- fdbclient/ActorLineageProfiler.h | 1 + fdbclient/CMakeLists.txt | 2 ++ fdbclient/FluentDSampleIngestor.cpp | 18 +++++++++--- fdbclient/SpecialKeySpace.actor.cpp | 45 +++++++++++++++++++---------- fdbclient/StackLineage.cpp | 29 +++++++++++++++++++ fdbclient/StackLineage.h | 41 ++++++++++++++++++++++++++ fdbserver/SigStack.cpp | 1 + flow/Net2.actor.cpp | 2 ++ flow/actorcompiler/ActorCompiler.cs | 6 ++-- flow/flow.cpp | 4 --- flow/flow.h | 14 ++++----- 12 files changed, 146 insertions(+), 44 deletions(-) create mode 100644 fdbclient/StackLineage.cpp create mode 100644 fdbclient/StackLineage.h diff --git a/fdbclient/ActorLineageProfiler.cpp b/fdbclient/ActorLineageProfiler.cpp index 9be7a60704..fe23e41c17 100644 --- a/fdbclient/ActorLineageProfiler.cpp +++ b/fdbclient/ActorLineageProfiler.cpp @@ -63,6 +63,8 @@ class Packer : public msgpack::packer { std::string, std::string_view, std::vector, + std::vector, + std::vector, std::map, std::map, std::vector>>::populate(visitorMap); @@ -196,6 +198,10 @@ std::shared_ptr 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> 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 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 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 const& config) { throw err; } setBackend(std::make_shared( - useTCP ? FluentDIngestor::Protocol::TCP : FluentDIngestor::Protocol::TCP, address)); + useTCP ? FluentDIngestor::Protocol::TCP : FluentDIngestor::Protocol::UDP, address)); } } diff --git a/fdbclient/ActorLineageProfiler.h b/fdbclient/ActorLineageProfiler.h index 30a3eec3e7..319b045b29 100644 --- a/fdbclient/ActorLineageProfiler.h +++ b/fdbclient/ActorLineageProfiler.h @@ -108,6 +108,7 @@ private: // construction void setBackend(std::shared_ptr ingestor) { this->ingestor = ingestor; } public: + void ingest(std::shared_ptr sample) { ingestor->ingest(sample); } void reset(std::map const& config); std::map getConfig() const; }; diff --git a/fdbclient/CMakeLists.txt b/fdbclient/CMakeLists.txt index 1e669f2e00..7138962115 100644 --- a/fdbclient/CMakeLists.txt +++ b/fdbclient/CMakeLists.txt @@ -76,6 +76,8 @@ set(FDBCLIENT_SRCS StorageServerInterface.h Subspace.cpp Subspace.h + StackLineage.h + StackLineage.cpp SystemData.cpp SystemData.h TagThrottle.actor.cpp diff --git a/fdbclient/FluentDSampleIngestor.cpp b/fdbclient/FluentDSampleIngestor.cpp index 08d5bfe55f..f4c9909253 100644 --- a/fdbclient/FluentDSampleIngestor.cpp +++ b/fdbclient/FluentDSampleIngestor.cpp @@ -52,6 +52,7 @@ class SampleSender : public std::enable_shared_from_this sample_; // to keep from being deallocated struct Buf { const char* data; @@ -72,20 +73,25 @@ class SampleSender : public std::enable_shared_from_this 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 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 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 socket; + std::shared_ptr socket; boost::asio::steady_timer retryTimer; FluentDIngestorImpl(Protocol protocol, NetworkAddress const& endpoint) : protocol(protocol), endpoint(endpoint), io_context(ActorLineageProfiler::instance().context()), diff --git a/fdbclient/SpecialKeySpace.actor.cpp b/fdbclient/SpecialKeySpace.actor.cpp index 8d0c7e5361..6802ed3196 100644 --- a/fdbclient/SpecialKeySpace.actor.cpp +++ b/fdbclient/SpecialKeySpace.actor.cpp @@ -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 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 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 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 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 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))); didWrite = true; } diff --git a/fdbclient/StackLineage.cpp b/fdbclient/StackLineage.cpp new file mode 100644 index 0000000000..16a19a8a19 --- /dev/null +++ b/fdbclient/StackLineage.cpp @@ -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 getActorStackTrace() { + return currentLineage->stack(&StackLineage::actorName); +} + +namespace { +StackLineageCollector stackLineageCollector; +} diff --git a/fdbclient/StackLineage.h b/fdbclient/StackLineage.h new file mode 100644 index 0000000000..a94d20a836 --- /dev/null +++ b/fdbclient/StackLineage.h @@ -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 + +#include "flow/flow.h" +#include "fdbclient/ActorLineageProfiler.h" + +extern std::vector getActorStackTrace(); + +struct StackLineageCollector : IALPCollector { + StackLineageCollector() : IALPCollector() {} + std::optional collect(ActorLineage* lineage) override { + auto vec = lineage->stack(&StackLineage::actorName); + + std::vector res; + for (const auto& str : vec) { + res.push_back(std::string_view(reinterpret_cast(str.begin()), str.size())); + } + return res; + } +}; diff --git a/fdbserver/SigStack.cpp b/fdbserver/SigStack.cpp index 0c35326766..6ad74ddb63 100644 --- a/fdbserver/SigStack.cpp +++ b/fdbserver/SigStack.cpp @@ -1,4 +1,5 @@ #include "flow/flow.h" +#include "fdbclient/StackLineage.h" #include #include #include diff --git a/flow/Net2.actor.cpp b/flow/Net2.actor.cpp index 7317476beb..4af22e5ff5 100644 --- a/flow/Net2.actor.cpp +++ b/flow/Net2.actor.cpp @@ -1435,6 +1435,8 @@ void Net2::run() { checkForSlowTask(tscBegin, timestampCounter(), taskEnd - taskBegin, TaskPriority::RunCycleFunction); } + currentLineage = Reference(); + currentLineageThreadSafe.replace(Reference()); double sleepTime = 0; bool b = ready.empty(); if (b) { diff --git a/flow/actorcompiler/ActorCompiler.cs b/flow/actorcompiler/ActorCompiler.cs index 28771f4503..e167cc5f61 100644 --- a/flow/actorcompiler/ActorCompiler.cs +++ b/flow/actorcompiler/ActorCompiler.cs @@ -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); diff --git a/flow/flow.cpp b/flow/flow.cpp index ec65640fe2..6d59e27e4f 100644 --- a/flow/flow.cpp +++ b/flow/flow.cpp @@ -43,10 +43,6 @@ using namespace std::literals; const std::string_view StackLineage::name = "StackLineage"sv; -std::vector 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) { diff --git a/flow/flow.h b/flow/flow.h index 4127bf6c8b..991371d066 100644 --- a/flow/flow.h +++ b/flow/flow.h @@ -448,7 +448,7 @@ struct LineageProperties : LineagePropertiesBase { } }; -struct ActorLineage : ThreadSafeReferenceCounted, public FastAllocated { +struct ActorLineage : ThreadSafeReferenceCounted { friend class LocalLineage; private: @@ -516,6 +516,11 @@ public: extern thread_local Reference currentLineage; extern WriteOnlyVariable currentLineageThreadSafe; +struct StackLineage : LineageProperties { + 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 { - static const std::string_view name; - StringRef actorName; -}; - -extern std::vector getActorStackTrace(); - // SAV is short for Single Assignment Variable: It can be assigned for only once! template struct SAV : private Callback, FastAllocated> {