проблема с распараллеливанием

Дата: Воскресенье, 12 Ноября 2017, 12:09 | Сообщение # 1

Пчёлка Зоя

Сейчас нет на сайте

Хочу разделить игру на 4 потока (пока что) . Пока что это чисто тестовые потоки, которые ничего не делают, хочу хоть как-то заставить их работать.

Идея такова:
1)инициализирую эти потоки перед вхождением в главный игровой цикл.
2)эти потоки внутри крутят бесконечный цикл, но вначале каждой итерации ждут оповещения от главного потока, чтобы он разрешил им работать
3)внутри главного игрового цикла каждую итерацию я оповещаю эти потоки о том, что они могут выполнить очередную итерацию по формированию кадра.
4)Тогда эти потоки снимаются с паузы и выполняют итерацию, после чего они должны оповестить главный поток о том, что они завершили работу и снова находятся на паузе.
5)главный поток ждёт оповещения ото всех запущенных потоков
6)следующая итерация главного игрового цикла.

Проблема: где-то получаю дедлок на рандомной итерации главного игрового цикла и в упор не понимаю, почему. Это мой первый опыт с многопоточкой, буду рад, если скажете, что у меня в коде не так и почему, потому что я ещё не совсем влился в тему.

Код делает дедлок даже если запустить 1 поток вместо 4-х, поэтому я 3 других даже закомментил, чтобы не мешались.

Главной функцией потока является ф-ция action()
Она принимает аргументом ту функцию, которую должен вызывать поток собственно для работы.
Внутри она крутит бесконечный цикл и ждёт разрешения на работу.
Когда разрешение получено, она выполняет нужную функцию (например, render()), которая, в свою очередь, по завершении, даёт главному потоку информацию о том, что работа завершилась

Ф-ция update() вызывается из главного цикла и оповещает потоки о том, что они могут начинать работу, после чего ждёт завершения их выполнения.

В общем, нужен хелп, долбусь уже целый день.

Код

#include <Windows.h>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <SFML/Graphics.hpp>

void init();

void update(float dt);
void postUpdate(float dt);

void action(void(*func) (float), int mapId);

void render(float dt);
void physics(float dt);
void AI(float dt);
void misc(float dt); //do something (чисто чтоб нашкрести на 4 потока)

void notifyFinishFrame();

void finish();
void quit();

sf::Window Wnd;

bool running;

std::mutex mutIterationStarted;
std::condition_variable condNewIterationStarted;

std::mutex mutFinishedFrame;
std::condition_variable condFinishedFrame;
int finishedFrameNum;

#define THREADS_AMOUNT 1
bool threadReadyMap[THREADS_AMOUNT];

void init()
{
    setlocale(LC_ALL, "ru_RU.UTF-8");

    //WindowWidth  = GetSystemMetrics(SM_CXSCREEN);
    //WindowHeight = GetSystemMetrics(SM_CYSCREEN);
    int WindowWidth = GetSystemMetrics(SM_CXSCREEN) - 80;
    int WindowHeight = GetSystemMetrics(SM_CYSCREEN) - 15;

    Wnd.create(sf::VideoMode(WindowWidth, WindowHeight), "SNS 1.5 v0.0.1", sf::Style::Default, sf::ContextSettings(0, 0, 4));
    Wnd.setPosition(sf::Vector2i(80, 0));
    Wnd.setVerticalSyncEnabled(true);
}

int WINAPI WinMain(HINSTANCE hInstance, HINSTANCE hPrevInstance, PSTR pCmdLine, int nShowCmd)
{
    init();

    running = true;
    std::thread threads[THREADS_AMOUNT] = {
  std::thread(action, std::ref(render), 0),
  //std::thread(action, std::ref(physics), 1),
  //std::thread(action, std::ref(AI), 2),
  //std::thread(action, std::ref(misc), 3),
    };

    while (running && Wnd.isOpen())
    {
  sf::Event event;
  while (Wnd.pollEvent(event))
  {
  }

  update(1/30.0);
    }

    finish();

    for (int i = 0; i < THREADS_AMOUNT; i++) {
  threads[i].join();
    }

    return 0;
}

