[Implemented] Thread safety and thread synchronisation

[Iria]

I know its been done before but PB threads really need sorting out. Thread safety and thread synchronisation are a priority for my Linux applications, and would be nice in Windows too. There are a number of windows based API hacks to help sort out some of the issues, but to be frank there not that helpful when working on a Linux platform.

I've found forum threads way back into the beginning of time on the subject and wondered if and when this issue is likely to be looked at?

As for suggested syntax, nothing fancy

Code: Select all

RequestMutex()
ReleaseMutex()

or maybe ...

Code: Select all

Lock()
Unlock()

Then a little more fancy

Code: Select all

RequestMutexTimout(1000) 

for example wait 1 second for a lock before moving on...

Naturally the above is more about preserving the 'valid' state of a PB procedure, threads also need a level of variable/data thread safetly, so I suggest (rather than making all variables thread safe!) you also add a thread safe Strucutre for example,

Code: Select all

MutexStructure ThreadSafeMemoryStructure
     AnyTypeOfLongWordDouble.l/w/d
     AnyTypeOfVariable.b[512]
    ..
EndStructure

Basically Im saying that a structure can contain a limited set of variable types word/long/double and byte the idea being Im happy to avoid the issue of no string buffers in PB if its easy to implment at least 'some' sort of thread safety on even limited variables types.

To use the above strucutre in a thread one would then do something like this

Code: Select all

Lock(a.ThreadSafeMemoryStructure)

or 

Mutex(a.ThreadSafeMemoryStructure)

or

RequestLock(a.ThreadSafeMemoryStructure)

RequestMutex(a.ThreadSafeMemoryStructure)

Then 

ReleaseLock(a.ThreadSafeMemoryStructure)

or

ReleaseMutex(a.ThreadSafeMemoryStructure)

Only a suggestion

BTW is there a place I can go to read up on whats planned for the next release of PB? or a PB road map

[Danilo]

Some code for Windows, to see how its done on Win.

The first code creates 2000 Threads, 1000 for writing and
1000 for reading.
All 2000 Threads access the same string:

Code: Select all

Structure MyThread
  mutex.l
  msg.l
  String$
EndStructure

#StopIt = 1

OpenConsole()

