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 "constants.h" 23 #include "foundation/log.h" 24 #include "osal/thread/condition_variable.h" 25 #include "osal/thread/mutex.h" 26 #include "osal/thread/scoped_lock.h" 27 28 namespace OHOS { 29 namespace Media { 30 template <typename T> 31 class BlockingQueue { 32 public: 33 explicit BlockingQueue(const std::string& name, size_t capacity = DEFAULT_QUEUE_SIZE) name_(name)34 : name_(name), capacity_(capacity), isActive(true) 35 { 36 } 37 ~BlockingQueue() = default; Size()38 size_t Size() 39 { 40 OSAL::ScopedLock lock(mutex_); 41 return que_.size(); 42 } Capacity()43 size_t Capacity() 44 { 45 return capacity_; 46 } Empty()47 size_t Empty() 48 { 49 OSAL::ScopedLock lock(mutex_); 50 return que_.empty(); 51 } Push(const T & value)52 bool Push(const T& value) 53 { 54 OSAL::ScopedLock lock(mutex_); 55 if (!isActive) { 56 MEDIA_LOG_D("blocking queue %" PUBLIC_LOG "s is inactive for Push.", name_.c_str()); 57 return false; 58 } 59 if (que_.size() >= capacity_) { 60 MEDIA_LOG_D("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 "d, 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_D("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 OSAL::ScopedLock 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 "d, isFull: %" PUBLIC_LOG "d", 86 isActive, 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_D("blocking queue %" PUBLIC_LOG "s Pop enter.", name_.c_str()); 96 OSAL::ScopedLock 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_D("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_D("blocking queue %" PUBLIC_LOG "s Pop succeed.", name_.c_str()); 112 return el; 113 } Pop(int timeoutMs)114 T Pop(int timeoutMs) 115 { 116 OSAL::ScopedLock 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 OSAL::ScopedLock lock(mutex_); 135 ClearUnprotected(); 136 } SetActive(bool active)137 void SetActive(bool active) 138 { 139 OSAL::ScopedLock lock(mutex_); 140 MEDIA_LOG_D("SetActive for %" PUBLIC_LOG "s: %" PUBLIC_LOG "d.", name_.c_str(), active); 141 isActive = active; 142 if (!active) { 143 ClearUnprotected(); 144 cvEmpty_.NotifyOne(); 145 } 146 } 147 148 private: ClearUnprotected()149 void ClearUnprotected() 150 { 151 if (que_.empty()) { 152 return; 153 } 154 bool needNotify = que_.size() == capacity_; 155 std::queue<T>().swap(que_); 156 if (needNotify) { 157 cvFull_.NotifyOne(); 158 } 159 } 160 161 OSAL::Mutex mutex_; 162 OSAL::ConditionVariable cvFull_; 163 OSAL::ConditionVariable cvEmpty_; 164 165 std::string name_; 166 std::queue<T> que_; 167 const size_t capacity_; 168 std::atomic<bool> isActive; 169 }; 170 } // namespace Media 171 } // namespace OHOS 172 #endif // HISTREAMER_FOUNDATION_BLOCKING_QUEUE_H 173