¡¾Chromium¡¿Base¿âµÄSimpleThread

//?ÔÚÕâÀïÃæʵÏÖ×Ô¶¨ÒåµÄÏß³ÌÔËÐÐÂß¼­
 class?MyThreadRunner?:?public?DelegateSimpleThread::Delegate?{?...?};

 MyThreadRunner?runner;
 DelegateSimpleThread?thread(&runner,?"good\_name\_here");

 //?Start?·½·¨½«Ôڳɹ¦Æô¶¯ºÍ³õʼ»¯Ï̺߳󷵻ء£
 //?д´½¨µÄÏ߳̽«µ÷ÓÃ?runner->Run()£¬²¢Ò»Ö±ÔËÐÐÖ±µ½·µ»Ø¡£
 thread.Start();

 //?thread.Join()?·½·¨½«µÈ´ýÏß³ÌÍ˳ö¡£±ØÐë?µ÷ÓÃ?Join£¡
 thread.Join();
 //?SimpleThread?¶ÔÏóÈÔÈ»ÓÐЧ£¬µ«ÊÇÄã²»ÄÜÔٴε÷ÓÃ?Join?»ò?Start?·½·¨¡£

struct BASE_EXPORT Options {
? public:
? Options() = default;
? explicit Options(ThreadPriority priority_in) : priority(priority_in) {}
? ~Options() = default;

? // Allow copies.
? Options(const Options& other) = default;
? Options& operator=(const Options& other) = default;

? // A custom stack size, or 0 for the system default.
? size_t stack_size = 0;

? ThreadPriority priority = ThreadPriority::NORMAL;

? // If false, the underlying thread's PlatformThreadHandle will not be kept
? // around and as such the SimpleThread instance will not be Join()able and
? // must not be deleted before Run() is invoked. After that, it's up to
? // the subclass to determine when it is safe to delete itself.
? bool joinable = true;
};

// Valid values for priority of Thread::Options and SimpleThread::Options, and
// SetCurrentThreadPriority(), listed in increasing order of importance.
enum class ThreadPriority : int {
??// ÊÊÓÃÓÚ²»Ó¦ÖжϸßÓÅÏȼ¶¹¤×÷µÄÏ̡߳£ ??
? BACKGROUND, ??? 
? // ĬÈÏÓÅÏȼ¶¼¶±ð¡£ ???
? NORMAL, ??? 
? // ÊÊÓÃÓÚΪÏÔʾÉú³ÉÊý¾ÝµÄỊ̈߳¨´óÔ¼ 60Hz£©¡£
? DISPLAY, ??? 
? // ÊÊÓÃÓÚµÍÑÓ³Ù¡¢¿¹¸ÉÈŵÄÒôƵ¡£
? REALTIME_AUDIO,
};

// ´´½¨Ò»¸ö SimpleThread¡£|options| Ó¦¸ÃÓÃÓÚ¹ÜÀíÉæ¼°Ï̴߳´½¨ºÍ¹ÜÀíµÄÌض¨ÅäÖá£
// ÿ¸öÏ̶߳¼ÓÐÒ»¸öÃû³Æ£¬ËüÊÇÒ»¸öÏÔʾ×Ö·û´®£¬ÓÃÓÚ±êʶÏ̡߳£
// Ö±µ½µ÷Óà Start() ·½·¨Ö®Ç°£¬Ï̲߳»»á±»´´½¨¡£
explicit SimpleThread(const std::string& name);
SimpleThread(const std::string& name, const Options& options);

~SimpleThread() override;

// Æô¶¯Ị̈߳¬²¢ÔÚÏß³ÌÆô¶¯ºÍ³õʼ»¯ºó£¨¼´µ÷Óà ThreadMain() ºó£©²Å·µ»Ø¡£
void Start();

// ¼ÓÈëÏ̡߳£Èç¹ûʹÓà StartAsync() Æô¶¯Ị̈߳¬ÔòÊ×ÏȵȴýÏ̸߳ɾ»µØÆô¶¯£¬È»ºó¼ÓÈëÏ̡߳£
void Join();

