Завершение потока в C++ из другого потока

Question

В основном я нахожу простые числа с двумя потоками. Я разбил диапазон возможных простых чисел пополам для каждого потока или иначе статически распределял диапазон между потоками. Однако поток, который должен иметь дело с меньшими числами, неизбежно заканчивается до того, который вычисляет более крупные. То, что я хочу сделать, - это как только нить проходит через его диапазон, прекратите оба потока, а затем отдайте половину остальной части потока, который не был доведен до того, который сделал так, чтобы они рекурсивно выровнялись и всегда будет работать параллельно. Ex: A (1-100) и B (100-200), A заканчивается первым, а B все еще на 150. Оба останавливаются, A начинается как A (150-175) и B, как B (175-200).

Вот мой код до сих пор:

#include <windows.h> 
#include <stdio.h> 
#include <process.h> 
#include <queue> 
#include <cmath> 
#include <iostream> 
using namespace std; 
priority_queue<int> primes; 
CRITICAL_SECTION critical; 
struct args 
{ 
    int begin; 
    int end; 
}par1, par2; 

int e_prosto(int n) 
{ 
    for(int i = 2; i*i<(n + 1) ; i++) 
    if (n & 1 == 0 || n % i == 0) return 0; 
    return 1; 
} 

unsigned int __stdcall rabotnik(void* n) 
{ 
    struct args *lPar = (args*) n; 
    for(int i = lPar->begin; i < lPar->end; i++) 
    { 
     if(e_prosto(i)){ 
      EnterCriticalSection(&critical); 
      primes.push(i); 
      LeaveCriticalSection(&critical); 
     } 
    } 
    return 0; 
} 

int main() 
{ 
    int foo; 
    printf(" Tarsene na prosti do: "); 
    scanf("%d", &foo); 
    par1.begin=1; 
    par1.end=foo/2+1; 
    par2.begin=foo/2+1; 
    par2.end=foo; 

    HANDLE hvadkaA, hvadkaB; 
    InitializeCriticalSection(&critical); 
    SYSTEMTIME st, now, then; 

    hvadkaA = (HANDLE)_beginthreadex(0, 0, &rabotnik, (void*)&par1, 0, 0); 
    hvadkaB = (HANDLE)_beginthreadex(0, 0, &rabotnik, (void*)&par2, 0, 0); 
    ::GetSystemTime(&then); 
    WaitForSingleObject(hvadkaA, INFINITE); 
    WaitForSingleObject(hvadkaB, INFINITE); 

    CloseHandle(hvadkaA); 
    CloseHandle(hvadkaB); 

    ::GetSystemTime(&now); 
    while(!primes.empty()) 
    { 
     printf("%d \t", primes.top()); 
     primes.pop(); 
    } 
    printf("\nGotov za %d milisec", abs(now.wMilliseconds - then.wMilliseconds)); 
    system("pause>nul"); 
    return 0; 
} 

Answer 1

Я предлагаю вместо этого дать каждому потоку один блок A (1-100) и B (101-200), которым каждому потоку назначается по модулю. Например, A возьмет все нечетные индексы, B возьмет все четные индексы, в результате получится A {1,3,5,7,9, ..., 191,193,195,197,199} и B {2,4,6,8, ..., 190.192.194.196.198.200}. Вероятно, это самый быстрый и простой способ загрузить баланс потоков, поскольку сложность вычислений будет равномерно распределена.

Следующее предложение - добавить bool, который указывает, нормально ли продолжать обработку. Затем перед началом каждого расчета вы проверяете, нормально ли оно действовать. Таким образом, вы можете остановить вычисление без завершения потока, за счет одного дополнительного сравнения каждого цикла.

#include <windows.h> 
#include <stdio.h> 
#include <process.h> 
#include <queue> 
#include <cmath> 
#include <iostream> 
using namespace std; 
bool run; 
priority_queue<int> primes; 
CRITICAL_SECTION critical; 
struct args 
{ 
    int begin; 
    int end; 
}par1, par2; 

int e_prosto(int n) 
{ 
    for(int i = 2; i*i<(n + 1) ; i++) 
    if (n & 1 == 0 || n % i == 0) return 0; 
    return 1; 
} 

unsigned int __stdcall rabotnik(void* n) 
{ 
    struct args *lPar = (args*) n; 
    for(int i = lPar->begin; i < lPar->end && run; i++) 
    { 
     if(e_prosto(i)){ 
      EnterCriticalSection(&critical); 
      primes.push(i); 
      LeaveCriticalSection(&critical); 
     } 
    } 
    run = false; 
    return 0; 
} 

