void* ThreadPoolWorker::Callback(void* arg) {
ThreadPoolWorker* worker = reinterpret_cast<ThreadPoolWorker*>(arg);
Runtime* runtime = Runtime::Current();
- CHECK(runtime->AttachCurrentThread(worker->name_.c_str(), true, nullptr, false));
+ CHECK(runtime->AttachCurrentThread(worker->name_.c_str(),
+ true,
+ nullptr,
+ worker->thread_pool_->create_peers_));
worker->thread_ = Thread::Current();
// Thread pool workers cannot call into java.
worker->thread_->SetCanCallIntoJava(false);
tasks_.clear();
}
-ThreadPool::ThreadPool(const char* name, size_t num_threads)
+ThreadPool::ThreadPool(const char* name, size_t num_threads, bool create_peers)
: name_(name),
task_queue_lock_("task queue lock"),
task_queue_condition_("task queue condition", task_queue_lock_),
total_wait_time_(0),
// Add one since the caller of constructor waits on the barrier too.
creation_barier_(num_threads + 1),
- max_active_workers_(num_threads) {
+ max_active_workers_(num_threads),
+ create_peers_(create_peers) {
Thread* self = Thread::Current();
while (GetThreadCount() < num_threads) {
const std::string worker_name = StringPrintf("%s worker thread %zu", name_.c_str(),
void ThreadPool::Wait(Thread* self, bool do_work, bool may_hold_locks) {
if (do_work) {
+ CHECK(!create_peers_);
Task* task = nullptr;
while ((task = TryGetTask(self)) != nullptr) {
task->Run(self);
// Remove all tasks in the queue.
void RemoveAllTasks(Thread* self) REQUIRES(!task_queue_lock_);
- ThreadPool(const char* name, size_t num_threads);
+ // Create a named thread pool with the given number of threads.
+ //
+ // If create_peers is true, all worker threads will have a Java peer object. Note that if the
+ // pool is asked to do work on the current thread (see Wait), a peer may not be available. Wait
+ // will conservatively abort if create_peers and do_work are true.
+ ThreadPool(const char* name, size_t num_threads, bool create_peers = false);
virtual ~ThreadPool();
// Wait for all tasks currently on queue to get completed. If the pool has been stopped, only
// wait till all already running tasks are done.
+ // When the pool was created with peers for workers, do_work must not be true (see ThreadPool()).
void Wait(Thread* self, bool do_work, bool may_hold_locks) REQUIRES(!task_queue_lock_);
size_t GetTaskCount(Thread* self) REQUIRES(!task_queue_lock_);
uint64_t total_wait_time_;
Barrier creation_barier_;
size_t max_active_workers_ GUARDED_BY(task_queue_lock_);
+ const bool create_peers_;
private:
friend class ThreadPoolWorker;
#include "atomic.h"
#include "common_runtime_test.h"
+#include "scoped_thread_state_change-inl.h"
#include "thread-inl.h"
namespace art {
EXPECT_EQ((1 << depth) - 1, count.LoadSequentiallyConsistent());
}
+class PeerTask : public Task {
+ public:
+ PeerTask() {}
+
+ void Run(Thread* self) {
+ ScopedObjectAccess soa(self);
+ CHECK(self->GetPeer() != nullptr);
+ }
+
+ void Finalize() {
+ delete this;
+ }
+};
+
+class NoPeerTask : public Task {
+ public:
+ NoPeerTask() {}
+
+ void Run(Thread* self) {
+ ScopedObjectAccess soa(self);
+ CHECK(self->GetPeer() == nullptr);
+ }
+
+ void Finalize() {
+ delete this;
+ }
+};
+
+// Tests for create_peer functionality.
+TEST_F(ThreadPoolTest, PeerTest) {
+ Thread* self = Thread::Current();
+ {
+ ThreadPool thread_pool("Thread pool test thread pool", 1);
+ thread_pool.AddTask(self, new NoPeerTask());
+ thread_pool.StartWorkers(self);
+ thread_pool.Wait(self, false, false);
+ }
+
+ {
+ // To create peers, the runtime needs to be started.
+ self->TransitionFromSuspendedToRunnable();
+ bool started = runtime_->Start();
+ ASSERT_TRUE(started);
+
+ ThreadPool thread_pool("Thread pool test thread pool", 1, true);
+ thread_pool.AddTask(self, new PeerTask());
+ thread_pool.StartWorkers(self);
+ thread_pool.Wait(self, false, false);
+ }
+}
+
} // namespace art