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
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 __attribute__((no_sanitize("cfi"))) Capacity()
44     {
45         AutoLock lock(mutex_);
46         return capacity_;
47     }
Empty()48     bool Empty()
49     {
50         AutoLock lock(mutex_);
51         return que_.empty();
52     }
Push(const T & value)53     bool Push(const T& value)
54     {
55         AutoLock lock(mutex_);
56         if (!isActive) {
57             MEDIA_LOG_DD("blocking queue " PUBLIC_LOG_S " is inactive for Push.", name_.c_str());
58             return false;
59         }
60         if (que_.size() >= capacity_) {
61             MEDIA_LOG_DD("blocking queue " PUBLIC_LOG_S " is full, waiting for pop.", name_.c_str());
62             cvFull_.Wait(lock, [this] { return !isActive || que_.size() < capacity_; });
63         }
64         if (!isActive) {
65             MEDIA_LOG_D("blocking queue " PUBLIC_LOG_S ": inactive: " PUBLIC_LOG_D32 ", isFull: " PUBLIC_LOG
66                         "d", name_.c_str(), isActive.load(), que_.size() < capacity_);
67             return false;
68         }
69         que_.push(value);
70         cvEmpty_.NotifyAll();
71         MEDIA_LOG_DD("blocking queue " PUBLIC_LOG_S " Push succeed.", name_.c_str());
72         return true;
73     }
Push(const T & value,int timeoutMs)74     bool Push(const T& value, int timeoutMs)
75     {
76         AutoLock lock(mutex_);
77         if (!isActive) {
78             MEDIA_LOG_D("blocking queue " PUBLIC_LOG_S " is inactive for Push.", name_.c_str());
79             return false;
80         }
81         if (que_.size() >= capacity_) {
82             MEDIA_LOG_D("blocking queue is full, waiting for pop...");
83             cvFull_.WaitFor(lock, timeoutMs, [this] { return !isActive || que_.size() < capacity_; });
84         }
85         if (!isActive || (que_.size() == capacity_)) {
86             MEDIA_LOG_D("blocking queue: inactive: " PUBLIC_LOG_D32 ", isFull: " PUBLIC_LOG_D32,
87                         isActive.load(), que_.size() < capacity_);
88             return false;
89         }
90         que_.push(value);
91         cvEmpty_.NotifyAll();
92         return true;
93     }
Pop()94     T Pop()
95     {
96         MEDIA_LOG_DD("blocking queue " PUBLIC_LOG_S " Pop enter.", name_.c_str());
97         AutoLock lock(mutex_);
98         if (!isActive) {
99             MEDIA_LOG_D("blocking queue " PUBLIC_LOG_S " is inactive.", name_.c_str());
100             return {};
101         }
102         if (que_.empty()) {
103             MEDIA_LOG_DD("blocking queue " PUBLIC_LOG_S " is empty, waiting for push", name_.c_str());
104             cvEmpty_.Wait(lock, [this] { return !isActive || !que_.empty(); });
105         }
106         if (!isActive) {
107             return {};
108         }
109         T el = que_.front();
110         que_.pop();
111         cvFull_.NotifyOne();
112         MEDIA_LOG_DD("blocking queue " PUBLIC_LOG_S " Pop succeed.", name_.c_str());
113         return el;
114     }
Pop(int timeoutMs)115     T Pop(int timeoutMs)
116     {
117         AutoLock lock(mutex_);
118         if (!isActive) {
119             MEDIA_LOG_D("blocking queue " PUBLIC_LOG_S " is inactive.", name_.c_str());
120             return {};
121         }
122         if (que_.empty()) {
123             cvEmpty_.WaitFor(lock, timeoutMs, [this] { return !isActive || !que_.empty(); });
124         }
125         if (!isActive || que_.empty()) {
126             return {};
127         }
128         T el = que_.front();
129         que_.pop();
130         cvFull_.NotifyOne();
131         return el;
132     }
Clear()133     void Clear()
134     {
135         AutoLock lock(mutex_);
136         ClearUnprotected();
137     }
138     void SetActive(bool active, bool cleanData = true)
139     {
140         AutoLock lock(mutex_);
141         MEDIA_LOG_D("SetActive for " PUBLIC_LOG_S ": " PUBLIC_LOG_D32 ".", name_.c_str(), active);
142         isActive = active;
143         if (!active) {
144             if (cleanData) {
145                 ClearUnprotected();
146             }
147             cvEmpty_.NotifyOne();
148         }
149     }
150 
ResetCapacity(size_t capacity)151     void ResetCapacity(size_t capacity)
152     {
153         {
154             AutoLock lock(mutex_);
155             capacity_ = capacity;
156         }
157         cvEmpty_.NotifyAll();
158         MEDIA_LOG_D("ResetCapacity: capacity_ is " PUBLIC_LOG_ZU, capacity_);
159     }
160 
161 private:
162     static constexpr OHOS::HiviewDFX::HiLogLabel LABEL = { LOG_CORE, LOG_DOMAIN_FOUNDATION, "BlockingQueue" };
ClearUnprotected()163     void ClearUnprotected()
164     {
165         if (que_.empty()) {
166             return;
167         }
168         bool needNotify = que_.size() == capacity_;
169         std::queue<T>().swap(que_);
170         if (needNotify) {
171             cvFull_.NotifyOne();
172         }
173     }
174 
175     Mutex mutex_;
176     ConditionVariable cvFull_;
177     ConditionVariable cvEmpty_;
178 
179     std::string name_;
180     std::queue<T> que_;
181     size_t capacity_;
182     std::atomic<bool> isActive;
183 };
184 } // namespace Media
185 } // namespace OHOS
186 #endif // HISTREAMER_FOUNDATION_BLOCKING_QUEUE_H
187