Govno

// Lab2.cpp : This file contains the 'main' function. Program execution begins and ends there.
//

#include <iostream>
#include <pthread.h>
#include <fstream>
#include <string>
#include <vector>
#include <queue>
#include <Windows.h>
#include <chrono>

#pragma comment (lib, "pthreadVCE2.lib")
#pragma warning (disable : 4996)

pthread_mutex_t mutex;
pthread_mutex_t end_mutex;
bool end = false;

auto factorize(const int number) -> std::vector<int>
{
    auto result = std::vector<int>();
    auto cur_number = number;
    auto probe = 2;

    while (cur_number != 1)
    {
        if (cur_number % probe != 0)
            probe++;
        else
        {
            cur_number /= probe;
            result.push_back(probe);
        }
    }

    return result;
}

auto string_to_char_array(const std::string& source) -> char*
{
    const auto destination = new char[source.size() + 1];
    source.copy(destination, source.size() + 1);
    destination[source.size()] = '\0';

    return destination;
}

auto array_to_vector(const int* buffer, const int size) -> std::vector<int>
{
    std::vector<int> vec(size);

    for (auto i = 0; i < size; i++)
    {
        vec[i] = *(buffer + i);
    }

    return vec;
}

typedef struct
{
    std::queue<std::vector<int>>* q;
    long tid;
    pthread_mutex_t* queue_mutex;
} evaluator_params;

auto write_numbers(const std::vector<int>& numbers, std::ofstream& stream) -> void
{

    for (auto n : numbers)
    {
        stream << n << " = ";

        auto result = factorize(n);
        for (auto number : result)
        {
            stream << number << " * ";
        }

        stream << "\n";
    }
}

auto evaluate(void* params) -> void*
{
    auto p = static_cast<evaluator_params*>(params);
    auto q = p->q;
    auto path = "utils/output-" + std::to_string(p->tid) + ".txt";
    auto mutex = p->queue_mutex;
    std::ofstream stream(path);

    while (true)
    {
        if (q->empty())
        {
            if (end)
                return nullptr;

            continue;
        }

        pthread_mutex_lock(mutex);
        auto numbers = q->front();
        q->pop();
        pthread_mutex_unlock(mutex);

        write_numbers(numbers, stream);
    }
}

struct reader_params_t
{
    FILE* f;
    int batch_size;
    int index;
};

auto read(void* params) -> void*
{
    // const auto p = static_cast<reader_params_t*>(params);
    const auto p = (reader_params_t*)params;
    int* buffer = new int[p->batch_size];
    std::queue<std::vector<int>> q;
    pthread_t tid;
    pthread_mutex_t queue_mutex;

    pthread_mutex_init(&queue_mutex, nullptr);

    evaluator_params eval_params{ &q, p->index, &queue_mutex };

    pthread_create(&tid, nullptr, evaluate, &eval_params);

    while (true)
    {
        pthread_mutex_lock(&mutex);
        const auto read_count = fread(buffer, sizeof(int), p->batch_size, p->f);
        pthread_mutex_unlock(&mutex);

        auto numbers = array_to_vector(buffer, p->batch_size);

        q.push(numbers);

        if (feof(p->f))
            break;
    }

    pthread_mutex_lock(&end_mutex);
    end = true;
    pthread_mutex_unlock(&end_mutex);

    pthread_join(tid, nullptr);
    return nullptr;
}

class pair
{
private:
    FILE* f;
    pthread_t tid;
    int index;
public:
    auto start(const int batch_size) -> void
    {
        auto p = new reader_params_t();
        p->f = f;
        p->batch_size = batch_size;
        p->index = index;

        pthread_create(&tid, nullptr, read, p);
    }
    auto join() -> void
    {
        pthread_join(tid, nullptr);
    }
    auto set_file(FILE** f) -> void
    {
        pair::f = *f;
    }
    auto set_index(int i)
    {
        index = i;
    }
};


auto test(int thread_count) -> void
{
    auto f = fopen("data.bin", "rb");

    const auto pairs = new pair[thread_count];

    pthread_mutex_init(&mutex, nullptr);
    pthread_mutex_init(&end_mutex, nullptr);
    const auto before = std::chrono::system_clock::now();
    for (auto i = 0; i < thread_count; i++)
    {
        pairs[i].set_file(&f);
        pairs[i].set_index(i);
        pairs[i].start(50);
    }

    for (auto i = 0; i < thread_count; i++)
    {
        pairs[i].join();
    }
    const auto after = std::chrono::system_clock::now();
    const auto elapsed = std::chrono::duration_cast<std::chrono::milliseconds>(after - before).count();
    pthread_mutex_destroy(&mutex);
    pthread_mutex_init(&end_mutex, nullptr);

    printf("Number of threads: %d, Elapsed: %lld\n", thread_count, elapsed);
}

auto main() -> int
{
    for (auto i = 1; i <= 16; i *= 2)
    {
        test(i);
    }

    return 0;
}