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