source/resources/resourcemanager.cpp

   1 #include <msp/core/algorithm.h>
   2 #include <typeinfo>
   3 #include <msp/debug/demangle.h>
   4 #include <msp/strings/format.h>
   5 #include <msp/time/utils.h>
   6 #include "resourcemanager.h"
   7 #include "resources.h"
   8 #include "resourceobserver.h"
   9
  10 using namespace std;
  11
  12 namespace Msp {
  13 namespace GL {
  14
  15 resource_load_error::resource_load_error(const string &name, const string &err):
  16         runtime_error(format("%s: %s", name, err))
  17 { }
  18
  19 resource_load_error::resource_load_error(const string &name, const exception &exc):
  20         runtime_error(format("%s: %s: %s", name, Debug::demangle(typeid(exc).name()), exc.what()))
  21 { }
  22
  23
  24 ResourceManager::ResourceManager():
  25         policy(LOAD_ON_DEMAND),
  26         async_loads(true),
  27         total_data_size(0),
  28         size_limit(0),
  29         frame(0),
  30         min_retain_frames(30),
  31         max_retain_frames(0),
  32         next_unload(0)
  33 { }
  34
  35 ResourceManager::~ResourceManager()
  36 {
  37         thread.terminate();
  38
  39         while(!resources.empty())
  40                 resources.begin()->second.resource->set_manager(0);
  41 }
  42
  43 void ResourceManager::set_loading_policy(LoadingPolicy p)
  44 {
  45         policy = p;
  46 }
  47
  48 void ResourceManager::set_async_loads(bool a)
  49 {
  50         async_loads = a;
  51 }
  52
  53 void ResourceManager::set_size_limit(uint64_t s)
  54 {
  55         size_limit = s;
  56 }
  57
  58 void ResourceManager::set_min_retain_frames(unsigned f)
  59 {
  60         min_retain_frames = f;
  61 }
  62
  63 void ResourceManager::set_max_retain_frames(unsigned f)
  64 {
  65         max_retain_frames = f;
  66 }
  67
  68 void ResourceManager::add_resource(Resource &r)
  69 {
  70         MutexLock lock(map_mutex);
  71         insert_unique(resources, &r, ManagedResource(r));
  72 }
  73
  74 const ResourceManager::ManagedResource &ResourceManager::get_managed_resource(const Resource &r) const
  75 {
  76         MutexLock lock(map_mutex);
  77         return get_item(resources, &r);
  78 }
  79
  80 ResourceManager::ManagedResource &ResourceManager::get_managed_resource(const Resource &r)
  81 {
  82         MutexLock lock(map_mutex);
  83         return get_item(resources, &r);
  84 }
  85
  86 void ResourceManager::set_resource_location(Resource &r, DataFile::Collection &c, const string &n)
  87 {
  88         set_resource_location(r, ResourceLocation(c, n));
  89 }
  90
  91 void ResourceManager::set_resource_location(Resource &r, const ResourceLocation &l)
  92 {
  93         {
  94                 MutexLock lock(map_mutex);
  95                 ManagedResource &managed = get_item(resources, &r);
  96                 managed.location = l;
  97         }
  98
  99         if(policy==LOAD_IMMEDIATELY)
 100                 load_resource(r);
 101 }
 102
 103 const ResourceManager::ResourceLocation *ResourceManager::get_resource_location(const Resource &r) const
 104 {
 105         const ManagedResource &managed = get_managed_resource(r);
 106         return managed.location.collection ? &managed.location : 0;
 107 }
 108
 109 void ResourceManager::load_resource(Resource &r)
 110 {
 111         ManagedResource &managed = get_managed_resource(r);
 112         if(!managed.location.collection)
 113                 throw runtime_error("no location");
 114
 115         if(managed.state!=ManagedResource::NOT_LOADED)
 116                 return;
 117
 118         if(async_loads)
 119         {
 120                 managed.state = ManagedResource::LOAD_QUEUED;
 121                 auto i = find_if(queue, [&managed](ManagedResource *q){ return q->load_priority>managed.load_priority; });
 122                 queue.insert(i, &managed);
 123         }
 124         else
 125         {
 126                 managed.start_loading();
 127                 while(!managed.loader->process()) ;
 128                 managed.finish_loading(true);
 129                 total_data_size += managed.data_size;
 130         }
 131 }
 132
 133 bool ResourceManager::is_resource_loaded(const Resource &r) const
 134 {
 135         ManagedResource *managed = reinterpret_cast<ManagedResource *>(r.get_manager_data());
 136         return managed ? managed->state==ManagedResource::LOADED : false;
 137 }
 138
 139 void ResourceManager::resource_used(const Resource &r)
 140 {
 141         ManagedResource *managed = reinterpret_cast<ManagedResource *>(r.get_manager_data());
 142         if(!managed)
 143                 return;
 144         if(managed->state==ManagedResource::NOT_LOADED && policy!=LOAD_MANUALLY)
 145                 load_resource(*managed->resource);
 146
 147         managed->last_used = frame;
 148         if(max_retain_frames && !next_unload)
 149                 next_unload = frame+max_retain_frames+1;
 150 }
 151
 152 void ResourceManager::remove_resource(Resource &r)
 153 {
 154         ManagedResource &managed = get_managed_resource(r);
 155         ManagedResource::State state = managed.state;
 156         if(state==ManagedResource::LOAD_QUEUED)
 157         {
 158                 auto i = find(queue, &managed);
 159                 if(i!=queue.end())
 160                         queue.erase(i);
 161         }
 162         else if(state>ManagedResource::LOAD_QUEUED && state<ManagedResource::LOADED)
 163                 thread.remove_resource(managed);
 164
 165         for(ResourceObserver *o: managed.observers)
 166                 o->resource_removed(r);
 167
 168         MutexLock lock(map_mutex);
 169         remove_existing(resources, &r);
 170 }
 171
 172 void ResourceManager::observe_resource(const Resource &r, ResourceObserver &w)
 173 {
 174         get_managed_resource(r).add_observer(w);
 175 }
 176
 177 void ResourceManager::unobserve_resource(const Resource &r, ResourceObserver &w)
 178 {
 179         get_managed_resource(r).remove_observer(w);
 180 }
 181
 182 void ResourceManager::tick()
 183 {
 184         ++frame;
 185
 186         bool do_unload = (frame>=next_unload);
 187         if(thread.sync())
 188         {
 189                 total_data_size += thread.get_and_reset_loaded_data_size();
 190                 do_unload = true;
 191         }
 192
 193         if(thread.needs_work() && !queue.empty())
 194                 dispatch_work();
 195
 196         if(do_unload)
 197         {
 198                 MutexLock lock(map_mutex);
 199                 if(max_retain_frames && frame>=next_unload)
 200                 {
 201                         unload_by_age();
 202
 203                         next_unload = frame;
 204                         for(const auto &kvp: resources)
 205                                 if(kvp.second.state==ManagedResource::LOADED)
 206                                         next_unload = min(next_unload, kvp.second.last_used);
 207                         next_unload = (next_unload<frame ? next_unload+max_retain_frames : 0);
 208                 }
 209
 210                 if(size_limit)
 211                         unload_by_size();
 212         }
 213 }
 214
 215 void ResourceManager::dispatch_work()
 216 {
 217         sort(queue, age_order);
 218
 219         if(queue.front()->last_used+min_retain_frames<frame)
 220         {
 221                 for(ManagedResource *r: queue)
 222                         r->state = ManagedResource::NOT_LOADED;
 223                 queue.clear();
 224                 return;
 225         }
 226
 227         while(thread.needs_work() && !queue.empty())
 228         {
 229                 ManagedResource *managed = queue.front();
 230                 queue.pop_front();
 231                 thread.add_resource(*managed);
 232         }
 233 }
 234
 235 void ResourceManager::unload_by_age()
 236 {
 237         unsigned unload_limit = frame-max_retain_frames;
 238
 239         for(auto &kvp: resources)
 240                 if(kvp.second.state==ManagedResource::LOADED && kvp.second.last_used<unload_limit)
 241                 {
 242                         kvp.second.unload();
 243                         total_data_size -= kvp.second.data_size;
 244                 }
 245 }
 246
 247 void ResourceManager::unload_by_size()
 248 {
 249         unsigned unload_limit = frame-min_retain_frames;
 250
 251         while(total_data_size>size_limit)
 252         {
 253                 ManagedResource *best = 0;
 254                 uint64_t best_impact = 0;
 255                 for(auto &kvp: resources)
 256                         if(kvp.second.state==ManagedResource::LOADED && kvp.second.last_used<unload_limit)
 257                         {
 258                                 uint64_t impact = kvp.second.data_size*(frame-kvp.second.last_used);
 259                                 if(!best || impact>best_impact)
 260                                 {
 261                                         best = &kvp.second;
 262                                         best_impact = impact;
 263                                 }
 264                         }
 265
 266                 if(!best)
 267                         break;
 268
 269                 best->unload();
 270                 total_data_size -= best->data_size;
 271         }
 272 }
 273
 274 bool ResourceManager::age_order(ManagedResource *mr1, ManagedResource *mr2)
 275 {
 276         return mr1->last_used>mr2->last_used;
 277 }
 278
 279
 280 ResourceManager::ResourceLocation::ResourceLocation():
 281         collection(0)
 282 { }
 283
 284 ResourceManager::ResourceLocation::ResourceLocation(DataFile::Collection &c, const string &n):
 285         collection(&c),
 286         name(n)
 287 { }
 288
 289
 290 ResourceManager::ManagedResource::ManagedResource(Resource &r):
 291         resource(&r),
 292         load_priority(r.get_load_priority()),
 293         io(0),
 294         loader(0),
 295         state(NOT_LOADED),
 296         last_used(0),
 297         data_size(0)
 298 { }
 299
 300 void ResourceManager::ManagedResource::start_loading()
 301 {
 302         io = location.collection->open_raw(location.name);
 303         if(!io)
 304                 throw resource_load_error(location.name, "open failed");
 305
 306         const Resources *res = dynamic_cast<Resources *>(location.collection);
 307         loader = resource->load(*io, res);
 308         if(!loader)
 309         {
 310                 delete io;
 311                 io = 0;
 312                 throw logic_error("no loader created");
 313         }
 314         state = LOADING;
 315 }
 316
 317 bool ResourceManager::ManagedResource::process(bool sync)
 318 {
 319         while(state!=LOAD_FINISHED && loader->needs_sync()==sync)
 320                 if(loader->process())
 321                         state = LOAD_FINISHED;
 322
 323         return state==LOAD_FINISHED;
 324 }
 325
 326 void ResourceManager::ManagedResource::finish_loading(bool successful)
 327 {
 328         delete loader;
 329         loader = 0;
 330         delete io;
 331         io = 0;
 332
 333         if(successful)
 334         {
 335                 state = LOADED;
 336                 data_size = resource->get_data_size();
 337
 338                 for(ResourceObserver *o: observers)
 339                         o->resource_loaded(*resource);
 340         }
 341         else
 342         {
 343                 resource->unload();
 344                 state = NOT_LOADED;
 345         }
 346 }
 347
 348 void ResourceManager::ManagedResource::finish_loading()
 349 {
 350         finish_loading(state==LOAD_FINISHED);
 351 }
 352
 353 void ResourceManager::ManagedResource::unload()
 354 {
 355         resource->unload();
 356         state = NOT_LOADED;
 357
 358         for(ResourceObserver *o: observers)
 359                 o->resource_unloaded(*resource);
 360 }
 361
 362 void ResourceManager::ManagedResource::add_observer(ResourceObserver &w)
 363 {
 364         if(find(observers.begin(), observers.end(), &w)==observers.end())
 365                 observers.push_back(&w);
 366 }
 367
 368 void ResourceManager::ManagedResource::remove_observer(ResourceObserver &w)
 369 {
 370         auto end = remove(observers.begin(), observers.end(), &w);
 371         if(end!=observers.end())
 372                 observers.erase(end, observers.end());
 373 }
 374
 375
 376 ResourceManager::LoadingThread::LoadingThread():
 377         Thread("ResourceManager"),
 378         sem(1),
 379         capacity(2),
 380         size(0),
 381         loaded_data_size(0),
 382         done(false)
 383 {
 384         launch();
 385 }
 386
 387 void ResourceManager::LoadingThread::main()
 388 {
 389         bool wait_for_work = false;
 390         while(!done)
 391         {
 392                 if(wait_for_work)
 393                         sem.wait();
 394
 395                 if(ManagedResource *managed = front(async_queue))
 396                 {
 397                         try
 398                         {
 399                                 managed->process(false);
 400                         }
 401                         catch(const exception &e)
 402                         {
 403                                 MutexLock lock(queue_mutex);
 404                                 error_queue.push_back(resource_load_error(managed->location.name, e));
 405                                 managed->state = ManagedResource::LOAD_ERROR;
 406                         }
 407
 408                         MutexLock lock(queue_mutex);
 409                         sync_queue.push_back(async_queue.front());
 410                         async_queue.pop_front();
 411                         wait_for_work = async_queue.empty();
 412                 }
 413                 else
 414                         wait_for_work = true;
 415         }
 416 }
 417
 418 ResourceManager::ManagedResource *ResourceManager::LoadingThread::front(deque<ManagedResource *> &que)
 419 {
 420         MutexLock lock(queue_mutex);
 421         if(que.empty())
 422                 return 0;
 423
 424         return que.front();
 425 }
 426
 427 void ResourceManager::LoadingThread::add_resource(ManagedResource &r)
 428 {
 429         r.start_loading();
 430
 431         MutexLock lock(queue_mutex);
 432         if(r.loader->needs_sync())
 433                 sync_queue.push_back(&r);
 434         else
 435         {
 436                 bool was_empty = async_queue.empty();
 437                 async_queue.push_back(&r);
 438                 if(was_empty)
 439                         sem.signal();
 440         }
 441
 442         ++size;
 443 }
 444
 445 void ResourceManager::LoadingThread::remove_resource(ManagedResource &r)
 446 {
 447         while(!try_remove_resource(r))
 448                 Time::sleep(Time::msec);
 449
 450         r.finish_loading();
 451         if(r.state==ManagedResource::LOADED)
 452         {
 453                 MutexLock lock(data_size_mutex);
 454                 loaded_data_size += r.data_size;
 455         }
 456 }
 457
 458 bool ResourceManager::LoadingThread::try_remove_resource(ManagedResource &r)
 459 {
 460         MutexLock lock(queue_mutex);
 461
 462         auto i = find(async_queue, &r);
 463         if(i==async_queue.end())
 464         {
 465                 i = find(sync_queue, &r);
 466                 if(i!=sync_queue.end())
 467                 {
 468                         sync_queue.erase(i);
 469                         --size;
 470                 }
 471         }
 472         else if(i==async_queue.begin())
 473                 return false;
 474         else
 475         {
 476                 async_queue.erase(i);
 477                 --size;
 478         }
 479
 480         return true;
 481 }
 482
 483 bool ResourceManager::LoadingThread::sync()
 484 {
 485         {
 486                 MutexLock lock(queue_mutex);
 487
 488                 if(!error_queue.empty())
 489                 {
 490                         resource_load_error err = error_queue.front();
 491                         error_queue.pop_front();
 492                         throw err;
 493                 }
 494
 495                 unsigned async_size = async_queue.size();
 496                 if(async_size==0 && size==capacity)
 497                         ++capacity;
 498                 else if(async_size>2 && capacity>2)
 499                         --capacity;
 500         }
 501
 502         bool any_finished = false;
 503         while(ManagedResource *managed = front(sync_queue))
 504         {
 505                 if(managed->state==ManagedResource::LOAD_ERROR || managed->process(true))
 506                 {
 507                         managed->finish_loading();
 508                         if(managed->state==ManagedResource::LOADED)
 509                         {
 510                                 MutexLock lock(data_size_mutex);
 511                                 loaded_data_size += managed->data_size;
 512                         }
 513                         any_finished = true;
 514                         --size;
 515
 516                         MutexLock lock(queue_mutex);
 517                         sync_queue.pop_front();
 518                 }
 519                 else
 520                 {
 521                         MutexLock lock(queue_mutex);
 522                         bool was_empty = async_queue.empty();
 523                         async_queue.push_back(sync_queue.front());
 524                         sync_queue.pop_front();
 525                         if(was_empty)
 526                                 sem.signal();
 527                 }
 528         }
 529
 530         return any_finished;
 531 }
 532
 533 uint64_t ResourceManager::LoadingThread::get_and_reset_loaded_data_size()
 534 {
 535         MutexLock lock(data_size_mutex);
 536         uint64_t result = loaded_data_size;
 537         loaded_data_size = 0;
 538         return result;
 539 }
 540
 541 void ResourceManager::LoadingThread::terminate()
 542 {
 543         done = true;
 544         sem.signal();
 545         join();
 546         async_queue.clear();
 547         sync_queue.clear();
 548 }
 549
 550 } // namespace GL
 551 } // namespace Msp