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