// 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/exec_env.h"

#include <gen_cpp/HeartbeatService_types.h>

#include <glog/logging.h>

#include <mutex>

#include <utility>

#include "common/config.h"

#include "common/logging.h"

#include "olap/olap_define.h"

#include "olap/storage_engine.h"

#include "olap/tablet_manager.h"

#include "runtime/fragment_mgr.h"

#include "runtime/frontend_info.h"

#include "runtime/load_stream_mgr.h"

#include "util/debug_util.h"

#include "util/time.h"

#include "vec/runtime/vdata_stream_mgr.h"

#include "vec/sink/delta_writer_v2_pool.h"

#include "vec/sink/load_stream_map_pool.h"

namespace doris {

#ifdef BE_TEST

void ExecEnv::set_inverted_index_searcher_cache(

        segment_v2::InvertedIndexSearcherCache* inverted_index_searcher_cache) {

    _inverted_index_searcher_cache = inverted_index_searcher_cache;

}

void ExecEnv::set_storage_engine(std::unique_ptr<BaseStorageEngine>&& engine) {

    _storage_engine = std::move(engine);

}

void ExecEnv::set_write_cooldown_meta_executors() {

    _write_cooldown_meta_executors = std::make_unique<WriteCooldownMetaExecutors>();

}

#endif // BE_TEST

Result<BaseTabletSPtr> ExecEnv::get_tablet(int64_t tablet_id, SyncRowsetStats* sync_stats,

                                           bool force_use_only_cached, bool cache_on_miss) {

    auto storage_engine = GetInstance()->_storage_engine.get();

    return storage_engine != nullptr

                   ? storage_engine->get_tablet(tablet_id, sync_stats, force_use_only_cached,

                                                cache_on_miss)

                   : ResultError(Status::InternalError("failed to get tablet {}", tablet_id));

}

Status ExecEnv::get_tablet_meta(int64_t tablet_id, TabletMetaSharedPtr* tablet_meta,

                                bool force_use_only_cached) {

    auto storage_engine = GetInstance()->_storage_engine.get();

    if (storage_engine == nullptr) {

        return Status::InternalError("storage engine is not initialized");

    }

    return storage_engine->get_tablet_meta(tablet_id, tablet_meta, force_use_only_cached);

}

const std::string& ExecEnv::token() const {

    return _cluster_info->token;

}

void ExecEnv::clear_stream_mgr() {

    if (_vstream_mgr) {

        SAFE_DELETE(_vstream_mgr);

    }

}

std::vector<TFrontendInfo> ExecEnv::get_frontends() {

    std::lock_guard<std::mutex> lg(_frontends_lock);

    std::vector<TFrontendInfo> infos;

    for (const auto& cur_fe : _frontends) {

        infos.push_back(cur_fe.second.info);

    }

    return infos;

}

void ExecEnv::update_frontends(const std::vector<TFrontendInfo>& new_fe_infos) {

    std::lock_guard<std::mutex> lg(_frontends_lock);

    std::set<TNetworkAddress> dropped_fes;

    for (const auto& cur_fe : _frontends) {

        dropped_fes.insert(cur_fe.first);

    }

    for (const auto& coming_fe_info : new_fe_infos) {

        auto itr = _frontends.find(coming_fe_info.coordinator_address);

        if (itr == _frontends.end()) {

            LOG(INFO) << "A completely new frontend, " << PrintFrontendInfo(coming_fe_info);

            _frontends.insert(std::pair<TNetworkAddress, FrontendInfo>(

                    coming_fe_info.coordinator_address,

                    FrontendInfo {coming_fe_info, GetCurrentTimeMicros() / 1000, /*first time*/

                                  GetCurrentTimeMicros() / 1000 /*last time*/}));

            continue;

        }

        dropped_fes.erase(coming_fe_info.coordinator_address);

        if (coming_fe_info.process_uuid == 0) {

            LOG(WARNING) << "Frontend " << PrintFrontendInfo(coming_fe_info)

                         << " is in an unknown state.";

        }

        if (coming_fe_info.process_uuid == itr->second.info.process_uuid) {

            itr->second.last_reveiving_time_ms = GetCurrentTimeMicros() / 1000;

            continue;

        }

        // If we get here, means this frontend has already restarted.

        itr->second.info.process_uuid = coming_fe_info.process_uuid;

        itr->second.first_receiving_time_ms = GetCurrentTimeMicros() / 1000;

        itr->second.last_reveiving_time_ms = GetCurrentTimeMicros() / 1000;

        LOG(INFO) << "Update frontend " << PrintFrontendInfo(coming_fe_info);

    }

    for (const auto& dropped_fe : dropped_fes) {

        LOG(INFO) << "Frontend " << PrintThriftNetworkAddress(dropped_fe)

                  << " has already been dropped, remove it";

        _frontends.erase(dropped_fe);

    }

}

std::map<TNetworkAddress, FrontendInfo> ExecEnv::get_running_frontends() {

    std::lock_guard<std::mutex> lg(_frontends_lock);

    std::map<TNetworkAddress, FrontendInfo> res;

    const int expired_duration = config::fe_expire_duration_seconds * 1000;

    const auto now = GetCurrentTimeMicros() / 1000;

    for (const auto& pair : _frontends) {

        auto& brpc_addr = pair.first;

        auto& fe_info = pair.second;

        if (fe_info.info.process_uuid == 0) {

            // FE is in an unknown state, regart it as alive. conservative

            res[brpc_addr] = fe_info;

        } else {

            if (now - fe_info.last_reveiving_time_ms < expired_duration) {

                // If fe info has just been update in last expired_duration, regard it as running.

                res[brpc_addr] = fe_info;

            } else {

                // Fe info has not been udpate for more than expired_duration, regard it as an abnormal.

                // Abnormal means this fe can not connect to master, and it is not dropped from cluster.

                // or fe do not have master yet.

                LOG_EVERY_N(WARNING, 50) << fmt::format(

                        "Frontend {}:{} has not update its hb for more than {} secs, regard it as "

                        "abnormal",

                        brpc_addr.hostname, brpc_addr.port, config::fe_expire_duration_seconds);

            }

        }

    }

    return res;

}

void ExecEnv::wait_for_all_tasks_done() {

    // For graceful shutdown, need to wait for all running queries to stop

    int32_t wait_seconds_passed = 0;

    while (true) {

        int num_queries = _fragment_mgr->running_query_num();

        if (num_queries < 1) {

            break;

        }

        if (wait_seconds_passed > doris::config::grace_shutdown_wait_seconds) {

            LOG(INFO) << "There are still " << num_queries << " queries running, but "

                      << wait_seconds_passed << " seconds passed, has to exist now";

            break;

        }

        LOG(INFO) << "There are still " << num_queries << " queries running, waiting...";

        sleep(1);

        ++wait_seconds_passed;

    }

    // This is a conservative strategy.

    // Because a query might still have fragments running on other BE nodes.

    // In other words, the query hasn't truly terminated.

    // If the current BE is shut down at this point,

    // the FE will detect the downtime of a related BE and cancel the entire query,

    // defeating the purpose of a graceful stop.

    sleep(config::grace_shutdown_post_delay_seconds);

}

bool ExecEnv::check_auth_token(const std::string& auth_token) {

    return _cluster_info->curr_auth_token == auth_token ||

           _cluster_info->last_auth_token == auth_token;

}

} // namespace doris