Refactor waiting for timeout in PhysicalSocketServer - Move timeout related logic into dedicated helper class, thus unifying that logic across multiple call sites. - Switch function to query current time from TimeMicros to SystemTime to stress this class doesn't support injected clock. - Use TimeDelta type instead of plain int to pass around timeout. - Delete unused function GetSocketRecvTimestamp Bug: webrtc:42223992 Change-Id: I1f483985c8764539f20c51846d35daa62f90b108 Reviewed-on: https://webrtc-review.googlesource.com/c/src/+/465702 Reviewed-by: Evan Shrubsole <eshr@webrtc.org> Commit-Queue: Danil Chapovalov <danilchap@webrtc.org> Cr-Commit-Position: refs/heads/main@{#47518}
diff --git a/rtc_base/BUILD.gn b/rtc_base/BUILD.gn index 976cf53..6df20cb 100644 --- a/rtc_base/BUILD.gn +++ b/rtc_base/BUILD.gn
@@ -1036,6 +1036,7 @@ "../api/units:timestamp", "synchronization:mutex", "system:rtc_export", + "system:system_time", "system:unused", "//third_party/abseil-cpp/absl/algorithm:container", "//third_party/abseil-cpp/absl/base:core_headers",
diff --git a/rtc_base/physical_socket_server.cc b/rtc_base/physical_socket_server.cc index c439d5f..9d9927e 100644 --- a/rtc_base/physical_socket_server.cc +++ b/rtc_base/physical_socket_server.cc
@@ -27,7 +27,6 @@ #include "rtc_base/async_dns_resolver.h" #include "rtc_base/checks.h" #include "rtc_base/deprecated/recursive_critical_section.h" -#include "rtc_base/event.h" #include "rtc_base/ip_address.h" #include "rtc_base/logging.h" #include "rtc_base/net_helpers.h" @@ -35,8 +34,8 @@ #include "rtc_base/socket.h" #include "rtc_base/socket_address.h" #include "rtc_base/synchronization/mutex.h" +#include "rtc_base/system/system_time.h" #include "rtc_base/thread_annotations.h" -#include "rtc_base/time_utils.h" #ifdef MEMORY_SANITIZER #include <sanitizer/msan_interface.h> @@ -52,7 +51,6 @@ #elif defined(WEBRTC_USE_POLL) #include <poll.h> #endif -#include <sys/ioctl.h> #include <sys/select.h> #include <unistd.h> #endif @@ -67,7 +65,6 @@ #if defined(WEBRTC_LINUX) #include <asm-generic/socket.h> -#include <linux/sockios.h> #include <sys/epoll.h> #endif @@ -82,24 +79,6 @@ typedef void* SockOptArg; #endif // WEBRTC_POSIX -#if defined(WEBRTC_POSIX) && !defined(WEBRTC_MAC) -int64_t GetSocketRecvTimestamp(int socket) { - struct timeval tv_ioctl; - int ret = ioctl(socket, SIOCGSTAMP, &tv_ioctl); - if (ret != 0) - return -1; - int64_t timestamp = - webrtc::kNumMicrosecsPerSec * static_cast<int64_t>(tv_ioctl.tv_sec) + - static_cast<int64_t>(tv_ioctl.tv_usec); - return timestamp; -} - -#else -int64_t GetSocketRecvTimestamp(int /* socket */) { - return -1; -} -#endif - #if defined(WEBRTC_WIN) typedef char* SockOptArg; #endif @@ -156,6 +135,72 @@ bool* value_; }; +// Tracks time until given 'timeout' pass. +// Converts timeout into platform-specfic time delta representation. +template <typename TimeType> +class DeadlineTracker { + public: + // Memorizes time to wait until and converts 'timeout' into platform-specific + // representation saving result into output variable 'wait_time'. + DeadlineTracker(TimeDelta timeout, TimeType& wait_time) + : deadline_(SetInitialTimeout(timeout, wait_time)) {} + + // Returns non-negative remaining time until `timeout`. + TimeType WaitTime() { + if (deadline_ == Timestamp::PlusInfinity()) { + return Infinite(); + } + if (deadline_ == Timestamp::MinusInfinity()) { + return Finite(TimeDelta::Zero()); + } + TimeDelta remaining = std::max(deadline_ - SystemTime(), TimeDelta::Zero()); + return Finite(remaining.RoundUpTo(TimeDelta::Millis(1))); + } + + private: + Timestamp SetInitialTimeout(TimeDelta timeout, TimeType& wait_time) { + if (timeout == TimeDelta::PlusInfinity()) { + wait_time = Infinite(); + return Timestamp::PlusInfinity(); + } + if (timeout <= TimeDelta::Zero()) { + wait_time = Finite(TimeDelta::Zero()); + return Timestamp::MinusInfinity(); + } + wait_time = Finite(timeout); + return SystemTime() + timeout; + } + +#if defined(WEBRTC_WIN) + DWORD Infinite() { return WSA_INFINITE; } + DWORD Finite(TimeDelta t) { return t.ms<DWORD>(); } +#else + TimeType Infinite() { + if constexpr (std::is_same_v<TimeType, int>) { + return -1; + } + if constexpr (std::is_same_v<TimeType, timeval*>) { + return nullptr; + } + } + + TimeType Finite(TimeDelta t) { + if constexpr (std::is_same_v<TimeType, int>) { + return t.ms<int>(); + } + if constexpr (std::is_same_v<TimeType, timeval*>) { + wait_time_.tv_sec = t.us() / TimeDelta::Seconds(1).us(); + wait_time_.tv_usec = t.us() % TimeDelta::Seconds(1).us(); + return &wait_time_; + } + } + + timeval wait_time_; +#endif + + const Timestamp deadline_; +}; + } // namespace @@ -565,8 +610,9 @@ if (timestamp && cmsg->cmsg_type == SCM_TIMESTAMP) { timeval ts; std::memcpy(static_cast<void*>(&ts), CMSG_DATA(cmsg), sizeof(ts)); - *timestamp = kNumMicrosecsPerSec * static_cast<int64_t>(ts.tv_sec) + - static_cast<int64_t>(ts.tv_usec); + *timestamp = + (Timestamp::Seconds(ts.tv_sec) + TimeDelta::Micros(ts.tv_usec)) + .us(); } } } @@ -1376,34 +1422,27 @@ #endif } -int PhysicalSocketServer::ToCmsWait(TimeDelta max_wait_duration) { - return max_wait_duration == Event::kForever - ? kForeverMs - : max_wait_duration.RoundUpTo(TimeDelta::Millis(1)).ms(); -} - #if defined(WEBRTC_POSIX) bool PhysicalSocketServer::Wait(TimeDelta max_wait_duration, bool process_io) { // We don't support reentrant waiting. RTC_DCHECK(!waiting_); ScopedSetTrue s(&waiting_); - const int cmsWait = ToCmsWait(max_wait_duration); #if defined(WEBRTC_USE_POLL) - return WaitPoll(cmsWait, process_io); + return WaitPoll(max_wait_duration, process_io); #else #if defined(WEBRTC_USE_EPOLL) // We don't keep a dedicated "epoll" descriptor containing only the non-IO // (i.e. signaling) dispatcher, so "poll" will be used instead of the default // "select" to support sockets larger than FD_SETSIZE. if (!process_io) { - return WaitPollOneDispatcher(cmsWait, signal_wakeup_); + return WaitPollOneDispatcher(max_wait_duration, signal_wakeup_); } else if (epoll_fd_ != INVALID_SOCKET) { - return WaitEpoll(cmsWait); + return WaitEpoll(max_wait_duration); } #endif - return WaitSelect(cmsWait, process_io); + return WaitSelect(max_wait_duration, process_io); #endif } @@ -1510,21 +1549,10 @@ } #endif // WEBRTC_USE_POLL || WEBRTC_USE_EPOLL -bool PhysicalSocketServer::WaitSelect(int cmsWait, bool process_io) { +bool PhysicalSocketServer::WaitSelect(TimeDelta timeout, bool process_io) { // Calculate timing information - - struct timeval* ptvWait = nullptr; - struct timeval tvWait; - int64_t stop_us; - if (cmsWait != kForeverMs) { - // Calculate wait timeval - tvWait.tv_sec = cmsWait / 1000; - tvWait.tv_usec = (cmsWait % 1000) * 1000; - ptvWait = &tvWait; - - // Calculate when to return - stop_us = TimeMicros() + cmsWait * 1000; - } + timeval* wait_time = nullptr; + DeadlineTracker<timeval*> deadline(timeout, wait_time); fd_set fdsRead; fd_set fdsWrite; @@ -1573,7 +1601,7 @@ // < 0 means error // 0 means timeout // > 0 means count of descriptors ready - int n = select(fdmax + 1, &fdsRead, &fdsWrite, nullptr, ptvWait); + int n = select(fdmax + 1, &fdsRead, &fdsWrite, nullptr, wait_time); // If error, return error. if (n < 0) { @@ -1619,15 +1647,7 @@ // Recalc the time remaining to wait. Doing it here means it doesn't get // calced twice the first time through the loop - if (ptvWait) { - ptvWait->tv_sec = 0; - ptvWait->tv_usec = 0; - int64_t time_left_us = stop_us - TimeMicros(); - if (time_left_us > 0) { - ptvWait->tv_sec = time_left_us / kNumMicrosecsPerSec; - ptvWait->tv_usec = time_left_us % kNumMicrosecsPerSec; - } - } + wait_time = deadline.WaitTime(); } return true; @@ -1708,14 +1728,10 @@ } } -bool PhysicalSocketServer::WaitEpoll(int cmsWait) { +bool PhysicalSocketServer::WaitEpoll(TimeDelta timeout) { RTC_DCHECK(epoll_fd_ != INVALID_SOCKET); - int64_t msWait = -1; - int64_t msStop = -1; - if (cmsWait != kForeverMs) { - msWait = cmsWait; - msStop = TimeAfter(cmsWait); - } + int wait_time_ms = -1; + DeadlineTracker<int> deadline(timeout, wait_time_ms); fWait_ = true; while (fWait_) { @@ -1724,7 +1740,7 @@ // 0 means timeout // > 0 means count of descriptors ready int n = epoll_wait(epoll_fd_, epoll_events_.data(), epoll_events_.size(), - static_cast<int>(msWait)); + wait_time_ms); if (n < 0) { if (errno != EINTR) { RTC_LOG_E(LS_ERROR, EN, errno) << "epoll"; @@ -1757,27 +1773,22 @@ } } - if (cmsWait != kForeverMs) { - msWait = TimeDiff(msStop, TimeMillis()); - if (msWait <= 0) { - // Return success on timeout. - return true; - } + wait_time_ms = deadline.WaitTime(); + if (wait_time_ms == 0) { + // Return success on timeout. + return true; } } return true; } -bool PhysicalSocketServer::WaitPollOneDispatcher(int cmsWait, +bool PhysicalSocketServer::WaitPollOneDispatcher(TimeDelta timeout, Dispatcher* dispatcher) { RTC_DCHECK(dispatcher); - int64_t msWait = -1; - int64_t msStop = -1; - if (cmsWait != kForeverMs) { - msWait = cmsWait; - msStop = TimeAfter(cmsWait); - } + + int wait_time_ms = -1; + DeadlineTracker<int> deadline(timeout, wait_time_ms); fWait_ = true; const int fd = dispatcher->GetDescriptor(); @@ -1789,7 +1800,7 @@ // < 0 means error // 0 means timeout // > 0 means count of descriptors ready - int n = poll(&fds, 1, static_cast<int>(msWait)); + int n = poll(&fds, 1, wait_time_ms); if (n < 0) { if (errno != EINTR) { RTC_LOG_E(LS_ERROR, EN, errno) << "poll"; @@ -1809,12 +1820,10 @@ ProcessPollEvents(dispatcher, fds); } - if (cmsWait != kForeverMs) { - msWait = TimeDiff(msStop, TimeMillis()); - if (msWait < 0) { - // Return success on timeout. - return true; - } + wait_time_ms = deadline.WaitTime(); + if (wait_time_ms == 0) { + // Return success on timeout. + return true; } } @@ -1823,13 +1832,9 @@ #elif defined(WEBRTC_USE_POLL) -bool PhysicalSocketServer::WaitPoll(int cmsWait, bool process_io) { - int64_t msWait = -1; - int64_t msStop = -1; - if (cmsWait != kForeverMs) { - msWait = cmsWait; - msStop = TimeAfter(cmsWait); - } +bool PhysicalSocketServer::WaitPoll(TimeDelta timeout, bool process_io) { + int wait_time_ms = -1; + DeadlineTracker<int> deadline(timeout, wait_time_ms); std::vector<pollfd> pollfds; fWait_ = true; @@ -1855,7 +1860,7 @@ // < 0 means error // 0 means timeout // > 0 means count of descriptors ready - int n = poll(pollfds.data(), pollfds.size(), static_cast<int>(msWait)); + int n = poll(pollfds.data(), pollfds.size(), wait_time_ms); if (n < 0) { if (errno != EINTR) { RTC_LOG_E(LS_ERROR, EN, errno) << "poll"; @@ -1882,12 +1887,10 @@ } } - if (cmsWait != kForeverMs) { - msWait = TimeDiff(msStop, TimeMillis()); - if (msWait < 0) { - // Return success on timeout. - return true; - } + wait_time_ms = deadline.WaitTime(); + if (wait_time_ms == 0) { + // Return success on timeout. + return true; } } @@ -1904,10 +1907,8 @@ RTC_DCHECK(!waiting_); ScopedSetTrue set(&waiting_); - int cmsWait = ToCmsWait(max_wait_duration); - int64_t cmsTotal = cmsWait; - int64_t cmsElapsed = 0; - int64_t msStart = Time(); + DWORD wait_time_ms = 0; + DeadlineTracker<DWORD> deadline(max_wait_duration, wait_time_ms); fWait_ = true; while (fWait_) { @@ -1948,19 +1949,9 @@ } } - // Which is shorter, the delay wait or the asked wait? - - int64_t cmsNext; - if (cmsWait == kForeverMs) { - cmsNext = cmsWait; - } else { - cmsNext = std::max<int64_t>(0, cmsTotal - cmsElapsed); - } - // Wait for one of the events to signal - DWORD dw = - WSAWaitForMultipleEvents(static_cast<DWORD>(events.size()), &events[0], - false, static_cast<DWORD>(cmsNext), false); + DWORD dw = WSAWaitForMultipleEvents(static_cast<DWORD>(events.size()), + &events[0], false, wait_time_ms, false); if (dw == WSA_WAIT_FAILED) { // Failed? @@ -2066,9 +2057,11 @@ // Break? if (!fWait_) break; - cmsElapsed = TimeSince(msStart); - if ((cmsWait != kForeverMs) && (cmsElapsed >= cmsWait)) { - break; + + wait_time_ms = deadline.WaitTime(); + if (wait_time_ms == 0) { + // Return success on timeout. + return true; } }
diff --git a/rtc_base/physical_socket_server.h b/rtc_base/physical_socket_server.h index 80743f5..d35e8f5 100644 --- a/rtc_base/physical_socket_server.h +++ b/rtc_base/physical_socket_server.h
@@ -106,20 +106,16 @@ private: // The number of events to process with one call to "epoll_wait". static constexpr size_t kNumEpollEvents = 128; - // A local historical definition of "foreverness", in milliseconds. - static constexpr int kForeverMs = -1; - - static int ToCmsWait(TimeDelta max_wait_duration); #if defined(WEBRTC_POSIX) - bool WaitSelect(int cmsWait, bool process_io); + bool WaitSelect(TimeDelta timeout, bool process_io); #if defined(WEBRTC_USE_EPOLL) void AddEpoll(Dispatcher* dispatcher, uint64_t key); void RemoveEpoll(Dispatcher* dispatcher); void UpdateEpoll(Dispatcher* dispatcher, uint64_t key); - bool WaitEpoll(int cmsWait); - bool WaitPollOneDispatcher(int cmsWait, Dispatcher* dispatcher); + bool WaitEpoll(TimeDelta timeout); + bool WaitPollOneDispatcher(TimeDelta timeout, Dispatcher* dispatcher); // This array is accessed in isolation by a thread calling into Wait(). // It's useless to use a SequenceChecker to guard it because a socket @@ -129,7 +125,7 @@ const int epoll_fd_ = INVALID_SOCKET; #elif defined(WEBRTC_USE_POLL) - bool WaitPoll(int cmsWait, bool process_io); + bool WaitPoll(TimeDelta timeout, bool process_io); #endif // WEBRTC_USE_EPOLL, WEBRTC_USE_POLL #endif // WEBRTC_POSIX
diff --git a/rtc_base/system/BUILD.gn b/rtc_base/system/BUILD.gn index f74e951..da45543 100644 --- a/rtc_base/system/BUILD.gn +++ b/rtc_base/system/BUILD.gn
@@ -101,6 +101,21 @@ } } +# Function to query uninjectable system time. +# Visibility is limited to webrtc low level system utilities that can't use +# injectable clock. +rtc_library("system_time") { + visibility = [ "..:threading" ] + sources = [ + "system_time.cc", + "system_time.h", + ] + deps = [ + "..:timeutils", + "../../api/units:timestamp", + ] +} + rtc_library("warn_current_thread_is_deadlocked") { sources = [ "warn_current_thread_is_deadlocked.h" ] deps = []
diff --git a/rtc_base/system/system_time.cc b/rtc_base/system/system_time.cc new file mode 100644 index 0000000..6d703ca --- /dev/null +++ b/rtc_base/system/system_time.cc
@@ -0,0 +1,26 @@ +/* + * Copyright 2026 The WebRTC Project Authors. All rights reserved. + * + * Use of this source code is governed by a BSD-style license + * that can be found in the LICENSE file in the root of the source + * tree. An additional intellectual property rights grant can be found + * in the file PATENTS. All contributing project authors may + * be found in the AUTHORS file in the root of the source tree. + */ +#ifndef RTC_BASE_SYSTEM_SYSTEM_TIME_H_ +#define RTC_BASE_SYSTEM_SYSTEM_TIME_H_ + +#include "rtc_base/system/system_time.h" + +#include "api/units/timestamp.h" +#include "rtc_base/system_time.h" + +namespace webrtc { + +Timestamp SystemTime() { + return Timestamp::Micros(SystemTimeNanos() / 1'000); +} + +} // namespace webrtc + +#endif // RTC_BASE_SYSTEM_SYSTEM_TIME_H_
diff --git a/rtc_base/system/system_time.h b/rtc_base/system/system_time.h new file mode 100644 index 0000000..ec2a17d --- /dev/null +++ b/rtc_base/system/system_time.h
@@ -0,0 +1,21 @@ +/* + * Copyright 2026 The WebRTC Project Authors. All rights reserved. + * + * Use of this source code is governed by a BSD-style license + * that can be found in the LICENSE file in the root of the source + * tree. An additional intellectual property rights grant can be found + * in the file PATENTS. All contributing project authors may + * be found in the AUTHORS file in the root of the source tree. + */ +#ifndef RTC_BASE_SYSTEM_SYSTEM_TIME_H_ +#define RTC_BASE_SYSTEM_SYSTEM_TIME_H_ + +#include "api/units/timestamp.h" + +namespace webrtc { + +Timestamp SystemTime(); + +} // namespace webrtc + +#endif // RTC_BASE_SYSTEM_SYSTEM_TIME_H_