Fea, 添加TaskRunner::AsyncSleep、TaskRunner::AsyncWaitForObject

Author: mingkuang-Chuyu

UnitTest/TaskRunnerUnitTest.cpp

@@ -367,7 +367,7 @@ namespace TaskRunnerUnitTest
                     return false;
                 });
 
-            Assert::IsTrue(((TaskEntry*)_pWait.Get())->Wait(600ul));
+            Assert::IsTrue(((TaskEntry*)_pWait.Get())->WaitTask(YY::TimeSpan::FromMilliseconds(600ul)));
             Assert::AreEqual(DWORD(_uWaitResult), DWORD(WAIT_TIMEOUT));
 
             _pWait = _pTaskRunner->CreateWait(
@@ -379,7 +379,7 @@ namespace TaskRunnerUnitTest
                     return false;
                 });
             SetEvent(_hEvent);
-            Assert::IsTrue(((TaskEntry*)_pWait.Get())->Wait(100ul));
+            Assert::IsTrue(((TaskEntry*)_pWait.Get())->WaitTask(YY::TimeSpan::FromMilliseconds(100ul)));
             Assert::AreEqual(DWORD(_uWaitResult), DWORD(WAIT_OBJECT_0));
             // CloseHandle(_hEvent);
         }
@@ -836,7 +836,7 @@ namespace TaskRunnerUnitTest
                     return false;
                 });
 
-            Assert::IsTrue(((TaskEntry*)_pWait.Get())->Wait(600ul));
+            Assert::IsTrue(_pWait->WaitTask(YY::TimeSpan::FromMilliseconds(600ul)));
             Assert::AreEqual(DWORD(_uWaitResult), DWORD(WAIT_TIMEOUT));
 
             _pWait = _pTaskRunner->CreateWait(
@@ -848,8 +848,76 @@ namespace TaskRunnerUnitTest
                     return false;
                 });
             SetEvent(_hEvent);
-            Assert::IsTrue(((TaskEntry*)_pWait.Get())->Wait(100ul));
+            Assert::IsTrue((_pWait->WaitTask(YY::TimeSpan::FromMilliseconds(100ul))));
             // CloseHandle(_hEvent);
         }
     };
+
+    TEST_CLASS(CommonTaskRunnerUnitTest)
+    {
+    public:
+#if defined(_HAS_CXX20) && _HAS_CXX20
+        TEST_METHOD(AsyncSleep)
+        {
+            const auto _hrPending = E_PENDING;
+
+            auto _pTaskRunner = SequencedTaskRunner::Create();
+
+            auto _pHr = std::make_shared<HRESULT>(_hrPending);
+            std::weak_ptr<HRESULT> _pWeakHr = _pHr;
+            _pTaskRunner->PostTask(
+                [_pWeakHr]() -> YY::Coroutine<void>
+                {
+                    auto _pHr = _pWeakHr.lock();
+                    if (!_pHr)
+                        co_return;
+
+                    *_pHr = co_await YY::TaskRunner::AsyncSleep(YY::TimeSpan::FromMilliseconds(500));
+                    WakeByAddressAll(_pHr.get());
+                    co_return;
+                });
+
+            ::WaitOnAddress((volatile void*)_pHr.get(), (void*)&_hrPending, sizeof(_hrPending), 2000);
+
+            Assert::AreEqual(HRESULT(S_OK), HRESULT(*_pHr));
+        }
+#endif
+
+#if defined(_HAS_CXX20) && _HAS_CXX20 && defined(_WIN32)
+        TEST_METHOD(AsyncWaitForObject)
+        {
+            const auto _hrPending = E_PENDING;
+
+            auto _pTaskRunner = SequencedTaskRunner::Create();
+
+            auto _pHr = std::make_shared<HRESULT>(_hrPending);
+            std::weak_ptr<HRESULT> _pWeakHr = _pHr;
+            _pTaskRunner->PostTask(
+                [_pWeakHr]() -> YY::Coroutine<void>
+                {
+                    auto _pHr = _pWeakHr.lock();
+                    if (!_pHr)
+                        co_return;
+
+                    auto _hEvent = CreateEventW(nullptr, FALSE, FALSE, nullptr);
+
+                    auto _uResult = co_await YY::TaskRunner::AsyncWaitForObject(_hEvent, YY::TimeSpan::FromMilliseconds(100));
+
+                    Assert::AreEqual(DWORD(WAIT_TIMEOUT), DWORD(_uResult));
+
+                    SetEvent(_hEvent);
+                    _uResult = co_await YY::TaskRunner::AsyncWaitForObject(_hEvent, YY::TimeSpan::FromMilliseconds(100));
+                    Assert::AreEqual(DWORD(WAIT_OBJECT_0), DWORD(_uResult));
+
+                    *_pHr.get() = S_OK;
+                    WakeByAddressAll(_pHr.get());
+                    co_return;
+                });
+
+            ::WaitOnAddress((volatile void*)_pHr.get(), (void*)&_hrPending, sizeof(_hrPending), 2000);
+
+            Assert::AreEqual(HRESULT(S_OK), HRESULT(*_pHr));
+        }
+#endif
+    };
 } // namespace UnitTest