void finish()
{
    {
  std::unique_lock<std::mutex> iterationLock(mutIterationStarted);
  running = false;
  condNewIterationStarted.notify_all();
    }

    Wnd.setMouseCursorVisible(true);
}

void quit() {
    std::unique_lock<std::mutex> iterationLock(mutIterationStarted);
    running = false;
}

void action(void(*func) (float), int mapId)
{
    while (true)
    {
  std::unique_lock<std::mutex> iterationLock(mutIterationStarted);
  bool exit = false;
  while (true) {
   condNewIterationStarted.wait(iterationLock);

   if (!running) {
    exit = true;
    break;
   }
   else if (threadReadyMap[mapId]) {
    threadReadyMap[mapId] = false;
    break;
   }
  }
  if (exit) break;

  func(1/30.0);
    }
}

void update(float dt)
{
    {
  std::unique_lock<std::mutex> iterationLock(mutIterationStarted);
  for (int i = 0; i < THREADS_AMOUNT; i++) {
   threadReadyMap[i] = true;
  }
  condNewIterationStarted.notify_all();
    }

    {
  std::unique_lock<std::mutex> frameLock(mutFinishedFrame);
  finishedFrameNum = 0;
  condFinishedFrame.wait(frameLock, []() { return finishedFrameNum == THREADS_AMOUNT; });
    }

    postUpdate(dt);
}

void postUpdate(float dt)
{
    static int a = 0;

    a++;
    if (a > 100000) {
  quit();
    }
}

void render(float dt)
{
    //do render

    notifyFinishFrame();
}

void physics(float dt)
{
    //do physics

    notifyFinishFrame();
}

void AI(float dt)
{
    //do AI

    notifyFinishFrame();
}

void misc(float dt)
{
    //do something

    notifyFinishFrame();
}

void notifyFinishFrame()
{
    std::unique_lock<std::mutex> frameLock(mutFinishedFrame);
    finishedFrameNum++;
    if (finishedFrameNum == THREADS_AMOUNT) {
  condFinishedFrame.notify_one();
    }
}

Добавлено (11 ноября 2017, 17:49)
---------------------------------------------
Кажись, решил проблему небольшим костылём
Как было устроено: все потоки в начале фрейма ждут разрешение на работу, главный поток им его даёт. Потом он ждёт когда потоки оповестят его о завершении работы. При этом, для того чтобы обновить инфу о прогрессе работы потоков, использовался тот же мьютекс, который использовался в условной переменной, которая ждала результат. Возможно, потому я и ловил дедлок
как решил: убрал условную переменную, которая ждала пока потоки не завершат кадр и не лочу в главном потоке соответствующий мьютекс.
Вместо этого заставляю главный поток в цикле проверять, закончена ли работа и ждать. Так что, теперь итерация главного цикла (ф-ция update) выглядит так

Сами потоки в конце работы вызывают ф-цию

Результат: 10к пустых итераций главного цикла проходят за 11-12 секунд => 900 кадров в секунду, не так уж и плохо.

Код

#include <Windows.h>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <SFML/Graphics.hpp>

void init();

void update(float dt);
void postUpdate(float dt);

void action(void(*func) (float), int mapId);

void render(float dt);
void physics(float dt);
void AI(float dt);
void misc(float dt); //do something

void notifyFinishFrame();

void finish();
void quit();

sf::Window Wnd;

bool running;

#define THREADS_AMOUNT 4
bool threadReadyMap[THREADS_AMOUNT];

std::mutex mutIterationStarted[THREADS_AMOUNT];
std::condition_variable condIterationStarted[THREADS_AMOUNT];

std::mutex mutFinishedFrame;
int finishedFrameNum;

void init()
{
    setlocale(LC_ALL, "ru_RU.UTF-8");

    //WindowWidth  = GetSystemMetrics(SM_CXSCREEN);
    //WindowHeight = GetSystemMetrics(SM_CYSCREEN);
    int WindowWidth = GetSystemMetrics(SM_CXSCREEN) - 80;
    int WindowHeight = GetSystemMetrics(SM_CYSCREEN) - 15;

    Wnd.create(sf::VideoMode(WindowWidth, WindowHeight), "SNS 1.5 v0.0.1", sf::Style::Default, sf::ContextSettings(0, 0, 4));
    Wnd.setPosition(sf::Vector2i(80, 0));
    Wnd.setVerticalSyncEnabled(true);
}