int main() 
{ 
    int foo; 
    printf(" Tarsene na prosti do: "); 
    scanf("%d", &foo); 
    par1.begin=1; 
    par1.end=foo/2+1; 
    par2.begin=foo/2+1; 
    par2.end=foo; 
    run = true; 

    HANDLE hvadkaA, hvadkaB; 
    InitializeCriticalSection(&critical); 
    SYSTEMTIME st, now, then; 

    hvadkaA = (HANDLE)_beginthreadex(0, 0, &rabotnik, (void*)&par1, 0, 0); 
    hvadkaB = (HANDLE)_beginthreadex(0, 0, &rabotnik, (void*)&par2, 0, 0); 
    ::GetSystemTime(&then); 
    WaitForSingleObject(hvadkaA, INFINITE); 
    WaitForSingleObject(hvadkaB, INFINITE); 

    CloseHandle(hvadkaA); 
    CloseHandle(hvadkaB); 

    ::GetSystemTime(&now); 
    while(!primes.empty()) 
    { 
     printf("%d \t", primes.top()); 
     primes.pop(); 
    } 
    printf("\nGotov za %d milisec", abs(now.wMilliseconds - then.wMilliseconds)); 
    system("pause>nul"); 
    return 0; 
} 

Другой способ заключается в разделении вашего диапазона на множество блоков, а затем, когда поток завершает работу, он даст новый блок для обработки. Это имеет преимущество, заключающееся в том, что вы не добавляете дополнительные затраты на вычисление, но требует немного больше кода (так что вы повторно используете потоки и слушаете какой-либо поток, а не только один). Чтобы иметь какую-либо ценность, вам, вероятно, понадобится больший диапазон, и вам может потребоваться изменить размер блока по сложности (размеры блоков {1-100}, {101-150}, {151-175}, {176-183}, {184-187}, ...).Быстрый пример использования коды (с симметричными размерами блоков):

#include <windows.h> 
#include <stdio.h> 
#include <process.h> 
#include <queue> 
#include <cmath> 
#include <iostream> 
using namespace std; 
priority_queue<int> primes; 
CRITICAL_SECTION critical; 
typedef struct args 
{ 
    int begin; 
    int end; 

    //Helper method for initalizing struct 
    void setAll(int inBegin, bool inEnd) 
    { 
    } 

} *PArgs; 

int e_prosto(int n) 
{ 
    for(int i = 2; i*i<(n + 1) ; i++) 
    if (n & 1 == 0 || n % i == 0) return 0; 
    return 1; 
} 

static DWORD WINAPI rabotnik(LPVOID lpParam) 
{ 
    struct args *lPar = (args*) lpParam; 
    for(int i = lPar->begin; i < lPar->end; i++) 
    { 
     if(e_prosto(i)){ 
      EnterCriticalSection(&critical); 
      primes.push(i); 
      LeaveCriticalSection(&critical); 
     } 
    } 
    return 0; 
} 

int main() 
{ 
    const int NUM_THREAD = 2;   //Use named constant incase you want to change later. 
    DWORD returnedThreadID; 
    DWORD threadID[NUM_THREAD]; 
    HANDLE threadHandle[NUM_THREAD]; //Holds the handels for the threads. 
    int foo,       //Range size. 
     fooBlockSize,     //Number of elements in a block. 
     nextBlock; 
    PArgs par[NUM_THREAD]; 

    printf(" Tarsene na prosti do: "); 
    scanf("%d", &foo);     //Get range size from user. 
    fooBlockSize = foo/(NUM_THREAD * 10); //Set number of elements in a block. 

    InitializeCriticalSection(&critical); 
    SYSTEMTIME st, now, then; 

    for (int i = 0; i < NUM_THREAD; i++) 
    { 
     par[i] = (PArgs) HeapAlloc(GetProcessHeap(), HEAP_ZERO_MEMORY, sizeof(PArgs)); 
      // If the allocation fails, the system is out of memory so terminate execution. 
     if(par[i] == NULL){ cout<<"par HeapAlloc failed with error# "<<GetLastError()<<endl<<"Will now quit."<<endl; ExitProcess(2);} 
    } 

    for(int i = 0; i < NUM_THREAD; i++) 
    { 
     par[i]->begin = (fooBlockSize * i) + 1; 
     par[i]->end = par[i]->begin + fooBlockSize; 
     threadHandle[i] = CreateThread(NULL, 0, rabotnik, par[i], CREATE_SUSPENDED, &threadID[i]); 
    } 
    nextBlock = NUM_THREAD; 

    ::GetSystemTime(&then); 
    for (int i = 0; i < NUM_THREAD; i++) 
    { 
     ResumeThread(threadHandle[i]);  //Start threads 
    } 

    while(((fooBlockSize * nextBlock) + 1) < foo) 
    { 
     returnedThreadID = WaitForMultipleObjects(NUM_THREAD, threadHandle, false, INFINITE); //Wait for a thread to complete. 
     for(int i = 0; i<NUM_THREAD;i++) 
     { 
      if(returnedThreadID = threadID[i]) 
      { 
       //Update the thread's arguments with the new block. 
       par[i]->begin = (fooBlockSize * nextBlock) + 1; 
       par[i]->end = par[i]->begin + fooBlockSize; 

       //Restart the thread. 
       ResumeThread(threadHandle[i]); 
       nextBlock++; 
      } 
     } 
    } 

    for (int i = 0; i < NUM_THREAD; i++) 
    { 
     //Return heap memorry (good practice, though Windows should return it all when the process terminates). 
     if (HeapFree(GetProcessHeap(), 0, par[i]) == 0) 
     { 
      cout<<"HeapFree failed for par["<<i<<"]"<<endl; 
     } 

     //Not sure we need to close the threads, but it was in original version. 
     CloseHandle(threadHandle[i]); 
    } 

    ::GetSystemTime(&now); 
    while(!primes.empty()) 
    { 
     printf("%d \t", primes.top()); 
     primes.pop(); 
    } 
    printf("\nGotov za %d milisec", abs(now.wMilliseconds - then.wMilliseconds)); 
    system("pause>nul"); 
    return 0; 
} 

