1

Для примера есть вот такая функция:

auto lambda = [](int k)
{
    size_t size = 1920 * 1080 * 3;
    size_t repeatCount = 100;
    char* ch = new char [size];
    auto start = std::chrono::high_resolution_clock::now();
    for (int i = 0; i < repeatCount; ++i)
    {
        for (int j = 0; j < size; ++j)
        {
            ch[j] = j % 255;
        }

    }
    auto duration = std::chrono::duration<double, std::milli>(std::chrono::high_resolution_clock::now() - start).count() / repeatCount;
    std::cout << k << " " << std::this_thread::get_id() << " " << duration << std::endl << std::flush;
    delete [] ch;
};

Среднее время её выполнения при наличии в программе лишь основного потока: 4.60426 мс

Далее выполняю данную функцию в n-ом числе потоков (в данном случае - 9):

std::vector <std::future <void>> futures;

for (int k = 0; k < 9; ++k)
{
    futures.push_back(std::async(std::launch::async, lambda, k));
}

Времена соответсвенно:

6.66124
6.86142
6.96151
7.12166
7.31183
7.32184
7.33185
7.47198
7.51201

Почему возросло время выполнения? В моей машине 6 ядер, 12 потоков. Машина во время выполнения тестов ничего другого не делала (нагрузка процессора не превышала 80%)

CPU : Ryzen 3600X 6-Core  
OS : Win10 x64  
compiler : mingw x64

Опции компилятора: g++ -c -fno-keep-inline-dllexport -O2 -std=gnu++11 -Wall -Wextra -Wextra -fexceptions -mthreads -DUNICODE -D_UNICODE -DWIN32 -DMINGW_HAS_SECURE_API=1 -DQT_DEPRECATED_WARNINGS -DQT_NO_DEBUG -DQT_CORE_LIB

UPD. На всякий случай минимально воспроизводимый пример:

#include <iostream>
#include <chrono>
#include <thread>
#include <vector>
#include <future>


int main(int argc, char *argv[])
{
    auto lambda = [](int k)
    {
        size_t size = 1920 * 1080 * 3;
        size_t repeatCount = 100;
        char* ch = new char [size];
        auto start = std::chrono::high_resolution_clock::now();
        for (int i = 0; i < repeatCount; ++i)
        {
            for (int j = 0; j < size; ++j)
            {
                ch[j] = j % 255;
            }

        }
        auto duration = std::chrono::duration<double, std::milli>(std::chrono::high_resolution_clock::now() - start).count() / repeatCount;
        std::cout << k << " " << std::this_thread::get_id() << " " << duration << std::endl << std::flush;
        delete [] ch;
    };

    lambda(-1);
    std::vector <std::future <void>> futures;

    for (int k = 0; k < 9; ++k)
    {
        futures.push_back(std::async(std::launch::async, lambda, k));
    }

    return 0;

}

UPD 2: использование сырых тредов не меняет ситуации:

std::vector <std::thread> threads;

for (int k = 0; k < 9; ++k)
{
    threads.push_back(std::move(std::thread(lambda, k)));
}
for (auto& thread : threads)
{
    thread.join();
}

5.20472
6.15558
6.22565
6.59598
6.67605
6.68598
6.73603
7.50681
7.52675
5
  • 1
    Добавьте в вопрос сразу используемые опции компилятора. Тестируемый цикл for (int i = 0; i < repeatCount; ++i) оптимизатор, по идее, должен вообще выбросить, т.к. он не влияет на наблюдаемое поведение программы...
    – wololo
    20 мар 2021 в 10:17
  • @wololo, добавил. да, я понимаю, что это не принципиально 20 мар 2021 в 10:25
  • А если просто запустить аналогичную функцию в 9-ти потоках (без всяких прибамбасов, типа future, async, launch ...)? Она ведь ничего полезного, корме вывода duration в stdout не делает
    – avp
    20 мар 2021 в 13:32
  • @avp, обновил, сути дела не меняет. как считаете, возможно ли такое, что здесь играет роль пропускная способность оперативной памяти? 21 мар 2021 в 14:10
  • imho, да, это память, точнее трафик из кэша в память. В случае одного потока мы имеем 1920х1080х3 = ~6Mbyte. Скорее всего у вас эти данные помещаются в L3 кэш (а вот 54М, требуемые 9-ти потокам туда уже не влазят и это вызывает постоянный обмен между кэшем и RAM)
    – avp
    21 мар 2021 в 14:41