int WINAPI WinMain(HINSTANCE hInstance, HINSTANCE hPrevInstance, PSTR pCmdLine, int nShowCmd)
{
    init();

    running = true;
    std::thread threads[THREADS_AMOUNT] = {
  std::thread(action, std::ref(render), 0),
  std::thread(action, std::ref(physics), 1),
  std::thread(action, std::ref(AI), 2),
  std::thread(action, std::ref(misc), 3),
    };

    while (running && Wnd.isOpen())
    {
  sf::Event event;
  while (Wnd.pollEvent(event))
  {
  }

  update(1/30.0);
    }

    finish();

    for (int i = 0; i < THREADS_AMOUNT; i++) {
  threads[i].detach();
    }

    return 0;
}

void finish()
{
    Wnd.setMouseCursorVisible(true);
}

void quit() {
    std::unique_lock<std::mutex> iterationLock[THREADS_AMOUNT] = {
  std::unique_lock<std::mutex>(mutIterationStarted[0]),
  std::unique_lock<std::mutex>(mutIterationStarted[1]),
  std::unique_lock<std::mutex>(mutIterationStarted[2]),
  std::unique_lock<std::mutex>(mutIterationStarted[3]),
    };
    running = false;
}

void action(void(*func) (float), int mapId)
{
    while (true)
    {
  std::unique_lock<std::mutex> iterationLock(mutIterationStarted[mapId]);
  bool exit = false;
  while (true) {
   condIterationStarted[mapId].wait(iterationLock);

   if (!running) {
    exit = true;
    break;
   }
   else if (threadReadyMap[mapId]) {
    threadReadyMap[mapId] = false;
    break;
   }
  }
  if (exit) break;

  func(1/30.0);
    }
}

void update(float dt)
{
    finishedFrameNum = 0;

    for (int i = 0; i < THREADS_AMOUNT; i++) {
  std::unique_lock<std::mutex> iterationLock(mutIterationStarted[i]);
  threadReadyMap[i] = true;
  condIterationStarted[i].notify_one();
    }


    while (finishedFrameNum != THREADS_AMOUNT) {
  std::this_thread::sleep_for(std::chrono::milliseconds(1));
    }

    postUpdate(dt);
}

void postUpdate(float dt)
{
    static int a = 0;

    a++;
    if (a > 10000) {
  quit();
    }
}

void render(float dt)
{
    //do render
    int k = 6;
    for (int i = 0; i < 70000; i++) {
  k += i / 2;
    }

    notifyFinishFrame();
}

void physics(float dt)
{
    //do physics

    notifyFinishFrame();
}

void AI(float dt)
{
    //do AI

    notifyFinishFrame();
}

void misc(float dt)
{
    //do something

    notifyFinishFrame();
}

void notifyFinishFrame()
{
    std::lock_guard<std::mutex> frameLock(mutFinishedFrame);
    finishedFrameNum++;
}

Добавлено (11 ноября 2017, 18:02)
---------------------------------------------
попробую теперь на этапе инициализации проверять, сколько ядер в компе юзера.
Если 4+ - запускать 3-4 потока
Если 2 - запускать 2 потока, в первом - рендер+что-то не слишком тяжёлое, во втором - всё остальное
Если 1 - 2 потока, рендер в одном, всё остальное во втором

Добавлено (11 ноября 2017, 22:15)
---------------------------------------------
Всё, обернул всё это дело в класс. Можно создать произвольное количество потоков, в каждый из потоков запихнуть произвольное количество последовательно исполняемых задач.

Только одна проблема есть: сделал синглтон, отнаследовался от него и попытался получать инстанс нужного класса, не получалось - психанул и удалил наследование от синглтона, всё равно я один над проектом сижу, и так знаю, что класс в единственном экземпляре должен быть))

Код

#include "Threading.h"

extern bool running;

Threading::Threading()
{
}

Threading::~Threading()
{
    quit();
}

