1 /*
2  * Copyright (c) 2023-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 HISTREAMER_RING_BUFFER_H
17 #define HISTREAMER_RING_BUFFER_H
18 
19 #include <atomic>
20 #include <memory>
21 #include "cpp_ext/memory_ext.h"
22 #include "common/log.h"
23 #include "osal/task/condition_variable.h"
24 #include "osal/task/mutex.h"
25 #include "osal/task/autolock.h"
26 #include "securec.h"
27 
28 namespace OHOS {
29 namespace Media {
30 class RingBuffer {
31 public:
RingBuffer(size_t bufferSize)32     explicit RingBuffer(size_t bufferSize) : bufferSize_(bufferSize)
33     {
34     }
35 
36     ~RingBuffer() = default;
37 
Init()38     bool Init()
39     {
40         buffer_ = CppExt::make_unique<uint8_t[]>(bufferSize_);
41         return buffer_ != nullptr;
42     }
43 
44     size_t ReadBuffer(void* ptr, size_t readSize, int waitTimes = 0)
45     {
46         AutoLock lck(writeMutex_);
47         if (!isActive_ || !isReadBlockingAllowed_) {
48             return 0;
49         }
50         auto available = tail_ - head_;
51         while (waitTimes > 0 && available == 0) {
52             MEDIA_LOG_DD("ReadBuffer wait , waitTimes is " PUBLIC_LOG_U64, waitTimes);
53             writeCondition_.Wait(lck);
54             if (!isActive_ || !isReadBlockingAllowed_) {
55                 return 0;
56             }
57             available = tail_ - head_;
58             waitTimes--;
59         }
60         available = (available > readSize) ? readSize : available;
61         size_t index = head_ % bufferSize_;
62         if (index + available < bufferSize_) {
63             (void)memcpy_s(ptr, available, buffer_.get() + index, available);
64         } else {
65             (void)memcpy_s(ptr, bufferSize_ - index, buffer_.get() + index, bufferSize_ - index);
66             (void)memcpy_s(((uint8_t*)ptr) + (bufferSize_ - index), available - (bufferSize_ - index), buffer_.get(),
67                            available - (bufferSize_ - index));
68         }
69         head_ += available;
70         mediaOffset_ += available;
71         MEDIA_LOG_DD("ReadBuffer finish available is " PUBLIC_LOG_ZU ", mediaOffset_ " PUBLIC_LOG_U64, available,
72             mediaOffset_);
73         writeCondition_.NotifyAll();
74         return available;
75     }
76 
WriteBuffer(void * ptr,size_t writeSize)77     bool WriteBuffer(void* ptr, size_t writeSize)
78     {
79         AutoLock lck(writeMutex_);
80         if (!isActive_) {
81             return false;
82         }
83         while (writeSize + tail_ > head_ + bufferSize_) {
84             MEDIA_LOG_DD("WriteBuffer wait writeSize is " PUBLIC_LOG_U64, writeSize);
85             writeCondition_.Wait(lck);
86             if (!isActive_) {
87                 return false;
88             }
89         }
90         size_t index = tail_ % bufferSize_;
91         if (index + writeSize < bufferSize_) {
92             (void)memcpy_s(buffer_.get() + index, writeSize, ptr, writeSize);
93         } else {
94             (void)memcpy_s(buffer_.get() + index, bufferSize_ - index, ptr, bufferSize_ - index);
95             (void)memcpy_s(buffer_.get(), writeSize - (bufferSize_ - index), ((uint8_t*)ptr) + bufferSize_ - index,
96                            writeSize - (bufferSize_ - index));
97         }
98         tail_ += writeSize;
99         writeCondition_.NotifyAll();
100         return true;
101     }
102 
103     void SetActive(bool active, bool cleanData = true)
104     {
105         AutoLock lck(writeMutex_);
106         isActive_ = active;
107         if (!active) {
108             if (cleanData) {
109                 head_ = 0;
110                 tail_ = 0;
111             }
112             writeCondition_.NotifyAll();
113         }
114     }
115 
SetReadBlocking(bool isReadBlockingAllowed)116     void SetReadBlocking(bool isReadBlockingAllowed)
117     {
118         {
119             AutoLock lck(writeMutex_);
120             isReadBlockingAllowed_ = isReadBlockingAllowed;
121         }
122         writeCondition_.NotifyAll();
123     }
124 
GetSize()125     size_t GetSize()
126     {
127         return (tail_ - head_);
128     }
129 
GetHead()130     inline size_t GetHead()
131     {
132         return head_;
133     }
134 
GetTail()135     inline size_t GetTail()
136     {
137         return tail_;
138     }
139 
GetMediaOffset()140     uint64_t GetMediaOffset()
141     {
142         return mediaOffset_;
143     }
144 
SetMediaOffset(uint64_t offset)145     void SetMediaOffset(uint64_t offset)
146     {
147         mediaOffset_ = offset;
148     }
149 
Clear()150     void Clear()
151     {
152         AutoLock lck(writeMutex_);
153         head_ = 0;
154         tail_ = 0;
155         writeCondition_.NotifyAll();
156     }
157 
SetTail(size_t newTail)158     void SetTail(size_t newTail)
159     {
160         {
161             AutoLock lck(writeMutex_);
162             MEDIA_LOG_I("SetTail: current tail " PUBLIC_LOG_ZU ", to tail " PUBLIC_LOG_ZU, tail_, newTail);
163             if (newTail >= 0 && newTail >= head_) {
164                 tail_ = newTail;
165             }
166         }
167         writeCondition_.NotifyAll();
168     }
169 
Seek(uint64_t offset)170     bool Seek(uint64_t offset)
171     {
172         AutoLock lck(writeMutex_);
173         MEDIA_LOG_I("Seek: buffer size " PUBLIC_LOG_ZU ", offset " PUBLIC_LOG_U64
174                     ", mediaOffset_ " PUBLIC_LOG_U64, GetSize(), offset, mediaOffset_);
175         bool result = false;
176         // case1: seek forward success without dropping data already downloaded
177         if (offset >= mediaOffset_ && (offset - mediaOffset_ < GetSize())) {
178             head_ += offset - mediaOffset_;
179             mediaOffset_ = offset;
180             result = true;
181         } else if (offset < mediaOffset_ &&
182             (mediaOffset_ - offset <= bufferSize_ - GetSize())) { // case2: seek backward
183             size_t minPosition = tail_ > bufferSize_ ? tail_ - bufferSize_ : 0;
184             size_t maxInterval = head_ - minPosition;
185             size_t interval = static_cast<size_t>(mediaOffset_ - offset);
186             // Seek backward success without dropping data already downloaded
187             if (interval <= maxInterval) {
188                 MEDIA_LOG_I("Seek backward success, size:" PUBLIC_LOG_ZU ", head:" PUBLIC_LOG_ZU ", tail:" PUBLIC_LOG_ZU
189                     ", minPosition:" PUBLIC_LOG_ZU ", maxInterval:" PUBLIC_LOG_ZU ", interval:" PUBLIC_LOG_ZU
190                     ", target offset:" PUBLIC_LOG_U64 ", current offset:" PUBLIC_LOG_U64,
191                     GetSize(), head_, tail_, minPosition, maxInterval, interval, offset, mediaOffset_);
192                 head_ -= interval;
193                 mediaOffset_ = offset;
194                 result = true;
195             }
196         }
197         writeCondition_.NotifyAll();
198         return result;
199     }
200 
SetHead(size_t newHead)201     bool SetHead(size_t newHead)
202     {
203         bool result = false;
204         {
205             AutoLock lck(writeMutex_);
206             MEDIA_LOG_I("SetHead: current head " PUBLIC_LOG_ZU ", to head " PUBLIC_LOG_ZU, head_, newHead);
207             if (newHead >= head_ && newHead <= tail_) {
208                 mediaOffset_ += (newHead - head_);
209                 head_ = newHead;
210                 result = true;
211             }
212         }
213         writeCondition_.NotifyAll();
214         return result;
215     }
216 private:
217     static constexpr OHOS::HiviewDFX::HiLogLabel LABEL = { LOG_CORE, LOG_DOMAIN_FOUNDATION, "RingBuffer" };
218     const size_t bufferSize_;
219     std::unique_ptr<uint8_t[]> buffer_;
220     size_t head_ {0}; // head
221     size_t tail_ {0}; // tail
222     Mutex writeMutex_ {};
223     ConditionVariable writeCondition_ {};
224     bool isActive_ {true};
225     uint64_t mediaOffset_ {0};
226     bool isReadBlockingAllowed_ {true};
227 };
228 } // namespace Media
229 } // namespace OHOS
230 
231 #endif // HISTREAMER_RING_BUFFER_H
232