我创建了一个创建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()）。如何让线程绘制分形的不同部分？
你实现这个（相当多）比它需要的更难。这是OpenMP几乎非常适合的任务。对于这项任务，它只需最少的努力即可实现几乎完美的缩放。
我通过在外部draw_mandelbrot循环之前插入一个pragma来修改你的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 <width; ++ ix）compute_thread（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％）：
......这正是我所期待的。
这种方法的一个主要问题是不同的区域需要不同的时间来计算。
更通用的方法是。
通过这种方式划分工作，所有工作线程将一直处于忙碌状态。
您可以通过将分形的开始和结束除以屏幕尺寸来将分形分割成碎片：
$this->stepsRe = (double)((($this->startRe * -1) + ($this->endeRe)) / ($this->size_x-1));
$this->stepsIm = (double)((($this->startIm * -1) + ($this->endeIm)) / ($this->size_y-1));