【问题标题】:Multithreading computation of Mandelbrot setMandelbrot集的多线程计算
【发布时间】:2015-05-06 07:35:35
【问题描述】:

我创建了一个创建Mandelbrot 集的程序。现在我正在尝试使其成为多线程。

// mandelbrot.cpp
// compile with: g++ -std=c++11 mandelbrot.cpp -o mandelbrot
// view output with: eog mandelbrot.ppm

#include <fstream>
#include <complex> // if you make use of complex number facilities in C++
#include <iostream>
#include <cstdlib>
#include <thread>
#include <mutex>
#include <vector>


using namespace std;

template <class T> struct RGB { T r, g, b; };

template <class T>
class Matrix {
public:
Matrix(const size_t rows, const size_t cols) : _rows(rows), _cols(cols) {
    _matrix = new T*[rows];
    for (size_t i = 0; i < rows; ++i) {
        _matrix[i] = new T[cols];
    }
}
Matrix(const Matrix &m) : _rows(m._rows), _cols(m._cols) {
    _matrix = new T*[m._rows];
    for (size_t i = 0; i < m._rows; ++i) {
        _matrix[i] = new T[m._cols];
        for (size_t j = 0; j < m._cols; ++j) {
            _matrix[i][j] = m._matrix[i][j];
        }
    }
}
~Matrix() {
    for (size_t i = 0; i < _rows; ++i) {
        delete [] _matrix[i];
    }
    delete [] _matrix;
}
T *operator[] (const size_t nIndex)
{
    return _matrix[nIndex];
}
size_t width() const { return _cols; }
size_t height() const { return _rows; }
protected:
size_t _rows, _cols;
T **_matrix;
};

// Portable PixMap image
class PPMImage : public Matrix<RGB<unsigned char> >
{
public:
   unsigned int size; 

PPMImage(const size_t height, const size_t width) : Matrix(height, width) { }
void save(const std::string &filename)
{
    std::ofstream out(filename, std::ios_base::binary);
    out <<"P6" << std::endl << _cols << " " << _rows << std::endl << 255 << std::endl;
    for (size_t y=0; y<_rows; y++)
        for (size_t x=0; x<_cols; x++) 
            out << _matrix[y][x].r << _matrix[y][x].g << _matrix[y][x].b;
}    
};

/*Draw mandelbrot according to the provided parameters*/
void draw_Mandelbrot(PPMImage & image, const unsigned width, const unsigned height, double cxmin, double cxmax, double cymin, double cymax,unsigned int max_iterations)                         
{

for (std::size_t ix = 0; ix < width; ++ix)
    for (std::size_t iy = 0; iy < height; ++iy)
    {
        std::complex<double> c(cxmin + ix / (width - 1.0)*(cxmax - cxmin), cymin + iy / (height - 1.0)*(cymax - cymin));
        std::complex<double> z = 0;
        unsigned int iterations;

        for (iterations = 0; iterations < max_iterations && std::abs(z) < 2.0; ++iterations)
            z = z*z + c;

        image[iy][ix].r = image[iy][ix].g = image[iy][ix].b = iterations;

    }
}

int main()
{
const unsigned width = 1600;
const unsigned height = 1600;

PPMImage image(height, width);


int parts = 8;

std::vector<int>bnd (parts, image.size);

std::thread *tt = new std::thread[parts - 1];

time_t start, end;
time(&start);
//Lauch parts-1 threads
for (int i = 0; i < parts - 1; ++i) {
    tt[i] = std::thread(draw_Mandelbrot,ref(image), width, height, -2.0, 0.5, -1.0, 1.0, 10);
}

//Use the main thread to do part of the work !!!
for (int i = parts - 1; i < parts; ++i) {
    draw_Mandelbrot(ref(image), width, height, -2.0, 0.5, -1.0, 1.0, 10);
}

//Join parts-1 threads
for (int i = 0; i < parts - 1; ++i)
    tt[i].join();

time(&end);
std::cout << difftime(end, start) << " seconds" << std::endl;


image.save("mandelbrot.ppm");

delete[] tt;

return 0;
}

现在每个thread 都绘制完整的分形(查看main())。如何让线程绘制分形的不同部分?

【问题讨论】:

  • 相关(Java):Multi-threading Java
  • 基本算术;找出每个部分的边界,将它们以及相应的 x 和 y 边界传递给线程。您需要将其中一个角的坐标添加到线程函数中。我建议您首先单线程执行(通过从main 调用函数),这样您就不必同时解决数学问题和线程问题..
  • @molbdnilo 你有这样的例子吗?
  • @Sybren 数学的例子?
  • @molbdnilo 不,在 C++ 中完成与您解释的类似操作的示例。

标签: c++ multithreading fractals mandelbrot


【解决方案1】:

你让这(相当多)变得比它需要的更难。这是 OpenMP 几乎完全适合的任务。对于这项任务,它以几乎最小的努力提供了几乎完美的缩放。

我修改了你的draw_mandelbrot,在外部for 循环之前插入了一个编译指示:

#pragma omp parallel for
for (int ix = 0; ix < width; ++ix)
    for (int iy = 0; iy < height; ++iy)

然后我将您的main 简化为:

int main() {
    const unsigned width = 1600;
    const unsigned height = 1600;

    PPMImage image(height, width);

    clock_t start = clock();
    draw_Mandelbrot(image, width, height, -2.0, 0.5, -1.0, 1.0, 10);
    clock_t stop = clock();

    std::cout << (double(stop - start) / CLOCKS_PER_SEC) << " seconds\n";

    image.save("mandelbrot.ppm");

    return 0;
}

在我的(相当慢的)机器上,您的原始代码在 4.73 秒内运行。我修改后的代码在 1.38 秒内运行。这比普通的单线程版本几乎无法区分代码的 3.4 倍。

为了它的价值,我做了更多的重写来得到这个:

// mandelbrot.cpp
// compile with: g++ -std=c++11 mandelbrot.cpp -o mandelbrot
// view output with: eog mandelbrot.ppm

#include <fstream>
#include <complex> // if you make use of complex number facilities in C++
#include <iostream>
#include <cstdlib>
#include <thread>
#include <mutex>
#include <vector>

using namespace std;

template <class T> struct RGB { T r, g, b; };

template <class T>
struct Matrix
{
    std::vector<T> data;
    size_t rows;
    size_t cols;

    class proxy {
        Matrix &m;
        size_t index_1;
    public:
        proxy(Matrix &m, size_t index_1) : m(m), index_1(index_1) { }

        T &operator[](size_t index) { return m.data[index * m.rows + index_1]; }
    };

    class const_proxy {
        Matrix const &m;
        size_t index_1;
    public:
        const_proxy(Matrix const &m, size_t index_1) : m(m), index_1(index_1) { }

        T const &operator[](size_t index) const { return m.data[index * m.rows + index_1]; }
    };


public:
    Matrix(size_t rows, size_t cols) : data(rows * cols), rows(rows), cols(cols) { }

    proxy operator[](size_t index) { return proxy(*this, index); }
    const_proxy operator[](size_t index) const { return const_proxy(*this, index); }

};

template <class T>
std::ostream &operator<<(std::ostream &out, Matrix<T> const &m) {
    out << "P6" << std::endl << m.cols << " " << m.rows << std::endl << 255 << std::endl;
    for (size_t y = 0; y < m.rows; y++)
        for (size_t x = 0; x < m.cols; x++) {
            T pixel = m[y][x];
            out << pixel.r << pixel.g << pixel.b;
        }
    return out;
}

/*Draw Mandelbrot according to the provided parameters*/
template <class T>
void draw_Mandelbrot(T & image, const unsigned width, const unsigned height, double cxmin, double cxmax, double cymin, double cymax, unsigned int max_iterations) {

#pragma omp parallel for
    for (int ix = 0; ix < width; ++ix)
        for (int iy = 0; iy < height; ++iy)
        {
            std::complex<double> c(cxmin + ix / (width - 1.0)*(cxmax - cxmin), cymin + iy / (height - 1.0)*(cymax - cymin));
            std::complex<double> z = 0;
            unsigned int iterations;

            for (iterations = 0; iterations < max_iterations && std::abs(z) < 2.0; ++iterations)
                z = z*z + c;

            image[iy][ix].r = image[iy][ix].g = image[iy][ix].b = iterations;

        }
}

int main() {
    const unsigned width = 1600;
    const unsigned height = 1600;

    Matrix<RGB<unsigned char>> image(height, width);

    clock_t start = clock();
    draw_Mandelbrot(image, width, height, -2.0, 0.5, -1.0, 1.0, 255);
    clock_t stop = clock();

    std::cout << (double(stop - start) / CLOCKS_PER_SEC) << " seconds\n";

    std::ofstream out("mandelbrot.ppm", std::ios::binary);
    out << image;

    return 0;
}

在我的机器上,这段代码运行大约需要 0.5 到 0.6 秒。

至于我为什么做出这些改变:主要是为了让它更快、更干净、更简单。您的 Matrix 类为每一行(或者可能是列——没有非常关注)分配了一个单独的内存块。这将分配整个矩阵的一个连续块。这消除了获取数据的间接级别,并增加了引用的局部性,从而提高了缓存的使用率。它还减少了使用的数据总量。

从使用 time 更改为使用 clock 进行计时是为了测量 CPU 时间而不是时钟时间(通常也会显着提高精度)。

