Commit 7469e0b5 authored by Misha Shneerson's avatar Misha Shneerson Committed by Facebook GitHub Bot

Lift RequestContext root-id into task stats

Summary: We want to use observer API to correlate tasks across same Thrift request. `RequestContext::getRootId` is the thing for it.

Reviewed By: yfeldblum

Differential Revision: D20600485

fbshipit-source-id: 1a79bde0f6c5b3692b4fb3805c98f377b1b317c9
parent 0938cf41
...@@ -78,7 +78,6 @@ class EDFThreadPoolExecutor::Task { ...@@ -78,7 +78,6 @@ class EDFThreadPoolExecutor::Task {
std::atomic<int> iter_{0}; std::atomic<int> iter_{0};
int total_; int total_;
uint64_t deadline_; uint64_t deadline_;
TaskStats stats_;
std::shared_ptr<RequestContext> context_ = RequestContext::saveContext(); std::shared_ptr<RequestContext> context_ = RequestContext::saveContext();
std::chrono::steady_clock::time_point enqueueTime_ = std::chrono::steady_clock::time_point enqueueTime_ =
std::chrono::steady_clock::now(); std::chrono::steady_clock::now();
...@@ -369,7 +368,13 @@ void EDFThreadPoolExecutor::threadRun(ThreadPtr thread) { ...@@ -369,7 +368,13 @@ void EDFThreadPoolExecutor::threadRun(ThreadPtr thread) {
thread->idle = false; thread->idle = false;
auto startTime = std::chrono::steady_clock::now(); auto startTime = std::chrono::steady_clock::now();
task->stats_.waitTime = startTime - task->enqueueTime_; TaskStats stats;
stats.enqueueTime = task->enqueueTime_;
if (task->context_) {
stats.requestId = task->context_->getRootId();
}
stats.waitTime = startTime - stats.enqueueTime;
try { try {
task->run(iter); task->run(iter);
} catch (const std::exception& e) { } catch (const std::exception& e) {
...@@ -379,7 +384,7 @@ void EDFThreadPoolExecutor::threadRun(ThreadPtr thread) { ...@@ -379,7 +384,7 @@ void EDFThreadPoolExecutor::threadRun(ThreadPtr thread) {
LOG(ERROR) LOG(ERROR)
<< "EDFThreadPoolExecutor: func threw unhandled non-exception object"; << "EDFThreadPoolExecutor: func threw unhandled non-exception object";
} }
task->stats_.runTime = std::chrono::steady_clock::now() - startTime; stats.runTime = std::chrono::steady_clock::now() - startTime;
thread->idle = true; thread->idle = true;
thread->lastActiveTime = std::chrono::steady_clock::now(); thread->lastActiveTime = std::chrono::steady_clock::now();
thread->taskStatsCallbacks->callbackList.withRLock([&](auto& callbacks) { thread->taskStatsCallbacks->callbackList.withRLock([&](auto& callbacks) {
...@@ -389,7 +394,7 @@ void EDFThreadPoolExecutor::threadRun(ThreadPtr thread) { ...@@ -389,7 +394,7 @@ void EDFThreadPoolExecutor::threadRun(ThreadPtr thread) {
}; };
try { try {
for (auto& callback : callbacks) { for (auto& callback : callbacks) {
callback(task->stats_); callback(stats);
} }
} catch (const std::exception& e) { } catch (const std::exception& e) {
LOG(ERROR) << "EDFThreadPoolExecutor: task stats callback threw " LOG(ERROR) << "EDFThreadPoolExecutor: task stats callback threw "
......
...@@ -76,10 +76,16 @@ ThreadPoolExecutor::Task::Task( ...@@ -76,10 +76,16 @@ ThreadPoolExecutor::Task::Task(
void ThreadPoolExecutor::runTask(const ThreadPtr& thread, Task&& task) { void ThreadPoolExecutor::runTask(const ThreadPtr& thread, Task&& task) {
thread->idle = false; thread->idle = false;
auto startTime = std::chrono::steady_clock::now(); auto startTime = std::chrono::steady_clock::now();
task.stats_.waitTime = startTime - task.enqueueTime_; TaskStats stats;
stats.enqueueTime = task.enqueueTime_;
if (task.context_) {
stats.requestId = task.context_->getRootId();
}
stats.waitTime = startTime - task.enqueueTime_;
if (task.expiration_ > std::chrono::milliseconds(0) && if (task.expiration_ > std::chrono::milliseconds(0) &&
task.stats_.waitTime >= task.expiration_) { stats.waitTime >= task.expiration_) {
task.stats_.expired = true; stats.expired = true;
if (task.expireCallback_ != nullptr) { if (task.expireCallback_ != nullptr) {
task.expireCallback_(); task.expireCallback_();
} }
...@@ -94,7 +100,7 @@ void ThreadPoolExecutor::runTask(const ThreadPtr& thread, Task&& task) { ...@@ -94,7 +100,7 @@ void ThreadPoolExecutor::runTask(const ThreadPtr& thread, Task&& task) {
LOG(ERROR) << "ThreadPoolExecutor: func threw unhandled non-exception " LOG(ERROR) << "ThreadPoolExecutor: func threw unhandled non-exception "
"object"; "object";
} }
task.stats_.runTime = std::chrono::steady_clock::now() - startTime; stats.runTime = std::chrono::steady_clock::now() - startTime;
} }
thread->idle = true; thread->idle = true;
thread->lastActiveTime = std::chrono::steady_clock::now(); thread->lastActiveTime = std::chrono::steady_clock::now();
...@@ -105,7 +111,7 @@ void ThreadPoolExecutor::runTask(const ThreadPtr& thread, Task&& task) { ...@@ -105,7 +111,7 @@ void ThreadPoolExecutor::runTask(const ThreadPtr& thread, Task&& task) {
}; };
try { try {
for (auto& callback : callbacks) { for (auto& callback : callbacks) {
callback(task.stats_); callback(stats);
} }
} catch (const std::exception& e) { } catch (const std::exception& e) {
LOG(ERROR) << "ThreadPoolExecutor: task stats callback threw " LOG(ERROR) << "ThreadPoolExecutor: task stats callback threw "
......
...@@ -125,10 +125,12 @@ class ThreadPoolExecutor : public DefaultKeepAliveExecutor { ...@@ -125,10 +125,12 @@ class ThreadPoolExecutor : public DefaultKeepAliveExecutor {
std::string getName() const; std::string getName() const;
struct TaskStats { struct TaskStats {
TaskStats() : expired(false), waitTime(0), runTime(0) {} TaskStats() : expired(false), waitTime(0), runTime(0), requestId(0) {}
bool expired; bool expired;
std::chrono::nanoseconds waitTime; std::chrono::nanoseconds waitTime;
std::chrono::nanoseconds runTime; std::chrono::nanoseconds runTime;
std::chrono::steady_clock::time_point enqueueTime;
uint64_t requestId;
}; };
using TaskStatsCallback = std::function<void(TaskStats)>; using TaskStatsCallback = std::function<void(TaskStats)>;
...@@ -204,7 +206,6 @@ class ThreadPoolExecutor : public DefaultKeepAliveExecutor { ...@@ -204,7 +206,6 @@ class ThreadPoolExecutor : public DefaultKeepAliveExecutor {
std::chrono::milliseconds expiration, std::chrono::milliseconds expiration,
Func&& expireCallback); Func&& expireCallback);
Func func_; Func func_;
TaskStats stats_;
std::chrono::steady_clock::time_point enqueueTime_; std::chrono::steady_clock::time_point enqueueTime_;
std::chrono::milliseconds expiration_; std::chrono::milliseconds expiration_;
Func expireCallback_; Func expireCallback_;
......
...@@ -266,14 +266,20 @@ template <class TPE> ...@@ -266,14 +266,20 @@ template <class TPE>
static void taskStats() { static void taskStats() {
TPE tpe(1); TPE tpe(1);
std::atomic<int> c(0); std::atomic<int> c(0);
auto now = std::chrono::steady_clock::now();
tpe.subscribeToTaskStats([&](ThreadPoolExecutor::TaskStats stats) { tpe.subscribeToTaskStats([&](ThreadPoolExecutor::TaskStats stats) {
int i = c++; int i = c++;
EXPECT_LT(now, stats.enqueueTime);
EXPECT_LT(milliseconds(0), stats.runTime); EXPECT_LT(milliseconds(0), stats.runTime);
if (i == 1) { if (i == 1) {
EXPECT_LT(milliseconds(0), stats.waitTime); EXPECT_LT(milliseconds(0), stats.waitTime);
EXPECT_NE(0, stats.requestId);
} else {
EXPECT_EQ(0, stats.requestId);
} }
}); });
tpe.add(burnMs(10)); tpe.add(burnMs(10));
RequestContextScopeGuard rctx;
tpe.add(burnMs(10)); tpe.add(burnMs(10));
tpe.join(); tpe.join();
EXPECT_EQ(2, c); EXPECT_EQ(2, c);
......
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