drogon
C++14/17-based HTTP application framework
Loading...
Searching...
No Matches
coroutine.h
Go to the documentation of this file.
1
14#pragma once
15
16#include <trantor/utils/NonCopyable.h>
17#include <trantor/net/EventLoop.h>
18#include <trantor/utils/Logger.h>
19#include <algorithm>
20#include <atomic>
21#include <cassert>
22#include <condition_variable>
23#include <coroutine>
24#include <exception>
25#include <future>
26#include <mutex>
27#include <type_traits>
28#include <optional>
29
30namespace drogon
31{
32namespace internal
33{
34template <typename T>
35auto getAwaiterImpl(T &&value) noexcept(
36 noexcept(static_cast<T &&>(value).operator co_await()))
37 -> decltype(static_cast<T &&>(value).operator co_await())
38{
39 return static_cast<T &&>(value).operator co_await();
40}
41
42template <typename T>
43auto getAwaiterImpl(T &&value) noexcept(
44 noexcept(operator co_await(static_cast<T &&>(value))))
45 -> decltype(operator co_await(static_cast<T &&>(value)))
46{
47 return operator co_await(static_cast<T &&>(value));
48}
49
50template <typename T>
51auto getAwaiter(T &&value) noexcept(
52 noexcept(getAwaiterImpl(static_cast<T &&>(value))))
53 -> decltype(getAwaiterImpl(static_cast<T &&>(value)))
54{
55 return getAwaiterImpl(static_cast<T &&>(value));
56}
57
58template <typename T>
59using void_to_false_t =
60 std::conditional_t<std::is_same_v<T, void>, std::false_type, T>;
61
62} // end namespace internal
63
64template <typename T>
66{
67 using awaiter_t = decltype(internal::getAwaiter(std::declval<T>()));
68 using type = decltype(std::declval<awaiter_t>().await_resume());
69};
70
71template <typename T>
72using await_result_t = typename await_result<T>::type;
73
74template <typename T, typename = std::void_t<>>
75struct is_awaitable : std::false_type
76{
77};
78
79template <typename T>
81 T,
82 std::void_t<decltype(internal::getAwaiter(std::declval<T>()))>>
83 : std::true_type
84{
85};
86
87template <typename T>
88constexpr bool is_awaitable_v = is_awaitable<T>::value;
89
96{
97 bool await_ready() noexcept
98 {
99 return false;
100 }
101
102 template <typename T>
103 auto await_suspend(std::coroutine_handle<T> handle) noexcept
104 {
105 return handle.promise().continuation_;
106 }
107
108 void await_resume() noexcept
109 {
110 }
111};
112
121template <typename Promise>
122struct task_awaiter
123{
124 using handle_type = std::coroutine_handle<Promise>;
125
126 public:
127 explicit task_awaiter(handle_type coro) : coro_(coro)
128 {
129 }
130
131 bool await_ready() noexcept
132 {
133 return !coro_ || coro_.done();
134 }
135
136 auto await_suspend(std::coroutine_handle<> handle) noexcept
137 {
138 coro_.promise().setContinuation(handle);
139 return coro_;
140 }
141
142 auto await_resume()
143 {
144 if constexpr (std::is_void_v<decltype(coro_.promise().result())>)
145 {
146 coro_.promise().result(); // throw exception if any
147 return;
148 }
149 else
150 {
151 return std::move(coro_.promise().result());
152 }
153 }
154
155 private:
156 handle_type coro_;
157};
158
159template <typename T = void>
160struct [[nodiscard]] Task
161{
162 struct promise_type;
163 using handle_type = std::coroutine_handle<promise_type>;
164
165 Task(handle_type h) : coro_(h)
166 {
167 }
168
169 Task(const Task &) = delete;
170
171 Task(Task &&other) noexcept
172 {
173 coro_ = other.coro_;
174 other.coro_ = nullptr;
175 }
176
177 ~Task()
178 {
179 if (coro_)
180 coro_.destroy();
181 }
182
183 Task &operator=(const Task &) = delete;
184
185 Task &operator=(Task &&other) noexcept
186 {
187 if (std::addressof(other) == this)
188 return *this;
189 if (coro_)
190 coro_.destroy();
191
192 coro_ = other.coro_;
193 other.coro_ = nullptr;
194 return *this;
195 }
196
198 {
199 Task<T> get_return_object()
200 {
201 return Task<T>{handle_type::from_promise(*this)};
202 }
203
204 std::suspend_always initial_suspend()
205 {
206 return {};
207 }
208
209 void return_value(const T &v)
210 {
211 value = v;
212 }
213
214 void return_value(T &&v)
215 {
216 value = std::move(v);
217 }
218
219 auto final_suspend() noexcept
220 {
221 return final_awaiter{};
222 }
223
224 void unhandled_exception()
225 {
226 exception_ = std::current_exception();
227 }
228
229 T &&result() &&
230 {
231 if (exception_ != nullptr)
232 std::rethrow_exception(exception_);
233 assert(value.has_value() == true);
234 return std::move(value.value());
235 }
236
237 T &result() &
238 {
239 if (exception_ != nullptr)
240 std::rethrow_exception(exception_);
241 assert(value.has_value() == true);
242 return value.value();
243 }
244
245 void setContinuation(std::coroutine_handle<> handle)
246 {
247 continuation_ = handle;
248 }
249
250 std::optional<T> value;
251 std::exception_ptr exception_;
252 std::coroutine_handle<> continuation_{std::noop_coroutine()};
253 };
254
255 auto operator co_await() const noexcept
256 {
257 return task_awaiter(coro_);
258 }
259
260 handle_type coro_;
261};
262
263template <>
264struct [[nodiscard]] Task<void>
265{
266 struct promise_type;
267 using handle_type = std::coroutine_handle<promise_type>;
268
269 Task(handle_type handle) : coro_(handle)
270 {
271 }
272
273 Task(const Task &) = delete;
274
275 Task(Task &&other) noexcept
276 {
277 coro_ = other.coro_;
278 other.coro_ = nullptr;
279 }
280
281 ~Task()
282 {
283 if (coro_)
284 coro_.destroy();
285 }
286
287 Task &operator=(const Task &) = delete;
288
289 Task &operator=(Task &&other) noexcept
290 {
291 if (std::addressof(other) == this)
292 return *this;
293 if (coro_)
294 coro_.destroy();
295
296 coro_ = other.coro_;
297 other.coro_ = nullptr;
298 return *this;
299 }
300
302 {
303 Task<> get_return_object()
304 {
305 return Task<>{handle_type::from_promise(*this)};
306 }
307
308 std::suspend_always initial_suspend()
309 {
310 return {};
311 }
312
313 void return_void()
314 {
315 }
316
317 auto final_suspend() noexcept
318 {
319 return final_awaiter{};
320 }
321
322 void unhandled_exception()
323 {
324 exception_ = std::current_exception();
325 }
326
327 void result()
328 {
329 if (exception_ != nullptr)
330 std::rethrow_exception(exception_);
331 }
332
333 void setContinuation(std::coroutine_handle<> handle)
334 {
335 continuation_ = handle;
336 }
337
338 std::exception_ptr exception_;
339 std::coroutine_handle<> continuation_{std::noop_coroutine()};
340 };
341
342 auto operator co_await() const noexcept
343 {
344 return task_awaiter(coro_);
345 }
346
347 handle_type coro_;
348};
349
352// NOTE: AsyncTask is designed to be not awaitable. And kills the entire process
353// if exception escaped.
354struct AsyncTask
355{
356 struct promise_type;
357 using handle_type = std::coroutine_handle<promise_type>;
358
359 AsyncTask() = default;
360
361 AsyncTask(handle_type h) : coro_(h)
362 {
363 }
364
365 AsyncTask(const AsyncTask &) = delete;
366
367 AsyncTask(AsyncTask &&other) noexcept
368 {
369 coro_ = other.coro_;
370 other.coro_ = nullptr;
371 }
372
373 AsyncTask &operator=(const AsyncTask &) = delete;
374
375 AsyncTask &operator=(AsyncTask &&other) noexcept
376 {
377 if (std::addressof(other) == this)
378 return *this;
379
380 coro_ = other.coro_;
381 other.coro_ = nullptr;
382 return *this;
383 }
384
386 {
387 AsyncTask get_return_object() noexcept
388 {
389 return {std::coroutine_handle<promise_type>::from_promise(*this)};
390 }
391
392 std::suspend_never initial_suspend() const noexcept
393 {
394 return {};
395 }
396
397 void unhandled_exception()
398 {
399 LOG_FATAL << "Exception escaping AsyncTask.";
400 std::terminate();
401 }
402
403 void return_void() noexcept
404 {
405 }
406
407 std::suspend_never final_suspend() const noexcept
408 {
409 return {};
410 }
411 };
412
413 handle_type coro_;
414};
415
418// The user is responsible to fill in `await_suspend()` and constructors.
419template <typename T = void>
420struct CallbackAwaiter : public trantor::NonCopyable
421{
422 bool await_ready() noexcept
423 {
424 return false;
425 }
426
427 bool hasException() const noexcept
428 {
429 return exception_ != nullptr;
430 }
431
432 const T &await_resume() const noexcept(false)
433 {
434 // await_resume() should always be called after co_await
435 // (await_suspend()) is called. Therefore the value should always be set
436 // (or there's an exception)
437 assert(result_.has_value() == true || exception_ != nullptr);
438
439 if (exception_)
440 std::rethrow_exception(exception_);
441 return result_.value();
442 }
443
444 private:
445 // HACK: Not all desired types are default constructable. But we need the
446 // entire struct to be constructed for awaiting. std::optional takes care of
447 // that.
448 std::optional<T> result_;
449 std::exception_ptr exception_{nullptr};
450
451 protected:
452 void setException(const std::exception_ptr &e)
453 {
454 exception_ = e;
455 }
456
457 void setValue(const T &v)
458 {
459 result_.emplace(v);
460 }
461
462 void setValue(T &&v)
463 {
464 result_.emplace(std::move(v));
465 }
466};
467
468template <>
469struct CallbackAwaiter<void> : public trantor::NonCopyable
470{
471 bool await_ready() noexcept
472 {
473 return false;
474 }
475
476 void await_resume() noexcept(false)
477 {
478 if (exception_)
479 std::rethrow_exception(exception_);
480 }
481
482 bool hasException() const noexcept
483 {
484 return exception_ != nullptr;
485 }
486
487 private:
488 std::exception_ptr exception_{nullptr};
489
490 protected:
491 void setException(const std::exception_ptr &e)
492 {
493 exception_ = e;
494 }
495};
496
497// An ok implementation of sync_await. This allows one to call
498// coroutines and wait for the result from a function.
499template <typename Await>
500auto sync_wait(Await &&await)
501{
502 static_assert(is_awaitable_v<std::decay_t<Await>>);
503 using value_type = typename await_result<Await>::type;
504 std::condition_variable cv;
505 std::mutex mtx;
506 std::atomic<bool> flag = false;
507 std::exception_ptr exception_ptr;
508 std::unique_lock lk(mtx);
509
510 if constexpr (std::is_same_v<value_type, void>)
511 {
512 auto task = [&]() -> AsyncTask {
513 try
514 {
515 co_await await;
516 }
517 catch (...)
518 {
519 exception_ptr = std::current_exception();
520 }
521 std::unique_lock lk(mtx);
522 flag = true;
523 cv.notify_all();
524 };
525
526 std::thread thr([&]() { task(); });
527 cv.wait(lk, [&]() { return (bool)flag; });
528 thr.join();
529 if (exception_ptr)
530 std::rethrow_exception(exception_ptr);
531 }
532 else
533 {
534 std::optional<value_type> value;
535 auto task = [&]() -> AsyncTask {
536 try
537 {
538 value = co_await await;
539 }
540 catch (...)
541 {
542 exception_ptr = std::current_exception();
543 }
544 std::unique_lock lk(mtx);
545 flag = true;
546 cv.notify_all();
547 };
548
549 std::thread thr([&]() { task(); });
550 cv.wait(lk, [&]() { return (bool)flag; });
551 assert(value.has_value() == true || exception_ptr);
552 thr.join();
553
554 if (exception_ptr)
555 std::rethrow_exception(exception_ptr);
556
557 return std::move(value.value());
558 }
559}
560
561// Converts a task (or task like) promise into std::future for old-style async
562template <typename Await>
563inline auto co_future(Await &&await) noexcept
564 -> std::future<await_result_t<Await>>
565{
566 using Result = await_result_t<Await>;
567 std::promise<Result> prom;
568 auto fut = prom.get_future();
569 [](std::promise<Result> prom, Await await) -> AsyncTask {
570 try
571 {
572 if constexpr (std::is_void_v<Result>)
573 {
574 co_await std::move(await);
575 prom.set_value();
576 }
577 else
578 prom.set_value(co_await std::move(await));
579 }
580 catch (...)
581 {
582 prom.set_exception(std::current_exception());
583 }
584 }(std::move(prom), std::move(await));
585 return fut;
586}
587
588namespace internal
589{
590struct [[nodiscard]] TimerAwaiter : CallbackAwaiter<void>
591{
592 TimerAwaiter(trantor::EventLoop *loop,
593 const std::chrono::duration<double> &delay)
594 : loop_(loop), delay_(delay.count())
595 {
596 }
597
598 TimerAwaiter(trantor::EventLoop *loop, double delay)
599 : loop_(loop), delay_(delay)
600 {
601 }
602
603 void await_suspend(std::coroutine_handle<> handle)
604 {
605 loop_->runAfter(delay_, [handle]() { handle.resume(); });
606 }
607
608 private:
609 trantor::EventLoop *loop_;
610 double delay_;
611};
612
613struct [[nodiscard]] LoopAwaiter : CallbackAwaiter<void>
614{
615 LoopAwaiter(trantor::EventLoop *workLoop,
616 std::function<void()> &&taskFunc,
617 trantor::EventLoop *resumeLoop = nullptr)
618 : workLoop_(workLoop),
619 resumeLoop_(resumeLoop),
620 taskFunc_(std::move(taskFunc))
621 {
622 assert(workLoop);
623 }
624
625 void await_suspend(std::coroutine_handle<> handle)
626 {
627 workLoop_->queueInLoop([handle, this]() {
628 try
629 {
630 taskFunc_();
631 if (resumeLoop_ && resumeLoop_ != workLoop_)
632 resumeLoop_->queueInLoop([handle]() { handle.resume(); });
633 else
634 handle.resume();
635 }
636 catch (...)
637 {
638 setException(std::current_exception());
639 if (resumeLoop_ && resumeLoop_ != workLoop_)
640 resumeLoop_->queueInLoop([handle]() { handle.resume(); });
641 else
642 handle.resume();
643 }
644 });
645 }
646
647 private:
648 trantor::EventLoop *workLoop_{nullptr};
649 trantor::EventLoop *resumeLoop_{nullptr};
650 std::function<void()> taskFunc_;
651};
652
653struct [[nodiscard]] SwitchThreadAwaiter : CallbackAwaiter<void>
654{
655 explicit SwitchThreadAwaiter(trantor::EventLoop *loop) : loop_(loop)
656 {
657 }
658
659 void await_suspend(std::coroutine_handle<> handle)
660 {
661 loop_->runInLoop([handle]() { handle.resume(); });
662 }
663
664 private:
665 trantor::EventLoop *loop_;
666};
667
668struct [[nodiscard]] EndAwaiter : CallbackAwaiter<void>
669{
670 EndAwaiter(trantor::EventLoop *loop) : loop_(loop)
671 {
672 assert(loop);
673 }
674
675 void await_suspend(std::coroutine_handle<> handle)
676 {
677 loop_->runOnQuit([handle]() { handle.resume(); });
678 }
679
680 private:
681 trantor::EventLoop *loop_{nullptr};
682};
683
684} // namespace internal
685
686inline internal::TimerAwaiter sleepCoro(
687 trantor::EventLoop *loop,
688 const std::chrono::duration<double> &delay) noexcept
689{
690 assert(loop);
691 return {loop, delay};
692}
693
694inline internal::TimerAwaiter sleepCoro(trantor::EventLoop *loop,
695 double delay) noexcept
696{
697 assert(loop);
698 return {loop, delay};
699}
700
701inline internal::LoopAwaiter queueInLoopCoro(
702 trantor::EventLoop *workLoop,
703 std::function<void()> taskFunc,
704 trantor::EventLoop *resumeLoop = nullptr)
705{
706 assert(workLoop);
707 return {workLoop, std::move(taskFunc), resumeLoop};
708}
709
710inline internal::SwitchThreadAwaiter switchThreadCoro(
711 trantor::EventLoop *loop) noexcept
712{
713 assert(loop);
715}
716
717inline internal::EndAwaiter untilQuit(trantor::EventLoop *loop)
718{
719 assert(loop);
720 return {loop};
721}
722
723template <typename T, typename = std::void_t<>>
724struct is_resumable : std::false_type
725{
726};
727
728template <typename T>
730 T,
731 std::void_t<decltype(internal::getAwaiter(std::declval<T>()))>>
732 : std::true_type
733{
734};
735
736template <>
737struct is_resumable<AsyncTask, std::void_t<AsyncTask>> : std::true_type
738{
739};
740
741template <typename T>
742constexpr bool is_resumable_v = is_resumable<T>::value;
743
748template <typename Coro>
749void async_run(Coro &&coro)
750{
751 using CoroValueType = std::decay_t<Coro>;
752 auto functor = [](CoroValueType coro) -> AsyncTask {
753 auto frame = coro();
754
755 using FrameType = std::decay_t<decltype(frame)>;
756 static_assert(is_awaitable_v<FrameType>);
757
758 co_await frame;
759 co_return;
760 };
761 functor(std::forward<Coro>(coro));
762}
763
768template <typename Coro>
769std::function<void()> async_func(Coro &&coro)
770{
771 return [coro = std::forward<Coro>(coro)]() mutable {
772 async_run(std::move(coro));
773 };
774}
775
776namespace internal
777{
778template <typename T>
779struct [[nodiscard]] EventLoopAwaiter : public drogon::CallbackAwaiter<T>
780{
781 EventLoopAwaiter(std::function<T()> &&task, trantor::EventLoop *loop)
782 : task_(std::move(task)), loop_(loop)
783 {
784 }
785
786 void await_suspend(std::coroutine_handle<> handle)
787 {
788 loop_->queueInLoop([this, handle]() {
789 try
790 {
791 if constexpr (!std::is_same_v<T, void>)
792 {
793 this->setValue(task_());
794 handle.resume();
795 }
796 else
797 {
798 task_();
799 handle.resume();
800 }
801 }
802 catch (const std::exception &err)
803 {
804 LOG_ERROR << err.what();
805 this->setException(std::current_exception());
806 handle.resume();
807 }
808 });
809 }
810
811 private:
812 std::function<T()> task_;
813 trantor::EventLoop *loop_;
814};
815
816template <typename... Tasks>
817struct WhenAllAwaiter
818 : public CallbackAwaiter<
819 std::tuple<internal::void_to_false_t<await_result_t<Tasks>>...>>
820{
821 WhenAllAwaiter(Tasks... tasks)
822 : tasks_(std::forward<Tasks>(tasks)...), counter_(sizeof...(tasks))
823 {
824 }
825
826 void await_suspend(std::coroutine_handle<> handle)
827 {
828 if (counter_ == 0)
829 {
830 handle.resume();
831 return;
832 }
833
834 await_suspend_impl(handle, std::index_sequence_for<Tasks...>{});
835 }
836
837 private:
838 std::tuple<Tasks...> tasks_;
839 std::atomic<size_t> counter_;
840 std::tuple<internal::void_to_false_t<await_result_t<Tasks>>...> results_;
841 std::atomic_flag exceptionFlag_;
842
843 template <size_t Idx>
844 void launch_task(std::coroutine_handle<> handle)
845 {
846 using Self = WhenAllAwaiter<Tasks...>;
847 [](Self *self, std::coroutine_handle<> handle) -> AsyncTask {
848 try
849 {
850 using TaskType = std::tuple_element_t<
851 Idx,
852 std::remove_cvref_t<decltype(results_)>>;
853 if constexpr (std::is_same_v<TaskType, std::false_type>)
854 {
855 co_await std::get<Idx>(self->tasks_);
856 std::get<Idx>(self->results_) = std::false_type{};
857 }
858 else
859 {
860 std::get<Idx>(self->results_) =
861 co_await std::get<Idx>(self->tasks_);
862 }
863 }
864 catch (...)
865 {
866 if (self->exceptionFlag_.test_and_set() == false)
867 self->setException(std::current_exception());
868 }
869
870 if (self->counter_.fetch_sub(1, std::memory_order_acq_rel) == 1)
871 {
872 if (!self->hasException())
873 self->setValue(std::move(self->results_));
874 handle.resume();
875 }
876 }(this, handle);
877 }
878
879 template <size_t... Is>
880 void await_suspend_impl(std::coroutine_handle<> handle,
881 std::index_sequence<Is...>)
882 {
883 ((launch_task<Is>(handle)), ...);
884 }
885};
886
887template <typename T>
888struct WhenAllAwaiter<std::vector<Task<T>>>
889 : public CallbackAwaiter<std::vector<T>>
890{
891 WhenAllAwaiter(std::vector<Task<T>> tasks)
892 : tasks_(std::move(tasks)),
893 counter_(tasks_.size()),
894 results_(tasks_.size())
895 {
896 }
897
898 void await_suspend(std::coroutine_handle<> handle)
899 {
900 if (tasks_.empty())
901 {
902 this->setValue(std::vector<T>{});
903 handle.resume();
904 return;
905 }
906
907 const size_t count = tasks_.size();
908 for (size_t i = 0; i < count; ++i)
909 {
910 [](WhenAllAwaiter *self,
911 std::coroutine_handle<> handle,
912 Task<T> task,
913 size_t index) -> AsyncTask {
914 try
915 {
916 auto result = co_await task;
917 self->results_[index] = std::move(result);
918 }
919 catch (...)
920 {
921 if (self->exceptionFlag_.test_and_set() == false)
922 self->setException(std::current_exception());
923 }
924
925 if (self->counter_.fetch_sub(1, std::memory_order_acq_rel) == 1)
926 {
927 if (!self->hasException())
928 {
929 self->setValue(std::move(self->results_));
930 }
931 handle.resume();
932 }
933 }(this, handle, std::move(tasks_[i]), i);
934 }
935 }
936
937 private:
938 std::vector<Task<T>> tasks_;
939 std::atomic<size_t> counter_;
940 std::vector<T> results_;
941 std::atomic_flag exceptionFlag_;
942};
943
944template <>
945struct WhenAllAwaiter<std::vector<Task<void>>> : public CallbackAwaiter<void>
946{
947 WhenAllAwaiter(std::vector<Task<void>> &&t)
948 : tasks_(std::move(t)), counter_(tasks_.size())
949 {
950 }
951
952 void await_suspend(std::coroutine_handle<> handle)
953 {
954 if (tasks_.empty())
955 {
956 handle.resume();
957 return;
958 }
959
960 const size_t count =
961 tasks_
962 .size(); // capture the size fist (see lifetime comment beflow)
963 for (size_t i = 0; i < count; ++i)
964 {
965 [](WhenAllAwaiter *self,
966 std::coroutine_handle<> handle,
967 Task<> task) -> AsyncTask {
968 try
969 {
970 co_await task;
971 }
972 catch (...)
973 {
974 if (self->exceptionFlag_.test_and_set() == false)
975 self->setException(std::current_exception());
976 }
977 if (self->counter_.fetch_sub(1, std::memory_order_acq_rel) == 1)
978 // This line CAN delete `this` at last iteration. We MUST
979 // NOT depend on this after last iteration
980 handle.resume();
981 }(this, handle, std::move(tasks_[i]));
982 }
983 }
984
985 std::vector<Task<void>> tasks_;
986 std::atomic<size_t> counter_;
987 std::atomic_flag exceptionFlag_;
988};
989} // namespace internal
990
995template <typename T>
996inline internal::EventLoopAwaiter<T> queueInLoopCoro(trantor::EventLoop *loop,
997 std::function<T()> task)
998{
999 return internal::EventLoopAwaiter<T>(std::move(task), loop);
1000}
1001
1002class Mutex final
1003{
1004 class ScopedCoroMutexAwaiter;
1005 class CoroMutexAwaiter;
1006
1007 public:
1008 Mutex() noexcept : state_(unlockedValue()), waiters_(nullptr)
1009 {
1010 }
1011
1012 Mutex(const Mutex &) = delete;
1013 Mutex(Mutex &&) = delete;
1014 Mutex &operator=(const Mutex &) = delete;
1015 Mutex &operator=(Mutex &&) = delete;
1016
1017 ~Mutex()
1018 {
1019 [[maybe_unused]] auto state = state_.load(std::memory_order_relaxed);
1020 assert(state == unlockedValue() || state == nullptr);
1021 assert(waiters_ == nullptr);
1022 }
1023
1024 bool try_lock() noexcept
1025 {
1026 void *oldValue = unlockedValue();
1027 return state_.compare_exchange_strong(oldValue,
1028 nullptr,
1029 std::memory_order_acquire,
1030 std::memory_order_relaxed);
1031 }
1032
1033 [[nodiscard]] ScopedCoroMutexAwaiter scoped_lock(
1034 trantor::EventLoop *loop =
1035 trantor::EventLoop::getEventLoopOfCurrentThread()) noexcept
1036 {
1037 return ScopedCoroMutexAwaiter(*this, loop);
1038 }
1039
1040 [[nodiscard]] CoroMutexAwaiter lock(
1041 trantor::EventLoop *loop =
1042 trantor::EventLoop::getEventLoopOfCurrentThread()) noexcept
1043 {
1044 return CoroMutexAwaiter(*this, loop);
1045 }
1046
1047 void unlock() noexcept
1048 {
1049 assert(state_.load(std::memory_order_relaxed) != unlockedValue());
1050 auto *waitersHead = waiters_;
1051 if (waitersHead == nullptr)
1052 {
1053 void *currentState = state_.load(std::memory_order_relaxed);
1054 if (currentState == nullptr)
1055 {
1056 const bool releasedLock =
1057 state_.compare_exchange_strong(currentState,
1058 unlockedValue(),
1059 std::memory_order_release,
1060 std::memory_order_relaxed);
1061 if (releasedLock)
1062 {
1063 return;
1064 }
1065 }
1066 currentState = state_.exchange(nullptr, std::memory_order_acquire);
1067 assert(currentState != unlockedValue());
1068 assert(currentState != nullptr);
1069 auto *waiter = static_cast<CoroMutexAwaiter *>(currentState);
1070 do
1071 {
1072 auto *temp = waiter->next_;
1073 waiter->next_ = waitersHead;
1074 waitersHead = waiter;
1075 waiter = temp;
1076 } while (waiter != nullptr);
1077 }
1078 assert(waitersHead != nullptr);
1079 waiters_ = waitersHead->next_;
1080 if (waitersHead->loop_)
1081 {
1082 auto handle = waitersHead->handle_;
1083 waitersHead->loop_->runInLoop([handle] { handle.resume(); });
1084 }
1085 else
1086 {
1087 waitersHead->handle_.resume();
1088 }
1089 }
1090
1091 private:
1092 class CoroMutexAwaiter
1093 {
1094 public:
1095 CoroMutexAwaiter(Mutex &mutex, trantor::EventLoop *loop) noexcept
1096 : mutex_(mutex), loop_(loop)
1097 {
1098 }
1099
1100 bool await_ready() noexcept
1101 {
1102 return mutex_.try_lock();
1103 }
1104
1105 bool await_suspend(std::coroutine_handle<> handle) noexcept
1106 {
1107 handle_ = handle;
1108 return mutex_.asynclockImpl(this);
1109 }
1110
1111 void await_resume() noexcept
1112 {
1113 }
1114
1115 private:
1116 friend class Mutex;
1117
1118 Mutex &mutex_;
1119 trantor::EventLoop *loop_;
1120 std::coroutine_handle<> handle_;
1121 CoroMutexAwaiter *next_;
1122 };
1123
1124 class ScopedCoroMutexAwaiter : public CoroMutexAwaiter
1125 {
1126 public:
1127 ScopedCoroMutexAwaiter(Mutex &mutex, trantor::EventLoop *loop)
1128 : CoroMutexAwaiter(mutex, loop)
1129 {
1130 }
1131
1132 [[nodiscard]] auto await_resume() noexcept
1133 {
1134 return std::unique_lock<Mutex>{mutex_, std::adopt_lock};
1135 }
1136 };
1137
1138 bool asynclockImpl(CoroMutexAwaiter *awaiter)
1139 {
1140 void *oldValue = state_.load(std::memory_order_relaxed);
1141 while (true)
1142 {
1143 if (oldValue == unlockedValue())
1144 {
1145 void *newValue = nullptr;
1146 if (state_.compare_exchange_weak(oldValue,
1147 newValue,
1148 std::memory_order_acquire,
1149 std::memory_order_relaxed))
1150 {
1151 return false;
1152 }
1153 }
1154 else
1155 {
1156 void *newValue = awaiter;
1157 awaiter->next_ = static_cast<CoroMutexAwaiter *>(oldValue);
1158 if (state_.compare_exchange_weak(oldValue,
1159 newValue,
1160 std::memory_order_release,
1161 std::memory_order_relaxed))
1162 {
1163 return true;
1164 }
1165 }
1166 }
1167 }
1168
1169 void *unlockedValue() noexcept
1170 {
1171 return this;
1172 }
1173
1174 std::atomic<void *> state_;
1175 CoroMutexAwaiter *waiters_;
1176};
1177
1178template <typename... Tasks>
1179internal::WhenAllAwaiter<Tasks...> when_all(Tasks... tasks)
1180{
1181 return internal::WhenAllAwaiter<Tasks...>(std::move(tasks)...);
1182}
1183
1184template <typename T>
1185internal::WhenAllAwaiter<std::vector<Task<T>>> when_all(
1186 std::vector<Task<T>> tasks)
1187{
1188 return internal::WhenAllAwaiter(std::move(tasks));
1189}
1190
1192 std::vector<Task<void>> tasks)
1193{
1194 return internal::WhenAllAwaiter(std::move(tasks));
1195}
1196
1197} // namespace drogon
Drogon Test is a minimal effort test framework developed because the major C++ test frameworks doesn'...
Definition Attribute.h:23
void async_run(Coro &&coro)
Runs a coroutine from a regular function.
Definition coroutine.h:749
std::function< void()> async_func(Coro &&coro)
returns a function that calls a coroutine
Definition coroutine.h:769
STL namespace.
Definition coroutine.h:386
Definition coroutine.h:355
Definition coroutine.h:421
Definition coroutine.h:198
Definition coroutine.h:302
Definition coroutine.h:161
Definition coroutine.h:66
An awaiter for Task::promise_type::final_suspend(). Transfer execution back to the coroutine who is c...
Definition coroutine.h:96
Definition coroutine.h:669
Definition coroutine.h:780
Definition coroutine.h:614
Definition coroutine.h:654
Definition coroutine.h:591
Definition coroutine.h:820
Definition coroutine.h:76
Definition coroutine.h:725
Convert Task to an awaiter when it is co_awaited. Following things will happen:
Definition coroutine.h:123