Method setupStorage

programs/keeper-bench/StorageRunner.cpp:178–248

}

void StorageRunner::setupStorage()
{
    auto settings = std::make_shared<DB::CoordinationSettings>();
    settings->loadFromConfig("storage.coordination_settings", *config_ptr);
    keeper_context = std::make_shared<DB::KeeperContext>(/*standalone_keeper=*/true, settings);
    keeper_context->setLocalLogsPreprocessed();
    keeper_context->setRocksDBOptions();
    keeper_context->setServerState(DB::KeeperContext::Phase::RUNNING);

    storage = std::make_unique<Storage>(tick_time_ms, /*superdigest=*/"", keeper_context);

    /// Allocate one session for setup and one per generator thread.
    /// All subsequent requests from a generator use its dedicated session.
    const int64_t setup_timeout_ms = std::max<int64_t>(60'000, tick_time_ms * 1000);
    setup_session_id = storage->getSessionID(setup_timeout_ms);
    for (size_t i = 0; i < concurrency; ++i)
        generator_session_ids.push_back(storage->getSessionID(setup_timeout_ms));

    /// Recursively create the setup tree. The real `BenchmarkContext::startup` uses a
    /// ZooKeeper client, so we walk the node tree here and issue preprocess+commit
    /// directly against the storage, populating `tagged_paths` as we go.
    std::function<void(const BenchmarkContext::Node &, const std::string &)> create_subtree;
    auto & tagged = benchmark_context.getTaggedPaths();
    const Coordination::ACLs & default_acls = benchmark_context.getDefaultAcls();

    create_subtree = [&](const BenchmarkContext::Node & node, const std::string & parent_path)
    {
        std::string path = parent_path == "/" ? "/" + node.name.getString() : parent_path + "/" + node.name.getString();

        auto request = std::make_shared<Coordination::ZooKeeperCreateRequest>();
        request->path = path;
        request->data = node.data ? node.data->getString() : "";
        request->acls = default_acls;

        int64_t zxid = next_zxid.fetch_add(1);
        try
        {
            storage->preprocessRequest(request, setup_session_id, 0, zxid);
            auto responses = storage->processRequest(request, setup_session_id, zxid);
            for (const auto & response : responses)
            {
                if (response.response->error != Coordination::Error::ZOK)
                    throw zkutil::KeeperException::fromPath(response.response->error, path);
            }
        }
        catch (...)
        {
            std::cerr << "Setup request failed for path " << path << ": "
                << DB::getCurrentExceptionMessage(false) << std::endl;
            throw;
        }

        if (node.tag)
            tagged[*node.tag].push_back(path);

        for (const auto & child : node.children)
            create_subtree(*child, path);
    };