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 GetMediaOffset()130 uint64_t GetMediaOffset() 131 { 132 return mediaOffset_; 133 } 134 SetMediaOffset(uint64_t offset)135 void SetMediaOffset(uint64_t offset) 136 { 137 mediaOffset_ = offset; 138 } 139 Clear()140 void Clear() 141 { 142 AutoLock lck(writeMutex_); 143 head_ = 0; 144 tail_ = 0; 145 writeCondition_.NotifyAll(); 146 } 147 Seek(uint64_t offset)148 bool Seek(uint64_t offset) 149 { 150 AutoLock lck(writeMutex_); 151 MEDIA_LOG_I("Seek: buffer size " PUBLIC_LOG_ZU ", offset " PUBLIC_LOG_U64 152 ", mediaOffset_ " PUBLIC_LOG_U64, GetSize(), offset, mediaOffset_); 153 bool result = false; 154 if (offset >= mediaOffset_ && offset - mediaOffset_ < GetSize()) { 155 head_ += offset - mediaOffset_; 156 mediaOffset_ = offset; 157 result = true; 158 } 159 writeCondition_.NotifyAll(); 160 return result; 161 } 162 private: 163 const size_t bufferSize_; 164 std::unique_ptr<uint8_t[]> buffer_; 165 size_t head_ {0}; // head 166 size_t tail_ {0}; // tail 167 Mutex writeMutex_ {}; 168 ConditionVariable writeCondition_ {}; 169 bool isActive_ {true}; 170 uint64_t mediaOffset_ {0}; 171 bool isReadBlockingAllowed_ {true}; 172 }; 173 } // namespace Media 174 } // namespace OHOS 175 176 #endif // HISTREAMER_RING_BUFFER_H 177