// Æô¶¯Ị̈߳¬µ«Á¢¼´·µ»Ø£¬¶ø²»µÈ´ýÏß³ÌÊ×ÏȽøÐгõʼ»¯£¨¼´²»µÈ´ý ThreadMain() ±»ÔËÐУ©¡£
void StartAsync();

// ×ÓÀàÓ¦¸ÃÖØд Run ·½·¨¡£
virtual void Run() = 0;

// ·µ»ØÏß³Ì ID£¬ÔÚÏß³ÌÆô¶¯ºó²ÅÓÐЧ¡£Èç¹ûʹÓà Start() Æô¶¯Ị̈߳¬ÔòÔÚµ÷Óà Start() ºóËü½«ÓÐЧ¡£
// Èç¹ûʹÓà StartAsync() Æô¶¯Ị̈߳¬ÔòÔÚ HasBeenStarted() ·µ»Ø True ֮ǰ²»Äܵ÷Óô˷½·¨¡£
PlatformThreadId tid();

// Èç¹ûÏß³ÌÒÑÆô¶¯²¢³õʼ»¯£¨¼´ ThreadMain() ÒÑÔËÐУ©£¬Ôò·µ»Ø True¡£
// Èç¹ûʹÓà StartAsync() Æô¶¯Ị̈߳¬µ«ÉÐδ³õʼ»¯£¨¼´ ThreadMain() ÉÐδÔËÐУ©£¬Ôò·µ»Ø False¡£
bool HasBeenStarted();

// Èç¹ûÔø¾­µ÷Óùý Join()£¬Ôò·µ»Ø True¡£
bool HasBeenJoined() const { return joined_; }

// Èç¹ûµ÷Óùý Start() »ò StartAsync()£¬Ôò·µ»Ø true¡£
bool HasStartBeenAttempted() { return start_called_; }

// ´Ó PlatformThread::Delegate ÖÐÖØд£º
void ThreadMain() override;

const std::string name_;
const Options options_;
PlatformThreadHandle thread_;? // PlatformThread handle, reset after Join.
WaitableEvent event_;????????? // Signaled if Start() was ever called.
PlatformThreadId tid_ = kInvalidThreadId;? // The backing thread's id.
bool joined_ = false;????????????????????? // True if Join has been called.
// Set to true when the platform-thread creation has started.
bool start_called_ = false;

void SimpleThread::Start() {
? StartAsync();
? ScopedAllowBaseSyncPrimitives allow_wait;
? event_.Wait();? // Wait for the thread to complete initialization.
}

void SimpleThread::StartAsync() {
? DCHECK(!HasStartBeenAttempted()) << "Tried to Start a thread multiple times.";
? start_called_ = true;
? BeforeStart();
? bool success =
? ? ? options_.joinable
? ? ? ? ? ? PlatformThread::CreateWithPriority(options_.stack_size, this,
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?? &thread_, options_.priority)
? ? ? ? ? : PlatformThread::CreateNonJoinableWithPriority(
? ? ? ? ? ? ? ? options_.stack_size, this, options_.priority);
? CHECK(success);
}

void SimpleThread::ThreadMain() {
? tid_ = PlatformThread::CurrentId();
? PlatformThread::SetName(name_);

? // We've initialized our new thread, signal that we're done to Start().
? event_.Signal();

? BeforeRun();
? Run();
}

void SimpleThread::Join() {
? DCHECK(options_.joinable) << "A non-joinable thread can't be joined.";
? DCHECK(HasStartBeenAttempted()) << "Tried to Join a never-started thread.";
? DCHECK(!HasBeenJoined()) << "Tried to Join a thread multiple times.";
? BeforeJoin();
? PlatformThread::Join(thread_);
? thread_ = PlatformThreadHandle();
? joined_ = true;
}

? // Subclasses should override the Run method.
? virtual void Run() = 0;

