stencilparallelpattern/main.cpp

/* -*- Mode: C++; tab-width: 4; c-basic-offset: 4; indent-tabs-mode: nil -*- */
#include <algorithm>
#include <cmath>
#include <cstdint>
#include <cstdio>
#include <filesystem>
#include <fstream>
#include <future>
#include <iostream>
#include <memory>
#include <numeric>
#include <thread>
#include <utility>
#include <vector>

#include <ff/ff.hpp>
// important ff.hpp before other ff includes
#include <ff/pipeline.hpp>

#include "reader.hpp"
#include "stencil.hpp"
#include "timer.hpp"
#include "writer.hpp"

using namespace std;
using namespace ff;

long long int fastflow(vector<string> images,
                       vector<pair<int, int>> neighborhood, int iterations,
                       std::function<char(std::vector<char>)> f,
                       int maxworkers) {
    Reader<char> reader(images);
    Stencil<char> stencil(f, neighborhood, iterations, maxworkers);
    Writer<char> writer;

    ff::ff_Pipe<> pipe(reader, stencil, writer);

    UTimer timer;
    timer.start();
    if (pipe.run_and_wait_end() < 0) {
        error("running pipeline\n");
        return 1;
    }
    timer.stop();
    long long int ret = timer.print("\tElapsed time: ");
    return ret;
}

long long int stdthreads(vector<string> images,
                         vector<pair<int, int>> neighborhood, int iterations,
                         std::function<char(std::vector<char>)> f,
                         int maxworkers) {
    long num_images = images.size();

    Reader<char> reader(images);
    Stencil<char> *stencil = // pointer so that we can delete only once and not
                             // when out of scope of the thread
        new Stencil<char>(f, neighborhood, iterations, maxworkers);
    Writer<char> writer;

    std::vector<std::promise<Task<char> *> *> *tv =
        new std::vector<std::promise<Task<char> *> *>();
    std::vector<std::shared_future<Task<char> *>> *in =
        new std::vector<std::shared_future<Task<char> *>>();
    std::vector<std::promise<Task<char> *> *> *out =
        new std::vector<std::promise<Task<char> *> *>();
    std::vector<std::shared_future<Task<char> *>> *tvw =
        new std::vector<std::shared_future<Task<char> *>>();
    for (int i = 0; i < num_images; ++i) {
        std::promise<Task<char> *> *p = new std::promise<Task<char> *>();
        std::future<Task<char> *> f = p->get_future();
        std::promise<Task<char> *> *o = new std::promise<Task<char> *>();
        std::future<Task<char> *> l = o->get_future();
        tv->push_back(p);
        in->push_back(f.share());
        out->push_back(o);
        tvw->push_back(l.share());
    }

    UTimer timer;
    timer.start();
    std::thread readerT(reader, images, tv);
    std::thread stencilT(&Stencil<char>::stdthread, stencil, in, out);
    std::thread writerT(writer, tvw);
    readerT.join();
    stencilT.join();
    writerT.join();
    timer.stop();
    long long int ret = timer.print("\tElapsed time: ");

    for (int i = 0; i < num_images; ++i) {
        delete ((*tv)[i]);
        delete ((*out)[i]);
    }
    delete tv;
    delete in;
    delete out;
    delete tvw;

    delete stencil;
    return ret;
}

long long int sequential(vector<string> images,
                         vector<pair<int, int>> neighborhood, int iterations,
                         std::function<char(std::vector<char>)> f) {
    long num_images = images.size();

    Reader<char> reader(images);
    Stencil<char> *stencil = // pointer so that we can delete only once and not
                             // when out of scope of the thread
        new Stencil<char>(f, neighborhood, iterations);
    Writer<char> writer;

    std::vector<std::promise<Task<char> *> *> *tv =
        new std::vector<std::promise<Task<char> *> *>();
    std::vector<std::shared_future<Task<char> *>> *in =
        new std::vector<std::shared_future<Task<char> *>>();
    std::vector<std::promise<Task<char> *> *> *out =
        new std::vector<std::promise<Task<char> *> *>();
    std::vector<std::shared_future<Task<char> *>> *tvw =
        new std::vector<std::shared_future<Task<char> *>>();
    for (int i = 0; i < num_images; ++i) {
        std::promise<Task<char> *> *p = new std::promise<Task<char> *>();
        std::future<Task<char> *> f = p->get_future();
        std::promise<Task<char> *> *o = new std::promise<Task<char> *>();
        std::future<Task<char> *> l = o->get_future();
        tv->push_back(p);
        in->push_back(f.share());
        out->push_back(o);
        tvw->push_back(l.share());
    }

    UTimer timer;
    timer.start();
    std::thread readerT(reader, images, tv);
    std::thread stencilT(&Stencil<char>::sequential, stencil, in, out);
    std::thread writerT(writer, tvw);
    readerT.join();
    stencilT.join();
    writerT.join();
    timer.stop();
    long long int ret = timer.print("\tElapsed time: ");

    for (int i = 0; i < num_images; ++i) {
        delete ((*tv)[i]);
        delete ((*out)[i]);
    }
    delete tv;
    delete in;
    delete out;
    delete tvw;

    delete stencil;
    return ret;
}

