Простая многопоточная помощь? C++, WaitForSingleObject и синхронизация

Question

Итак, я пытаюсь создать простую многопоточную программу, которая проверяет гипотезу Collatz для большого набора чисел и возвращает общее количество проверенных номеров. Каждый поток (всего 4) выполняет интервал чисел и обновляет «проверенную» переменную, когда число достигает 1. Я также определяю весь процесс (для сравнения с одним потоковым вычислением)

Проблема, что никогда не бывает согласованности, когда я печатаю «проверенный» int в конце программы, поэтому я предполагаю, что ни потоки пишут друг над другом, либо основной поток заканчивается перед другими, тем самым печатая неполное число. Я также предполагаю, что вычисления clock() также будут выключены, если основной поток завершится перед другими. Итак, как мне остановить главный поток от продолжения до тех пор, пока другие потоки не будут завершены (что заставило бы ждать обновления проверенного счета и завершения точного измерения времени)? Это то, что я думал WaitForSingleObject, но я предполагаю, что он просто останавливает основной поток от EXITING, все еще позволяя ему вычислять другие его функции.

Это мой первый переход на любую многопоточность, и я не думаю, что я вполне понимаю работу синхронизации и команду WaitForSingleObject. Вот что у меня есть до сих пор в моей основной функции:

EDIT: Вот моя обновленная функция Main и функция Collatz. Я изменил его так, чтобы каждый поток обращался к отдельной переменной, чтобы избежать проблемы синхронизации, но проблема все еще сохраняется. Там нет никакого последовательного значения, когда я распечатать «подтверждено» *

EDIT СНОВА: Хорошо, так что я удалил «нить» ИНТ на Младен Янкович, а просто использовал простой счетчик, чтобы распределить различные интервалы созданные потоки. В настоящее время существует согласованное, правильное значение для «проверенных». ОДНАКО, я до сих пор не могу заставить программу фактически закончить, когда есть 1 000 000 номеров. Тестирование его на 100 000 или даже 10000 работает безупречно, но когда я до него до 1 000 000 номеров, программа работает неопределенно (часы) без фактического возврата значений. Я предполагаю, что он застревает в определенном значении (например, 750831, как указал Мартин Джеймс). Я попытался заменить int на long int, но кажется, что он все еще страдает от переполнения. Какие-либо предложения? И спасибо за огромную помощь.

ПОСЛЕДНИЙ EDIT: Хорошо, так что я просто использовал долго долго вместо междунар и теперь программа работает безупречно. Спасибо всем за помощь!

void main() 
{ 
    clock_t start; 
    clock_t finish; 
    unsigned int thread = 0; 

    start = clock(); 

    HANDLE h1 = (HANDLE)_beginthreadex(NULL, 0, collatz_thread, NULL, 0, NULL); 

    HANDLE h2 = (HANDLE)_beginthreadex(NULL, 0, collatz_thread, NULL, 0, NULL); 

    HANDLE h3 = (HANDLE)_beginthreadex(NULL, 0, collatz_thread, NULL, 0, NULL); 

    for (int i = 750001 ; i <= 1000000 ; i++) { collatz(i, 4); } 

    WaitForSingleObject(h1, INFINITE); 
    WaitForSingleObject(h2, INFINITE); 
    WaitForSingleObject(h3, INFINITE); 

    finish = clock() - start; 
    double time = finish/(double) CLOCKS_PER_SEC; 

    validated = v1 + v2 + v3 + v4; 
    cout << validated << " numbers validated." << endl; 
    cout << endl << time << " seconds." << endl; 
} 

unsigned _stdcall collatz_thread (void* n) 
{ 
    selection++; // selects a different interval each time collatz_thread is called 

    switch (selection) { 
    case 1: 
     for (int i = 1 ; i <= 250000; i++)  { collatz(i, 1); } 
     break; 
    case 2: 
     for (int i = 250001 ; i <= 500000; i++) { collatz(i, 2); } 
     break; 
    case 3: 
     for (int i = 500001 ; i <= 750000; i++) { collatz(i, 3); } 
     break; 
    } 
    return 0; 
} 

int collatz (int n, int thread) 
{ 
    int original = n; 

    while (n != 1) { 
    if (n%2 == 0) 
     n = (n/2); 
    else 
     n = (3*n + 1); 
    } 

    if (n == 1) { 
    switch (thread) { 
     case 1: 
      v1++; 
      break; 
     case 2: 
      v2++; 
      break; 
     case 3: 
      v3++; 
      break; 
     case 4: 
      v4++; 
      break; 
    } 
    return n; 
} 

}

Answer 1

