xbmc/filesystem/FileCache.cpp

   1 /*
   2  *  Copyright (C) 2005-2018 Team Kodi
   3  *  This file is part of Kodi - https://kodi.tv
   4  *
   5  *  SPDX-License-Identifier: GPL-2.0-or-later
   6  *  See LICENSES/README.md for more information.
   7  */
   8
   9 #include "FileCache.h"
  10
  11 #include "CircularCache.h"
  12 #include "ServiceBroker.h"
  13 #include "URL.h"
  14 #include "settings/Settings.h"
  15 #include "settings/SettingsComponent.h"
  16 #include "threads/Thread.h"
  17 #include "utils/log.h"
  18
  19 #include <mutex>
  20
  21 #if !defined(TARGET_WINDOWS)
  22 #include "platform/posix/ConvUtils.h"
  23 #endif
  24
  25 #include <algorithm>
  26 #include <cassert>
  27 #include <inttypes.h>
  28 #include <memory>
  29
  30 #ifdef TARGET_POSIX
  31 #include "platform/posix/ConvUtils.h"
  32 #endif
  33
  34 using namespace XFILE;
  35
  36 class CWriteRate
  37 {
  38 public:
  39   CWriteRate() : m_stamp(std::chrono::steady_clock::now()), m_time(std::chrono::milliseconds(0))
  40   {
  41     m_pos   = 0;
  42     m_size = 0;
  43   }
  44
  45   void Reset(int64_t pos, bool bResetAll = true)
  46   {
  47     m_stamp = std::chrono::steady_clock::now();
  48     m_pos   = pos;
  49
  50     if (bResetAll)
  51     {
  52       m_size  = 0;
  53       m_time = std::chrono::milliseconds(0);
  54     }
  55   }
  56
  57   uint32_t Rate(int64_t pos, uint32_t time_bias = 0)
  58   {
  59     auto ts = std::chrono::steady_clock::now();
  60
  61     m_size += (pos - m_pos);
  62     m_time += std::chrono::duration_cast<std::chrono::milliseconds>(ts - m_stamp);
  63     m_pos = pos;
  64     m_stamp = ts;
  65
  66     if (m_time == std::chrono::milliseconds(0))
  67       return 0;
  68
  69     return static_cast<uint32_t>(1000 * (m_size / (m_time.count() + time_bias)));
  70   }
  71
  72 private:
  73   std::chrono::time_point<std::chrono::steady_clock> m_stamp;
  74   int64_t  m_pos;
  75   std::chrono::milliseconds m_time;
  76   int64_t  m_size;
  77 };
  78
  79 CFileCache::CFileCache(const unsigned int flags)
  80   : CThread("FileCache"), m_fileSize(0), m_flags(flags)
  81 {
  82 }
  83
  84 CFileCache::~CFileCache()
  85 {
  86   Close();
  87 }
  88
  89 IFile *CFileCache::GetFileImp()
  90 {
  91   return m_source.GetImplementation();
  92 }
  93
  94 bool CFileCache::Open(const CURL& url)
  95 {
  96   Close();
  97
  98   std::unique_lock<CCriticalSection> lock(m_sync);
  99
 100   m_sourcePath = url.GetRedacted();
 101
 102   CLog::Log(LOGDEBUG, "CFileCache::{} - <{}> opening", __FUNCTION__, m_sourcePath);
 103
 104   // opening the source file.
 105   if (!m_source.Open(url.Get(), READ_NO_CACHE | READ_TRUNCATED | READ_CHUNKED))
 106   {
 107     CLog::Log(LOGERROR, "CFileCache::{} - <{}> failed to open", __FUNCTION__, m_sourcePath);
 108     Close();
 109     return false;
 110   }
 111
 112   const auto settings = CServiceBroker::GetSettingsComponent()->GetSettings();
 113   if (!settings)
 114     return false;
 115
 116   const unsigned int cacheMemSize =
 117       settings->GetInt(CSettings::SETTING_FILECACHE_MEMORYSIZE) * 1024 * 1024;
 118
 119   m_source.IoControl(IOCTRL_SET_CACHE, this);
 120
 121   bool retry = false;
 122   m_source.IoControl(IOCTRL_SET_RETRY, &retry); // We already handle retrying ourselves
 123
 124   // check if source can seek
 125   m_seekPossible = m_source.IoControl(IOCTRL_SEEK_POSSIBLE, NULL);
 126
 127   // Determine the best chunk size we can use
 128   m_chunkSize = CFile::DetermineChunkSize(m_source.GetChunkSize(),
 129                                           settings->GetInt(CSettings::SETTING_FILECACHE_CHUNKSIZE));
 130   CLog::Log(LOGDEBUG,
 131             "CFileCache::{} - <{}> source chunk size is {}, setting cache chunk size to {}",
 132             __FUNCTION__, m_sourcePath, m_source.GetChunkSize(), m_chunkSize);
 133
 134   m_fileSize = m_source.GetLength();
 135
 136   if (!m_pCache)
 137   {
 138     if (cacheMemSize == 0)
 139     {
 140       // Use cache on disk
 141       m_pCache = std::make_unique<CSimpleFileCache>();
 142       m_forwardCacheSize = 0;
 143       m_maxForward = m_fileSize;
 144     }
 145     else
 146     {
 147       size_t cacheSize;
 148       if (m_fileSize > 0 && m_fileSize < cacheMemSize && !(m_flags & READ_AUDIO_VIDEO))
 149       {
 150         // Cap cache size by filesize, but not for audio/video files as those may grow.
 151         // We don't need to take into account READ_MULTI_STREAM here as that's only used for audio/video
 152         cacheSize = m_fileSize;
 153
 154         // Cap chunk size by cache size
 155         if (m_chunkSize > cacheSize)
 156           m_chunkSize = cacheSize;
 157       }
 158       else
 159       {
 160         cacheSize = cacheMemSize;
 161
 162         // NOTE: READ_MULTI_STREAM is only used with READ_AUDIO_VIDEO
 163         if (m_flags & READ_MULTI_STREAM)
 164         {
 165           // READ_MULTI_STREAM requires double buffering, so use half the amount of memory for each buffer
 166           cacheSize /= 2;
 167         }
 168
 169         // Make sure cache can at least hold 2 chunks
 170         if (cacheSize < m_chunkSize * 2)
 171           cacheSize = m_chunkSize * 2;
 172       }
 173
 174       if (m_flags & READ_MULTI_STREAM)
 175         CLog::Log(LOGDEBUG, "CFileCache::{} - <{}> using double memory cache each sized {} bytes",
 176                   __FUNCTION__, m_sourcePath, cacheSize);
 177       else
 178         CLog::Log(LOGDEBUG, "CFileCache::{} - <{}> using single memory cache sized {} bytes",
 179                   __FUNCTION__, m_sourcePath, cacheSize);
 180
 181       const size_t back = cacheSize / 4;
 182       const size_t front = cacheSize - back;
 183
 184       m_pCache = std::make_unique<CCircularCache>(front, back);
 185       m_forwardCacheSize = front;
 186       m_maxForward = m_forwardCacheSize;
 187     }
 188
 189     if (m_flags & READ_MULTI_STREAM)
 190     {
 191       // If READ_MULTI_STREAM flag is set: Double buffering is required
 192       m_pCache = std::make_unique<CDoubleCache>(m_pCache.release());
 193     }
 194   }
 195
 196   // open cache strategy
 197   if (!m_pCache || m_pCache->Open() != CACHE_RC_OK)
 198   {
 199     CLog::Log(LOGERROR, "CFileCache::{} - <{}> failed to open cache", __FUNCTION__, m_sourcePath);
 200     Close();
 201     return false;
 202   }
 203
 204   m_readPos = 0;
 205   m_writePos = 0;
 206   m_writeRate = 1024 * 1024;
 207   m_writeRateActual = 0;
 208   m_writeRateLowSpeed = 0;
 209   m_bFilling = true;
 210   m_seekEvent.Reset();
 211   m_seekEnded.Reset();
 212
 213   CThread::Create(false);
 214
 215   return true;
 216 }
 217
 218 void CFileCache::Process()
 219 {
 220   if (!m_pCache)
 221   {
 222     CLog::Log(LOGERROR, "CFileCache::{} - <{}> sanity failed. no cache strategy", __FUNCTION__,
 223               m_sourcePath);
 224     return;
 225   }
 226
 227   // create our read buffer
 228   std::unique_ptr<char[]> buffer(new char[m_chunkSize]);
 229   if (buffer == nullptr)
 230   {
 231     CLog::Log(LOGERROR, "CFileCache::{} - <{}> failed to allocate read buffer", __FUNCTION__,
 232               m_sourcePath);
 233     return;
 234   }
 235
 236   const auto settings = CServiceBroker::GetSettingsComponent()->GetSettings();
 237   if (!settings)
 238     return;
 239
 240   float readFactor = settings->GetInt(CSettings::SETTING_FILECACHE_READFACTOR) / 100.0f;
 241
 242   const bool useAdaptativeReadFactor = (readFactor < 1.0f);
 243
 244   CWriteRate limiter;
 245   CWriteRate average;
 246
 247   while (!m_bStop)
 248   {
 249     // Update filesize
 250     m_fileSize = m_source.GetLength();
 251
 252     // check for seek events
 253     if (m_seekEvent.Wait(0ms))
 254     {
 255       m_seekEvent.Reset();
 256       const int64_t cacheMaxPos = m_pCache->CachedDataEndPosIfSeekTo(m_seekPos);
 257       const bool cacheReachEOF = (cacheMaxPos == m_fileSize);
 258
 259       bool sourceSeekFailed = false;
 260       if (!cacheReachEOF)
 261       {
 262         m_nSeekResult = m_source.Seek(cacheMaxPos, SEEK_SET);
 263         if (m_nSeekResult != cacheMaxPos)
 264         {
 265           CLog::Log(LOGERROR, "CFileCache::{} - <{}> error {} seeking. Seek returned {}",
 266                     __FUNCTION__, m_sourcePath, GetLastError(), m_nSeekResult);
 267           m_seekPossible = m_source.IoControl(IOCTRL_SEEK_POSSIBLE, NULL);
 268           sourceSeekFailed = true;
 269         }
 270       }
 271
 272       if (!sourceSeekFailed)
 273       {
 274         const bool bCompleteReset = m_pCache->Reset(m_seekPos);
 275         m_readPos = m_seekPos;
 276         m_writePos = m_pCache->CachedDataEndPos();
 277         assert(m_writePos == cacheMaxPos);
 278         average.Reset(m_writePos, bCompleteReset); // Can only recalculate new average from scratch after a full reset (empty cache)
 279         limiter.Reset(m_writePos);
 280         m_nSeekResult = m_seekPos;
 281         if (bCompleteReset)
 282         {
 283           CLog::Log(LOGDEBUG,
 284                     "CFileCache::{} - <{}> cache completely reset for seek to position {}",
 285                     __FUNCTION__, m_sourcePath, m_seekPos);
 286           m_bFilling = true;
 287           m_writeRateLowSpeed = 0;
 288         }
 289       }
 290
 291       m_seekEnded.Set();
 292     }
 293
 294     // variable read factor based on cache level
 295     if (useAdaptativeReadFactor)
 296     {
 297       // cache level [0.0 - 1.0]
 298       const double level = static_cast<double>(m_writePos - m_readPos) / m_maxForward;
 299       readFactor = static_cast<float>(level * -2.5 + 4.0); // read factor [4.0x - 1.5x]
 300     }
 301
 302     while (m_writeRate)
 303     {
 304       if (m_writePos - m_readPos < m_writeRate * readFactor)
 305       {
 306         limiter.Reset(m_writePos);
 307         break;
 308       }
 309
 310       if (limiter.Rate(m_writePos) < m_writeRate * readFactor)
 311         break;
 312
 313       if (m_seekEvent.Wait(m_processWait))
 314       {
 315         if (!m_bStop)
 316           m_seekEvent.Set();
 317         break;
 318       }
 319     }
 320
 321     const int64_t maxWrite = m_pCache->GetMaxWriteSize(m_chunkSize);
 322     int64_t maxSourceRead = m_chunkSize;
 323     // Cap source read size by space available between current write position and EOF
 324     if (m_fileSize != 0)
 325       maxSourceRead = std::min(maxSourceRead, m_fileSize - m_writePos);
 326
 327     /* Only read from source if there's enough write space in the cache
 328      * else we may keep disposing data and seeking back on (slow) source
 329      */
 330     if (maxWrite < maxSourceRead)
 331     {
 332       // Wait until sufficient cache write space is available
 333       m_pCache->m_space.Wait(5ms);
 334       continue;
 335     }
 336
 337     ssize_t iRead = 0;
 338     if (maxSourceRead > 0)
 339       iRead = m_source.Read(buffer.get(), maxSourceRead);
 340     if (iRead <= 0)
 341     {
 342       // Check for actual EOF and retry as long as we still have data in our cache
 343       if (m_writePos < m_fileSize && m_pCache->WaitForData(0, 0ms) > 0)
 344       {
 345         CLog::Log(LOGWARNING, "CFileCache::{} - <{}> source read returned {}! Will retry",
 346                   __FUNCTION__, m_sourcePath, iRead);
 347
 348         // Wait a bit:
 349         if (m_seekEvent.Wait(2000ms))
 350         {
 351           if (!m_bStop)
 352             m_seekEvent.Set(); // hack so that later we realize seek is needed
 353         }
 354
 355         // and retry:
 356         continue; // while (!m_bStop)
 357       }
 358       else
 359       {
 360         if (iRead < 0)
 361           CLog::Log(LOGERROR,
 362                     "{} - <{}> source read failed with {}!", __FUNCTION__, m_sourcePath, iRead);
 363         else if (m_fileSize == 0)
 364           CLog::Log(LOGDEBUG,
 365                     "CFileCache::{} - <{}> source read didn't return any data! Hit eof(?)",
 366                     __FUNCTION__, m_sourcePath);
 367         else if (m_writePos < m_fileSize)
 368           CLog::Log(LOGERROR,
 369                     "CFileCache::{} - <{}> source read didn't return any data before eof!",
 370                     __FUNCTION__, m_sourcePath);
 371         else
 372           CLog::Log(LOGDEBUG, "CFileCache::{} - <{}> source read hit eof", __FUNCTION__,
 373                     m_sourcePath);
 374
 375         m_pCache->EndOfInput();
 376
 377         // The thread event will now also cause the wait of an event to return a false.
 378         if (AbortableWait(m_seekEvent) == WAIT_SIGNALED)
 379         {
 380           m_pCache->ClearEndOfInput();
 381           if (!m_bStop)
 382             m_seekEvent.Set(); // hack so that later we realize seek is needed
 383         }
 384         else
 385           break; // while (!m_bStop)
 386       }
 387     }
 388
 389     int iTotalWrite = 0;
 390     while (!m_bStop && (iTotalWrite < iRead))
 391     {
 392       int iWrite = 0;
 393       iWrite = m_pCache->WriteToCache(buffer.get() + iTotalWrite, iRead - iTotalWrite);
 394
 395       // write should always work. all handling of buffering and errors should be
 396       // done inside the cache strategy. only if unrecoverable error happened, WriteToCache would return error and we break.
 397       if (iWrite < 0)
 398       {
 399         CLog::Log(LOGERROR, "CFileCache::{} - <{}> error writing to cache", __FUNCTION__,
 400                   m_sourcePath);
 401         m_bStop = true;
 402         break;
 403       }
 404       else if (iWrite == 0)
 405       {
 406         m_pCache->m_space.Wait(5ms);
 407       }
 408
 409       iTotalWrite += iWrite;
 410
 411       // check if seek was asked. otherwise if cache is full we'll freeze.
 412       if (m_seekEvent.Wait(0ms))
 413       {
 414         if (!m_bStop)
 415           m_seekEvent.Set(); // make sure we get the seek event later.
 416         break;
 417       }
 418     }
 419
 420     m_writePos += iTotalWrite;
 421
 422     // under estimate write rate by a second, to
 423     // avoid uncertainty at start of caching
 424     m_writeRateActual = average.Rate(m_writePos, 1000);
 425
 426    /* NOTE: We can only reliably test for low speed condition, when the cache is *really*
 427     * filling. This is because as soon as it's full the average-
 428     * rate will become approximately the current-rate which can flag false
 429     * low read-rate conditions.
 430     */
 431     if (m_bFilling && m_forwardCacheSize != 0)
 432     {
 433       const int64_t forward = m_pCache->WaitForData(0, 0ms);
 434       if (forward + m_chunkSize >= m_forwardCacheSize)
 435       {
 436         if (m_writeRateActual < m_writeRate)
 437           m_writeRateLowSpeed = m_writeRateActual;
 438
 439         m_bFilling = false;
 440       }
 441     }
 442   }
 443 }
 444
 445 void CFileCache::OnExit()
 446 {
 447   m_bStop = true;
 448
 449   // make sure cache is set to mark end of file (read may be waiting).
 450   if (m_pCache)
 451     m_pCache->EndOfInput();
 452
 453   // just in case someone's waiting...
 454   m_seekEnded.Set();
 455 }
 456
 457 bool CFileCache::Exists(const CURL& url)
 458 {
 459   return CFile::Exists(url.Get());
 460 }
 461
 462 int CFileCache::Stat(const CURL& url, struct __stat64* buffer)
 463 {
 464   return CFile::Stat(url.Get(), buffer);
 465 }
 466
 467 ssize_t CFileCache::Read(void* lpBuf, size_t uiBufSize)
 468 {
 469   std::unique_lock<CCriticalSection> lock(m_sync);
 470   if (!m_pCache)
 471   {
 472     CLog::Log(LOGERROR, "CFileCache::{} - <{}> sanity failed. no cache strategy!", __FUNCTION__,
 473               m_sourcePath);
 474     return -1;
 475   }
 476   int64_t iRc;
 477
 478   if (uiBufSize > SSIZE_MAX)
 479     uiBufSize = SSIZE_MAX;
 480
 481 retry:
 482   // attempt to read
 483   iRc = m_pCache->ReadFromCache((char *)lpBuf, uiBufSize);
 484   if (iRc > 0)
 485   {
 486     m_readPos += iRc;
 487     return (int)iRc;
 488   }
 489
 490   if (iRc == CACHE_RC_WOULD_BLOCK)
 491   {
 492     // just wait for some data to show up
 493     iRc = m_pCache->WaitForData(1, 10s);
 494     if (iRc > 0)
 495       goto retry;
 496   }
 497
 498   if (iRc == CACHE_RC_TIMEOUT)
 499   {
 500     CLog::Log(LOGWARNING, "CFileCache::{} - <{}> timeout waiting for data", __FUNCTION__,
 501               m_sourcePath);
 502     return -1;
 503   }
 504
 505   if (iRc == 0)
 506     return 0;
 507
 508   // unknown error code
 509   CLog::Log(LOGERROR, "CFileCache::{} - <{}> cache strategy returned unknown error code {}",
 510             __FUNCTION__, m_sourcePath, (int)iRc);
 511   return -1;
 512 }
 513
 514 int64_t CFileCache::Seek(int64_t iFilePosition, int iWhence)
 515 {
 516   std::unique_lock<CCriticalSection> lock(m_sync);
 517
 518   if (!m_pCache)
 519   {
 520     CLog::Log(LOGERROR, "CFileCache::{} - <{}> sanity failed. no cache strategy!", __FUNCTION__,
 521               m_sourcePath);
 522     return -1;
 523   }
 524
 525   int64_t iCurPos = m_readPos;
 526   int64_t iTarget = iFilePosition;
 527   if (iWhence == SEEK_END)
 528     iTarget = m_fileSize + iTarget;
 529   else if (iWhence == SEEK_CUR)
 530     iTarget = iCurPos + iTarget;
 531   else if (iWhence != SEEK_SET)
 532     return -1;
 533
 534   if (iTarget == m_readPos)
 535     return m_readPos;
 536
 537   if ((m_nSeekResult = m_pCache->Seek(iTarget)) != iTarget)
 538   {
 539     if (m_seekPossible == 0)
 540       return m_nSeekResult;
 541
 542     // Never request closer to end than one chunk. Speeds up tag reading
 543     m_seekPos = std::min(iTarget, std::max((int64_t)0, m_fileSize - m_chunkSize));
 544
 545     m_seekEvent.Set();
 546     while (!m_seekEnded.Wait(100ms))
 547     {
 548       // SeekEnded will never be set if FileCache thread is not running
 549       if (!CThread::IsRunning())
 550         return -1;
 551     }
 552
 553     /* wait for any remaining data */
 554     if(m_seekPos < iTarget)
 555     {
 556       CLog::Log(LOGDEBUG, "CFileCache::{} - <{}> waiting for position {}", __FUNCTION__,
 557                 m_sourcePath, iTarget);
 558       if (m_pCache->WaitForData(static_cast<uint32_t>(iTarget - m_seekPos), 10s) <
 559           iTarget - m_seekPos)
 560       {
 561         CLog::Log(LOGWARNING, "CFileCache::{} - <{}> failed to get remaining data", __FUNCTION__,
 562                   m_sourcePath);
 563         return -1;
 564       }
 565       m_pCache->Seek(iTarget);
 566     }
 567     m_readPos = iTarget;
 568     m_seekEvent.Reset();
 569   }
 570   else
 571     m_readPos = iTarget;
 572
 573   return iTarget;
 574 }
 575
 576 void CFileCache::Close()
 577 {
 578   StopThread();
 579
 580   std::unique_lock<CCriticalSection> lock(m_sync);
 581   if (m_pCache)
 582     m_pCache->Close();
 583
 584   m_source.Close();
 585 }
 586
 587 int64_t CFileCache::GetPosition()
 588 {
 589   return m_readPos;
 590 }
 591
 592 int64_t CFileCache::GetLength()
 593 {
 594   return m_fileSize;
 595 }
 596
 597 void CFileCache::StopThread(bool bWait /*= true*/)
 598 {
 599   m_bStop = true;
 600   //Process could be waiting for seekEvent
 601   m_seekEvent.Set();
 602   CThread::StopThread(bWait);
 603 }
 604
 605 const std::string CFileCache::GetProperty(XFILE::FileProperty type, const std::string &name) const
 606 {
 607   if (!m_source.GetImplementation())
 608     return IFile::GetProperty(type, name);
 609
 610   return m_source.GetImplementation()->GetProperty(type, name);
 611 }
 612
 613 int CFileCache::IoControl(EIoControl request, void* param)
 614 {
 615   if (request == IOCTRL_CACHE_STATUS)
 616   {
 617     SCacheStatus* status = (SCacheStatus*)param;
 618     status->maxforward = m_maxForward;
 619     status->forward = m_pCache->WaitForData(0, 0ms);
 620     status->maxrate = m_writeRate;
 621     status->currate = m_writeRateActual;
 622     status->lowrate = m_writeRateLowSpeed;
 623     m_writeRateLowSpeed = 0; // Reset low speed condition
 624     return 0;
 625   }
 626
 627   if (request == IOCTRL_CACHE_SETRATE)
 628   {
 629     m_writeRate = *static_cast<uint32_t*>(param);
 630
 631     const double mBits = m_writeRate / 1024.0 / 1024.0 * 8.0; // Mbit/s
 632
 633     // calculates wait time inversely proportional to the bitrate
 634     // and limited between 30 - 100 ms
 635     const int wait = std::clamp(static_cast<int>(110.0 - mBits), 30, 100);
 636
 637     m_processWait = std::chrono::milliseconds(wait);
 638
 639     CLog::Log(LOGDEBUG,
 640               "CFileCache::IoControl - setting maxRate to {:.2f} Mbit/s with processWait of {} ms",
 641               mBits, wait);
 642     return 0;
 643   }
 644
 645   if (request == IOCTRL_SEEK_POSSIBLE)
 646     return m_seekPossible;
 647
 648   return -1;
 649 }