Threads in Windows

[HelpfulPerson]

I've created threads before using begin_thread_ex() and end_thread_ex(), but I've never quite got how to pass variables. I also never learned how to make my code "thread safe". All I know that is there is something called a mutex which is used for data storage( might be wrong on that ), and you can "lock" parts to make sure your data doesn't go bad. Can someone explain this subject more to me or point me to some links on more detailed explanation than the MSDN explanation.

Edit : I found an example on MSDN that compiles fine. I understand what they're doing because of the comments, but I don't get the reason for the thread-safe part or why they have to use specialized functions for everything.

#include <windows.h>
#include <tchar.h>
#include <strsafe.h>


#define MAX_THREADS 3
#define BUF_SIZE 255


DWORD WINAPI MyThreadFunction( LPVOID lpParam );
void ErrorHandler(LPTSTR lpszFunction);


// Sample custom data structure for threads to use.
// This is passed by void pointer so it can be any data type
// that can be passed using a single void pointer (LPVOID).
typedef struct MyData {
    int val1;
    int val2;
} MYDATA, *PMYDATA;




int _tmain()
{
    PMYDATA pDataArray[MAX_THREADS];
    DWORD   dwThreadIdArray[MAX_THREADS];
    HANDLE  hThreadArray[MAX_THREADS];


    int i = 0;


    // Create MAX_THREADS worker threads.


    for( ; i<MAX_THREADS; i++ )
    {
        // Allocate memory for thread data.


        pDataArray[i] = (PMYDATA) HeapAlloc(GetProcessHeap(), HEAP_ZERO_MEMORY,
                sizeof(MYDATA));


        if( pDataArray[i] == NULL )
        {
           // If the array allocation fails, the system is out of memory
           // so there is no point in trying to print an error message.
           // Just terminate execution.
            ExitProcess(2);
        }


        // Generate unique data for each thread to work with.


        pDataArray[i]->val1 = i;
        pDataArray[i]->val2 = i+100;


        // Create the thread to begin execution on its own.


        hThreadArray[i] = CreateThread(
            NULL,                   // default security attributes
            0,                      // use default stack size
            MyThreadFunction,       // thread function name
            pDataArray[i],          // argument to thread function
            0,                      // use default creation flags
            &dwThreadIdArray[i]);   // returns the thread identifier




        // Check the return value for success.
        // If CreateThread fails, terminate execution.
        // This will automatically clean up threads and memory.


        if (hThreadArray[i] == NULL)
        {
           ErrorHandler(TEXT("CreateThread"));
           ExitProcess(3);
        }
    } // End of main thread creation loop.


    // Wait until all threads have terminated.


    WaitForMultipleObjects(MAX_THREADS, hThreadArray, TRUE, INFINITE);


    // Close all thread handles and free memory allocations.


    for(i=0; i<MAX_THREADS; i++)
    {
        CloseHandle(hThreadArray[i]);
        if(pDataArray[i] != NULL)
        {
            HeapFree(GetProcessHeap(), 0, pDataArray[i]);
            pDataArray[i] = NULL;    // Ensure address is not reused.
        }
    }


    return 0;
}




DWORD WINAPI MyThreadFunction( LPVOID lpParam )
{
    HANDLE hStdout;
    PMYDATA pDataArray;


    TCHAR msgBuf[BUF_SIZE];
    size_t cchStringSize = 0;
    DWORD dwChars;


    // Make sure there is a console to receive output results.


    hStdout = GetStdHandle(STD_OUTPUT_HANDLE);
    if( hStdout == INVALID_HANDLE_VALUE )
        return 1;


    // Cast the parameter to the correct data type.
    // The pointer is known to be valid because
    // it was checked for NULL before the thread was created.


    pDataArray = (PMYDATA)lpParam;


    // Print the parameter values using thread-safe functions.


    StringCchPrintf(msgBuf, BUF_SIZE, TEXT("Parameters = %d, %d\n"),
        pDataArray->val1, pDataArray->val2);
    StringCchLength(msgBuf, BUF_SIZE, &cchStringSize);
    WriteConsole(hStdout, msgBuf, (DWORD)cchStringSize, &dwChars, NULL);


    return 0;
}






