fileopstest.cpp

Go to the documentation of this file.

#include <thread>
#include <mutex>
#include <condition_variable>
#include <string>
#include <vector>
#include <deque>
#include <iostream>
#include <iomanip>
#include <system_error>
#include <sstream>
#include <chrono>
#include <algorithm>
#include <random>
#include <map>
#include <cstdlib>
#include <list>
 
#include <Options.h>
#include "throwcall.h"
 
#include <sys/types.h>
#include <sys/stat.h>
#include <sys/statvfs.h>
#include <fcntl.h>
#include <unistd.h>
#include <string.h>
typedef std::chrono::system_clock clock_type;
//typedef std::chrono::high_resolution_clock clock_type;
 
 
 
 
class testFile {
  public:
    std::string name;
    clock_type::duration deltaTCreate;
    clock_type::duration deltaTWrite;
    clock_type::duration deltaTClose1;
    clock_type::duration deltaTOpen;
    clock_type::duration deltaTRead;
    clock_type::duration deltaTClose2;
    clock_type::duration deltaTUnlink;
    clock_type::duration deltaTQueue;
    clock_type::time_point tCreate;
    clock_type::time_point tOpen;
    clock_type::time_point tClose1;
    clock_type::time_point tClose2;
    size_t bytes;
};
 
class buffqueue {
  private:
    std::deque<testFile*> queue;
    std::mutex queue_mutex;
    std::condition_variable cv;
    bool expectingInput;
  public:
    buffqueue(): expectingInput(true) {};
    void enqueue(testFile *f) {
        std::unique_lock<decltype(queue_mutex)> lock(queue_mutex);
        queue.emplace_back(f);
        cv.notify_all();
    }
    testFile* dequeue() {
        std::unique_lock<decltype(queue_mutex)> lock(queue_mutex);
 
        while (queue.empty()) {
            if (!expectingInput) {
                return nullptr;
            }
 
            cv.wait(lock);
        }
 
        auto f = queue.front();
        queue.pop_front();
        return f;
    }
    decltype(queue.size()) size() {
        return queue.size();
    }
    void endOfInput() {
        expectingInput = false;
        cv.notify_all();
    }
};
 
 
class buffstuff {
  public:
    buffstuff(unsigned int aBlockSize): blockSize(aBlockSize) {
        writeBuffer.reserve(blockSize / sizeof(int));
        readBuffer.reserve(blockSize / sizeof(int));
 
        for (unsigned int i = 0; i < writeBuffer.capacity(); i++) {
            writeBuffer.emplace_back(~writeBuffer.capacity() - i);
        }
    }
    std::vector<int> writeBuffer;
    decltype(writeBuffer) readBuffer;
    unsigned int blockSize;
};
 
testFile* createFile(const char *workdir, unsigned int index, buffstuff& buffer, size_t size, std::chrono::system_clock::time_point& stat_time) {
    testFile* f = new testFile();
    std::ostringstream name;
    name << workdir
         << "/testfile_"
         << std::this_thread::get_id()
         << "_"
         << index;
    f->name = name.str();
    f->bytes = size;
    f->tCreate = clock_type::now();
    auto fd = throwcall::badval(open(f->name.c_str(), O_CREAT | O_WRONLY, 0755), -1,
                                "can't open ", f->name);
    auto now = clock_type::now();
    f->deltaTCreate = now - f->tCreate;
    {
        struct stat newstat;
        throwcall::good0(fstat(fd, &newstat), "can't stat file");
        stat_time = std::chrono::time_point<std::chrono::system_clock>(std::chrono::nanoseconds(newstat.st_mtim.tv_sec * 1000000000 + newstat.st_mtim.tv_nsec));
    }
    auto before = clock_type::now();
    throwcall::badval(write(fd, buffer.writeBuffer.data(), f->bytes), -1,
                      "write ", f->name);
    now = clock_type::now();
    f->deltaTWrite = now - before;
    before = now;
    throwcall::good0(close(fd), "can't close ", f->name);
    f->tClose1 = clock_type::now();
    f->deltaTClose1 = f->tClose1 - before;
    return f;
}
void readAndDeleteFile(testFile *f, buffstuff& buffer) {
    f->tOpen = clock_type::now();
    auto fd = throwcall::badval(open(f->name.c_str(), O_RDONLY), -1,
                                "can't open fotr read", f->name);
    auto now = clock_type::now();
    f->deltaTOpen = now - f->tOpen;
    f->deltaTQueue = f->tOpen - f->tClose1;
    auto before = now;
    throwcall::badval(read(fd, buffer.readBuffer.data(), f->bytes), -1,
                      "can't read from ", f->name);
    now = clock_type::now();
    f->deltaTRead = now - before;
 
    if (memcmp(buffer.writeBuffer.data(), buffer.readBuffer.data(), f->bytes) != 0) {
        throw std::runtime_error("values in file " + f->name);
    }
 
    before = clock_type::now();
    throwcall::good0(close(fd), "can't close after reading ", f->name);
    now = clock_type::now();
    f->tClose2 = now;
    f->deltaTClose2 = now - before;
    before = now;
    throwcall::good0(unlink(f->name.c_str()), "can't unlink ", f->name);
    now = clock_type::now();
    f->deltaTUnlink = now - before;
}
 
 
void bunch_wait(long int bunchTime) {
    if (bunchTime > 0) {
        auto now = std::chrono::system_clock::now();
        auto seconds = std::chrono::duration_cast<std::chrono::seconds>(now.time_since_epoch()).count();
        auto divresult = std::div(seconds, bunchTime);
        auto then = std::chrono::system_clock::time_point(std::chrono::seconds(divresult.quot * bunchTime + bunchTime));
        std::this_thread::sleep_until(then);
    }
}
 
