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