text:
// Copyright 2022 PingCAP, 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.

#include "Server.h"

#include <AggregateFunctions/registerAggregateFunctions.h>
#include <Common/CPUAffinityManager.h>
#include <Common/ClickHouseRevision.h>
#include <Common/Config/ConfigReloader.h>
#include <Common/CurrentMetrics.h>
#include <Common/DynamicThreadPool.h>
#include <Common/Macros.h>
#include <Common/RedactHelpers.h>
#include <Common/StringUtils/StringUtils.h>
#include <Common/ThreadManager.h>
#include <Common/TiFlashBuildInfo.h>
#include <Common/TiFlashException.h>
#include <Common/TiFlashMetrics.h>
#include <Common/assert_cast.h>
#include <Common/config.h>
#include <Common/escapeForFileName.h>
#include <Common/formatReadable.h>
#include <Common/getFQDNOrHostName.h>
#include <Common/getMultipleKeysFromConfig.h>
#include <Common/getNumberOfPhysicalCPUCores.h>
#include <Common/setThreadName.h>
#include <Encryption/DataKeyManager.h>
#include <Encryption/FileProvider.h>
#include <Encryption/MockKeyManager.h>
#include <Encryption/RateLimiter.h>
#include <Flash/DiagnosticsService.h>
#include <Flash/FlashService.h>
#include <Flash/Mpp/GRPCCompletionQueuePool.h>
#include <Functions/registerFunctions.h>
#include <IO/HTTPCommon.h>
#include <IO/ReadHelpers.h>
#include <IO/createReadBufferFromFileBase.h>
#include <Interpreters/AsynchronousMetrics.h>
#include <Interpreters/IDAsPathUpgrader.h>
#include <Interpreters/ProcessList.h>
#include <Interpreters/loadMetadata.h>
#include <Poco/DirectoryIterator.h>
#include <Poco/Net/HTTPServer.h>
#include <Poco/Net/NetException.h>
#include <Poco/StringTokenizer.h>
#include <Poco/Timestamp.h>
#include <Server/RaftConfigParser.h>
#include <Server/StorageConfigParser.h>
#include <Server/UserConfigParser.h>
#include <Storages/FormatVersion.h>
#include <Storages/IManageableStorage.h>
#include <Storages/PathCapacityMetrics.h>
#include <Storages/System/attachSystemTables.h>
#include <Storages/Transaction/FileEncryption.h>
#include <Storages/Transaction/KVStore.h>
#include <Storages/Transaction/ProxyFFI.h>
#include <Storages/Transaction/TMTContext.h>
#include <Storages/registerStorages.h>
#include <TableFunctions/registerTableFunctions.h>
#include <TiDB/Schema/SchemaSyncer.h>
#include <WindowFunctions/registerWindowFunctions.h>
#include <common/ErrorHandlers.h>
#include <common/config_common.h>
#include <common/getMemoryAmount.h>
#include <common/logger_useful.h>
#include <sys/resource.h>

#include <boost/algorithm/string/classification.hpp>
#include <boost/algorithm/string/split.hpp>
#include <ext/scope_guard.h>
#include <limits>
#include <memory>

#include "HTTPHandlerFactory.h"
#include "MetricsPrometheus.h"
#include "MetricsTransmitter.h"
#include "StatusFile.h"
#include "TCPHandlerFactory.h"

#if Poco_NetSSL_FOUND
#include <Poco/Net/Context.h>
#include <Poco/Net/SecureServerSocket.h>
#include <grpc++/grpc++.h>
#endif

#if USE_JEMALLOC
#include <jemalloc/jemalloc.h>
#endif

#if USE_MIMALLOC
#include <Poco/JSON/Parser.h>
#include <mimalloc.h>

#include <fstream>
#endif

#ifdef FIU_ENABLE
#include <fiu.h>
#endif

