<commit_before>#include <libdariadb/storage/dropper.h>
#include <libdariadb/storage/manifest.h>
#include <libdariadb/storage/pages/page.h>
#include <libdariadb/storage/settings.h>
#include <libdariadb/utils/async/thread_manager.h>
#include <ctime>

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

std::shared_ptr<SplitedById> splitById(const MeasArray &ma) {
  std::shared_ptr<SplitedById> result = std::make_shared<SplitedById>();

  std::map<Id, size_t> id2count;
  for (auto v : ma) {
    auto iter = id2count.find(v.id);
    if (iter == id2count.end()) {
      id2count.insert(std::make_pair(v.id, size_t(0)));
    } else {
      id2count[v.id] += 1;
    }
  }

  for (auto i2c : id2count) {
    (*result)[i2c.first].reserve(i2c.second);
  }

  for (auto v : ma) {
    (*result)[v.id].push_back(v);
  }
  return result;
}

Dropper::Dropper(EngineEnvironment_ptr engine_env, PageManager_ptr page_manager,
                 WALManager_ptr wal_manager)
    : _page_manager(page_manager), _wal_manager(wal_manager), _engine_env(engine_env) {
  _active_operations = 0;
  _stop = false;
  _is_stoped = false;
  _settings =
      _engine_env->getResourceObject<Settings>(EngineEnvironment::Resource::SETTINGS);
  _thread_handle = std::thread(&Dropper::drop_wal_internal, this);
}

Dropper::~Dropper() {
  logger("engine", _settings->alias, ": dropper - stop begin.");
  _stop = true;
  while (!_is_stoped) {
    _cond_var.notify_all();
    _dropper_cond_var.notify_all();
  }
  _thread_handle.join();
  logger("engine", _settings->alias, ": dropper - stop end.");
}

DropperDescription Dropper::description() const {
  std::lock_guard<std::mutex> lg(_queue_locker);
  DropperDescription result;
  result.wal = _files_queue.size();
  result.active_works = _active_operations;
  return result;
}

void Dropper::dropWAL(const std::string &fname) {
  std::lock_guard<std::mutex> lg(_queue_locker);
  if (std::find(_files_queue.begin(), _files_queue.end(), fname) != _files_queue.end()) {
    return;
  }
  auto storage_path = _settings->raw_path.value();
  if (utils::fs::path_exists(utils::fs::append_path(storage_path, fname))) {
    _files_queue.emplace_back(fname);
    _cond_var.notify_all();
  }
}

void Dropper::cleanStorage(const std::string &storagePath) {
  logger_info("engine: dropper - check storage ", storagePath);
  auto wals_lst = fs::ls(storagePath, WAL_FILE_EXT);
  auto page_lst = fs::ls(storagePath, PAGE_FILE_EXT);

  for (auto &wal : wals_lst) {
    auto wal_fname = fs::filename(wal);
    for (auto &pagef : page_lst) {
      auto page_fname = fs::filename(pagef);
      if (page_fname == wal_fname) {
        logger_info("engine: fsck wal drop not finished: ", wal_fname);
        logger_info("engine: fsck rm ", pagef);
        PageManager::erase(storagePath, fs::extract_filename(pagef));
      }
    }
  }
}

void Dropper::drop_wal_internal() {
  _is_stoped = false;
  while (!_stop) {
    std::string fname;
    {
      std::unique_lock<std::mutex> ul(_queue_locker);
      _cond_var.wait(ul, [this]() { return !this->_files_queue.empty() || _stop; });
      if (_stop) {
        break;
      }
      if (_files_queue.empty()) {
        continue;
      }
      fname = _files_queue.front();
    }
    while (true) {
      if (_stop) {
        break;
      }
      if (_dropper_lock.try_lock()) {
        break;
      }
    }

    if (_stop) {
      break;
    }
    ENSURE(_active_operations == 0);
    ++_active_operations;
    _state = DROPPER_STATE::OK;
    AsyncTask at = [fname, this](const ThreadInfo &ti) {
      try {
        TKIND_CHECK(THREAD_KINDS::DISK_IO, ti.kind);
        ENSURE(_active_operations == 1);
        this->drop_stage_read(fname);
      } catch (std::exception &ex) {
        _state = DROPPER_STATE::ERROR;
        THROW_EXCEPTION("Dropper::drop_wal_internal: ", ex.what());
      }
      return false;
    };

    ThreadManager::instance()->post(THREAD_KINDS::DISK_IO,
                                    AT_PRIORITY(at, TASK_PRIORITY::DEFAULT));
    while (_active_operations != size_t(0) && !_stop) {
      std::this_thread::sleep_for(std::chrono::milliseconds(100));
    }
    if (_state != DROPPER_STATE::ERROR) {
      std::lock_guard<std::mutex> lg(_queue_locker);
      _files_queue.pop_front();
      _dropper_lock.unlock();
    }
  }
  _is_stoped = true;
}

