1 /* 2 * Copyright (c) 2021-2021 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 HISTREAMER_FOUNDATION_BLOCKING_QUEUE_H 17 #define HISTREAMER_FOUNDATION_BLOCKING_QUEUE_H 18 19 #include <atomic> 20 #include <queue> 21 #include <string> 22 #include <utility> 23 #include "common/log.h" 24 #include "osal/task/condition_variable.h" 25 #include "osal/task/mutex.h" 26 #include "osal/task/autolock.h" 27 28 namespace OHOS { 29 namespace Media { 30 template <typename T> 31 class BlockingQueue { 32 public: 33 explicit BlockingQueue(std::string name, size_t capacity = 10) // 10 means default queue size name_(std::move (name))34 : name_(std::move(name)), capacity_(capacity), isActive(true) 35 { 36 } 37 ~BlockingQueue() = default; Size()38 size_t Size() 39 { 40 AutoLock lock(mutex_); 41 return que_.size(); 42 } Capacity()43 size_t Capacity() 44 { 45 return capacity_; 46 } Empty()47 bool Empty() 48 { 49 AutoLock lock(mutex_); 50 return que_.empty(); 51 } Push(const T & value)52 bool Push(const T& value) 53 { 54 AutoLock lock(mutex_); 55 if (!isActive) { 56 MEDIA_LOG_DD("blocking queue " PUBLIC_LOG_S " is inactive for Push.", name_.c_str()); 57 return false; 58 } 59 if (que_.size() >= capacity_) { 60 MEDIA_LOG_DD("blocking queue " PUBLIC_LOG_S " is full, waiting for pop.", name_.c_str()); 61 cvFull_.Wait(lock, [this] { return !isActive || que_.size() < capacity_; }); 62 } 63 if (!isActive) { 64 MEDIA_LOG_D("blocking queue " PUBLIC_LOG_S ": inactive: " PUBLIC_LOG_D32 ", isFull: " PUBLIC_LOG 65 "d", name_.c_str(), isActive.load(), que_.size() < capacity_); 66 return false; 67 } 68 que_.push(value); 69 cvEmpty_.NotifyAll(); 70 MEDIA_LOG_DD("blocking queue " PUBLIC_LOG_S " Push succeed.", name_.c_str()); 71 return true; 72 } Push(const T & value,int timeoutMs)73 bool Push(const T& value, int timeoutMs) 74 { 75 AutoLock lock(mutex_); 76 if (!isActive) { 77 MEDIA_LOG_D("blocking queue " PUBLIC_LOG_S " is inactive for Push.", name_.c_str()); 78 return false; 79 } 80 if (que_.size() >= capacity_) { 81 MEDIA_LOG_D("blocking queue is full, waiting for pop..."); 82 cvFull_.WaitFor(lock, timeoutMs, [this] { return !isActive || que_.size() < capacity_; }); 83 } 84 if (!isActive || (que_.size() == capacity_)) { 85 MEDIA_LOG_D("blocking queue: inactive: " PUBLIC_LOG_D32 ", isFull: " PUBLIC_LOG_D32, 86 isActive.load(), que_.size() < capacity_); 87 return false; 88 } 89 que_.push(value); 90 cvEmpty_.NotifyAll(); 91 return true; 92 } Pop()93 T Pop() 94 { 95 MEDIA_LOG_DD("blocking queue " PUBLIC_LOG_S " Pop enter.", name_.c_str()); 96 AutoLock lock(mutex_); 97 if (!isActive) { 98 MEDIA_LOG_D("blocking queue " PUBLIC_LOG_S " is inactive.", name_.c_str()); 99 return {}; 100 } 101 if (que_.empty()) { 102 MEDIA_LOG_DD("blocking queue " PUBLIC_LOG_S " is empty, waiting for push", name_.c_str()); 103 cvEmpty_.Wait(lock, [this] { return !isActive || !que_.empty(); }); 104 } 105 if (!isActive) { 106 return {}; 107 } 108 T el = que_.front(); 109 que_.pop(); 110 cvFull_.NotifyOne(); 111 MEDIA_LOG_DD("blocking queue " PUBLIC_LOG_S " Pop succeed.", name_.c_str()); 112 return el; 113 } Pop(int timeoutMs)114 T Pop(int timeoutMs) 115 { 116 AutoLock lock(mutex_); 117 if (!isActive) { 118 MEDIA_LOG_D("blocking queue " PUBLIC_LOG_S " is inactive.", name_.c_str()); 119 return {}; 120 } 121 if (que_.empty()) { 122 cvEmpty_.WaitFor(lock, timeoutMs, [this] { return !isActive || !que_.empty(); }); 123 } 124 if (!isActive || que_.empty()) { 125 return {}; 126 } 127 T el = que_.front(); 128 que_.pop(); 129 cvFull_.NotifyOne(); 130 return el; 131 } Clear()132 void Clear() 133 { 134 AutoLock lock(mutex_); 135 ClearUnprotected(); 136 } 137 void SetActive(bool active, bool cleanData = true) 138 { 139 AutoLock lock(mutex_); 140 MEDIA_LOG_D("SetActive for " PUBLIC_LOG_S ": " PUBLIC_LOG_D32 ".", name_.c_str(), active); 141 isActive = active; 142 if (!active) { 143 if (cleanData) { 144 ClearUnprotected(); 145 } 146 cvEmpty_.NotifyOne(); 147 } 148 } 149 150 private: ClearUnprotected()151 void ClearUnprotected() 152 { 153 if (que_.empty()) { 154 return; 155 } 156 bool needNotify = que_.size() == capacity_; 157 std::queue<T>().swap(que_); 158 if (needNotify) { 159 cvFull_.NotifyOne(); 160 } 161 } 162 163 Mutex mutex_; 164 ConditionVariable cvFull_; 165 ConditionVariable cvEmpty_; 166 167 std::string name_; 168 std::queue<T> que_; 169 const size_t capacity_; 170 std::atomic<bool> isActive; 171 }; 172 } // namespace Media 173 } // namespace OHOS 174 #endif // HISTREAMER_FOUNDATION_BLOCKING_QUEUE_H 175