Procedure Thread1(*p.MyThread)
  Repeat
    WaitForSingleObject_(*p\mutex,#INFINITE) ; wait until object is available
      ; now access object
      *p\String$ = "$"+RSet(Hex(Random($7FFFFFFF)),8,"0")
      If *p\msg = #StopIt
        Quit = #TRUE
      EndIf
    ReleaseMutex_(*p\mutex)                  ; release object
  Until Quit
EndProcedure

Procedure Thread2(*p.MyThread)
  Repeat
    WaitForSingleObject_(*p\mutex,#INFINITE) ; wait until object is available
      ; now access object
      PrintN(*p\String$)
      If *p\msg = #StopIt
        Quit = #TRUE
      EndIf
    ReleaseMutex_(*p\mutex)                  ; release object
  Until Quit
EndProcedure


; create Thread with mutex object
x.MyThread
x\mutex = CreateMutex_(0,0,0)

For a = 1 To 1000
  CreateThread(@Thread1(),@x) ; write
  CreateThread(@Thread2(),@x) ; read
Next

Delay(30000)

For a = 1 To 100
  x\msg = #StopIt
Next a

Delay(2000)

Beep_(800,100)

CloseHandle_(x\mutex)

Note: you can use this system with any "object", means
structures, variables, strings, allocated memory, ...
You just assign 1 mutex to every object and use the 2 functions
WaitForSingleObject_() and ReleaseMutex_() around every part
where you access the object.

The second code contains some memory-string functions,
all made thread safe:

Code: Select all

;
; Thread safe string test, by Danilo
;
; first version without thread safety by NicTheQuick
;
Structure _string
  mutex.l    ; mutex
  len.l      ; length of string
  adr.l      ; address of string in memory
EndStructure

;#StrMutex = 0
;#StrLen   = 4
;#StrAdr   = 8
;#SIZEOF_STROBJ = 12

NewList String_Collector.l() ; string stack

; Procedure StrCpy(*Dest.BYTE,*Src.BYTE)
;   ; helper function
;   ; String copy
;   If *Dest And *Src
;     While *Src\b <> 0
;       *Dest\b = *Src\b
;       *Src  + 1
;       *Dest + 1
;     Wend
;     *Dest\b = 0 ; end
;   EndIf
; EndProcedure


Procedure StrNewAlloc(*String._STRING,len)
  ; helper function
  ; allocate new string in mem and delete old string-mem
  If *String
    mem = GlobalAlloc_(#GMEM_FIXED|#GMEM_ZEROINIT,len+1)
    If mem
      GlobalFree_(*String\adr)         ; free old String
      *String\adr = mem
      *String\len = len
    EndIf
    ProcedureReturn mem
  EndIf
EndProcedure

Procedure StrReAlloc(*String._STRING,len)
  ; helper function
  ; allocate new string in mem but DONT delete old string-mem!
  If *String
    mem = GlobalAlloc_(#GMEM_FIXED|#GMEM_ZEROINIT,len+1)
    If mem
      *String\adr = mem
      *String\len = len
    EndIf
    ProcedureReturn mem
  EndIf
EndProcedure


Procedure StrCreate2(chars.s)
  Mutex = CreateMutex_(0,0,0)
  Debug "Mutex created: "+StrU(Mutex,#LONG)
  If Mutex
    *NewStr._string = GlobalAlloc_(#GMEM_FIXED|#GMEM_ZEROINIT,SizeOf(_string))
    If *NewStr
      *NewStr\adr = GlobalAlloc_(#GMEM_FIXED|#GMEM_ZEROINIT,Len(chars))
      If *NewStr\adr
        If chars
          CopyMemory(@chars,*NewStr\adr,Len(chars))
        EndIf
        *NewStr\mutex = Mutex
        *NewStr\len   = Len(chars)
        AddElement(String_Collector())
        String_Collector() = *NewStr
        ProcedureReturn *NewStr
      Else
        CloseHandle_(Mutex)
        GlobalFree_(*NewStr)
        ProcedureReturn 0
      EndIf
    Else
      CloseHandle_(Mutex)
      ProcedureReturn 0
    EndIf
  Else
    ProcedureReturn 0
  EndIf
EndProcedure


Procedure StrCreate()
  ProcedureReturn StrCreate2("")
EndProcedure


Procedure StrAdr(*String._STRING)
  ; Get address of string in memory
  If *String
    WaitForSingleObject_(*String\mutex,#INFINITE)
    retval = *String\adr
    ReleaseMutex_(*String\mutex)
    ProcedureReturn retval
  EndIf
EndProcedure


Procedure StrLen(*Str._STRING)
  ; Get length of string in memory
  ;
  ; The length is stored in the string object,
  ; but could also be: len = ( GlobalSize_(*Str\adr) - 1 )
  If *Str

    WaitForSingleObject_(*Str\mutex,#INFINITE)
      retval = *Str\len
    ReleaseMutex_(*Str\mutex)

    ProcedureReturn retval
  EndIf
EndProcedure

Procedure.s StrLenS(*Str._STRING)
  ; returns string length as string
  ProcedureReturn StrU(StrLen(*Str._STRING),#LONG)
EndProcedure


Procedure.s StrGet(*Str._STRING)
  ; Get string
  If *Str

    WaitForSingleObject_(*Str\mutex,#INFINITE)
      retval.s = PeekS(*Str\adr)
    ReleaseMutex_(*Str\mutex)

    ProcedureReturn retval
  EndIf
EndProcedure


Procedure StrChgCh(*Dest._STRING,char.s)
  ; change string to new chars
  If *Dest And char
    WaitForSingleObject_(*Dest\mutex,#INFINITE)
      If StrNewAlloc(*Dest,Len(char)) ; alloc new
        CopyMemory(@char,*Dest\adr,Len(char))
        retval = *Dest
      EndIf
    ReleaseMutex_(*Dest\mutex)
  EndIf
  ProcedureReturn retval
EndProcedure


Procedure StrChg(*Dest._STRING,*Src._STRING)
  ; change string to new String
  If *Dest And *Source
    WaitForSingleObject_(*Dest\mutex,#INFINITE)
    WaitForSingleObject_(*Src\mutex ,#INFINITE)
      If StrNewAlloc(*Dest,*Src\len) ; alloc new
        CopyMemory(*Src\adr,*Dest\adr,*Src\len)
        retval = *String
      EndIf
    ReleaseMutex_(*Dest\mutex)
    ReleaseMutex_(*Src\mutex)
  EndIf
  ProcedureReturn retval
EndProcedure


Procedure StrAddCh(*Dest._STRING,char.s)
  ; add chars to string
  If *Dest And char
    WaitForSingleObject_(*Dest\mutex,#INFINITE)
      oldStr = *Dest\adr
      oldLen = *Dest\len
      If StrReAlloc(*Dest,oldLen+Len(char)) ; alloc new
        CopyMemory(oldStr, *Dest\adr        ,oldLen)
        CopyMemory(@char ,(*Dest\adr)+oldLen,Len(char)+1)
        GlobalFree_(oldStr)
        retval = *Dest
      EndIf
    ReleaseMutex_(*Dest\mutex)
  EndIf
EndProcedure


Procedure StrAdd(*Dest._STRING,*Src._STRING)
  ; add string to string
  If *Dest And *Src
    WaitForSingleObject_(*Dest\mutex,#INFINITE)
    WaitForSingleObject_(*Src\mutex ,#INFINITE)
      oldStr = *Dest\adr
      oldLen = *Dest\len
      If StrReAlloc(*Dest,(*Dest\len) + (*Src\len)) ; alloc new
        CopyMemory(oldStr,*Dest\adr,oldLen)
        CopyMemory(*Src\adr,(*Dest\adr)+oldLen,*Src\len)
        GlobalFree_(oldStr)
        retval = *Dest
      EndIf
    ReleaseMutex_(*Dest\mutex)
    ReleaseMutex_(*Src\mutex)
  EndIf
EndProcedure


Procedure EndStrings()
  ; clears and deletes internal string list
  ResetList(String_Collector())
  Debug "Deleting "+StrU(CountList(String_Collector()),#LONG)+" strings."

  While NextElement(String_Collector())
    *String._String = String_Collector()
    GlobalFree_ (*String\adr)
    CloseHandle_(*String\mutex)
    GlobalFree_ (*String)
  Wend

  ClearList(String_Collector())
EndProcedure


;- App Start
Global As, Bs
Global ThreadEnd

Procedure Thread1(thread)
  Shared As, Bs
  Repeat
    StrChgCh(Bs,"CDE")
    StrAdd  (As,Bs)
    ;PrintN( StrGet(As) )
  Until ThreadEnd
EndProcedure

Procedure Thread2(thread)
  Repeat
    StrChgCh(Bs,"")
    StrAdd  (As,Bs)
    ;Delay(10)
  Until ThreadEnd
EndProcedure



As = StrCreate()
Bs = StrCreate2("Hello World!")


OpenConsole()
  PrintN( StrLenS(As) )
  PrintN( StrGet (As) )
  PrintN( StrLenS(Bs) )
  PrintN( StrGet (Bs) )

  PrintN("")
  StrChgCh(As,"Yo! Yo!")
  PrintN( StrLenS(As) )
  PrintN( StrGet (As) )

  StrAddCh(As," - add somthing. ")
  PrintN( StrLenS(As) )
  PrintN( StrGet (As) )

  StrAdd(As,Bs)
  PrintN( StrLenS(As) )
  PrintN( StrGet (As) )

  PrintN(""):PrintN("Starting Multithreading Test. Please press <RETURN>"):Input()
  Beep_(800,100)

  t1 = CreateThread(@Thread1(),0)
  t2 = CreateThread(@Thread2(),0)
  ;CreateThread(@Thread3(),0)

  Delay(10000) ; Run Threads for a while...
  ;Delay(60000) ; 1 minute
  ThreadEnd = #TRUE
  Beep_(800,100)

  WaitThread(t1)
  WaitThread(t2)

  PrintN(""):PrintN("End. Please press <RETURN>"):Input()
CloseConsole()


;- App End
Cleanup:
EndStrings()

Maybe it helps if somebody needs thread safety NOW.

[Iria]

Thanks for the reply, yes I have indeed looked at the code previously provided on the forums and as 'people should' they have looked to the underlying OS to provide managed thread behaviour.

However my issue is I want to do this on a Linux platform and ideally not have to write an alternative version for the Windows platform.

There are thrid party libraries that provide cross platform thread management, however I really think this is an intrinsic set of functionality and a real need for PB to provide it (even though it will ultimately be an abstract of the underlying OS thread management API)

Its this PB abstract thats needed.

[Danilo]

Its this PB abstract thats needed.

I understood that, Iria - and i would also prefer Lock() and UnLock()
or thread safety directly built-in in PB.

Its just a thing Fred cant add in 2 days, because every single
PB command needs the thread stuff added - and all this on
all platforms.
Lets hope the best, maybe it will come one day...

[blueznl]

just ran into my first thread issue... grrr... now this would be nice to have in pb4... (hint)