Refactor samples to include wait state

fdbclient/ActorLineageProfiler.cpp

@@ -150,12 +150,9 @@ public:
 		}
 	}
 
-	std::shared_ptr<Sample> done(double time) {
-		auto res = std::make_shared<Sample>();
-		res->time = time;
-		res->size = sbuffer.size();
-		res->data = sbuffer.release();
-		return res;
+	std::pair<char*, unsigned> getbuf() {
+		unsigned size = sbuffer.size();
+		return std::make_pair(sbuffer.release(), size);
 	}
 };
 
@@ -175,11 +172,11 @@ std::map<std::string_view, std::any> SampleCollectorT::collect(ActorLineage* lin
 }
 
 std::shared_ptr<Sample> SampleCollectorT::collect() {
-	Packer packer;
-	std::map<std::string_view, std::any> res;
+	auto sample = std::make_shared<Sample>();
 	double time = g_network->now();
-	res["time"sv] = time;
+	sample->time = time;
 	for (auto& p : getSamples) {
+		Packer packer;
 		std::vector<std::map<std::string_view, std::any>> samples;
 		auto sampleVec = p.second();
 		for (auto& val : sampleVec) {
@@ -189,11 +186,11 @@ std::shared_ptr<Sample> SampleCollectorT::collect() {
 			}
 		}
 		if (!samples.empty()) {
-			res[to_string(p.first)] = samples;
+			packer.pack(samples);
+			sample->data[p.first] = packer.getbuf();
 		}
 	}
-	packer.pack(res);
-	return packer.done(time);
+	return sample;
 }
 
 void SampleCollection_t::refresh() {

fdbclient/ActorLineageProfiler.h

@@ -47,9 +47,12 @@ struct IALPCollector : IALPCollectorBase {
 
 struct Sample : std::enable_shared_from_this<Sample> {
 	double time = 0.0;
-	unsigned size = 0u;
-	char* data = nullptr;
-	~Sample() { ::free(data); }
+	std::unordered_map<WaitState, std::pair<char*, unsigned>> data;
+	~Sample() {
+		std::for_each(data.begin(), data.end(), [](std::pair<WaitState, std::pair<char*, unsigned>> entry) {
+			::free(entry.second.first);
+		});
+	}
 };
 
 class SampleCollectorT {

fdbclient/ProcessInterface.h

@@ -62,14 +62,13 @@ struct EchoRequest {
 struct SerializedSample {
 	constexpr static FileIdentifier file_identifier = 15785634;
 
-	WaitState waitState;
 	double time;
 	int seq;
-	std::string data;
+	std::unordered_map<WaitState, std::string> data;
 
 	template <class Ar>
 	void serialize(Ar& ar) {
-		serializer(ar, waitState, time, seq, data);
+		serializer(ar, time, seq, data);
 	}
 };
 

fdbclient/SpecialKeySpace.actor.cpp

@@ -2095,15 +2095,37 @@ ACTOR static Future<Standalone<RangeResultRef>> actorLineageGetRangeActor(ReadYo
 	actorLineageRequest.seqEnd = seqEnd;
 	ActorLineageReply reply = wait(process.actorLineage.getReply(actorLineageRequest));
 
+	time_t dt = 0;
+	int seq = -1;
 	for (const auto& sample : reply.samples) {
-		msgpack::object_handle oh = msgpack::unpack(sample.data.data(), sample.data.size());
-		msgpack::object deserialized = oh.get();
+		for (const auto& [waitState, data] : sample.data) {
+			time_t datetime = (time_t)sample.time;
+			seq = dt == datetime ? seq + 1 : 0;
+			dt = datetime;
 
-		std::ostringstream stream;
-		stream << deserialized;
-		// TODO: Fix return value for ranges
-		Key returnKey = prefix.withSuffix(host.toString() + "/" + std::to_string(sample.seq));
-		result.push_back_deep(result.arena(), KeyValueRef(returnKey, stream.str()));
+			if (seq < seqStart) {
+				continue;
+			} else if (seq >= seqEnd) {
+				break;
+			}
+
+			char buf[200];
+			struct tm* tm;
+			tm = localtime(&datetime);
+			size_t size = strftime(buf, 200, "%FT%T%z", tm);
+			std::string date(buf, size);
+
+			msgpack::object_handle oh = msgpack::unpack(data.data(), data.size());
+			msgpack::object deserialized = oh.get();
+
+			std::ostringstream stream;
+			stream << deserialized;
+
+			// TODO: Fix return value for time range
+			Key returnKey = prefix.withSuffix(host.toString() + "/" + std::string(to_string(waitState)) + "/" + date +
+			                                  "/" + std::to_string(seq));
+			result.push_back_deep(result.arena(), KeyValueRef(returnKey, stream.str()));
+		}
 	}
 
 	return result;

fdbserver/worker.actor.cpp

@@ -2060,13 +2060,18 @@ ACTOR Future<Void> serveProcess() {
 				int maxSeq = std::min(req.seqEnd, static_cast<int>(samples.size()));
 
 				std::vector<SerializedSample> serializedSamples;
-				for (int i = req.seqStart; i < maxSeq; ++i) {
-					auto samplePtr = samples.at(i);
-					auto serialized = SerializedSample{ .waitState = WaitState::Network, // TODO: Currently unused
-						                                .time = samplePtr->time,
-						                                .seq = i,
-						                                .data = std::string(samplePtr->data, samplePtr->size) };
+				for (const auto& samplePtr : samples) {
+					int seq = 0;
+					auto serialized = SerializedSample{ .time = samplePtr->time, .seq = seq };
+					for (const auto& [waitState, pair] : samplePtr->data) {
+						serialized.data[waitState] = std::string(pair.first, pair.second);
+					}
 					serializedSamples.push_back(std::move(serialized));
+
+					// TODO: Don't need to transmit seq over the network anymore
+					if (++seq >= maxSeq) {
+						continue;
+					};
 				}
 				ActorLineageReply reply{ serializedSamples };
 				req.reply.send(reply);