Threading in QB64pe (again)

[justsomeguy]

I have updated the "Worker demo" based upon some of DSMan195276's suggestions.

Code: (Select All)

// workers.h
// Threading Header
#include "pthread.h"
// Only needed for the SIGTERM Constant
#include <signal.h>
// needed the sleep()
#include <unistd.h>
// needed for clock()
#include <time.h>

#define RETRYCOUNT 5
// Initialize Threads
static pthread_t thread[9];

// thread arguments
typedef struct thread_data {
  int id;
} thread_data;

// Easy was to determine if a thread is running
static bool threadRunning[] = {false,false,false,false,false,false,false,false,false,false};

// Setup Mutexes.
static pthread_mutex_t mutex[9];

// QB's names for the threaded Subs
// You can locate these in your ''qb64pe/internal/temp'' folder.
// I found these in the 'main.txt'
void SUB_WORKERTHREAD(int32*_SUB_WORKERTHREAD_LONG_ID);

// wrap the subs so that you can easily get the void* for pthread
void* RunWorker(void *arg){
    thread_data *tdata=(thread_data *)arg;
    int id = tdata->id;
    threadRunning[id] = true;
    SUB_WORKERTHREAD((int32*)&id);
}

// These are the commands that are accessed by you program
int invokeWorker(int id){
    // setup arguments to be sent to the thread
    thread_data tdata;
    tdata.id = id;
    // Threads are not always successfully started
    // So, we retry until it sticks or we exceed the retry count.
    int retry_count = 0;
    // Is it already running?
    if (!threadRunning[id]) {
        // try to start thread
        while(pthread_create( &thread[id], NULL, RunWorker, (void *)&tdata))
        {
            // thread is not running
            if (retry_count++ > RETRYCOUNT){
                // jumpout if tried more than RETRYCOUNT
                return threadRunning[id];
            }
            // wait a sec before trying again
            sleep(1);
        };
        // Give thread a sec to spool up
        int timeout_time = clock() * 1000 / CLOCKS_PER_SEC + 1000;
        do {} while (clock() * 1000 / CLOCKS_PER_SEC <= timeout_time && threadRunning[id] != true);
    }
    return threadRunning[id];
}

void joinThread(int id){
    pthread_join(thread[id],NULL);
    threadRunning[id] = false;
}

void exitThread(){
    pthread_exit(NULL);
}

void killThread(int id){
    if (threadRunning[id]) {
        int iret = pthread_kill(thread[id], SIGTERM);
    }
}

void lockThread(int id){
    pthread_mutex_lock(&mutex[id]);
}

void unlockThread(int id){
    pthread_mutex_unlock(&mutex[id]);
}

void initLock(int id)
{
    pthread_mutex_init(&mutex[id], NULL);
}

Code: (Select All)

'****************************************************************************

' Worker Test

' by justsomeguy

'****************************************************************************



' Header interface

DECLARE LIBRARY "./workers"

  FUNCTION invokeWorker (BYVAL id AS LONG) ' start worker thread

  SUB joinThread (BYVAL id AS LONG) ' wait til thread is finished

  SUB exitThread ' must be called as thread exits

  SUB killThread (BYVAL id AS LONG) ' kill the thread

  SUB initLock (BYVAL id AS LONG) ' initialize mutex

  SUB lockThread (BYVAL id AS LONG) ' mutex lock

  SUB unlockThread (BYVAL id AS LONG) ' mutex unlock

END DECLARE



' UDT type definition

' this is going to be shared with the workers.

TYPE tWorker

  id AS _BYTE

  xstart AS LONG

  ystart AS LONG

  xsize AS LONG

  ysize AS LONG

  xdir AS _BYTE

  x AS LONG

  y AS LONG

  colour AS LONG

  command AS LONG

  sc AS LONG

  img AS _MEM

  offset AS _OFFSET

  dly AS _BYTE

  dlyCount AS INTEGER

