// 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 "runtime/routine_load/data_consumer_pool.h"

namespace doris {

Status DataConsumerPool::get_consumer(
        StreamLoadContext* ctx,
        std::shared_ptr<DataConsumer>* ret) {

    std::unique_lock<std::mutex> l(_lock);

    // check if there is an available consumer.
    // if has, return it, also remove it from the pool
    auto iter = std::begin(_pool);
    while (iter != std::end(_pool)) {
        if ((*iter)->match(ctx)) {
            VLOG(3) << "get an available data consumer from pool: " << (*iter)->id();
            (*iter)->reset();
            *ret = *iter;
            iter = _pool.erase(iter);
            return Status::OK; 
        } else {
            ++iter;
        }
    }

    // no available consumer, create a new one
    std::shared_ptr<DataConsumer> consumer;
    switch (ctx->load_src_type) {
        case TLoadSourceType::KAFKA:
            consumer = std::make_shared<KafkaDataConsumer>(ctx);
            break;
        default:
            std::stringstream ss;
            ss << "unknown routine load task type: " << ctx->load_type;
            return Status(ss.str());
    }
    
    // init the consumer
    RETURN_IF_ERROR(consumer->init(ctx));

    VLOG(3) << "create new data consumer: " << consumer->id();
    *ret = consumer;
    return Status::OK;
}

void DataConsumerPool::return_consumer(std::shared_ptr<DataConsumer> consumer) {
    std::unique_lock<std::mutex> l(_lock);

    if (_pool.size() == _max_pool_size) {
        VLOG(3) << "data consumer pool is full: " << _pool.size()
                << "-" << _max_pool_size << ", discard the returned consumer: "
                << consumer->id();
        return;
    }

    consumer->reset();
    _pool.push_back(consumer);
    VLOG(3) << "return the data consumer: " << consumer->id()
            << ", current pool size: " << _pool.size();
    return;
}

Status DataConsumerPool::start_bg_worker() {
    _clean_idle_consumer_thread = std::thread(
        [this] {
            #ifdef GOOGLE_PROFILER
                ProfilerRegisterThread();
            #endif

            uint32_t interval = 60;
            while (true) {
                _clean_idle_consumer_bg();
                sleep(interval);
            }
        });
    _clean_idle_consumer_thread.detach();
    return Status::OK;
}

void DataConsumerPool::_clean_idle_consumer_bg() {
    const static int32_t max_idle_time_second = 600;

    std::unique_lock<std::mutex> l(_lock);
    time_t now = time(nullptr);    

    auto iter = std::begin(_pool);
    while (iter != std::end(_pool)) {
        if (difftime(now, (*iter)->last_visit_time()) >= max_idle_time_second) {
            LOG(INFO) << "remove data consumer " << (*iter)->id()
                    << ", since it last visit: " << (*iter)->last_visit_time()
                    << ", now: " << now;
            iter = _pool.erase(iter);
        } else {
            ++iter;
        }
    }
}

} // end namespace doris