static options::single<int>bunchTime('B', "bunchTime", "bunching time for requests (seconds)", 0);
static options::single<int>bunchStart('s', "startTime", "bunching time for starts (seconds)", 0);
 
class timeRecords {
    typedef std::chrono::time_point<std::chrono::system_clock, std::chrono::milliseconds> mapTimeType;
    std::mutex recordMutex;
    std::map<mapTimeType, unsigned int> timeMap;
  public:
    template <typename InputIt> void record(InputIt begin, InputIt end) {
        for (auto it = begin; it != end; ++it) {
            record(*it);
        }
    }
    void record(std::chrono::system_clock::time_point when) {
        std::unique_lock<decltype(recordMutex)> lock(recordMutex);
        ++(timeMap[std::chrono::time_point_cast<mapTimeType::duration>(when)]);
    }
    void print(const std::string& prefix) {
        for (const auto& it : timeMap) {
            auto then = std::chrono::system_clock::to_time_t(it.first);
            auto milliseconds = std::chrono::duration_cast<std::chrono::milliseconds>(it.first.time_since_epoch()).count() % 1000;
            std::cout << prefix <<
                      //std::put_time(std::localtime(&then), "%F %T")
                      then
                      << "." << std::setw(3) << std::setfill('0') << milliseconds << "   " << it.second << "\n";
        }
    }
};
 
static timeRecords creationRecords;
static timeRecords mtimeRecords;
static timeRecords consumptionRecords;
 
 
class fileSizeGenerator {
  public:
    enum distributionType {
        constant,
        uniform,
        geometric,
        gauss
    };
  private:
    std::mt19937 generator;
    std::geometric_distribution<size_t> geometric_dist;
    std::uniform_int_distribution<size_t> uniform_dist;
    std::normal_distribution<double> normal_dist;
    distributionType distribution;
    size_t parameter1;
    size_t parameter2;
    size_t maxFileSize;
  public:
    fileSizeGenerator(const std::string& aDistribution,
                      size_t size_parameter1,
                      size_t size_parameter2,
                      size_t max_size) :
        parameter1(size_parameter1),
        parameter2(size_parameter2),
        maxFileSize(max_size) {
        static std::random_device rd;
        generator.seed(rd());
 
        if (aDistribution == "constant") {
            distribution = constant;
        } else if (aDistribution == "uniform") {
            distribution = uniform;
            uniform_dist.param(decltype(uniform_dist)::param_type(parameter1, parameter2));
        } else if (aDistribution == "geometric") {
            distribution = geometric;
            geometric_dist.param(decltype(geometric_dist)::param_type(1. / parameter1));
        } else if (aDistribution == "gauss") {
            distribution = gauss;
            normal_dist.param(decltype(normal_dist)::param_type(static_cast<double>(parameter1),
                              static_cast<double>(parameter2)));
        } else {
            throw std::runtime_error("bad distribution parameter");
        }
    }
    size_t getNumber() {
        size_t result;
 
        switch (distribution) {
            case constant:
                result = parameter1;
                break;
 
            case uniform:
                result = uniform_dist(generator);
                break;
 
            case geometric:
                result = geometric_dist(generator);
                break;
 
            case gauss:
                result = normal_dist(generator);
                break;
 
            default:
                return 0;
        }
 
        if (result > maxFileSize) {
            result = maxFileSize;
        }
 
        return result;
    }
};
 
