1 /*
2 * Copyright (c) 2024-2024 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 #include <mutex>
17 #include <list>
18 #include <algorithm>
19 #include <cassert>
20 #include <limits>
21 #include <securec.h>
22 #include "media_cached_buffer.h"
23 #include "common/log.h"
24 #include "avcodec_log.h"
25 #include "avcodec_errors.h"
26
27 namespace OHOS {
28 namespace Media {
29 constexpr size_t CACHE_FRAGMENT_MAX_NUM_DEFAULT = 300; // Maximum number of fragment nodes
30 constexpr size_t CACHE_FRAGMENT_MAX_NUM_LARGE = 10; // Maximum number of fragment nodes
31 constexpr size_t CACHE_FRAGMENT_MIN_NUM_DEFAULT = 3; // Minimum number of fragment nodes
32 constexpr double NEW_FRAGMENT_INIT_CHUNK_NUM = 128.0; // Restricting the cache size of seek operation, 128 = 2MB
33 constexpr double NEW_FRAGMENT_NIT_DEFAULT_DENOMINATOR = 0.25;
34 constexpr double CACHE_RELEASE_FACTOR_DEFAULT = 10;
35 constexpr double TO_PERCENT = 100;
36 constexpr int64_t MAX_TOTAL_READ_SIZE = 2000000;
37 constexpr int64_t UP_LIMIT_MAX_TOTAL_READ_SIZE = 3000000;
38 constexpr int64_t ACCESS_OFFSET_MAX_LENGTH = 2 * 1024;
39
BoundedIntervalComp(int64_t mid,uint64_t start,int64_t end)40 inline constexpr bool BoundedIntervalComp(int64_t mid, uint64_t start, int64_t end)
41 {
42 return (static_cast<int64_t>(start) <= mid && mid <= end);
43 }
44
LeftBoundedRightOpenComp(int64_t mid,uint64_t start,int64_t end)45 inline constexpr bool LeftBoundedRightOpenComp(int64_t mid, uint64_t start, int64_t end)
46 {
47 return (static_cast<int64_t>(start) <= mid && mid < end);
48 }
49
IncreaseStep(uint8_t * & src,uint64_t & offset,size_t & writeSize,size_t step)50 inline void IncreaseStep(uint8_t*& src, uint64_t& offset, size_t& writeSize, size_t step)
51 {
52 src += step;
53 offset += static_cast<uint64_t>(step);
54 writeSize += step;
55 }
56
InitChunkInfo(CacheChunk & chunkInfo,uint64_t offset)57 inline void InitChunkInfo(CacheChunk& chunkInfo, uint64_t offset)
58 {
59 chunkInfo.offset = offset;
60 chunkInfo.dataLength = 0;
61 }
62
CacheMediaChunkBufferImpl()63 CacheMediaChunkBufferImpl::CacheMediaChunkBufferImpl()
64 : totalBuffSize_(0), totalReadSize_(0), chunkMaxNum_(0), chunkSize_(0), bufferAddr_(nullptr),
65 fragmentMaxNum_(CACHE_FRAGMENT_MAX_NUM_DEFAULT),
66 lruCache_(CACHE_FRAGMENT_MAX_NUM_DEFAULT) {}
67
~CacheMediaChunkBufferImpl()68 CacheMediaChunkBufferImpl::~CacheMediaChunkBufferImpl()
69 {
70 std::lock_guard lock(mutex_);
71 freeChunks_.clear();
72 fragmentCacheBuffer_.clear();
73 readPos_ = fragmentCacheBuffer_.end();
74 writePos_ = fragmentCacheBuffer_.end();
75 chunkMaxNum_ = 0;
76 totalReadSize_ = 0;
77 if (bufferAddr_ != nullptr) {
78 free(bufferAddr_);
79 bufferAddr_ = nullptr;
80 }
81 }
82
Init(uint64_t totalBuffSize,uint32_t chunkSize)83 bool CacheMediaChunkBufferImpl::Init(uint64_t totalBuffSize, uint32_t chunkSize)
84 {
85 if (isLargeOffsetSpan_) {
86 lruCache_.ReCacheSize(CACHE_FRAGMENT_MAX_NUM_LARGE);
87 } else {
88 lruCache_.ReCacheSize(CACHE_FRAGMENT_MAX_NUM_DEFAULT);
89 }
90 if (totalBuffSize == 0 || chunkSize == 0 || totalBuffSize < chunkSize) {
91 return false;
92 }
93 double newFragmentInitChunkNum = NEW_FRAGMENT_INIT_CHUNK_NUM;
94 uint64_t diff = (totalBuffSize + chunkSize) > 1 ? (totalBuffSize + chunkSize) - 1 : 0;
95 int64_t chunkNum = static_cast<int64_t>(diff / chunkSize) + 1;
96 if ((chunkNum - static_cast<int64_t>(newFragmentInitChunkNum)) < 0 ||
97 chunkNum > MAX_CACHE_BUFFER_SIZE) {
98 return false;
99 }
100 if (newFragmentInitChunkNum > static_cast<double>(chunkNum) * NEW_FRAGMENT_NIT_DEFAULT_DENOMINATOR) {
101 newFragmentInitChunkNum = std::max(1.0, static_cast<double>(chunkNum) * NEW_FRAGMENT_NIT_DEFAULT_DENOMINATOR);
102 }
103 std::lock_guard lock(mutex_);
104 if (bufferAddr_ != nullptr) {
105 return false;
106 }
107 readPos_ = fragmentCacheBuffer_.end();
108 writePos_ = fragmentCacheBuffer_.end();
109 size_t sizePerChunk = sizeof(CacheChunk) + chunkSize;
110 int64_t totalSize = static_cast<int64_t>(sizePerChunk) * chunkNum;
111 if (totalSize <= 0 || totalSize > MAX_CACHE_BUFFER_SIZE * CHUNK_SIZE) {
112 return false;
113 }
114 bufferAddr_ = static_cast<uint8_t*>(malloc(totalSize));
115 if (bufferAddr_ == nullptr) {
116 return false;
117 }
118 uint8_t* temp = bufferAddr_;
119 for (auto i = 0; i < chunkNum; ++i) {
120 auto chunkInfo = reinterpret_cast<CacheChunk*>(temp);
121 chunkInfo->offset = 0;
122 chunkInfo->dataLength = 0;
123 chunkInfo->chunkSize = static_cast<uint32_t>(chunkSize);
124 freeChunks_.push_back(chunkInfo);
125 temp += sizePerChunk;
126 }
127 chunkMaxNum_ = chunkNum >= 1 ? static_cast<uint32_t>(chunkNum) - 1 : 0; // -1
128 totalBuffSize_ = totalBuffSize;
129 chunkSize_ = chunkSize;
130 initReadSizeFactor_ = newFragmentInitChunkNum / (chunkMaxNum_ - newFragmentInitChunkNum);
131 loopInterruptClock_.Reset();
132 return true;
133 }
134
135 // Upadate the chunk read from the fragment
UpdateAccessPos(FragmentIterator & fragmentPos,ChunkIterator & chunkPos,uint64_t offsetChunk)136 void CacheMediaChunkBufferImpl::UpdateAccessPos(FragmentIterator& fragmentPos, ChunkIterator& chunkPos,
137 uint64_t offsetChunk)
138 {
139 if (chunkPos != fragmentPos->chunks.begin() && chunkPos == fragmentPos->chunks.end()) {
140 auto preChunkPos = std::prev(chunkPos);
141 if (((*preChunkPos)->offset + (*preChunkPos)->chunkSize) == offsetChunk) {
142 fragmentPos->accessPos = chunkPos;
143 } else {
144 fragmentPos->accessPos = preChunkPos;
145 }
146 } else if ((*chunkPos)->offset == offsetChunk) {
147 fragmentPos->accessPos = chunkPos;
148 } else {
149 fragmentPos->accessPos = std::prev(chunkPos);
150 }
151 }
152
Read(void * ptr,uint64_t offset,size_t readSize)153 size_t CacheMediaChunkBufferImpl::Read(void* ptr, uint64_t offset, size_t readSize)
154 {
155 std::lock_guard lock(mutex_);
156 size_t hasReadSize = 0;
157 uint8_t* dst = static_cast<uint8_t*>(ptr);
158 uint64_t hasReadOffset = offset;
159 size_t oneReadSize = ReadInner(dst, hasReadOffset, readSize);
160 hasReadSize = oneReadSize;
161 int64_t loopStartTime = loopInterruptClock_.ElapsedSeconds();
162 while (hasReadSize < readSize && oneReadSize != 0) {
163 if (CheckLoopTimeout(loopStartTime)) {
164 break;
165 }
166 dst += oneReadSize;
167 hasReadOffset += static_cast<uint64_t>(oneReadSize);
168 oneReadSize = ReadInner(dst, hasReadOffset, readSize - hasReadSize);
169 hasReadSize += oneReadSize;
170 }
171 return hasReadSize;
172 }
173
ReadInner(void * ptr,uint64_t offset,size_t readSize)174 size_t CacheMediaChunkBufferImpl::ReadInner(void* ptr, uint64_t offset, size_t readSize)
175 {
176 auto fragmentPos = GetOffsetFragmentCache(readPos_, offset, LeftBoundedRightOpenComp);
177 if (readSize == 0 || fragmentPos == fragmentCacheBuffer_.end()) {
178 return 0;
179 }
180 auto chunkPos = fragmentPos->accessPos;
181 if (chunkPos == fragmentPos->chunks.end() ||
182 offset < (*chunkPos)->offset ||
183 offset > (*chunkPos)->offset + (*chunkPos)->dataLength) {
184 chunkPos = GetOffsetChunkCache(fragmentPos->chunks, offset, LeftBoundedRightOpenComp);
185 }
186
187 uint8_t* dst = static_cast<uint8_t*>(ptr);
188 uint64_t offsetChunk = offset;
189 if (chunkPos != fragmentPos->chunks.end()) {
190 uint64_t readOffset = offset > fragmentPos->offsetBegin ? offset - fragmentPos->offsetBegin : 0;
191 uint64_t temp = readOffset > static_cast<uint64_t>(fragmentPos->accessLength) ?
192 readOffset - static_cast<uint64_t>(fragmentPos->accessLength) : 0;
193 if (temp >= ACCESS_OFFSET_MAX_LENGTH) {
194 chunkPos = SplitFragmentCacheBuffer(fragmentPos, offset, chunkPos);
195 }
196 size_t hasReadSize = 0;
197 int64_t loopStartTime = loopInterruptClock_.ElapsedSeconds();
198 while (hasReadSize < readSize && chunkPos != fragmentPos->chunks.end()) {
199 if (CheckLoopTimeout(loopStartTime)) {
200 break;
201 }
202 auto chunkInfo = *chunkPos;
203 uint64_t diff = offsetChunk > chunkInfo->offset ? offsetChunk - chunkInfo->offset : 0;
204 if (offsetChunk < chunkInfo->offset || diff > chunkInfo->dataLength) {
205 DumpAndCheckInner();
206 return 0;
207 }
208 uint64_t readDiff = chunkInfo->dataLength > diff ? chunkInfo->dataLength - diff : 0;
209 auto readOne = std::min(static_cast<size_t>(readDiff), readSize - hasReadSize);
210 errno_t res = memcpy_s(dst + hasReadSize, readOne, (*chunkPos)->data + diff, readOne);
211 FALSE_RETURN_V_MSG_E(res == EOK, 0, "memcpy_s data err");
212 hasReadSize += readOne;
213 offsetChunk += static_cast<uint64_t>(readOne);
214 chunkPos++;
215 }
216 UpdateAccessPos(fragmentPos, chunkPos, offsetChunk);
217 UpdateFragment(fragmentPos, hasReadSize, offsetChunk);
218 return hasReadSize;
219 }
220 return 0;
221 }
222
UpdateFragment(FragmentIterator & fragmentPos,size_t hasReadSize,uint64_t offsetChunk)223 void CacheMediaChunkBufferImpl::UpdateFragment(FragmentIterator& fragmentPos, size_t hasReadSize,
224 uint64_t offsetChunk)
225 {
226 uint64_t lengthDiff = offsetChunk > fragmentPos->offsetBegin ? offsetChunk - fragmentPos->offsetBegin : 0;
227 fragmentPos->accessLength = static_cast<int64_t>(lengthDiff);
228 fragmentPos->readTime = Clock::now();
229 fragmentPos->totalReadSize += hasReadSize;
230 totalReadSize_ += hasReadSize;
231 readPos_ = fragmentPos;
232 lruCache_.Refer(fragmentPos->offsetBegin, fragmentPos);
233 }
234
WriteInPlace(FragmentIterator & fragmentPos,uint8_t * ptr,uint64_t inOffset,size_t inWriteSize,size_t & outWriteSize)235 bool CacheMediaChunkBufferImpl::WriteInPlace(FragmentIterator& fragmentPos, uint8_t* ptr, uint64_t inOffset,
236 size_t inWriteSize, size_t& outWriteSize)
237 {
238 uint64_t offset = inOffset;
239 size_t writeSize = inWriteSize;
240 uint8_t* src = ptr;
241 auto& chunkList = fragmentPos->chunks;
242 outWriteSize = 0;
243 ChunkIterator chunkPos = std::upper_bound(chunkList.begin(), chunkList.end(), offset,
244 [](auto inputOffset, const CacheChunk* chunk) {
245 return (inputOffset <= chunk->offset + chunk->dataLength);
246 });
247 if (chunkPos == chunkList.end()) {
248 DumpInner(0);
249 return false;
250 }
251 size_t writeSizeTmp = 0;
252 auto chunkInfoTmp = *chunkPos;
253 uint64_t accessLengthTmp = inOffset > writePos_->offsetBegin ? inOffset - writePos_->offsetBegin : 0;
254 if (chunkInfoTmp->offset <= offset &&
255 offset < chunkInfoTmp->offset + static_cast<uint64_t>(chunkInfoTmp->dataLength)) {
256 size_t diff = static_cast<size_t>(offset > chunkInfoTmp->offset ? offset - chunkInfoTmp->offset : 0);
257 size_t copyLen = static_cast<size_t>(chunkInfoTmp->dataLength - diff);
258 copyLen = std::min(copyLen, writeSize);
259 errno_t res = memcpy_s(chunkInfoTmp->data + diff, copyLen, src, copyLen);
260 FALSE_RETURN_V_MSG_E(res == EOK, false, "memcpy_s data err");
261 IncreaseStep(src, offset, writeSizeTmp, copyLen);
262 if (writePos_->accessLength > static_cast<int64_t>(accessLengthTmp)) {
263 writePos_->accessPos = chunkPos;
264 writePos_->accessLength = static_cast<int64_t>(accessLengthTmp);
265 }
266 } else if (writePos_->accessLength > static_cast<int64_t>(accessLengthTmp)) {
267 writePos_->accessPos = std::next(chunkPos);
268 writePos_->accessLength = static_cast<int64_t>(accessLengthTmp);
269 }
270 ++chunkPos;
271 int64_t loopStartTime = loopInterruptClock_.ElapsedSeconds();
272 while (writeSizeTmp < writeSize && chunkPos != chunkList.end()) {
273 if (CheckLoopTimeout(loopStartTime)) {
274 break;
275 }
276 chunkInfoTmp = *chunkPos;
277 auto copyLen = std::min(chunkInfoTmp->dataLength, (uint32_t)(writeSize - writeSizeTmp));
278 errno_t res = memcpy_s(chunkInfoTmp->data, copyLen, src, copyLen);
279 FALSE_RETURN_V_MSG_E(res == EOK, false, "memcpy_s data err");
280 IncreaseStep(src, offset, writeSizeTmp, copyLen);
281 ++chunkPos;
282 }
283 outWriteSize = writeSizeTmp;
284 return true;
285 }
286
WriteMergerPre(uint64_t offset,size_t writeSize,FragmentIterator & nextFragmentPos)287 bool CacheMediaChunkBufferImpl::WriteMergerPre(uint64_t offset, size_t writeSize, FragmentIterator& nextFragmentPos)
288 {
289 nextFragmentPos = std::next(writePos_);
290 bool isLoop = true;
291 while (isLoop) {
292 if (nextFragmentPos == fragmentCacheBuffer_.end() ||
293 offset + static_cast<uint64_t>(writeSize) < nextFragmentPos->offsetBegin) {
294 nextFragmentPos = fragmentCacheBuffer_.end();
295 isLoop = false;
296 break;
297 }
298 if (offset + static_cast<uint64_t>(writeSize) <
299 nextFragmentPos->offsetBegin + static_cast<uint64_t>(nextFragmentPos->dataLength)) {
300 auto endPos = GetOffsetChunkCache(nextFragmentPos->chunks,
301 offset + static_cast<uint64_t>(writeSize), LeftBoundedRightOpenComp);
302 freeChunks_.splice(freeChunks_.end(), nextFragmentPos->chunks, nextFragmentPos->chunks.begin(), endPos);
303 if (endPos == nextFragmentPos->chunks.end()) {
304 nextFragmentPos = EraseFragmentCache(nextFragmentPos);
305 DumpInner(0);
306 return false;
307 }
308 auto &chunkInfo = *endPos;
309 uint64_t newOffset = offset + static_cast<uint64_t>(writeSize);
310 uint64_t dataLength = static_cast<uint64_t>(chunkInfo->dataLength);
311 uint64_t moveLen = std::max(chunkInfo->offset + dataLength, newOffset) - newOffset;
312 auto mergeDataLen = chunkInfo->dataLength > moveLen ? chunkInfo->dataLength - moveLen : 0;
313 errno_t res = memmove_s(chunkInfo->data, moveLen, chunkInfo->data + mergeDataLen, moveLen);
314 FALSE_RETURN_V_MSG_E(res == EOK, false, "memmove_s data err");
315 chunkInfo->offset = newOffset;
316 chunkInfo->dataLength = static_cast<uint32_t>(moveLen);
317 uint64_t lostLength = std::max(newOffset, nextFragmentPos->offsetBegin) - nextFragmentPos->offsetBegin;
318 nextFragmentPos->dataLength -= static_cast<int64_t>(lostLength);
319 lruCache_.Update(nextFragmentPos->offsetBegin, newOffset, nextFragmentPos);
320 nextFragmentPos->offsetBegin = newOffset;
321 nextFragmentPos->accessLength = 0;
322 nextFragmentPos->accessPos = nextFragmentPos->chunks.end();
323 isLoop = false;
324 break;
325 } else {
326 freeChunks_.splice(freeChunks_.end(), nextFragmentPos->chunks);
327 writePos_->totalReadSize += nextFragmentPos->totalReadSize;
328 nextFragmentPos->totalReadSize = 0; // avoid total size sub, chunk num reduce.
329 nextFragmentPos = EraseFragmentCache(nextFragmentPos);
330 }
331 }
332 return true;
333 }
334
WriteMergerPost(FragmentIterator & nextFragmentPos)335 void CacheMediaChunkBufferImpl::WriteMergerPost(FragmentIterator& nextFragmentPos)
336 {
337 if (nextFragmentPos == fragmentCacheBuffer_.end() || writePos_->chunks.empty() ||
338 nextFragmentPos->chunks.empty()) {
339 return;
340 }
341 auto preChunkInfo = writePos_->chunks.back();
342 auto nextChunkInfo = nextFragmentPos->chunks.front();
343 if (preChunkInfo->offset + preChunkInfo->dataLength != nextChunkInfo->offset) {
344 DumpAndCheckInner();
345 return;
346 }
347 writePos_->dataLength += nextFragmentPos->dataLength;
348 writePos_->totalReadSize += nextFragmentPos->totalReadSize;
349 nextFragmentPos->totalReadSize = 0; // avoid total size sub, chunk num reduce
350 writePos_->chunks.splice(writePos_->chunks.end(), nextFragmentPos->chunks);
351 EraseFragmentCache(nextFragmentPos);
352 }
353
Write(void * ptr,uint64_t inOffset,size_t inWriteSize)354 size_t CacheMediaChunkBufferImpl::Write(void* ptr, uint64_t inOffset, size_t inWriteSize)
355 {
356 std::lock_guard lock(mutex_);
357 uint64_t offset = inOffset;
358 size_t writeSize = inWriteSize;
359 uint8_t* src = static_cast<uint8_t*>(ptr);
360 size_t dupWriteSize = 0;
361
362 auto fragmentPos = GetOffsetFragmentCache(writePos_, offset, BoundedIntervalComp);
363 ChunkIterator chunkPos;
364 if (fragmentPos != fragmentCacheBuffer_.end()) {
365 auto& chunkList = fragmentPos->chunks;
366 writePos_ = fragmentPos;
367 if ((fragmentPos->offsetBegin + static_cast<uint64_t>(fragmentPos->dataLength)) != offset) {
368 auto ret = WriteInPlace(fragmentPos, src, offset, writeSize, dupWriteSize);
369 if (!ret || dupWriteSize >= writeSize) {
370 return ret ? writeSize : dupWriteSize;
371 }
372 src += dupWriteSize;
373 offset += dupWriteSize;
374 writeSize -= dupWriteSize;
375 }
376 chunkPos = std::prev(chunkList.end());
377 } else {
378 if (freeChunks_.empty()) {
379 MEDIA_LOG_D("no free chunk.");
380 }
381 MEDIA_LOG_D("not find fragment.");
382 chunkPos = AddFragmentCacheBuffer(offset);
383 }
384 FragmentIterator nextFragmentPos = fragmentCacheBuffer_.end();
385 auto success = WriteMergerPre(offset, writeSize, nextFragmentPos);
386 if (!success) {
387 return dupWriteSize;
388 }
389 auto writeSizeTmp = WriteChunk(*writePos_, chunkPos, src, offset, writeSize);
390 if (writeSize != writeSizeTmp) {
391 nextFragmentPos = fragmentCacheBuffer_.end();
392 }
393 WriteMergerPost(nextFragmentPos);
394 return writeSizeTmp + dupWriteSize;
395 }
396
Write(void * ptr,uint64_t inOffset,size_t inWriteSize)397 size_t CacheMediaChunkBufferHlsImpl::Write(void* ptr, uint64_t inOffset, size_t inWriteSize)
398 {
399 std::lock_guard lock(mutex_);
400 uint64_t offset = inOffset;
401 size_t writeSize = inWriteSize;
402 uint8_t* src = static_cast<uint8_t*>(ptr);
403 size_t dupWriteSize = 0;
404
405 auto fragmentPos = GetOffsetFragmentCache(writePos_, offset, BoundedIntervalComp);
406 ChunkIterator chunkPos;
407 if (fragmentPos != fragmentCacheBuffer_.end()) {
408 auto& chunkList = fragmentPos->chunks;
409 writePos_ = fragmentPos;
410 if ((fragmentPos->offsetBegin + static_cast<uint64_t>(fragmentPos->dataLength)) != offset) {
411 auto ret = WriteInPlace(fragmentPos, src, offset, writeSize, dupWriteSize);
412 if (!ret || dupWriteSize >= writeSize) {
413 return ret ? writeSize : dupWriteSize;
414 }
415 src += dupWriteSize;
416 offset += dupWriteSize;
417 writeSize -= dupWriteSize;
418 }
419 chunkPos = std::prev(chunkList.end());
420 } else {
421 if (freeChunks_.empty()) {
422 return 0; // 只有hls可以return 0
423 }
424 MEDIA_LOG_D("not find fragment.");
425 chunkPos = AddFragmentCacheBuffer(offset);
426 }
427 FragmentIterator nextFragmentPos = fragmentCacheBuffer_.end();
428 auto success = WriteMergerPre(offset, writeSize, nextFragmentPos);
429 if (!success) {
430 return dupWriteSize;
431 }
432 auto writeSizeTmp = WriteChunk(*writePos_, chunkPos, src, offset, writeSize);
433 if (writeSize != writeSizeTmp) {
434 nextFragmentPos = fragmentCacheBuffer_.end();
435 }
436 WriteMergerPost(nextFragmentPos);
437 return writeSizeTmp + dupWriteSize;
438 }
439
Seek(uint64_t offset)440 bool CacheMediaChunkBufferImpl::Seek(uint64_t offset)
441 {
442 std::lock_guard lock(mutex_);
443 auto readPos = GetOffsetFragmentCache(readPos_, offset, BoundedIntervalComp);
444 if (readPos != fragmentCacheBuffer_.end()) {
445 readPos_ = readPos;
446 bool isSeekHit = false;
447 auto chunkPos = GetOffsetChunkCache(readPos->chunks, offset, LeftBoundedRightOpenComp);
448 if (chunkPos != readPos->chunks.end()) {
449 auto readOffset = offset > readPos->offsetBegin ? offset - readPos->offsetBegin : 0;
450 uint64_t diff = readOffset > static_cast<uint64_t>(readPos->accessLength) ?
451 readOffset - static_cast<uint64_t>(readPos->accessLength) : 0;
452 if (diff >= ACCESS_OFFSET_MAX_LENGTH) {
453 chunkPos = SplitFragmentCacheBuffer(readPos, offset, chunkPos);
454 }
455
456 if (chunkPos == readPos->chunks.end()) {
457 return false;
458 }
459 lruCache_.Refer(readPos->offsetBegin, readPos);
460 (*readPos).accessPos = chunkPos;
461 auto tmpLength = offset > (*readPos).offsetBegin ? offset - (*readPos).offsetBegin : 0;
462 (*readPos).accessLength = static_cast<int64_t>(tmpLength);
463 readPos->readTime = Clock::now();
464 isSeekHit = true;
465 }
466 ResetReadSizeAlloc();
467 uint64_t newReadSizeInit = static_cast<uint64_t>(1 + initReadSizeFactor_ * static_cast<double>(totalReadSize_));
468 readPos->totalReadSize += newReadSizeInit;
469 totalReadSize_ += newReadSizeInit;
470 return isSeekHit;
471 }
472 return false;
473 }
474
GetBufferSize(uint64_t offset)475 size_t CacheMediaChunkBufferImpl::GetBufferSize(uint64_t offset)
476 {
477 std::lock_guard lock(mutex_);
478 auto readPos = GetOffsetFragmentCache(readPos_, offset, LeftBoundedRightOpenComp);
479 size_t bufferSize = 0;
480 while (readPos != fragmentCacheBuffer_.end()) {
481 uint64_t nextOffsetBegin = readPos->offsetBegin + static_cast<uint64_t>(readPos->dataLength);
482 bufferSize = static_cast<size_t>(nextOffsetBegin > offset ? nextOffsetBegin - offset : 0);
483 readPos++;
484 if (readPos == fragmentCacheBuffer_.end() || nextOffsetBegin != readPos->offsetBegin) {
485 break;
486 }
487 }
488 return bufferSize;
489 }
490
HandleFragmentPos(FragmentIterator & fragmentIter)491 void CacheMediaChunkBufferImpl::HandleFragmentPos(FragmentIterator& fragmentIter)
492 {
493 uint64_t nextOffsetBegin = fragmentIter->offsetBegin + static_cast<uint64_t>(fragmentIter->dataLength);
494 ++fragmentIter;
495 while (fragmentIter != fragmentCacheBuffer_.end()) {
496 if (nextOffsetBegin != fragmentIter->offsetBegin) {
497 break;
498 }
499 nextOffsetBegin = fragmentIter->offsetBegin + static_cast<uint64_t>(fragmentIter->dataLength);
500 ++fragmentIter;
501 }
502 }
503
GetNextBufferOffset(uint64_t offset)504 uint64_t CacheMediaChunkBufferImpl::GetNextBufferOffset(uint64_t offset)
505 {
506 std::lock_guard lock(mutex_);
507 auto fragmentIter = std::upper_bound(fragmentCacheBuffer_.begin(), fragmentCacheBuffer_.end(), offset,
508 [](auto inputOffset, const FragmentCacheBuffer& fragment) {
509 return (inputOffset < fragment.offsetBegin + fragment.dataLength);
510 });
511 if (fragmentIter != fragmentCacheBuffer_.end()) {
512 if (LeftBoundedRightOpenComp(offset, fragmentIter->offsetBegin,
513 fragmentIter->offsetBegin + fragmentIter->dataLength)) {
514 HandleFragmentPos(fragmentIter);
515 }
516 }
517 if (fragmentIter != fragmentCacheBuffer_.end()) {
518 return fragmentIter->offsetBegin;
519 }
520 return 0;
521 }
522
EraseFragmentCache(const FragmentIterator & iter)523 FragmentIterator CacheMediaChunkBufferImpl::EraseFragmentCache(const FragmentIterator& iter)
524 {
525 if (iter == readPos_) {
526 readPos_ = fragmentCacheBuffer_.end();
527 }
528 if (iter == writePos_) {
529 writePos_ = fragmentCacheBuffer_.end();
530 }
531 totalReadSize_ -= iter->totalReadSize;
532 lruCache_.Delete(iter->offsetBegin);
533 return fragmentCacheBuffer_.erase(iter);
534 }
535
WriteOneChunkData(CacheChunk & chunkInfo,uint8_t * src,uint64_t offset,size_t writeSize)536 inline size_t WriteOneChunkData(CacheChunk& chunkInfo, uint8_t* src, uint64_t offset, size_t writeSize)
537 {
538 uint64_t copyBegin = offset > chunkInfo.offset ? offset - chunkInfo.offset : 0;
539 if (copyBegin < 0 || copyBegin > chunkInfo.chunkSize) {
540 return 0;
541 }
542 size_t writePerOne = static_cast<size_t>(chunkInfo.chunkSize - static_cast<size_t>(copyBegin));
543 writePerOne = std::min(writePerOne, writeSize);
544 errno_t res = memcpy_s(chunkInfo.data + copyBegin, writePerOne, src, writePerOne);
545 FALSE_RETURN_V_MSG_E(res == EOK, 0, "memcpy_s data err");
546 chunkInfo.dataLength = static_cast<uint32_t>(static_cast<size_t>(copyBegin) + writePerOne);
547 return writePerOne;
548 }
549
PopFreeCacheChunk(CacheChunkList & freeChunks,uint64_t offset)550 inline CacheChunk* PopFreeCacheChunk(CacheChunkList& freeChunks, uint64_t offset)
551 {
552 if (freeChunks.empty()) {
553 return nullptr;
554 }
555 auto tmp = freeChunks.front();
556 freeChunks.pop_front();
557 InitChunkInfo(*tmp, offset);
558 return tmp;
559 }
560
WriteChunk(FragmentCacheBuffer & fragmentCacheBuffer,ChunkIterator & chunkPos,void * ptr,uint64_t offset,size_t writeSize)561 size_t CacheMediaChunkBufferImpl::WriteChunk(FragmentCacheBuffer& fragmentCacheBuffer, ChunkIterator& chunkPos,
562 void* ptr, uint64_t offset, size_t writeSize)
563 {
564 if (chunkPos == fragmentCacheBuffer.chunks.end()) {
565 MEDIA_LOG_D("input valid.");
566 return 0;
567 }
568 size_t writedTmp = 0;
569 auto chunkInfo = *chunkPos;
570 uint8_t* src = static_cast<uint8_t*>(ptr);
571 if (chunkInfo->chunkSize > chunkInfo->dataLength) {
572 writedTmp += WriteOneChunkData(*chunkInfo, src, offset, writeSize);
573 fragmentCacheBuffer.dataLength += static_cast<int64_t>(writedTmp);
574 }
575 int64_t loopStartTime = loopInterruptClock_.ElapsedSeconds();
576 while (writedTmp < writeSize && writedTmp >= 0) {
577 if (CheckLoopTimeout(loopStartTime)) {
578 break;
579 }
580 auto chunkOffset = offset + static_cast<uint64_t>(writedTmp);
581 auto freeChunk = GetFreeCacheChunk(chunkOffset);
582 if (freeChunk == nullptr) {
583 return writedTmp;
584 }
585 auto writePerOne = WriteOneChunkData(*freeChunk, src + writedTmp, chunkOffset, writeSize - writedTmp);
586 fragmentCacheBuffer.chunks.push_back(freeChunk);
587 writedTmp += writePerOne;
588 fragmentCacheBuffer.dataLength += static_cast<int64_t>(writePerOne);
589
590 if (fragmentCacheBuffer.accessPos == fragmentCacheBuffer.chunks.end()) {
591 fragmentCacheBuffer.accessPos = std::prev(fragmentCacheBuffer.chunks.end());
592 }
593 }
594 return writedTmp;
595 }
596
UpdateFragmentCacheForDelHead(FragmentIterator & fragmentIter)597 CacheChunk* CacheMediaChunkBufferImpl::UpdateFragmentCacheForDelHead(FragmentIterator& fragmentIter)
598 {
599 FragmentCacheBuffer& fragment = *fragmentIter;
600 if (fragment.chunks.empty()) {
601 return nullptr;
602 }
603 auto cacheChunk = fragment.chunks.front();
604 fragment.chunks.pop_front();
605
606 auto oldOffsetBegin = fragment.offsetBegin;
607 int64_t dataLength = static_cast<int64_t>(cacheChunk->dataLength);
608 fragment.offsetBegin += static_cast<uint64_t>(dataLength);
609 fragment.dataLength -= dataLength;
610 if (fragment.accessLength > dataLength) {
611 fragment.accessLength -= dataLength;
612 } else {
613 fragment.accessLength = 0;
614 }
615 lruCache_.Update(oldOffsetBegin, fragmentIter->offsetBegin, fragmentIter);
616 return cacheChunk;
617 }
618
UpdateFragmentCacheForDelTail(FragmentCacheBuffer & fragment)619 CacheChunk* UpdateFragmentCacheForDelTail(FragmentCacheBuffer& fragment)
620 {
621 if (fragment.chunks.empty()) {
622 return nullptr;
623 }
624 if (fragment.accessPos == std::prev(fragment.chunks.end())) {
625 fragment.accessPos = fragment.chunks.end();
626 }
627
628 auto cacheChunk = fragment.chunks.back();
629 fragment.chunks.pop_back();
630
631 auto dataLength = cacheChunk->dataLength;
632 if (fragment.accessLength > fragment.dataLength - static_cast<int64_t>(dataLength)) {
633 fragment.accessLength = fragment.dataLength - static_cast<int64_t>(dataLength);
634 }
635 fragment.dataLength -= static_cast<int64_t>(dataLength);
636 return cacheChunk;
637 }
638
CheckThresholdFragmentCacheBuffer(FragmentIterator & currWritePos)639 bool CacheMediaChunkBufferImpl::CheckThresholdFragmentCacheBuffer(FragmentIterator& currWritePos)
640 {
641 int64_t offset = -1;
642 FragmentIterator fragmentIterator = fragmentCacheBuffer_.end();
643 auto ret = lruCache_.GetLruNode(offset, fragmentIterator);
644 if (!ret) {
645 return false;
646 }
647 if (fragmentIterator == fragmentCacheBuffer_.end()) {
648 return false;
649 }
650 if (currWritePos == fragmentIterator) {
651 lruCache_.Refer(offset, currWritePos);
652 ret = lruCache_.GetLruNode(offset, fragmentIterator);
653 if (!ret) {
654 return false;
655 }
656 if (fragmentIterator == fragmentCacheBuffer_.end()) {
657 return false;
658 }
659 }
660 if (fragmentIterator != fragmentCacheBuffer_.end()) {
661 freeChunks_.splice(freeChunks_.end(), fragmentIterator->chunks);
662 EraseFragmentCache(fragmentIterator);
663 }
664 return true;
665 }
666
667 /***
668 * 总体策略：
669 * 计算最大允许Fragment数，大于 FRAGMENT_MAX_NUM(4)则剔除最近为未读取的Fragment（不包含当前写的节点）
670 * 新分配的节点固定分配 个chunk大小，通过公式计算，保证其能够下载；
671 * 每个Fragment最大允许的Chunk数：（本Fragment读取字节（fragmentReadSize）/ 总读取字节（totalReadSize））* 总Chunk个数
672 * 计算改Fragment最大允许的chunk个数
673 * 如果超过，则删除对应已读chunk，如果没有已读chunk，还超则返回不允许继续写，返回失败；（说明该Fragment不能再写更多的内容）
674 * 如果没有超过则从空闲队列中获取chunk，没有则
675 * for循环其他Fragment，计算每个Fragment的最大允许chunk个数：
676 * 如果超过，则删除对应已读chunk
677 * 如果还不够，则
678 * for循环其他Fragment，计算每个Fragment的最大允许chunk个数：
679 * 如果超过，则删除对应末尾未读chunk
680 *
681 * 如果还没有则返回失败
682 *
683 * 备注：是否一开始：优先从空闲队列中获取，没有则继续。
684 */
DeleteHasReadFragmentCacheBuffer(FragmentIterator & fragmentIter,size_t allowChunkNum)685 void CacheMediaChunkBufferImpl::DeleteHasReadFragmentCacheBuffer(FragmentIterator& fragmentIter, size_t allowChunkNum)
686 {
687 auto& fragmentCacheChunks = *fragmentIter;
688 while (fragmentCacheChunks.chunks.size() >= allowChunkNum &&
689 fragmentCacheChunks.accessLength > static_cast<int64_t>(static_cast<double>(fragmentCacheChunks.dataLength) *
690 CACHE_RELEASE_FACTOR_DEFAULT / TO_PERCENT)) {
691 if (fragmentCacheChunks.accessPos != fragmentCacheChunks.chunks.begin()) {
692 auto tmp = UpdateFragmentCacheForDelHead(fragmentIter);
693 if (tmp != nullptr) {
694 freeChunks_.push_back(tmp);
695 }
696 } else {
697 MEDIA_LOG_D("judge has read finish.");
698 break;
699 }
700 }
701 }
702
DeleteUnreadFragmentCacheBuffer(FragmentIterator & fragmentIter,size_t allowChunkNum)703 void CacheMediaChunkBufferImpl::DeleteUnreadFragmentCacheBuffer(FragmentIterator& fragmentIter, size_t allowChunkNum)
704 {
705 auto& fragmentCacheChunks = *fragmentIter;
706 while (fragmentCacheChunks.chunks.size() > allowChunkNum) {
707 if (!fragmentCacheChunks.chunks.empty()) {
708 auto tmp = UpdateFragmentCacheForDelTail(fragmentCacheChunks);
709 if (tmp != nullptr) {
710 freeChunks_.push_back(tmp);
711 }
712 } else {
713 break;
714 }
715 }
716 }
717
GetFreeCacheChunk(uint64_t offset,bool checkAllowFailContinue)718 CacheChunk* CacheMediaChunkBufferImpl::GetFreeCacheChunk(uint64_t offset, bool checkAllowFailContinue)
719 {
720 if (writePos_ == fragmentCacheBuffer_.end()) {
721 return nullptr;
722 }
723 if (!freeChunks_.empty()) {
724 return PopFreeCacheChunk(freeChunks_, offset);
725 }
726 auto currWritePos = GetOffsetFragmentCache(writePos_, offset, BoundedIntervalComp);
727 size_t allowChunkNum = 0;
728 if (currWritePos != fragmentCacheBuffer_.end()) {
729 allowChunkNum = CalcAllowMaxChunkNum(currWritePos->totalReadSize, currWritePos->offsetBegin);
730 DeleteHasReadFragmentCacheBuffer(currWritePos, allowChunkNum);
731 if (currWritePos->chunks.size() >= allowChunkNum && !checkAllowFailContinue) {
732 return nullptr;
733 }
734 }
735 if (!freeChunks_.empty()) {
736 return PopFreeCacheChunk(freeChunks_, offset);
737 }
738 for (auto iter = fragmentCacheBuffer_.begin(); iter != fragmentCacheBuffer_.end(); ++iter) {
739 if (iter != currWritePos) {
740 allowChunkNum = CalcAllowMaxChunkNum(iter->totalReadSize, iter->offsetBegin);
741 DeleteHasReadFragmentCacheBuffer(iter, allowChunkNum);
742 }
743 }
744 if (!freeChunks_.empty()) {
745 return PopFreeCacheChunk(freeChunks_, offset);
746 }
747 while (fragmentCacheBuffer_.size() > CACHE_FRAGMENT_MIN_NUM_DEFAULT) {
748 auto result = CheckThresholdFragmentCacheBuffer(currWritePos);
749 if (!freeChunks_.empty()) {
750 return PopFreeCacheChunk(freeChunks_, offset);
751 }
752 if (!result) {
753 break;
754 }
755 }
756 for (auto iter = fragmentCacheBuffer_.begin(); iter != fragmentCacheBuffer_.end(); ++iter) {
757 if (iter != currWritePos) {
758 allowChunkNum = CalcAllowMaxChunkNum(iter->totalReadSize, iter->offsetBegin);
759 DeleteUnreadFragmentCacheBuffer(iter, allowChunkNum);
760 }
761 }
762 if (!freeChunks_.empty()) {
763 return PopFreeCacheChunk(freeChunks_, offset);
764 }
765 return nullptr;
766 }
767
GetFreeCacheChunk(uint64_t offset,bool checkAllowFailContinue)768 CacheChunk* CacheMediaChunkBufferHlsImpl::GetFreeCacheChunk(uint64_t offset, bool checkAllowFailContinue)
769 {
770 if (writePos_ == fragmentCacheBuffer_.end()) {
771 return nullptr;
772 }
773 if (!freeChunks_.empty()) {
774 return PopFreeCacheChunk(freeChunks_, offset);
775 }
776 auto currWritePos = GetOffsetFragmentCache(writePos_, offset, BoundedIntervalComp);
777 size_t allowChunkNum = 0;
778 if (currWritePos != fragmentCacheBuffer_.end()) {
779 allowChunkNum = CalcAllowMaxChunkNum(currWritePos->totalReadSize, currWritePos->offsetBegin);
780 DeleteHasReadFragmentCacheBuffer(currWritePos, allowChunkNum);
781 if (currWritePos->chunks.size() >= allowChunkNum && !checkAllowFailContinue) {
782 MEDIA_LOG_D("allowChunkNum limit.");
783 return nullptr;
784 }
785 } else {
786 MEDIA_LOG_D("curr write is new fragment.");
787 }
788 MEDIA_LOG_D("clear other fragment has read chunk.");
789 for (auto iter = fragmentCacheBuffer_.begin(); iter != fragmentCacheBuffer_.end(); ++iter) {
790 if (iter != currWritePos) {
791 allowChunkNum = CalcAllowMaxChunkNum(iter->totalReadSize, iter->offsetBegin);
792 DeleteHasReadFragmentCacheBuffer(iter, allowChunkNum);
793 }
794 }
795 if (!freeChunks_.empty()) {
796 return PopFreeCacheChunk(freeChunks_, offset);
797 }
798 return nullptr;
799 }
800
GetFragmentIterator(FragmentIterator & currFragmentIter,uint64_t offset,ChunkIterator chunkPos,CacheChunk * splitHead,CacheChunk * & chunkInfo)801 FragmentIterator CacheMediaChunkBufferImpl::GetFragmentIterator(FragmentIterator& currFragmentIter,
802 uint64_t offset, ChunkIterator chunkPos, CacheChunk* splitHead, CacheChunk*& chunkInfo)
803 {
804 auto newFragmentPos = fragmentCacheBuffer_.emplace(std::next(currFragmentIter), offset);
805 if (splitHead == nullptr) {
806 newFragmentPos->chunks.splice(newFragmentPos->chunks.end(), currFragmentIter->chunks, chunkPos,
807 currFragmentIter->chunks.end());
808 } else {
809 splitHead->dataLength = 0;
810 newFragmentPos->chunks.splice(newFragmentPos->chunks.end(), currFragmentIter->chunks, std::next(chunkPos),
811 currFragmentIter->chunks.end());
812 newFragmentPos->chunks.push_front(splitHead);
813 splitHead->offset = offset;
814 uint64_t diff = offset > chunkInfo->offset ? offset - chunkInfo->offset : 0;
815 if (chunkInfo->dataLength >= diff) {
816 splitHead->dataLength = chunkInfo->dataLength - static_cast<uint32_t>(diff);
817 chunkInfo->dataLength = static_cast<uint32_t>(diff);
818 memcpy_s(splitHead->data, splitHead->dataLength, chunkInfo->data + diff, splitHead->dataLength);
819 }
820 }
821 newFragmentPos->offsetBegin = offset;
822 uint64_t diff = offset > currFragmentIter->offsetBegin ? offset - currFragmentIter->offsetBegin : 0;
823 newFragmentPos->dataLength = currFragmentIter->dataLength > static_cast<int64_t>(diff) ?
824 currFragmentIter->dataLength - static_cast<int64_t>(diff) : 0;
825 newFragmentPos->accessLength = 0;
826 uint64_t newReadSizeInit = static_cast<uint64_t>(1 + initReadSizeFactor_ * static_cast<double>(totalReadSize_));
827 newReadSizeInit = std::max(newReadSizeInit, currFragmentIter->totalReadSize);
828
829 newFragmentPos->totalReadSize = newReadSizeInit;
830 totalReadSize_ += newReadSizeInit;
831 newFragmentPos->readTime = Clock::now();
832 newFragmentPos->accessPos = newFragmentPos->chunks.begin();
833 newFragmentPos->isSplit = currFragmentIter->isSplit;
834 currFragmentIter->isSplit = true;
835 currFragmentIter->dataLength = static_cast<int64_t>(offset > currFragmentIter->offsetBegin ?
836 offset - currFragmentIter->offsetBegin : 0);
837 return newFragmentPos;
838 }
839
SplitFragmentCacheBuffer(FragmentIterator & currFragmentIter,uint64_t offset,ChunkIterator chunkPos)840 ChunkIterator CacheMediaChunkBufferImpl::SplitFragmentCacheBuffer(FragmentIterator& currFragmentIter,
841 uint64_t offset, ChunkIterator chunkPos)
842 {
843 ResetReadSizeAlloc();
844 auto& chunkInfo = *chunkPos;
845 CacheChunk* splitHead = nullptr;
846 if (offset != chunkInfo->offset) {
847 splitHead = freeChunks_.empty() ? GetFreeCacheChunk(offset, true) : PopFreeCacheChunk(freeChunks_, offset);
848 if (splitHead == nullptr) {
849 return chunkPos;
850 }
851 }
852 auto newFragmentPos = GetFragmentIterator(currFragmentIter, offset, chunkPos, splitHead, chunkInfo);
853 currFragmentIter = newFragmentPos;
854 if (fragmentCacheBuffer_.size() > CACHE_FRAGMENT_MAX_NUM_DEFAULT) {
855 CheckThresholdFragmentCacheBuffer(currFragmentIter);
856 }
857 lruCache_.Refer(newFragmentPos->offsetBegin, newFragmentPos);
858 return newFragmentPos->accessPos;
859 }
860
SplitFragmentCacheBuffer(FragmentIterator & currFragmentIter,uint64_t offset,ChunkIterator chunkPos)861 ChunkIterator CacheMediaChunkBufferHlsImpl::SplitFragmentCacheBuffer(FragmentIterator& currFragmentIter,
862 uint64_t offset, ChunkIterator chunkPos)
863 {
864 ResetReadSizeAlloc();
865 auto& chunkInfo = *chunkPos;
866 CacheChunk* splitHead = nullptr;
867 if (offset != chunkInfo->offset) {
868 splitHead = freeChunks_.empty() ? GetFreeCacheChunk(offset, true) : PopFreeCacheChunk(freeChunks_, offset);
869 if (splitHead == nullptr) {
870 return chunkPos;
871 }
872 }
873 auto newFragmentPos = fragmentCacheBuffer_.emplace(std::next(currFragmentIter), offset);
874 if (splitHead == nullptr) {
875 newFragmentPos->chunks.splice(newFragmentPos->chunks.end(), currFragmentIter->chunks, chunkPos,
876 currFragmentIter->chunks.end());
877 } else {
878 newFragmentPos->chunks.splice(newFragmentPos->chunks.end(), currFragmentIter->chunks, std::next(chunkPos),
879 currFragmentIter->chunks.end());
880 newFragmentPos->chunks.push_front(splitHead);
881 splitHead->offset = offset;
882 uint64_t diff = offset > chunkInfo->offset ? offset - chunkInfo->offset : 0;
883 if (chunkInfo->dataLength >= diff) {
884 splitHead->dataLength = chunkInfo->dataLength > static_cast<uint32_t>(diff) ?
885 chunkInfo->dataLength - static_cast<uint32_t>(diff) : 0;
886 chunkInfo->dataLength = static_cast<uint32_t>(diff);
887 memcpy_s(splitHead->data, splitHead->dataLength, chunkInfo->data + diff, splitHead->dataLength);
888 } else {
889 splitHead->dataLength = 0; // It can't happen. us_asan can check.
890 }
891 }
892 newFragmentPos->offsetBegin = offset;
893 uint64_t diff = offset > currFragmentIter->offsetBegin ? offset - currFragmentIter->offsetBegin : 0;
894 newFragmentPos->dataLength = currFragmentIter->dataLength > static_cast<int64_t>(diff) ?
895 currFragmentIter->dataLength - static_cast<int64_t>(diff) : 0;
896 newFragmentPos->accessLength = 0;
897 uint64_t newReadSizeInit = static_cast<uint64_t>(1 + initReadSizeFactor_ * static_cast<double>(totalReadSize_));
898 if (currFragmentIter->totalReadSize > newReadSizeInit) {
899 newReadSizeInit = currFragmentIter->totalReadSize;
900 }
901 newFragmentPos->totalReadSize = newReadSizeInit;
902 totalReadSize_ += newReadSizeInit;
903 newFragmentPos->readTime = Clock::now();
904 newFragmentPos->accessPos = newFragmentPos->chunks.begin();
905 currFragmentIter->dataLength = static_cast<int64_t>(offset > diff ? offset - diff : 0);
906 currFragmentIter = newFragmentPos;
907 lruCache_.Refer(newFragmentPos->offsetBegin, newFragmentPos);
908 return newFragmentPos->accessPos;
909 }
910
AddFragmentCacheBuffer(uint64_t offset)911 ChunkIterator CacheMediaChunkBufferImpl::AddFragmentCacheBuffer(uint64_t offset)
912 {
913 size_t fragmentThreshold = CACHE_FRAGMENT_MAX_NUM_DEFAULT;
914 if (isLargeOffsetSpan_) {
915 fragmentThreshold = CACHE_FRAGMENT_MAX_NUM_LARGE;
916 }
917 if (fragmentCacheBuffer_.size() >= fragmentThreshold) {
918 auto fragmentIterTmp = fragmentCacheBuffer_.end();
919 CheckThresholdFragmentCacheBuffer(fragmentIterTmp);
920 }
921 ResetReadSizeAlloc();
922 auto fragmentInsertPos = std::upper_bound(fragmentCacheBuffer_.begin(), fragmentCacheBuffer_.end(), offset,
923 [](auto mediaOffset, const FragmentCacheBuffer& fragment) {
924 if (mediaOffset <= fragment.offsetBegin + fragment.dataLength) {
925 return true;
926 }
927 return false;
928 });
929 auto newFragmentPos = fragmentCacheBuffer_.emplace(fragmentInsertPos, offset);
930 uint64_t newReadSizeInit = static_cast<uint64_t>(1 + initReadSizeFactor_ * static_cast<double>(totalReadSize_));
931 totalReadSize_ += newReadSizeInit;
932 newFragmentPos->totalReadSize = newReadSizeInit;
933 writePos_ = newFragmentPos;
934 writePos_->accessPos = writePos_->chunks.end();
935 lruCache_.Refer(newFragmentPos->offsetBegin, newFragmentPos);
936 auto freeChunk = GetFreeCacheChunk(offset);
937 if (freeChunk == nullptr) {
938 MEDIA_LOG_D("get free cache chunk fail.");
939 return writePos_->chunks.end();
940 }
941 writePos_->accessPos = newFragmentPos->chunks.emplace(newFragmentPos->chunks.end(), freeChunk);
942 return writePos_->accessPos;
943 }
944
AddFragmentCacheBuffer(uint64_t offset)945 ChunkIterator CacheMediaChunkBufferHlsImpl::AddFragmentCacheBuffer(uint64_t offset)
946 {
947 ResetReadSizeAlloc();
948 auto fragmentInsertPos = std::upper_bound(fragmentCacheBuffer_.begin(), fragmentCacheBuffer_.end(), offset,
949 [](auto mediaOffset, const FragmentCacheBuffer& fragment) {
950 if (mediaOffset <= fragment.offsetBegin + fragment.dataLength) {
951 return true;
952 }
953 return false;
954 });
955 auto newFragmentPos = fragmentCacheBuffer_.emplace(fragmentInsertPos, offset);
956 uint64_t newReadSizeInit = static_cast<uint64_t>(1 + initReadSizeFactor_ * static_cast<double>(totalReadSize_));
957 totalReadSize_ += newReadSizeInit;
958 newFragmentPos->totalReadSize = newReadSizeInit;
959 writePos_ = newFragmentPos;
960 writePos_->accessPos = writePos_->chunks.end();
961 lruCache_.Refer(newFragmentPos->offsetBegin, newFragmentPos);
962 auto freeChunk = GetFreeCacheChunk(offset);
963 if (freeChunk == nullptr) {
964 MEDIA_LOG_D("get free cache chunk fail.");
965 return writePos_->chunks.end();
966 }
967 writePos_->accessPos = newFragmentPos->chunks.emplace(newFragmentPos->chunks.end(), freeChunk);
968 return writePos_->accessPos;
969 }
970
ResetReadSizeAlloc()971 void CacheMediaChunkBufferImpl::ResetReadSizeAlloc()
972 {
973 size_t chunkNum = chunkMaxNum_ + 1 >= freeChunks_.size() ?
974 chunkMaxNum_ + 1 - freeChunks_.size() : 0;
975 if (totalReadSize_ > static_cast<size_t>(UP_LIMIT_MAX_TOTAL_READ_SIZE) && chunkNum > 0) {
976 size_t preChunkSize = static_cast<size_t>(MAX_TOTAL_READ_SIZE - 1) / chunkNum;
977 for (auto iter = fragmentCacheBuffer_.begin(); iter != fragmentCacheBuffer_.end(); ++iter) {
978 iter->totalReadSize = preChunkSize * iter->chunks.size();
979 }
980 totalReadSize_ = preChunkSize * chunkNum;
981 }
982 }
983
Dump(uint64_t param)984 void CacheMediaChunkBufferImpl::Dump(uint64_t param)
985 {
986 std::lock_guard lock(mutex_);
987 DumpInner(param);
988 }
989
DumpInner(uint64_t param)990 void CacheMediaChunkBufferImpl::DumpInner(uint64_t param)
991 {
992 (void)param;
993 MEDIA_LOG_D("cacheBuff total buffer size : " PUBLIC_LOG_U64, totalBuffSize_);
994 MEDIA_LOG_D("cacheBuff total chunk size : " PUBLIC_LOG_U32, chunkSize_);
995 MEDIA_LOG_D("cacheBuff total chunk num : " PUBLIC_LOG_U32, chunkMaxNum_);
996 MEDIA_LOG_D("cacheBuff total read size : " PUBLIC_LOG_U64, totalReadSize_);
997 MEDIA_LOG_D("cacheBuff read size factor : " PUBLIC_LOG_F, initReadSizeFactor_);
998 MEDIA_LOG_D("cacheBuff free chunk num: : " PUBLIC_LOG_ZU, freeChunks_.size());
999 MEDIA_LOG_D("cacheBuff fragment num: : " PUBLIC_LOG_ZU, fragmentCacheBuffer_.size());
1000 for (auto const & fragment : fragmentCacheBuffer_) {
1001 MEDIA_LOG_D("cacheBuff - fragment offset : " PUBLIC_LOG_U64, fragment.offsetBegin);
1002 MEDIA_LOG_D("cacheBuff fragment length : " PUBLIC_LOG_D64, fragment.dataLength);
1003 MEDIA_LOG_D("cacheBuff chunk num : " PUBLIC_LOG_ZU, fragment.chunks.size());
1004 MEDIA_LOG_D("cacheBuff access length : " PUBLIC_LOG_U64, fragment.accessLength);
1005 MEDIA_LOG_D("cacheBuff read size : " PUBLIC_LOG_U64, fragment.totalReadSize);
1006 if (fragment.accessPos != fragment.chunks.end()) {
1007 auto &chunkInfo = *fragment.accessPos;
1008 MEDIA_LOG_D("cacheBuff access offset: " PUBLIC_LOG_D64 ", len: " PUBLIC_LOG_U32,
1009 chunkInfo->offset, chunkInfo->dataLength);
1010 } else {
1011 MEDIA_LOG_D("cacheBuff access ended");
1012 }
1013 if (!fragment.chunks.empty()) {
1014 auto &chunkInfo = fragment.chunks.back();
1015 MEDIA_LOG_D("cacheBuff last chunk offset: " PUBLIC_LOG_D64 ", len: " PUBLIC_LOG_U32,
1016 chunkInfo->offset, chunkInfo->dataLength);
1017 }
1018 MEDIA_LOG_D("cacheBuff ");
1019 }
1020 }
1021
CheckLoopTimeout(int64_t loopStartTime)1022 bool CacheMediaChunkBufferImpl::CheckLoopTimeout(int64_t loopStartTime)
1023 {
1024 int64_t now = loopInterruptClock_.ElapsedSeconds();
1025 int64_t loopDuration = now > loopStartTime ? now - loopStartTime : 0;
1026 bool isLoopTimeout = loopDuration > LOOP_TIMEOUT;
1027 if (isLoopTimeout) {
1028 MEDIA_LOG_E("loop timeout.");
1029 }
1030 return isLoopTimeout;
1031 }
1032
Check()1033 bool CacheMediaChunkBufferImpl::Check()
1034 {
1035 std::lock_guard lock(mutex_);
1036 return CheckInner();
1037 }
1038
Clear()1039 void CacheMediaChunkBufferImpl::Clear()
1040 {
1041 std::lock_guard lock(mutex_);
1042 auto iter = fragmentCacheBuffer_.begin();
1043 while (iter != fragmentCacheBuffer_.end()) {
1044 freeChunks_.splice(freeChunks_.end(), iter->chunks);
1045 iter = EraseFragmentCache(iter);
1046 }
1047 lruCache_.Reset();
1048 totalReadSize_ = 0;
1049 }
1050
GetFreeSize()1051 uint64_t CacheMediaChunkBufferImpl::GetFreeSize()
1052 {
1053 std::lock_guard lock(mutex_);
1054 uint64_t totalFreeSize = totalBuffSize_;
1055 for (auto iter = fragmentCacheBuffer_.begin(); iter != fragmentCacheBuffer_.end(); iter++) {
1056 uint64_t fragmentDataLen = static_cast<uint64_t>(iter->dataLength);
1057 totalFreeSize = totalFreeSize > fragmentDataLen ? totalFreeSize - fragmentDataLen : 0;
1058 }
1059 return totalFreeSize;
1060 }
1061
1062 // Release all fragments before the offset.
ClearChunksOfFragment(uint64_t offset)1063 bool CacheMediaChunkBufferImpl::ClearChunksOfFragment(uint64_t offset)
1064 {
1065 std::lock_guard lock(mutex_);
1066 bool res = false;
1067 auto fragmentPos = GetOffsetFragmentCache(readPos_, offset, LeftBoundedRightOpenComp);
1068 if (fragmentPos == fragmentCacheBuffer_.end()) {
1069 return false;
1070 }
1071 auto& fragment = *fragmentPos;
1072 if (fragment.chunks.empty()) {
1073 return false;
1074 }
1075 uint32_t chunkSize = fragment.chunks.size();
1076 for (uint32_t i = 0; i < chunkSize; ++i) {
1077 auto chunkIter = fragment.chunks.front();
1078 if (fragmentPos->accessPos == fragmentPos->chunks.end() || chunkIter == nullptr ||
1079 chunkIter->offset + chunkIter->dataLength >= offset) {
1080 break;
1081 }
1082
1083 auto chunkPos = fragmentPos->accessPos;
1084 if ((*chunkPos) != nullptr && chunkIter->offset >= (*chunkPos)->offset) { // Update accessPos of fragment
1085 chunkPos = GetOffsetChunkCache(fragmentPos->chunks, chunkIter->offset + chunkIter->dataLength,
1086 LeftBoundedRightOpenComp);
1087 (*fragmentPos).accessPos = chunkPos;
1088 }
1089
1090 MEDIA_LOG_D("ClearChunksOfFragment clear chunk, offsetBegin: " PUBLIC_LOG_U64 " offsetEnd " PUBLIC_LOG_U64,
1091 chunkIter->offset, chunkIter->offset + chunkIter->dataLength);
1092 auto tmp = UpdateFragmentCacheForDelHead(fragmentPos);
1093 if (tmp != nullptr) {
1094 res = true;
1095 freeChunks_.push_back(tmp);
1096 }
1097 }
1098 return res;
1099 }
1100
1101 // Release all chunks before the offset in the fragment to which the specified offset belongs.
ClearFragmentBeforeOffset(uint64_t offset)1102 bool CacheMediaChunkBufferImpl::ClearFragmentBeforeOffset(uint64_t offset)
1103 {
1104 std::lock_guard lock(mutex_);
1105 bool res = false;
1106 for (auto iter = fragmentCacheBuffer_.begin(); iter != fragmentCacheBuffer_.end();) {
1107 if (iter->offsetBegin >= offset) {
1108 break;
1109 }
1110 if (iter->offsetBegin + static_cast<uint64_t>(iter->dataLength) <= offset) {
1111 MEDIA_LOG_D("ClearFragmentBeforeOffset clear fragment, offsetBegin: " PUBLIC_LOG_U64 " offsetEnd "
1112 PUBLIC_LOG_U64, iter->offsetBegin, iter->offsetBegin + iter->dataLength);
1113 freeChunks_.splice(freeChunks_.end(), iter->chunks);
1114 iter = EraseFragmentCache(iter);
1115 res = true;
1116 continue;
1117 }
1118 iter++;
1119 }
1120 return res;
1121 }
1122
1123 // Release all chunks of read fragment between minReadOffset and maxReadOffset.
ClearMiddleReadFragment(uint64_t minOffset,uint64_t maxOffset)1124 bool CacheMediaChunkBufferImpl::ClearMiddleReadFragment(uint64_t minOffset, uint64_t maxOffset)
1125 {
1126 std::lock_guard lock(mutex_);
1127 bool res = false;
1128 for (auto iter = fragmentCacheBuffer_.begin(); iter != fragmentCacheBuffer_.end(); iter++) {
1129 if (iter->offsetBegin + static_cast<uint64_t>(iter->dataLength) < minOffset) {
1130 continue;
1131 }
1132 if (iter->offsetBegin > maxOffset) {
1133 break;
1134 }
1135 if (iter->accessLength <= chunkSize_) {
1136 continue;
1137 }
1138 MEDIA_LOG_D("ClearMiddleReadFragment, minOffset: " PUBLIC_LOG_U64 " maxOffset: "
1139 PUBLIC_LOG_U64 " offsetBegin: " PUBLIC_LOG_U64 " dataLength: " PUBLIC_LOG_D64 " accessLength "
1140 PUBLIC_LOG_D64, minOffset, maxOffset, iter->offsetBegin, iter->dataLength, iter->accessLength);
1141 auto& fragment = *iter;
1142 uint32_t chunksSize = fragment.chunks.size();
1143 for (uint32_t i = 0; i < chunksSize; ++i) {
1144 auto chunkIter = fragment.chunks.front();
1145 if (chunkIter->dataLength >= iter->accessLength ||
1146 (chunkIter->offset + chunkIter->dataLength >= maxOffset &&
1147 chunkIter->offset <= minOffset)) {
1148 break;
1149 }
1150 auto tmp = UpdateFragmentCacheForDelHead(iter);
1151 if (tmp != nullptr) {
1152 freeChunks_.push_back(tmp);
1153 }
1154 }
1155 }
1156 return res;
1157 }
1158
IsReadSplit(uint64_t offset)1159 bool CacheMediaChunkBufferImpl::IsReadSplit(uint64_t offset)
1160 {
1161 std::lock_guard lock(mutex_);
1162 auto readPos = GetOffsetFragmentCache(readPos_, offset, LeftBoundedRightOpenComp);
1163 if (readPos != fragmentCacheBuffer_.end()) {
1164 return readPos->isSplit;
1165 }
1166 return false;
1167 }
1168
SetIsLargeOffsetSpan(bool isLargeOffsetSpan)1169 void CacheMediaChunkBufferImpl::SetIsLargeOffsetSpan(bool isLargeOffsetSpan)
1170 {
1171 isLargeOffsetSpan_ = isLargeOffsetSpan;
1172 }
1173
DumpAndCheckInner()1174 bool CacheMediaChunkBufferImpl::DumpAndCheckInner()
1175 {
1176 DumpInner(0);
1177 return CheckInner();
1178 }
1179
CheckFragment(const FragmentCacheBuffer & fragment,bool & checkSuccess)1180 void CacheMediaChunkBufferImpl::CheckFragment(const FragmentCacheBuffer& fragment, bool& checkSuccess)
1181 {
1182 if (fragment.accessPos != fragment.chunks.end()) {
1183 auto& accessChunk = *fragment.accessPos;
1184 auto accessLength = accessChunk->offset > fragment.offsetBegin ?
1185 accessChunk->offset - fragment.offsetBegin : 0;
1186 if (fragment.accessLength < accessLength ||
1187 fragment.accessLength >
1188 (static_cast<int64_t>(accessLength) + static_cast<int64_t>(accessChunk->dataLength))) {
1189 checkSuccess = false;
1190 }
1191 }
1192 }
1193
CheckInner()1194 bool CacheMediaChunkBufferImpl::CheckInner()
1195 {
1196 uint64_t chunkNum = 0;
1197 uint64_t totalReadSize = 0;
1198 bool checkSuccess = true;
1199 chunkNum = freeChunks_.size();
1200 for (auto const& fragment : fragmentCacheBuffer_) {
1201 int64_t dataLength = 0;
1202 chunkNum += fragment.chunks.size();
1203 totalReadSize += fragment.totalReadSize;
1204
1205 auto prev = fragment.chunks.begin();
1206 auto next = fragment.chunks.end();
1207 if (!fragment.chunks.empty()) {
1208 dataLength += static_cast<int64_t>((*prev)->dataLength);
1209 next = std::next(prev);
1210 if ((*prev)->offset != fragment.offsetBegin) {
1211 checkSuccess = false;
1212 }
1213 }
1214 while (next != fragment.chunks.end()) {
1215 auto &chunkPrev = *prev;
1216 auto &chunkNext = *next;
1217 dataLength += static_cast<int64_t>(chunkNext->dataLength);
1218 if (chunkPrev->offset + chunkPrev->dataLength != chunkNext->offset) {
1219 checkSuccess = false;
1220 }
1221 ++next;
1222 ++prev;
1223 }
1224 if (dataLength != fragment.dataLength) {
1225 checkSuccess = false;
1226 }
1227 CheckFragment(fragment, checkSuccess);
1228 }
1229 if (chunkNum != chunkMaxNum_ + 1) {
1230 checkSuccess = false;
1231 }
1232
1233 if (totalReadSize != totalReadSize_) {
1234 checkSuccess = false;
1235 }
1236 return checkSuccess;
1237 }
1238
1239
CacheMediaChunkBuffer()1240 CacheMediaChunkBuffer::CacheMediaChunkBuffer()
1241 {
1242 MEDIA_LOG_D("enter");
1243 impl_ = std::make_unique<CacheMediaChunkBufferImpl>();
1244 };
1245
~CacheMediaChunkBuffer()1246 CacheMediaChunkBuffer::~CacheMediaChunkBuffer()
1247 {
1248 MEDIA_LOG_D("exit");
1249 }
1250
Init(uint64_t totalBuffSize,uint32_t chunkSize)1251 bool CacheMediaChunkBuffer::Init(uint64_t totalBuffSize, uint32_t chunkSize)
1252 {
1253 return impl_->Init(totalBuffSize, chunkSize);
1254 }
1255
Read(void * ptr,uint64_t offset,size_t readSize)1256 size_t CacheMediaChunkBuffer::Read(void* ptr, uint64_t offset, size_t readSize)
1257 {
1258 return impl_->Read(ptr, offset, readSize);
1259 }
1260
Write(void * ptr,uint64_t offset,size_t writeSize)1261 size_t CacheMediaChunkBuffer::Write(void* ptr, uint64_t offset, size_t writeSize)
1262 {
1263 return impl_->Write(ptr, offset, writeSize);
1264 }
1265
Seek(uint64_t offset)1266 bool CacheMediaChunkBuffer::Seek(uint64_t offset)
1267 {
1268 return impl_->Seek(offset);
1269 }
1270
GetBufferSize(uint64_t offset)1271 size_t CacheMediaChunkBuffer::GetBufferSize(uint64_t offset)
1272 {
1273 return impl_->GetBufferSize(offset);
1274 }
1275
GetNextBufferOffset(uint64_t offset)1276 uint64_t CacheMediaChunkBuffer::GetNextBufferOffset(uint64_t offset)
1277 {
1278 return impl_->GetNextBufferOffset(offset);
1279 }
1280
Clear()1281 void CacheMediaChunkBuffer::Clear()
1282 {
1283 return impl_->Clear();
1284 }
1285
GetFreeSize()1286 uint64_t CacheMediaChunkBuffer::GetFreeSize()
1287 {
1288 return impl_->GetFreeSize();
1289 }
1290
ClearFragmentBeforeOffset(uint64_t offset)1291 bool CacheMediaChunkBuffer::ClearFragmentBeforeOffset(uint64_t offset)
1292 {
1293 return impl_->ClearFragmentBeforeOffset(offset);
1294 }
1295
ClearChunksOfFragment(uint64_t offset)1296 bool CacheMediaChunkBuffer::ClearChunksOfFragment(uint64_t offset)
1297 {
1298 return impl_->ClearChunksOfFragment(offset);
1299 }
1300
ClearMiddleReadFragment(uint64_t minOffset,uint64_t maxOffset)1301 bool CacheMediaChunkBuffer::ClearMiddleReadFragment(uint64_t minOffset, uint64_t maxOffset)
1302 {
1303 return impl_->ClearMiddleReadFragment(minOffset, maxOffset);
1304 }
1305
IsReadSplit(uint64_t offset)1306 bool CacheMediaChunkBuffer::IsReadSplit(uint64_t offset)
1307 {
1308 return impl_->IsReadSplit(offset);
1309 }
1310
SetIsLargeOffsetSpan(bool isLargeOffsetSpan)1311 void CacheMediaChunkBuffer::SetIsLargeOffsetSpan(bool isLargeOffsetSpan)
1312 {
1313 return impl_->SetIsLargeOffsetSpan(isLargeOffsetSpan);
1314 }
1315
SetReadBlocking(bool isReadBlockingAllowed)1316 void CacheMediaChunkBuffer::SetReadBlocking(bool isReadBlockingAllowed)
1317 {
1318 (void)isReadBlockingAllowed;
1319 }
1320
Dump(uint64_t param)1321 void CacheMediaChunkBuffer::Dump(uint64_t param)
1322 {
1323 return impl_->Dump(param);
1324 }
1325
Check()1326 bool CacheMediaChunkBuffer::Check()
1327 {
1328 return impl_->Check();
1329 }
1330 }
1331 }