fileopstest.cpp File Reference

(v0.19-24-g0617ca1 with changes)

#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>

Include dependency graph for fileopstest.cpp:

Go to the source code of this file.

Classes
class	testFile
class	buffqueue
class	buffstuff
class	timeRecords
class	fileSizeGenerator
class	statCollectionBase< TC >
class	dummyMember
class	statCollection< T, TC >
class	timeSpan

Typedefs
typedef std::chrono::system_clock	clock_type

Functions
testFile *	createFile (const char *workdir, unsigned int index, buffstuff &buffer, size_t size, std::chrono::system_clock::time_point &stat_time)
void	readAndDeleteFile (testFile *f, buffstuff &buffer)
void	bunch_wait (long int bunchTime)
void	creator (buffqueue *queue, const char *workdir, unsigned int repeat, unsigned long blockSize, unsigned int maxQueueDepth)
void	consumer (buffqueue *queue, buffqueue *results, unsigned long blockSize)
template<typename T , typename TC , class ... Types>
statCollection< T, TC > *	statCollectionFactory (T TC::*member, Types...args)
void	collectStats (buffqueue *results)
int	main (int argc, const char *argv[])

Variables
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)
static timeRecords	creationRecords
static timeRecords	mtimeRecords
static timeRecords	consumptionRecords
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)

Typedef Documentation

clock_type

typedef std::chrono::system_clock clock_type

Definition at line 27 of file fileopstest.cpp.

Function Documentation

bunch_wait()

void bunch_wait

(

long int

bunchTime

)

Definition at line 164 of file fileopstest.cpp.

                                    {
    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);
    }
}

References bunchTime.

Referenced by consumer(), and creator().

Here is the caller graph for this function:

collectStats()

void collectStats

(

buffqueue *

results

)

Definition at line 460 of file fileopstest.cpp.

                                      {
    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";
}

References statCollection< T, TC >::addValue(), testFile::bytes, testFile::deltaTClose1, testFile::deltaTClose2, testFile::deltaTCreate, testFile::deltaTOpen, testFile::deltaTQueue, testFile::deltaTRead, testFile::deltaTUnlink, testFile::deltaTWrite, buffqueue::dequeue(), f(), timeSpan::span(), statCollectionFactory(), and timeSpan::update().

Referenced by main().

Here is the call graph for this function:

Here is the caller graph for this function:

consumer()

void consumer	(	buffqueue *	queue,
		buffqueue *	results,
		unsigned long	blockSize
	)

Definition at line 317 of file fileopstest.cpp.

                                                                             {
    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());
}

References bunch_wait(), bunchStart, bunchTime, consumptionRecords, buffqueue::dequeue(), buffqueue::enqueue(), f(), readAndDeleteFile(), and timeRecords::record().

Referenced by main().

Here is the call graph for this function:

Here is the caller graph for this function:

createFile()

testFile* createFile	(	const char *	workdir,
		unsigned int	index,
		buffstuff &	buffer,
		size_t	size,
		std::chrono::system_clock::time_point &	stat_time
	)

Definition at line 104 of file fileopstest.cpp.

                                                                                                                                          {
    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;
}

References throwcall::badval(), f(), throwcall::good0(), and buffstuff::writeBuffer.

Referenced by creator().

Here is the call graph for this function:

Here is the caller graph for this function:

creator()

void creator	(	buffqueue *	queue,
		const char *	workdir,
		unsigned int	repeat,
		unsigned long	blockSize,
		unsigned int	maxQueueDepth
	)

Definition at line 290 of file fileopstest.cpp.

                                                                                                                              {
    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());
}

References bunch_wait(), bunchStart, bunchTime, createFile(), creationRecords, distribution, buffqueue::enqueue(), f(), fileSizeGenerator::getNumber(), mtimeRecords, parameter1, parameter2, timeRecords::record(), and buffqueue::size().

Referenced by main().

Here is the call graph for this function:

Here is the caller graph for this function:

main()

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

Definition at line 562 of file fileopstest.cpp.

                                       {
    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;
}

References collectStats(), consumer(), consumptionRecords, creationRecords, creator(), buffqueue::endOfInput(), options::single< T >::fGetValue(), throwcall::good0(), mtimeRecords, and timeRecords::print().

Here is the call graph for this function:

readAndDeleteFile()

void readAndDeleteFile	(	testFile *	f,
		buffstuff &	buffer
	)

Definition at line 135 of file fileopstest.cpp.

                                                       {
    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;
}

References throwcall::badval(), f(), throwcall::good0(), buffstuff::readBuffer, and buffstuff::writeBuffer.

Referenced by consumer().

Here is the call graph for this function:

Here is the caller graph for this function:

statCollectionFactory()

template<typename T , typename TC , class ... Types>

statCollection<T, TC>* statCollectionFactory	(	T TC::*	member,
		Types...	args
	)

Definition at line 425 of file fileopstest.cpp.

                                                  {
    return new typename statCollection<T, TC>::statCollection(member, args...);
}

References statCollection< T, TC >::statCollection().

Referenced by collectStats().

Here is the call graph for this function:

Here is the caller graph for this function:

Variable Documentation

bunchStart

options::single<int> bunchStart( 's', "startTime", "bunching time for starts (seconds)", 0)

static

Referenced by consumer(), and creator().

bunchTime

options::single<int> bunchTime( 'B', "bunchTime", "bunching time for requests (seconds)", 0)

static

Referenced by bunch_wait(), consumer(), and creator().

consumptionRecords

timeRecords consumptionRecords

static

Definition at line 205 of file fileopstest.cpp.

Referenced by consumer(), and main().

creationRecords

timeRecords creationRecords

static

Definition at line 203 of file fileopstest.cpp.

Referenced by creator(), and main().

distribution

options::single<std::string> distribution( 'd', "distribution", "distribution type", "constant", {"constant", "uniform", "geometric", "gauss"})

static

Referenced by creator(), and inputHandler::dummy::DummyDirectory::getNextEntry().

mtimeRecords

timeRecords mtimeRecords

static

Definition at line 204 of file fileopstest.cpp.

Referenced by creator(), and main().

parameter1

options::single<size_t> parameter1('\0', "parameter1", "1st distribution paramater, avg/min file size", 1024)

static

Referenced by creator().

parameter2

options::single<size_t> parameter2('\0', "parameter2", "2nd distribution paramater, sigma/max file size", 2048)

static

Referenced by creator().