piece.cc

#include "piece.h"
#include "epoch.h"
#include "gopp/gopp.h"
#include "gopp/channels.h"

#include <queue>
#include "util/objects.h"
#include "util/arch.h"
#include "opts.h"
#include "mem.h"
#include "coro_sched.h"

using util::Instance;
using util::Impl;

namespace felis {

static constexpr size_t kPromiseRoutineHeader = sizeof(go::Routine);
static_assert(kPromiseRoutineHeader % 8 == 0); // Has to be aligned!

PieceRoutine *PieceRoutine::CreateFromCapture(size_t capture_len)
{
  auto r = (PieceRoutine *) BasePieceCollection::Alloc(sizeof(PieceRoutine));
  r->capture_len = capture_len;
  r->capture_data = (uint8_t *) BasePieceCollection::Alloc(util::Align(capture_len));
  r->sched_key = 0;
  r->level = 0;
  r->affinity = std::numeric_limits<uint64_t>::max();

  r->callback = nullptr;
  r->next = nullptr;
  r->fv_signals = 0;
  r->future_source_node_id = 0;
  std::fill(r->__padding__, r->__padding__ + 8, 0);
  return r;
}

PieceRoutine *
PieceRoutine::CreateFromPacket(uint8_t *p, size_t packet_len)
{
  auto r = (PieceRoutine *) BasePieceCollection::Alloc(sizeof(PieceRoutine));
  auto result_len = r->DecodeNode(p, packet_len);
  abort_if(result_len != packet_len,
           "DecodeNode() consumes {} but passed in {} bytes",
           result_len, packet_len);
  return r;
}

size_t PieceRoutine::NodeSize() const
{
  size_t s = util::Align(sizeof(PieceRoutine))
             + util::Align(capture_len)
             + 8;

  for (size_t i = 0; next && i < next->nr_handlers; i++) {
    auto *child = next->routine(i);
    s += child->NodeSize();
  }

  return s;
}

uint8_t *PieceRoutine::EncodeNode(uint8_t *p)
{
  memcpy(p, this, sizeof(PieceRoutine));
  auto node = (PieceRoutine *) p;
  p += util::Align(sizeof(PieceRoutine));

  memcpy(p, capture_data, capture_len);
  p += util::Align(capture_len);

  size_t nr_children = next ? next->nr_handlers : 0;
  memcpy(p, &nr_children, 8);
  p += 8;

  for (size_t i = 0; next && i < next->nr_handlers; i++) {
    auto *child = next->routine(i);
    p = child->EncodeNode(p);
  }
  return p;
}

size_t PieceRoutine::DecodeNode(uint8_t *p, size_t len)
{
  uint8_t *orig_p = p;
  size_t off = util::Align(sizeof(PieceRoutine));
  memcpy(this, p, off);
  p += off;

  off = util::Align(capture_len);
  capture_data = (uint8_t *) BasePieceCollection::Alloc(off);
  memcpy(capture_data, p, off);
  p += off;

  next = nullptr;

  size_t nr_children = 0;
  memcpy(&nr_children, p, 8);
  p += 8;

  if (nr_children > 0) {
    next = new BasePieceCollection(nr_children);
    for (int i = 0; i < nr_children; i++) {
      auto child = (PieceRoutine *) BasePieceCollection::Alloc(sizeof(PieceRoutine));
      off = child->DecodeNode(p, len - (p - orig_p));
      p += off;
      next->Add(child);
    }
  }
  return p - orig_p;
}

size_t BasePieceCollection::g_nr_threads = 0;

BasePieceCollection::BasePieceCollection(int limit)
    : limit(limit), nr_handlers(0), extra_handlers(nullptr)
{
  if (limit > kInlineLimit) {
    extra_handlers = (PieceRoutine **)Alloc(util::Align(
        sizeof(PieceRoutine *) * (limit - kInlineLimit), CACHE_LINE_SIZE));
  }
}

void *BasePieceCollection::operator new(std::size_t size)
{
  return BasePieceCollection::Alloc(size);
}

void *BasePieceCollection::Alloc(size_t size)
{
  return util::Impl<PromiseAllocationService>().Alloc(size);
}

void BasePieceCollection::AssignSchedulingKey(uint64_t key)
{
  for (int i = 0; i < nr_handlers; i++) {
    auto *child = routine(i);
    if (child->sched_key == 0)
      child->sched_key = key;
    if (child->next)
      child->next->AssignSchedulingKey(key);
  }
}

void BasePieceCollection::AssignAffinity(uint64_t aff)
{
  for (int i = 0; i < nr_handlers; i++) {
    auto *child = routine(i);
    if (child->affinity == std::numeric_limits<uint64_t>::max())
      child->affinity = aff;
    if (child->next)
      child->next->AssignAffinity(aff);
  }
}

void BasePieceCollection::Add(PieceRoutine *child)
{
  abort_if(nr_handlers >= limit,
           "nr_handlers {} exceeding limits!", nr_handlers);
  routine(nr_handlers++) = child;
}

void BasePieceCollection::Complete()
{
  auto &transport = util::Impl<PromiseRoutineTransportService>();
  for (size_t i = 0; i < nr_handlers; i++) {
    auto r = routine(i);
    transport.TransportPromiseRoutine(r);
  }
}

void BasePieceCollection::ExecutionRoutine::AddToReadyQueue(go::Scheduler::Queue *q, bool next_ready)
{
  auto p = q;
  while (p->next != q) {
    if (!((Routine *) p->next)->is_urgent())
      break;
    p = p->next;
  }
  Add(p);
}

void BasePieceCollection::ExecutionRoutine::Run()
{
  if (CoroSched::g_use_coro_sched) {
    // hijacking ExecutionRoutine to use coroutine scheduler
    coro_sched->StartCoroExec();
    return;
  }

  auto &svc = util::Impl<PromiseRoutineDispatchService>();
  auto &transport = util::Impl<PromiseRoutineTransportService>();

  int core_id = scheduler()->thread_pool_id() - 1;
  trace(TRACE_EXEC_ROUTINE "new ExecutionRoutine up and running on {}", core_id);

  PieceRoutine *next_r;
  bool give_up = false;  /*!< should I exit and let another ExecutionRoutine continue? */
  // BasePromise::ExecutionRoutine *next_state = nullptr;
  go::Scheduler *sched = scheduler();

  auto should_pop = PromiseRoutineDispatchService::GenericDispatchPeekListener(
      [&next_r, &give_up, sched]
      (PieceRoutine *r, BasePieceCollection::ExecutionRoutine *state) -> bool {
        // If there's another ExecutionRoutine to wake up, then wake up that one and kill myself
        if (state != nullptr) {
          if (state->is_detached()) {
            trace(TRACE_EXEC_ROUTINE "Wakeup Coroutine {}", (void *) state);
            state->Init();
            sched->WakeUp(state);
          } else {
            trace(TRACE_EXEC_ROUTINE "Found a sleeping Coroutine, but it's already awaken.");
          }
          give_up = true;
          return false;
        }
        // if the PieceRoutine was not run before (and preempted)
        give_up = false;
        next_r = r;
        // next_state = state;
        return true;
      });


  unsigned long cnt = 0x01F;

  do {
    while (svc.Peek(core_id, should_pop)) {
      // Periodic flush
      cnt++;
      if ((cnt & 0x01F) == 0) {
        transport.PeriodicIO(core_id);
      }

      auto rt = next_r;
      if (rt->sched_key != 0)
        debug(TRACE_EXEC_ROUTINE "Run {} sid {}", (void *) rt, rt->sched_key);

      rt->callback(rt);
      svc.Complete(core_id);
    }
  } while (!give_up && svc.IsReady(core_id) && transport.PeriodicIO(core_id));

  trace(TRACE_EXEC_ROUTINE "Coroutine Exit on core {} give up {}", core_id, give_up);
}

bool BasePieceCollection::ExecutionRoutine::Preempt(uint64_t sid, uint64_t ver)
{
//  return false; // FIXME: Disable preemption from now
 
  auto &svc = util::Impl<PromiseRoutineDispatchService>();
  int core_id = scheduler()->thread_pool_id() - 1;
  bool spawn = true;

  //logger->info("preempt core: {} sid: {} ver: {}",core_id,sid,ver);

  // checks if there's something else to switch to (other waiting routines or new pieces to run)
  // if so, preempt
  if (svc.Preempt(core_id, this, sid, ver)) {
 sleep:

    // Shujian: why does it always spawn a new ExecutionRoutine first?
    // what if there's some other waiting routine to run? shouldn't we run that first?
    if (spawn) {
      sched->WakeUp(new ExecutionRoutine());
    }
    trace(TRACE_EXEC_ROUTINE "Sleep. Spawning a new coroutine = {}.", spawn);
    sched->RunNext(go::Scheduler::SleepState);

    spawn = true;
    auto should_pop = PromiseRoutineDispatchService::GenericDispatchPeekListener(
        [this, &spawn]
        (PieceRoutine *, BasePieceCollection::ExecutionRoutine *state) -> bool {
          if (state == this)
            // if I myself should run, then return from Preempt call
            return true;
          if (state != nullptr) {
            // if there's a ExecutionRoutine to wake up, don't spawn and just run that
            trace(TRACE_EXEC_ROUTINE "Unfinished encoutered, no spawn.");
            if (state->is_detached()) {
              state->Init();
              sched->WakeUp(state);
            }
            spawn = false;
          } else {
            // when there's a new piece routine to run, spawn
            trace(TRACE_EXEC_ROUTINE "Spawning because I saw a new piece");
          }
          return false;
        });

    trace(TRACE_EXEC_ROUTINE "Just got up!");

    // when there's anything other than a waiting ExecutionRoutine to run, spawn a new ExecutionRoutine
    if (!svc.Peek(core_id, should_pop))
      goto sleep;

    set_busy_poll(false);
    return true;
  }

  // There's nothing to switch to, continue spinning
  return false;
}

void BasePieceCollection::QueueRoutine(PieceRoutine **routines, size_t nr_routines, int core_id)
{
  util::Impl<PromiseRoutineDispatchService>().Add(core_id, routines, nr_routines);
}

void BasePieceCollection::FlushScheduler()
{
  auto &svc = util::Impl<PromiseRoutineDispatchService>();
  for (int i = 0; i < g_nr_threads; i++) {
    if (!svc.IsRunning(i)) {
      go::GetSchedulerFromPool(i + 1)->WakeUp(new ExecutionRoutine());
    }
  }
}

}