class BASE_EXPORT DelegateSimpleThread : public SimpleThread {
?public:
? class BASE_EXPORT Delegate {
?? public:
??? virtual ~Delegate() = default;
??? virtual void Run() = 0;
? };

? DelegateSimpleThread(Delegate* delegate,
?????????????????????? const std::string& name_prefix);
? DelegateSimpleThread(Delegate* delegate,
?????????????????????? const std::string& name_prefix,
?????????????????????? const Options& options);

? ~DelegateSimpleThread() override;
? void Run() override;

?private:
? Delegate* delegate_;

? DISALLOW_COPY_AND_ASSIGN(DelegateSimpleThread);
};

void DelegateSimpleThread::Run() {
? DCHECK(delegate_) << "Tried to call Run without a delegate (called twice?)";

? // Non-joinable DelegateSimpleThreads are allowed to be deleted during Run().
? // Member state must not be accessed after invoking Run().
? Delegate* delegate = delegate_;
? delegate_ = nullptr;
? delegate->Run();
}

class BASE_EXPORT DelegateSimpleThreadPool
??? : public DelegateSimpleThread::Delegate {
?public:
? typedef DelegateSimpleThread::Delegate Delegate;

? DelegateSimpleThreadPool(const std::string& name_prefix, int num_threads);
? ~DelegateSimpleThreadPool() override;

? // Start up all of the underlying threads, and start processing work if we
? // have any.
? void Start();

? // Make sure all outstanding work is finished, and wait for and destroy all
? // of the underlying threads in the pool.
? void JoinAll();

? // It is safe to AddWork() any time, before or after Start().
? // Delegate* should always be a valid pointer, NULL is reserved internally.
? void AddWork(Delegate* work, int repeat_count);
? void AddWork(Delegate* work) {
??? AddWork(work, 1);
? }

? // We implement the Delegate interface, for running our internal threads.
? void Run() override;

?private:
? const std::string name_prefix_;
? int num_threads_;
? std::vector<DelegateSimpleThread*> threads_;
? base::queue<Delegate*> delegates_;
? base::Lock lock_;??????????? // Locks delegates_
? WaitableEvent dry_;??? // Not signaled when there is no work to do.

? DISALLOW_COPY_AND_ASSIGN(DelegateSimpleThreadPool);
};

DelegateSimpleThreadPool::DelegateSimpleThreadPool(
??? const std::string& name_prefix,
??? int num_threads)
??? : name_prefix_(name_prefix),
????? num_threads_(num_threads),
????? dry_(WaitableEvent::ResetPolicy::MANUAL,
?????????? WaitableEvent::InitialState::NOT_SIGNALED) {}

void DelegateSimpleThreadPool::Start() {
? DCHECK(threads_.empty()) << "Start() called with outstanding threads.";
? for (int i = 0; i < num_threads_; ++i) {
??? std::string name(name_prefix_);
??? name.push_back('/');
??? name.append(NumberToString(i));
??? DelegateSimpleThread* thread = new DelegateSimpleThread(this, name);
??? thread->Start();
??? threads_.push_back(thread);
? }
}

void DelegateSimpleThreadPool::Run() {
? Delegate* work = nullptr;

? while (true) {
??? dry_.Wait();
??? {
????? AutoLock locked(lock_);
????? if (!dry_.IsSignaled())
??????? continue;

????? DCHECK(!delegates_.empty());
????? work = delegates_.front();
????? delegates_.pop();

????? // Signal to any other threads that we're currently out of work.
????? if (delegates_.empty())
??????? dry_.Reset();
??? }

??? // A NULL delegate pointer signals us to quit.
??? if (!work)
????? break;

??? work->Run();
? }
}

void DelegateSimpleThreadPool::JoinAll() {
? DCHECK(!threads_.empty()) << "JoinAll() called with no outstanding threads.";

? // Tell all our threads to quit their worker loop.
? AddWork(nullptr, num_threads_);

? // Join and destroy all the worker threads.
? for (int i = 0; i < num_threads_; ++i) {
? ? threads_[i]->Join();
? ? delete threads_[i];
? }
? threads_.clear();
? DCHECK(delegates_.empty());
}