include/YY/Base/Threading/TaskRunner.h

@@ -333,6 +333,30 @@ namespace YY
 
                 virtual TaskRunnerStyle __YYAPI GetStyle() const noexcept = 0;
 
+#if defined(_HAS_CXX20) && _HAS_CXX20
+                /// <summary>
+                /// 返回一个用于异步睡眠的等待器，该等待器在指定时间间隔后完成并提供 HRESULT 结果。
+                /// </summary>
+                /// <param name="_uAfter">要等待的时间间隔（TimeSpan），表示异步睡眠的持续时间。</param>
+                /// <returns>一个 TaskAwaiter<HRESULT>，可用于等待异步睡眠操作完成并获取表示操作结果的 HRESULT。</returns>
+                static TaskAwaiter<HRESULT> __YYAPI AsyncSleep(_In_ TimeSpan _uAfter);
+#endif
+
+#if defined(_HAS_CXX20) && _HAS_CXX20 && defined(_WIN32)
+                /// <summary>
+                /// 异步等待指定对象并返回表示等待结果的 awaiter。
+                /// </summary>
+                /// <param name="_hHandle">要等待的对象句柄。</param>
+                /// <param name="_iWaitTimeOut">等待超时时间；可选，默认为 TimeSpan::GetMax()（表示最大/无限超时）。</param>
+                /// <returns>返回一个 TaskAwaiter<DWORD>，用于异步等待并在完成后获取等待结果，其结果可能值为：
+                /// * WAIT_ABANDONED
+                /// * WAIT_OBJECT_0
+                /// * WAIT_TIMEOUT
+                /// * WAIT_FAILED
+                /// </returns>
+                static TaskAwaiter<DWORD> __YYAPI AsyncWaitForObject(_In_ HANDLE _hHandle, _In_ TimeSpan _iWaitTimeOut = TimeSpan::GetMax());
+#endif
+
                 /// <summary>
                 /// 将任务异步执行。
                 /// </summary>

src/YY/Base/Threading/TaskRunner.cpp

@@ -162,6 +162,170 @@ namespace YY
                 return TaskRunnerDispatch::Get()->StartIo();
             }
 