static options::single<std::string> distribution('d', "distribution", "distribution type", "constant",
{"constant", "uniform", "geometric", "gauss"});
static options::single<size_t> parameter1('\0', "parameter1", "1st distribution paramater, avg/min file size", 1024);
static options::single<size_t> parameter2('\0', "parameter2", "2nd distribution paramater, sigma/max file size", 2048);
 
 
void creator(buffqueue* queue, const char *workdir, unsigned int repeat, unsigned long blockSize, unsigned int maxQueueDepth) {
    buffstuff buffer(blockSize);
    fileSizeGenerator generator(distribution, parameter1, parameter2, blockSize);
 
    std::list<std::chrono::system_clock::time_point> startTimes;
    std::list<std::chrono::system_clock::time_point> statTimes;
 
    bunch_wait(bunchStart);
 
    for (decltype(repeat) index = 0; index < repeat; index++) {
        while (queue->size() > maxQueueDepth) {
            std::this_thread::yield();
        }
 
        auto size = generator.getNumber();
        bunch_wait(bunchTime);
        startTimes.emplace_back(std::chrono::system_clock::now());
        std::chrono::system_clock::time_point stat_time;
        auto f = createFile(workdir, index, buffer, size, stat_time);
        statTimes.emplace_back(stat_time);
        queue->enqueue(f);
    }
 
    creationRecords.record(startTimes.begin(), startTimes.end());
    mtimeRecords.record(statTimes.begin(), statTimes.end());
}
 
void consumer(buffqueue* queue, buffqueue* results, unsigned long blockSize) {
    buffstuff buffer(blockSize);
    std::list<std::chrono::system_clock::time_point> startTimes;
 
    bunch_wait(bunchStart);
 
    while (true) {
        auto f = queue->dequeue();
 
        if (f == nullptr) {
            break;
        }
 
        bunch_wait(bunchTime);
        startTimes.emplace_back(std::chrono::system_clock::now());
        readAndDeleteFile(f, buffer);
        results->enqueue(f);
    }
 
    consumptionRecords.record(startTimes.begin(), startTimes.end());
}
template <typename TC> class statCollectionBase {
  private:
    std::string name;
    std::string unit;
  public:
    statCollectionBase(const std::string& aName,
                       const std::string& aUnit = "s") : name(aName), unit(aUnit) {};
    virtual void print(unsigned long n) = 0;
    virtual void add(TC* /*what*/) {};
    const std::string& getName() const {
        return name;
    };
    const std::string& getUnit() const {
        return unit;
    };
};
class dummyMember {};
template <typename T, typename TC = testFile> class statCollection: public statCollectionBase<TC> {
  public:
    typename std::conditional<std::is_void<TC>::value, dummyMember, T TC::*>::type sourceMember;
    T sum;
    T min;
    T max;
    void addValue(T value) {
        sum += value;
 
        if (value < min) {
            min = value;
        }
 
        if (value > max) {
            max = value;
        }
    };
    void add(TC* what) {
        addValue(what->*sourceMember);
    }
    template < typename TT = T, typename TCC = TC, typename std::enable_if < std::is_arithmetic<TT>::value && ! std::is_void<TCC>::value >::type..., class ... Types >
    statCollection(decltype(sourceMember) aSourceMember, Types ... args) :
        statCollectionBase<TC>(args...),
        sourceMember(aSourceMember),
        sum(0),
        min(std::numeric_limits<T>::max()),
        max(std::numeric_limits<T>::lowest()) {
    };
    template <typename TT = T, typename std::enable_if <std::is_arithmetic<TT>::value>::type..., class ... Types>
    statCollection(Types ... args) :
        statCollectionBase<TC>(args...),
        sum(0),
        min(std::numeric_limits<T>::max()),
        max(std::numeric_limits<T>::lowest()) {
    };
    template < typename TT = T, typename std::enable_if < !std::is_arithmetic<TT>::value >::type..., class ... Types >
    statCollection(decltype(sourceMember) aSourceMember, Types ... args) :
        statCollectionBase<TC>(args...),
        sourceMember(aSourceMember),
        sum(T::zero()),
        min(T::max()),
        max(T::min()) {
    };
    template<typename TT = T>
    typename std::enable_if < !std::is_arithmetic<TT>::value >::type
    _print(unsigned long n) {
        std::cout << this->getName() << " avg: " << std::scientific
                  << std::chrono::duration_cast<std::chrono::duration<double>>(sum / n).count()
                  << this->getUnit() << ",  min: "
                  << std::chrono::duration_cast<std::chrono::duration<double>>(min).count()
                  << this->getUnit() << ",  max: "
                  << std::chrono::duration_cast<std::chrono::duration<double>>(max).count()
                  << this->getUnit() << "\n";
    }
    template<typename TT = T>
    typename std::enable_if <std::is_arithmetic<TT>::value>::type
    _print(unsigned long n) {
        std::cout << this->getName() << " avg: "
                  << sum / n
                  << this->getUnit() << ",  min: "
                  << min
                  << this->getUnit() << ",  max: "
                  << max
                  << this->getUnit() << "\n";
    }
    void print(unsigned long n) override {
        _print(n);
    }
};
template <typename T, typename TC, class ... Types> statCollection<T, TC>*
statCollectionFactory(T TC::* member, Types...args) {
    return new typename statCollection<T, TC>::statCollection(member, args...);
}
 