Вам необходимо синхронизировать доступ к validated, если она разделяет переменную. Легким способом является использование функции InterlockedIncrement вместо стандартного оператора ++, если вы хотите его увеличить. Другим подходом является использование какого-то объекта синхронизации, такого как spinlock или mutex, при доступе к общей переменной, но это слишком сложно, если вам просто нужно синхронизировать операцию увеличения.

Если вы хотите получить более подробную информацию, предоставьте код collatz функции.

Как предлагается «usr», для лучшей производительности вы можете использовать отдельную переменную для каждого потока, а затем суммировать их в основном потоке. В этом случае вы должны поместить эти переменные таким образом, чтобы они не делили одну и ту же линию кэша, чтобы избежать false sharing.

Вы не указали функцию collatz_thread, что может стать еще одной причиной непоследовательных результатов. Причина в том, что вы передаете указатель на переменную (&thread), в которой хранится поток #, который изменяется между вызовами, которые создают новые потоки, поэтому в зависимости от состояния планировщика ОС новые потоки могут не получить возможность запускать, а переменная thread уже изменилось, чтобы иметь другое значение, поэтому у вас будет больше одного потока, делающего один и тот же набор данных, тогда как другие наборы могут быть пропущены. Поскольку поведение зависит от текущего состояния планировщика потоков, это в значительной степени непредсказуемо.

Раствор отливают thread переменной void* вместо передачи его адрес, а затем в collatz_thread функции брось назад int:

HANDLE h1 = (HANDLE)_beginthreadex(NULL, 0, collatz_thread, (void*)thread, 0, NULL);

И как предложил Мартин, вы, возможно, имеют целочисленное переполнение, но не должны вызывать противоречивые результаты, просто неправильные результаты, но, тем не менее, непротиворечивые.

Answer 2

Попробуйте взглянуть на это:

Semaphore and threads explenation from MSDN

Это, пожалуй, лучшая документация вы найдете в Интернете.

Теперь, в связи с вашим кодом, я предполагаю, что он работает не очень хорошо, и именно по этой причине: WaitForSingleObject - В основном это означает, что вы пытаетесь сделать -1 на семафоре h1 (или h2 или h3) и если вы не можете сделать это -1 (т.е. семафор находится в 0), тогда подождите бесконечное время. WaitForSimgleObject действительно должен быть вызван в вашей потоковой программе, а не в вас.

Кроме того, если вы закончили работу с общей переменной, вам нужно освободить семафор, чтобы другой поток мог получить блокировку на этом конкретном Семафоре.

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

Answer 3

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

8 tests, 
8 tasks, 
counting to 1000000, 
using 14 threads: 
Validated: 1000000 in 670ms 
Validated: 1000000 in 671ms 
Validated: 1000000 in 624ms 
Validated: 1000000 in 656ms 
Validated: 1000000 in 655ms 
Validated: 1000000 in 655ms 
Validated: 1000000 in 640ms 
Validated: 1000000 in 686ms 
Average time: 657ms 
Total validated: 8000000 

8 tests, 
8 tasks, 
counting to 10000000, 
using 14 threads: 
Validated: 10000000 in 8081ms 
Validated: 10000000 in 7441ms 
Validated: 10000000 in 7784ms 
Validated: 10000000 in 7598ms 
Validated: 10000000 in 7956ms 
Validated: 10000000 in 7534ms 
Validated: 10000000 in 7816ms 
Validated: 10000000 in 7769ms 
Average time: 7747ms 
Total validated: 80000000 

Обратите внимание, что, хотя он говорит, что 14 потоков, то есть для всего пула Один потока всегда ожидание для других задач, чтобы закончить, так что только 13 потоки были фактически доступны для запуска проверки. По какой-то причине это было оптимальным.