Существует компромисс между увеличением числа блоков против размера блока. При увеличении количества блоков означает, что будет меньше времени, затраченного только на один блок, способный обрабатывать (поток [0] завершается, и нет ничего, что можно было бы обработать, пока поток [1] ​​заканчивается), но также означает, что мне больше времени уделялось ожиданию цикла диспетчера, чтобы назначить новый блок для обработки. С вашим заявлением о проблемах я ожидаю, что это займет так много времени, чтобы найти простые числа, которые не имеют значения.

Как уже отмечалось в других ответах, не используйте один и тот же стек для хранения основных значений, найденных каждым потоком (количество времени, которое требуется для работы блокировки, будет чрезмерным). Если вы хотите, чтобы простые числа возвращались в числовом порядке, я предлагаю либо переписать цикл, который печатает простые числа, так что он проходит через оба стека одновременно, печатая следующее значение (по порядку). Некоторые вещи, как:

while(!primes1.empty() && !primes2.empty()) 
    { 
     if(primes1.top() > primes2.top()) 
     { 
      printf("%d \t", primes1.top()); 
      primes1.pop(); 
     } 
     else 
     { 
      printf("%d \t", primes2.top()); 
      primes2.pop(); 
     } 
    } 

Вам придется иметь дело со значениями, оставшихся в одном стеке разок другой пустой (или, возможно, поместить -1 в нижней части каждого стека, так что если вы либо стек пуст, то все значения, превышающие -1, уже будут напечатаны).

Другим решением будет поддерживать отсортированный список простых чисел, который обновляется каждый раз, когда поток возвращается. Затем это можно было бы скопировать в структуру par для более быстрого первичного обнаружения (если число делится равномерно существующим простым числом, то число не является простым).

Примечание: Я не тестировал эти примеры, хотя они должны быть достаточно близко, чтобы дать вам общую идею.

Answer 2

Нагрузочный нить сильно плохая идея, так как вы это может оставить свой процесс в плохом состоянии. Если поток работает в цикле, вы можете установить некоторый флаг извне, чтобы поток мог протестировать и решить, следует ли его завершать (это можно сделать, просто выйдя из функции thead). Просто не забудьте защитить свой флаг мьютексом.

Есть и другие шаблоны, которые вы можете посмотреть. Возможно, вы захотите поставить свой ряд простых чисел в очередь. Каждый рабочий поток может вывести значения из очереди и выполнить поиск. Таким образом, вы можете равномерно разделить нагрузку, не убивая и не перезапуская потоки.

Answer 3

Вашего алгоритм прайм-находка ужасно неэффективен, но я хочу говорить о том, что сейчас ..

Threading:

Настройка очереди заданий для каждого потока .. убийство/прекращения короткоживущих потоков, особенно для первичная находка звучит как действительно плохая идея.

Избегайте конкуренции. Не используйте блокировку для primes.push(i);. установите отдельную очередь для каждого потока, нажмите там результаты. Когда нить закончила с диапазоном, , то введите критический раздел и объедините результаты. Таким образом, вы вряд ли блокируете.