void ErrorHandler(LPTSTR lpszFunction)
{
    // Retrieve the system error message for the last-error code.


    LPVOID lpMsgBuf;
    LPVOID lpDisplayBuf;
    DWORD dw = GetLastError();


    FormatMessage(
        FORMAT_MESSAGE_ALLOCATE_BUFFER |
        FORMAT_MESSAGE_FROM_SYSTEM |
        FORMAT_MESSAGE_IGNORE_INSERTS,
        NULL,
        dw,
        MAKELANGID(LANG_NEUTRAL, SUBLANG_DEFAULT),
        (LPTSTR) &lpMsgBuf,
        0, NULL );


    // Display the error message.


    lpDisplayBuf = (LPVOID)LocalAlloc(LMEM_ZEROINIT,
        (lstrlen((LPCTSTR) lpMsgBuf) + lstrlen((LPCTSTR) lpszFunction) + 40) * sizeof(TCHAR));
    StringCchPrintf((LPTSTR)lpDisplayBuf,
        LocalSize(lpDisplayBuf) / sizeof(TCHAR),
        TEXT("%s failed with error %d: %s"),
        lpszFunction, dw, lpMsgBuf);
    MessageBox(NULL, (LPCTSTR) lpDisplayBuf, TEXT("Error"), MB_OK);


    // Free error-handling buffer allocations.


    LocalFree(lpMsgBuf);
    LocalFree(lpDisplayBuf);
}

[ledow]

How to pass variables? Look at the example, it shows you how to do it. Almost all threading libraries provide a way to send an arbitrary value through to threads on creation and/or there are methods to use "shared memory" where many threads can see and use the same piece of memory.

A mutex? Mutex stands for "mutually exclusive". It doesn't store any data. It's basically a way of ensuring that ONE and ONLY ONE thread can access a variable (or piece of data, or file handle, or whatever) at a time.

With threads, things can happen at the same time. While you're reading the first half of a number from memory, another thread could be changing the number to something else and you end up with a mess where the number you read isn't what it was when you started, and isn't what the number was changing to but some half-and-half mess of them both (this is a simplified example).

If you grab the "lock", the mutex, inside a thread, it's a way of ensuring that ONLY you have access to that piece of memory. You're basically saying to the rest of the system "It's my turn now, don't touch". Any other thread that tries to grab the lock while you have it will just get an error until you say you are done with it. So you KNOW that you're safe to modify things and that nobody else is going to change things underneath you. When you are done, you "release" the lock, and then other threads are free to use both the lock AND whatever it was protecting.

Now, the link between what lock you use, and what it's protecting is largely determined by the programmer. It's not an automatic thing and libraries, etc. tend to provide their own locking because only they know their code well enough to effectively lock it.

Let's take a customer database with names and addresses. You might have one thread (thread A) that is running through the database, cleaning up entries. While that thread wants to change the customer's address to their new address, another thread (thread B) might be reading the database wanting to print out a label to send a product there. 99.9% of the time, these things won't happen at exactly the same time so you can get away with a lot of bad coding. But let's say they happen at almost exactly the same time.

Thread B is first in by a fraction of a second. It grabs the lock / mutex.
Thread A is just pipped to the post. It tries to grab the lock but can't because thread B already has it.
Thread B can now do whatever it needs to - the information it has shouldn't change. It prints out a label (the fact that it's a label of the old address doesn't matter - at least it got full, complete data that wasn't invalid, just slightly out-of-date).
Thread A sits around waiting while the label is printed, constantly trying to grab the lock but it's still always "in use".
Thread B finally finishes with the data, so it "releases" the lock.
Now Thread A doesn't get an error when it tries to grab the lock, but instead gets it successfully. It's Thread A's "turn" to do something with it. It changes the address, and then releases the lock.

