ParProg2 (8)

#include <agents.h>
#include <cmath>
#include <iostream>
#include <locale.h>
#include <time.h>
using namespace concurrency;
using namespace std;

int const N = 100; //Размерность рядов
int const NNN = 1000; //Длина массива на входе конвейера
int const Chunk = 100; //Порция обрабатываемых элементов

double Func1(double x) // 1-ая ступень конвейера
{
    double Tmp = 0, Tmpa = x * x;

    for (int n = 0; n <= N; n++) {
        double Tmpn4 = n * n * n * n, Tmpxn = sqrt(Tmpa + Tmpn4);

        for (int k = 0; k <= N; k++) {
            double Tmpk = k * k;

            for (int j = 0; j <= N; j++)
                Tmp += (x + j) / (1 + Tmpxn + Tmpk + j * j * j);
        }
    }

    return Tmp;
}

double Func2(double x) // 2-ая ступень конвейера
{
    double Tmp = 0, Tmpa = x * x;

    for (int n = 0; n <= N; n++) {
        double Tmpn = n * n * n;

        for (int k = 0; k <= N; k++) {
            double Tmpk = k * k * k;

            for (int j = 0; j <= N; j++)
                Tmp += (x * x - k) / (1 + Tmpa + Tmpn + Tmpk + j * j);
        }
    }

    return x * Tmp;
}

class Stage_Agent : public agent // Отдельная ступень конвейера
{

private:
    ISource<double>& _source; // Буфер - источник информации
    ITarget<double>& _target; // Буфер - приемник информации
    double(*_func) (double); //Функция для обработки

public:

    //Конструктор агента
    Stage_Agent(ISource<double>& source, ITarget<double>& target, double func(double)) : _target(target), _source(source)
    {
        _func = func;
    }

protected:
    void run() //Выполняемая агентом работа
    {
        while (true) { asend(_target, _func(receive(_source))); }
    }
};

int main()
{
    setlocale(LC_ALL, "Russian");

    vector<double> X(NNN), Y(NNN), Z(NNN);

    for (int k = 0; k < X.size(); k++) //Заполнение входного вектора
        X[k] = 10 * sin((double)k);

    double Tms = clock();

    for (int k = 0; k < X.size(); k++) { //Последовательные вычисления
        Y[k] = Func1(X[k]); Y[k] = Func2(Y[k]);
    }

    Tms = (clock() - Tms) / CLOCKS_PER_SEC;
    cout << "Время последовательного алгоритма: " << Tms << " c." << endl;

    //Буферы между ступенями конвейера
    unbounded_buffer<double> buf0, buf1, buf2;

    //Первая ступень конвейера
    Stage_Agent Stage_Agent1(buf0, buf1, Func1);

    //Вторая ступень конвейера
    Stage_Agent Stage_Agent2(buf1, buf2, Func2);

    Stage_Agent1.start(); Stage_Agent2.start();// Запуск

    Tms = clock();
    for (int k = 0; k < Chunk; k++) // Параллельные вычисления
        send(buf0, X[k]);

    for (int k = Chunk; k < X.size(); k++) {
        Z[k - Chunk] = receive(buf2); send(buf0, X[k]);
    }

    for (int k = X.size() - Chunk; k < X.size(); k++) {
        Z[k] = receive(buf2);
    }

    Tms = (clock() - Tms) / CLOCKS_PER_SEC;
    cout << "Время параллельного алгоритма: " << Tms << " c." << endl;
}