From ec5dbe56f1fcf1c3204f1deecfff88acf51bce22 Mon Sep 17 00:00:00 2001 From: Francis Tsui Date: Thu, 14 Nov 2024 11:54:03 -0800 Subject: [PATCH] Add worker queue class. PiperOrigin-RevId: 696602320 --- sharing/BUILD | 25 +++++++ sharing/worker_queue.h | 133 +++++++++++++++++++++++++++++++++++ sharing/worker_queue_test.cc | 114 ++++++++++++++++++++++++++++++ 3 files changed, 272 insertions(+) create mode 100644 sharing/worker_queue.h create mode 100644 sharing/worker_queue_test.cc diff --git a/sharing/BUILD b/sharing/BUILD index 32e884fa..be7614c3 100644 --- a/sharing/BUILD +++ b/sharing/BUILD @@ -132,6 +132,18 @@ cc_library( ], ) +cc_library( + name = "worker_queue", + hdrs = ["worker_queue.h"], + deps = [ + "//internal/platform:types", + "//sharing/internal/public:logging", + "@com_google_absl//absl/base:core_headers", + "@com_google_absl//absl/functional:any_invocable", + "@com_google_absl//absl/synchronization", + ], +) + cc_library( name = "incoming_frame_reader", srcs = ["incoming_frames_reader.cc"], @@ -863,3 +875,16 @@ cc_test( "@com_google_googletest//:gtest_main", ], ) + +cc_test( + name = "worker_queue_test", + srcs = ["worker_queue_test.cc"], + deps = [ + ":worker_queue", + "//internal/platform/implementation/g3", # fixdeps: keep + "//internal/test", + "@com_github_protobuf_matchers//protobuf-matchers", + "@com_google_absl//absl/synchronization", + "@com_google_googletest//:gtest_main", + ], +) diff --git a/sharing/worker_queue.h b/sharing/worker_queue.h new file mode 100644 index 00000000..1632d978 --- /dev/null +++ b/sharing/worker_queue.h @@ -0,0 +1,133 @@ +// Copyright 2024 Google LLC +// +// Licensed 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 +// +// https://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. + +#ifndef THIRD_PARTY_NEARBY_SHARING_WORKER_QUEUE_H_ +#define THIRD_PARTY_NEARBY_SHARING_WORKER_QUEUE_H_ + +#include +#include +#include + +#include "absl/base/thread_annotations.h" +#include "absl/functional/any_invocable.h" +#include "absl/synchronization/mutex.h" +#include "internal/platform/task_runner.h" +#include "sharing/internal/public/logging.h" + +namespace nearby::sharing { + +// A queue that runs a callback on a worker thread to handle queued items. +// +// The callback is edge triggered, i.e. it will only be scheduled when the queue +// changes from empty to non-empty. This object guarantees that only 1 callback +// is scheduled at a time until the `ReadAll` method is called which resets the +// scheduling state. +// +// This class is thread-safe. +template +class WorkerQueue { + public: + explicit WorkerQueue(TaskRunner* task_runner) : task_runner_(task_runner) {} + + ~WorkerQueue() { Stop(); } + + // Starts the queue. `callback` will be called on the worker thread when + // there are items in the queue. + // Returns true if the queue is started successfully. + // Returns false if the queue is already started or stopped. Queue cannot be + // restarted. + bool Start(absl::AnyInvocable callback) { + if (is_started_.exchange(true)) { + LOG(ERROR) << "WorkerQueue is already started."; + return false; + } + if (is_stopped_) { + LOG(ERROR) << "WorkerQueue is already stopped, cannot restart."; + return false; + } + callback_ = std::move(callback); + { + absl::MutexLock lock(&mutex_); + if (!queue_.empty()) { + ScheduleCallback(); + } + } + return true; + } + + // Stops the queue. No new callback will be scheduled. + void Stop() { + bool already_stopped = is_stopped_.exchange(true); + if (already_stopped || !is_started_) { + return; + } + is_scheduled_ = true; + } + + // Queues an item to be processed by the callback. + // Callback are edge triggered. + void Queue(T item) { + absl::MutexLock lock(&mutex_); + queue_.push(std::move(item)); + ScheduleCallback(); + } + + // Returns all the items in the queue and clears the queue. + // This resets the callback scheduling state and new callbacks will be + // scheduled when new items are queued. + std::queue ReadAll() { + is_scheduled_ = false; + absl::MutexLock lock(&mutex_); + std::queue queue; + queue.swap(queue_); + return queue; + } + + private: + void ScheduleCallback() { + // Skip if not started or stopped + if (!is_started_ || is_stopped_) { + return; + } + if (is_scheduled_.exchange(true)) { + LOG(ERROR) << "Already scheduled"; + // Already scheduled. + return; + } + LOG(ERROR) << "Scheduling callback"; + task_runner_->PostTask([this]() { + if (is_stopped_) { + return; + } + callback_(); + }); + } + + TaskRunner* const task_runner_ = nullptr; + absl::AnyInvocable callback_; + // Tracks whether Start() has been called. + std::atomic is_started_ = false; + // Tracks whether Stop() has been called. + std::atomic is_stopped_ = false; + absl::Mutex mutex_; + std::queue queue_ ABSL_GUARDED_BY(mutex_); + // This is used track whether the callback is already scheduled so as to avoid + // scheduling multiple callbacks. + // `is_scheduled_` must be false if is_started_ is false. + std::atomic is_scheduled_ = false; +}; + +} // namespace nearby::sharing + +#endif // THIRD_PARTY_NEARBY_SHARING_WORKER_QUEUE_H_ diff --git a/sharing/worker_queue_test.cc b/sharing/worker_queue_test.cc new file mode 100644 index 00000000..e9b6e3b0 --- /dev/null +++ b/sharing/worker_queue_test.cc @@ -0,0 +1,114 @@ +// Copyright 2024 Google LLC +// +// Licensed 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 +// +// https://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 "sharing/worker_queue.h" + +#include + +#include "gtest/gtest.h" +#include "absl/synchronization/notification.h" +#include "internal/test/fake_clock.h" +#include "internal/test/fake_task_runner.h" + +namespace nearby::sharing { +namespace { + +TEST(WorkerQueueTest, SecondStartReturnsFalse) { + FakeClock fake_clock; + FakeTaskRunner task_runner(&fake_clock, 1); + WorkerQueue queue(&task_runner); + EXPECT_TRUE(queue.Start([]() {})); + EXPECT_FALSE(queue.Start([]() {})); +} + +TEST(WorkerQueueTest, StartAfterStopReturnsFalse) { + FakeClock fake_clock; + FakeTaskRunner task_runner(&fake_clock, 1); + WorkerQueue queue(&task_runner); + EXPECT_TRUE(queue.Start([]() {})); + queue.Stop(); + EXPECT_FALSE(queue.Start([]() {})); +} + +TEST(WorkerQueueTest, QueueItems) { + FakeClock fake_clock; + FakeTaskRunner task_runner(&fake_clock, 1); + WorkerQueue queue(&task_runner); + absl::Notification notification; + // Block the task runner thread. + task_runner.PostTask( + [¬ification]() { notification.WaitForNotification(); }); + EXPECT_TRUE(queue.Start([&queue]() { + std::queue items = queue.ReadAll(); + EXPECT_EQ(items.size(), 2); + EXPECT_EQ(items.front(), 1); + EXPECT_EQ(items.back(), 2); + })); + queue.Queue(1); + queue.Queue(2); + notification.Notify(); + + task_runner.Sync(); + // No more callbacks. +} + +TEST(WorkerQueueTest, QueueItemsWhileCallbackRunning) { + FakeClock fake_clock; + FakeTaskRunner task_runner(&fake_clock, 1); + WorkerQueue queue(&task_runner); + absl::Notification notification1; + absl::Notification notification2; + EXPECT_TRUE(queue.Start([&queue, ¬ification1, ¬ification2]() { + notification1.Notify(); + // Block the task runner thread. + notification2.WaitForNotification(); + std::queue items = queue.ReadAll(); + EXPECT_EQ(items.size(), 2); + EXPECT_EQ(items.front(), 1); + EXPECT_EQ(items.back(), 2); + })); + queue.Queue(1); + // Wait for the callback to start. + notification1.WaitForNotification(); + queue.Queue(2); + notification2.Notify(); + + task_runner.Sync(); + // No more callbacks. +} + +TEST(WorkerQueueTest, StopStopsCallback) { + FakeClock fake_clock; + FakeTaskRunner task_runner(&fake_clock, 1); + WorkerQueue queue(&task_runner); + queue.Queue(1); + queue.Queue(2); + absl::Notification notification; + EXPECT_TRUE(queue.Start([&queue, ¬ification]() { + std::queue items = queue.ReadAll(); + EXPECT_EQ(items.size(), 2); + EXPECT_EQ(items.front(), 1); + EXPECT_EQ(items.back(), 2); + notification.Notify(); + })); + // Wait for the callback to start. + notification.WaitForNotification(); + queue.Stop(); + queue.Queue(3); + task_runner.Sync(); + // No more callbacks. +} + +} // namespace +} // namespace nearby::sharing