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 "foundation/log.h"
24 #include "foundation/osal/thread/condition_variable.h"
25 #include "foundation/osal/thread/mutex.h"
26 #include "foundation/osal/thread/scoped_lock.h"
27 #include "foundation/utils/constants.h"
28 
29 namespace OHOS {
30 namespace Media {
31 template <typename T>
32 class BlockingQueue {
33 public:
34     explicit BlockingQueue(std::string name, size_t capacity = DEFAULT_QUEUE_SIZE)
35         : name_(std::move(name)), capacity_(capacity), isActive(true)
36     {
37     }
38     ~BlockingQueue() = default;
Size()39     size_t Size()
40     {
41         OSAL::ScopedLock lock(mutex_);
42         return que_.size();
43     }
Capacity()44     size_t Capacity()
45     {
46         return capacity_;
47     }
Empty()48     size_t Empty()
49     {
50         OSAL::ScopedLock lock(mutex_);
51         return que_.empty();
52     }
Push(const T & value)53     bool Push(const T& value)
54     {
55         OSAL::ScopedLock 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         OSAL::ScopedLock 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         OSAL::ScopedLock 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         OSAL::ScopedLock 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         OSAL::ScopedLock lock(mutex_);
136         ClearUnprotected();
137     }
138     void SetActive(bool active, bool cleanData = true)
139     {
140         OSAL::ScopedLock 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 
151 private:
ClearUnprotected()152     void ClearUnprotected()
153     {
154         if (que_.empty()) {
155             return;
156         }
157         bool needNotify = que_.size() == capacity_;
158         std::queue<T>().swap(que_);
159         if (needNotify) {
160             cvFull_.NotifyOne();
161         }
162     }
163 
164     OSAL::Mutex mutex_;
165     OSAL::ConditionVariable cvFull_;
166     OSAL::ConditionVariable cvEmpty_;
167 
168     std::string name_;
169     std::queue<T> que_;
170     const size_t capacity_;
171     std::atomic<bool> isActive;
172 };
173 } // namespace Media
174 } // namespace OHOS
175 #endif // HISTREAMER_FOUNDATION_BLOCKING_QUEUE_H
176