void Threading::init(int threadsNum, threadingActionFunction postUpdate, int waitIntervalMicroseconds)
{
    threadsNum_ = threadsNum;
    postUpdateFunction_ = postUpdate;
    waitIntervalMicroseconds_ = waitIntervalMicroseconds;

    initiated_ = true;

    functions_.reserve( threadsNum_);
    threadReadyMap_.reserve( threadsNum_);
    mutIterationStarted_.reserve( threadsNum_);
    condIterationStarted_.reserve( threadsNum_);
    for (int i = 0; i < threadsNum_; i++) {
  functions_.emplace_back( std::vector <threadingActionFunction>() );
  threadReadyMap_.emplace_back(false);
  mutIterationStarted_.emplace_back( new std::mutex() );
  condIterationStarted_.emplace_back( new std::condition_variable() );
    }
}

void Threading::pushFunction(int threadId, threadingActionFunction func)
{
    functions_[threadId].push_back(func);
}

void Threading::finalizeInitiation()
{
    threads_.reserve(threadsNum_);
    for (int i = 0; i < threadsNum_; i++) {
  threads_.emplace_back( new std::thread(&Threading::action, this, i) );
    }
}

void Threading::update(float deltaTime)
{
    deltaTime_ = deltaTime;
    finishedFrameNum_ = 0;

    for (int i = 0; i < threadsNum_; i++) {
  std::unique_lock<std::mutex> iterationLock(*mutIterationStarted_[i]);
  threadReadyMap_[i] = true;
  condIterationStarted_[i]->notify_one();
    }

    while (finishedFrameNum_ != threadsNum_) {
  std::this_thread::sleep_for(std::chrono::microseconds(waitIntervalMicroseconds_));
    }

    postUpdateFunction_(deltaTime_);
}

void Threading::action(int threadId)
{
    while (true)
    {
  std::unique_lock<std::mutex> iterationLock(*mutIterationStarted_[threadId]);
  bool exit = false;
  while (true) {
   //wait for starting new game iteration
   condIterationStarted_[threadId]->wait(iterationLock);

   if (!running) {
    exit = true;
    break;
   }
   else if (threadReadyMap_[threadId]) {
    //to avoid spurious wakeup
    threadReadyMap_[threadId] = false;
    break;
   }
  }
  if (exit) break;

  //call all functions of the thread
  for (int i = 0; i < functions_[threadId].size(); i++) {
   functions_[threadId][i](deltaTime_);
  }

  //increase counter of threads that finished making this frame
  notifyFinishFrame();
    }
}

void Threading::notifyFinishFrame()
{
    std::lock_guard<std::mutex> frameLock(mutFinishedFrame_);
    finishedFrameNum_++;
}

void Threading::quit()
{
    if (!initiated_) {
  return;
    }
    initiated_ = false;

    for (int i = 0; i < threadsNum_; i++) {
  mutIterationStarted_[i]->lock();
    }

    running = false;

    for (int i = 0; i < threadsNum_; i++) {
  mutIterationStarted_[i]->unlock();
    }

    for (int i = 0; i < threadsNum_; i++) {
  condIterationStarted_[i]->notify_one();
    }

    for (int i = 0; i < threadsNum_; i++) {
  if (threads_[i]->joinable()) {
   threads_[i]->join();
  }

  delete threads_[i];
    }
}

Добавлено (12 ноября 2017, 12:09)
---------------------------------------------
упс, ещё один дедлок был, который случался довольно редко (тобишь прога стабильно работала несколько десятков секунд в полной нагрузке, прежде чем валилась)
На этот раз связанный со "spurious wakeup"

Дело в том, что может так совпасть, что поток сорвётся по ложному прерыванию за павру мгновений до того, как главный поток оповестит его о начале работы. Таким макаром, когда главный поток его оповещает, тот всё ещё занят обработкой ложного прерывания и усыплением себя, а ведь оповещение из главного потока прошло мимо! А потом мы ждём завершения работы потока, в то время как он пропустил оповещение о начале работы.
Вылечил так: в цикле спамим оповещения, разблокируя при этом мьютекс условной переменной до тех пор, пока не убедимся, что наше сообщение дошло

https://vk.com/beezoya

Сообщение отредактировал puksus - Суббота, 11 Ноября 2017, 19:45