/*
 * task_service.h
 *
 *  Copyright (c) 2021 Rockchip Electronics Co., Ltd.
 *
 * Licensed under the Apache License, Version 2.0 (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.apache.org/licenses/LICENSE-2.0
 *
 * 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.
 *
 * Author: Cody Xie <cody.xie@rock-chips.com>
 */
#ifndef XCORE_TASK_SERVICE_H
#define XCORE_TASK_SERVICE_H

#include <algorithm>
#include <atomic>
#include <cassert>
#include <chrono>
#include <condition_variable>
#include <cstdint>
#include <deque>
#include <functional>
#include <iostream>
#include <mutex>
#include <thread>
#include <utility>

#include "task_traits.h"

#define DEBUG 1
#ifndef TASK_LOG
#define TASK_LOG(format, ...) \
    printf("DEBUG: " #format "\n", ##__VA_ARGS__)
#endif

namespace XCam {

// SERVICE CONCEPTS:
// * A aynchronous threaded service to execute specific tasks that deal with
//   INPUTS and OUTPUTS parameters
// * Service will calculate the MAX_PROCESS_TIME of the task consumed
// * Support run ONCE/SYNC/ASYNC on a task
//   * ASYNC: Run task in a threaded loop
//   * SYNC: Call ENQUEUE/DEQUEUE synchronous by user
//   * ONCE: RunOnce for one param input
// * Support ENQUEUE/DEQUEUE methods
//   * ENQUEUE will signal loop immediately to run task
//   * DEQUEUE will wait until results not empty and the MAY_BLOCK is true
//   * DEQUEUE may return empty param if no result and the MAY_BLOCK is false
// * Support buffer overrun/underrun behaviors on
//   * BUSY (fifo full)
//   * IDLE (fifo empty for longer than 2 period)
//   * LATE (execution time longer than 1 period)
//   * ERROR (process error)
// * Support config and config change event ???
//   to negotiate with other service ???
//
// * Users can ENQUEUE new params, DEQUEUE PROCESSED/ERROR/SKIPPED params
// * Users can trigger START, STOP the service
// * Users can FLUSH the proccessing queue, mark param to SKIPPED
// * Users can specific the MAXIMUM PROCESS TIME
//
// * Parameters are UNIQUE, FIFO even on ERROR/LATE happened
// * Parameters have a BUFFER POOL, with MAX_PARAM_COUNT
// * Parameters can be ALLOCATED inside or outside of Service
// * Last result will be HOLD A COPY if current task failed
//
// * Task is a FUNCTOR object
// * Task does NOT OWN the param
// * Task may HOLD several input buffers
// * Task may return process result
//   * SUCCESS - returned, successful processed
//   * SKIPPED - returned, but this param skipped
//   * FAILED - returned, but failed in processing
//   * AGAIN - required more inputs
// * Task inputs and outputs may have different id
// * Task outputs may have inputs param
//

template <typename T, class Container = std::deque<ServiceParam<T>>>
class TaskService {
 public:
    typedef TaskService<T> value_type;
    typedef ServiceTask<T> task_type;
    typedef ServiceParam<T> param_type;

    TaskService()                      = delete;
    TaskService(const TaskService<T>&) = delete;
    const TaskService<T>& operator=(const TaskService<T>&) = delete;

    explicit TaskService(std::unique_ptr<ServiceTask<T>> task,
                         const bool may_block_               = default_may_block,
                         const uint8_t max_param_count       = default_max_param_count,
                         const TaskDuration max_process_time = default_process_time)
        : task_(std::move(task)),
          may_block_(default_may_block),
          max_param_count_(max_param_count),
          max_process_time_(max_process_time),
          started_(false) {
        assert(task_.get() != nullptr);
        allocServiceParam();
    }

    virtual ~TaskService() {
        stop();
        clear();
    }

    void loop() {
        while (started_) {
            std::unique_lock<std::mutex> in_lock(in_mutex_);
            auto wait_ret = in_cond_.wait_for(in_lock, max_process_time_, [this]() {
                return (!started_ || (!in_params_.empty() &&
                                      in_params_.front().state == ParamState::kReadyForProcess));
            });
            if (wait_ret == false) {
                // IDLE
                continue;
            }
            if (!started_) break;

            auto param = std::move(in_params_.front());
            in_params_.pop_front();
            in_lock.unlock();
            param.state = ParamState::kProcessing;
            const auto start = std::chrono::steady_clock::now();

            auto ret = (*task_)(param);
            if (ret == TaskResult::kSuccess) {
                param.state = ParamState::kProcessedSuccess;
            } else {
                param.state = ParamState::kProcessedError;
                TASK_LOG("task processs failed");
#ifdef USE_TASK_CALLBACK
                callbacks_.onError();
#endif
            }

            {
                std::lock_guard<std::mutex> out_lock(out_mutex_);
                // TODO(Cody): if we have alllocated params in front,
                // this will block the result until these params are
                // dequeued. if we use push_front, we break the sequence
                // of params
                out_params_.push_back(std::move(param));
                out_cond_.notify_one();
            }

            const auto end = std::chrono::steady_clock::now();

            const std::chrono::duration<double, std::milli> elapsed = end - start;
            if (elapsed >= max_process_time_) {
                // TODO(Cody): match the frame id ?
                TASK_LOG("params processs elapsed %lf exceeds %lf", elapsed.count(),
                     max_process_time_.count());
#if USE_TASK_CALLBACK
                callbacks_.onLate();
#endif
            }
        }
    }

    void start() {
        if (started_) {
            return;
        }
        started_  = true;
        executor_ = std::thread([this] { loop(); });
        // executor_.detach();
#if USE_TASK_CALLBACK
        callbacks_.onStarted();
#endif
    }

    void stop() {
        if (started_) {
            started_ = false;
            executor_.join();
        }
#if USE_TASK_CALLBACK
        callbacks_.onStoped();
#endif
    }

    void allocServiceParam() noexcept {
        std::lock_guard<std::mutex> lock(out_mutex_);
        while (out_params_.size() < max_param_count_) {
            // TODO(Cody): Must add try block if -fexceptions enabled
            // Ensure new object does not throw any exceptions
            ServiceParam<T> param;
            param.state = ParamState::kAllocated;
            param.unique_id = -1;
            param.payload = std::make_shared<T>();
            out_params_.push_back(std::move(param));
            out_cond_.notify_one();
        }
    }

    void enqueue(ServiceParam<T>& input) {
        std::lock_guard<std::mutex> lock(in_mutex_);
        // TODO(Cody): deal with busy case
#if 0
        if (in_params_.size() == max_param_count_) {
            TASK_LOG("input params exceeds max count!");
#if USE_TASK_CALLBACK
            callbacks_.onFull(input);
#endif
            return;
        }
#endif
        input.state = ParamState::kReadyForProcess;
        in_params_.push_back(input);
        in_cond_.notify_one();
    }

    void enqueue(ServiceParam<T>&& input) {
        std::lock_guard<std::mutex> lock(in_mutex_);
        // TODO(Cody): deal with busy case
#if 0
        if (in_params_.size() == max_param_count_) {
            TASK_LOG("input params exceeds max count!");
#if USE_TASK_CALLBACK
            callbacks_.onFull(input);
#endif
            return;
        }
#endif
        input.state = ParamState::kReadyForProcess;
        in_params_.push_back(std::move(input));
        in_cond_.notify_one();
    }

    ServiceParam<T> dequeue() {
        std::unique_lock<std::mutex> lock(out_mutex_);
        out_cond_.wait(lock, [this]() {
            if ((!out_params_.empty() &&
                 (out_params_.front().state == ParamState::kAllocated ||
                  out_params_.front().state == ParamState::kProcessedSuccess ||
                  out_params_.front().state == ParamState::kProcessedError))) {
                return true;
            } else if (!may_block_) {
                ServiceParam<T> empty = {
                    .state = ParamState::kNull,
                    .unique_id = -1,
                    .payload = std::shared_ptr<T>(nullptr),
                };
                out_params_.push_front(std::move(empty));
                return true;
            } else {
                return false;
            }
        });
        auto p = std::move(out_params_.front());
        out_params_.pop_front();
        return p;
    }

    void flush() {
        std::lock(in_mutex_, out_mutex_);
        std::lock_guard<std::mutex> in_lock(in_mutex_, std::adopt_lock);
        std::lock_guard<std::mutex> out_lock(in_mutex_, std::adopt_lock);
        if (in_params_.empty()) {
            return;
        }
        auto paramIt =
            std::remove_if(in_params_.begin(), in_params_.end(), [](const ServiceParam<T>& p) {
                return (p.state != ParamState::kReadyForProcess) &&
                       (p.state != ParamState::kAllocated);
            });
        for (auto param = paramIt; param != in_params_.end(); param++) {
            param.state = ParamState::kAllocated;
            param.unique_id = -1;
            out_params_.push_back(param);
            out_cond_.notify_one();
        }
    }

    void clear() {
        std::lock(in_mutex_, out_mutex_);
        std::lock_guard<std::mutex> in_lock(in_mutex_, std::adopt_lock);
        std::lock_guard<std::mutex> out_lock(in_mutex_, std::adopt_lock);
        in_params_.clear();
        out_params_.clear();
    }

    const uint8_t getMaxParamCount() { return max_param_count_; }
    void setMaxParamCount(const uint8_t count) { max_param_count_ = count; }

    const TaskDuration getMaxProceedTime() { return max_process_time_; }
    void setMaxProceedTime(const TaskDuration duration) { max_process_time_ = duration; }
    void setMaxProceedTimeByFps(const int fps) {
        max_process_time_ = TaskDuration(std::chrono::milliseconds(1000 / fps));
    }

    // TODO(Cody): add support link to other services
#if 0
    template <typename O>
    friend void link(TaskService<O>* other);
    template <typename O>
    friend void unlink(TaskService<O>* other);
#endif

 private:
    uint8_t max_param_count_;
    TaskDuration max_process_time_;
    bool may_block_;
    bool started_;

    std::mutex in_mutex_;
    std::condition_variable in_cond_;
    std::mutex out_mutex_;
    std::condition_variable out_cond_;
    std::unique_ptr<ServiceTask<T>> task_;
    std::thread executor_;

    Container in_params_;
    Container out_params_;

#if USE_TASK_CALLBACK
    TaskCallbacks callbacks_;
#endif
};

};  // namespace XCam

#endif  // ALGOS_AEIS_TASK_SERVICE_H