Оптимізація сигналізації між потоками у C++ за допомогою умовних змінних

5 Березня 2024 в 21:13 33

При роботі з багатьма виробниками та одним споживачем у потоковому середовищі виникає проблема ефективної сигналізації споживачеві про готовність даних. Стандартним методом для цього є використання умовних змінних та атомарних операцій. Проте, існує можливість оптимізації, яка дозволить уникнути зайвої конкуренції за блокуванням м’ютексу.

constexpr int N_WORKERS = 10;
struct WorkData {
    int input[N_WORKERS];
    int result[N_WORKERS];
    std::atomic<int> remainingWorkers;
    std::condition_variable cv;
    std::mutex mutex;
};
void workerFunc(WorkData* wd, int index) {
    // Виконуємо обчислення
    wd->result[index] = wd->input[index] * wd->input[index];
    // Зменшуємо лічильник
    if (--wd->remainingWorkers == 0) {
        wd->mutex.lock();
        wd->mutex.unlock();
        wd->cv.notify_one();
    }
}
int main() {
    WorkData wd;
    wd.remainingWorkers.store(N_WORKERS);
    std::thread workerThreads[N_WORKERS];
    for (int i = 0; i < N_WORKERS; i++) {
        wd.input[i] = i;
        wd.result[i] = 0;
        workerThreads[i] = std::thread(workerFunc, &wd, i);
    }
    // Чекаємо завершення роботи виробників
    if (wd.remainingWorkers.load() > 0) {
        std::unique_lock<std::mutex> lock(wd.mutex);
        while (wd.remainingWorkers.load() > 0)
            wd.cv.wait(lock);
    }
    // Обробляємо результати обчислень
    for (int i = 0; i < N_WORKERS; i++)
        std::cout << wd.input[i] << "^2 = " << wd.result[i] << std::endl;
    for (std::thread& t : workerThreads)
        t.join();
    return 0;
}

constexpr int N_WORKERS = 10;

struct WorkData {

int input[N_WORKERS];

int result[N_WORKERS];

std::atomic<int> remainingWorkers;

std::condition_variable cv;

std::mutex mutex;

};

void workerFunc(WorkData* wd, int index) {

// Виконуємо обчислення

wd->result[index] = wd->input[index] * wd->input[index];

// Зменшуємо лічильник

if (--wd->remainingWorkers == 0) {

wd->mutex.lock();

wd->mutex.unlock();

wd->cv.notify_one();

}

int main() {

WorkData wd;

wd.remainingWorkers.store(N_WORKERS);

std::thread workerThreads[N_WORKERS];

for (int i = 0; i < N_WORKERS; i++) {

wd.input[i] = i;

wd.result[i] = 0;

workerThreads[i] = std::thread(workerFunc, &wd, i);

}

// Чекаємо завершення роботи виробників

if (wd.remainingWorkers.load() > 0) {

std::unique_lock<std::mutex> lock(wd.mutex);

while (wd.remainingWorkers.load() > 0)

wd.cv.wait(lock);

}

// Обробляємо результати обчислень

for (int i = 0; i < N_WORKERS; i++)

std::cout << wd.input[i] << "^2 = " << wd.result[i] << std::endl;

for (std::thread& t : workerThreads)

t.join();

return 0;

}

Підхід, який дозволяє виробникам зменшувати лічильник атомарно без блокуванням м’ютексу, здається безпечним, оскільки використовує атомарні операції, які гарантують відсутність гонок. Потрібно забезпечити, щоб м’ютекс був захоплений під час відправлення сигналу, щоб уникнути його втрати через перевірку умови і вступ до очікування умовної змінної. Цей підхід дозволяє зменшити накладні витрати на блокування м’ютексу для кожного виробника та покращити ефективність системи в цілому.