Thread pooling

[uncut glen]

I'm very new to multithreading and found some thread pooling examples online and made a class out of them: https://pastebin.com/pWddf9m0

I can get it to do some jobs with a simple example:

int j = 0;
for (int i = 0; i < 4; i++) {
    this->pool.Queue([&]() {
        j++;
        std::cout << "\n " << j;
        });
}

Which sort of works, it out puts:

1
3
4
4

But for larger jobs, like setting color data from a file, it crashes after jobs and I get a read access violation from the vector its filling.

So, with all that, what am I doing wrong or is the POOL object even correct? And how about the accuracy issue?

I'll gladly provide more info as needed. Hopefully I explained my problem right.

[sturdy panther]

When your question is answered use !solved to mark the question as resolved.

Remember to ask specific questions, provide necessary details, and reduce your question to its simplest form. For tips on how to ask a good question use !howto ask.

[uncut glen]

So I managed to get rid of some of the errors, atleast the heap errors, by setting the filled vector to a new object. So what I guess I have a problem with now, is the accuracy, out of 64x64 lines, around 2k only read each debug.

[oak geode]

[&]() what does that do?

[uncut glen]

I'm not too sure. It's called a lambda function.

[oak geode]

be sure.

[uncut glen]

From what I gather, [] is the capture clause? & captures used variables so they store any modification to the variable itself? And () would have the arguments, if there were any. I could be wrong, very wrong. Very new to using lambdas aswell.

[slim cobalt]

It automatically binds all variables used in the lamda by reference (similar to [=], which is by-value):

& (implicitly capture the used variables with automatic storage duration by reference)
- cppref

[oak geode]

so you're referencing local vars, an acccessing them from a different thread?

[uncut glen]

Pretty much

[slim cobalt]

Oh, if that's what's going on, you have to move the needed variables in as freestore-variables, also you probably need [=] to pass by value, since you usually can't access the stack from other threads

[uncut glen]

"freestore-variables"?

[slim cobalt]

variables on the heap, e.g. new int

Or in other other words: new pointers

[uncut glen]

Like so?

std::vector<int> *ints = new std::vector<int>;

for (int i = 0; i < 5; i++) {
    this->pool.Queue([&]() {
        int* p = new int(i);
        ints->push_back(*p);
        delete p;
        });
}

[slim cobalt]

You can use the direct int values inside the function since it copies the values into the std::vector's array on the freestore; also you definitely need to have a mutex as well, since otherwise it will corrupt the vector

[uncut glen]

Sorry, mutex? How would that be done?

[slim cobalt]

The same way as in your pastebin, which uses an std::mutex for the same purpose

[uncut glen]

std::mutex m;
std::vector<int> *ints = new std::vector<int>;

for (int i = 0; i < 5; i++) {
    this->pool.Queue([&]() {
        {
            std::unique_lock<std::mutex> lock(m);
            ints->push_back(i);
        }
        });
}

?

[slim cobalt]

That should work, since IIRC the mutex has it's stuff on the freestore too, so you just need to change the [&] to a [=]

[uncut glen]

If I use = instead of &

Also if I run with = and without the mutex, the vector doesn't seem corrupt, prints:

0
1
2
3
4

[slim cobalt]

For 5 iterations it probably doesn't execute in the right way, but if you let it run long enough, it likely will; this works (with new mutex, the problem is that [=] makes all context vars const):

std::mutex *m = new std::mutex;
std::vector<int> *ints = new std::vector<int>;
POOL pool;
for (int i = 0; i < 5; i++) {
    pool.Queue([=]() {
        {
            std::unique_lock<std::mutex> lock(*m);
            ints->push_back(i);
        }
        });
}

[uncut glen]

Will give it a try, just fixing up my file reader for multithreading. (which like put as before, did also read correct amount of lines (64 x 64 = 4096))

[uncut glen]

Well, I'm getting a "Breakpoint Instruction Executed"

A breakpoint instruction (__debugbreak() statement or a similar call) was executed in build.exe.