END TYPE



' Local Variables

DIM AS LONG indx

DIM AS STRING ky

DIM AS INTEGER tc



' Setup Screen

_TITLE "Worker Thread Test"

SCREEN _NEWIMAGE(800, 600, 32)



' Poll user on how many worker threads

DO

  CLS

  PRINT "How many worker threads? Low spec machines should start less than 4."

  INPUT "(1 - 8, 0 to quit):"; tc

  IF tc = 0 THEN SYSTEM

LOOP UNTIL tc >= 1 AND tc <= 8

tc = tc - 1

CLS



' Initialize our worker array

DIM SHARED AS tWorker worker(tc)



' Setup initial values for each of the workers

FOR indx = 0 TO tc

  initWorker indx

  ' These are the actual screens that the workers output to.

  worker(indx).sc = _NEWIMAGE(worker(indx).xsize + 1, worker(indx).ysize + 1, 32)

  worker(indx).img = _MEMIMAGE(worker(indx).sc)

NEXT



LOCATE 1

' Start workers

FOR indx = 0 TO tc

  PRINT "Starting thread:"; indx

  IF invokeWorker(indx) = 0 THEN PRINT "Failed to start worker thread!": END

NEXT

CLS



'main loop

DO

  ky = INKEY$

  ' toggle the built in delay for the worker

  IF ky = "d" THEN

    FOR indx = 0 TO tc

      worker(indx).dly = NOT worker(indx).dly

    NEXT

  END IF

  ' start/stop the workers

  IF ky = "s" THEN

    CLS

    FOR indx = 0 TO tc

      ' toggle worker command

      ' this could be more elaborate adding more functionality

      worker(indx).command = NOT worker(indx).command



      'if worker command is -1 then worker is going to quit

      IF worker(indx).command THEN

        ' if worker quits then wait until all the threads quit

        joinThread indx

      ELSE

        ' restart threads when command is 0

        PRINT "Restarting thread:"; indx

        IF invokeWorker(indx) = 0 THEN PRINT "Failed to start worker thread!": END

        _DISPLAY

      END IF

    NEXT

    CLS

  END IF

  ' stop the workers and quit

  IF ky = CHR$(27) THEN

    FOR indx = 0 TO tc

      worker(indx).command = -1

      joinThread indx

      SYSTEM

    NEXT

  END IF



  'Draw the results of the work

  FOR indx = 0 TO tc

    ' display the id that was passed to worker

    _PRINTSTRING (indx * 100 + 10, 220), "Worker:" + STR$(worker(indx).id)

    ' lock image to keep worker from writing to it while

    ' QB64 is drawing it to the screen.

    ' suggested by DSMan195276

    lockThread indx

    _PUTIMAGE (indx * 100, 250), worker(indx).sc

    unlockThread indx

  NEXT

  ' on screen instructions

  _PRINTSTRING (10, 380), "Press 'd' toggle thread slowdown. Press 's' to start/stop threads. Press 'ESC' to exit."

  _LIMIT 60

  _DISPLAY

LOOP



' Initialize worker

SUB initWorker (id AS LONG)

  worker(id).xsize = 100

  worker(id).ysize = 100

  worker(id).xstart = 0

  worker(id).ystart = 0

  worker(id).x = worker(id).xstart

  worker(id).y = worker(id).ystart

  worker(id).xsize = 100

  worker(id).ysize = 100

  worker(id).xdir = 1

  worker(id).colour = _RGB32(RND * 255, RND * 255, RND * 255)

END SUB



' Threaded Worker

