<commit_before>#include <algorithm>
#include <libdariadb/config.h>
#include <libdariadb/engine.h>
#include <libdariadb/flags.h>
#include <libdariadb/storage/cursors.h>
#include <libdariadb/storage/dropper.h>
#include <libdariadb/storage/engine_environment.h>
#include <libdariadb/storage/manifest.h>
#include <libdariadb/storage/memstorage/memstorage.h>
#include <libdariadb/storage/pages/page_manager.h>
#include <libdariadb/storage/subscribe.h>
#include <libdariadb/timeutil.h>
#include <libdariadb/utils/async/locker.h>
#include <libdariadb/utils/async/thread_manager.h>
#include <libdariadb/utils/exception.h>
#include <libdariadb/utils/fs.h>
#include <libdariadb/utils/logger.h>
#include <libdariadb/utils/strings.h>
#include <libdariadb/utils/utils.h>

#include <cstring>
#include <fstream>
#include <iomanip>
#include <shared_mutex>
#include <sstream>

using namespace dariadb;
using namespace dariadb::storage;
using namespace dariadb::utils::async;

class Engine::Private {
public:
  Private(Settings_ptr settings, bool ignore_lock_file) {
    _settings = settings;
    _strategy = _settings->strategy.value();

    _engine_env = EngineEnvironment::create();
    _engine_env->addResource(EngineEnvironment::Resource::SETTINGS,
                             _settings.get());

    logger_info("engine: project version - ", version());
    logger_info("engine: storage format - ", format());
    logger_info("engine: strategy - ", _settings->strategy.value());
    _stoped = false;

    if (!dariadb::utils::fs::path_exists(_settings->storage_path.value())) {
      dariadb::utils::fs::mkdir(_settings->storage_path.value());
      dariadb::utils::fs::mkdir(_settings->raw_path.value());
    }

    lockfile_lock_or_die(ignore_lock_file);

    auto manifest_file_name = utils::fs::append_path(
        _settings->storage_path.value(), MANIFEST_FILE_NAME);

    bool is_new_storage = !utils::fs::file_exists(manifest_file_name);
    if (is_new_storage) {
      logger_info("engine: init new storage.");
    }
    _subscribe_notify.start();
    ThreadManager::Params tpm_params(_settings->thread_pools_params());
    ThreadManager::start(tpm_params);

    _manifest = Manifest::create(_settings);
    _engine_env->addResource(EngineEnvironment::Resource::MANIFEST,
                             _manifest.get());

    if (is_new_storage) {
      _manifest->set_format(std::to_string(format()));
    } else {
      check_storage_version();
      Dropper::cleanStorage(_settings->raw_path.value());
    }

    _page_manager = PageManager::create(_engine_env);

    if (_settings->load_min_max) {
      _min_max_map = _page_manager->loadMinMax();
    }

    if (_strategy != STRATEGY::MEMORY) {
      _wal_manager = WALManager::create(_engine_env);

      _dropper =
          std::make_unique<Dropper>(_engine_env, _page_manager, _wal_manager);
      _wal_manager->setDownlevel(_dropper.get());
      this->_top_level_storage = _wal_manager;
    }

    if (_strategy == STRATEGY::CACHE || _strategy == STRATEGY::MEMORY) {
      _memstorage = MemStorage::create(_engine_env, _min_max_map.size());
      if (_strategy != STRATEGY::CACHE) {
        _memstorage->setDownLevel(_page_manager.get());
      }
      _top_level_storage = _memstorage;
      if (_strategy == STRATEGY::CACHE) {
        _memstorage->setDiskStorage(_wal_manager.get());
      }
    }

    if (_strategy == STRATEGY::WAL) {
      if (_settings->load_min_max) {
        auto amm = _top_level_storage->loadMinMax();
        minmax_append(_min_max_map, amm);
      }
    }

    logger_info("engine: start - OK ");
  }
  ~Private() { this->stop(); }

  void init_storages() {}
  void stop() {
    if (!_stoped) {
      _top_level_storage = nullptr;
      _subscribe_notify.stop();

      this->flush();
      if (_memstorage != nullptr) {
        _memstorage = nullptr;
      }
      _wal_manager = nullptr;
      _page_manager = nullptr;
      _manifest = nullptr;
      _dropper = nullptr;
      _stoped = true;

      ThreadManager::stop();
      lockfile_unlock();
    }
  }

  std::string lockfile_path() {
    return utils::fs::append_path(_settings->storage_path.value(), "lockfile");
  }

  void lockfile_lock_or_die(bool ignore_lock_file) {
    auto lfile = lockfile_path();
    if (utils::fs::path_exists(lfile)) {
      if (!ignore_lock_file) {
        throw_lock_error(_settings->storage_path.value());
      } else {
        lockfile_unlock();
      }
    }
    // TODO use lockfile from Boost.
    std::ofstream ofs;
    ofs.open(lfile, std::ios_base::out | std::ios_base::binary);
    if (!ofs.is_open()) {
      throw_lock_error(_settings->storage_path.value());
    }
    ofs << "locked.";
    ofs.close();
  }

