Как правильно распараллелить программу?
1,00
р.
В качестве ответа на вопрос Какой алгоритм использовать для решения задачи? написал свою программку (приведена ниже) - ветвления с отсечением. Бывает, меня, как поющего Кобзона :), не остановить - словом, мне захотелось выжать все, что можно. В однопоточном режиме, как мне кажется, выжал все, что мог (но, конечно, дальнейшее ускорение приветствуется). Захотел попробовать ускорить за счет параллельности - и вот тут я застрял. Любой из вариантов у меня оказывался резко хуже однопоточного.
Начал я с того, что создавал потоков по количеству ядер. Каждый поток поочередно (с синхронизацией с помощью мьютексов) брал очередную ветвь на первом уровне и полностью обрабатывал ее. Получилось плохо, как как каждая ветвь отрабатывалась полностью независимо, т.е. искалось полное решение для нее - в то время как она, быть может, была бы отброшена сразу - из-за наличия другой более короткой ветви.
Далее сделал общим значение достигнутого минимума среди всех потоков - но это значение опять же пришлось защищать мьютексом, и это привело к очередному увеличению времени работы.
Посему хотелось бы посмотреть на то, как эту задачу решат реальные эксперты в области многопоточности (к каковым себя отнести ну никак не могу).
Так сказать, конкурс на самый быстрый вариант. Интересуют именно параллельные вычисления.
А вот обещанный однопоточный код (да, я знаю, что он написан ужасно - обсуждается не это, ладно?):
#include #include #include
using namespace std
struct pnt { double x, y }
using distance_t = vector>
struct Func { long long calls double min // Текущее минимальное расстояние vector save // Сохраненная перестановка const distance_t &dist // Расстония между точками
Func(const distance_t& dist, const vector& x):dist(dist) { calls = 0 min = 0.0 // Инициализация путем 0-1-2-... save.push_back(0) for(size_t i = 1 i < x.size() ++i) { min += dist[i-1][i] save.push_back(i) } }
// Проверка ветви bool operator()(const vector& x, size_t l, double cur) { ++calls // Режем все, где неверный конец (неверное начало невозможно) if (l != x.size()-1 && x[l] == int(x.size() - 1)) return false
// Текущая длина + расстояние до последней точки, если еще не достали if (l != x.size()-1) cur += dist[x[l]][x.size()-1]
// Если больше минимальной - режем эту ветвь if (cur > min + min*DBL_EPSILON) return false // Сохранение нового пути if (l == x.size()-1) { if (abs(min-cur) < min*DBL_EPSILON) { // Только лексикографически меньший путь if (save > x) save = x } else { min = cur save = x } } return true } }
// Ветвление и обрезка bool branches(size_t N, Func& f, const distance_t& dist, size_t level = 1, vector*v_ = nullptr, double cur_dist = 0.0) { // Вспомогательный вектор перестановок пути vector * vv = (level == 1) ? new vector : v_ if (level == 1) for(size_t i = 0 i < N ++i) vv->push_back(i) vector& v = *vv
for(size_t i = level i < N ++i) { // Очередная перестановка std::swap(v[level],v[i]) // Длина для нее double length = cur_dist + dist[v[level]][v[level-1]] if (f(v,level,length) && level < N-1) branches(N,f,dist,level+1,vv,length) // Возвращаем все, как было std::swap(v[i],v[level]) } if (level == 1) delete vv return true }
int main(/*int argc, const char * argv[]*/) { int N cin >> N vector x for(int i = 0 i < N ++i) { // cout <<"Point " << (i+1) << ": " double xx, yy cin >> xx >> yy x.push_back(pnt{xx,yy}) }
distance_t dist(N,vector(N,0.0)) for(int i = 0 i < N ++i) for(int j = i+1 j < N ++j) dist[i][j] = dist[j][i] = sqrt((x[i].x-x[j].x)*(x[i].x-x[j].x)+ (x[i].y-x[j].y)*(x[i].y-x[j].y))
Func f(dist,x) branches(N,f,dist) cout << "<br>" << f.min << endl for(auto i: f.save) cout << (i+1) << " " cout << endl <br> cout << "Calls Func(): " << setw(12) << f.calls << endl <br>}
Ответ
Вот как надо:
#include #include #include #include
using namespace std
#define DBL_EPSILON 0.0000000000001
struct pnt { double x, y }
using distance_t = vector>
struct Func { long long calls double min // Текущее минимальное расстояние vector save // Сохраненная перестановка const distance_t &dist // Расстония между точками
Func(const distance_t& dist, const vector& x):dist(dist) { calls = 0 min = 0.0 // Инициализация путем 0-1-2-... save.push_back(0) for(size_t i = 1 i < x.size() ++i) { min += dist[i-1][i] save.push_back(i) } }
// Проверка ветви bool operator()(const vector& x, size_t l, double cur) { ++calls // Режем все, где неверный конец (неверное начало невозможно) if (l != x.size()-1 && x[l] == int(x.size() - 1)) return false
// Текущая длина + расстояние до последней точки, если еще не достали if (l != x.size()-1) cur += dist[x[l]][x.size()-1]
// Если больше минимальной - режем эту ветвь if (cur > min + min*DBL_EPSILON) return false // Сохранение нового пути if (l == x.size()-1) { if (abs(min-cur) < min*DBL_EPSILON) { // Только лексикографически меньший путь if (save > x) save = x } else { min = cur save = x } } return true } }
// Ветвление и обрезка bool branches(size_t N, Func& f, const distance_t& dist, size_t level = 1, vector*v_ = nullptr, double cur_dist = 0.0) { // Вспомогательный вектор перестановок пути vector * vv = (level == 1) ? new vector : v_ if (level == 1) for(size_t i = 0 i < N ++i) vv->push_back(i) vector& v = *vv
for(size_t i = level i < N ++i) { // Очередная перестановка std::swap(v[level],v[i]) // Длина для нее double length = cur_dist + dist[v[level]][v[level-1]] if (f(v,level,length) && level < N-1) branches(N,f,dist,level+1,vv,length) // Возвращаем все, как было std::swap(v[i],v[level]) } if (level == 1) delete vv return true }
int main(/*int argc, const char * argv[]*/) { int N cin >> N vector x for(int i = 0 i < N ++i) { // cout <<"Point " << (i+1) << ": " double xx, yy cin >> xx >> yy x.push_back(pnt{xx,yy}) }
distance_t dist(N,vector(N,0.0)) for(int i = 0 i < N ++i) for(int j = i+1 j < N ++j) dist[i][j] = dist[j][i] = sqrt((x[i].x-x[j].x)*(x[i].x-x[j].x)+ (x[i].y-x[j].y)*(x[i].y-x[j].y))
Func f(dist,x) branches(N,f,dist) cout << "<br>" << f.min << endl for(auto i: f.save) cout << (i+1) << " " cout << endl <br> cout << "Calls Func(): " << setw(12) << f.calls << endl <br>}
А вот ошибки: надо #include добавить и задать DBL_EPSILON.
Начал я с того, что создавал потоков по количеству ядер. Каждый поток поочередно (с синхронизацией с помощью мьютексов) брал очередную ветвь на первом уровне и полностью обрабатывал ее. Получилось плохо, как как каждая ветвь отрабатывалась полностью независимо, т.е. искалось полное решение для нее - в то время как она, быть может, была бы отброшена сразу - из-за наличия другой более короткой ветви.
Далее сделал общим значение достигнутого минимума среди всех потоков - но это значение опять же пришлось защищать мьютексом, и это привело к очередному увеличению времени работы.
Посему хотелось бы посмотреть на то, как эту задачу решат реальные эксперты в области многопоточности (к каковым себя отнести ну никак не могу).
Так сказать, конкурс на самый быстрый вариант. Интересуют именно параллельные вычисления.
А вот обещанный однопоточный код (да, я знаю, что он написан ужасно - обсуждается не это, ладно?):
#include #include #include
using namespace std
struct pnt { double x, y }
using distance_t = vector>
struct Func { long long calls double min // Текущее минимальное расстояние vector save // Сохраненная перестановка const distance_t &dist // Расстония между точками
Func(const distance_t& dist, const vector& x):dist(dist) { calls = 0 min = 0.0 // Инициализация путем 0-1-2-... save.push_back(0) for(size_t i = 1 i < x.size() ++i) { min += dist[i-1][i] save.push_back(i) } }
// Проверка ветви bool operator()(const vector& x, size_t l, double cur) { ++calls // Режем все, где неверный конец (неверное начало невозможно) if (l != x.size()-1 && x[l] == int(x.size() - 1)) return false
// Текущая длина + расстояние до последней точки, если еще не достали if (l != x.size()-1) cur += dist[x[l]][x.size()-1]
// Если больше минимальной - режем эту ветвь if (cur > min + min*DBL_EPSILON) return false // Сохранение нового пути if (l == x.size()-1) { if (abs(min-cur) < min*DBL_EPSILON) { // Только лексикографически меньший путь if (save > x) save = x } else { min = cur save = x } } return true } }
// Ветвление и обрезка bool branches(size_t N, Func& f, const distance_t& dist, size_t level = 1, vector*v_ = nullptr, double cur_dist = 0.0) { // Вспомогательный вектор перестановок пути vector * vv = (level == 1) ? new vector : v_ if (level == 1) for(size_t i = 0 i < N ++i) vv->push_back(i) vector& v = *vv
for(size_t i = level i < N ++i) { // Очередная перестановка std::swap(v[level],v[i]) // Длина для нее double length = cur_dist + dist[v[level]][v[level-1]] if (f(v,level,length) && level < N-1) branches(N,f,dist,level+1,vv,length) // Возвращаем все, как было std::swap(v[i],v[level]) } if (level == 1) delete vv return true }
int main(/*int argc, const char * argv[]*/) { int N cin >> N vector x for(int i = 0 i < N ++i) { // cout <<"Point " << (i+1) << ": " double xx, yy cin >> xx >> yy x.push_back(pnt{xx,yy}) }
distance_t dist(N,vector(N,0.0)) for(int i = 0 i < N ++i) for(int j = i+1 j < N ++j) dist[i][j] = dist[j][i] = sqrt((x[i].x-x[j].x)*(x[i].x-x[j].x)+ (x[i].y-x[j].y)*(x[i].y-x[j].y))
Func f(dist,x) branches(N,f,dist) cout << "<br>" << f.min << endl for(auto i: f.save) cout << (i+1) << " " cout << endl <br> cout << "Calls Func(): " << setw(12) << f.calls << endl <br>}
Ответ
Вот как надо:
#include #include #include #include
using namespace std
#define DBL_EPSILON 0.0000000000001
struct pnt { double x, y }
using distance_t = vector>
struct Func { long long calls double min // Текущее минимальное расстояние vector save // Сохраненная перестановка const distance_t &dist // Расстония между точками
Func(const distance_t& dist, const vector& x):dist(dist) { calls = 0 min = 0.0 // Инициализация путем 0-1-2-... save.push_back(0) for(size_t i = 1 i < x.size() ++i) { min += dist[i-1][i] save.push_back(i) } }
// Проверка ветви bool operator()(const vector& x, size_t l, double cur) { ++calls // Режем все, где неверный конец (неверное начало невозможно) if (l != x.size()-1 && x[l] == int(x.size() - 1)) return false
// Текущая длина + расстояние до последней точки, если еще не достали if (l != x.size()-1) cur += dist[x[l]][x.size()-1]
// Если больше минимальной - режем эту ветвь if (cur > min + min*DBL_EPSILON) return false // Сохранение нового пути if (l == x.size()-1) { if (abs(min-cur) < min*DBL_EPSILON) { // Только лексикографически меньший путь if (save > x) save = x } else { min = cur save = x } } return true } }
// Ветвление и обрезка bool branches(size_t N, Func& f, const distance_t& dist, size_t level = 1, vector*v_ = nullptr, double cur_dist = 0.0) { // Вспомогательный вектор перестановок пути vector * vv = (level == 1) ? new vector : v_ if (level == 1) for(size_t i = 0 i < N ++i) vv->push_back(i) vector& v = *vv
for(size_t i = level i < N ++i) { // Очередная перестановка std::swap(v[level],v[i]) // Длина для нее double length = cur_dist + dist[v[level]][v[level-1]] if (f(v,level,length) && level < N-1) branches(N,f,dist,level+1,vv,length) // Возвращаем все, как было std::swap(v[i],v[level]) } if (level == 1) delete vv return true }
int main(/*int argc, const char * argv[]*/) { int N cin >> N vector x for(int i = 0 i < N ++i) { // cout <<"Point " << (i+1) << ": " double xx, yy cin >> xx >> yy x.push_back(pnt{xx,yy}) }
distance_t dist(N,vector(N,0.0)) for(int i = 0 i < N ++i) for(int j = i+1 j < N ++j) dist[i][j] = dist[j][i] = sqrt((x[i].x-x[j].x)*(x[i].x-x[j].x)+ (x[i].y-x[j].y)*(x[i].y-x[j].y))
Func f(dist,x) branches(N,f,dist) cout << "<br>" << f.min << endl for(auto i: f.save) cout << (i+1) << " " cout << endl <br> cout << "Calls Func(): " << setw(12) << f.calls << endl <br>}
А вот ошибки: надо #include добавить и задать DBL_EPSILON.
