// Copyright 2019 The Fuchsia Authors. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.

#include <stdio.h>
#include <zircon/syscalls.h>
#include <zircon/types.h>

#include <io-scheduler/io-scheduler.h>
#include <io-scheduler/worker.h>

namespace ioscheduler {

zx_status_t Worker::Create(Scheduler* sched, uint32_t id, fbl::unique_ptr<Worker>* out) {
  fbl::AllocChecker ac;
  fbl::unique_ptr<Worker> worker(new (&ac) Worker(sched, id));
  if (!ac.check()) {
    fprintf(stderr, "Failed to allocate worker.\n");
    return ZX_ERR_NO_MEMORY;
  }
  if (thrd_create_with_name(&worker->thread_, worker->ThreadEntry, worker.get(), "io-worker") !=
      thrd_success) {
    fprintf(stderr, "Failed to create worker thread.\n");
    return ZX_ERR_NO_MEMORY;
  }
  worker->thread_started_ = true;
  *out = std::move(worker);
  return ZX_OK;
}

Worker::Worker(Scheduler* sched, uint32_t id) : sched_(sched), id_(id) {}

Worker::~Worker() {
  if (thread_started_) {
    thrd_join(thread_, nullptr);
  }
}

int Worker::ThreadEntry(void* arg) {
  Worker* w = static_cast<Worker*>(arg);
  w->WorkerLoop();
  return 0;
}

void Worker::WorkerLoop() {
  const size_t max_ops = 10;
  SchedulerClient* client = sched_->client();
  zx_status_t status;
  while (!cancelled_) {
    // Fetch ops from the client.

    size_t acquire_count = 0;
    StreamOp* op_list[max_ops];
    status = client->Acquire(op_list, max_ops, &acquire_count, true);
    if (status == ZX_ERR_CANCELED) {
      // Cancel received, no more ops to read. Drain the streams and exit.
      break;
    }
    if (status != ZX_OK) {
      fprintf(stderr, "Unexpected return status from Acquire() %d\n", status);
      client->Fatal();
      break;
    }

    // Containerize all ops for safety.
    UniqueOp uop_list[max_ops];
    for (size_t i = 0; i < acquire_count; i++) {
      uop_list[i].set(op_list[i]);
    }

    // Enqueue ops in the scheduler's priority queue.

    size_t num_error = 0;
    sched_->Enqueue(uop_list, acquire_count, uop_list, &num_error);
    // Any ops remaining in the list have encountered an error and should be released.
    for (size_t i = 0; i < num_error; i++) {
      client->Release(uop_list[i].release());
    }

    // Drain the priority queue.

    for (;;) {
      // Fetch an op.

      UniqueOp op;
      status = sched_->Dequeue(&op, false);
      if (status == ZX_ERR_SHOULD_WAIT) {
        // No more ops.
        break;
      } else if (status == ZX_ERR_CANCELED) {
        // Shutdown initiated.
        cancelled_ = true;
        break;
      } else if (status != ZX_OK) {
        fprintf(stderr, "Dequeue() failed %d\n", status);
        return;
      }

      // Execute it.

      status = client->Issue(op.get());

      Stream* stream = op->stream();
      ZX_DEBUG_ASSERT(stream != nullptr);

      if (status == ZX_OK) {
        // Op completed successfully or encountered a synchronous error.
        stream->ReleaseOp(std::move(op), client);
      } else if (status == ZX_ERR_ASYNC) {
        // Op queued for async completion. Released when completed.
        // Op is retained in stream.
        op.release();
      } else {
        fprintf(stderr, "Unexpected return status from Issue() %d\n", status);
        // Mark op as failed.
        op->set_result(ZX_ERR_IO);
        stream->ReleaseOp(std::move(op), client);
      }
    }
  }
}

}  // namespace ioscheduler