#if USE_MIMALLOC
#define TRY_LOAD_CONF(NAME)                          \
    {                                                \
        try                                          \
        {                                            \
            auto value = obj->getValue<long>(#NAME); \
            mi_option_set(NAME, value);              \
        }                                            \
        catch (...)                                  \
        {                                            \
        }                                            \
    }

void loadMiConfig(Logger * log)
{
    auto config = getenv("MIMALLOC_CONF");
    if (config)
    {
        LOG_FMT_INFO(log, "Got environment variable MIMALLOC_CONF: {}", config);
        Poco::JSON::Parser parser;
        std::ifstream data{config};
        Poco::Dynamic::Var result = parser.parse(data);
        auto obj = result.extract<Poco::JSON::Object::Ptr>();
        TRY_LOAD_CONF(mi_option_show_errors);
        TRY_LOAD_CONF(mi_option_show_stats);
        TRY_LOAD_CONF(mi_option_verbose);
        TRY_LOAD_CONF(mi_option_eager_commit);
        TRY_LOAD_CONF(mi_option_eager_region_commit);
        TRY_LOAD_CONF(mi_option_large_os_pages);
        TRY_LOAD_CONF(mi_option_reserve_huge_os_pages);
        TRY_LOAD_CONF(mi_option_segment_cache);
        TRY_LOAD_CONF(mi_option_page_reset);
        TRY_LOAD_CONF(mi_option_segment_reset);
        TRY_LOAD_CONF(mi_option_reset_delay);
        TRY_LOAD_CONF(mi_option_use_numa_nodes);
        TRY_LOAD_CONF(mi_option_reset_decommits);
        TRY_LOAD_CONF(mi_option_eager_commit_delay);
        TRY_LOAD_CONF(mi_option_os_tag);
    }
}
#undef TRY_LOAD_CONF
#endif
namespace
{
[[maybe_unused]] void tryLoadBoolConfigFromEnv(Poco::Logger * log, bool & target, const char * name)
{
    auto * config = getenv(name);
    if (config)
    {
        LOG_FMT_INFO(log, "Got environment variable {} = {}", name, config);
        try
        {
            auto result = std::stoul(config);
            if (result != 0 && result != 1)
            {
                LOG_FMT_ERROR(log, "Environment variable{} = {} is not valid", name, result);
                return;
            }
            target = result;
        }
        catch (...)
        {
        }
    }
}
} // namespace

namespace CurrentMetrics
{
extern const Metric Revision;
}

namespace DB
{
namespace ErrorCodes
{
extern const int NO_ELEMENTS_IN_CONFIG;
extern const int SUPPORT_IS_DISABLED;
extern const int ARGUMENT_OUT_OF_BOUND;
extern const int INVALID_CONFIG_PARAMETER;
} // namespace ErrorCodes

namespace Debug
{
extern void setServiceAddr(const std::string & addr);
}

static std::string getCanonicalPath(std::string path)
{
    Poco::trimInPlace(path);
    if (path.empty())
        throw Exception("path configuration parameter is empty");
    if (path.back() != '/')
        path += '/';
    return path;
}

void Server::uninitialize()
{
    logger().information("shutting down");
    BaseDaemon::uninitialize();
}

void Server::initialize(Poco::Util::Application & self)
{
    BaseDaemon::initialize(self);
    logger().information("starting up");
}

std::string Server::getDefaultCorePath() const
{
    return getCanonicalPath(config().getString("path")) + "cores";
}

struct TiFlashProxyConfig
{
    static const std::string config_prefix;
    std::vector<const char *> args;
    std::unordered_map<std::string, std::string> val_map;
    bool is_proxy_runnable = false;

    // TiFlash Proxy will set the default value of "flash.proxy.addr", so we don't need to set here.
    const String engine_store_version = "engine-version";
    const String engine_store_git_hash = "engine-git-hash";
    const String engine_store_address = "engine-addr";
    const String engine_store_advertise_address = "advertise-engine-addr";
    const String pd_endpoints = "pd-endpoints";
    const String engine_label = "engine-label";
    const String engine_label_value = "tiflash";

    explicit TiFlashProxyConfig(Poco::Util::LayeredConfiguration & config)
    {
        if (!config.has(config_prefix))
            return;

        Poco::Util::AbstractConfiguration::Keys keys;
        config.keys(config_prefix, keys);
        {
            std::unordered_map<std::string, std::string> args_map;
            for (const auto & key : keys)
            {
                const auto k = config_prefix + "." + key;
                args_map[key] = config.getString(k);
            }
            args_map[pd_endpoints] = config.getString("raft.pd_addr");
            args_map[engine_store_version] = TiFlashBuildInfo::getReleaseVersion();
            args_map[engine_store_git_hash] = TiFlashBuildInfo::getGitHash();
            if (!args_map.count(engine_store_address))
                args_map[engine_store_address] = config.getString("flash.service_addr");
            else
                args_map[engine_store_advertise_address] = args_map[engine_store_address];
            args_map[engine_label] = engine_label_value;

            for (auto && [k, v] : args_map)
            {
                val_map.emplace("--" + k, std::move(v));
            }
        }

        args.push_back("TiFlash Proxy");
        for (const auto & v : val_map)
        {
            args.push_back(v.first.data());
            args.push_back(v.second.data());
        }
        is_proxy_runnable = true;
    }
};

const std::string TiFlashProxyConfig::config_prefix = "flash.proxy";

pingcap::ClusterConfig getClusterConfig(const TiFlashSecurityConfig & security_config, const TiFlashRaftConfig & raft_config)
{
    pingcap::ClusterConfig config;
    config.tiflash_engine_key = raft_config.engine_key;
    config.tiflash_engine_value = raft_config.engine_value;
    config.ca_path = security_config.ca_path;
    config.cert_path = security_config.cert_path;
    config.key_path = security_config.key_path;
    return config;
}

Poco::Logger * grpc_log = nullptr;

void printGRPCLog(gpr_log_func_args * args)
{
    String log_msg = fmt::format("{}, line number: {}, log msg : {}", args->file, args->line, args->message);
    if (args->severity == GPR_LOG_SEVERITY_DEBUG)
    {
        LOG_DEBUG(grpc_log, log_msg);
    }
    else if (args->severity == GPR_LOG_SEVERITY_INFO)
    {
        LOG_INFO(grpc_log, log_msg);
    }
    else if (args->severity == GPR_LOG_SEVERITY_ERROR)
    {
        LOG_ERROR(grpc_log, log_msg);
    }
}

struct HTTPServer : Poco::Net::HTTPServer
{
    HTTPServer(Poco::Net::HTTPRequestHandlerFactory::Ptr pFactory, Poco::ThreadPool & threadPool, const Poco::Net::ServerSocket & socket, Poco::Net::HTTPServerParams::Ptr pParams)
        : Poco::Net::HTTPServer(pFactory, threadPool, socket, pParams)
    {}

protected:
    void run() override
    {
        setThreadName("HTTPServer");
        Poco::Net::HTTPServer::run();
    }
};

struct TCPServer : Poco::Net::TCPServer
{
    TCPServer(Poco::Net::TCPServerConnectionFactory::Ptr pFactory, Poco::ThreadPool & threadPool, const Poco::Net::ServerSocket & socket, Poco::Net::TCPServerParams::Ptr pParams)
        : Poco::Net::TCPServer(pFactory, threadPool, socket, pParams)
    {}

protected:
    void run() override
    {
        setThreadName("TCPServer");
        Poco::Net::TCPServer::run();
    }
};

void UpdateMallocConfig([[maybe_unused]] Poco::Logger * log)
{
#ifdef RUN_FAIL_RETURN
    static_assert(false);
#endif
#define RUN_FAIL_RETURN(f)                                        \
    do                                                            \
    {                                                             \
        if (f)                                                    \
        {                                                         \
            LOG_FMT_ERROR(log, "Fail to update jemalloc config"); \
            return;                                               \
        }                                                         \
    } while (0)
#if USE_JEMALLOC
    const char * version;
    bool old_b, new_b = true;
    size_t old_max_thd, new_max_thd = 1;
    size_t sz_b = sizeof(bool), sz_st = sizeof(size_t), sz_ver = sizeof(version);

    RUN_FAIL_RETURN(je_mallctl("version", &version, &sz_ver, nullptr, 0));
    LOG_FMT_INFO(log, "Got jemalloc version: {}", version);

    auto * malloc_conf = getenv("MALLOC_CONF");
    if (malloc_conf)
    {
        LOG_FMT_INFO(log, "Got environment variable MALLOC_CONF: {}", malloc_conf);
    }
    else
    {
        LOG_FMT_INFO(log, "Not found environment variable MALLOC_CONF");
    }

    RUN_FAIL_RETURN(je_mallctl("opt.background_thread", (void *)&old_b, &sz_b, nullptr, 0));
    RUN_FAIL_RETURN(je_mallctl("opt.max_background_threads", (void *)&old_max_thd, &sz_st, nullptr, 0));

    LOG_FMT_INFO(log, "Got jemalloc config: opt.background_thread {}, opt.max_background_threads {}", old_b, old_max_thd);

    if (!malloc_conf && !old_b)
    {
        LOG_FMT_INFO(log, "Try to use background_thread of jemalloc to handle purging asynchronously");

        RUN_FAIL_RETURN(je_mallctl("max_background_threads", nullptr, nullptr, (void *)&new_max_thd, sz_st));
        LOG_FMT_INFO(log, "Set jemalloc.max_background_threads {}", new_max_thd);

        RUN_FAIL_RETURN(je_mallctl("background_thread", nullptr, nullptr, (void *)&new_b, sz_b));
        LOG_FMT_INFO(log, "Set jemalloc.background_thread {}", new_b);
    }
#endif

#if USE_MIMALLOC
#define MI_OPTION_SHOW(OPTION) LOG_FMT_INFO(log, "mimalloc." #OPTION ": {}", mi_option_get(OPTION));

    int version = mi_version();
    LOG_FMT_INFO(log, "Got mimalloc version: {}.{}.{}", (version / 100), ((version % 100) / 10), (version % 10));
    loadMiConfig(log);
    MI_OPTION_SHOW(mi_option_show_errors);
    MI_OPTION_SHOW(mi_option_show_stats);
    MI_OPTION_SHOW(mi_option_verbose);
    MI_OPTION_SHOW(mi_option_eager_commit);
    MI_OPTION_SHOW(mi_option_eager_region_commit);
    MI_OPTION_SHOW(mi_option_large_os_pages);
    MI_OPTION_SHOW(mi_option_reserve_huge_os_pages);
    MI_OPTION_SHOW(mi_option_segment_cache);
    MI_OPTION_SHOW(mi_option_page_reset);
    MI_OPTION_SHOW(mi_option_segment_reset);
    MI_OPTION_SHOW(mi_option_reset_delay);
    MI_OPTION_SHOW(mi_option_use_numa_nodes);
    MI_OPTION_SHOW(mi_option_reset_decommits);
    MI_OPTION_SHOW(mi_option_eager_commit_delay);
    MI_OPTION_SHOW(mi_option_os_tag);
#undef MI_OPTION_SHOW
#endif
#undef RUN_FAIL_RETURN
}

extern "C" {
void run_raftstore_proxy_ffi(int argc, const char * const * argv, const EngineStoreServerHelper *);
}

struct RaftStoreProxyRunner : boost::noncopyable
{
    struct RunRaftStoreProxyParms
    {
        const EngineStoreServerHelper * helper;
        const TiFlashProxyConfig & conf;

        /// set big enough stack size to avoid runtime error like stack-overflow.
        size_t stack_size = 1024 * 1024 * 20;
    };

    RaftStoreProxyRunner(RunRaftStoreProxyParms && parms_, Poco::Logger * log_)
        : parms(std::move(parms_))
        , log(log_)
    {}

    void join() const
    {
        if (!parms.conf.is_proxy_runnable)
            return;
        pthread_join(thread, nullptr);
    }

    void run()
    {
        if (!parms.conf.is_proxy_runnable)
            return;
        pthread_attr_t attribute;
        pthread_attr_init(&attribute);
        pthread_attr_setstacksize(&attribute, parms.stack_size);
        LOG_FMT_INFO(log, "start raft store proxy");
        pthread_create(&thread, &attribute, runRaftStoreProxyFFI, &parms);
        pthread_attr_destroy(&attribute);
    }

private:
    static void * runRaftStoreProxyFFI(void * pv)
    {
        setThreadName("RaftStoreProxy");
        const auto & parms = *static_cast<const RunRaftStoreProxyParms *>(pv);
        run_raftstore_proxy_ffi(static_cast<int>(parms.conf.args.size()), parms.conf.args.data(), parms.helper);
        return nullptr;
    }

    RunRaftStoreProxyParms parms;
    pthread_t thread{};
    Poco::Logger * log;
};

// We only need this task run once.
void initStores(Context & global_context, Poco::Logger * log, bool lazily_init_store)
{
    auto do_init_stores = [&global_context, log]() {
        auto storages = global_context.getTMTContext().getStorages().getAllStorage();
        int init_cnt = 0;
        int err_cnt = 0;
        for (auto & [table_id, storage] : storages)
        {
            // This will skip the init of storages that do not contain any data. TiFlash now sync the schema and
            // create all tables regardless the table have define TiFlash replica or not, so there may be lots
            // of empty tables in TiFlash.
            // Note that we still need to init stores that contains data (defined by the stable dir of this storage
            // is exist), or the data used size reported to PD is not correct.
            try
            {
                init_cnt += storage->initStoreIfDataDirExist() ? 1 : 0;
                LOG_FMT_INFO(log, "Storage inited done [table_id={}]", table_id);
            }
            catch (...)
            {
                err_cnt++;
                tryLogCurrentException(log, fmt::format("Storage inited fail, [table_id={}]", table_id));
            }
        }
        LOG_FMT_INFO(
            log,
            "Storage inited finish. [total_count={}] [init_count={}] [error_count={}]",
            storages.size(),
            init_cnt,
            err_cnt);
    };
    if (lazily_init_store)
    {
        LOG_FMT_INFO(log, "Lazily init store.");
        // apply the inited in another thread to shorten the start time of TiFlash
        std::thread(do_init_stores).detach();
    }
    else
    {
        LOG_FMT_INFO(log, "Not lazily init store.");
        do_init_stores();
    }
}

void handleRpcs(grpc::ServerCompletionQueue * curcq, Poco::Logger * log)
{
    GET_METRIC(tiflash_thread_count, type_total_rpc_async_worker).Increment();
    SCOPE_EXIT({
        GET_METRIC(tiflash_thread_count, type_total_rpc_async_worker).Decrement();
    });
    void * tag = nullptr; // uniquely identifies a request.
    bool ok = false;
    while (true)
    {
        String err_msg;
        try
        {
            // Block waiting to read the next event from the completion queue. The
            // event is uniquely identified by its tag, which in this case is the
            // memory address of a EstablishCallData instance.
            // The return value of Next should always be checked. This return value
            // tells us whether there is any kind of event or cq is shutting down.
            if (!curcq->Next(&tag, &ok))
            {
                LOG_FMT_INFO(grpc_log, "CQ is fully drained and shut down");
                break;
            }
            GET_METRIC(tiflash_thread_count, type_active_rpc_async_worker).Increment();
            SCOPE_EXIT({
                GET_METRIC(tiflash_thread_count, type_active_rpc_async_worker).Decrement();
            });
            // If ok is false, it means server is shutdown.
            // We need not log all not ok events, since the volumn is large which will pollute the content of log.
            if (ok)
                static_cast<EstablishCallData *>(tag)->proceed();
            else
                static_cast<EstablishCallData *>(tag)->cancel();
        }
        catch (Exception & e)
        {
            err_msg = e.displayText();
            LOG_FMT_ERROR(log, "handleRpcs meets error: {} Stack Trace : {}", err_msg, e.getStackTrace().toString());
        }
        catch (pingcap::Exception & e)
        {
            err_msg = e.message();
            LOG_FMT_ERROR(log, "handleRpcs meets error: {}", err_msg);
        }
        catch (std::exception & e)
        {
            err_msg = e.what();
            LOG_FMT_ERROR(log, "handleRpcs meets error: {}", err_msg);
        }
        catch (...)
        {
            err_msg = "unrecovered error";
            LOG_FMT_ERROR(log, "handleRpcs meets error: {}", err_msg);
            throw;
        }
    }
}

class Server::FlashGrpcServerHolder
{
public:
    FlashGrpcServerHolder(Server & server, const TiFlashRaftConfig & raft_config, Poco::Logger * log_)
        : log(log_)
        , is_shutdown(std::make_shared<std::atomic<bool>>(false))
    {
        grpc::ServerBuilder builder;
        if (server.security_config.has_tls_config)
        {
            grpc::SslServerCredentialsOptions server_cred(GRPC_SSL_REQUEST_AND_REQUIRE_CLIENT_CERTIFICATE_AND_VERIFY);
            auto options = server.security_config.readAndCacheSecurityInfo();
            server_cred.pem_root_certs = options.pem_root_certs;
            server_cred.pem_key_cert_pairs.push_back(
                grpc::SslServerCredentialsOptions::PemKeyCertPair{options.pem_private_key, options.pem_cert_chain});
            builder.AddListeningPort(raft_config.flash_server_addr, grpc::SslServerCredentials(server_cred));
        }
        else
        {
            builder.AddListeningPort(raft_config.flash_server_addr, grpc::InsecureServerCredentials());
        }

        /// Init and register flash service.
        bool enable_async_server = server.context().getSettingsRef().enable_async_server;
        if (enable_async_server)
            flash_service = std::make_unique<AsyncFlashService>(server);
        else
            flash_service = std::make_unique<FlashService>(server);
        diagnostics_service = std::make_unique<DiagnosticsService>(server);
        builder.SetOption(grpc::MakeChannelArgumentOption(GRPC_ARG_HTTP2_MIN_RECV_PING_INTERVAL_WITHOUT_DATA_MS, 5 * 1000));
        builder.SetOption(grpc::MakeChannelArgumentOption(GRPC_ARG_HTTP2_MIN_SENT_PING_INTERVAL_WITHOUT_DATA_MS, 10 * 1000));
        builder.SetOption(grpc::MakeChannelArgumentOption(GRPC_ARG_KEEPALIVE_PERMIT_WITHOUT_CALLS, 1));
        // number of grpc thread pool's non-temporary threads, better tune it up to avoid frequent creation/destruction of threads
        auto max_grpc_pollers = server.context().getSettingsRef().max_grpc_pollers;
        if (max_grpc_pollers > 0 && max_grpc_pollers <= std::numeric_limits<int>::max())
            builder.SetSyncServerOption(grpc::ServerBuilder::SyncServerOption::MAX_POLLERS, max_grpc_pollers);
        builder.RegisterService(flash_service.get());
        LOG_FMT_INFO(log, "Flash service registered");
        builder.RegisterService(diagnostics_service.get());
        LOG_FMT_INFO(log, "Diagnostics service registered");

        /// Kick off grpc server.
        // Prevent TiKV from throwing "Received message larger than max (4404462 vs. 4194304)" error.
        builder.SetMaxReceiveMessageSize(-1);
        builder.SetMaxSendMessageSize(-1);
        thread_manager = DB::newThreadManager();
        int async_cq_num = server.context().getSettingsRef().async_cqs;
        if (enable_async_server)
        {
            for (int i = 0; i < async_cq_num; ++i)
            {
                cqs.emplace_back(builder.AddCompletionQueue());
                notify_cqs.emplace_back(builder.AddCompletionQueue());
            }
        }
        flash_grpc_server = builder.BuildAndStart();
        LOG_FMT_INFO(log, "Flash grpc server listening on [{}]", raft_config.flash_server_addr);
        Debug::setServiceAddr(raft_config.flash_server_addr);
        if (enable_async_server)
        {
            int preallocated_request_count_per_poller = server.context().getSettingsRef().preallocated_request_count_per_poller;
            int pollers_per_cq = server.context().getSettingsRef().async_pollers_per_cq;
            for (int i = 0; i < async_cq_num * pollers_per_cq; ++i)
            {
                auto * cq = cqs[i / pollers_per_cq].get();
                auto * notify_cq = notify_cqs[i / pollers_per_cq].get();
                for (int j = 0; j < preallocated_request_count_per_poller; ++j)
                {
                    // EstablishCallData will handle its lifecycle by itself.
                    EstablishCallData::spawn(assert_cast<AsyncFlashService *>(flash_service.get()), cq, notify_cq, is_shutdown);
                }
                thread_manager->schedule(false, "async_poller", [cq, this] { handleRpcs(cq, log); });
                thread_manager->schedule(false, "async_poller", [notify_cq, this] { handleRpcs(notify_cq, log); });
            }
        }
    }

    ~FlashGrpcServerHolder()
    {
        try
        {
            /// Shut down grpc server.
            LOG_FMT_INFO(log, "Begin to shut down flash grpc server");
            flash_grpc_server->Shutdown();
            *is_shutdown = true;
            // Wait all existed MPPTunnels done to prevent crash.
            // If all existed MPPTunnels are done, almost in all cases it means all existed MPPTasks and ExchangeReceivers are also done.
            const int max_wait_cnt = 300;
            int wait_cnt = 0;
            while (GET_METRIC(tiflash_object_count, type_count_of_mpptunnel).Value() >= 1 && (wait_cnt++ < max_wait_cnt))
                std::this_thread::sleep_for(std::chrono::seconds(1));

            for (auto & cq : cqs)
                cq->Shutdown();
            for (auto & cq : notify_cqs)
                cq->Shutdown();
            thread_manager->wait();
            flash_grpc_server->Wait();
            flash_grpc_server.reset();
            LOG_FMT_INFO(log, "Shut down flash grpc server");

            /// Close flash service.
            LOG_FMT_INFO(log, "Begin to shut down flash service");
            flash_service.reset();
            LOG_FMT_INFO(log, "Shut down flash service");
        }
        catch (...)
        {
            auto message = getCurrentExceptionMessage(false);
            LOG_FMT_FATAL(log, "Exception happens in destructor of FlashGrpcServerHolder with message: {}", message);
            std::terminate();
        }
    }

private:
    Poco::Logger * log;
    std::shared_ptr<std::atomic<bool>> is_shutdown;
    std::unique_ptr<FlashService> flash_service = nullptr;
    std::unique_ptr<DiagnosticsService> diagnostics_service = nullptr;
    std::unique_ptr<grpc::Server> flash_grpc_server = nullptr;
    // cqs and notify_cqs are used for processing async grpc events (currently only EstablishMPPConnection).
    std::vector<std::unique_ptr<grpc::ServerCompletionQueue>> cqs;
    std::vector<std::unique_ptr<grpc::ServerCompletionQueue>> notify_cqs;
    std::shared_ptr<ThreadManager> thread_manager;
};

class Server::TcpHttpServersHolder
{
public:
    TcpHttpServersHolder(Server & server_, const Settings & settings, Poco::Logger * log_)
        : server(server_)
        , log(log_)
        , server_pool(1, server.config().getUInt("max_connections", 1024))
    {
        auto & config = server.config();
        auto & security_config = server.security_config;

        Poco::Timespan keep_alive_timeout(config.getUInt("keep_alive_timeout", 10), 0);
        Poco::Net::HTTPServerParams::Ptr http_params = new Poco::Net::HTTPServerParams; // NOLINT
        http_params->setTimeout(settings.receive_timeout);
        http_params->setKeepAliveTimeout(keep_alive_timeout);

        std::vector<std::string> listen_hosts = DB::getMultipleValuesFromConfig(config, "", "listen_host");

        bool listen_try = config.getBool("listen_try", false);
        if (listen_hosts.empty())
        {
            listen_hosts.emplace_back("0.0.0.0");
            listen_try = true;
        }

        auto make_socket_address = [&](const std::string & host, UInt16 port) {
            Poco::Net::SocketAddress socket_address;
            try
            {
                socket_address = Poco::Net::SocketAddress(host, port);
            }
            catch (const Poco::Net::DNSException & e)
            {
                const auto code = e.code();
                if (code == EAI_FAMILY
#if defined(EAI_ADDRFAMILY)
                    || code == EAI_ADDRFAMILY
#endif
                )
                {
                    LOG_FMT_ERROR(
                        log,
                        "Cannot resolve listen_host ({}), error {}: {}."
                        "If it is an IPv6 address and your host has disabled IPv6, then consider to "
                        "specify IPv4 address to listen in <listen_host> element of configuration "
                        "file. Example: <listen_host>0.0.0.0</listen_host>",
                        host,
                        e.code(),
                        e.message());
                }

                throw;
            }
            return socket_address;
        };

        auto socket_bind_listen = [&](auto & socket, const std::string & host, UInt16 port, bool secure = false) {
            auto address = make_socket_address(host, port);
#if !POCO_CLICKHOUSE_PATCH || POCO_VERSION <= 0x02000000 // TODO: fill correct version
            if (secure)
                /// Bug in old poco, listen() after bind() with reusePort param will fail because have no implementation in SecureServerSocketImpl
                /// https://github.com/pocoproject/poco/pull/2257
                socket.bind(address, /* reuseAddress = */ true);
            else
#endif
#if POCO_VERSION < 0x01080000
                socket.bind(address, /* reuseAddress = */ true);
#else
            socket.bind(address, /* reuseAddress = */ true, /* reusePort = */ config.getBool("listen_reuse_port", false));
#endif

            socket.listen(/* backlog = */ config.getUInt("listen_backlog", 64));

            return address;
        };

        for (const auto & listen_host : listen_hosts)
        {
            /// For testing purposes, user may omit tcp_port or http_port or https_port in configuration file.
            try
            {
                /// HTTPS
                if (config.has("https_port"))
                {
#if Poco_NetSSL_FOUND
                    if (!security_config.has_tls_config)
                    {
                        LOG_FMT_ERROR(log, "https_port is set but tls config is not set");
                    }
                    Poco::Net::Context::Ptr context = new Poco::Net::Context(Poco::Net::Context::TLSV1_2_SERVER_USE,
                                                                             security_config.key_path,
                                                                             security_config.cert_path,
                                                                             security_config.ca_path,
                                                                             Poco::Net::Context::VerificationMode::VERIFY_STRICT);
                    std::function<bool(const Poco::Crypto::X509Certificate &)> check_common_name
                        = [&](const Poco::Crypto::X509Certificate & cert) {
                              if (security_config.allowed_common_names.empty())
                              {
                                  return true;
                              }
                              return security_config.allowed_common_names.count(cert.commonName()) > 0;
                          };
                    context->setAdhocVerification(check_common_name);
                    std::call_once(ssl_init_once, SSLInit);

                    Poco::Net::SecureServerSocket socket(context);
                    auto address = socket_bind_listen(socket, listen_host, config.getInt("https_port"), /* secure = */ true);
                    socket.setReceiveTimeout(settings.http_receive_timeout);
                    socket.setSendTimeout(settings.http_send_timeout);
                    servers.emplace_back(
                        new HTTPServer(new HTTPHandlerFactory(server, "HTTPSHandler-factory"), server_pool, socket, http_params));

                    LOG_FMT_INFO(log, "Listening https://{}", address.toString());
#else
                    throw Exception{"HTTPS protocol is disabled because Poco library was built without NetSSL support.",
                                    ErrorCodes::SUPPORT_IS_DISABLED};
#endif
                }
                else
                {
                    /// HTTP
                    if (security_config.has_tls_config)
                    {
                        throw Exception("tls config is set but https_port is not set ", ErrorCodes::INVALID_CONFIG_PARAMETER);
                    }
                    Poco::Net::ServerSocket socket;
                    auto address = socket_bind_listen(socket, listen_host, config.getInt("http_port", DEFAULT_HTTP_PORT));
                    socket.setReceiveTimeout(settings.http_receive_timeout);
                    socket.setSendTimeout(settings.http_send_timeout);
                    servers.emplace_back(
                        new HTTPServer(new HTTPHandlerFactory(server, "HTTPHandler-factory"), server_pool, socket, http_params));

                    LOG_FMT_INFO(log, "Listening http://{}", address.toString());
                }

                /// TCP
                if (config.has("tcp_port"))
                {
                    if (security_config.has_tls_config)
                    {
                        LOG_FMT_ERROR(log, "tls config is set but tcp_port_secure is not set.");
                    }
                    std::call_once(ssl_init_once, SSLInit);
                    Poco::Net::ServerSocket socket;
                    auto address = socket_bind_listen(socket, listen_host, config.getInt("tcp_port"));
                    socket.setReceiveTimeout(settings.receive_timeout);
                    socket.setSendTimeout(settings.send_timeout);
                    servers.emplace_back(new TCPServer(new TCPHandlerFactory(server), server_pool, socket, new Poco::Net::TCPServerParams));

                    LOG_FMT_INFO(log, "Listening tcp: {}", address.toString());
                }
                else if (security_config.has_tls_config)
                {
                    LOG_FMT_INFO(log, "tcp_port is closed because tls config is set");
                }

                /// TCP with SSL
                if (config.has("tcp_port_secure") && !security_config.has_tls_config)
                {
#if Poco_NetSSL_FOUND
                    Poco::Net::Context::Ptr context = new Poco::Net::Context(Poco::Net::Context::TLSV1_2_SERVER_USE,
                                                                             security_config.key_path,
                                                                             security_config.cert_path,
                                                                             security_config.ca_path);
                    Poco::Net::SecureServerSocket socket(context);
                    auto address = socket_bind_listen(socket, listen_host, config.getInt("tcp_port_secure"), /* secure = */ true);
                    socket.setReceiveTimeout(settings.receive_timeout);
                    socket.setSendTimeout(settings.send_timeout);
                    servers.emplace_back(new TCPServer(
                        new TCPHandlerFactory(server, /* secure= */ true),
                        server_pool,
                        socket,
                        new Poco::Net::TCPServerParams));
                    LOG_FMT_INFO(log, "Listening tcp_secure: {}", address.toString());
#else
                    throw Exception{"SSL support for TCP protocol is disabled because Poco library was built without NetSSL support.",
                                    ErrorCodes::SUPPORT_IS_DISABLED};
#endif
                }
                else if (security_config.has_tls_config)
                {
                    LOG_FMT_INFO(log, "tcp_port is closed because tls config is set");
                }

                /// At least one of TCP and HTTP servers must be created.
                if (servers.empty())
                    throw Exception("No 'tcp_port' and 'http_port' is specified in configuration file.", ErrorCodes::NO_ELEMENTS_IN_CONFIG);
            }
            catch (const Poco::Net::NetException & e)
            {
                if (listen_try)
                    LOG_FMT_ERROR(
                        log,
                        "Listen [{}]: {}: {}: {}"
                        "  If it is an IPv6 or IPv4 address and your host has disabled IPv6 or IPv4, then consider to "
                        "specify not disabled IPv4 or IPv6 address to listen in <listen_host> element of configuration "
                        "file. Example for disabled IPv6: <listen_host>0.0.0.0</listen_host> ."
                        " Example for disabled IPv4: <listen_host>::</listen_host>",
                        listen_host,
                        e.code(),
                        e.what(),
                        e.message());
                else
                    throw;
            }
        }

        if (servers.empty())
            throw Exception("No servers started (add valid listen_host and 'tcp_port' or 'http_port' to configuration file.)",
                            ErrorCodes::NO_ELEMENTS_IN_CONFIG);

        for (auto & server : servers)
            server->start();
    }

    void onExit()
    {
        auto & config = server.config();

        LOG_FMT_DEBUG(log, "Received termination signal.");
        LOG_FMT_DEBUG(log, "Waiting for current connections to close.");

        int current_connections = 0;
        for (auto & server : servers)
        {
            server->stop();
            current_connections += server->currentConnections();
        }
        String debug_msg = "Closed all listening sockets.";

        if (current_connections)
            LOG_FMT_DEBUG(
                log,
                "{} Waiting for {} outstanding connections.",
                debug_msg,
                current_connections);
        else
            LOG_DEBUG(log, debug_msg);

        if (current_connections)
        {
            const int sleep_max_ms = 1000 * config.getInt("shutdown_wait_unfinished", 5);
            const int sleep_one_ms = 100;
            int sleep_current_ms = 0;
            while (sleep_current_ms < sleep_max_ms)
            {
                current_connections = 0;
                for (auto & server : servers)
                    current_connections += server->currentConnections();
                if (!current_connections)
                    break;
                sleep_current_ms += sleep_one_ms;
                std::this_thread::sleep_for(std::chrono::milliseconds(sleep_one_ms));
            }
        }

        debug_msg = "Closed connections.";

        if (current_connections)
            LOG_FMT_DEBUG(
                log,
                "{} But {} remains."
                " Tip: To increase wait time add to config: <shutdown_wait_unfinished>60</shutdown_wait_unfinished>",
                debug_msg,
                current_connections);
        else
            LOG_DEBUG(log, debug_msg);
    }

    const std::vector<std::unique_ptr<Poco::Net::TCPServer>> & getServers() const { return servers; }

private:
    Server & server;
    Poco::Logger * log;
    Poco::ThreadPool server_pool;
    std::vector<std::unique_ptr<Poco::Net::TCPServer>> servers;
};

int Server::main(const std::vector<std::string> & /*args*/)
{
    setThreadName("TiFlashMain");

    Poco::Logger * log = &logger();
#ifdef FIU_ENABLE
    fiu_init(0); // init failpoint
#endif

    UpdateMallocConfig(log);

#ifdef TIFLASH_ENABLE_AVX_SUPPORT
    tryLoadBoolConfigFromEnv(log, simd_option::ENABLE_AVX, "TIFLASH_ENABLE_AVX");
#endif

#ifdef TIFLASH_ENABLE_AVX512_SUPPORT
    tryLoadBoolConfigFromEnv(log, simd_option::ENABLE_AVX512, "TIFLASH_ENABLE_AVX512");
#endif

#ifdef TIFLASH_ENABLE_ASIMD_SUPPORT
    tryLoadBoolConfigFromEnv(log, simd_option::ENABLE_ASIMD, "TIFLASH_ENABLE_ASIMD");
#endif

#ifdef TIFLASH_ENABLE_SVE_SUPPORT
    tryLoadBoolConfigFromEnv(log, simd_option::ENABLE_SVE, "TIFLASH_ENABLE_SVE");
#endif

    registerFunctions();
    registerAggregateFunctions();
    registerWindowFunctions();
    registerTableFunctions();
    registerStorages();

    TiFlashErrorRegistry::instance(); // This invocation is for initializing

    TiFlashProxyConfig proxy_conf(config());
    EngineStoreServerWrap tiflash_instance_wrap{};
    auto helper = GetEngineStoreServerHelper(
        &tiflash_instance_wrap);

    RaftStoreProxyRunner proxy_runner(RaftStoreProxyRunner::RunRaftStoreProxyParms{&helper, proxy_conf}, log);

    proxy_runner.run();

    if (proxy_conf.is_proxy_runnable)
    {
        LOG_FMT_INFO(log, "wait for tiflash proxy initializing");
        while (!tiflash_instance_wrap.proxy_helper)
            std::this_thread::sleep_for(std::chrono::milliseconds(200));
        LOG_FMT_INFO(log, "tiflash proxy is initialized");
        if (tiflash_instance_wrap.proxy_helper->checkEncryptionEnabled())
        {
            auto method = tiflash_instance_wrap.proxy_helper->getEncryptionMethod();
            LOG_FMT_INFO(log, "encryption is enabled, method is {}", IntoEncryptionMethodName(method));
        }
        else
            LOG_FMT_INFO(log, "encryption is disabled");
    }

    SCOPE_EXIT({
        if (!proxy_conf.is_proxy_runnable)
        {
            proxy_runner.join();
            return;
        }
        LOG_FMT_INFO(log, "Let tiflash proxy shutdown");
        tiflash_instance_wrap.status = EngineStoreServerStatus::Terminated;
        tiflash_instance_wrap.tmt = nullptr;
        LOG_FMT_INFO(log, "Wait for tiflash proxy thread to join");
        proxy_runner.join();
        LOG_FMT_INFO(log, "tiflash proxy thread is joined");
    });

    CurrentMetrics::set(CurrentMetrics::Revision, ClickHouseRevision::get());

    // print necessary grpc log.
    grpc_log = &Poco::Logger::get("grpc");
    gpr_set_log_verbosity(GPR_LOG_SEVERITY_DEBUG);
    gpr_set_log_function(&printGRPCLog);

    /** Context contains all that query execution is dependent:
      *  settings, available functions, data types, aggregate functions, databases...
      */
    global_context = std::make_unique<Context>(Context::createGlobal());
    global_context->setGlobalContext(*global_context);
    global_context->setApplicationType(Context::ApplicationType::SERVER);

    /// Init File Provider
    if (proxy_conf.is_proxy_runnable)
    {
        bool enable_encryption = tiflash_instance_wrap.proxy_helper->checkEncryptionEnabled();
        if (enable_encryption)
        {
            auto method = tiflash_instance_wrap.proxy_helper->getEncryptionMethod();
            enable_encryption = (method != EncryptionMethod::Plaintext);
        }
        KeyManagerPtr key_manager = std::make_shared<DataKeyManager>(&tiflash_instance_wrap);
        global_context->initializeFileProvider(key_manager, enable_encryption);
    }
    else
    {
        KeyManagerPtr key_manager = std::make_shared<MockKeyManager>(false);
        global_context->initializeFileProvider(key_manager, false);
    }

    /// ===== Paths related configuration initialized start ===== ///
    /// Note that theses global variables should be initialized by the following order:
    // 1. capacity
    // 2. path pool
    // 3. TMTContext

    // Deprecated settings.
    // `global_capacity_quota` will be ignored if `storage_config.main_capacity_quota` is not empty.
    // "0" by default, means no quota, the actual disk capacity is used.
    size_t global_capacity_quota = 0;
    TiFlashStorageConfig storage_config;
    std::tie(global_capacity_quota, storage_config) = TiFlashStorageConfig::parseSettings(config(), log);

    if (storage_config.format_version)
    {
        setStorageFormat(storage_config.format_version);
        LOG_FMT_INFO(log, "Using format_version={} (explicit stable storage format detected).", storage_config.format_version);
    }
    else
    {
        LOG_FMT_INFO(log, "Using format_version={} (default settings).", STORAGE_FORMAT_CURRENT.identifier);
    }

    global_context->initializePathCapacityMetric( //
        global_capacity_quota, //
        storage_config.main_data_paths,
        storage_config.main_capacity_quota, //
        storage_config.latest_data_paths,
        storage_config.latest_capacity_quota);
    TiFlashRaftConfig raft_config = TiFlashRaftConfig::parseSettings(config(), log);
    global_context->setPathPool( //
        storage_config.main_data_paths, //
        storage_config.latest_data_paths, //
        storage_config.kvstore_data_path, //
        raft_config.enable_compatible_mode, //
        global_context->getPathCapacity(),
        global_context->getFileProvider());

    global_context->initializePageStorageMode(global_context->getPathPool(), STORAGE_FORMAT_CURRENT.page);
    global_context->initializeGlobalStoragePoolIfNeed(global_context->getPathPool());
    LOG_FMT_INFO(log, "Global PageStorage run mode is {}", static_cast<UInt8>(global_context->getPageStorageRunMode()));

    // Use pd address to define which default_database we use by default.
    // For mock test, we use "default". For deployed with pd/tidb/tikv use "system", which is always exist in TiFlash.
    std::string default_database = config().getString("default_database", raft_config.pd_addrs.empty() ? "default" : "system");
    Strings all_normal_path = storage_config.getAllNormalPaths();
    const std::string path = all_normal_path[0];
    global_context->setPath(path);

    /// ===== Paths related configuration initialized end ===== ///

    security_config = TiFlashSecurityConfig(config(), log);
    Redact::setRedactLog(security_config.redact_info_log);

    // Create directories for 'path' and for default database, if not exist.
    for (const String & candidate_path : all_normal_path)
    {
        Poco::File(candidate_path + "data/" + default_database).createDirectories();
    }
    Poco::File(path + "metadata/" + default_database).createDirectories();

    StatusFile status{path + "status"};

    SCOPE_EXIT({
        /** Explicitly destroy Context. It is more convenient than in destructor of Server, because logger is still available.
          * At this moment, no one could own shared part of Context.
          */
        global_context.reset();

        LOG_FMT_DEBUG(log, "Destroyed global context.");
    });

    /// Try to increase limit on number of open files.
    {
        rlimit rlim{};
        if (getrlimit(RLIMIT_NOFILE, &rlim))
            throw Poco::Exception("Cannot getrlimit");

        if (rlim.rlim_cur == rlim.rlim_max)
        {
            LOG_FMT_DEBUG(log, "rlimit on number of file descriptors is {}", rlim.rlim_cur);
        }
        else
        {
            rlim_t old = rlim.rlim_cur;
            rlim.rlim_cur = config().getUInt("max_open_files", rlim.rlim_max);
            int rc = setrlimit(RLIMIT_NOFILE, &rlim);
            if (rc != 0)
                LOG_FMT_WARNING(
                    log,
                    "Cannot set max number of file descriptors to {}"
                    ". Try to specify max_open_files according to your system limits. error: {}",
                    rlim.rlim_cur,
                    strerror(errno));
            else
                LOG_FMT_DEBUG(log, "Set max number of file descriptors to {} (was {}).", rlim.rlim_cur, old);
        }
    }

    static ServerErrorHandler error_handler;
    Poco::ErrorHandler::set(&error_handler);

    /// Initialize DateLUT early, to not interfere with running time of first query.
    LOG_FMT_DEBUG(log, "Initializing DateLUT.");
    DateLUT::instance();
    LOG_FMT_TRACE(log, "Initialized DateLUT with time zone `{}`.", DateLUT::instance().getTimeZone());

    /// Directory with temporary data for processing of heavy queries.
    {
        std::string tmp_path = config().getString("tmp_path", path + "tmp/");
        global_context->setTemporaryPath(tmp_path);
        Poco::File(tmp_path).createDirectories();

        /// Clearing old temporary files.
        Poco::DirectoryIterator dir_end;
        for (Poco::DirectoryIterator it(tmp_path); it != dir_end; ++it)
        {
            if (it->isFile() && startsWith(it.name(), "tmp"))
            {
                LOG_FMT_DEBUG(log, "Removing old temporary file {}", it->path());
                global_context->getFileProvider()->deleteRegularFile(it->path(), EncryptionPath(it->path(), ""));
            }
        }
    }

    /** Directory with 'flags': files indicating temporary settings for the server set by system administrator.
      * Flags may be cleared automatically after being applied by the server.
      * Examples: do repair of local data; clone all replicated tables from replica.
      */
    {
        Poco::File(path + "flags/").createDirectories();
        global_context->setFlagsPath(path + "flags/");
    }

    /** Directory with user provided files that are usable by 'file' table function.
      */
    {
        std::string user_files_path = config().getString("user_files_path", path + "user_files/");
        global_context->setUserFilesPath(user_files_path);
        Poco::File(user_files_path).createDirectories();
    }

    if (config().has("macros"))
        global_context->setMacros(std::make_unique<Macros>(config(), "macros"));

    /// Init TiFlash metrics.
    global_context->initializeTiFlashMetrics();

    /// Initialize users config reloader.
    auto users_config_reloader = UserConfig::parseSettings(config(), config_path, global_context, log);

    /// Load global settings from default_profile and system_profile.
    /// It internally depends on UserConfig::parseSettings.
    global_context->setDefaultProfiles(config());
    Settings & settings = global_context->getSettingsRef();

    /// Initialize the background thread pool.
    /// It internally depends on settings.background_pool_size,
    /// so must be called after settings has been load.
    auto & bg_pool = global_context->getBackgroundPool();
    auto & blockable_bg_pool = global_context->getBlockableBackgroundPool();

    /// Initialize RateLimiter.
    global_context->initializeRateLimiter(config(), bg_pool, blockable_bg_pool);

    /// Initialize main config reloader.
    auto main_config_reloader = std::make_unique<ConfigReloader>(
        config_path,
        [&](ConfigurationPtr config) {
            buildLoggers(*config);
            global_context->setMacros(std::make_unique<Macros>(*config, "macros"));
            global_context->getTMTContext().reloadConfig(*config);
            global_context->getIORateLimiter().updateConfig(*config);
            global_context->reloadDeltaTreeConfig(*config);
        },
        /* already_loaded = */ true);

    /// Reload config in SYSTEM RELOAD CONFIG query.
    global_context->setConfigReloadCallback([&]() {
        main_config_reloader->reload();
        users_config_reloader->reload();
    });

    /// Limit on total number of concurrently executed queries.
    global_context->getProcessList().setMaxSize(config().getInt("max_concurrent_queries", 0));

    /// Setup protection to avoid accidental DROP for big tables (that are greater than 50 GB by default)
    if (config().has("max_table_size_to_drop"))
        global_context->setMaxTableSizeToDrop(config().getUInt64("max_table_size_to_drop"));

    /// Size of cache for uncompressed blocks. Zero means disabled.
    size_t uncompressed_cache_size = config().getUInt64("uncompressed_cache_size", 0);
    if (uncompressed_cache_size)
        global_context->setUncompressedCache(uncompressed_cache_size);

    bool use_l0_opt = config().getBool("l0_optimize", false);
    global_context->setUseL0Opt(use_l0_opt);

    /// Size of cache for marks (index of MergeTree family of tables). It is necessary.
    size_t mark_cache_size = config().getUInt64("mark_cache_size", DEFAULT_MARK_CACHE_SIZE);
    if (mark_cache_size)
        global_context->setMarkCache(mark_cache_size);

    /// Size of cache for minmax index, used by DeltaMerge engine.
    size_t minmax_index_cache_size = config().getUInt64("minmax_index_cache_size", mark_cache_size);
    if (minmax_index_cache_size)
        global_context->setMinMaxIndexCache(minmax_index_cache_size);

    /// Size of max memory usage of DeltaIndex, used by DeltaMerge engine.
    size_t delta_index_cache_size = config().getUInt64("delta_index_cache_size", 0);
    global_context->setDeltaIndexManager(delta_index_cache_size);

    /// Set path for format schema files
    auto format_schema_path = Poco::File(config().getString("format_schema_path", path + "format_schemas/"));
    global_context->setFormatSchemaPath(format_schema_path.path() + "/");
    format_schema_path.createDirectories();

    LOG_FMT_INFO(log, "Loading metadata.");
    loadMetadataSystem(*global_context); // Load "system" database. Its engine keeps as Ordinary.
    /// After attaching system databases we can initialize system log.
    global_context->initializeSystemLogs();
    /// After the system database is created, attach virtual system tables (in addition to query_log and part_log)
    attachSystemTablesServer(*global_context->getDatabase("system"));

    {
        /// create TMTContext
        auto cluster_config = getClusterConfig(security_config, raft_config);
        global_context->createTMTContext(raft_config, std::move(cluster_config));
        global_context->getTMTContext().reloadConfig(config());
    }

    {
        // Note that this must do before initialize schema sync service.
        do
        {
            // Check whether we need to upgrade directories hierarchy
            // If some database can not find in TiDB, they will be dropped
            // if theirs name is not in reserved_databases.
            // Besides, database engine in reserved_databases just keep as
            // what they are.
            IDAsPathUpgrader upgrader(
                *global_context,
                /*is_mock=*/raft_config.pd_addrs.empty(),
                /*reserved_databases=*/raft_config.ignore_databases);
            if (!upgrader.needUpgrade())
                break;
            upgrader.doUpgrade();
        } while (false);

        /// Then, load remaining databases
        loadMetadata(*global_context);
    }
    LOG_FMT_DEBUG(log, "Load metadata done.");

    /// Then, sync schemas with TiDB, and initialize schema sync service.
    for (int i = 0; i < 60; i++) // retry for 3 mins
    {
        try
        {
            global_context->getTMTContext().getSchemaSyncer()->syncSchemas(*global_context);
            break;
        }
        catch (Poco::Exception & e)
        {
            const int wait_seconds = 3;
            LOG_FMT_ERROR(
                log,
                "Bootstrap failed because sync schema error: {}\nWe will sleep for {}"
                " seconds and try again.",
                e.displayText(),
                wait_seconds);
            ::sleep(wait_seconds);
        }
    }
    LOG_FMT_DEBUG(log, "Sync schemas done.");

    initStores(*global_context, log, storage_config.lazily_init_store);

    // After schema synced, set current database.
    global_context->setCurrentDatabase(default_database);

    global_context->initializeSchemaSyncService();
    CPUAffinityManager::initCPUAffinityManager(config());
    LOG_FMT_INFO(log, "CPUAffinity: {}", CPUAffinityManager::getInstance().toString());
    SCOPE_EXIT({
        /** Ask to cancel background jobs all table engines,
          *  and also query_log.
          * It is important to do early, not in destructor of Context, because
          *  table engines could use Context on destroy.
          */
        LOG_FMT_INFO(log, "Shutting down storages.");
        global_context->shutdown();
        LOG_FMT_DEBUG(log, "Shutted down storages.");
    });

    {
        if (proxy_conf.is_proxy_runnable && !tiflash_instance_wrap.proxy_helper)
            throw Exception("Raft Proxy Helper is not set, should not happen");
        /// initialize TMTContext
        global_context->getTMTContext().restore(tiflash_instance_wrap.proxy_helper);
    }

    /// setting up elastic thread pool
    if (settings.enable_elastic_threadpool)
        DynamicThreadPool::global_instance = std::make_unique<DynamicThreadPool>(
            settings.elastic_threadpool_init_cap,
            std::chrono::milliseconds(settings.elastic_threadpool_shrink_period_ms));

    if (settings.enable_async_grpc_client)
    {
        auto size = settings.grpc_completion_queue_pool_size;
        if (size == 0)
            size = std::thread::hardware_concurrency();
        GRPCCompletionQueuePool::global_instance = std::make_unique<GRPCCompletionQueuePool>(size);
    }

    /// Then, startup grpc server to serve raft and/or flash services.
    FlashGrpcServerHolder flash_grpc_server_holder(*this, raft_config, log);

    {
        TcpHttpServersHolder tcpHttpServersHolder(*this, settings, log);

        main_config_reloader->start();
        users_config_reloader->start();

        {
            // on ARM processors it can show only enabled at current moment cores
            LOG_FMT_INFO(
                log,
                "Available RAM = {}; physical cores = {}; threads = {}.",
                formatReadableSizeWithBinarySuffix(getMemoryAmount()),
                getNumberOfPhysicalCPUCores(),
                std::thread::hardware_concurrency());
        }

        LOG_FMT_INFO(log, "Ready for connections.");

        SCOPE_EXIT({
            is_cancelled = true;

            tcpHttpServersHolder.onExit();

            main_config_reloader.reset();
            users_config_reloader.reset();
        });

        /// try to load dictionaries immediately, throw on error and die
        try
        {
            if (!config().getBool("dictionaries_lazy_load", true))
            {
                global_context->tryCreateEmbeddedDictionaries();
                global_context->tryCreateExternalDictionaries();
            }
        }
        catch (...)
        {
            LOG_FMT_ERROR(log, "Caught exception while loading dictionaries.");
            throw;
        }

        /// This object will periodically calculate some metrics.
        /// should init after `createTMTContext` cause we collect some data from the TiFlash context object.
        AsynchronousMetrics async_metrics(*global_context);
        attachSystemTablesAsync(*global_context->getDatabase("system"), async_metrics);

        std::vector<std::unique_ptr<MetricsTransmitter>> metrics_transmitters;
        for (const auto & graphite_key : DB::getMultipleKeysFromConfig(config(), "", "graphite"))
        {
            metrics_transmitters.emplace_back(std::make_unique<MetricsTransmitter>(*global_context, async_metrics, graphite_key));
        }

        auto metrics_prometheus = std::make_unique<MetricsPrometheus>(*global_context, async_metrics, security_config);

        SessionCleaner session_cleaner(*global_context);

        auto & tmt_context = global_context->getTMTContext();
        if (proxy_conf.is_proxy_runnable)
        {
            tiflash_instance_wrap.tmt = &tmt_context;
            LOG_FMT_INFO(log, "Let tiflash proxy start all services");
            tiflash_instance_wrap.status = EngineStoreServerStatus::Running;
            while (tiflash_instance_wrap.proxy_helper->getProxyStatus() == RaftProxyStatus::Idle)
                std::this_thread::sleep_for(std::chrono::milliseconds(200));

            // proxy update store-id before status set `RaftProxyStatus::Running`
            assert(tiflash_instance_wrap.proxy_helper->getProxyStatus() == RaftProxyStatus::Running);
            LOG_FMT_INFO(log, "store {}, tiflash proxy is ready to serve, try to wake up all regions' leader", tmt_context.getKVStore()->getStoreID(std::memory_order_seq_cst));
            size_t runner_cnt = config().getUInt("flash.read_index_runner_count", 1); // if set 0, DO NOT enable read-index worker
            tmt_context.getKVStore()->initReadIndexWorkers(
                [&]() {
                    // get from tmt context
                    return std::chrono::milliseconds(tmt_context.readIndexWorkerTick());
                },
                /*running thread count*/ runner_cnt);
            tmt_context.getKVStore()->asyncRunReadIndexWorkers();
            WaitCheckRegionReady(tmt_context, terminate_signals_counter);
        }
        SCOPE_EXIT({
            if (proxy_conf.is_proxy_runnable && tiflash_instance_wrap.status != EngineStoreServerStatus::Running)
            {
                LOG_FMT_ERROR(log, "Current status of engine-store is NOT Running, should not happen");
                exit(-1);
            }
            LOG_FMT_INFO(log, "Set store context status Stopping");
            tmt_context.setStatusStopping();
            {
                // Wait until there is no read-index task.
                while (tmt_context.getKVStore()->getReadIndexEvent())
                    std::this_thread::sleep_for(std::chrono::milliseconds(200));
            }
            tmt_context.setStatusTerminated();
            tmt_context.getKVStore()->stopReadIndexWorkers();
            LOG_FMT_INFO(log, "Set store context status Terminated");
            {
                // update status and let proxy stop all services except encryption.
                tiflash_instance_wrap.status = EngineStoreServerStatus::Stopping;
                LOG_FMT_INFO(log, "Set engine store server status Stopping");
            }
            // wait proxy to stop services
            if (proxy_conf.is_proxy_runnable)
            {
                LOG_FMT_INFO(log, "Let tiflash proxy to stop all services");
                while (tiflash_instance_wrap.proxy_helper->getProxyStatus() != RaftProxyStatus::Stopped)
                    std::this_thread::sleep_for(std::chrono::milliseconds(200));
                LOG_FMT_INFO(log, "All services in tiflash proxy are stopped");
            }
        });

        {
            // Report the unix timestamp, git hash, release version
            Poco::Timestamp ts;
            GET_METRIC(tiflash_server_info, start_time).Set(ts.epochTime());
        }

        tmt_context.setStatusRunning();

        try
        {
            // Bind CPU affinity after all threads started.
            CPUAffinityManager::getInstance().bindThreadCPUAffinity();
        }
        catch (...)
        {
            LOG_FMT_ERROR(log, "CPUAffinityManager::bindThreadCPUAffinity throws exception.");
        }

        LOG_FMT_INFO(log, "Start to wait for terminal signal");
        waitForTerminationRequest();

        {
            // Set limiters stopping and wakeup threads in waitting queue.
            global_context->getIORateLimiter().setStop();
        }
    }

    return Application::EXIT_OK;
}
} // namespace DB

int mainEntryClickHouseServer(int argc, char ** argv)
{
    DB::Server app;
    try
    {
        return app.run(argc, argv);
    }
    catch (...)
    {
        std::cerr << DB::getCurrentExceptionMessage(true) << "\n";
        auto code = DB::getCurrentExceptionCode();
        return code ? code : 1;
    }
}

max_stars_repo_name: pingcap/tiflash

max_issues_repo_name: pingcap/tiflash

max_forks_repo_name: pingcap/tiflash


text:
//
// Created by 范炜东 on 2019/4/23.
//

#include <iostream>
#include <exception>
#include <boost/program_options.hpp>
#include "Poco/Net/SocketAddress.h"
#include "Poco/Net/StreamSocket.h"

using Poco::Net::SocketAddress;
using Poco::Net::StreamSocket;

int main(int argc, char *argv[]) {
  boost::program_options::options_description cmdline_options("命令行参数");
  cmdline_options.add_options()
          ("port", boost::program_options::value<int>()->default_value(2000), "服务器端口");

  try {
    boost::program_options::variables_map vm;
    boost::program_options::store(boost::program_options::parse_command_line(argc, argv, cmdline_options), vm);
    boost::program_options::notify(vm);

    SocketAddress sa("127.0.0.1", vm["port"].as<int>());
    StreamSocket socket(sa);
    char buffer[64] = {};
    int n = socket.receiveBytes(buffer, sizeof(buffer));
    if (n > 0)
      std::cout << buffer << std::endl;
  } catch (std::exception &e) {
    std::cerr << "Error: " << e.what() << std::endl;
  }

  return 0;
}

ext: cc

max_stars_repo_name: silvermagic/Practice

max_issues_repo_name: silvermagic/Practice

max_forks_repo_name: silvermagic/Practice


text:
#include <boost/circular_buffer.hpp>
#include <iostream>

using namespace std;

int main() {
  typedef boost::circular_buffer<int> circular_buffer;
  circular_buffer cb{3};

  cout << "capacity = " << cb.capacity() << endl;
  cout << "size = " << cb.size() << endl;

  cb.push_back(0);
  cb.push_back(1);
  cb.push_back(2);

  cout << "size = " << cb.size() << endl;

  for (int i: cb)
    cout << i << " ";
  cout << endl;

  cb.push_back(3);
  cb.push_back(4);
  cb.push_back(5);

  cout << "size = " << cb.size() << endl;

  for (int i: cb)
    cout << i << " ";
  cout << endl;
}

ext: cpp


text:
/*************************************************************************
 * odil - Copyright (C) Universite de Strasbourg
 * Distributed under the terms of the CeCILL-B license, as published by
 * the CEA-CNRS-INRIA. Refer to the LICENSE file or to
 * http://www.cecill.info/licences/Licence_CeCILL-B_V1-en.html
 * for details.
 ************************************************************************/

#include "odil/BasicDirectoryCreator.h"

#include <algorithm>
#include <fstream>
#include <iterator>
#include <map>
#include <memory>
#include <sstream>
#include <stack>
#include <string>
#include <utility>
#include <vector>

#include <boost/filesystem.hpp>
#include <boost/filesystem/fstream.hpp>

#include "odil/DataSet.h"
#include "odil/Exception.h"
#include "odil/Reader.h"
#include "odil/registry.h"
#include "odil/Tag.h"
#include "odil/uid.h"

namespace odil
{

std::map<std::string, std::vector<BasicDirectoryCreator::RecordKey>> const
BasicDirectoryCreator
::default_record_keys = {
    { "PATIENT", {
        { registry::PatientName, 2},
        { registry::PatientID, 1 } } },
    { "STUDY", {
        { registry::StudyDate, 1 },
        { registry::StudyTime, 1 },
        { registry::StudyDescription, 2 },
        { registry::StudyInstanceUID, 1 }, // only if (0004,1511) is absent (see Note)
        { registry::StudyID, 1 },
        { registry::AccessionNumber, 2 } } },
    { "SERIES", {
        { registry::Modality, 1 },
        { registry::SeriesInstanceUID, 1 },
        { registry::SeriesNumber, 1 },
        //registry::IconImageSequence
        } },
    { "IMAGE", {
        { registry::InstanceNumber, 1 },
        //registry::IconImageSequence
        } },
};

BasicDirectoryCreator
::BasicDirectoryCreator(
    std::string const & root, std::vector<std::string> const & files,
    RecordKeyMap const & extra_record_keys, Writer::ItemEncoding item_encoding)
: root(root), files(files), extra_record_keys(extra_record_keys),
    item_encoding(item_encoding)
{
    // Nothing else.
}

void
BasicDirectoryCreator
::operator()() const
{
    auto const record_tree = this->_create_records();
    auto linearized_tree = this->_linearize_records(record_tree);
    auto const relative_offsets = this->_get_relative_offsets(
        linearized_tree.records);
    this->_write(linearized_tree, relative_offsets);
}

BasicDirectoryCreator::RecordMap
BasicDirectoryCreator
::_create_records() const
{
    RecordMap patients;

    for(auto const & file: files)
    {
        auto const absolute_path = boost::filesystem::path(this->root)/file;
        if(!boost::filesystem::is_regular_file(absolute_path))
        {
            throw Exception("No such file: "+absolute_path.string());
        }

        boost::filesystem::ifstream stream(
            absolute_path, std::ifstream::in | std::ifstream::binary);
        auto const meta_info_and_data_set = Reader::read_file(stream);
        auto const & data_set = meta_info_and_data_set.second;

        auto const & patient_id = data_set.as_string(
            registry::PatientID)[0];
        auto & patient = this->_find_record(patients, patient_id);
        if(patient.data_set.empty())
        {
            this->_fill_record(data_set, patient, "PATIENT");
        }

        auto const & study_instance_uid = data_set.as_string(
            registry::StudyInstanceUID)[0];
        auto & study = this->_find_record(patient.children, study_instance_uid);
        if(study.data_set.empty())
        {
            this->_fill_record(data_set, study, "STUDY");
        }

        auto const & series_instance_uid = data_set.as_string(
            registry::SeriesInstanceUID)[0];
        auto & series = this->_find_record(study.children, series_instance_uid);
        if(series.data_set.empty())
        {
            this->_fill_record(data_set, series, "SERIES");
        }

        auto const & sop_instance_uid = data_set.as_string(
            registry::SOPInstanceUID)[0];
        auto & image = this->_find_record(series.children, sop_instance_uid);
        if(image.data_set.empty())
        {
            this->_fill_record(data_set, image, "IMAGE");

            Value::Strings file_id;
            int begin = 0;
            while(begin != std::string::npos)
            {
                auto const end = file.find('/', begin);
                auto const size =
                    (end==std::string::npos)?std::string::npos:(end-begin);
                std::string const item = file.substr(begin, size);
                file_id.push_back(item);
                if(end == std::string::npos)
                {
                    begin = end;
                }
                else
                {
                    begin = end+1;
                }
            }

            // Required if the Directory Record references a SOP Instance (PS3.3 F.3.2.2)
            image.data_set.add(registry::ReferencedFileID, file_id);
            image.data_set.add(
                registry::ReferencedSOPClassUIDInFile,
                data_set[registry::SOPClassUID]);
            image.data_set.add(
                registry::ReferencedSOPInstanceUIDInFile,
                data_set[registry::SOPInstanceUID]);

            auto const & meta_info = meta_info_and_data_set.first;
            image.data_set.add(
                registry::ReferencedTransferSyntaxUIDInFile,
                meta_info[registry::TransferSyntaxUID]);
        }
    }

    return patients;
}

BasicDirectoryCreator::Record &
BasicDirectoryCreator
::_find_record(RecordMap & records, std::string const & key) const
{
    auto iterator = records.find(key);
    if(iterator == records.end())
    {
        iterator = records.insert({key, Record::Pointer(new Record())}).first;
    }
    return *(iterator->second);
}

void
BasicDirectoryCreator
::_fill_record(
    DataSet const & data_set, Record & record, std::string const & type) const
{
    record.data_set.add(registry::DirectoryRecordType, {type});
    record.data_set.add(registry::OffsetOfTheNextDirectoryRecord, {0});
    record.data_set.add(registry::RecordInUseFlag, {0xffff});
    record.data_set.add(
        registry::OffsetOfReferencedLowerLevelDirectoryEntity, {0});

    this->_update_record(data_set, record, this->default_record_keys.at(type));

    auto const extra_record_keys_it = this->extra_record_keys.find(type);
    if(extra_record_keys_it != this->extra_record_keys.end())
    {
        this->_update_record(data_set, record, extra_record_keys_it->second);
    }
}

void
BasicDirectoryCreator
::_update_record(
    DataSet const & data_set, Record & record,
    std::vector<RecordKey> const & keys) const
{
    for(auto const & key: keys)
    {
        auto const & tag = key.first;
        auto const & type = key.second;

        if(data_set.has(tag))
        {
            record.data_set.add(tag, data_set[tag]);
        }
        else
        {
            if(type == 1)
            {
                throw Exception("Type 1 tag missing from data set: "+std::string(tag));
            }
            else if(type == 2)
            {
                record.data_set.add(tag);
            }
            // Otherwise do nothing.
        }
    }

    // TODO: registry::RecordInUseFlag;
}

BasicDirectoryCreator::LinearizedTree
BasicDirectoryCreator
::_linearize_records(RecordMap const & record_tree) const
{
    LinearizedTree linearized_tree;

    // Create a depth-first list of records and a sibling map.
    // Prime the stack and sibling map with all top-level records.
    std::stack<Record::Pointer> stack;
    std::map<Record const *, long> location_map;
    std::map<Record const *, Record const *> sibling_map;
    for(auto it=record_tree.begin(); it!=record_tree.end(); ++it)
    {
        stack.push(it->second);
        auto const sibling_it = std::next(it);
        if(sibling_it != record_tree.end())
        {
            auto const & sibling = sibling_it->second;
            sibling_map[it->second.get()] = sibling.get();
        }
    }

    // Create a depth-first list of records and a sibling map.
    while(!stack.empty())
    {
        auto const & record = stack.top();
        stack.pop();

        linearized_tree.records.push_back(record);
        location_map[record.get()] = linearized_tree.records.size()-1;

        auto const & children = record->children;

        for(auto child_it=children.begin(); child_it!=children.end(); ++child_it)
        {
            auto const & child_record = child_it->second;
            auto const sibling_it = std::next(child_it);
            if(sibling_it != children.end())
            {
                auto const & sibling = sibling_it->second;
                sibling_map[child_record.get()] = sibling.get();
            }
        }

        // Use reverse iterators to keep the relative order of children during
        // the depth-first search.
        std::for_each(
            children.rbegin(), children.rend(),
            [&] (RecordMap::value_type const & child) { stack.push(child.second); });
    }

    // Update the topological information.

    linearized_tree.child.resize(location_map.size());
    std::fill(linearized_tree.child.begin(), linearized_tree.child.end(), -1);

    linearized_tree.sibling.resize(location_map.size());
    std::fill(linearized_tree.sibling.begin(), linearized_tree.sibling.end(), -1);

    for(auto const & location_it: location_map)
    {
        auto const & record = location_it.first;
        auto const & location = location_it.second;
        if(!record->children.empty())
        {
            linearized_tree.child[location] =
                location_map.at(record->children.begin()->second.get());
        }
        else
        {
            linearized_tree.child[location] = -1;
        }

        auto const sibling_it = sibling_map.find(record);
        if(sibling_it != sibling_map.end())
        {
            linearized_tree.sibling[location] =
                location_map.at(sibling_it->second);
        }
        else
        {
            linearized_tree.sibling[location] = -1;
        }
    }

    return linearized_tree;
}

std::vector<std::streampos>
BasicDirectoryCreator
::_get_relative_offsets(std::vector<Record::Pointer> const & records) const
{
    // Create an in-memory stream to compute the position of the start of the
    // records
    std::ostringstream stream;

    // PS3.10, 8.6: The DICOMDIR File shall use the Explicit VR Little Endian
    // Transfer Syntax
    Writer const writer(
        stream, registry::ExplicitVRLittleEndian, this->item_encoding);

    std::vector<std::streampos> relative_offsets;

    std::streampos offset = 0;

    for(auto const & record: records)
    {
        relative_offsets.push_back(stream.tellp()+offset);

        // Item tag: 4 bytes
        // Item length: 4 bytes
        offset += 8;

        writer.write_data_set(record->data_set);

        if(this->item_encoding == Writer::ItemEncoding::UndefinedLength)
        {
            // Item Delimitation Item tag: 4 bytes
            // Item Delimitation Item length: 4 bytes
            offset += 8;
        }
    }

    return relative_offsets;
}

void
BasicDirectoryCreator
::_write(
    LinearizedTree const & linearized_tree,
    std::vector<std::streampos> const & relative_offsets) const
{
    DataSet meta_information;
    meta_information.add(
        registry::FileMetaInformationVersion, Value::Binary({{0x00, 0x01}}));
    meta_information.add(
        registry::MediaStorageSOPClassUID,
        { registry::MediaStorageDirectoryStorage });
    meta_information.add(
        registry::MediaStorageSOPInstanceUID, { generate_uid() });
    meta_information.add(
        registry::TransferSyntaxUID, { registry::ExplicitVRLittleEndian });
    meta_information.add(
        registry::ImplementationClassUID, { implementation_class_uid });
    meta_information.add(
        registry::ImplementationVersionName, { implementation_version_name });

    DataSet basic_directory;

    // File-Set Identification Module (PS3.3 F.3.2.1)
    basic_directory.add(registry::FileSetID, { "" });
    // Don't include user comments yet.
    // registry::FileSetDescriptorFileID;
    // registry::SpecificCharacterSetOfFileSetDescriptorFile;
    // Directory Information Module (PS3.3 F.3.2.2)
    basic_directory.add(
        registry::OffsetOfTheFirstDirectoryRecordOfTheRootDirectoryEntity,
        { 0 });
    basic_directory.add(
        registry::OffsetOfTheLastDirectoryRecordOfTheRootDirectoryEntity,
        { 0 });
    basic_directory.add(registry::FileSetConsistencyFlag, { 0 });

    std::ostringstream buffer;
    Writer buffer_writer(
        buffer, meta_information.as_string(registry::TransferSyntaxUID)[0],
        this->item_encoding);
    buffer_writer.write_data_set(meta_information);
    buffer_writer.write_data_set(basic_directory);

    // Preamble + prefix + data + Directory Record Sequence (tag + VR + reserved + VL)
    auto const start_of_directory_records = 128+4+buffer.tellp()+4+2+2+4;
    std::vector<std::streampos> absolute_offsets(relative_offsets.size());
    std::transform(
        relative_offsets.begin(), relative_offsets.end(), absolute_offsets.begin(),
        [&](std::streampos offset) { return offset+start_of_directory_records; });

    basic_directory.as_int(
        registry::OffsetOfTheFirstDirectoryRecordOfTheRootDirectoryEntity)[0] =
            *(absolute_offsets.begin());

    int last_top_level = 0;
    while(linearized_tree.sibling[last_top_level] != -1)
    {
        last_top_level = linearized_tree.sibling[last_top_level];
    }
    basic_directory.as_int(
        registry::OffsetOfTheLastDirectoryRecordOfTheRootDirectoryEntity)[0] =
            absolute_offsets[last_top_level];

    Value::DataSets record_sequence;
    for(int record_index=0; record_index<linearized_tree.records.size(); ++record_index)
    {
        auto & record = linearized_tree.records[record_index];
        auto & item = record->data_set;

        auto const sibling = linearized_tree.sibling[record_index];
        if(sibling != -1)
        {
            item.as_int(
                registry::OffsetOfTheNextDirectoryRecord)[0] =
                    absolute_offsets[sibling];
        }

        auto const child = linearized_tree.child[record_index];
        if(child != -1)
        {
            item.as_int(
                registry::OffsetOfReferencedLowerLevelDirectoryEntity)[0] =
                    absolute_offsets[child];
        }

        record_sequence.push_back(item);
    }
    basic_directory.add(registry::DirectoryRecordSequence, record_sequence);

    boost::filesystem::ofstream file(
        boost::filesystem::path(root)/"DICOMDIR",
        std::ofstream::out | std::ofstream::binary);
    Writer file_writer(
        file, buffer_writer.byte_ordering, buffer_writer.explicit_vr,
        buffer_writer.item_encoding, buffer_writer.use_group_length);

    // File preamble
    for(unsigned int i=0; i<128; ++i)
    {
        file.put(0);
        if(!file)
        {
            throw Exception("Could not write to stream");
        }
    }

    // DICOM prefix
    std::string const prefix("DICM");
    file.write(&prefix[0], prefix.size());
    if(!file)
    {
        throw Exception("Could not write to stream");
    }

    file_writer.write_data_set(meta_information);
    file_writer.write_data_set(basic_directory);
}

}

max_stars_repo_name: icryrainix/odil

max_issues_repo_name: icryrainix/odil

max_forks_repo_name: icryrainix/odil


text:
/**
 * Copyright (c) 2009 Carnegie Mellon University.
 *     All rights reserved.
 *
 *  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.
 *
 *
 */

#include <cassert>
#include <iostream>
#include <fstream>
#include <vector>
#include <algorithm>
#include <limits>
#include <cmath>

#include <boost/random.hpp>

#include <graphlab.hpp>

#include "image.hpp"

#include <graphlab/macros_def.hpp>

void image::save_vec(const char* filename) const {
  std::ofstream os(filename);
  ASSERT_TRUE(os.good());
  for(size_t i = 0; i < pixels(); ++i) {
    os << pixel(i) << "\n";
  }
  os.flush();
  os.close();
}

double image::min() const {
  return *std::min_element(data.begin(), data.end());
}

double image::max() const {
  return *std::max_element(data.begin(), data.end());
}

void image::save(graphlab::oarchive &oarc) const {
  oarc << _rows;
  oarc << _cols;
  oarc << data;
}

void image::load(graphlab::iarchive &iarc) {
  iarc >> _rows;
  iarc >> _cols;
  iarc >> data;
}

/** Generate a normally distributed random number N(mu, sigma^2) */
// std::pair<double, double> randn(double mu = 0, double sigma = 1 );

// IMPLEMENTATION =============================================================>

void image::resize(size_t rows, size_t cols) {
  _rows = rows;
  _cols = cols;
  data.resize(rows * cols, 0);
}

/** Get the vertex id of a pixel */
size_t image::vertid(size_t i, size_t j) const {
  ASSERT_LT(i, _rows);
  ASSERT_LT(j, _cols);
  return i * _cols + j;
}

// static size_t image::vertid(size_t rows, size_t cols, size_t i, size_t j)  {
//   assert(i < rows);
//   assert(j < cols);
//   return i * cols + j;
// }

/** Get the vertex id of a pixel */
std::pair<size_t, size_t> image::loc(size_t vertexid) const {
  ASSERT_LT(vertexid, _rows * _cols);
  return std::make_pair( vertexid / _cols, vertexid % _cols);
}

void image::save(const char* filename) const {
  std::ofstream os(filename);
  os << "P2" << std::endl
     << _cols << " " << _rows << std::endl
     << 255 << std::endl;
  // Compute min and max pixel intensities
  double min = data[0]; double max = data[0];
  for(size_t i = 0; i < _rows * _cols; ++i) {
    min = std::min(min, data[i]);
    max = std::max(max, data[i]);
  }

  // Save the image (rescaled)
  for(size_t r = 0; r < _rows; ++r) {
    for(size_t c = 0; c < _cols; c++) {
      if(min != max) {
        int color =
          static_cast<int>(255.0 * (pixel(r,c) - min)/(max-min));
        os << color;
      } else { os << min; }
      if(c != _cols-1) os << "\t";
    }
    os << std::endl;
  }
  os.flush();
  os.close();
} // end of save

void image::paint_sunset(size_t states) {
  const double center_r = rows() / 2.0;
  const double center_c = cols() / 2.0;
  const double max_radius = std::min(rows(), cols()) / 2.0;
  // Fill out the image
  for(size_t r = 0; r < rows(); ++r) {
    for(size_t c = 0; c < cols(); ++c) {
      double distance = sqrt((r-center_r)*(r-center_r) +
                             (c-center_c)*(c-center_c));
      // If on top of image
      if(r < rows() / 2) {
        // Compute ring of sunset
        size_t ring =
          static_cast<size_t>(std::floor(std::min(1.0, distance/max_radius)
                                         * (states - 1) ) );
        pixel(r,c) = ring;
      } else {
        size_t blockx = r / 20;
        size_t blocky = (c + 20 * sin(10.0*r/rows())) / 20;
        size_t index = blockx + 2*blocky;
        pixel(r,c) = index % states;

      }
    }
  }
} // end of paint_beatiful_sunset

void image::paint_checkerboard(size_t states, size_t blocks) {
  size_t block_size = std::min(rows(), cols() ) / blocks;
  // Fill out the image
  for(size_t r = 0; r < rows(); ++r) {
    for(size_t c = 0; c < cols(); ++c) {
      size_t blockx = r / block_size;
      size_t blocky = c / block_size;
      size_t index = blockx + blocky * block_size;
      pixel(r,c) = index % states;
    }
  }
} // end of paint_beatiful_sunset

/** corrupt the image with gaussian noise */
void image::gaussian_corrupt(double sigma) {
  //  boost::mt19937 rng;
  boost::lagged_fibonacci607 rng;
  boost::normal_distribution<double> noise_model(0, sigma);
  for(size_t i = 0; i < rows() * cols();  ) {
    // Corrupt two pixels at a time.
    pixel(i++) += noise_model(rng);
  }
} // end of corrupt_image

/** flip_corrupt */
void image::flip_corrupt(size_t states, double prob_flip) {
  boost::mt19937 rng;
  boost::uniform_real<double> dist01;

  for(size_t i = 0; i < rows() * cols();  ++i) {
    double p = dist01(rng);
    if(p < prob_flip) pixel(i) = rand() % states;
  }
} // end of corrupt_image

// /** generate a normally distributed iid pair */
// std::pair<double, double> randn(double mu , double sigma ) {
//   // Generate a N(0,1) from a Unif(0,1) using Box-Muller generator:
//   double u1 = static_cast<double>(rand()) / RAND_MAX;
//   double u2 = static_cast<double>(rand()) / RAND_MAX;
//   double coeff = std::sqrt(-2.0 * std::log(u1));
//   double n1 = coeff * std::cos(2.0 * M_PI * u2) ;
//   double n2 = coeff * std::sin(2.0 * M_PI * u2) ;
//   // Adjust for mean and variance
//   n1 = sigma * n1 + mu;
//   n2 = sigma * n2 + mu;
//   return std::make_pair(n1, n2);
// } // end of randn

#include <graphlab/macros_undef.hpp>

max_issues_repo_name: HybridGraph/GraphLab-PowerGraph

max_forks_repo_name: HybridGraph/GraphLab-PowerGraph


text:
// practice 3.3

#include <iostream>
#include <boost/type_traits/is_reference.hpp>
#include <boost/type_traits/is_same.hpp>
#include <boost/type_traits/is_pointer.hpp>
#include <boost/type_traits/is_float.hpp>
#include <boost/type_traits/alignment_of.hpp>
#include <boost/type_traits/remove_reference.hpp>
#include <boost/type_traits/remove_pointer.hpp>
#include <boost/type_traits/add_pointer.hpp>
#include <boost/static_assert.hpp>
#include <boost/mpl/plus.hpp>
#include <boost/mpl/minus.hpp>
#include <boost/mpl/placeholders.hpp>
#include <boost/mpl/vector_c.hpp>
#include <boost/mpl/vector.hpp>
#include <boost/mpl/transform.hpp>
#include <boost/mpl/equal.hpp>
#include <boost/mpl/apply.hpp>
using namespace boost::mpl::placeholders;
#include <iterator>
#include <utility>
#include <list>
#include <vector>
#include <string>
#include <assert.h>

namespace mpl = boost::mpl;

struct add_ptr
{
	template<class T>
	struct apply
	{
		typedef T* type;
	};
};

// struct once
// {
// 	template<class F, class X>
// 	struct apply
// 	{
// 		typedef F::apply< X >::type type;
// 	};
// };

// F< F<X> >
template<class F, class X>
struct twice :
	mpl::apply<
		F,
		typename mpl::apply<F, X>::type
	>
{
// 	typedef typename F::apply< X >::type once;
// 	typedef typename F::apply< once >::type type;

// 	template<class F, class X>
// 	struct apply
// 	{
// 		typedef typename F::apply<
// 			F::apply< X >::type >::type type;
// 	};

// 	typedef typename mpl::lambda<
// 		//typename mpl::lambda<F::apply, X>::type,
// 		F,
// 		X >::type type;

};

template<class T>
struct twoptr :
	twice<
		typename mpl::lambda< boost::add_pointer< _1 > >::type
	,	T >
{};

void main()
{
	bool r1 = boost::is_same<
			add_ptr::apply<int>::type
		,	int*
	>::value;

	bool r2 = boost::is_same<
		twoptr<int>::type
		,	int**
	>::value;

	// practice 3-3
	bool r3 = boost::is_same<
		twoptr<
			typename twoptr<int>::type >::type
		,	int****
	>::value;

	// practice 3-4
	bool r4 = boost::is_same<

	typename twice<
		typename boost::add_pointer<_>,

		typename twice<
			typename boost::add_pointer<_>
			, int >::type
	>::type

		,	int****
	>::value;

}

max_stars_repo_name: jjuiddong/TemplateMetaProgramming

max_issues_repo_name: jjuiddong/TemplateMetaProgramming

max_forks_repo_name: jjuiddong/TemplateMetaProgramming


text:
// ----------------------------------------------------------------------------
// -                        Open3D: www.open3d.org                            -
// ----------------------------------------------------------------------------
// The MIT License (MIT)
//
// Copyright (c) 2018 www.open3d.org
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in
// all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
// IN THE SOFTWARE.
// ----------------------------------------------------------------------------

#include "TransformationEstimation.h"

#include <Eigen/Core>
#include <Eigen/Geometry>
#include <Eigen/LU>
#include <Eigen/SVD>
#include <Eigen/Dense>
#include <Open3D/Geometry/PointCloud.h>
#include <Open3D/Utility/Eigen.h>
#include <iostream>

namespace Eigen {

#ifndef EIGEN_PARSED_BY_DOXYGEN

    // These helpers are required since it allows to use mixed types as parameters
    // for the Umeyama. The problem with mixed parameters is that the return type
    // cannot trivially be deduced when float and double types are mixed.
    namespace internal {

        // Compile time return type deduction for different MatrixBase types.
        // Different means here different alignment and parameters but the same underlying
        // real scalar type.
        template<typename MatrixType, typename OtherMatrixType>
            struct umeyama_transform_matrix_type2
            {
                enum {
                    MinRowsAtCompileTime = EIGEN_SIZE_MIN_PREFER_DYNAMIC(MatrixType::RowsAtCompileTime, OtherMatrixType::RowsAtCompileTime),

                    // When possible we want to choose some small fixed size value since the result
                    // is likely to fit on the stack. So here, EIGEN_SIZE_MIN_PREFER_DYNAMIC is not what we want.
                    HomogeneousDimension = int(MinRowsAtCompileTime) == Dynamic ? Dynamic : int(MinRowsAtCompileTime)+1
                };

                typedef Matrix<typename traits<MatrixType>::Scalar,
                        HomogeneousDimension,
                        HomogeneousDimension,
                        AutoAlign | (traits<MatrixType>::Flags & RowMajorBit ? RowMajor : ColMajor),
                        HomogeneousDimension,
                        HomogeneousDimension
                            > type;
            };

    }

#endif

    // Similar to the original Eigen::Umeyama  https://eigen.tuxfamily.org/dox/Umeyama_8h_source.html
    template <typename Derived, typename OtherDerived>
    Eigen::Matrix4d umeyama_constrained(const Eigen::MatrixXd& _src, const Eigen::MatrixXd& _dst, bool with_scaling = true) {
        typedef typename Eigen::internal::umeyama_transform_matrix_type2<Derived, OtherDerived>::type TransformationMatrixType;
        typedef typename Eigen::internal::traits<TransformationMatrixType>::Scalar Scalar;
        typedef typename Eigen::NumTraits<Scalar>::Real RealScalar;

        enum { Dimension = EIGEN_SIZE_MIN_PREFER_DYNAMIC(Derived::RowsAtCompileTime, OtherDerived::RowsAtCompileTime) };

        typedef Eigen::Matrix<Scalar, Dimension, 1> VectorType;
        typedef Eigen::Matrix<Scalar, Dimension, Dimension> MatrixType;
        typedef typename Eigen::internal::plain_matrix_type_row_major<Derived>::type RowMajorMatrixType;

        const Index m = _src.rows(); // dimension
        const Index n = _src.cols(); // number of measurements

        Eigen::MatrixXd src(m, n);
        src << _src.topRows(2), Eigen::MatrixXd::Zero(1, n);
        Eigen::MatrixXd dst(m, n);
        dst << _dst.topRows(2), Eigen::MatrixXd::Zero(1, n);

        // required for demeaning ...
        const RealScalar one_over_n = RealScalar(1) / static_cast<RealScalar>(n);

        // computation of mean
        const VectorType src_mean = src.rowwise().sum() * one_over_n;
        const VectorType dst_mean = dst.rowwise().sum() * one_over_n;

        // demeaning of src and dst points
        const RowMajorMatrixType src_demean = src.colwise() - src_mean;
        const RowMajorMatrixType dst_demean = dst.colwise() - dst_mean;

        // Eq. (36)-(37)
        const Scalar src_var = src_demean.rowwise().squaredNorm().sum() * one_over_n;

        // Eq. (38)
        const MatrixType sigma = one_over_n * dst_demean * src_demean.transpose();

        Eigen::JacobiSVD<MatrixType> svd(sigma, Eigen::ComputeFullU | Eigen::ComputeFullV);

        // Initialize the resulting transformation with an identity matrix...
        TransformationMatrixType Rt = TransformationMatrixType::Identity(m+1,m+1);

        // Eq. (39)
        VectorType S = VectorType::Ones(m);

        if  ( svd.matrixU().determinant() * svd.matrixV().determinant() < 0 ) {
            // S(m-1) = -1;
            Rt.col(m).head(m) = dst_mean-src_mean;
            return Rt;
        }

        // Eq. (40) and (43)
        Rt.block(0,0,m,m).noalias() = svd.matrixU() * S.asDiagonal() * svd.matrixV().transpose();

        if (with_scaling)
        {
            // Eq. (42)
            const Scalar c = Scalar(1)/src_var * svd.singularValues().dot(S);

            // Eq. (41)
            Rt.col(m).head(m) = dst_mean;
            Rt.col(m).head(m).noalias() -= c*Rt.topLeftCorner(m,m)*src_mean;
            Rt.block(0,0,m,m) *= c;
        }
        else
        {
            Rt.col(m).head(m) = dst_mean;
            Rt.col(m).head(m).noalias() -= Rt.topLeftCorner(m,m)*src_mean;
        }
        return Rt;
    }
}

namespace open3d {
namespace registration {

double TransformationEstimationPointToPoint::ComputeRMSE(
        const geometry::PointCloud &source,
        const geometry::PointCloud &target,
        const CorrespondenceSet &corres) const {
    if (corres.empty()) return 0.0;
    double err = 0.0;
    for (const auto &c : corres) {
        err += (source.points_[c[0]] - target.points_[c[1]]).squaredNorm();
    }
    return std::sqrt(err / (double)corres.size());
}

Eigen::Matrix4d TransformationEstimationPointToPoint::ComputeTransformation(
        const geometry::PointCloud &source,
        const geometry::PointCloud &target,
        const CorrespondenceSet &corres) const {
    if (corres.empty()) return Eigen::Matrix4d::Identity();
    Eigen::MatrixXd source_mat(3, corres.size());
    Eigen::MatrixXd target_mat(3, corres.size());
    for (size_t i = 0; i < corres.size(); i++) {
        source_mat.block<3, 1>(0, i) = source.points_[corres[i][0]];
        target_mat.block<3, 1>(0, i) = target.points_[corres[i][1]];
    }
    if (with_constraint_) {
        return Eigen::umeyama_constrained<Eigen::MatrixXd, Eigen::MatrixXd>(source_mat, target_mat, with_scaling_);
    }
    else {
        return Eigen::umeyama(source_mat, target_mat, with_scaling_);
    }
}

double TransformationEstimationPointToPlane::ComputeRMSE(
        const geometry::PointCloud &source,
        const geometry::PointCloud &target,
        const CorrespondenceSet &corres) const {
    if (corres.empty() || target.HasNormals() == false) return 0.0;
    double err = 0.0, r;
    for (const auto &c : corres) {
        r = (source.points_[c[0]] - target.points_[c[1]])
                    .dot(target.normals_[c[1]]);
        err += r * r;
    }
    return std::sqrt(err / (double)corres.size());
}

Eigen::Matrix4d TransformationEstimationPointToPlane::ComputeTransformation(
        const geometry::PointCloud &source,
        const geometry::PointCloud &target,
        const CorrespondenceSet &corres) const {
    if (corres.empty() || target.HasNormals() == false)
        return Eigen::Matrix4d::Identity();

    auto compute_jacobian_and_residual = [&](int i, Eigen::Vector6d &J_r,
                                             double &r) {
        const Eigen::Vector3d &vs = source.points_[corres[i][0]];
        const Eigen::Vector3d &vt = target.points_[corres[i][1]];
        const Eigen::Vector3d &nt = target.normals_[corres[i][1]];
        r = (vs - vt).dot(nt);
        J_r.block<3, 1>(0, 0) = vs.cross(nt);
        J_r.block<3, 1>(3, 0) = nt;
    };

    Eigen::Matrix6d JTJ;
    Eigen::Vector6d JTr;
    double r2;
    std::tie(JTJ, JTr, r2) =
            utility::ComputeJTJandJTr<Eigen::Matrix6d, Eigen::Vector6d>(
                    compute_jacobian_and_residual, (int)corres.size());

    bool is_success;
    Eigen::Matrix4d extrinsic;
    std::tie(is_success, extrinsic) =
            utility::SolveJacobianSystemAndObtainExtrinsicMatrix(JTJ, JTr);

    return is_success ? extrinsic : Eigen::Matrix4d::Identity();
}

}  // namespace registration
}  // namespace open3d