/*
 * Licensed to the OpenAirInterface (OAI) Software Alliance under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The OpenAirInterface Software Alliance licenses this file to You under
 * the OAI Public License, Version 1.1  (the "License"); you may not use this
 * file except in compliance with the License. You may obtain a copy of the
 * License at
 *
 *      http://www.openairinterface.org/?page_id=698
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *-------------------------------------------------------------------------------
 * For more information about the OpenAirInterface (OAI) Software Alliance:
 *      contact@openairinterface.org
 */
/*! \file itti.cpp
 \brief
 \author  Lionel GAUTHIER
 \date 2018
 \email: lionel.gauthier@eurecom.fr
 */
#include "itti.hpp"

#include <signal.h>
#include <sys/epoll.h>
#include <sys/eventfd.h>
#include <time.h>

#include <algorithm>
#include <csignal>

#include "common_defs.h"
#include "logger.hpp"

extern itti_mw* itti_inst;

static itti_timer null_timer(
    ITTI_INVALID_TIMER_ID, TASK_NONE, 0xFFFFFFFF, 0xFFFFFFFF, 0, 0);

//------------------------------------------------------------------------------
void itti_mw::timer_manager_task(
    const util::thread_sched_params& sched_params) {
  Logger::itti().info("Starting timer_manager_task");
  sched_params.apply(TASK_ITTI_TIMER, Logger::itti());
  while (true) {
    if (itti_inst->terminate) return;
    {
      std::unique_lock<std::mutex> lx(itti_inst->m_timers);
      while (itti_inst->timers.empty()) {
        itti_inst->c_timers.wait(lx);
      }
      std::set<itti_timer>::iterator it = itti_inst->timers.begin();
      itti_inst->current_timer          = std::ref(*it);
      itti_inst->timers.erase(it);
      lx.unlock();

      // check time-out
      if (itti_inst->current_timer.time_out >
          std::chrono::system_clock::now()) {
        std::unique_lock<std::mutex> lto(itti_inst->m_timeout);
        auto diff = itti_inst->current_timer.time_out -
                    std::chrono::system_clock::now();
        auto rc = itti_inst->c_timeout.wait_for(lto, diff);
        lto.unlock();
        if (std::cv_status::timeout == rc) {
          // signal time-out
          itti_msg_timeout mto(
              TASK_ITTI_TIMER, itti_inst->current_timer.task_id,
              itti_inst->current_timer.id, itti_inst->current_timer.arg1_user,
              itti_inst->current_timer.arg2_user);
          std::shared_ptr<itti_msg_timeout> msgsh =
              std::make_shared<itti_msg_timeout>(mto);
          int ret = itti_inst->send_msg(msgsh);
        } else {
          // other timer required ?
          itti_inst->m_timers.lock();
          if (itti_inst->current_timer.id != ITTI_INVALID_TIMER_ID) {
            itti_inst->timers.insert(itti_inst->current_timer);
          }
          itti_inst->current_timer = null_timer;
          itti_inst->m_timers.unlock();
          // cout << "cv.wait released by other triggered timer end" << endl;
        }
      } else {
        // signal time-out
        itti_msg_timeout mto(
            TASK_ITTI_TIMER, itti_inst->current_timer.task_id,
            itti_inst->current_timer.id, itti_inst->current_timer.arg1_user,
            itti_inst->current_timer.arg2_user);
        std::shared_ptr<itti_msg_timeout> msgsh =
            std::make_shared<itti_msg_timeout>(mto);
        itti_inst->send_msg(msgsh);
      }
    }
  }
}
//------------------------------------------------------------------------------
itti_mw::itti_mw()
    : timer_id(0),
      msg_number(0),
      created_tasks(0),
      ready_tasks(0),
      m_timers(),
      current_timer(null_timer),
      m_timeout(),
      m_timer_id(),
      terminate(false) {
  std::fill(itti_task_ctxts, itti_task_ctxts + TASK_MAX, nullptr);
}

//------------------------------------------------------------------------------
itti_mw::~itti_mw() {
  std::cout << "~itti()" << std::endl;
  timer_thread.detach();
  // wake up thread timer if necessary
  std::unique_lock<std::mutex> l2(m_timeout);
  c_timeout.notify_one();

  for (int t = TASK_FIRST; t < TASK_MAX; t++) {
    if (itti_task_ctxts[t]) {
      delete itti_task_ctxts[t];
    }
  }
  std::cout << "~itti() Done!" << std::endl;
}

