f20eb12457
Thread pool design point:
All tasks submitted directly to the thread pool enter a FIFO queue and are
dispatched to a worker thread when one becomes free. Tasks may also be
submitted via ThreadPoolTokens. The token wait() and shutdown() functions
can then be used to block on logical groups of tasks.
A token operates in one of two ExecutionModes, determined at token
construction time:
1. SERIAL: submitted tasks are run one at a time.
2. CONCURRENT: submitted tasks may be run in parallel.
This isn't unlike submitted without a token, but the logical grouping that tokens
impart can be useful when a pool is shared by many contexts (e.g. to
safely shut down one context, to derive context-specific metrics, etc.).
Tasks submitted without a token or via ExecutionMode::CONCURRENT tokens are
processed in FIFO order. On the other hand, ExecutionMode::SERIAL tokens are
processed in a round-robin fashion, one task at a time. This prevents them
from starving one another. However, tokenless (and CONCURRENT token-based)
tasks can starve SERIAL token-based tasks.
Thread design point:
1. It is a thin wrapper around pthread that can register itself with the singleton ThreadMgr
(a private class implemented in thread.cpp entirely, which tracks all live threads so
that they may be monitored via the debug webpages). This class has a limited subset of
boost::thread's API. Construction is almost the same, but clients must supply a
category and a name for each thread so that they can be identified in the debug web
UI. Otherwise, join() is the only supported method from boost::thread.
2. Each Thread object knows its operating system thread ID (TID), which can be used to
attach debuggers to specific threads, to retrieve resource-usage statistics from the
operating system, and to assign threads to resource control groups.
3. Threads are shared objects, but in a degenerate way. They may only have
up to two referents: the caller that created the thread (parent), and
the thread itself (child). Moreover, the only two methods to mutate state
(join() and the destructor) are constrained: the child may not join() on
itself, and the destructor is only run when there's one referent left.
These constraints allow us to access thread internals without any locks.
127 lines
3.8 KiB
C++
127 lines
3.8 KiB
C++
// Licensed to the Apache Software Foundation (ASF) under one
|
|
// or more contributor license agreements. See the NOTICE file
|
|
// distributed with this work for additional information
|
|
// regarding copyright ownership. The ASF licenses this file
|
|
// to you 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 "util/thread.h"
|
|
|
|
#include <sys/types.h>
|
|
#include <unistd.h>
|
|
|
|
#include <ostream>
|
|
#include <string>
|
|
#include <vector>
|
|
|
|
#include <gtest/gtest.h>
|
|
|
|
#include "common/status.h"
|
|
#include "common/logging.h"
|
|
#include "gutil/basictypes.h"
|
|
#include "gutil/ref_counted.h"
|
|
#include "util/countdown_latch.h"
|
|
#include "util/runtime_profile.h"
|
|
|
|
using std::string;
|
|
|
|
namespace doris {
|
|
|
|
class ThreadTest : public ::testing::Test {
|
|
public:
|
|
virtual void SetUp() {}
|
|
virtual void TearDown() {}
|
|
};
|
|
|
|
// Join with a thread and emit warnings while waiting to join.
|
|
// This has to be manually verified.
|
|
TEST_F(ThreadTest, TestJoinAndWarn) {
|
|
scoped_refptr<Thread> holder;
|
|
Status status = Thread::create("test", "sleeper thread", usleep, 1000*1000, &holder);
|
|
ASSERT_TRUE(status.ok());
|
|
status = ThreadJoiner(holder.get())
|
|
.warn_after_ms(10)
|
|
.warn_every_ms(100)
|
|
.join();
|
|
ASSERT_TRUE(status.ok());
|
|
}
|
|
|
|
TEST_F(ThreadTest, TestFailedJoin) {
|
|
scoped_refptr<Thread> holder;
|
|
Status status = Thread::create("test", "sleeper thread", usleep, 1000*1000, &holder);
|
|
ASSERT_TRUE(status.ok());
|
|
status = ThreadJoiner(holder.get())
|
|
.give_up_after_ms(50)
|
|
.join();
|
|
ASSERT_TRUE(status.is_aborted());
|
|
}
|
|
|
|
static void TryJoinOnSelf() {
|
|
Status s = ThreadJoiner(Thread::current_thread()).join();
|
|
// Use CHECK instead of ASSERT because gtest isn't thread-safe.
|
|
CHECK(s.is_invalid_argument());
|
|
}
|
|
|
|
// Try to join on the thread that is currently running.
|
|
TEST_F(ThreadTest, TestJoinOnSelf) {
|
|
scoped_refptr<Thread> holder;
|
|
ASSERT_TRUE(Thread::create("test", "test", TryJoinOnSelf, &holder).ok());
|
|
holder->join();
|
|
// Actual assertion is done by the thread spawned above.
|
|
}
|
|
|
|
TEST_F(ThreadTest, TestDoubleJoinIsNoOp) {
|
|
scoped_refptr<Thread> holder;
|
|
Status status = Thread::create("test", "sleeper thread", usleep, 0, &holder);
|
|
ASSERT_TRUE(status.ok());
|
|
ThreadJoiner joiner(holder.get());
|
|
status = joiner.join();
|
|
ASSERT_TRUE(status.ok());
|
|
status = joiner.join();
|
|
ASSERT_TRUE(status.ok());
|
|
}
|
|
|
|
TEST_F(ThreadTest, ThreadStartBenchmark) {
|
|
std::vector<scoped_refptr<Thread>> threads(1000);
|
|
{
|
|
int64_t thread_creation_ns = 0;
|
|
SCOPED_RAW_TIMER(&thread_creation_ns);
|
|
for (auto& t : threads) {
|
|
Status status = Thread::create("test", "TestCallOnExit", usleep, 0, &t);
|
|
ASSERT_TRUE(status.ok());
|
|
}
|
|
std::cout << "create 1000 threads use:"
|
|
<< thread_creation_ns << "ns" << std::endl;
|
|
}
|
|
{
|
|
int64_t thread_publish_tid_ns = 0;
|
|
SCOPED_RAW_TIMER(&thread_publish_tid_ns);
|
|
for (auto& t : threads) {
|
|
t->tid();
|
|
}
|
|
std::cout << "1000 threads publish TIDS use:"
|
|
<< thread_publish_tid_ns << "ns" << std::endl;
|
|
}
|
|
|
|
for (auto& t : threads) {
|
|
t->join();
|
|
}
|
|
}
|
|
|
|
} // namespace doris
|
|
|
|
int main(int argc, char* argv[]) {
|
|
::testing::InitGoogleTest(&argc, argv);
|
|
return RUN_ALL_TESTS();
|
|
}
|