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_RING_BUFFER_H
17 #define HISTREAMER_RING_BUFFER_H
18 
19 #include <atomic>
20 #include <memory>
21 #include "foundation/cpp_ext/memory_ext.h"
22 #include "foundation/log.h"
23 #include "foundation/osal/thread/condition_variable.h"
24 #include "foundation/osal/thread/mutex.h"
25 #include "foundation/osal/thread/scoped_lock.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         OSAL::ScopedLock lck(writeMutex_);
47         if (!isActive_) {
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_) {
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_.NotifyOne();
74         return available;
75     }
76 
WriteBuffer(void * ptr,size_t writeSize)77     bool WriteBuffer(void* ptr, size_t writeSize)
78     {
79         OSAL::ScopedLock 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_.NotifyOne();
100         return true;
101     }
102 
103     void SetActive(bool active, bool cleanData = true)
104     {
105         OSAL::ScopedLock lck(writeMutex_);
106         isActive_ = active;
107         if (!active) {
108             if (cleanData) {
109                 head_ = 0;
110                 tail_ = 0;
111             }
112             writeCondition_.NotifyOne();
113         }
114     }
115 
GetSize()116     size_t GetSize()
117     {
118         return (tail_ - head_);
119     }
120 
GetMediaOffset()121     uint64_t GetMediaOffset()
122     {
123         return mediaOffset_;
124     }
125 
SetMediaOffset(uint64_t offset)126     void SetMediaOffset(uint64_t offset)
127     {
128         mediaOffset_ = offset;
129     }
130 
Clear()131     void Clear()
132     {
133         OSAL::ScopedLock lck(writeMutex_);
134         head_ = 0;
135         tail_ = 0;
136         writeCondition_.NotifyOne();
137     }
138 
Seek(uint64_t offset)139     bool Seek(uint64_t offset)
140     {
141         OSAL::ScopedLock lck(writeMutex_);
142         MEDIA_LOG_I("Seek: buffer size " PUBLIC_LOG_ZU ", offset " PUBLIC_LOG_U64
143                     ", mediaOffset_ " PUBLIC_LOG_U64, GetSize(), offset, mediaOffset_);
144         bool result = false;
145         if (offset >= mediaOffset_ && offset - mediaOffset_ < GetSize()) {
146             head_ += offset - mediaOffset_;
147             result = true;
148         }
149         writeCondition_.NotifyOne();
150         return result;
151     }
152 private:
153     const size_t bufferSize_;
154     std::unique_ptr<uint8_t[]> buffer_;
155     size_t head_ {0}; // head
156     size_t tail_ {0}; // tail
157     OSAL::Mutex writeMutex_ {};
158     OSAL::ConditionVariable writeCondition_ {};
159     bool isActive_ {true};
160     uint64_t mediaOffset_ {0};
161 };
162 } // namespace Media
163 } // namespace OHOS
164 
165 #endif // HISTREAMER_RING_BUFFER_H
166