3 ответа 3

1

Это не ответ на вопрос, а скорее ответ @avp на предложение запустить в обычных потоках

А если просто запустить аналогичную функцию в 9-ти потоках (без всяких прибамбасов, типа future, async, launch ...)?

Я, как очень ленивый человек, решил попробовать вместо обычных потоков OpenMP (думаю, тут не будет принципиальных отличий). И еще, немного изменил программу: cout довольно неприлично ведет себя при выводе из нескольких потоков одновременно (выводит кашу из чисел), поэтому я для вывода использую printf(). Вот модифицированный код примера:

#include <iostream>
#include <chrono>
#include <thread>
#include <vector>
#include <future>
#include <cstdio>
#include <omp.h>
#include <sstream>


int main()
{
    const int n_threads = 8;
    
    auto lambda = [](int k)
    {
        size_t size = 1920 * 1080 * 3;
        size_t repeatCount = 100;
        char* ch = new char [size];
        std::stringstream ss;
        ss << std::this_thread::get_id();
        auto start = std::chrono::high_resolution_clock::now();
        for (size_t i = 0; i < repeatCount; ++i)
        {
            for (size_t j = 0; j < size; ++j)
            {
                ch[j] = j % 255;
            }

        }
        auto duration = std::chrono::duration<double, std::milli>(std::chrono::high_resolution_clock::now() - start).count() / repeatCount;
        //std::cout << k << "\tThread id:" << std::this_thread::get_id() << "\tTime: " << duration << " ms" << std::endl; /// << std::flush;
        std::printf("k = %d\tThread id: %s\tTime: %lf ms\n",k, ss.str().c_str(),duration);
///        std::printf("k = %d\tThread id: %d\tTime: %lf ms\n",k, omp_get_thread_num(),duration);
        delete [] ch;
    };

    std::printf("Linear version:\n");
    lambda(-1);
    std::printf("\nAsync version:\n");
    std::vector <std::future <void>> futures;

    for (int k = 0; k < n_threads; ++k)
    {
        futures.push_back(std::async(std::launch::async, lambda, k));
    }

    for (int k = 0; k < n_threads; ++k)
        futures[k].wait();


    printf("\n\nOMP version:\n");

    #pragma omp parallel for num_threads(n_threads)
    for (int k = 0; k < n_threads; ++k)
    {
        lambda(k);
    }
    
    return 0;

}

Программу собирал командой: g++ -O2 -Wall -o "futures" "futures.cpp" -fopenmp -pedantic -Wextra, компилятор GCC: gcc version 7.3.0 (x86_64-posix-seh-rev0, Built by MinGW-W64 project).

У меня 4-х ядерный Core i5, поэтому оптимальным для него было бы 4 потока... Но почему-то нет. Даже при 4-х потоках время выполнения <измеряемой части> функции в асинхронных потоках чуть больше, чем при однократном запуске функции. И только при двух потоках это время получается одинаковым.

Привожу пример расчетов для 8 потоков (время при асинхронных вызовах растет от количества потоков, но не линейно: при 100 потоках время было в диапазоне от 20 до 35мс для разных потоков).

введите сюда описание изображения

PS: с асинхронными потоками раньше дел не имел, поэтому сильно не пинайте, если не верно понял идею методов wait() с ожиданием завершения всех таких потоков :-).

UPD: заметил, что не изменяю количество потоков для OpenMP (по-умолчанию, OpenMP задет количество потоков равным количеству логических процессоров, т.е. у меня было всё время 4). Добавил в директиву num_threads(n_threads) и всё стало заметно хуже: для обычных потоков время выполнения блока тоже растет. Так что, вопрос остается открытым: что не так с тем тестируемым блоком кода и параллельными потоками?.. По идее, размер кванта времени в windows 10 должен быть около 25-30 мс, т.е. переключение потоков не должно бы мешать выполнению блока кода, "длиной" в 5 мс... Есть подозрение, что мы наблюдаем "квантовые эффекты", когда сам наблюдатель вносит изменения в эксперимент. Мы для измерения времени используем внутренний высокоточный таймер процессора из разных потоков одновременно - не он ли всё тормозит? Сколько "одновременных" таймеров может нам выделить CPU? И возможно, этот таймер работает через мютекс (допускаю даже, что он аппаратный).

UPD2: Из интереса, попробовал 100 потоков снова (с исправленной директивой для указания количества потоков). И тут оказалось, что OpenMP очень сильно проигрывает: у него получается время 135 мс (в среднем, диапазон от 130 до 140) против 30 мс с async. Наверное, нужно таки и честные потоки проверить.

