1 /* 2 * Copyright (c) 2023 Huawei Device Co., Ltd. 3 * Licensed under the Apache License, Version 2.0 (the "License"); 4 * you may not use this file except in compliance with the License. 5 * You may obtain a copy of the License at 6 * 7 * http://www.apache.org/licenses/LICENSE-2.0 8 * 9 * Unless required by applicable law or agreed to in writing, software 10 * distributed under the License is distributed on an "AS IS" BASIS, 11 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 12 * See the License for the specific language governing permissions and 13 * limitations under the License. 14 */ 15 16 #ifndef __WAITQUEUE_H__ 17 #define __WAITQUEUE_H__ 18 #include "sync/sync.h" 19 #include "cpp/mutex.h" 20 #include "sched/execute_ctx.h" 21 #include "util/IntrusiveList.h" 22 #include "sync/mutex_private.h" 23 #include "dfx/log/ffrt_log_api.h" 24 25 namespace ffrt { 26 class TaskBase; 27 struct TaskWithNode; 28 using TaskListNode = ListNode; 29 using TaskList = List<TaskWithNode, TaskListNode>; 30 31 struct TaskWithNode : public TaskListNode { 32 TaskWithNode(); 33 TaskBase* task = nullptr; 34 std::mutex lk; 35 std::condition_variable cv; 36 }; 37 38 class WaitQueue { 39 public: 40 using TimePoint = std::chrono::steady_clock::time_point; 41 void SuspendAndWait(mutexPrivate* lk); 42 int SuspendAndWaitUntil(mutexPrivate* lk, const TimePoint& tp) noexcept; NotifyAll()43 void NotifyAll() noexcept { Notify(false); } NotifyOne()44 void NotifyOne() noexcept { Notify(true); } 45 WaitQueue()46 WaitQueue() 47 { 48 whead = new WaitUntilEntry(); 49 whead->next = whead; 50 whead->prev = whead; 51 } 52 WaitQueue(WaitQueue const&) = delete; 53 void operator=(WaitQueue const&) = delete; 54 ~WaitQueue()55 ~WaitQueue() 56 { 57 std::lock_guard lg(wqlock); 58 ReleaseAll(); 59 delete whead; 60 whead = nullptr; 61 } 62 63 private: 64 fast_mutex wqlock; 65 WaitUntilEntry* whead; 66 67 private: 68 void WeNotifyProc(WaitUntilEntry* we); 69 void ThreadWait(WaitUntilEntry* wn, mutexPrivate* lk, TaskBase* task); 70 bool ThreadWaitUntil(WaitUntilEntry* wn, mutexPrivate* lk, const TimePoint& tp, TaskBase* task); 71 void Notify(bool one) noexcept; 72 empty()73 inline bool empty() const 74 { 75 if (whead == nullptr) { 76 return true; 77 } 78 return (whead->next == whead); 79 } 80 ReleaseAll()81 void ReleaseAll() 82 { 83 while (!empty()) { 84 FFRT_LOGE("There are still tasks in cv that have not been awakened"); 85 WaitUntilEntry *wue = pop_front(); 86 WeNotifyProc(wue); 87 } 88 } 89 push_back(WaitUntilEntry * we)90 inline void push_back(WaitUntilEntry* we) 91 { 92 if ((we == nullptr) || (whead == nullptr) || (whead->prev == nullptr)) { 93 FFRT_LOGE("we or whead or whead->prev is nullptr"); 94 return; 95 } 96 we->next = whead; 97 we->prev = whead->prev; 98 whead->prev->next = we; 99 whead->prev = we; 100 } 101 pop_front()102 inline WaitUntilEntry* pop_front() 103 { 104 if ((whead->next == nullptr) || (whead->next->next == nullptr)) { 105 FFRT_LOGE("whead->next or whead->next->next is nullptr"); 106 return nullptr; 107 } 108 WaitEntry *we = whead->next; 109 whead->next = we->next; 110 we->next->prev = whead; 111 we->next = nullptr; 112 we->prev = nullptr; 113 return static_cast<WaitUntilEntry*>(we); 114 } 115 remove(WaitUntilEntry * we)116 inline void remove(WaitUntilEntry* we) 117 { 118 if ((we->next == nullptr) || (we->prev == nullptr)) { 119 return; 120 } 121 we->prev->next = we->next; 122 we->next->prev = we->prev; 123 we->next = nullptr; 124 we->prev = nullptr; 125 return; 126 } 127 friend bool WeTimeoutProc(WaitQueue* wq, WaitUntilEntry* wue); 128 }; 129 } // namespace ffrt 130 #endif // _WAITQUEUE_H_ 131