ОК, мой i7 плоский на всех 4/8 сердечниках, но я не вижу времени работы от ушей краска в секунды, потому что у меня больше ядер и отделил работу от них :(

Это то, что я использовал. Это немного отличается от того, как вы это делали, потому что у меня было большинство инструментов/кода. Это Visual C++, для начала. Есть два класса. Каждый запуск управляется классом «PoolTest», который передает несколько экземпляров «TestTask» в threadpool, по одному для каждого сегмента полного диапазона целых чисел, подлежащих проверке. Вы заметите, что ваш код скопирован/вставлен в класс TestTask. Я выделил, где TestTask собран в коде PoolTest.Класс «PoolTest» затем ожидает Событие для всех экземпляров «TestTask» для завершения - он может это сделать, потому что вызов TestTask возвращается к методу «TaskComplete» PoolTest при завершении. Этот метод обращается к потокобезопасному счетчику, который подсчитывается, когда тестовые задачи отправляются и подсчитываются методом «taskComplete».

Этот код, который я повторно использовал, немного усложняет ситуацию, потому что он может повторять пробег несколько раз, чтобы получить среднее время, поэтому полный набор TestTasks может выдаваться несколько раз.

Когда последний тест TestTask отсчитывает до нуля, он сигнализирует о событии, и PoolTest затем снова запустит сигнализацию завершения всего тестового прогона в графический интерфейс (не беспокоит листинг содержимого графического интерфейса пользователя, что не имеет значения).

using System; 
using System.Collections.Generic; 
using System.Linq; 
using System.Text; 
using System.Threading; 


namespace WindowsFormsApplication1 
{ 

public class TestTask: Task{ 
public int validated; 
public int fromVal, toVal; 
public int ticks; 

    long collatz(long n) 
    { 
     while (n != 1) 
     { 
      if (n % 2 == 0) 
       n = (n/2); 
      else 
       n = (3 * n + 1); 
     } 
     return (n); 
    } 

    public override void run(){ 
     int index; 
     int localTo = toVal; 
     int localFrom = fromVal; 
     int localVal = 0; 
     for (index = localFrom; index < localTo; index++) 
     { // if not disproved, inc the stack 'validated' 
      if (1 == collatz(index + 1)) localVal++; 
     } 
     validated = localVal; // put stack result into instance field, 
    } 
    public TestTask(int paramx, EventHandler OnDone) 
     : base(paramx, OnDone){} 
}; 


/* a task that submits testTask instances. 
*/ 
public class PoolTest:Task{ 
    int FnumTasks; 
    int FnumTests; 
    int Fcount; 
    int FtestCount; 
    int taskCount; 
    int startTicks; 
    int endTicks; 
    int totalTicks; 
    EventHandler FonTaskComplete; 
    AutoResetEvent testCompleteEvent; 
    public int average; 
    public int testTicks; 
    public int Vone; 
    public int Vtot; 
    public TestTask thisTestTask; 

    public PoolTest(int testsNum, int tasks, int count, EventHandler taskDone, 
     EventHandler testDone) 
     : base(0, testDone) 
    { 
     FnumTests=testsNum; 
     FtestCount=testsNum; 
     FnumTasks=tasks; 
     Fcount=count; 
     Vtot = 0; 
     Vone = 0; 
     totalTicks = 0; 
     FonTaskComplete=taskDone; // call after each test to report ticks 
     testCompleteEvent= new AutoResetEvent(false); 
    } 
    void submitAtest(){ // queue up numTasks testTask instances 
     taskCount=FnumTasks; 
     startTicks = System.Environment.TickCount; 

//*********************THIS IS WHERE THE RANGES AND TASKS ARE ASSEMBLED 

     int startNum = 0; // here, start at 0 and build up the ranges 
     int countIncrement=Fcount/FnumTasks; // calc. range size 
     int endNum=startNum+countIncrement; // and so init. start/end 
     TestTask newTask; 
     for (int i = 1; i < FnumTasks; i++) // one less than requested 
     { 
      newTask=new TestTask(0, taskComplete); 
      newTask.fromVal=startNum; // load in the start/end for the loop 
      newTask.toVal = endNum; 
      myPool.submit(newTask);  // off it goes, see you later 
      startNum = endNum;   // now move range up for 
      endNum += countIncrement;  // next TestTask 
     } 
     // treat last range separately to cover div rounding when 
     // calculating 'countIncrement' 
     newTask = new TestTask(0, taskComplete); // do last one 
     newTask.fromVal = startNum; 
     newTask.toVal = Fcount; 
     myPool.submit(newTask); // send off the last one 
    } 

//***************************************************************** 

    public override void run(){ 
     submitAtest(); //start off the first run of testTasks 
     testCompleteEvent.WaitOne(); 
    } 
    void taskComplete(object sender, EventArgs e){ // called when a testTask 
     bool finishedTasks;       // instance is complete 
     lock (this) 
     { 
      thisTestTask = (TestTask)sender; 
      taskCount--;       // another one down 
      Vone += thisTestTask.validated;   // Vone - total for one run 
      Vtot += thisTestTask.validated;   // total for all runs 
      finishedTasks = (taskCount == 0);  // this run all done yet? 
      if (finishedTasks) 
      { 
       endTicks = System.Environment.TickCount; // yes, so calc. elapsed time 
       testTicks=endTicks-startTicks; 
       thisTestTask.ticks = testTicks; 
       totalTicks=totalTicks+testTicks; 
       if (0!=--FtestCount) {     // done all the test runs? 
        thisTestTask.validated = Vone;  // use this field to return run total 
        FonTaskComplete(thisTestTask, null); // and signal result of test 
        Vone = 0; 
        submitAtest();      // no, so send off another load 
       } 
       else{ 
         average=totalTicks/FnumTests;  // done all test runs! 
         testCompleteEvent.Set();   // signal all runs completed 
        }; 
      } 
     } 
    } 
}; 
}