//------------------------------------------------------------------------------
void itti_mw::start(const util::thread_sched_params& sched_params) {
  Logger::itti().startup("Starting...");
  timer_thread = std::thread(timer_manager_task, sched_params);
  Logger::itti().startup("Started");
}
//------------------------------------------------------------------------------
timer_id_t itti_mw::increment_timer_id() {
  return ++timer_id;
}

//------------------------------------------------------------------------------
unsigned int itti_mw::increment_message_number() {
  return __sync_fetch_and_add(&msg_number, 1);
}

//------------------------------------------------------------------------------
int itti_mw::create_task(
    const task_id_t task_id, void (*start_routine)(void*), void* args_p) {
  if (nullptr == start_routine) {
    Logger::itti().error("Null start routine for task %d", task_id);
    return RETURNerror;
  }
  if ((TASK_FIRST <= task_id) && (TASK_MAX > task_id)) {
    if (itti_task_ctxts[task_id] == nullptr) {
      itti_task_ctxts[task_id] = new itti_task_ctxt(task_id);
      {
        std::unique_lock<std::mutex> lk(itti_task_ctxts[task_id]->m_state);
        if (itti_task_ctxts[task_id]->task_state == TASK_STATE_NOT_CONFIGURED) {
          itti_task_ctxts[task_id]->task_state = TASK_STATE_STARTING;
        }
        created_tasks++;
        lk.unlock();
      }
      itti_task_ctxts[task_id]->thread = std::thread(start_routine, args_p);
      while ((itti_task_ctxts[task_id]->task_state != TASK_STATE_READY) &&
             (itti_task_ctxts[task_id]->task_state != TASK_STATE_ENDED))
        usleep(1000);
      return 0;
    } else {
      Logger::itti().error("Try to start an already started task %d", task_id);
    }
  } else {
    Logger::itti().error("Bad task id %d", task_id);
  }
  return RETURNerror;
}

//------------------------------------------------------------------------------
int itti_mw::notify_task_ready(const task_id_t task_id) {
  if (itti_task_ctxts[task_id]) {
    itti_task_ctxts[task_id]->m_state.lock();
    if (itti_task_ctxts[task_id]->task_state == TASK_STATE_STARTING) {
      itti_task_ctxts[task_id]->task_state = TASK_STATE_READY;
      itti_task_ctxts[task_id]->m_state.unlock();
      return RETURNok;
    }
    itti_task_ctxts[task_id]->m_state.unlock();
    Logger::itti().error(
        "Notify task ready, bad state %d",
        itti_task_ctxts[task_id]->task_state);
  } else {
    Logger::itti().error("Notify task ready, task not starting %d", task_id);
  }
  return RETURNerror;
}

//------------------------------------------------------------------------------
int itti_mw::send_msg(std::shared_ptr<itti_msg> message) {
  if ((TASK_FIRST <= message->destination) &&
      (TASK_MAX > message->destination)) {
    if (itti_task_ctxts[message->destination]) {
      if (itti_task_ctxts[message->destination]->task_state ==
          TASK_STATE_READY) {
        std::unique_lock<std::mutex> l(
            itti_task_ctxts[message->destination]->m_queue);
        // res =
        itti_task_ctxts[message->destination]->msg_queue.push(message);
        itti_task_ctxts[message->destination]->c_queue.notify_one();
        return RETURNok;
      } else if (
          itti_task_ctxts[message->destination]->task_state ==
          TASK_STATE_ENDED) {
        Logger::itti().warn(
            "Unicast message number %lu can not be sent from %d to %d, ended "
            "destination task!",
            message->msg_num, message->origin, message->destination);
        return RETURNerror;
      }
    } else {
      Logger::itti().warn(
          "Unicast message number %lu can not be sent from %d to %d, not "
          "started destination task!",
          message->msg_num, message->origin, message->destination);
    }
  } else if (message->destination == TASK_ALL) {
    return send_broadcast_msg(message);
  }
  return RETURNerror;
}

//------------------------------------------------------------------------------
int itti_mw::send_broadcast_msg(std::shared_ptr<itti_msg> message) {
  if (TASK_ALL == message->destination) {
    for (int t = TASK_FIRST; t < TASK_MAX; t++) {
      if (itti_task_ctxts[t]) {
        if (itti_task_ctxts[t]->task_state == TASK_STATE_READY) {
          std::unique_lock<std::mutex> l(itti_task_ctxts[t]->m_queue);
          itti_task_ctxts[t]->msg_queue.push(message);
          itti_task_ctxts[t]->c_queue.notify_one();
        } else if (itti_task_ctxts[t]->task_state == TASK_STATE_ENDED) {
          Logger::itti().warn(
              "Broadcast message number %lu can not be sent from %d to %d, "
              "ended destination task!",
              message->msg_num, message->origin, t);
        } else {
          Logger::itti().warn(
              "Broadcast message number %lu can not be sent from %d to %d, "
              "unknown state %d !",
              message->msg_num, message->origin, t,
              itti_task_ctxts[t]->task_state);
        }
      }
    }
    return RETURNok;
  }
  return RETURNerror;
}

