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