| OLD | NEW |
| 1 // Copyright (c) 2012 The Chromium Authors. All rights reserved. | 1 // Copyright (c) 2012 The Chromium Authors. All rights reserved. |
| 2 // Use of this source code is governed by a BSD-style license that can be | 2 // Use of this source code is governed by a BSD-style license that can be |
| 3 // found in the LICENSE file. | 3 // found in the LICENSE file. |
| 4 | 4 |
| 5 #include "base/threading/sequenced_worker_pool.h" | 5 #include "base/threading/sequenced_worker_pool.h" |
| 6 | 6 |
| 7 #include <stddef.h> | 7 #include <stddef.h> |
| 8 | 8 |
| 9 #include <algorithm> | 9 #include <algorithm> |
| 10 #include <memory> | 10 #include <memory> |
| 11 | 11 |
| 12 #include "base/bind.h" | 12 #include "base/bind.h" |
| 13 #include "base/compiler_specific.h" | 13 #include "base/compiler_specific.h" |
| 14 #include "base/macros.h" | 14 #include "base/macros.h" |
| 15 #include "base/memory/ref_counted.h" | 15 #include "base/memory/ref_counted.h" |
| 16 #include "base/message_loop/message_loop.h" | 16 #include "base/message_loop/message_loop.h" |
| 17 #include "base/sequence_checker_impl.h" |
| 17 #include "base/stl_util.h" | 18 #include "base/stl_util.h" |
| 18 #include "base/synchronization/condition_variable.h" | 19 #include "base/synchronization/condition_variable.h" |
| 19 #include "base/synchronization/lock.h" | 20 #include "base/synchronization/lock.h" |
| 21 #include "base/synchronization/waitable_event.h" |
| 20 #include "base/test/sequenced_task_runner_test_template.h" | 22 #include "base/test/sequenced_task_runner_test_template.h" |
| 21 #include "base/test/sequenced_worker_pool_owner.h" | 23 #include "base/test/sequenced_worker_pool_owner.h" |
| 22 #include "base/test/task_runner_test_template.h" | 24 #include "base/test/task_runner_test_template.h" |
| 23 #include "base/test/test_timeouts.h" | 25 #include "base/test/test_timeouts.h" |
| 24 #include "base/threading/platform_thread.h" | 26 #include "base/threading/platform_thread.h" |
| 25 #include "base/time/time.h" | 27 #include "base/time/time.h" |
| 26 #include "base/tracked_objects.h" | 28 #include "base/tracked_objects.h" |
| 27 #include "testing/gtest/include/gtest/gtest.h" | 29 #include "testing/gtest/include/gtest/gtest.h" |
| 28 | 30 |
| 29 namespace base { | 31 namespace base { |
| (...skipping 858 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
| 888 // Should be fine to call on an idle instance with all threads created, and | 890 // Should be fine to call on an idle instance with all threads created, and |
| 889 // spamming the method shouldn't deadlock or confuse the class. | 891 // spamming the method shouldn't deadlock or confuse the class. |
| 890 pool()->FlushForTesting(); | 892 pool()->FlushForTesting(); |
| 891 pool()->FlushForTesting(); | 893 pool()->FlushForTesting(); |
| 892 | 894 |
| 893 // Should be fine to call after shutdown too. | 895 // Should be fine to call after shutdown too. |
| 894 pool()->Shutdown(); | 896 pool()->Shutdown(); |
| 895 pool()->FlushForTesting(); | 897 pool()->FlushForTesting(); |
| 896 } | 898 } |
| 897 | 899 |
| 898 namespace { | 900 // Helper method for VerifyCurrentSequencedTaskRunner() and |
| 899 | 901 // VerifyCurrentSequencedTaskRunnerForUnsequencedTask(). |
| 900 void CheckWorkerPoolAndSequenceToken( | 902 void VerifySequencedTaskRunnerRunsOnCurrentThread( |
| 901 const scoped_refptr<SequencedWorkerPool>& expected_pool, | 903 SequencedTaskRunner* task_runner, |
| 902 SequencedWorkerPool::SequenceToken expected_token) { | 904 bool should_run_on_current_thread, |
| 903 SequencedWorkerPool::SequenceToken token = | 905 const Closure& callback) { |
| 904 SequencedWorkerPool::GetSequenceTokenForCurrentThread(); | 906 EXPECT_EQ(should_run_on_current_thread, |
| 905 EXPECT_EQ(expected_token.ToString(), token.ToString()); | 907 task_runner->RunsTasksOnCurrentThread()); |
| 906 | 908 callback.Run(); |
| 907 scoped_refptr<SequencedWorkerPool> pool = | |
| 908 SequencedWorkerPool::GetWorkerPoolForCurrentThread(); | |
| 909 EXPECT_EQ(expected_pool, pool); | |
| 910 } | 909 } |
| 911 | 910 |
| 912 } // namespace | 911 void VerifyCurrentSequencedTaskRunner( |
| 912 SequencedTaskRunner* expected_task_runner, |
| 913 bool expected_equal, |
| 914 const Closure& callback) { |
| 915 scoped_refptr<SequencedTaskRunner> task_runner = |
| 916 SequencedWorkerPool::GetSequencedTaskRunnerForCurrentThread(); |
| 913 | 917 |
| 914 TEST_F(SequencedWorkerPoolTest, GetWorkerPoolAndSequenceTokenForCurrentThread) { | 918 EXPECT_TRUE(task_runner->RunsTasksOnCurrentThread()); |
| 919 |
| 920 // SequencedTaskRunner does not allow directly checking for equality, but we |
| 921 // can post a task to one task runner and verify that the other task runner |
| 922 // is on the same sequence. |
| 923 task_runner->PostTask( |
| 924 FROM_HERE, |
| 925 Bind(&VerifySequencedTaskRunnerRunsOnCurrentThread, |
| 926 base::Unretained(expected_task_runner), expected_equal, callback)); |
| 927 } |
| 928 |
| 929 void VerifyCurrentSequencedTaskRunnerForUnsequencedTask( |
| 930 SequencedWorkerPool* pool, |
| 931 const Closure& callback) { |
| 932 EXPECT_FALSE( |
| 933 SequencedWorkerPool::GetSequenceTokenForCurrentThread().IsValid()); |
| 934 |
| 935 scoped_refptr<SequencedTaskRunner> task_runner = |
| 936 SequencedWorkerPool::GetSequencedTaskRunnerForCurrentThread(); |
| 937 |
| 938 EXPECT_TRUE(task_runner->RunsTasksOnCurrentThread()); |
| 939 |
| 940 scoped_refptr<SequencedTaskRunner> expected_task_runner = |
| 941 SequencedWorkerPool::GetSequencedTaskRunnerForCurrentThread(); |
| 942 |
| 943 // The pool should now be running a sequence. This also verifies that no other |
| 944 // thread will start running tasks with this sequence token. |
| 945 const SequencedWorkerPool::SequenceToken sequence_token = |
| 946 SequencedWorkerPool::GetSequenceTokenForCurrentThread(); |
| 947 ASSERT_TRUE(sequence_token.IsValid()); |
| 948 EXPECT_TRUE(pool->IsRunningSequence(sequence_token)); |
| 949 |
| 950 // The two sequenced task runners should be the same. See |
| 951 // VerifyCurrentSequencedTaskRunner() above for why the check is implemented |
| 952 // this way. |
| 953 const bool expected_equal = true; |
| 954 task_runner->PostTask(FROM_HERE, |
| 955 Bind(&VerifySequencedTaskRunnerRunsOnCurrentThread, |
| 956 RetainedRef(std::move(expected_task_runner)), |
| 957 expected_equal, callback)); |
| 958 } |
| 959 |
| 960 TEST_F(SequencedWorkerPoolTest, GetSequencedTaskRunnerForCurrentThread) { |
| 915 EnsureAllWorkersCreated(); | 961 EnsureAllWorkersCreated(); |
| 916 | 962 |
| 917 // The current thread should have neither a worker pool nor a sequence token. | 963 // The current thread should not have a sequenced task runner from a |
| 918 SequencedWorkerPool::SequenceToken local_token = | 964 // worker pool. |
| 919 SequencedWorkerPool::GetSequenceTokenForCurrentThread(); | 965 scoped_refptr<SequencedTaskRunner> local_task_runner = |
| 920 scoped_refptr<SequencedWorkerPool> local_pool = | 966 SequencedWorkerPool::GetSequencedTaskRunnerForCurrentThread(); |
| 921 SequencedWorkerPool::GetWorkerPoolForCurrentThread(); | 967 EXPECT_FALSE(local_task_runner); |
| 922 EXPECT_FALSE(local_token.IsValid()) << local_token.ToString(); | |
| 923 EXPECT_FALSE(local_pool); | |
| 924 | 968 |
| 925 SequencedWorkerPool::SequenceToken token1 = pool()->GetSequenceToken(); | 969 WaitableEvent event(WaitableEvent::ResetPolicy::AUTOMATIC, |
| 926 SequencedWorkerPool::SequenceToken token2 = pool()->GetSequenceToken(); | 970 WaitableEvent::InitialState::NOT_SIGNALED); |
| 927 pool()->PostSequencedWorkerTask( | 971 Closure signal = Bind(&WaitableEvent::Signal, Unretained(&event)); |
| 928 token1, FROM_HERE, | 972 scoped_refptr<SequencedTaskRunner> task_runner_1 = |
| 929 base::Bind(&CheckWorkerPoolAndSequenceToken, pool(), token1)); | 973 pool()->GetSequencedTaskRunner(SequencedWorkerPool::GetSequenceToken()); |
| 930 pool()->PostSequencedWorkerTask( | 974 scoped_refptr<SequencedTaskRunner> task_runner_2 = |
| 931 token2, FROM_HERE, | 975 pool()->GetSequencedTaskRunner(SequencedWorkerPool::GetSequenceToken()); |
| 932 base::Bind(&CheckWorkerPoolAndSequenceToken, pool(), token2)); | 976 task_runner_1->PostTask( |
| 977 FROM_HERE, Bind(&VerifyCurrentSequencedTaskRunner, |
| 978 base::Unretained(task_runner_1.get()), true, signal)); |
| 979 event.Wait(); |
| 980 task_runner_2->PostTask( |
| 981 FROM_HERE, Bind(&VerifyCurrentSequencedTaskRunner, |
| 982 base::Unretained(task_runner_2.get()), true, signal)); |
| 983 event.Wait(); |
| 933 | 984 |
| 934 pool()->PostWorkerTask(FROM_HERE, | 985 task_runner_1->PostTask( |
| 935 base::Bind(&CheckWorkerPoolAndSequenceToken, pool(), | 986 FROM_HERE, Bind(&VerifyCurrentSequencedTaskRunner, |
| 936 SequencedWorkerPool::SequenceToken())); | 987 base::Unretained(task_runner_2.get()), false, signal)); |
| 988 event.Wait(); |
| 937 | 989 |
| 938 pool()->FlushForTesting(); | 990 pool()->PostWorkerTask( |
| 991 FROM_HERE, Bind(&VerifyCurrentSequencedTaskRunnerForUnsequencedTask, |
| 992 RetainedRef(pool()), signal)); |
| 993 event.Wait(); |
| 994 } |
| 995 |
| 996 class ChecksSequenceOnDestruction |
| 997 : public RefCountedThreadSafe<ChecksSequenceOnDestruction> { |
| 998 public: |
| 999 void DoNothing() {} |
| 1000 |
| 1001 private: |
| 1002 friend class RefCountedThreadSafe<ChecksSequenceOnDestruction>; |
| 1003 |
| 1004 ~ChecksSequenceOnDestruction() { |
| 1005 EXPECT_TRUE(sequence_checker_.CalledOnValidSequencedThread()); |
| 1006 } |
| 1007 |
| 1008 SequenceCheckerImpl sequence_checker_; |
| 1009 }; |
| 1010 |
| 1011 void VerifySequenceOnDestruction(const Closure& callback) { |
| 1012 scoped_refptr<SequencedTaskRunner> task_runner = |
| 1013 SequencedWorkerPool::GetSequencedTaskRunnerForCurrentThread(); |
| 1014 scoped_refptr<ChecksSequenceOnDestruction> check_sequence( |
| 1015 new ChecksSequenceOnDestruction); |
| 1016 |
| 1017 // Post a task to an empty method. This will keep the only reference to the |
| 1018 // object, so it will be destroyed right after running the task. |
| 1019 task_runner->PostTask(FROM_HERE, Bind(&ChecksSequenceOnDestruction::DoNothing, |
| 1020 std::move(check_sequence))); |
| 1021 |
| 1022 // Post the callback afterwards, so we can be sure the first task completed. |
| 1023 task_runner->PostTask(FROM_HERE, callback); |
| 1024 } |
| 1025 |
| 1026 TEST_F(SequencedWorkerPoolTest, CheckSequenceOnDestruction) { |
| 1027 EnsureAllWorkersCreated(); |
| 1028 |
| 1029 WaitableEvent event(WaitableEvent::ResetPolicy::AUTOMATIC, |
| 1030 WaitableEvent::InitialState::NOT_SIGNALED); |
| 1031 Closure signal = Bind(&WaitableEvent::Signal, Unretained(&event)); |
| 1032 pool()->PostWorkerTask(FROM_HERE, Bind(&VerifySequenceOnDestruction, signal)); |
| 1033 event.Wait(); |
| 939 } | 1034 } |
| 940 | 1035 |
| 941 TEST_F(SequencedWorkerPoolTest, ShutsDownCleanWithContinueOnShutdown) { | 1036 TEST_F(SequencedWorkerPoolTest, ShutsDownCleanWithContinueOnShutdown) { |
| 942 scoped_refptr<SequencedTaskRunner> task_runner = | 1037 scoped_refptr<SequencedTaskRunner> task_runner = |
| 943 pool()->GetSequencedTaskRunnerWithShutdownBehavior( | 1038 pool()->GetSequencedTaskRunnerWithShutdownBehavior( |
| 944 pool()->GetSequenceToken(), | 1039 pool()->GetSequenceToken(), |
| 945 base::SequencedWorkerPool::CONTINUE_ON_SHUTDOWN); | 1040 base::SequencedWorkerPool::CONTINUE_ON_SHUTDOWN); |
| 946 | 1041 |
| 947 // Upon test exit, should shut down without hanging. | 1042 // Upon test exit, should shut down without hanging. |
| 948 pool()->Shutdown(); | 1043 pool()->Shutdown(); |
| (...skipping 118 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
| 1067 SequencedWorkerPoolSequencedTaskRunner, SequencedTaskRunnerTest, | 1162 SequencedWorkerPoolSequencedTaskRunner, SequencedTaskRunnerTest, |
| 1068 SequencedWorkerPoolSequencedTaskRunnerTestDelegate); | 1163 SequencedWorkerPoolSequencedTaskRunnerTestDelegate); |
| 1069 INSTANTIATE_TYPED_TEST_CASE_P( | 1164 INSTANTIATE_TYPED_TEST_CASE_P( |
| 1070 SequencedWorkerPoolSequencedTaskRunner, | 1165 SequencedWorkerPoolSequencedTaskRunner, |
| 1071 SequencedTaskRunnerDelayedTest, | 1166 SequencedTaskRunnerDelayedTest, |
| 1072 SequencedWorkerPoolSequencedTaskRunnerTestDelegate); | 1167 SequencedWorkerPoolSequencedTaskRunnerTestDelegate); |
| 1073 | 1168 |
| 1074 } // namespace | 1169 } // namespace |
| 1075 | 1170 |
| 1076 } // namespace base | 1171 } // namespace base |
| OLD | NEW |