Without the lock, it would have been quite possible to print, say, the first line of the old address and the second line of the new address if Thread A had changed the data JUST AS Thread B was printing it. That would form an impossible combination (e.g. Paris, USA) and the customer's products would get sent out to a non-existent address.

Now think of my computer which is currently running 867 threads (and I'm not even doing anything but loading a web browser!). Each of those threads might want to do something that another thread is doing. So they are constantly "passing the parcel" of locks around to determine who has access to certain bits of data. Eventually you'd hit a point where one of those threads interfered with another if you didn't do locking properly. There are locks for hardware (sound, graphics, etc.) there are locks for files, locks for unique names, locks for parts of memory, even locks for critical Windows information. Without the locks, there would be a massive free-for-all and any of those threads could change something under the nose of the others and NO-ONE WOULD KNOW! That's inevitably going to cause crashes and data loss and all sorts before long.

This is what we mean by "thread-safe" - not accessing ANY resources that another thread might be accessing at the same time, without proper locking and control to ensure you don't end up corrupting or "half-modifying" data. One way of doing this is with locks (basically a little binary flag in the system that says "Yes, this is in use" or "No, this is free") - there are many types of locks and many ways to deal with them (would you want your computer to let someone "lock" a critical part of the system forever? Or would you want to take that lock away from them? Would you want your antivirus to constantly be denied access to a file because another process had locked it, etc.?). A mutex is one particular type of lock.

Notice, that though most OS provide proper threading and (as a part of that) locking tools you can use, it's also possible to "roll your own" in your own code. The problem is that there's SO MUCH to take account of that it's usually best to try to use the OS-provided ones. For example, SDL provides it's own locking for any drawing surface it uses. If you want to modify the image, you have to "lock" the surface first. This is so that one thread cannot "draw" onto the screen while another thread is also drawing on it. They all have to wait their turn. Done improperly, it can be a major bottleneck, or a major security problem. Done properly, it lets just about everything happen simultaneously and - on the very rare occasion that two things want to use the same thing at the same time - ensures they take proper turns so as to not "break" the system.

Threading is really one of those things that is incredibly simple in principle but that making sure your code is thread-safe is one of the most difficult tasks you can take. There are still locking problems (deadlocks, security problems etc.) in the Linux kernel, for instance, and more are found all the time. But they are one of hardest class of problems to debug, reproduce and effectively fix unless you're an expert.

[HelpfulPerson]

That makes a lot more sense now. I didn't see in sections about locking on MSDN, where would I look up references for that? All I was able to read about was how you can use fibers in threads to call regular functions and how threads can use preemptive multitasking to create an illusion of simultaneous execution. Also, can you use regular functions in threads at all, or do you have to use fibers or use a special technique to make sure nothing bad happens? My last question would be how can I tell a thread that it no longer needs to execute( e.g. no more data is available to work with )?

[rcgldr]

Link to a zip of an example multithreaded C program that copies a file, reading into buffers in one thread (the current thread), writing from the buffers in another thread (created). It uses linked list messages to communicate and synchronize between threads, using mutexes for ownership of a list and semaphores for the count of message nodes on a list. The setup takes a bit of code, but the messaging functions and the read / write functions are simple.

http://rcgldr.net/misc/mtcopy.zip

[ledow]

I didn't see in sections about locking on MSDN, where would I look up references for that?

Find the Windows locking functions, read their specification. Google windows threading and look through code. This isn't an MS-specific technique, just the way that you have to implement it is and MSDN describes the interface quite simply. The concept of locking in itself is much larger and more complicated than that, though, which is why you won't find much of a "locking for dummies" post on MSDN.

Also, can you use regular functions in threads at all, or do you have to use fibers or use a special technique to make sure nothing bad happens?

It depends very much on the functions. Some you can, some you can't, but there's normally a warning in the function documentation if you read up on that particular function. Hell, even some standard-C functions aren't "thread-safe" just because of their usage (strtok springs to mind), unless you're very careful when you're handling them. It's as much about knowing the functions called and imagining what interactions might occur between threads and reading up on the functions themselves. Bear in mind that just because a function is thread-safe it doesn't mean it's automatic and you don't have to do anything to make it work.

My last question would be how can I tell a thread that it no longer needs to execute( e.g. no more data is available to work with )?

Machine-specific functions, or just exit the thread in the prescribed manner. Depending on the way threading is implemented, you can either just return from the function that was called in a threaded manner (and the return destroys the thread), or you have to call a specific function to terminate the thread.

Usually, threading is a matter of making a function that will run in the thread and then calling a "create thread" of some kind and telling it what function it needs to execute. That will create a thread and set it working on that function. When that function returns (i.e. it's finished), the thread is destroyed. But equally, you can forcibly terminate threads (e.g. TerminateThread on Windows), or you might have a prescribed function to tell the threading library that you're finished with a particular thread. That's not recommended, but it's also possible. More likely, you will signal your thread somehow that it's time to shut up shop, and it will clear up after itself and do what it needs to to terminate. How does it know if you're finished with it? The same way you can pass a value to the thread, or use some shared memory you can both see, or otherwise send signals to a thread.


P.S. If you're not able to find this stuff out for yourself by just Googling, I suggest you buy a book of some kind: Concurrent Programming on Windows is one I've seen recommended. The magic words you want to learn are things like threads, re-entrant, concurrent, etc.

Processes and Threads (Windows) would be my MSDN starting point, though.

[HelpfulPerson]

Thanks ledow, I had a book that taught me about threads( I didn't read the chapter, I probably should ), but it was only for Mac and Linux. Most of the functions they used seemed easy, but it may have just been the way they explained it. Probably if I had some kind of book on the Windows API that explained it like they did, it would seem easy too. Also, I already read the link you posted, I didn't do it until some time after I first made posts here.

[HelpfulPerson]

Here is my first program I made to use threads with shared data.

#include <stdio.h>
#include <stdlib.h>


#include <process.h>
#define WIN_32_LEAN_AND_MEAN
#include <windows.h>


#define MAX_THREADS 3


typedef struct thread_data
{
    DWORD value;
    HANDLE semaphore;
} thread_data;


unsigned __stdcall thread( void * parameter )
{
    thread_data * my_data = (thread_data *)parameter;


    DWORD access = 0;
    BOOL wait = TRUE;


    while ( wait )
    {
        access = WaitForSingleObject(my_data->semaphore, 0L);


        switch (access)
        {
            case WAIT_OBJECT_0 :
                printf("Thread id %d : Acessing value\n", (int)GetCurrentThreadId());
                wait = FALSE;


                my_data->value *= 2;


                printf("Value is now : %d\n", (int)my_data->value);


                if ( !ReleaseSemaphore(my_data->semaphore, 1, NULL))
                {
                    printf("ReleaseSemaphore error : %d\n", (int)GetLastError());
                    exit(1);
                }


                break;


            case WAIT_TIMEOUT :
                break;
        }
    }


    return 0;
}


int main( )
{
    void * data = NULL;


    unsigned thread_id[3];
    unsigned index;
    DWORD exit_code[3] = {0};
    HANDLE h_threads[3] ={0};


    thread_data * new_data = malloc(sizeof(thread_data));


    if (!new_data)
    {
        perror("Malloc");
        exit(1);
    }


    new_data->value = 400;
    new_data->semaphore = CreateSemaphore( NULL, 1, 1, NULL );


    data = new_data;


    if (new_data->semaphore == NULL)
    {
        printf("CreateSemaphore error : %d\n", (int)GetLastError());
        exit(1);
    }


    for (index = 0; index < MAX_THREADS; index++ )
        h_threads[index] = (HANDLE)_beginthreadex(NULL, 0, &thread, data, 0, &thread_id[index] );


    WaitForMultipleObjects(MAX_THREADS, h_threads, TRUE, INFINITE);


    for (index = 0; index < MAX_THREADS; index++ )
    {
        GetExitCodeThread(&h_threads[index], &exit_code[index]);


        CloseHandle(h_threads[index]);


    }


    for (index = 0; index < MAX_THREADS; index++ )
    {
        if ( exit_code[index] )
        {
            puts("An error has occured.");
            exit(1);
        }
    }


    CloseHandle(new_data->semaphore);


    printf("Value in main is now : %d\n", (int)new_data->value);


    return 0;
}

Here is the output :

Thread id 4084 : Acessing value
Value is now : 800
Thread id 4976 : Acessing value
Value is now : 1600
Thread id 1336 : Acessing value
Value is now : 3200
Value in main is now : 3200

Thanks Ledow, you helped a lot. Reading the documentation on MSDN helped me too. It will be a while before I memorize how to make threads and semaphores, but so far so good.

[Codeplug]

>> ...[3]
Use "MAX_THREADS" instead of "3".

Missing matching free().

Check for errors when calling _beginthreadex() and WaitForXXXObjects().

>> WaitForSingleObject(..., 0L);
That is a hard-spinning loop that does nothing. Never do that as it wastes CPU resources. If there is something that needs to be done periodically, use a non-zero timeout value. Otherwise use INFINITE to just wait for objects to be signaled.

If the synchronization is within a single process, and you don't need to use WaitForXXX semantics, then a CRITICAL_SECTION can also be used:
Critical Section Objects (Windows)

gg

[HelpfulPerson]

>> ...[3]
Use "MAX_THREADS" instead of "3".

Missing matching free().

Check for errors when calling _beginthreadex() and WaitForXXXObjects().

>> WaitForSingleObject(..., 0L);
That is a hard-spinning loop that does nothing. Never do that as it wastes CPU resources. If there is something that needs to be done periodically, use a non-zero timeout value. Otherwise use INFINITE to just wait for objects to be signaled.

If the synchronization is within a single process, and you don't need to use WaitForXXX semantics, then a CRITICAL_SECTION can also be used:
Critical Section Objects (Windows)

gg

With your suggestions, I changed my code to this

#include <stdio.h>
#include <stdlib.h>

#include <process.h>
#define WIN_32_LEAN_AND_MEAN
#include <windows.h>

#define MAX_THREADS 3

typedef struct thread_data
{
    DWORD value;
    HANDLE semaphore;
} thread_data;

unsigned __stdcall thread( void * parameter )
{
    thread_data * my_data = (thread_data *)parameter;
    WaitForSingleObject(my_data->semaphore, INFINITE);

    printf("Thread id %d : Acessing value\n", (int)GetCurrentThreadId());
    my_data->value *= 2;
    printf("Value is now : %d\n", (int)my_data->value);

    if ( !ReleaseSemaphore(my_data->semaphore, 1, NULL))
    {
        printf("ReleaseSemaphore error : %d\n", (int)GetLastError());
        exit(1);
    }

    return 0;
}

int main( )
{
    void * data = NULL;

    unsigned thread_id[MAX_THREADS];
    unsigned index;
    DWORD exit_code[MAX_THREADS] = {0};
    HANDLE h_threads[MAX_THREADS] ={0};

    thread_data * new_data = malloc(sizeof(thread_data));

    if (!new_data)
    {
        perror("Malloc");
        exit(1);
    }

    new_data->value = 400;
    new_data->semaphore = CreateSemaphore( NULL, 1, 1, NULL );

    data = new_data;

    if (new_data->semaphore == NULL)
    {
        printf("CreateSemaphore error : %d\n", (int)GetLastError());
        exit(1);
    }

    for (index = 0; index < MAX_THREADS; index++ )
        h_threads[index] = (HANDLE)_beginthreadex(NULL, 0, &thread, data, 0, &thread_id[index] );

    WaitForMultipleObjects(MAX_THREADS, h_threads, TRUE, INFINITE);

    for (index = 0; index < MAX_THREADS; index++ )
    {
        GetExitCodeThread(&h_threads[index], &exit_code[index]);
        CloseHandle(h_threads[index]);
    }

    for (index = 0; index < MAX_THREADS; index++ )
    {
        if ( exit_code[index] )
        {
            puts("An error has occured.");
            exit(1);
        }
    }

    CloseHandle(new_data->semaphore);

    printf("Value in main is now : %d\n", (int)new_data->value);
    free(new_data);

    return 0;
}

The output is of course the same, but thanks to your suggestions, there was a major performance increase. I will study critical sections more tomorrow, and I might re-post some code when I add it.

Edit : Forgot to to check beginthreadex's return value.

[HelpfulPerson]

Here's the new code

#include <stdio.h>
#include <stdlib.h>


#include <process.h>
#define WIN_32_LEAN_AND_MEAN
#include <windows.h>
#include <strsafe.h>


#define MAX_THREADS 10


typedef struct thread_data
{
    DWORD value;
} thread_data;


CRITICAL_SECTION critical_section;


/* Credit to Microsoft : Retrieving the Last-Error Code (Windows) */


void ErrorExit(LPTSTR lpszFunction)
{
    LPVOID lp_msg_buf;
    LPVOID lp_display_buf;
    DWORD dw = GetLastError();


    FormatMessage
    (
    FORMAT_MESSAGE_ALLOCATE_BUFFER | FORMAT_MESSAGE_FROM_SYSTEM | FORMAT_MESSAGE_IGNORE_INSERTS,
    NULL,
    dw,
    MAKELANGID(LANG_NEUTRAL, SUBLANG_DEFAULT),
    (LPTSTR) &lp_msg_buf,
    0,
    NULL
    );




    lp_display_buf = (LPVOID)LocalAlloc(LMEM_ZEROINIT, (lstrlen((LPCTSTR)lp_msg_buf) + lstrlen((LPCTSTR)lpszFunction) + 40) * sizeof(TCHAR));


    StringCchPrintf((LPTSTR)lp_display_buf, LocalSize(lp_display_buf) / sizeof(TCHAR), TEXT("%s failed with error %d: %s"), lpszFunction, dw, lp_msg_buf);
    MessageBox(NULL, (LPCTSTR)lp_display_buf, TEXT("Error"), MB_OK);


    LocalFree(lp_msg_buf);
    LocalFree(lp_display_buf);
    ExitProcess(dw);
}


unsigned __stdcall thread( void * parameter )
{
    thread_data * my_data = (thread_data *)parameter;


    EnterCriticalSection(&critical_section);


    printf("Thread id %d : Acessing value\n", (int)GetCurrentThreadId());
    my_data->value *= 2;
    printf("Value is now : %d\n", (int)my_data->value);


    LeaveCriticalSection(&critical_section);


    return 0;
}


int main( )
{
    void * data = NULL;


    unsigned thread_id[MAX_THREADS] = {0};
    unsigned index;
    DWORD exit_code[MAX_THREADS] = {0};
    HANDLE h_threads[MAX_THREADS] ={0};


    thread_data * new_data = malloc(sizeof(thread_data));


    if (!InitializeCriticalSectionAndSpinCount(&critical_section, 0x0000050))
        ErrorExit(TEXT("InitialCriticalSectionAndSpinCount"));


    if (!new_data)
    {
        perror("Malloc");
        exit(1);
    }


    new_data->value = 2;
    data = new_data;


    for (index = 0; index < MAX_THREADS; index++ )
        if (!(h_threads[index] = (HANDLE)_beginthreadex(NULL, 0, &thread, data, 0, &thread_id[index])))
            ErrorExit("_beginthreadex");


    WaitForMultipleObjects(MAX_THREADS, h_threads, TRUE, INFINITE);


    for (index = 0; index < MAX_THREADS; index++ )
    {
        GetExitCodeThread(&h_threads[index], &exit_code[index]);
        CloseHandle(h_threads[index]);
    }


    for (index = 0; index < MAX_THREADS; index++ )
    {
        if ( exit_code[index] )
        {
            puts("An error has occured.");
            exit(1);
        }
    }


    printf("Value in main is now : %d\n", (int)new_data->value);


    DeleteCriticalSection(&critical_section);
    free(new_data);


    return 0;
}

[vart]

It seems logical to me to check whether the malloc failed immediately. You do not need to create the Critical section when malloc failed.