Callstack points to thread, vector, their destructors and what not

I had the file reader, join threads after its read the file, which seemed to work, but then the program stalls on looping through the threads in the pool:

for (int i = 0; i < this->thread.size(); i++) {
    if (this->thread[i].joinable()) this->thread[i].join();
}

[uncut glen]

Well I got it to draw the read color data, and, which astonishingly looks as it almost should (few pixels out of place), just some colors mixed up (seems like some yellow shade)

Scaled version, with 2 threads reading.

After some writing, it doesn't seem to read it everytime, sometimes I get some assembly error while reading, and sometimes empty pointers from the color object, plus sometimes access violation reading color object location.

Also how would I go about fixing the pixels which are out of place? I tried to feed a counter variable to the threads, but same outcome.

[uncut glen]

Got the colors sorted, was reading wrong characters and missing a switch case.

So now left is to figure out how I get them ordered in as they are in the file by line. 😦

[jovial sigil]

I'm a bit sad about this end result of code, but it's also hard to say anything definitive without a little bit more context about how OP sets things up

taking objects by reference in lambda expressions can definitely cause issues, but since the jobs' only fill in the vector, it's also quite likely that immediately after the loop OP is waiting for the pool to complete all the job that have been queued, so as long as you properly deal with all the data race/concurrency part, you're actually fine with capturing by reference

[jovial sigil]

ok well I've only just opened the code for the thread pool, I genuinely dislike that pool a lot

the loop member function almost certainly shouldn't be public, and there is no simple way to wait for all the queued jobs to complete, so if you need any for of synchronization it has to be baked into the jobs directly

[uncut glen]

Simplified code from my line reader, which works, just some pixels out of place:

https://pastebin.com/7pJBzQ0g

[jovial sigil]

    this->pool.Queue([=, &counter] {
        std::unique_lock<std::mutex> lock(*m);
        /*
            Parse line
        */
        counter++;
    });      

this makes the pool mostly useless tbh

individual jobs must acquire the lock to do anything

so the jobs related to "reading" can only occur one at a time

so it'd be simpler to just do them directly in the loop and without the thread pool

[uncut glen]

Yeah, but then again it takes a long time read and parse larger files with more lines, which is what I had in the first place, which got me thinking to multithread since it was so slow

[jovial sigil]

this

do { std::this_thread::sleep_for(std::chrono::milliseconds(1)); } while (this->pool.job.size() > 0);

also isn't "correct" because you kinda have a data race on pool.job, that's pretty much why the queue and loop function in the pool work with a mutex when they manipulate the queue of jobs

[jovial sigil]

did you actually measure your performances

[uncut glen]

Yeah, smallers files, like 20 x 20 colors, get loaded in just about 20 - 50 ms, where as a 64 x 64 colors took over a second

[jovial sigil]

threading IO is kinda not great because the file IO generally has to be sequential
dealing with the parsing can be threaded if you do it correctly, but here the central mutex just kills it

and you also have to deal with the ordering issue

it's just generally not simple to deal with this, but it's also hard to say anything without seeing the actual code

because here you've just locked out the entire job

[uncut glen]

#1234848324650139683 message

[jovial sigil]

if you really want to try to thread the parsing part, the individual lines you need to parse should be parsable without depending on the rest of the lines

and then you still have to deal with the ordering of stuff

[uncut glen]

Yeah they are, each line is separate and not dependant on others

[uncut glen]

So how does the mutex then come in?

[jovial sigil]

you use a mutex and mutex is short for mutual exclusion

mutexes are used when only one thread at a time should be executing the critical section protected by the mutex

by locking a mutex before the entire parsing job you effectively say that only one thread is allowed to parse at any single time

and it's just unclear if that's actually required or not without seeing what the code does

like, why did you add that mutex in the first place

do you need the mutex, or do you not

I assume somewhere in the parsing job you add the color to whatever clrs is

and maybe that part needs to be mutex protected

[uncut glen]

would u like to see the whole parsing stuff?

[jovial sigil]

