Verwaltung mehrerer Threads

Hallo,
ich möchte mir in C++ ein Programm schreiben, was Rechenaufgaben mittels der C++11 Threads auf meine Kerne vereilt.
Ein Threadverwalter prüft, ob CPU Kerne unbelegt sind und ob Noch Berechnungen anstehen und spwant Threads.
Bedauerlicherweise funktioniert meine Implementation nicht. Hat jemand eine Idee, warum die Anwendung immer nur einen Thread erzeugt, anstatt 4?

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

void workThread(const uint64_t& seed) {

    // Logistische Gleichung berechen
    uint64_t numIterations = 1000000000 * (1 + rand() % 3);
    double xn = 0;
    double a = 4.0 * ((double) seed / 1000000.0);
    for(uint64_t i = 0; i < numIterations; ++ i) {
        xn = a * xn * (1.0 + xn);
    }

    std::cout << xn;
}

class Scheduler {
private:
    uint32_t numThreads = 0;

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

    bool haveTasks = true;
    uint64_t activeThreads = 0;
    const uint64_t seedStart = 0;
    const uint64_t seedEnd = 1000000;
    uint64_t seed = seedStart;

public:
    Scheduler() {
        numThreads = std::thread::hardware_concurrency();

        for(uint64_t i = 0; i < numThreads; ++ i) {
            threads.push_back(std::thread());
            threadStates.push_back(0);
        }

        activeThreads = 0;

    }

    void sheduleThreads() {

        while(haveTasks) {
            // Threads spawnen
            while(activeThreads < numThreads) {
                for(uint64_t t = 0; t < numThreads; ++ t) {
                    if(threadStates[t] == false) {
                        threads[t] = std::thread(workThread, seed);
                        seed ++;
                        activeThreads ++;
                        threadStates[t] = true;

                        std::async(std::launch::async, [this](uint64_t t){ std::cout << "join: " << t << std::endl; this->threads[t].join(); }, t);
                        break;
                    }
                }
            }

            // Pruefen, ob noch Seeds vorhanden
            if(seed >= seedEnd) {
                haveTasks = false;
            }

            // Status der Threads pruefen
            for(uint64_t t = 0; t < numThreads; ++ t) {
                if(!threads[t].joinable()) {
                    activeThreads --;
                    threadStates[t] = false;
                } else {
                    threadStates[t] = true;
                }
            }

//            std::cout << "active threads: " << activeThreads << std::endl;
            std::cout << 100.0 * (double) seed / (double) seedEnd << std::endl;
        }
    }

};

using namespace std;

int main() {
    Scheduler sh;
    sh.sheduleThreads();
    return 0;
}

Unmittelbares Problem ist, dass

std::async(std::launch::async, [this](uint64_t t) {
                                   std::cout << "join: " << t << std::endl;
                                   this->threads[t].join();
                                 },
             t);

nicht asynchron ist (der Destruktor der future "joint" die Berechnung).

Warum aber überhaupt den ganzen Overhead?

#include <boost/range/irange.hpp>
#include <boost/range/algorithm/for_each.hpp>

int main() {
  const unsigned seedStart = 0;
  const unsigned seedEnd = 1000000;
  const unsigned numTasks = 100 * std::thread::hardware_concurrency();
  const unsigned workSize = (seedEnd - seedStart + numTasks - 1) / numTasks;
  std::vector<std::future<void> > tasks;
  for (unsigned i : boost::irange(seedStart, seedEnd, workSize))
    tasks.push_back(std::async([=] {
      boost::for_each(boost::irange(i, std::min(i + workSize, seedEnd)),
                      workThread);
    }));
}

Hervorragend für die schnelle Antwort!

Wie kann ich

tasks.push_back(std::async([=] {
      boost::for_each(boost::irange(i, std::min(i + workSize, seedEnd)),
                      workThread);
    }));

ohne Boost realisieren?

Ist nur ein fancy way um eine for-Loop zu schreiben.

for (unsigned i=seedStart; i<seedEnd; i += workSize)
  tasks.push_back(std::async([=] {
        for (unsigned j=i; j<std::min(i + workSize, seedEnd); ++j)
          workThread(j);
  }));

Mit folgendem Code passiert gar nichts mehr Oo

void workThread(const uint64_t& seed) {

    // Logistische Gleichung berechen
    uint64_t numIterations = 1000;
    double xn = 1.0;
    const double a = 4.0 * ((double) seed / 1000000.0);
    for(uint64_t i = 0; i < numIterations; ++ i) {
        xn = a * xn * (1.0 + xn);
    }

    std::cout << "xn: " << xn << std::endl;
}

int main() {
  const uint64_t seedStart = 0;
  const uint64_t seedEnd = 100000000;
  const uint64_t numTasks = std::thread::hardware_concurrency();

  const uint64_t workSize = (seedEnd - seedStart + numTasks - 1) / numTasks;
  std::cout << "worksize: " << workSize << std::endl;
  std::vector<std::future<void> > tasks;
  for(uint64_t seed = seedStart; seed < seedEnd; seed += workSize) {
    tasks.push_back(std::async([=] {
        for(uint64_t t = 0; t < std::min(t + workSize, seedEnd); ++ t) {
            workThread(t);
        }
    }));
  }

    return 0;
}

Ja, sorry, hätte es testen sollen. Füg nach der Schleife noch ein

for (auto& f : tasks)
    f.get();

an, ich dachte das passiere automatisch.

Btw, du hast meinen Code verschlimmbessert. numTask sollte um einen konstanten Faktor grösser sein als die Anzahl logischer Cores (10 oder 100 ist da gut geeignet), uint64_t ist für std::thread::hardware_concurrency() der falsche Typ und was die innere for-Loop bei dir machen soll verstehe ich nicht (kurz: sie ist falsch).

@Ja, sorry, hätte es testen sollen. Füg nach der Schleife noch ein
Leider hatte ich hier kein Erfolg

int main()
{
    const uint64_t seedStart = 0;
    const uint64_t seedEnd = 1000000;
    const uint64_t numTasks = 100 * std::thread::hardware_concurrency();

    const uint64_t workSize = (seedEnd - seedStart + numTasks - 1) / numTasks;
    std::cout << "worksize: " << workSize << std::endl;
    std::vector<std::future<void> > tasks;
    for(uint64_t seed = seedStart; seed < seedEnd; seed += workSize)
    {
        tasks.push_back(std::async([=]
        {
            for(uint64_t t = 0; t < std::min(t + workSize, seedEnd); ++ t)
            {
                workThread(t);
            }
        }));
    }

    for (auto& f : tasks)
    {
        f.get();
    }

    return 0;
}

@Btw, du hast meinen Code verschlimmbessert. numTask sollte um einen konstanten Faktor grösser sein als die Anzahl logischer Cores (10 oder 100 ist da gut geeignet), uint64_t ist für std::thread::hardware_concurrency() der falsche Typ und was die innere for-Loop bei dir machen soll verstehe ich nicht (kurz: sie ist falsch).
Bezieht sich das auf folgende Schleife?

for(uint64_t t = 0; t < std::min(t + workSize, seedEnd); ++ t) {
            workThread(t);
        }

Mein Ziel:
Ich habe ein Problem, welches in Teilprobleme zerlegen kann.
Jedes Teilproblem kann durch den Seed beschrieben werden.
Das Programm soll solange n Prozesse spawnen, bis alle Probleme gelöst sind / keine Seeds mehr vorhanden.