void Dropper::flush() {
  logger_info("engine", _settings->alias, ": Dropper flush...");
  while (this->description().wal != 0 && _active_operations != size_t(0)) {
    this->_cond_var.notify_all();
    std::this_thread::sleep_for(std::chrono::milliseconds(100));
  }<commit_msg>droper: always wait drop io opertion.<commit_after>#include <libdariadb/storage/dropper.h>
#include <libdariadb/storage/manifest.h>
#include <libdariadb/storage/pages/page.h>
#include <libdariadb/storage/settings.h>
#include <libdariadb/utils/async/thread_manager.h>
#include <ctime>

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

std::shared_ptr<SplitedById> splitById(const MeasArray &ma) {
  std::shared_ptr<SplitedById> result = std::make_shared<SplitedById>();

  std::map<Id, size_t> id2count;
  for (auto v : ma) {
    auto iter = id2count.find(v.id);
    if (iter == id2count.end()) {
      id2count.insert(std::make_pair(v.id, size_t(0)));
    } else {
      id2count[v.id] += 1;
    }
  }

  for (auto i2c : id2count) {
    (*result)[i2c.first].reserve(i2c.second);
  }

  for (auto v : ma) {
    (*result)[v.id].push_back(v);
  }
  return result;
}

Dropper::Dropper(EngineEnvironment_ptr engine_env, PageManager_ptr page_manager,
                 WALManager_ptr wal_manager)
    : _page_manager(page_manager), _wal_manager(wal_manager), _engine_env(engine_env) {
  _active_operations = 0;
  _stop = false;
  _is_stoped = false;
  _settings =
      _engine_env->getResourceObject<Settings>(EngineEnvironment::Resource::SETTINGS);
  _thread_handle = std::thread(&Dropper::drop_wal_internal, this);
}

Dropper::~Dropper() {
  logger("engine", _settings->alias, ": dropper - stop begin.");
  _stop = true;
  while (!_is_stoped) {
    _cond_var.notify_all();
    _dropper_cond_var.notify_all();
  }
  _thread_handle.join();
  ENSURE(_active_operations == 0);
  logger("engine", _settings->alias, ": dropper - stop end.");
}

DropperDescription Dropper::description() const {
  std::lock_guard<std::mutex> lg(_queue_locker);
  DropperDescription result;
  result.wal = _files_queue.size();
  result.active_works = _active_operations;
  return result;
}

void Dropper::dropWAL(const std::string &fname) {
  std::lock_guard<std::mutex> lg(_queue_locker);
  if (std::find(_files_queue.begin(), _files_queue.end(), fname) != _files_queue.end()) {
    return;
  }
  auto storage_path = _settings->raw_path.value();
  if (utils::fs::path_exists(utils::fs::append_path(storage_path, fname))) {
    _files_queue.emplace_back(fname);
    _cond_var.notify_all();
  }
}

void Dropper::cleanStorage(const std::string &storagePath) {
  logger_info("engine: dropper - check storage ", storagePath);
  auto wals_lst = fs::ls(storagePath, WAL_FILE_EXT);
  auto page_lst = fs::ls(storagePath, PAGE_FILE_EXT);

  for (auto &wal : wals_lst) {
    auto wal_fname = fs::filename(wal);
    for (auto &pagef : page_lst) {
      auto page_fname = fs::filename(pagef);
      if (page_fname == wal_fname) {
        logger_info("engine: fsck wal drop not finished: ", wal_fname);
        logger_info("engine: fsck rm ", pagef);
        PageManager::erase(storagePath, fs::extract_filename(pagef));
      }
    }
  }
}

void Dropper::drop_wal_internal() {
  _is_stoped = false;
  while (!_stop) {
    std::string fname;
    {
      std::unique_lock<std::mutex> ul(_queue_locker);
      _cond_var.wait(ul, [this]() { return !this->_files_queue.empty() || _stop; });
      if (_stop) {
        break;
      }
      if (_files_queue.empty()) {
        continue;
      }
      fname = _files_queue.front();
    }
    while (true) {
      if (_stop) {
        break;
      }
      if (_dropper_lock.try_lock()) {
        break;
      }
    }

    if (_stop) {
      break;
    }
    ENSURE(_active_operations == 0);
    ++_active_operations;
    _state = DROPPER_STATE::OK;
    AsyncTask at = [fname, this](const ThreadInfo &ti) {
      try {
        TKIND_CHECK(THREAD_KINDS::DISK_IO, ti.kind);
        ENSURE(_active_operations == 1);
        this->drop_stage_read(fname);
      } catch (std::exception &ex) {
        _state = DROPPER_STATE::ERROR;
        THROW_EXCEPTION("Dropper::drop_wal_internal: ", ex.what());
      }
      return false;
    };

    ThreadManager::instance()->post(THREAD_KINDS::DISK_IO,
                                    AT_PRIORITY(at, TASK_PRIORITY::DEFAULT));
    while (_active_operations != size_t(0)) {
      std::this_thread::sleep_for(std::chrono::milliseconds(100));
    }
    if (_state != DROPPER_STATE::ERROR) {
      std::lock_guard<std::mutex> lg(_queue_locker);
      _files_queue.pop_front();
      _dropper_lock.unlock();
    }
  }
  _is_stoped = true;
}

void Dropper::flush() {
  logger_info("engine", _settings->alias, ": Dropper flush...");
  while (this->description().wal != 0 && _active_operations != size_t(0)) {
    this->_cond_var.notify_all();
    std::this_thread::sleep_for(std::chrono::milliseconds(100));
  }
{'fraction_non_alphanumeric': 0.15098351940457203, 'fraction_numerical': 0.004997341839447103, 'mean_word_length': 3.698301698301698, 'pattern_counts': {'":': 6, '<': 33, '<?xml version=': 0, '>': 53, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3285282', 'n_tokens_mistral': 3372, 'n_tokens_neox': 3346, 'n_words': 544}