#include "io_coro.h"
#include "../server/Global.h"
#include "../server/Mode_Msg_Buffer.h"
#include "Psc_Cpp_Core/Statistics/Frequency_Limit.h"
#include "Local_Server/Data_Feed/Data_Feed.h"
#include "Local_Server/Data_Source/Data_Source.h"
#include <asio/detached.hpp>
#include <asio/post.hpp>
#include <asio/steady_timer.hpp>
#include <asio/this_coro.hpp>
#include <asio/use_awaitable.hpp>
#include <chrono>
#include <exception>
#include <format>
#include <sstream>
#include <thread>
#include <utility>
static std::string thread_id_str() {
    std::ostringstream oss;
    oss << std::this_thread::get_id();
    return oss.str();
}
std::vector<std::shared_ptr<With_Loop_Coro>> get_all() {
    auto g = Global::instance();
    std::vector<std::shared_ptr<With_Loop_Coro>> ret;
    auto sources = g->mode_acs.data_source_config.map.list();
    for (auto& source : sources) {
        ret.emplace_back(source);
    }
    auto feeds = g->mode_acs.data_feed_config.map.list();
    for (auto& feed : feeds) {
        ret.emplace_back(feed);
    }
    return ret;
}
bool data_feed_debug = false;
bool data_source_debug = false;
// 不开启wait 性能巨差 数据积压
void Coro::start() {
    std::cout << "1================ start_io_coro" << std::endl;
    running.store(true, std::memory_order_release);
    io.restart();
    io_work = std::make_unique<asio::executor_work_guard<asio::io_context::executor_type>>(io.get_executor());
    io_thread = std::thread([this] {
        try {
            std::cout << std::format("协程io:{} 协程启动！\n", thread_id_str());
            io.run();
            std::cout << std::format("协程io:{} 销毁！\n", thread_id_str());
        }
        catch (const std::exception& e) {
            std::cerr << std::format("协程io异常: {}\n", e.what());
            std::terminate();
        }
        catch (...) {
            std::cerr << "协程io未知异常\n";
            std::terminate();
        }
    });
    auto n = std::max<unsigned int>(1, std::thread::hardware_concurrency());
    auto other_need = std::max<unsigned int>(2, n / 2);
    if (n > other_need) {
        n -= other_need;
    }
    else {
        n = n / 2;
    }
    n = std::max<unsigned int>(1, n);
    process_data = std::make_unique<asio::thread_pool>(n);
    std::cout << std::format("协程已经启动 process_data concurrency: 在n个线程上{}\n", n);
    auto promise = std::make_shared<std::promise<void>>();
    loop_task = std::make_unique<std::future<void>>(promise->get_future());
    asio::co_spawn(io, coro_thread(), [promise](std::exception_ptr ep) {
        std::cout << "发生了异常" << std::endl;
        if (ep) {
            promise->set_exception(ep);
            std::rethrow_exception(ep);
        }
        promise->set_value();
    });
    std::cout << std::flush;
}
asio::awaitable<void> Coro::sleep_for(std::chrono::milliseconds ms) {
    auto executor = co_await asio::this_coro::executor;
    asio::steady_timer timer(executor, ms);
    co_await timer.async_wait(asio::use_awaitable);
    co_return;
}
void Coro::stop() {
    std::cout << "stop_io_coro begin" << std::endl;
    auto list = get_all();
    for (auto& li : list) {
        if (li->running()) {
            std::cout << std::format("请求停止 {}:{}   当前状态为{}\n", li->type, li->key, Psc::to_string(li->state.load()));
            li->async_stop();
        }
    }
    // 同步等待
    for (auto& li : list) {
        li->sync_wait();
    }
    // 停止状态更新状态机
    running = false;
    asio::post(io, [] {});
    if (loop_task) {
        std::cout << "等待 coro_thread 退出" << std::endl;
        loop_task->get();
        loop_task.reset();
        std::cout << "coro_thread 已退出" << std::endl;
    }
    // 清理资源
    for (auto& li : list) {
        li->loop_task.reset();
    }
    if (process_data) {
        process_data->join();
        process_data.reset();
    }
    io_work.reset();
    io.stop();
    if (io_thread.joinable()) {
        io_thread.join();
    }
    std::cout << "stop_io_coro end" << std::endl;
}
asio::awaitable<void> Coro::coro_thread() {
    while (running.load(std::memory_order_acquire)) {
        auto list = get_all();
        for (auto& li : list) {
            co_await li->tick(); //状态机驱动函数
        }
        co_await sleep_for(std::chrono::milliseconds(10));
    }
    co_return;
}
asio::awaitable<void> Data_Feed::loop_coro() {
    auto feed = this;
    auto co = Coro::instance();
    auto& mode_acs = Global::instance()->mode_acs;
    while (feed->running()) {
        if (data_feed_debug) {
            std::cout << std::format("{} feed loop begin {}\n", key, thread_id_str());
        }
        BIN_Msg_Buffer::Batch messages;
        {
            if (!feed->running() || !feed->registered()) {
                break;
            }
            co_await feed->handle_in_loop_coro();
            auto s_num = feed->msg_buffer.mode_s_msg_num.load();
            auto other_num = feed->msg_buffer.mode_other_msg_num.load();
            messages = feed->msg_buffer.take_all();
            const auto [num, monitor_msg_live] = mode_acs.settings.read([](const auto& value) {
                return std::pair{value.report_data_feed_msg_mum, value.monitor_msg_live};
            });
            if (monitor_msg_live && num != 0 && messages.size() > num && too_many_msg_limit.test()) {
                std::ostringstream oss;
                oss << "feed_key:" << feed->key << "   ";
                oss << "recv_s:" << s_num << "   ";
                oss << "recv_other:" << other_num << "   ";
                std::cout << oss.str() << std::endl;
            }
        }
        if (messages.empty()) {
            co_await co->sleep_for(std::chrono::milliseconds(10));
            continue;
        }
        if (!feed->running() || !feed->registered()) {
            break;
        }
        for (auto& msg : messages) {
            if (!feed->enabled()) {
                break;
            }
            feed->record_dispatched(msg.size());
            if (data_feed_debug) {
                std::cout << std::format("{} before send {}\n", key, thread_id_str());
            }
            co_await feed->send_coro(std::string_view(msg.data(), msg.size()));
            if (data_feed_debug) {
                std::cout << std::format("{} after send {}\n", key, thread_id_str());
            }
        }
        co_await resume_on(co->io.get_executor());
    }
    std::cout << std::format("{} feed loop exit {}\n", key, thread_id_str());
    co_return;
}
asio::awaitable<void> Data_Source::loop_coro() {
    auto co = Coro::instance();
    auto process_strand = std::make_shared<asio::strand<asio::thread_pool::executor_type>>(co->process_data->get_executor());
    while (running()) {
        if (data_source_debug) {
            std::cout << std::format("[1] ds:{} source loop begin {}\n", key, thread_id_str()) << std::flush;;
        }
        if (!running() || !registered()) {
            break;
        }
        if (data_source_debug) {
            std::cout << std::format("[2] ds:{} before handle thread:{}\n", key, thread_id_str()) << std::flush;;
        }
        co_await handle_in_loop_coro();
        if (!enabled() || !running() || !registered()) {
            break;
        }
        if (data_source_debug) {
            std::cout << std::format("[3]  ds:{} before read thread:{}\n", key, thread_id_str()) << std::flush;;
        }
        auto mode_data = co_await read_coro();
        if (data_source_debug) {
            std::cout << std::format("[4]  ds:{} after read thread:{}\n", key, thread_id_str()) << std::flush;;
        }
        const auto wait = Global::instance()->mode_acs.settings.member<&Mode_ACS_Config_Data::wait_process_msg>().read([](const auto& value) { return value; });
        if (wait) {
            // 调度到别的线程
            co_await resume_on(co->process_data->get_executor());
            if (data_source_debug) {
                std::cout << std::format("wait  [5]  ds:{} before process {}  thread:{} \n", key, thread_id_str(),
                                         mode_data.size()) <<
                std::flush;;
            }
            auto num = process_mode_acs_data(mode_data);
            if (data_source_debug) {
                std::cout << std::format("[6]  ds:{} after process {} {}\n", key, num, thread_id_str()) << std::flush;;
            }
            // 调度回io主线程
            co_await resume_on(co->io.get_executor());
            static Frequency_Limit_Multi mt(0.1);
            auto name = type + ":" + key;
            if (mt.test(name)) {
                if (num == 0) {
                    if (data_source_debug)
                        std::cout << std::format("[7]  {} 没解析到数据睡眠10ms {} {}\n", name, num,
                                                 thread_id_str()) << std::flush;;
                    ask_sleep = true;
                }
                else {
                    if (data_source_debug)
                        std::cout << std::format("[7]   {} 解析到数据 {} {}\n", name, num,
                                                 thread_id_str()) << std::flush;;
                }
            }
            if (num == 0) {
                co_await co->sleep_for(std::chrono::milliseconds(10));
                ask_sleep = false;
            }
        }
        else {
            asio::post(*process_strand, [this, mode_data = std::move(mode_data)]() mutable {
                try {
                    if (data_source_debug) {
                        std::cout << std::format("no wait [5]  ds:{} before process {}  size:{}\n", key,
                                                 thread_id_str(),
                                                 mode_data.size());
                    }
                    auto num = process_mode_acs_data(mode_data);
                    if (data_source_debug) {
                        std::cout << std::format("[6] ds:{} after process {} {}\n", key, num, thread_id_str());
                    }
                    static Frequency_Limit_Multi mt(0.1);
                    auto name = type + ":" + key;
                    if (mt.test(name)) {
                        if (num == 0) {
                            std::cout << std::format("[7]  {} 没解析到数据 {} {}\n", name, num, thread_id_str());
                            ask_sleep = true;
                        }
                        else {
                            std::cout << std::format("[8]  {} 解析到数据 {} {}\n", name, num, thread_id_str());
                        }
                    }
                }
                catch (const std::exception& e) {
                    std::cout << std::format("[9]  {} process exception: {}\n", key, e.what());
                }
                catch (...) {
                    std::cout << std::format("[9]  {} process unknown exception\n", key);
                }
            });
            co_await asio::post(co->io, asio::use_awaitable);
            if (ask_sleep) {
                co_await co->sleep_for(std::chrono::milliseconds(10));
                ask_sleep = false;
            }
        }
    }
    std::cout << std::format("{} source loop exit {}\n", key, thread_id_str());
    co_return;
}
