Threading in QB64pe (again)

[justsomeguy]

@DSMan195276 First, Thank you for your help! Without your insight this would have been another failed attempt.

Since I'm starting and stopping threads, it seems to me that if I re-initialize the mutex it might cause problems.  So, my solution is to destroy the mutex after they join. Kinda following the example here https://www.thegeekstuff.com/2012/05/c-mutex-examples/

Mutex initialize:

Code: (Select All)

// 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
        pthread_mutex_init(&mutex[id], NULL);
        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];
}

Mutex destroy:

Code: (Select All)

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

So, I hope this is the last iteration of this example.

Full Header "workers.h"

Code: (Select All)

// workers.h
// Threading Header
#include "pthread.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
        pthread_mutex_init(&mutex[id], NULL);
        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;
    pthread_mutex_destroy(&mutex[id]);
}

void exitThread(){
    pthread_exit(NULL);
}

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

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

 
And full QB64

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 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

  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 think I'm going to work up a new example, perhaps Mandelbrot, we will see.