Commit 7f8a21e8 authored by Etienne Pierre-doray's avatar Etienne Pierre-doray Committed by Commit Bot

Reland "Reland "[Jobs API]: Use worker_lock in JobTaskSource.""

This is a reland of fb56fbc5
Issue: Trace event under lock cause PostTask and lock inversion.
Fix: Move trace event outside of lock (to both callsites of
WaitForParticipationOpportunity), this
should give us mostly the same information but scope is a bit bigger.
See patchset 2

Original change's description:
> Reland "[Jobs API]: Use worker_lock in JobTaskSource."
>
> This is a reland of 6eb566d2
>
> Reason for revert: Cause failure in CheckedLockImpl::Acquire
> crbug.com/1099649
> Mark worker_released_condition_ declare_only_used_while_idle
> to prevent priority queue from being acquired in ScopedBlockingCall.
>
> Original change's description:
> > [Jobs API]: Use worker_lock in JobTaskSource.
> >
> > Possible race when Join():
> > - thread A: Join: worker_released_condition_->Wait()
> > - thread C: WillRunTask:
> >               GetMaxConcurrency() returns > 0
> > - thread B: already running, finishes all the work
> >               GetMaxConcurrency() goes 0
> > - thread B: DidProcessTask:
> >               worker_released_condition_->Signal(),
> > - thread A: Join returns (GetMaxConcurrency() is 0)
> > - thread C: TryIncrementWorkerCountFromWorkerRelease
> >               worker count goes 1
> > - thread C: runs worker_task after Join
> >
> > To fix race when Joining, all writes to |state_| are protected by
> > |worker_lock|. Memory ordering is no longer necessary.
> >
> > Alternative: cancel before Join returns with a compare and swap and
> > loop again if new workers.
> >
> > Change-Id: I4e478ffae2bdaec56386739f78de089d0e74e42c
> > Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/2248159
> > Commit-Queue: Etienne Pierre-Doray <etiennep@chromium.org>
> > Reviewed-by: Gabriel Charette <gab@chromium.org>
> > Cr-Commit-Position: refs/heads/master@{#781453}
>
> Change-Id: I1c7c0054a52b9b12dd6d0edd049ab2a7912df361
> Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/2272942
> Commit-Queue: Etienne Pierre-Doray <etiennep@chromium.org>
> Reviewed-by: Gabriel Charette <gab@chromium.org>
> Cr-Commit-Position: refs/heads/master@{#789526}

Change-Id: I0b08ee9fb7efd6b5cfdf9d171f2a280ccbb1fa5d
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/2309572Reviewed-by: default avatarGabriel Charette <gab@chromium.org>
Commit-Queue: Etienne Pierre-Doray <etiennep@chromium.org>
Cr-Commit-Position: refs/heads/master@{#790489}
parent 30003e96
This diff is collapsed.
...@@ -15,7 +15,7 @@ ...@@ -15,7 +15,7 @@
#include "base/macros.h" #include "base/macros.h"
#include "base/optional.h" #include "base/optional.h"
#include "base/synchronization/condition_variable.h" #include "base/synchronization/condition_variable.h"
#include "base/synchronization/lock.h" #include "base/task/common/checked_lock.h"
#include "base/task/post_job.h" #include "base/task/post_job.h"
#include "base/task/task_traits.h" #include "base/task/task_traits.h"
#include "base/task/thread_pool/sequence_sort_key.h" #include "base/task/thread_pool/sequence_sort_key.h"
...@@ -91,7 +91,9 @@ class BASE_EXPORT JobTaskSource : public TaskSource { ...@@ -91,7 +91,9 @@ class BASE_EXPORT JobTaskSource : public TaskSource {
private: private:
// Atomic internal state to track the number of workers running a task from // Atomic internal state to track the number of workers running a task from
// this JobTaskSource and whether this JobTaskSource is canceled. // this JobTaskSource and whether this JobTaskSource is canceled. All
// operations are performed with std::memory_order_relaxed as State is only
// ever modified under a lock or read atomically (optimistic read).
class State { class State {
public: public:
static constexpr size_t kCanceledMask = 1; static constexpr size_t kCanceledMask = 1;
...@@ -109,28 +111,17 @@ class BASE_EXPORT JobTaskSource : public TaskSource { ...@@ -109,28 +111,17 @@ class BASE_EXPORT JobTaskSource : public TaskSource {
State(); State();
~State(); ~State();
// Sets as canceled using std::memory_order_relaxed. Returns the state // Sets as canceled. Returns the state
// before the operation. // before the operation.
Value Cancel(); Value Cancel();
// Increments the worker count by 1 if smaller than |max_concurrency| and if // Increments the worker count by 1. Returns the state before the operation.
// |!is_canceled()|, using std::memory_order_release, and returns the state Value IncrementWorkerCount();
// before the operation. Equivalent to Load() otherwise.
Value TryIncrementWorkerCountFromWorkerRelease(size_t max_concurrency);
// Decrements the worker count by 1 using std::memory_order_acquire. Returns // Decrements the worker count by 1. Returns the state before the operation.
// the state before the operation. Value DecrementWorkerCount();
Value DecrementWorkerCountFromWorkerAcquire();
// Increments the worker count by 1 using std::memory_order_relaxed. Returns // Loads and returns the state.
// the state before the operation.
Value IncrementWorkerCountFromJoiningThread();
// Decrements the worker count by 1 using std::memory_order_relaxed. Returns
// the state before the operation.
Value DecrementWorkerCountFromJoiningThread();
// Loads and returns the state, using std::memory_order_relaxed.
Value Load() const; Value Load() const;
private: private:
...@@ -186,7 +177,7 @@ class BASE_EXPORT JobTaskSource : public TaskSource { ...@@ -186,7 +177,7 @@ class BASE_EXPORT JobTaskSource : public TaskSource {
// DidProcessTask()). Returns true if the joining thread should run a task, or // DidProcessTask()). Returns true if the joining thread should run a task, or
// false if joining was completed and all other workers returned because // false if joining was completed and all other workers returned because
// either there's no work remaining or Job was cancelled. // either there's no work remaining or Job was cancelled.
bool WaitForParticipationOpportunity(); bool WaitForParticipationOpportunity() EXCLUSIVE_LOCKS_REQUIRED(worker_lock_);
// TaskSource: // TaskSource:
RunStatus WillRunTask() override; RunStatus WillRunTask() override;
...@@ -195,15 +186,20 @@ class BASE_EXPORT JobTaskSource : public TaskSource { ...@@ -195,15 +186,20 @@ class BASE_EXPORT JobTaskSource : public TaskSource {
bool DidProcessTask(TaskSource::Transaction* transaction) override; bool DidProcessTask(TaskSource::Transaction* transaction) override;
SequenceSortKey GetSortKey() const override; SequenceSortKey GetSortKey() const override;
// Current atomic state. // Synchronizes access to workers state.
State state_; mutable CheckedLock worker_lock_{UniversalSuccessor()};
std::atomic<uint32_t> assigned_task_ids_{0};
// Current atomic state (atomic despite the lock to allow optimistic reads
// without the lock).
State state_ GUARDED_BY(worker_lock_);
// Normally, |join_flag_| is protected by |lock_|, except in ShouldYield() // Normally, |join_flag_| is protected by |lock_|, except in ShouldYield()
// hence the use of atomics. // hence the use of atomics.
JoinFlag join_flag_ GUARDED_BY(lock_); JoinFlag join_flag_ GUARDED_BY(worker_lock_);
// Signaled when |join_flag_| is kWaiting* and a worker returns. // Signaled when |join_flag_| is kWaiting* and a worker returns.
std::unique_ptr<ConditionVariable> worker_released_condition_ std::unique_ptr<ConditionVariable> worker_released_condition_
GUARDED_BY(lock_); GUARDED_BY(worker_lock_);
std::atomic<uint32_t> assigned_task_ids_{0};
const Location from_here_; const Location from_here_;
RepeatingCallback<size_t()> max_concurrency_callback_; RepeatingCallback<size_t()> max_concurrency_callback_;
...@@ -217,12 +213,10 @@ class BASE_EXPORT JobTaskSource : public TaskSource { ...@@ -217,12 +213,10 @@ class BASE_EXPORT JobTaskSource : public TaskSource {
PooledTaskRunnerDelegate* delegate_; PooledTaskRunnerDelegate* delegate_;
#if DCHECK_IS_ON() #if DCHECK_IS_ON()
// Synchronizes accesses to |increase_version_|. // Signaled whenever |increase_version_| is updated.
mutable Lock version_lock_; std::unique_ptr<ConditionVariable> version_condition_for_dcheck_;
// Signaled whenever increase_version_ is updated.
ConditionVariable version_condition_{&version_lock_};
// Incremented every time max concurrency is increased. // Incremented every time max concurrency is increased.
size_t increase_version_ GUARDED_BY(version_lock_) = 0; size_t increase_version_ GUARDED_BY(worker_lock_) = 0;
#endif // DCHECK_IS_ON() #endif // DCHECK_IS_ON()
DISALLOW_COPY_AND_ASSIGN(JobTaskSource); DISALLOW_COPY_AND_ASSIGN(JobTaskSource);
......
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