+#if defined(_HAS_CXX20) && _HAS_CXX20
+            TaskAwaiter<HRESULT>__YYAPI TaskRunner::AsyncSleep(TimeSpan _uAfter)
+            {
+                const auto _uExpire = TickCount::GetNow() + _uAfter;
+
+                struct AsyncTaskEntry
+                    : public Timer
+                    , public TaskAwaiter<HRESULT>::RefData
+                {
+                    HRESULT _hrValue = E_PENDING;
+
+                    uint32_t __YYAPI AddRef() noexcept override
+                    {
+                        return Timer::AddRef();
+                    }
+
+                    uint32_t __YYAPI Release() noexcept override
+                    {
+                        return Timer::Release();
+                    }
+
+                    HRESULT __YYAPI RunTask() override
+                    {
+                        if (IsCanceled())
+                            return YY::Base::HRESULT_From_LSTATUS(ERROR_CANCELLED);
+
+                        _hrValue = S_OK;
+                        auto _hHandle = (void*)YY::Base::Sync::Exchange(&hCoroutineHandle, /*hReadyHandle*/ (intptr_t)-1);
+                        if (_hHandle)
+                        {
+                            try
+                            {
+                                std::coroutine_handle<>::from_address(_hHandle).resume();
+                            }
+                            catch (const YY::Base::OperationCanceledException& _Exception)
+                            {
+                                _hrValue = YY::Base::HRESULT_From_LSTATUS(ERROR_CANCELLED);
+                            }
+                        }
+
+                        Wakeup(_hrValue);
+                        return _hrValue;
+                    }
+
+                    HRESULT __YYAPI Resume() noexcept override
+                    {
+                        return _hrValue;
+                    }
+                };
+
+                auto _pAsyncTaskEntry = RefPtr<AsyncTaskEntry>::Create();
+                if (!_pAsyncTaskEntry)
+                    throw Exception();
+
+                _pAsyncTaskEntry->uExpire = _uExpire;
+                HRESULT _hr = S_OK;
+                do
+                {
+                    auto _pTaskRunner = TaskRunner::GetCurrent();
+                    if (!_pTaskRunner)
+                    {
+                        _hr = E_UNEXPECTED;
+                        break;
+                    }
+
+                    _hr = _pTaskRunner->SetTimerInternal(_pAsyncTaskEntry);
+
+                } while (false);
+
+                if (_hr != S_OK)
+                {
+                    _pAsyncTaskEntry->hCoroutineHandle = (intptr_t)-1;
+                    _pAsyncTaskEntry->_hrValue = _hr;
+                }
+
+                return TaskAwaiter<HRESULT>(std::move(_pAsyncTaskEntry));
+            }
+#endif
+
+#if defined(_HAS_CXX20) && _HAS_CXX20 && defined(_WIN32)
+            TaskAwaiter<DWORD>__YYAPI TaskRunner::AsyncWaitForObject(HANDLE _hHandle, TimeSpan _iWaitTimeOut)
+            {
+                struct AsyncTaskEntry
+                    : public Wait
+                    , public TaskAwaiter<DWORD>::RefData
+                {
+                    uint32_t __YYAPI AddRef() noexcept override
+                    {
+                        return Wait::AddRef();
+                    }
+
+                    uint32_t __YYAPI Release() noexcept override
+                    {
+                        return Wait::Release();
+                    }
+
+                    HRESULT __YYAPI RunTask() override
+                    {
+                        if (IsCanceled())
+                            return YY::Base::HRESULT_From_LSTATUS(ERROR_CANCELLED);
+
+                        HRESULT _hr = S_OK;
+                        auto _hHandle = (void*)YY::Base::Sync::Exchange(&hCoroutineHandle, /*hReadyHandle*/ (intptr_t)-1);
+                        if (_hHandle)
+                        {
+                            try
+                            {
+                                std::coroutine_handle<>::from_address(_hHandle).resume();
+                            }
+                            catch (const YY::Base::OperationCanceledException& _Exception)
+                            {
+                                _hr = YY::Base::HRESULT_From_LSTATUS(ERROR_CANCELLED);
+                            }
+                        }
+
+                        Wakeup(_hr);
+                        return _hr;
+                    }
+
+                    DWORD __YYAPI Resume() noexcept override
+                    {
+                        return uWaitResult;
+                    }
+                };
+
+                auto _pAsyncTaskEntry = RefPtr<AsyncTaskEntry>::Create();
+                if (!_pAsyncTaskEntry)
+                    throw Exception();
+
+                _pAsyncTaskEntry->hHandle = _hHandle;
+
+                // >= UINT32_MAX 时认为是无限等待。
+                if (_iWaitTimeOut >= TimeSpan::FromMilliseconds(UINT32_MAX))
+                {
+                    _pAsyncTaskEntry->uTimeOut = TickCount::GetMax();
+                }
+                else
+                {
+                    _pAsyncTaskEntry->uTimeOut = TickCount::GetNow() + _iWaitTimeOut;
+                }
+
+                HRESULT _hr = S_OK;
+                do
+                {
+                    auto _pTaskRunner = TaskRunner::GetCurrent();
+                    if (!_pTaskRunner)
+                    {
+                        _hr = E_UNEXPECTED;
+                        break;
+                    }
+                    
+                    _hr = _pTaskRunner->SetWaitInternal(_pAsyncTaskEntry);
+
+                } while (false);
+
+                if (_hr != S_OK)
+                {
+                    _pAsyncTaskEntry->hCoroutineHandle = (intptr_t)-1;
+                }
+
+                return TaskAwaiter<DWORD>(std::move(_pAsyncTaskEntry));
+            }
+#endif
+
             HRESULT __YYAPI TaskRunner::PostDelayTask(TimeSpan _uAfter, std::function<void(void)>&& _pfnTaskCallback)
             {
                 auto _uExpire = TickCount::GetNow() + _uAfter;