class timeSpan {
    clock_type::time_point start;
    clock_type::time_point stop;
  public:
    timeSpan(): start(clock_type::time_point::max()),
        stop(clock_type::time_point::min()) {
    }
    void update(clock_type::time_point aStart,
                clock_type::time_point aStop) {
        if (aStart < start) {
            start = aStart;
        }
 
        if (aStop > stop) {
            stop = aStop;
        }
    }
    void update(clock_type::time_point aTime) {
        if (aTime < start) {
            start = aTime;
        }
 
        if (aTime > stop) {
            stop = aTime;
        }
    }
    std::chrono::duration<double> span() {
        return std::chrono::duration_cast<std::chrono::duration<double>>(stop - start);
    }
};
 
void collectStats(buffqueue* results) {
    unsigned long n(0);
    statCollection<double>                                writeSpeed("write speed  ", "MiB/s");
    statCollection<double>                                 readSpeed("read speed   ", "MiB/s");
    std::vector<statCollectionBase<testFile>*> stats;
    stats.emplace_back(statCollectionFactory(&testFile::deltaTCreate, "File creation"));
    stats.emplace_back(statCollectionFactory(&testFile::deltaTWrite, "Write to file"));
    stats.emplace_back(statCollectionFactory(&testFile::deltaTClose1, "Close file 1 "));
    stats.emplace_back(statCollectionFactory(&testFile::deltaTOpen,  "Open for read"));
    stats.emplace_back(statCollectionFactory(&testFile::deltaTRead,  "Read fr. file"));
    stats.emplace_back(statCollectionFactory(&testFile::deltaTClose2, "Close file 2 "));
    stats.emplace_back(statCollectionFactory(&testFile::deltaTUnlink, "Unlink file  "));
    stats.emplace_back(statCollectionFactory(&testFile::deltaTQueue, "Wait wr/rd   "));
    auto fileSize = statCollectionFactory(&testFile::bytes,          "file size    ", "Bytes");
    stats.emplace_back(fileSize);
 
    timeSpan tWrite;
    timeSpan tRead;
    timeSpan tCreations;
    timeSpan tOpens;
 
 
    std::map<size_t, unsigned int> sizeHisto;
 
    while (true) {
        auto f = results->dequeue();
 
        if (f == nullptr) {
            break;
        }
 
        n++;
        tWrite.update(f->tCreate, f->tClose1);
        tRead.update(f->tOpen, f->tClose2);
        tCreations.update(f->tCreate);
        tOpens.update(f->tOpen);
        {
            unsigned int i = 1;
 
            while (i < f->bytes) {
                i <<= 1;
            }
 
            ++sizeHisto[i];
        }
 
        for (auto stat : stats) {
            stat->add(f);
        }
 
        writeSpeed.addValue(f->bytes / (1024 * 1024 *
                                        std::chrono::duration_cast<std::chrono::duration<double>>(f->deltaTWrite).count()));
        readSpeed.addValue(f->bytes / (1024 * 1024 *
                                       std::chrono::duration_cast<std::chrono::duration<double>>(f->deltaTRead).count()));
        /*
          std::cout << f->name
          << " cr: " << std::chrono::duration_cast<std::chrono::nanoseconds>(f->deltaTCreate).count()
          << " wr: " << std::chrono::duration_cast<std::chrono::nanoseconds>(f->deltaTWrite).count()
          << " cl: " << std::chrono::duration_cast<std::chrono::nanoseconds>(f->deltaTClose1).count()
          << " op: " << std::chrono::duration_cast<std::chrono::nanoseconds>(f->deltaTOpen).count()
          << " rd: " << std::chrono::duration_cast<std::chrono::nanoseconds>(f->deltaTRead).count()
          << " cl: " << std::chrono::duration_cast<std::chrono::nanoseconds>(f->deltaTClose2).count()
          << " rm: " << std::chrono::duration_cast<std::chrono::nanoseconds>(f->deltaTUnlink).count()
          << " qu: " << std::chrono::duration_cast<std::chrono::nanoseconds>(f->deltaTQueue).count()
          << "\n";
        */
    }
 
    std::cout << "file sizes in bytes:\n";
    size_t from = 0;
 
    for (auto bin : sizeHisto) {
        std::cout << std::setw(10) << std::right << from << " to " << bin.first << "\t" << bin.second << "\n";
        from = bin.first;
    }
 
    stats.push_back(&writeSpeed);
    stats.push_back(&readSpeed);
 
    for (auto stat : stats) {
        stat->print(n);
    }
 
    auto deltaTWrite = tWrite.span().count();
    std::cout << "total time spent writing: "
              << deltaTWrite
              << "s, with "
              << fileSize->sum / (1024 * 1024 * deltaTWrite)
              << "MB/s \n";
    auto deltaTRead = tRead.span().count();
    std::cout << "total time spent reading: "
              << deltaTRead
              << "s, with "
              << fileSize->sum / (1024 * 1024 * deltaTRead)
              << "MB/s \n";
    auto deltaTCreate = tCreations.span().count();
    auto deltaTOpen = tOpens.span().count();
    std::cout << n << " files created in " << deltaTCreate << "s: " << n / deltaTCreate << "Creations/s\n";
    std::cout << n << " files opene   in " << deltaTOpen << "s: " << n / deltaTOpen << "Opens/s\n";
}
 
 
int main(int argc, const char *argv[]) {
    options::parser parser("fileopstest", "", {});
    options::single<unsigned>nCreators('w', "nCreators", "number of creator threads", sysconf(_SC_NPROCESSORS_ONLN));
    options::single<unsigned>nConsumers('r', "nConsumers", "number of consumer threads", sysconf(_SC_NPROCESSORS_ONLN));
    options::single<unsigned>nRepeat('c', "nRepeat", "number of files per thread", 1);
    options::single<unsigned>maxQueueDepth('q', "maxQueueDepth", "maximum depth of file queue", 100);
    options::positional<options::single<std::string>>workdir(10, "workdir", "directory to work in");
    options::single<size_t>blockSize('b', "blockSize", "block size: 0 fs native block size", 0);
    options::single<int>consumerDelay('D', "consumerDelay", "consumer delay", 0);
 
    parser.fParse(argc, argv);
 
    if (blockSize == 0) {
        struct statvfs statbuf;
        throwcall::good0(statvfs(workdir.c_str(), &statbuf), "can't stat workdir fs");
        size_t& bla = blockSize;
        bla = statbuf.f_bsize;
    }
 
    std::cout << "Will create " << nRepeat << " files in " << nCreators << " threads with " << blockSize << std::hex << " (0x" << blockSize << ", " << std::dec << blockSize / 1024 << "k, " << blockSize / (1024 * 1024) <<  "M) blocks," << std::endl;
    std::cout << "Consuming them with " << nConsumers << " threads, the queue should not grow above " << maxQueueDepth << std::endl;
 
    buffqueue queue;
    buffqueue results;
    std::vector<std::thread> creators;
    std::vector<std::thread> consumers;
 
    for (unsigned index = 0; index < nCreators; index++) {
        creators.emplace_back(std::thread(creator, &queue, workdir.c_str(), nRepeat, blockSize, maxQueueDepth));
    }
 
    std::this_thread::sleep_for(std::chrono::seconds(consumerDelay.fGetValue()));
 
    for (unsigned index = 0; index < nConsumers; index++) {
        consumers.emplace_back(std::thread(consumer, &queue, &results, blockSize));
    }
 
    auto collector = std::thread(collectStats, &results);
 
    for (auto & tr : creators) {
        tr.join();
    }
 
    queue.endOfInput();
 
    for (auto & tr : consumers) {
        tr.join();
    }
 
    results.endOfInput();
    collector.join();
    std::cout << "creation times:\n";
    creationRecords.print("file creation time: ");
    std::cout << "\nstat times:\n";
    mtimeRecords.print("file stat time: ");
    std::cout << "\nconsumption times:\n";
    consumptionRecords.print("file consumption time: ");
    return 0;
}