scheduler.cpp

#include "pls/internal/scheduling/scheduler.h"
#include "pls/internal/scheduling/thread_state.h"
#include "pls/internal/scheduling/task.h"

#include "pls/internal/base/error_handling.h"

namespace pls {
namespace internal {
namespace scheduling {

scheduler::scheduler(scheduler_memory &memory, const unsigned int num_threads, bool reuse_thread) :
    num_threads_{num_threads},
    reuse_thread_{reuse_thread},
    memory_{memory},
    sync_barrier_{num_threads + 1 - reuse_thread},
    terminated_{false} {
  if (num_threads_ > memory.max_threads()) {
    PLS_ERROR("Tried to create scheduler with more OS threads than pre-allocated memory.");
  }

  for (unsigned int i = 0; i < num_threads_; i++) {
    // Placement new is required, as the memory of `memory_` is not required to be initialized.
    memory.thread_state_for(i).scheduler_ = this;
    memory.thread_state_for(i).id_ = i;

    if (reuse_thread && i == 0) {
      continue; // Skip over first/main thread when re-using the users thread, as this one will replace the first one.
    }
    memory.thread_for(i) = base::thread(&scheduler::work_thread_main_loop, &memory_.thread_state_for(i));
  }
}

scheduler::~scheduler() {
  terminate();
}

void scheduler::work_thread_main_loop() {
  auto &scheduler = thread_state::get().get_scheduler();
  while (true) {
    // Wait to be triggered
    scheduler.sync_barrier_.wait();

    // Check for shutdown
    if (scheduler.terminated_) {
      return;
    }

    scheduler.work_thread_work_section();

    // Sync back with main thread
    scheduler.sync_barrier_.wait();
  }
}

void scheduler::work_thread_work_section() {
  auto &my_state = thread_state::get();

  if (my_state.get_id() == 0) {
    // Main Thread, kick of by executing the user's main code block.
    main_thread_starter_function_->run();
  }

  do {
    // TODO: Implement other threads, for now we are happy if it compiles and runs on one thread
    //       For now we can test without this, as the fast path should never hit this.
    // 1) Try Steal
    // 2) Copy Over
    // 3) Finish Steal
    // 4) Execute Local Copy
  } while (!work_section_done_);
}

void scheduler::terminate() {
  if (terminated_) {
    return;
  }

  terminated_ = true;
  sync_barrier_.wait();

  for (unsigned int i = 0; i < num_threads_; i++) {
    if (reuse_thread_ && i == 0) {
      continue;
    }
    memory_.thread_for(i).join();
  }
}

thread_state &scheduler::thread_state_for(size_t id) { return memory_.thread_state_for(id); }

}
}
}