long long int reading_writing(vector<string> images) {
    Reader<char> reader(images);
    Writer<char> writer;

    ff::ff_Pipe<> pipe(reader, writer);

    UTimer timer;
    timer.start();
    if (pipe.run_and_wait_end() < 0) {
        error("running pipeline\n");
        return 1;
    }
    timer.stop();
    long long int ret = timer.print("\tElapsed time: ");
    return ret;
}

// -----------------------------------------------------------------------------
// stencil used for the examples
char gameOfLife(vector<char> in) {
    auto v = std::accumulate(in.begin() + 1, in.end(), 0);
    if (in[0] && v < 2) {
        return 0;
    } else if (in[0] && v > 3) {
        return 0;
    } else if (!in[0] && v != 3) {
        return 0;
    }
    return 1;
}

// -----------------------------------------------------------------------------
int main(int argc, char *argv[]) {
    int average_max = 5;
    int average_max_rw = 15;

    int iter_for_num_workers_images1 = 128;
    vector<string> images1 = {
        "../tests/empty2x2",
        "../tests/increasing4x6",
        "../tests/increasing300x200",
        "../tests/random400x2500"
    };

    int iter_for_num_workers_images2 = 8;
    vector<string> images2 = {
        "../tests/equation",
        "../tests/equation2"
    };

    vector<pair<int, int>> neig = {make_pair(-1, 1),  make_pair(-1, 0),
                                   make_pair(-1, -1), make_pair(0, 1),
                                   make_pair(0, -1),  make_pair(1, 1),
                                   make_pair(1, 0),   make_pair(1, -1)};    

    ofstream csvfile;
    csvfile.open("performance.csv");

    for (std::string image : images1) {
        cout << endl
             << "\033[1;31mProcessing: \t" << image << "\033[0m" << endl;

        csvfile << ",Name,Size(B)\n";
        csvfile << "Image," << image << ",";
        csvfile << std::filesystem::file_size(image) << "\n";

        csvfile << "\n";
        csvfile << "Number of iterations,1,2,4,8,16,32\n";
        csvfile << "fastflow:,";

        cout << "\033[1;31mFastflow\033[0m" << endl;
        for (int iterations : {1, 2, 4, 8, 16, 32}) {
            vector<long long int> results;
            for (int i = 0; i < average_max; ++i) {
                results.push_back(
                    fastflow({image}, neig, iterations, &gameOfLife, 0));
            }
            csvfile << *std::min_element(results.begin(), results.end())
                    << ",";
        }
        csvfile << "\n";

        csvfile << "stdthread:,";
        cout << "\033[1;31mStdthread\033[0m" << endl;
        for (int iterations : {1, 2, 4, 8, 16, 32}) {
            vector<long long int> results;
            for (int i = 0; i < average_max; ++i) {
                results.push_back(
                    stdthreads({image}, neig, iterations, &gameOfLife, 0));
            }
            csvfile << *std::min_element(results.begin(), results.end())
                    << ",";
        }
        csvfile << "\n";

        // ------------------------------------------------------------------ //
        csvfile << "Number of Workers,sequential,fastflow,stdthreads,";
        csvfile << "Iterations:," << iter_for_num_workers_images1 << "\n";
        cout << "\033[1;31mDifferent number of workers\033[0m" << endl;

        int hardware_concurrency = std::thread::hardware_concurrency();

        vector<int> list_max_workers = vector<int>();
        for (int i = 1; i < hardware_concurrency; i*=2) {
            list_max_workers.push_back(i);
        }
        list_max_workers.push_back(hardware_concurrency);

        for (int max_workers : list_max_workers) {
            csvfile << max_workers << ",";

            if(max_workers == 1) {
                cout << "\033[1;31mSequential\033[0m" << endl;
                vector<long long int> results;
                for (int i = 0; i < average_max; ++i) {
                    results.push_back(sequential(
                        {image}, neig, iter_for_num_workers_images1, &gameOfLife));
                }
                csvfile << *std::min_element(results.begin(), results.end());
            }
            csvfile << ",";

            cout << "\033[1;31mFastflow with " << max_workers
                 << " workers\033[0m" << endl;
            {
                vector<long long int> results;
                for (int i = 0; i < average_max; ++i) {
                    results.push_back(fastflow({image}, neig,
                                               iter_for_num_workers_images1,
                                               &gameOfLife, max_workers));
                }
                csvfile << *std::min_element(results.begin(), results.end());
            }
            csvfile << ",";

            cout << "\033[1;31mStdthread with " << max_workers
                 << " workers\033[0m" << endl;
            {
                vector<long long int> results;
                for (int i = 0; i < average_max; ++i) {
                    results.push_back(stdthreads({image}, neig,
                                                 iter_for_num_workers_images1,
                                                 &gameOfLife, max_workers));
                }
                csvfile << *std::min_element(results.begin(), results.end());
            }
            csvfile << "\n";
        }
    }

    for (std::string image : images2) {
        cout << endl
             << "\033[1;31mProcessing: \t" << image << "\033[0m" << endl;

        csvfile << ",Name,Size(B)\n";
        csvfile << "Image," << image << ",";
        csvfile << std::filesystem::file_size(image) << "\n";

        csvfile << "\n";
        csvfile << "Number of iterations,1,2,4,8\n";
        csvfile << "fastflow:,";

        cout << "\033[1;31mFastflow\033[0m" << endl;
        for (int iterations : {1, 2, 4, 8}) {
            vector<long long int> results;
            for (int i = 0; i < average_max; ++i) {
                results.push_back(
                    fastflow({image}, neig, iterations, &gameOfLife, 0));
            }
            csvfile << *std::min_element(results.begin(), results.end())
                    << ",";
        }
        csvfile << "\n";

        csvfile << "stdthread:,";
        cout << "\033[1;31mStdthread\033[0m" << endl;
        for (int iterations : {1, 2, 4, 8}) {
            vector<long long int> results;
            for (int i = 0; i < average_max; ++i) {
                results.push_back(
                    stdthreads({image}, neig, iterations, &gameOfLife, 0));
            }
            csvfile << *std::min_element(results.begin(), results.end())
                    << ",";
        }
        csvfile << "\n";

        // ------------------------------------------------------------------ //
        csvfile << "Number of Workers,sequential,fastflow,stdthreads,";
        csvfile << "Iterations:," << iter_for_num_workers_images2 << "\n";
        cout << "\033[1;31mDifferent number of workers\033[0m" << endl;

        int hardware_concurrency = std::thread::hardware_concurrency();

        vector<int> list_max_workers = vector<int>();
        for (int i = 1; i < hardware_concurrency; i*=2) {
            list_max_workers.push_back(i);
        }
        list_max_workers.push_back(hardware_concurrency);

        for (int max_workers : list_max_workers) {
            csvfile << max_workers << ",";

            if (max_workers == 1) {
                cout << "\033[1;31mSequential\033[0m" << endl;
                vector<long long int> results;
                for (int i = 0; i < average_max; ++i) {
                    results.push_back(sequential(
                                          {image}, neig, iter_for_num_workers_images2, &gameOfLife));
                }
                csvfile << *std::min_element(results.begin(), results.end());
            }
            csvfile << ",";

            cout << "\033[1;31mFastflow with " << max_workers
                 << " workers\033[0m" << endl;
            {
                vector<long long int> results;
                for (int i = 0; i < average_max; ++i) {
                    results.push_back(fastflow({image}, neig,
                                               iter_for_num_workers_images2,
                                               &gameOfLife, max_workers));
                }
                csvfile << *std::min_element(results.begin(), results.end());
            }
            csvfile << ",";

            cout << "\033[1;31mStdthread with " << max_workers
                 << " workers\033[0m" << endl;
            {
                vector<long long int> results;
                for (int i = 0; i < average_max; ++i) {
                    results.push_back(stdthreads({image}, neig,
                                                 iter_for_num_workers_images2,
                                                 &gameOfLife, max_workers));
                }
                csvfile << *std::min_element(results.begin(), results.end());
            }
            csvfile << "\n";
        }
    }

    cout << "Computing reading and writing speeds" << endl;
    for (auto image : images1) {
        cout << endl
             << "\033[1;31mProcessing: \t" << image << "\033[0m" << endl;
        csvfile << "RWtime:," << image << ",";
        vector<long long int> results;
        for (int i = 0; i < average_max_rw; ++i) {
            results.push_back(reading_writing({image}));
        }
        csvfile << *std::min_element(results.begin(), results.end())
                << "\n";
    }
    for (auto image : images2) {
        cout << endl
             << "\033[1;31mProcessing: \t" << image << "\033[0m" << endl;
        csvfile << "RWtime:," << image << ",";
        vector<long long int> results;
        for (int i = 0; i < average_max_rw; ++i) {
            results.push_back(reading_writing({image}));
        }
        csvfile << *std::min_element(results.begin(), results.end())
                << "\n";
    }

    csvfile.close();

    return 0;
}