but it's impossible to tell for us outsider without knowing what clrs is and how it's manipulated and what operations on it are or aren't thread safe

I mean ideally you understand how threading and data race work and we don't need to look at it, but we wouldn't be here if it was that simple :p

I mean sure you can show that, but since you seem to be using a bunch of custom types it's pretty likely you'll have to show a lot more than just that

e.g. what IMAP is

[uncut glen]

sure thing, gimme a sec

[jovial sigil]

because based on what has been said so far, it's very likely that you're manipulating clrs in that parsing code, and clrs has type IMAP

[uncut glen]

Well, theres the parsing, a whole switch. https://pastebin.com/tzxG3zFn

The IMAP https://pastebin.com/vK4x251W

(which is short for index map)

(or indexed map)

the STRING object: https://pastebin.com/gSbqr2AW

(be gentle? )

[jovial sigil]

no

[uncut glen]

So what youre saying is i'm screwd with my attempts?

[jovial sigil]

no I'm saying I'm not inclined at all to be "gentle", and I honestly would have been more inclined to try and be "gentle" if you hadn't said anything

[uncut glen]

Alright, give it to me

[jovial sigil]

that's how "bad" those custom structures are

[uncut glen]

Yeah I sort of figured as much

[jovial sigil]

where does "line" come from in that snippet

[jovial sigil]

I'm honestly not gonna spend that amount of time going through things right now

[uncut glen]

woops before the eof loop:

char line[256];

[jovial sigil]

do you seriously have only one line

that you share amongst threads

one line object/variable*

[uncut glen]

Yes, only one line variable, didn't even think of having plural

I mean I wasn't expecting much, I'm amazed someone has put that much thought into looking at the mess

[jovial sigil]

if you seriously want to multi-thread the parsing part, you cannot share the line that the threads have to work on

each job needs its own copy of the line

with exactly the line data that it has to parse

[uncut glen]

Gotcha, I'll get on it, but the mutex part, what operations are such which require locking?

[jovial sigil]

there's no shot at actually threading this mess if you don't fix that

also I assume the counter variable is there to kinda let the job know what line number it's processing, so that it then insert the line into the right place into you IMAP

you shouldn't be capturing that by reference

[uncut glen]

You sir, are correct

[jovial sigil]

the line number has to be attributed when the job receives its line

the job has absolutely no business messing with the counter

the loop that reads in the lines should just increment the counter when it deals with a new line/finishes reading a line

then you queue a job with an independent copy of the line, and the line number

the "position" function on imap is not thread safe, so that specific part has to be protected by a mutex

the parsing doesn't, assuming you're parsing an independent line object that isn't shared across threads

no clue what these parts do though

case 'w': {
    DINT i = 2;
    do {
        w.App(line[i]);
        i++;
    } while (line[i] != '-');
    wt = drx::fns::Cti(w.text);
} break;
case 'h': {
    DINT i = 2;
    do {
        h.App(line[i]);
        i++;
    } while (line[i] != '-');
    ht = drx::fns::Cti(h.text);
} break;

so just gonna ignore it

[uncut glen]

the color data has two last lines defining width and height

[jovial sigil]

right, I also don't know what this do

drx::util::RESULT<drx::gfx::COLOR, SINT> result = { color, counter };

dunno if its relevant

like it's not clear what color is in that context, and you don't seem to be using result

[uncut glen]

thats something i was gonna try, left it there (has no effect)

[jovial sigil]

ok but my point is that it's unclear if there's any thread safety to put there or not

because it's unclear to me what objects are going to be manipulated or not

[uncut glen]

Ok ok, got it.

[jovial sigil]

for the other case, it's "fairly" clear that you only need to acquire the mutex right before you do

clrs->Position({ r, g, b, e, bk }, counter);

which will also release the mutex immediately after the call because in your original code, that's the end of the switch case scope

[uncut glen]

But you said a central mutex is just gonna kill it, does that mean I should make an array of mutexes?

[jovial sigil]

I said a central mutex for the entirety of the parsing job would kill the threading for the parsing, not that you shouldn't use a central mutex