3
  • Отличный результат, видимо future async задуман для чего-то другого, чем просто параллельное исполнение кода.
    – avp
    20 мар 2021 в 15:48
  • @avp, прошу прошения, немного накосячил :-) - сейчас исправлю и это будет хуже. Я же не задаю число потоков в OpenMP... А по умолчанию оно мне всегда ставит 4 потока (4 ядра же).
    – Vladimir
    20 мар 2021 в 16:03
  • @Vladimir, спасибо за ваши мысли, однако предположения не оправдались. с сырыми тредами ситуация не меняется (обновил пост). что касается таймера, то сделал эксперимент, где таймер вызывается лишь для одного из потоков (выкладывать не стал, чтобы пост не разросся до неимоверных размеров) в общем на время выполнения не влияет 21 мар 2021 в 14:12
1

Это пока тоже не ответ на вопрос. Я так же заинтересовался этим вопросом и решил поэкспериментировать. Для чистоты эксперимента я модифицировал, вернее почти полностью переписал код, постаравшись убрать все моменты, которые могли бы влиять на увеличение времени выполнения потоков, а именно: 1) отказался от использования библиотек std в пользу прямой работы с WinAPI; 2) синхронизировал потоки таким образрм, чтобы измеряемый цикл начинал выполнялся только после того, как запустятся все потоки, а сами потоки не завершались прежде, чем завершится последний цикл; 3) операторы new[], delete[] и вывод результатов вынес за пределы функции потока. Также добавил информативности в вывод.

В общем, вот код:

#include <iostream>
#include <assert.h>
#include <windows.h>

#define NUM_THREADS         12
#define DATA_SIZE           (1920*1080*3)
#define MESURE_COUNT        100

#define INLINE __attribute__ ((always_inline)) inline


INLINE int GetCpuCoreId()
{
    int cpu_core_id;

    asm (
        ".intel_syntax noprefix\n"
        "   rdtscp              \n"
        "   mov %0,ecx          \n"
        :"=rm"(cpu_core_id)
        :
        :"eax","edx","ecx"
    );

    return cpu_core_id;
}


#define MS_PER_TICKS    0.0001

class CTimer
{
    union _TIME
    {
        FILETIME    fileTime;
        long long   qwTime;
    };

    typedef void WINAPI(*_PFN_GSTPAFT)(LPFILETIME lpSystemTimeAsFileTime);
    static _PFN_GSTPAFT s_pfnGetSystemTimePreciseAsFileTime;

    _TIME   m_StartTime;
    _TIME   m_CurrentTime;

public:
    static INLINE bool GlobalInit()
    {
        HINSTANCE hKernel32 = GetModuleHandle("kernel32.dll");
        if (hKernel32)
        {
            s_pfnGetSystemTimePreciseAsFileTime = (_PFN_GSTPAFT)GetProcAddress(hKernel32,"GetSystemTimePreciseAsFileTime");
            if (s_pfnGetSystemTimePreciseAsFileTime) return true;
        }
        return false;
    }

    INLINE void Start() 
    {
        assert (s_pfnGetSystemTimePreciseAsFileTime!=NULL);
        s_pfnGetSystemTimePreciseAsFileTime(&m_StartTime.fileTime);
    }

    INLINE double GetTime()
    {
        assert (s_pfnGetSystemTimePreciseAsFileTime!=NULL);
        s_pfnGetSystemTimePreciseAsFileTime(&m_CurrentTime.fileTime);
        return (double)(m_CurrentTime.qwTime-m_StartTime.qwTime)*MS_PER_TICKS;
    }
};

CTimer::_PFN_GSTPAFT CTimer::s_pfnGetSystemTimePreciseAsFileTime = NULL;


UINT volatile g_uNumThreads;
LONG volatile g_nThreadCount;
double g_lfBaseDuration = 0.0;

struct THREAD_TAG
{
    //input data
    char*   pData;
    HANDLE  hThread;

    //output data
    DWORD   dwThreadId;
    int     nPriority;
    int     nCpuCoreId;
    double  lfDuration;

    INLINE THREAD_TAG()
    {
        pData = new char[DATA_SIZE];
    }

    INLINE ~THREAD_TAG()
    {
        delete[] pData;
    }
};