SUB workerThread (id AS LONG)

  ' turn checking off because it would useless in the context of a thread

  ' suggested by DSMan195276

  $CHECKING:OFF

  initLock id

  worker(id).id = id ' this is silly, but it is to test if 'id' get passed successfully.

  DO

    ' This worker fills in a square by snaking side to side

    ' when it reaches bottom it changes color and returns to the top.

    IF worker(id).x + worker(id).xdir > worker(id).xstart + worker(id).xsize OR worker(id).x + worker(id).xdir < worker(id).xstart THEN

      worker(id).xdir = -worker(id).xdir

      IF worker(id).y + 1 > worker(id).ysize THEN

        worker(id).x = worker(id).xstart

        worker(id).y = worker(id).ystart

        worker(id).xdir = 1

        worker(id).colour = _RGB32(RND * 255, RND * 255, RND * 255)

      ELSE

        worker(id).y = worker(id).y + 1

      END IF

    ELSE

      worker(id).x = worker(id).x + worker(id).xdir

    END IF

    ' draw pixels directly to the image

    worker(id).offset = worker(id).img.OFFSET + (worker(id).x * 4) + (worker(id).y * worker(id).xsize * 4)

    ' lock image to keep QB64 from trying to draw it while the worker is writing to it

    ' suggested by DSMan195276

    lockThread id

    _MEMPUT worker(id).img, worker(id).offset + 0, _BLUE32(worker(id).colour) AS _UNSIGNED _BYTE

    _MEMPUT worker(id).img, worker(id).offset + 1, _GREEN32(worker(id).colour) AS _UNSIGNED _BYTE

    _MEMPUT worker(id).img, worker(id).offset + 2, _RED32(worker(id).colour) AS _UNSIGNED _BYTE

    _MEMPUT worker(id).img, worker(id).offset + 3, 255 AS _UNSIGNED _BYTE

    ' unlock after done

    unlockThread id

    ' Put a small delay so that you can see the worker snake

    IF NOT worker(id).dly THEN worker(id).dlyCount = 0: DO: worker(id).dlyCount = worker(id).dlyCount + 1: LOOP UNTIL worker(id).dlyCount < 0

  LOOP WHILE worker(id).command = 0

  exitThread

  $CHECKING:ON

END SUB

I would recommend reading up on stack vs. heap allocation in C and C++. That said the gist is that `thread_data tdata` declares the `thread_data` structure on the stack, that means it only exists while the main thread is in the call to `invokeWorker`. Once `invokeWorker` returns, the thread_data variable goes away and its memory will be reused for something else.

The issue that comes up is that a pointer to the `tdata` variable is passed to `pthread_create()` (as the parameter to the new thread). That starts a separate thread and the original thread can can get the success result from `pthread_create` and then return from `invokeWorker` all before the new thread actually starts running any code. Then you're in the situation where the new thread will be reading from the `tdata` variable while it is already gone from the original thread (and its memory is likely being reused). It's a situation where the issue is probably avoided in this simplified example, but will start to blow up once your code gets more complex and you're calling more functions.

It looks like worked around this problem by adding a 1 second sleep timer after starting the thread. This gave time for the thread to spool up.

In this latest version, it checks if the thread is started before 'invokeWorker' returns. 'invokeWorker' is now a function that returns a success or fail based on if the thread started. Now the threads start a lot faster.

I double check if the arguments going to the worker by recording the 'id' to its UDT. I then print that result above each worker. So far they have all been aligned. 

I have been reading and I'm seeing a few examples of using malloc to pass arguments to the thread. I'm pondering the easiest and simplest way to implement it.

Maybe we're talking about the same thing, but I'm talking about the issue that your code writes four bytes (via four _MEMPUT commands) to the image. The main thread may read the image data while you have only written two of the four bytes, displaying an incorrect color as the result. A mutex is used to avoid this kind of issue, you give each image a mutex and then lock/unlock the mutex every time the image is used (so, around the `_MEMPUT`s and the `_PUTIMAGE`). That way the main thread will only read from an image when the corresponding worker thread is not writing to it.

I did as you suggested. I hope I implemented it right. I did notice a bit of a slowdown. I suspect its a race condition that I won't have much control over.

Thank you for your suggestions and keep them coming.