// Licensed to the Apache Software Foundation (ASF) under one

// or more contributor license agreements.  See the NOTICE file

// distributed with this work for additional information

// regarding copyright ownership.  The ASF licenses this file

// to you 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.

#include "util/concurrency_stats.h"

#include <chrono>

#include <sstream>

#include "common/config.h"

#include "common/logging.h"

#include "util/thread.h"

namespace doris {

ConcurrencyStatsManager::ConcurrencyStatsManager() : _running(false) {

    // Initialize all counters in the order of read path (top to bottom)

    vscanner_get_block = new ConcurrencyCounter("vscanner");

    segment_iterator_next_batch = new ConcurrencyCounter("segment_iterator");

    column_reader_read_page = new ConcurrencyCounter("column_reader");

    page_io_decompress = new ConcurrencyCounter("page_io.decompress");

    page_io_pre_decode = new ConcurrencyCounter("page_io.pre_decode");

    page_io_insert_page_cache = new ConcurrencyCounter("page_io.insert_page_cache");

    cached_remote_reader_read_at = new ConcurrencyCounter("file_cache.read_at");

    cached_remote_reader_get_or_set = new ConcurrencyCounter("file_cache.get_or_set");

    cached_remote_reader_get_or_set_wait_lock =

            new ConcurrencyCounter("file_cache.get_or_set_wait_lock");

    cached_remote_reader_write_back = new ConcurrencyCounter("file_cache.write_back");

    cached_remote_reader_blocking = new ConcurrencyCounter("file_cache.blocking");

    cached_remote_reader_local_read = new ConcurrencyCounter("file_cache.local_read");

    s3_file_reader_read = new ConcurrencyCounter("s3.read");

    // Add to vector in the order they should be printed

    _counters.push_back(vscanner_get_block);

    _counters.push_back(segment_iterator_next_batch);

    _counters.push_back(column_reader_read_page);

    _counters.push_back(page_io_decompress);

    _counters.push_back(page_io_pre_decode);

    _counters.push_back(page_io_insert_page_cache);

    _counters.push_back(cached_remote_reader_read_at);

    _counters.push_back(cached_remote_reader_get_or_set);

    _counters.push_back(cached_remote_reader_get_or_set_wait_lock);

    _counters.push_back(cached_remote_reader_write_back);

    _counters.push_back(cached_remote_reader_blocking);

    _counters.push_back(cached_remote_reader_local_read);

    _counters.push_back(s3_file_reader_read);

}

ConcurrencyStatsManager::~ConcurrencyStatsManager() {

    stop();

    // Clean up counters

    for (auto* counter : _counters) {

        delete counter;

    }

    _counters.clear();

}

ConcurrencyStatsManager& ConcurrencyStatsManager::instance() {

    static ConcurrencyStatsManager instance;

    return instance;

}

void ConcurrencyStatsManager::start() {

    if (_running.exchange(true)) {

        return; // Already running

    }

    _dump_thread = std::make_unique<std::thread>([this]() { _dump_thread_func(); });

}

void ConcurrencyStatsManager::stop() {

    if (!_running.exchange(false)) {

        return; // Not running

    }

    if (_dump_thread && _dump_thread->joinable()) {

        _dump_thread->join();

    }

    _dump_thread.reset();

}

void ConcurrencyStatsManager::dump_to_log() {

    if (_counters.empty()) {

        return;

    }

    // Build single line output: CONCURRENCY_STATS name1=value1 name2=value2 ...

    std::stringstream ss;

    ss << "CONCURRENCY_STATS";

    for (const auto* counter : _counters) {

        int64_t value = counter->value();

        ss << " " << counter->name() << "=" << value;

    }

    LOG(INFO) << ss.str();

}

void ConcurrencyStatsManager::_dump_thread_func() {

    Thread::set_self_name("ConcurrencyStatsManager_dump_thread");

    while (_running.load(std::memory_order_relaxed)) {

        // Check if dumping is enabled

        if (config::enable_concurrency_stats_dump) {

            dump_to_log();

        }

        // Sleep for the configured interval

        int32_t interval_ms = config::concurrency_stats_dump_interval_ms;

        if (interval_ms <= 0) {

            interval_ms = 100; // Default to 100ms if invalid

        }

        std::this_thread::sleep_for(std::chrono::milliseconds(interval_ms));

    }

}

} // namespace doris