  void lockfile_unlock() {<commit_msg>rm todo.<commit_after>#include <algorithm>
#include <libdariadb/config.h>
#include <libdariadb/engine.h>
#include <libdariadb/flags.h>
#include <libdariadb/storage/cursors.h>
#include <libdariadb/storage/dropper.h>
#include <libdariadb/storage/engine_environment.h>
#include <libdariadb/storage/manifest.h>
#include <libdariadb/storage/memstorage/memstorage.h>
#include <libdariadb/storage/pages/page_manager.h>
#include <libdariadb/storage/subscribe.h>
#include <libdariadb/timeutil.h>
#include <libdariadb/utils/async/locker.h>
#include <libdariadb/utils/async/thread_manager.h>
#include <libdariadb/utils/exception.h>
#include <libdariadb/utils/fs.h>
#include <libdariadb/utils/logger.h>
#include <libdariadb/utils/strings.h>
#include <libdariadb/utils/utils.h>

#include <cstring>
#include <fstream>
#include <iomanip>
#include <shared_mutex>
#include <sstream>

using namespace dariadb;
using namespace dariadb::storage;
using namespace dariadb::utils::async;

class Engine::Private {
public:
  Private(Settings_ptr settings, bool ignore_lock_file) {
    _settings = settings;
    _strategy = _settings->strategy.value();

    _engine_env = EngineEnvironment::create();
    _engine_env->addResource(EngineEnvironment::Resource::SETTINGS,
                             _settings.get());

    logger_info("engine: project version - ", version());
    logger_info("engine: storage format - ", format());
    logger_info("engine: strategy - ", _settings->strategy.value());
    _stoped = false;

    if (!dariadb::utils::fs::path_exists(_settings->storage_path.value())) {
      dariadb::utils::fs::mkdir(_settings->storage_path.value());
      dariadb::utils::fs::mkdir(_settings->raw_path.value());
    }

    lockfile_lock_or_die(ignore_lock_file);

    auto manifest_file_name = utils::fs::append_path(
        _settings->storage_path.value(), MANIFEST_FILE_NAME);

    bool is_new_storage = !utils::fs::file_exists(manifest_file_name);
    if (is_new_storage) {
      logger_info("engine: init new storage.");
    }
    _subscribe_notify.start();
    ThreadManager::Params tpm_params(_settings->thread_pools_params());
    ThreadManager::start(tpm_params);

    _manifest = Manifest::create(_settings);
    _engine_env->addResource(EngineEnvironment::Resource::MANIFEST,
                             _manifest.get());

    if (is_new_storage) {
      _manifest->set_format(std::to_string(format()));
    } else {
      check_storage_version();
      Dropper::cleanStorage(_settings->raw_path.value());
    }

    _page_manager = PageManager::create(_engine_env);

    if (_settings->load_min_max) {
      _min_max_map = _page_manager->loadMinMax();
    }

    if (_strategy != STRATEGY::MEMORY) {
      _wal_manager = WALManager::create(_engine_env);

      _dropper =
          std::make_unique<Dropper>(_engine_env, _page_manager, _wal_manager);
      _wal_manager->setDownlevel(_dropper.get());
      this->_top_level_storage = _wal_manager;
    }

    if (_strategy == STRATEGY::CACHE || _strategy == STRATEGY::MEMORY) {
      _memstorage = MemStorage::create(_engine_env, _min_max_map.size());
      if (_strategy != STRATEGY::CACHE) {
        _memstorage->setDownLevel(_page_manager.get());
      }
      _top_level_storage = _memstorage;
      if (_strategy == STRATEGY::CACHE) {
        _memstorage->setDiskStorage(_wal_manager.get());
      }
    }

    if (_strategy == STRATEGY::WAL) {
      if (_settings->load_min_max) {
        auto amm = _top_level_storage->loadMinMax();
        minmax_append(_min_max_map, amm);
      }
    }

    logger_info("engine: start - OK ");
  }
  ~Private() { this->stop(); }

  void init_storages() {}
  void stop() {
    if (!_stoped) {
      _top_level_storage = nullptr;
      _subscribe_notify.stop();

      this->flush();
      if (_memstorage != nullptr) {
        _memstorage = nullptr;
      }
      _wal_manager = nullptr;
      _page_manager = nullptr;
      _manifest = nullptr;
      _dropper = nullptr;
      _stoped = true;

      ThreadManager::stop();
      lockfile_unlock();
    }
  }

  std::string lockfile_path() {
    return utils::fs::append_path(_settings->storage_path.value(), "lockfile");
  }

  void lockfile_lock_or_die(bool ignore_lock_file) {
    auto lfile = lockfile_path();
    if (utils::fs::path_exists(lfile)) {
      if (!ignore_lock_file) {
        throw_lock_error(_settings->storage_path.value());
      } else {
        lockfile_unlock();
      }
    }
    std::ofstream ofs;
    ofs.open(lfile, std::ios_base::out | std::ios_base::binary);
    if (!ofs.is_open()) {
      throw_lock_error(_settings->storage_path.value());
    }
    ofs << "locked.";
    ofs.close();
  }

  void lockfile_unlock() {
{'fraction_non_alphanumeric': 0.14173562486738808, 'fraction_numerical': 0.0, 'mean_word_length': 3.7371859296482413, 'pattern_counts': {'":': 0, '<': 57, '<?xml version=': 0, '>': 101, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3998808', 'n_tokens_mistral': 3211, 'n_tokens_neox': 3139, 'n_words': 460}