void DelegateSimpleThreadPool::AddWork(Delegate* delegate, int repeat_count) {
? AutoLock locked(lock_);
? for (int i = 0; i < repeat_count; ++i)
??? delegates_.push(delegate);
? // If we were empty, signal that we have work now.
? if (!dry_.IsSignaled())
??? dry_.Signal();
}

class SetIntRunner : public DelegateSimpleThread::Delegate {
public:
??? SetIntRunner(int* ptr, int val) : ptr_(ptr), val_(val) {}
??? ~SetIntRunner() override = default;

private:
??? void Run() override { *ptr_ = val_; }

??? int* ptr_;
??? int val_;

??? DISALLOW_COPY_AND_ASSIGN(SetIntRunner);
};

int stack_int = 0;

SetIntRunner runner(&stack_int, 7);
EXPECT_EQ(0, stack_int);

DelegateSimpleThread thread(&runner, "int_setter");
EXPECT_FALSE(thread.HasBeenStarted());
EXPECT_FALSE(thread.HasBeenJoined());
EXPECT_EQ(0, stack_int);

thread.Start();
EXPECT_TRUE(thread.HasBeenStarted());
EXPECT_FALSE(thread.HasBeenJoined());

thread.Join();
EXPECT_TRUE(thread.HasBeenStarted());
EXPECT_TRUE(thread.HasBeenJoined());
EXPECT_EQ(7, stack_int);

class WaitEventRunner : public DelegateSimpleThread::Delegate {
public:
??? explicit WaitEventRunner(WaitableEvent* event) : event_(event) {}
??? ~WaitEventRunner() override = default;

private:
??? void Run() override {
??????? EXPECT_FALSE(event_->IsSignaled());
? ? ? 	// ×öһЩÒì²½ºÄʱ²Ù×÷£¬×öÍêºó´¥·¢ÐźŸæËßÈÎÎñÒѾ­Íê³É
??????? event_->Signal();
??????? EXPECT_TRUE(event_->IsSignaled());
??? }

??? WaitableEvent* event_;

??? DISALLOW_COPY_AND_ASSIGN(WaitEventRunner);
};

// Create a thread, and wait for it to signal us.
WaitableEvent event(WaitableEvent::ResetPolicy::MANUAL,
??? WaitableEvent::InitialState::NOT_SIGNALED);

WaitEventRunner runner(&event);
DelegateSimpleThread thread(&runner, "event_waiter");

EXPECT_FALSE(event.IsSignaled());
thread.Start();
event.Wait();
EXPECT_TRUE(event.IsSignaled());
thread.Join();

// AtomicSequenceNumber is a thread safe increasing sequence number generator.
// Its constructor doesn't emit a static initializer, so it's safe to use as a
// global variable or static member.
class AtomicSequenceNumber {
?public:
? constexpr AtomicSequenceNumber() = default;

? // Returns an increasing sequence number starts from 0 for each call.
? // This function can be called from any thread without data race.
? inline int GetNext() { return seq_.fetch_add(1, std::memory_order_relaxed); }

?private:
? std::atomic_int seq_{0};

? DISALLOW_COPY_AND_ASSIGN(AtomicSequenceNumber);
};

class SeqRunner : public DelegateSimpleThread::Delegate {
public:
??? explicit SeqRunner(AtomicSequenceNumber* seq) : seq_(seq) {}

private:
??? void Run() override { seq_->GetNext(); }

??? AtomicSequenceNumber* seq_;

??? DISALLOW_COPY_AND_ASSIGN(SeqRunner);
};

AtomicSequenceNumber seq;
SeqRunner runner(&seq);
DelegateSimpleThreadPool pool("seq_runner", 10);

// Add work before we're running.
pool.AddWork(&runner, 300);

EXPECT_EQ(seq.GetNext(), 0);
pool.Start();

// Add work while we're running.
pool.AddWork(&runner, 300);

pool.JoinAll();

EXPECT_EQ(seq.GetNext(), 601);