the mutex is there to protect the parts that cannot be done simultaneously in multiple threads

the parsing part in and of itself has no business being done under a mutex lock

the part where you add the resulting color of the parsing into IMAP fundamentally needs to be centralized, with that structure at least, and has to be protected by a mutex

though if I wasn't lazy I'd actually read through your setup to see if this IMAP thing is salvageable or not

[uncut glen]

Is ok

[jovial sigil]

but with something different then yeah you could in theory bypass some of the need for a mutex, though it's still plenty awkward because of how you're dealing with the height/width

like it's just genuinely weird that you've put that at the end and not at the start

of the file

[uncut glen]

yeah it is quite tedious, ive noticed, I been meaning to change that, just havent got to it

[jovial sigil]

but then again it looks like the file is all text which is also weird

if you'd put that information at the start it'd be a lot more trivial to parallelize this parsing code

[uncut glen]

Yeah it is, but seems somethings working, pixels are now in theyre correct places, with 3 threads reading.

[uncut glen]

Well I thought it was working with multiple mutexes, but there seems to be a central mutex somewhere (does the POOLs mutex have anything to do with it?) which is blocking other threads, the measured time doesn't change no matter how many threads I go with.

std::vector<std::mutex*> mtx;
int counter = 0;
for (int i = 0; i < 25; i++) {
    mtx.push_back(new std::mutex);
    this->pool.Queue([=, &counter]() {
            int id = counter;
            std::unique_lock<std::mutex> lock(*mtx.at(id));
            counter++;
            std::cout << "\n [" << counter << " - " << std::this_thread::get_id() << "]";
        });
}

[jovial sigil]

mandatory mention that just throwing more threads isn't going to magically make things faster, and that's not just related to mutex usage

IO is notoriously "slow", and yeah sure there's an extent to which you don't want to be manipulating std::cout from multiple different threads to avoid interleaving different invocations of <<
yet at the same time the << is very likely the one thing that's going to dominate your runtime in this reduced examples, incrementing a global counter is inconsequential in comparison

and to make these even worse, your mutexes are useless in that context

this strongly suggests you do not understand the purpose of them, and also this example also strongly suggests you didn't understand my suggestion up there #1235123691906666496 message

at the same time it's unclear what purpose that counters actually fulfills now that I reread more closely

anyway going back to the topic of the mutexes, they are completely useless here and just add overhead, there's nothing that they can reliably protect, and if they provide any mutual exclusion it's only by coincidence since this construct to obtain an "id" is subject to data race, so there's just no point to it

it's nearly (but not quite) as if you had written

int counter = 0;
for (int i = 0; i < 25; i++) {
    this->pool.Queue([=, &counter]() {
            std::mutex mtx;
            std::unique_lock<std::mutex> lock(mtx);
            counter++;
            std::cout << "\n [" << counter << " - " << std::this_thread::get_id() << "]";
        });
}

and I say that because I assume you intended each "id" to be unique, for each individual job

a mutex by nature has to be shared/centralized over every thread that wants to make use of the shared resource, you can't just blindly multiply mutexes and hope that it will work out, that's not how this work

mutexes are like doors, behind which you agree to put the shared resources

and you agree that only one person can go into the room with the shared resources

the mutex/door is there so you can lock the door and not be disturbed by other people wanting to get access to the room/shared resources

if you break down the walls of the room and install new doors for everyone, there's no point to having installed a door with a lock in the first place, anyone can just come in and out as they so please and it defeats the point

[jovial sigil]

semi off-topic, but this earlier remark was wrong btw
since your lambda was capturing by copy, the line array (assuming it is a real array and not just a pointer to one) was also captured by copy, so the lambdas should have had individual line arrays to work with

[uncut glen]

Yeah sorry, I'm just trying to learn what I can and understand

[uncut glen]

Got that part yeah, I made some changes to the file reader to use only one mutex, which locks right before making changes to the poolshared storage.

[uncut glen]

Figured so yeah, and also made changes accordingly