multiwriter.cpp

// Standalone version of a trashing client inspired by mutliclient.cc in cephfs tests

#include <atomic>
#include <cstring>
#include <charconv>
#include <chrono>
#include <iostream>
#include <fstream>
#include <thread>
#include <vector>
#include <unistd.h>
#include <fcntl.h>
#include <sys/xattr.h>
#include <sys/stat.h>

std::atomic<bool> g_shutdown {false};
const size_t BUFFER_SIZE = 4096;
char WRITE_BUFFER[BUFFER_SIZE];


template <typename T>
T getIntfromEnv(const char* env_var, T def_val)
{
  if (const char* val = std::getenv(env_var)) {
    std::cerr << "Parsing " << env_var << " with value=" << val;
    T value;
    auto [ptr, ec] = std::from_chars(val, val + std::strlen(val), value);
    if (ec == std::errc()) {
      std::cerr << " Got parsed value=" << value << std::endl;
      return value;
    }  else if (ec == std::errc::invalid_argument)
      std::cerr << "This is not a number.\n";
    else if (ec == std::errc::result_out_of_range)
      std::cerr << "This number is larger than an int.\n";
  }
  return def_val;
}

void make_prefix(std::string& prefix) {
  if (!prefix.empty() && !prefix.ends_with('.')) {
    prefix += '.';
  }
}

std::string make_filename(const std::string& dir, int tid, std::string prefix)
{
  make_prefix(prefix);

  std::string fname = dir + "/" + prefix + "out." + std::to_string(tid);
  return fname;
}

void write_func(const std::string& dir, int tid, std::string prefix="") {
  std::string fname = make_filename(dir, tid, prefix);
  int fd = open(fname.c_str(), O_CREAT | O_RDWR, 644);
  if (fd == -1) {
    std::cerr << "Unable to open file: " << fname << " err= " << std::strerror(errno)  << std::endl;
    return;
  }

  while (!g_shutdown.load(std::memory_order_relaxed)) {
    ftruncate(fd, 4096);
    for (int i=0; i<1040; ++i) {
      write(fd, WRITE_BUFFER, BUFFER_SIZE);
    }
  }

  close(fd);
}

void attr_func(const std::string& dir, int tid, std::string prefix="") {
  std::string fname = make_filename(dir, tid, prefix);
  int fd = open(fname.c_str(), O_CREAT | O_RDWR, 644);
  if (fd == -1) {
    std::cerr << "Unable to open file: " << fname << " err= " << std::strerror(errno)  << std::endl;
    return;
  }


  while (!g_shutdown.load(std::memory_order_relaxed)) {
    if (fsetxattr(fd, "user.foo", "bar", 3, 0) == -1) {
      std::cerr << "Failed attr setting for file:" << fname << "err =" << std::strerror(errno) << std::endl;
    }
  }

  close(fd);
}


void read_func(const std::string& dir, int tid, std::string prefix="") {
  std::string fname = make_filename(dir, tid, prefix);
  struct stat file_stat;

  while (!g_shutdown.load(std::memory_order_relaxed)) {
    if (stat(fname.c_str(), &file_stat) == -1) {
      std::cerr << "Failed stat for file:" << fname << "err =" << std::strerror(errno) << std::endl;
    }
  }
}

int main(int argc, char* argv[])
{

  if (argc < 2 || argc > 4) {
    std::cerr << "Usage: " << argv[0] << " <path> [prefix]";
    return -1;
  }
  std::string dir, prefix;

  dir = argv[1];
  if (argc == 3) {
    prefix = argv[2];
  }

  std::fill_n(WRITE_BUFFER, BUFFER_SIZE, 1);

  std::vector<std::thread> write_threads;
  std::vector<std::thread> attr_threads;
  std::vector<std::thread> read_threads;
  int writer_thread_count = getIntfromEnv("WRITER_THREADS", (int)0);
  int attr_thread_count = getIntfromEnv("ATTR_THREADS", (int)0);
  int reader_thread_count = getIntfromEnv("READER_THREADS", (int)0);
  int wait = getIntfromEnv("WAIT_TIME",(int)300);

  if (writer_thread_count + attr_thread_count + reader_thread_count == 0) {
    std::cerr << "Nothing to do! specify at least of one  WRITER/ATTR/READER_THREADS in env" << std::endl;
    return -1;
  }

  for (int i=0; i < writer_thread_count; ++i) {
    write_threads.emplace_back(write_func, dir, i, prefix);
  }

  for (int i=0; i < attr_thread_count; ++i) {
    attr_threads.emplace_back(attr_func, dir, i, prefix);
  }

  for (int i=0; i < reader_thread_count; ++i) {
    read_threads.emplace_back(read_func, dir, i, prefix);
  }

  std::cerr << "Test will run for " << wait << " seconds" << std::endl;
  std::this_thread::sleep_for(std::chrono::seconds(wait));
  g_shutdown = true;
  for (auto &t: write_threads) {
    t.join();
  }

  for (auto &t: attr_threads) {
    t.join();
  }
  for (auto &t: read_threads) {
    t.join();
  }
  return 0;
}