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