void thread_func(THREAD_TAG* pTag);
double test_duration(char* pData, size_t uDataSize);
void write_thread_result(int nIndex, THREAD_TAG* pTag);


int main()
{
    THREAD_TAG t1;
    THREAD_TAG t[NUM_THREADS];

    if (!CTimer::GlobalInit()) return -1;  //need Win8 or higher

    g_uNumThreads = 1;
    g_nThreadCount = 0;
    thread_func(&t1);
    write_thread_result(-1,&t1);

    puts("");
    g_uNumThreads = NUM_THREADS;
    g_nThreadCount = 0;
    for (int i=0; i<NUM_THREADS; ++i)
        t[i].hThread = CreateThread(NULL,0,(LPTHREAD_START_ROUTINE)thread_func,(LPVOID)&t[i],0,NULL);

    for (int i=0; i<NUM_THREADS; ++i)
        if (t[i].hThread) WaitForSingleObject(t[i].hThread,INFINITE);

    double lfAvgDuration = 0;
    for (int i = 0; i<NUM_THREADS; ++i)
    {
        write_thread_result(i,&t[i]);
        lfAvgDuration+= t[i].lfDuration;
    }

    lfAvgDuration/= NUM_THREADS;
    puts("");
    std::cout << "average:\t\t\t\t\t\t" 
              << lfAvgDuration << "\t(" << lfAvgDuration/g_lfBaseDuration*100 << "%)" 
              << std::endl << std::flush;

    return 0;
}


void thread_func(THREAD_TAG* pTag)
{
    InterlockedIncrement(&g_nThreadCount);
    while (g_nThreadCount<g_uNumThreads);

    pTag->lfDuration = test_duration(pTag->pData,DATA_SIZE);

    InterlockedDecrement(&g_nThreadCount);
    while (g_nThreadCount>0);

    if (g_lfBaseDuration==0.0) g_lfBaseDuration = pTag->lfDuration;
    pTag->dwThreadId = GetCurrentThreadId();
    pTag->nPriority = GetThreadPriority(GetCurrentThread());
    pTag->nCpuCoreId = GetCpuCoreId();
};


double test_duration(char* pData, size_t uDataSize)
{
    CTimer timer;

    timer.Start();
    for (int i = 0; i < MESURE_COUNT; ++i)
        for (int j = 0; j < uDataSize; ++j)
            pData[j] = j % 255;
    return timer.GetTime()/MESURE_COUNT;
}


void write_thread_result(int nIndex, THREAD_TAG* pTag)
{
    std::cout << nIndex << "\t\t" << pTag->dwThreadId
          << "\t(priority: " << pTag->nPriority << ", core: " << pTag->nCpuCoreId << ")\t\t" 
          << pTag->lfDuration << "\t(" << pTag->lfDuration/g_lfBaseDuration*100 << "%)" 
          << std::endl << std::flush;
}

А теперь результаты. У меня процессор AMD Ryzen 5 2600X, 6 физических ядер, 12 виртуальных.

Запустил программу с NUM_THREADS==6, и получил увеличение времени выполнения в среднем на 18% (при каждом запуске показатель варьировался где-то от 10% до 25%). Вот скрин: введите сюда описание изображения

Но при запуске с 12 потоками, всякий раз получал увеличение времени выполнения практически ровно в два раза, т.е. так, как по идее должно было быть, будь у меня ровно 6 ядер без гипертрединга: введите сюда описание изображения

В общем, какие у кого есть ещё идеи? Мои предположения: такое увеличение времени при параллельном выполнении может быть связано с конкуренцией за шину данных при обращениях к памяти, либо с какими-то эффектами от кэша. Но это только лишь предположения...

1
  • Забыл предупредить: программа скомпилируется только на GCC (из-за ассемблерной вставки) и запустится на винде 8+ (из-за вызова GetSystemTimePreciseAsFileTime()).
    – LShadow77
    23 мар 2021 в 20:12
0

Мое предположение друзья, что все из за RAM и контроллера, он не дает возможности предоставить оперативку всем потокам одновременно. На RAM с двумя каналами пошло ускорение! Ну и как добавили - пресловутый кеш-коллектор.

1
  • 1
    добро пожаловать на stack overflow на русском! пожалуйста, постарайтесь оставлять чуть более развёрнутые ответы. дополнить ответ можно, нажав править 6 дек 2022 в 13:16

Ваш ответ

By clicking “Отправить ответ”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Всё ещё ищете ответ? Посмотрите другие вопросы с метками или задайте свой вопрос.