Remove OperationLoop
This change removes OperationLoop because it's only used in one place
and is arguably confusing. OperationLoop mixes sleep() and select()
usage, with separate timeouts, which has resulted in some "tuning" bugs
in the past. Since there's no benefit to the sleep portion and it is
also what causes the confusion, this change removes it and thing
directly uses SocketHandleWaiter in PlatformClientPosix.
Bug: None
Change-Id: I9f62c9253bee7d29b26d1838d34c1399f4d472b5
Reviewed-on: https://chromium-review.googlesource.com/c/openscreen/+/2798852
Commit-Queue: Jordan Bayles <jophba@chromium.org>
Reviewed-by: Jordan Bayles <jophba@chromium.org>
diff --git a/cast/common/discovery/e2e_test/tests.cc b/cast/common/discovery/e2e_test/tests.cc
index 9a02053..7c29441 100644
--- a/cast/common/discovery/e2e_test/tests.cc
+++ b/cast/common/discovery/e2e_test/tests.cc
@@ -143,7 +143,7 @@
// Sleep to let any packets clear off the network before further tests.
std::this_thread::sleep_for(milliseconds(500));
- PlatformClientPosix::Create(milliseconds(50), milliseconds(50));
+ PlatformClientPosix::Create(milliseconds(50));
task_runner_ = PlatformClientPosix::GetInstance()->GetTaskRunner();
}
diff --git a/cast/standalone_receiver/main.cc b/cast/standalone_receiver/main.cc
index 9faee41..9e305c8 100644
--- a/cast/standalone_receiver/main.cc
+++ b/cast/standalone_receiver/main.cc
@@ -230,9 +230,7 @@
}
auto* const task_runner = new TaskRunnerImpl(&Clock::now);
- // Cast has high networking demands--network operation timing and timeout must
- // be kept extremely small.
- PlatformClientPosix::Create(microseconds(50), microseconds(50),
+ PlatformClientPosix::Create(milliseconds(50),
std::unique_ptr<TaskRunnerImpl>(task_runner));
RunCastService(task_runner, interface, std::move(creds.value()),
friendly_name, model_name, discovery_enabled);
diff --git a/cast/standalone_sender/main.cc b/cast/standalone_sender/main.cc
index 7479adc..75b5055 100644
--- a/cast/standalone_sender/main.cc
+++ b/cast/standalone_sender/main.cc
@@ -174,9 +174,7 @@
#endif
auto* const task_runner = new TaskRunnerImpl(&Clock::now);
- // Cast has high networking demands--network operation timing and timeout must
- // be kept extremely small.
- PlatformClientPosix::Create(microseconds(50), microseconds(50),
+ PlatformClientPosix::Create(milliseconds(50),
std::unique_ptr<TaskRunnerImpl>(task_runner));
IPEndpoint remote_endpoint = ParseAsEndpoint(iface_or_endpoint);
diff --git a/cast/test/cast_socket_e2e_test.cc b/cast/test/cast_socket_e2e_test.cc
index 1f3273f..446a880 100644
--- a/cast/test/cast_socket_e2e_test.cc
+++ b/cast/test/cast_socket_e2e_test.cc
@@ -136,8 +136,7 @@
class CastSocketE2ETest : public ::testing::Test {
public:
void SetUp() override {
- PlatformClientPosix::Create(std::chrono::milliseconds(10),
- std::chrono::milliseconds(0));
+ PlatformClientPosix::Create(std::chrono::milliseconds(10));
task_runner_ = PlatformClientPosix::GetInstance()->GetTaskRunner();
sender_router_ = MakeSerialDelete<VirtualConnectionRouter>(task_runner_);
diff --git a/osp/demo/osp_demo.cc b/osp/demo/osp_demo.cc
index 443536f..952a925 100644
--- a/osp/demo/osp_demo.cc
+++ b/osp/demo/osp_demo.cc
@@ -644,8 +644,7 @@
// TODO(jophba): Mac on Mojave hangs on this command forever.
openscreen::SetLogFifoOrDie(log_filename);
- PlatformClientPosix::Create(std::chrono::milliseconds(50),
- std::chrono::milliseconds(50));
+ PlatformClientPosix::Create(std::chrono::milliseconds(50));
if (is_receiver_demo) {
OSP_LOG_INFO << "Running publisher demo...";
diff --git a/osp/impl/discovery/mdns/mdns_demo.cc b/osp/impl/discovery/mdns/mdns_demo.cc
index 6286099..1fc1513 100644
--- a/osp/impl/discovery/mdns/mdns_demo.cc
+++ b/osp/impl/discovery/mdns/mdns_demo.cc
@@ -361,8 +361,7 @@
openscreen::osp::ServiceMap services;
openscreen::osp::g_services = &services;
- PlatformClientPosix::Create(std::chrono::milliseconds(50),
- std::chrono::milliseconds(50));
+ PlatformClientPosix::Create(std::chrono::milliseconds(50));
openscreen::osp::BrowseDemo(
PlatformClientPosix::GetInstance()->GetTaskRunner(), labels[0], labels[1],
diff --git a/platform/impl/platform_client_posix.cc b/platform/impl/platform_client_posix.cc
index d141525..328b8bf 100644
--- a/platform/impl/platform_client_posix.cc
+++ b/platform/impl/platform_client_posix.cc
@@ -5,6 +5,7 @@
#include "platform/impl/platform_client_posix.h"
#include <functional>
+#include <utility>
#include <vector>
#include "platform/impl/udp_socket_reader_posix.h"
@@ -16,18 +17,14 @@
// static
void PlatformClientPosix::Create(Clock::duration networking_operation_timeout,
- Clock::duration networking_loop_interval,
std::unique_ptr<TaskRunnerImpl> task_runner) {
SetInstance(new PlatformClientPosix(networking_operation_timeout,
- networking_loop_interval,
std::move(task_runner)));
}
// static
-void PlatformClientPosix::Create(Clock::duration networking_operation_timeout,
- Clock::duration networking_loop_interval) {
- SetInstance(new PlatformClientPosix(networking_operation_timeout,
- networking_loop_interval));
+void PlatformClientPosix::Create(Clock::duration networking_operation_timeout) {
+ SetInstance(new PlatformClientPosix(networking_operation_timeout));
}
// static
@@ -67,7 +64,7 @@
}
OSP_DVLOG << "Shutting down network operations...";
- networking_loop_.RequestStopSoon();
+ networking_loop_running_.store(false);
networking_loop_thread_.join();
OSP_DVLOG << "\tNetwork operation shutdown complete!";
}
@@ -79,27 +76,21 @@
}
PlatformClientPosix::PlatformClientPosix(
- Clock::duration networking_operation_timeout,
- Clock::duration networking_loop_interval)
- : networking_loop_(networking_operations(),
- networking_operation_timeout,
- networking_loop_interval),
- task_runner_(new TaskRunnerImpl(Clock::now)),
- networking_loop_thread_(&OperationLoop::RunUntilStopped,
- &networking_loop_),
+ Clock::duration networking_operation_timeout)
+ : task_runner_(new TaskRunnerImpl(Clock::now)),
+ networking_loop_timeout_(networking_operation_timeout),
+ networking_loop_thread_(&PlatformClientPosix::RunNetworkLoopUntilStopped,
+ this),
task_runner_thread_(
std::thread(&TaskRunnerImpl::RunUntilStopped, task_runner_.get())) {}
PlatformClientPosix::PlatformClientPosix(
Clock::duration networking_operation_timeout,
- Clock::duration networking_loop_interval,
std::unique_ptr<TaskRunnerImpl> task_runner)
- : networking_loop_(networking_operations(),
- networking_operation_timeout,
- networking_loop_interval),
- task_runner_(std::move(task_runner)),
- networking_loop_thread_(&OperationLoop::RunUntilStopped,
- &networking_loop_) {}
+ : task_runner_(std::move(task_runner)),
+ networking_loop_timeout_(networking_operation_timeout),
+ networking_loop_thread_(&PlatformClientPosix::RunNetworkLoopUntilStopped,
+ this) {}
SocketHandleWaiterPosix* PlatformClientPosix::socket_handle_waiter() {
std::call_once(waiter_initialization_, [this]() {
@@ -109,20 +100,14 @@
return waiter_.get();
}
-void PlatformClientPosix::PerformSocketHandleWaiterActions(
- Clock::duration timeout) {
- if (!waiter_created_.load()) {
- return;
+void PlatformClientPosix::RunNetworkLoopUntilStopped() {
+ while (networking_loop_running_.load()) {
+ if (!waiter_created_.load()) {
+ std::this_thread::sleep_for(networking_loop_timeout_);
+ continue;
+ }
+ socket_handle_waiter()->ProcessHandles(networking_loop_timeout_);
}
-
- socket_handle_waiter()->ProcessHandles(timeout);
-}
-
-std::vector<std::function<void(Clock::duration)>>
-PlatformClientPosix::networking_operations() {
- return {[this](Clock::duration timeout) {
- PerformSocketHandleWaiterActions(timeout);
- }};
}
} // namespace openscreen
diff --git a/platform/impl/platform_client_posix.h b/platform/impl/platform_client_posix.h
index 9f086f5..bb320f5 100644
--- a/platform/impl/platform_client_posix.h
+++ b/platform/impl/platform_client_posix.h
@@ -17,7 +17,6 @@
#include "platform/impl/socket_handle_waiter_posix.h"
#include "platform/impl/task_runner.h"
#include "platform/impl/tls_data_router_posix.h"
-#include "util/operation_loop.h"
namespace openscreen {
@@ -50,13 +49,11 @@
//
// |task_runner| is a client-provided TaskRunner implementation.
static void Create(Clock::duration networking_operation_timeout,
- Clock::duration networking_loop_interval,
std::unique_ptr<TaskRunnerImpl> task_runner);
// Initializes the platform implementation and creates a new TaskRunner (which
// starts a new thread).
- static void Create(Clock::duration networking_operation_timeout,
- Clock::duration networking_loop_interval);
+ static void Create(Clock::duration networking_operation_timeout);
// Shuts down and deletes the PlatformClient instance currently stored as a
// singleton. This method is expected to be called before program exit. After
@@ -85,25 +82,15 @@
static void SetInstance(PlatformClientPosix* client);
private:
- PlatformClientPosix(Clock::duration networking_operation_timeout,
- Clock::duration networking_loop_interval);
+ explicit PlatformClientPosix(Clock::duration networking_operation_timeout);
PlatformClientPosix(Clock::duration networking_operation_timeout,
- Clock::duration networking_loop_interval,
std::unique_ptr<TaskRunnerImpl> task_runner);
// This method is thread-safe.
SocketHandleWaiterPosix* socket_handle_waiter();
- // Helper functions to use when creating and calling the OperationLoop used
- // for the networking thread.
- void PerformSocketHandleWaiterActions(Clock::duration timeout);
- void PerformTlsDataRouterActions(Clock::duration timeout);
- std::vector<std::function<void(Clock::duration)>> networking_operations();
-
- // Instance objects with threads are created at object-creation time.
- // NOTE: Delayed instantiation of networking_loop_ may be useful in future.
- OperationLoop networking_loop_;
+ void RunNetworkLoopUntilStopped();
std::unique_ptr<TaskRunnerImpl> task_runner_;
@@ -111,6 +98,10 @@
std::atomic_bool waiter_created_{false};
std::atomic_bool tls_data_router_created_{false};
+ // Parameters for networking loop.
+ std::atomic_bool networking_loop_running_{true};
+ Clock::duration networking_loop_timeout_;
+
// Flags used to ensure that initialization of below instance objects occurs
// only once across all threads.
std::once_flag waiter_initialization_;
diff --git a/util/BUILD.gn b/util/BUILD.gn
index d379fde..3f97e09 100644
--- a/util/BUILD.gn
+++ b/util/BUILD.gn
@@ -51,8 +51,6 @@
"json/json_serialization.h",
"json/json_value.cc",
"json/json_value.h",
- "operation_loop.cc",
- "operation_loop.h",
"osp_logging.h",
"saturate_cast.h",
"simple_fraction.cc",
@@ -112,7 +110,6 @@
"json/json_helpers_unittest.cc",
"json/json_serialization_unittest.cc",
"json/json_value_unittest.cc",
- "operation_loop_unittest.cc",
"saturate_cast_unittest.cc",
"simple_fraction_unittest.cc",
"stringprintf_unittest.cc",
diff --git a/util/operation_loop.cc b/util/operation_loop.cc
deleted file mode 100644
index a20f144..0000000
--- a/util/operation_loop.cc
+++ /dev/null
@@ -1,57 +0,0 @@
-// Copyright 2019 The Chromium Authors. All rights reserved.
-// Use of this source code is governed by a BSD-style license that can be
-// found in the LICENSE file.
-#include "util/operation_loop.h"
-
-#include <algorithm>
-
-#include "util/osp_logging.h"
-
-namespace openscreen {
-
-OperationLoop::OperationLoop(std::vector<OperationWithTimeout> operations,
- Clock::duration timeout,
- Clock::duration min_loop_execution_time)
- : perform_all_operations_min_execution_time_(min_loop_execution_time),
- operation_timeout_(timeout),
- operations_(operations) {
- OSP_DCHECK(operations_.size());
-}
-
-void OperationLoop::RunUntilStopped() {
- OSP_CHECK(!is_running_.exchange(true));
-
- while (is_running_.load()) {
- PerformAllOperations();
- }
-}
-
-void OperationLoop::RequestStopSoon() {
- {
- std::unique_lock<std::mutex> lock(wait_mutex_);
- is_running_.store(false);
- }
-
- perform_all_operations_waiter_.notify_all();
-}
-
-void OperationLoop::PerformAllOperations() {
- auto start_time = Clock::now();
-
- for (OperationWithTimeout operation : operations_) {
- if (is_running_.load()) {
- operation(operation_timeout_);
- }
- }
-
- const auto duration = Clock::now() - start_time;
- const auto remaining_duration =
- perform_all_operations_min_execution_time_ - duration;
- if (remaining_duration > Clock::duration{0} && is_running_.load()) {
- std::unique_lock<std::mutex> lock(wait_mutex_);
- perform_all_operations_waiter_.wait_for(
- lock, remaining_duration, [this]() { return !is_running_.load(); });
- }
-}
-
-} // namespace openscreen
diff --git a/util/operation_loop.h b/util/operation_loop.h
deleted file mode 100644
index ddb7846..0000000
--- a/util/operation_loop.h
+++ /dev/null
@@ -1,74 +0,0 @@
-// Copyright 2019 The Chromium Authors. All rights reserved.
-// Use of this source code is governed by a BSD-style license that can be
-// found in the LICENSE file.
-
-#ifndef UTIL_OPERATION_LOOP_H_
-#define UTIL_OPERATION_LOOP_H_
-
-#include <atomic>
-#include <condition_variable>
-#include <functional>
-#include <vector>
-
-#include "platform/api/time.h"
-#include "platform/base/macros.h"
-
-namespace openscreen {
-
-class OperationLoop {
- public:
- using OperationWithTimeout = std::function<void(Clock::duration)>;
-
- // Creates a new OperationLoop from a variable number of operations. The
- // provided functions will be called repeatedly, at a minimum interval equal
- // to min_loop_execution_time, and are expected to exit after the time period
- // provided to their call has passed. This is because some operations may not
- // be safe to be interrupted from this class.
- // NOTE: If n operations are provided with operation timeout T, each iteration
- // of the operation loop may take as long as n * T, and will not exit after
- // min_loop_execution_time has elapsed. In order to avoid this behavior, the
- // caller can set min_loop_execution_time = n * T.
- //
- // operations = Functions to execute repeatedly. All functions are expected to
- // be valid the duration of this object's lifetime.
- // timeout = Timeout for each individual function above.
- // min_loop_execution_time = Minimum time that OperationLoop should wait
- // before successive calls to members of the
- // provided operations vector.
- OperationLoop(std::vector<OperationWithTimeout> operations,
- Clock::duration timeout,
- Clock::duration min_loop_execution_time);
-
- // Runs the PerformAllOperations function in a loop until the below
- // RequestStopSoon function is called.
- void RunUntilStopped();
-
- // Signals for the RunUntilStopped loop to cease running.
- void RequestStopSoon();
-
- OSP_DISALLOW_COPY_AND_ASSIGN(OperationLoop);
-
- private:
- // Performs all operations which have been provided to this instance.
- void PerformAllOperations();
-
- const Clock::duration perform_all_operations_min_execution_time_;
-
- const Clock::duration operation_timeout_;
-
- // Used to wait in PerformAllOperations() if not enough time has elapsed.
- std::condition_variable perform_all_operations_waiter_;
-
- // Mutex used by the above condition_variable.
- std::mutex wait_mutex_;
-
- // Represents whether this instance is currently "running".
- std::atomic_bool is_running_{false};
-
- // Operations currently being run by this object.
- const std::vector<OperationWithTimeout> operations_;
-};
-
-} // namespace openscreen
-
-#endif // UTIL_OPERATION_LOOP_H_
diff --git a/util/operation_loop_unittest.cc b/util/operation_loop_unittest.cc
deleted file mode 100644
index 4b4ea5d..0000000
--- a/util/operation_loop_unittest.cc
+++ /dev/null
@@ -1,42 +0,0 @@
-// Copyright 2019 The Chromium Authors. All rights reserved.
-// Use of this source code is governed by a BSD-style license that can be
-// found in the LICENSE file.
-
-#include "util/operation_loop.h"
-
-#include <atomic>
-#include <chrono>
-#include <thread>
-
-#include "gtest/gtest.h"
-
-namespace openscreen {
-
-TEST(OperationsLoopTest, PerformAllOperationsWaits) {
- constexpr Clock::duration kTimeout{0};
- constexpr Clock::duration kMinRuntime{500};
-
- const auto start_time = Clock::now();
- std::atomic<Clock::time_point> last_run{start_time};
- std::atomic<Clock::time_point> current_run{start_time};
- std::function<void(Clock::duration)> test_function =
- [last = &last_run, current = ¤t_run](Clock::duration timeout) {
- last->store(current->load());
- current->store(Clock::now());
- };
- OperationLoop loop({test_function}, kTimeout, kMinRuntime);
-
- std::thread run_loop([&loop]() { loop.RunUntilStopped(); });
-
- while (last_run.load() == start_time) {
- }
-
- loop.RequestStopSoon();
- run_loop.join();
-
- EXPECT_GE(
- std::chrono::nanoseconds(current_run.load() - last_run.load()).count(),
- std::chrono::nanoseconds(kMinRuntime).count());
-}
-
-} // namespace openscreen