摆脱 PPMImage 类只是因为(IMO)拥有一个派生自 Matrix 类的 PPImage 类并没有多大意义(如果有的话)。我想它可以工作(对于“工作”的足够宽松的定义),但它并没有让我觉得它是好的设计。如果您坚持这样做,它至少应该是私有派生,因为您只是使用 Matrix 作为实现 PPMImage 类的一种方式,而不是(至少我当然不希望)试图对属性做出断言PPM 图像。

如果出于某种原因,您决定手动处理线程,在线程之间划分工作的明显方法仍然是查看draw_mandelbrot 内部的循环。显而易见的做法是不理会您的外部循环,而是将每次迭代的计算发送到线程池: for (int ix = 0; ix

compute_thread 的主体基本上就是这段代码:

        for (int iy = 0; iy < height; ++iy)
        {
            std::complex<double> c(cxmin + ix / (width - 1.0)*(cxmax - cxmin), cymin + iy / (height - 1.0)*(cymax - cymin));
            std::complex<double> z = 0;
            unsigned int iterations;

            for (iterations = 0; iterations < max_iterations && std::abs(z) < 2.0; ++iterations)
                z = z*z + c;

            image[iy][ix].r = image[iy][ix].g = image[iy][ix].b = iterations;

        }

将正确的数据传递给计算线程显然需要做一些工作(每个线程都应该传递对结果图片切片的引用),但这将是一个明显且相当干净的地方来划分事物向上。特别是,它将工作划分为足够多的任务,您可以半自动地获得相当好的负载平衡(即,您可以保持所有核心忙碌)但足够大,您不会浪费大量时间在通信和同步上线程。

至于结果,将迭代次数设置为 255,我得到以下结果(缩放到 25%):

...这和我预期的差不多。

【讨论】:

  • 这是一个巨大的改进,我将研究 OpenMP,但现在的主要问题是将图像分成几块并用多个线程处理这些块。该怎么做?
  • @Sybren:通过插入上述编译指示(即,这正是 OpenMP 正在做的事情)。
  • 按照您的解释插入了 pragma,并用简化的 main 替换了 main,但 GIMP 现在显示此错误:PNM Image Plug-in could not open image.(我使用 GIMP 打开图像)
  • @Sybren:我不知道——我进行了位比较,发现我得到的图像与之前生成的图像相同。
  • 我看到你做了更多的重写,然后添加了 pragma 并简化了 main。你不使用PPMImageclass,这是什么原因?我更喜欢让程序尽可能与我发布的代码保持一致。
【解决方案2】:

这种方法的一个大问题是不同的地区需要不同的时间来计算。

更通用的方法是。

  • 启动 1 个源线程。
  • 启动 N 个工作线程。
  • 启动 1 个接收器线程。
  • 创建 2 个线程安全队列(将它们称为源队列和接收器队列)。
  • 将图像分成 M(多于 N)块。
  • 源线程将片段推送到源队列中
  • 工作人员从源队列中提取片段,将片段转换为结果片段,并将这些片段推送到接收队列中。
  • 接收器线程从接收器队列中获取片段并将它们组合成最终图像。

通过这种方式划分工作,所有工作线程将一直处于忙碌状态。

【讨论】:

  • 如何将图片分割成碎片?我找不到任何例子。
  • @Sybren 老实说,很难相信您编写了绘制曼德布罗分形的代码,但不知道如何将矩形区域划分为多个部分。如果您有一个宽度为w 和高度为h 的矩形,那么您可以简单地将其分成两个宽度为w/2 和高度为h 的矩形,其中一个与原始矩形的x 偏移量为w/2。我会留下进一步的划分作为练习。
【解决方案3】:

您可以通过将分形的开始和结束与屏幕尺寸分开来将分形分成几块:

   $this->stepsRe = (double)((($this->startRe * -1) + ($this->endeRe)) / ($this->size_x-1));
   $this->stepsIm = (double)((($this->startIm * -1) + ($this->endeIm)) / ($this->size_y-1));

【讨论】:

  • $this-&gt; 到底应该是什么?您是否尝试编写 PHP 而不是 C++?
  • 是的。它的PHP。但这与语言无关!?
  • 问题是标记为C++。虽然数学的基本概念并不局限于任何一种语言,但他仍然相当明确地要求一种语言,而且使用 PHP 似乎没有任何好处。
  • IMO 的问题是如何细分图像。他在 cmets 中多次问过这个问题。
  • 好吧,我想如果您的意图是产生更多热量,并且产生结果的速度要慢得多,那么这将可能是一种有效的方法。 ..
猜你喜欢
  • 2020-12-08
  • 1970-01-01
  • 1970-01-01
  • 1970-01-01
  • 2023-03-28
  • 2016-09-06
  • 1970-01-01
  • 1970-01-01
  • 2023-03-12
相关资源
最近更新 更多