//------------------------------------------------------------------------------
int itti_mw::send_terminate_msg(task_id_t src_task_id) {
  itti_msg_terminate msg(src_task_id, TASK_ALL);
  terminate = true;
  std::shared_ptr<itti_msg_terminate> smsg =
      std::make_shared<itti_msg_terminate>(msg);
  int ret = itti_inst->send_broadcast_msg(smsg);

  return ret;
}
//------------------------------------------------------------------------------
std::shared_ptr<itti_msg> itti_mw::receive_msg(task_id_t task_id) {
  if ((TASK_FIRST <= task_id) && (TASK_MAX > task_id)) {
    if (itti_task_ctxts[task_id]) {
      std::unique_lock<std::mutex> lk(itti_task_ctxts[task_id]->m_queue);
      while (itti_task_ctxts[task_id]->msg_queue.empty()) {
        itti_task_ctxts[task_id]->c_queue.wait(lk);
      }
      std::shared_ptr<itti_msg> msg =
          itti_task_ctxts[task_id]->msg_queue.front();
      itti_task_ctxts[task_id]->msg_queue.pop();
      return msg;
    }
  }
  Logger::itti().warn("received message failed, bad task id");
  return nullptr;
}

//------------------------------------------------------------------------------
std::shared_ptr<itti_msg> itti_mw::poll_msg(task_id_t task_id) {
  if ((TASK_FIRST <= task_id) && (TASK_MAX > task_id)) {
    if (itti_task_ctxts[task_id]) {
      std::lock_guard<std::mutex> lk(itti_task_ctxts[task_id]->m_queue);
      if (!itti_task_ctxts[task_id]->msg_queue.empty()) {
        std::shared_ptr<itti_msg> msg =
            itti_task_ctxts[task_id]->msg_queue.front();
        itti_task_ctxts[task_id]->msg_queue.pop();
        return msg;
      }
    }
  }
  return nullptr;
}

//------------------------------------------------------------------------------
void itti_mw::wait_tasks_end(void) {
  Logger::itti().info("Waiting ITTI tasks closed");
  for (int task_idx = TASK_FIRST; task_idx < TASK_MAX; task_idx++) {
    if (itti_task_ctxts[task_idx]) {
      if (itti_task_ctxts[task_idx]->thread.joinable()) {
        itti_task_ctxts[task_idx]->thread.join();
        itti_task_ctxts[task_idx]->m_state.lock();
        itti_task_ctxts[task_idx]->task_state = TASK_STATE_ENDED;
        itti_task_ctxts[task_idx]->m_state.unlock();
      }
    }
  }
  Logger::itti().info("All ITTI tasks closed");
}

//------------------------------------------------------------------------------
timer_id_t itti_mw::timer_setup(
    uint32_t interval_sec, uint32_t interval_us, task_id_t task_id,
    uint64_t arg1_user, uint64_t arg2_user) {
  // Not sending to task timer
  if ((TASK_FIRST < task_id) && (TASK_MAX > task_id)) {
    itti_timer t(
        increment_timer_id(), task_id, interval_sec, interval_us, arg1_user,
        arg2_user);
    timer_id_t id = t.id;
    std::unique_lock<std::mutex> l(m_timers);
    timers.insert(t);
    c_timers.notify_one();

    // wake up thread timer if necessary
    std::unique_lock<std::mutex> l2(m_timeout);
    c_timeout.notify_one();
    return id;
  }
  return ITTI_INVALID_TIMER_ID;
}

//------------------------------------------------------------------------------
int itti_mw::timer_remove(timer_id_t timer_id) {
  std::lock_guard<std::mutex> lk(m_timers);
  if (current_timer.id == timer_id) {
    // cout << "timer_remove() current_timer.id == timer_id" << endl;
    current_timer = null_timer;
    c_timers.notify_one();
    // wake up thread timer if necessary
    std::unique_lock<std::mutex> l2(m_timeout);
    c_timeout.notify_one();
    return RETURNok;
  } else {
    std::set<itti_timer>::iterator it = itti_inst->timers.begin();
    while (it != itti_inst->timers.end()) {
      if (it->id == timer_id) {
        itti_inst->timers.erase(it);
        return RETURNok;
      }
      ++it;
    }
  }
  Logger::itti().trace("Removing timer 0x%lx: Not found", timer_id);
  return RETURNerror;
}
