using semaphores in threads and TrySemaphore(Semaphore)

[ss3e55]

Hello I'm using PB on windows, the application is user driven by buttons on the UI. These initiate the transmission of JSON serial messages over a serial port (actually emulated USB serial port) selected earlier in the UI flow. The messages cause actions on a remote embedded ESP32 device, that response with its own reply (also JSON) over the serial link. I have traffic working, when I use a simple approach, but I wanted a more sophisticated design...here's the issue

MY user via the UI posts messages to a queue and there's a separate Thread that removes messages off the queue when it sees the semaphore signal to telling it to take them off (in order) and send over the serial link - This approach allows async, messaging TX and RX (separate threads)..there's a receiving thread to collect messages and it does a similar thing...in reverse... it adds these replies to a receiver queue and signals the handler back in the UI thread to check and display their results.

The issue I have is so simple yet I can't seem to fix it...during debug I see the UI post the semaphore, and later in the other thread I stop the execution and NEVER see a valid semaphore count other than the quiescent ZERO.

Things I checked...

• Of course I run thread safe compilation
  I use mutex locks on all manipulation of the queues
  I use SHARE (nad use GLOBALS) on the named variables semaphores in case they are in different address spaces.
  When I look at the semaphores I see their addresses (32 bit) but never their values (I need a tutorial on the debugger for this issue) to check if it increments as expected.
  I do see the queues increase in length and I see the messages on the queues.
  BTW I had a difficult time using #PB_ANY to open the serial port and use the returned number and use it across different threads (again declaring SHARE) but this was avoided by using my own number and that worked (I don't know why) - I also tried passing it thought the createthread() paramter - no luck

So why doesn't TrySemaphore(Semaphore) ever see the semaphore increase from zero ? I see the code to set it being executed...

Code: Select all

  LockMutex(CommandQueueMutex)
  
  ; Add to command queue using list commands  
  AddElement(CommandQueue())
  CommandQueue()\CommandText = Command
  CommandQueue()\Timestamp = Date()
  CommandQueue()\CommandID = Random(999999)
  
  UnlockMutex(CommandQueueMutex)
  
  ; Signal sender thread
  SignalSemaphore(CommandAvailableSemaphore) ; <-- tells process  something in command list

Anyone know what might be wrong? Am I doing something simple but wrong?
I appreciate all the help you can lend me...
Phil

[idle]

would help if you show the threaded procedure that's waiting for the semephore

[PBJim]

We use a similar principle to what you describe, though for network traffic and it has always worked very well. I notice that you do not set the last element before you add a new element to the list. Perhaps you've accounted for that in your logic, but it's the first thing I noticed, on seeing your code.

You might need to set the current element after you've locked your list, because another thread will have moved the current element. Something like the below, sorry I didn't adapt it to yours, but taken from other code :

Code: Select all

              LockMutex(RecMutex)                                      ; Lock data received list mutex
              LastElement(*RecData())                                  ; Add received data to end of list
              AddElement(*RecData())

[ss3e55]

Thank you both for replying - I'm glad of any help on this...
The lists adding and removing are all (producer and consumer) behind mutex and so should be ok...I can't actually check that yet as the semaphore refuses to signal the thread to use the messages on the queue. I can add more messages from the producer onto the queue and that seems fine.
All producers use this pattern (here's the status producer)

Code: Select all

Procedure AddStatusUpdate(StatusText.s, StatusType.l)
  LockMutex(StatusQueueMutex)
  
  AddElement(StatusQueue())
  StatusQueue()\StatusText = StatusText
  StatusQueue()\Timestamp = Date()
  StatusQueue()\StatusType = StatusType
  
  UnlockMutex(StatusQueueMutex)
  
  SignalSemaphore(StatusAvailableSemaphore)
EndProcedure

2nd example (using the semaphore in question) - ALL Semaphores and Mutexs are GlOBAL variables

Code: Select all

Procedure SendJSONCommand(Command.s)

  
  ; Basic JSON validation
  If CountString(Command, "{") <> CountString(Command, "}")
    MessageRequester("Error", "Invalid JSON: Mismatched braces")
    ProcedureReturn
  EndIf
  
  LockMutex(CommandQueueMutex)
  
  ; Add to command queue using PureBasic list commands
  AddElement(CommandQueue())
  CommandQueue()\CommandText = Command
  CommandQueue()\Timestamp = Date()
  CommandQueue()\CommandID = Random(999999)
  
  ; here we should be loading the exact type of command on the queue and signal the sender to send!
  
  UnlockMutex(CommandQueueMutex)
  
  ; Signal serial handler thread
  SignalSemaphore(CommandAvailableSemaphore)  ; This SHOULD start a transmist in the SerialHanlerThread (ln 365) ????
  
  ; Update UI command list
  LockMutex(UIUpdateMutex)
  
  ; Extract command name for abbreviated display
  Protected CommandName.s = "unknown"
  Protected Pos = FindString(Command, Chr(34) + "command" + Chr(34) + ":")
  If Pos > 0
    Pos = FindString(Command, Chr(34), Pos + 10)
    If Pos > 0
      Protected EndPos = FindString(Command, Chr(34), Pos + 1)
      If EndPos > 0
        CommandName = Mid(Command, Pos + 1, EndPos - Pos - 1)
      EndIf
    EndIf
  EndIf
  
  AddGadgetItem(#CommandListGadget, -1, FormatDate("%hh:%ii:%ss", Date()) + " - " + CommandName)
  
  ; Keep list size manageable
  If CountGadgetItems(#CommandListGadget) > #MAX_COMMAND_HISTORY
    RemoveGadgetItem(#CommandListGadget, 0)
  EndIf
  
  ; Auto-scroll to bottom
  SetGadgetState(#CommandListGadget, CountGadgetItems(#CommandListGadget) - 1)
  
  UnlockMutex(UIUpdateMutex)
EndProcedure

And the consumers are the same and look like this pattern (some minor differences but the semaphore use is the same)

Code: Select all

; Message processor thread
Procedure ProcessorThread(*Dummy)
  Protected Reply.ReplyItem
  Protected DisplayText.s
  Protected HasReply.l = #False
  
  While Not ExitFlag
    ; Check for replies to process - non-blocking approach
    HasReply = #False
    
    ; Check if there are replies available without blocking
    If TrySemaphore(ReplyAvailableSemaphore) > 0
      ; Consume the semaphore signal
      ; SignalSemaphore(ReplyAvailableSemaphore)
      HasReply = #True
      
      ; Get reply from queue
      LockMutex(ReplyQueueMutex)
      
      If ListSize(ReplyQueue()) > 0
        FirstElement(ReplyQueue())
        Reply\ReplyText = ReplyQueue()\ReplyText
        Reply\Timestamp = ReplyQueue()\Timestamp
        Reply\MessageType = ReplyQueue()\MessageType
        Reply\RawData = ReplyQueue()\RawData
        DeleteElement(ReplyQueue())
      Else
        ; No replies in queue despite semaphore signal
        HasReply = #False
      EndIf
      
      UnlockMutex(ReplyQueueMutex)
      
      ; Process the reply
      If HasReply And Reply\ReplyText <> ""
        LockMutex(UIUpdateMutex)
        
        ; Format display text based on message type
        Select Reply\MessageType
          Case #MSG_IMAGE
            DisplayText = FormatDate("%hh:%ii:%ss", Reply\Timestamp) + " - [IMAGE] " + Left(Reply\ReplyText, 50) + "..."
            
            ; Decode and display image
            Protected ImagePos = FindString(Reply\ReplyText, "image_data")
            If ImagePos > 0
              ; Extract base64 data (simplified)
              Protected Base64Start = FindString(Reply\ReplyText, Chr(34), ImagePos + 10)
              If Base64Start > 0
                Protected Base64End = FindString(Reply\ReplyText, Chr(34), Base64Start + 1)
                If Base64End > 0
                  Protected Base64Data.s = Mid(Reply\ReplyText, Base64Start + 1, Base64End - Base64Start - 1)
                  DecodeBase64Image(Base64Data)
                EndIf
              EndIf
            EndIf
            
          Case #MSG_SENSOR_DATA
            DisplayText = FormatDate("%hh:%ii:%ss", Reply\Timestamp) + " - [SENSOR] " + Reply\ReplyText
            
          Default
            DisplayText = FormatDate("%hh:%ii:%ss", Reply\Timestamp) + " - " + Reply\ReplyText
        EndSelect
        
        ; Add to reply list
        AddGadgetItem(#ReplyListGadget, -1, DisplayText)
        
        ; Keep list size manageable
        If CountGadgetItems(#ReplyListGadget) > #MAX_REPLY_HISTORY
          RemoveGadgetItem(#ReplyListGadget, 0)
        EndIf
        
        ; Auto-scroll to bottom
        SetGadgetState(#ReplyListGadget, CountGadgetItems(#ReplyListGadget) - 1)
        
        UnlockMutex(UIUpdateMutex)
        
        ; Clear reply for next iteration
        Reply\ReplyText = ""
      EndIf
    EndIf
    
    ; Small delay to prevent CPU hogging
    Delay(1)
  Wend
  
  ProcedureReturn 0
EndProcedure

Apologies for the code being untidy - still working on logic a little...but first I need basic functionality to work. So why does
the try semaphore never see anything but zero.

Perhaps I can find this with a little help...
1) how can I look at the semaphores actual value
2) if in debug I see it IS non-zero then why on earth would the code "TrySemaphore()" step refuse to trigger the IF clause
OR
3) It never sets
Or
4) they are testing different addresses for the Semaphore (???) - that's why I tried SHARE

thanks Guys - any ideas??

[PBJim]

I found it difficult to debug your code and follow it, but there's just a logic problem. The below working code shows how it can be done and is runable. You might like to take this working code and then add what you need to it. Also TrySemaphore() that you've used is okay but you don't really need that in your loop, because there is nothing else to do anyway, so you might as well just use WaitSemaphore() as the trigger. Hope it helps!

Code: Select all

; ** Enable threadsafe compiler option

EnableExplicit

Global CommandMutex.i = CreateMutex()
Global Sema.i = CreateSemaphore()

Global NewList CommandList.s()

Define loop.i
Define text.s

; **
; **  Background thread, waits indefinitely for elements in CommandList() list, processes each that it finds, then deletes it
; **
Procedure ProcessThread(*dummy)
  
  Repeat
    
    Delay(1000)                                                         ; Wait a bit, so we at least get a chance to receive a few list elements at once
    WaitSemaphore(Sema.i)                                               ; Wait until we're told there's a new element in the list
    
    LockMutex(CommandMutex.i)                                           ; Grab the list exclusively for our thread
    
    While ListSize(CommandList())                                       ; Anything in the list?
      FirstElement(CommandList())                                       ; Pick up first element
      Debug "Picked up " + CommandList() + " from CommandList()"
      DeleteElement(CommandList())                                      ; Delete the first element
    Wend
    
    UnlockMutex(CommandMutex.i)                                         ; We're done with the elements, so release the mutex
  ForEver

EndProcedure


; **
; **  Main routine
; **
CreateThread(@ProcessThread(), 0)                                       ; Start our background thread

For loop.i = 1 To 50
  
  text.s = "Test command " + loop.i                                     ; Make an individual test string for the command list
  
  LockMutex(CommandMutex.i)                                             ; Make sure we have exclusive access to the list
  
  LastElement(CommandList())                                            ; Always put the new element at the end of the list
  AddElement(CommandList())
  CommandList() = text.s                                                ; Add our new element text
  UnlockMutex(CommandMutex.i)                                           ; Release our list for the thread to take over
  
  SignalSemaphore(Sema.i)                                               ; Tell the thread that there's something in the list
  
  Delay(Random(500,250))                                                ; Delay a bit so we send two or three commands together

Next

[idle]

In your process thread you should use waitsemephore so it effectively halts the thread at that point and it will execute as soon as it gets a signal then return to the wait.
Edit See pbjims above

[tored]

Here is an example of using TrySemaphore() within the main event loop that we don't want to block

Code: Select all

EnableExplicit

Global NewList messages.s()
Global mutex
Global semaphore

Enumeration
  #WINDOW
  #TEXT
EndEnumeration  

Procedure AddMessagesThread(*notused)
  Protected i
  Repeat
    Delay(1000)
    i + 1
    LockMutex(mutex)
    AddElement(messages())
    messages() = "Message " + Str(i)
    UnlockMutex(mutex)
    SignalSemaphore(semaphore)
  ForEver
EndProcedure


If OpenWindow(#WINDOW, 100, 100, 400, 300, "PureBasic Thread Semaphore Example", #PB_Window_SystemMenu | #PB_Window_ScreenCentered )
  TextGadget(#TEXT, 10, 10, 380, 280, "", #PB_Text_Border)
  
  mutex = CreateMutex()
  semaphore = CreateSemaphore()
  CreateThread(@AddMessagesThread(), 0)
  
  
  Repeat
    Select WaitWindowEvent(100)
      Case #PB_Event_CloseWindow
        Break
    EndSelect
    
    If TrySemaphore(semaphore)
      LockMutex(mutex)
      If ListSize(messages()) > 0
        Define msg.s = GetGadgetText(#TEXT)
        
        ForEach messages()
          msg + messages() + #CRLF$
        Next
        
        ClearList(Messages())
        SetGadgetText(#TEXT, msg)
      EndIf
      UnlockMutex(mutex)
    EndIf
    
  ForEver
  
  FreeSemaphore(semaphore)
  FreeMutex(mutex)
EndIf

Here is an example of only using mutex, note we use TryLockMutex() instead

Code: Select all

EnableExplicit

Global NewList messages.s()
Global mutex

Enumeration
  #WINDOW
  #TEXT
EndEnumeration  

Procedure AddMessagesThread(*notused)
  Protected i
  Repeat
    Delay(1000)
    i + 1
    LockMutex(mutex)
    AddElement(messages())
    messages() = "Message " + Str(i)
    UnlockMutex(mutex)
  ForEver
EndProcedure


If OpenWindow(#WINDOW, 100, 100, 400, 300, "PureBasic Thread Mutex Example", #PB_Window_SystemMenu | #PB_Window_ScreenCentered )
  TextGadget(#TEXT, 10, 10, 380, 280, "", #PB_Text_Border)
  
  mutex = CreateMutex()
  CreateThread(@AddMessagesThread(), 0)
  
  Repeat
    Select WaitWindowEvent(100)
      Case #PB_Event_CloseWindow
        Break
    EndSelect
    
    TryLockMutex(mutex)
    If ListSize(messages()) > 0
      Define msg.s = GetGadgetText(#TEXT)
      
      ForEach messages()
        msg + messages() + #CRLF$
      Next
      
      ClearList(Messages())
      SetGadgetText(#TEXT, msg)
    EndIf
    UnlockMutex(mutex)
    
  ForEver
  
  FreeMutex(mutex)
EndIf

[tored]

Here is another example where we instead of using a semaphore we use PostEvent() to send an event

Code: Select all

EnableExplicit

Global NewList messages.s()
Global mutex

Enumeration
  #WINDOW
  #TEXT
EndEnumeration

Enumeration #PB_Event_FirstCustomValue
  #MESSAGE
EndEnumeration  

Procedure AddMessagesThread(*notused)
  Protected i
  Repeat
    Delay(1000)
    i + 1
    LockMutex(mutex)
    AddElement(messages())
    messages() = "Message " + Str(i)
    UnlockMutex(mutex)
    
    PostEvent(#MESSAGE, #WINDOW, #PB_Ignore)
  ForEver
EndProcedure


If OpenWindow(#WINDOW, 100, 100, 400, 300, "PureBasic Thread Event Example", #PB_Window_SystemMenu | #PB_Window_ScreenCentered )
  TextGadget(#TEXT, 10, 10, 380, 280, "", #PB_Text_Border)
  
  mutex = CreateMutex()
  CreateThread(@AddMessagesThread(), 0)
  
  Define msg.s
  Repeat
    Select WaitWindowEvent(100)
      Case #PB_Event_CloseWindow
        Break
        
      Case #MESSAGE
        LockMutex(mutex)
        If ListSize(messages()) > 0
          msg = GetGadgetText(#TEXT)
          
          ForEach messages()
            msg + messages() + #CRLF$
          Next
          
          ClearList(Messages())
          SetGadgetText(#TEXT, msg)
        EndIf
        UnlockMutex(mutex)
    EndSelect
  
  ForEver
  
  FreeMutex(mutex)
EndIf

And here is yet another example where we include the data in the event rather than reading from a global message list

Code: Select all

EnableExplicit

Global mutex
Global NewList bufferMemory.String()
Global NewList *bufferPool.String()

Procedure AquireBuffer()
  Protected *string.String
  
  LockMutex(mutex)
  If Not ListSize(*bufferPool())
    *string = AddElement(bufferMemory()) 
  Else   
    FirstElement(*bufferPool())
    *string = *bufferPool()
    DeleteElement(*bufferPool())
  EndIf
  
  UnlockMutex(mutex)
  ProcedureReturn *string
EndProcedure  

Procedure ReleaseBuffer(*string.String)
  LockMutex(mutex)
  LastElement(*bufferPool())
  AddElement(*bufferPool())
  *bufferPool() = *string
  UnlockMutex(mutex)
EndProcedure

Enumeration
  #WINDOW
  #TEXT
EndEnumeration

Enumeration #PB_Event_FirstCustomValue
  #MESSAGE
EndEnumeration  

Procedure AddMessagesThread(*notused)
  Protected i, *buffer.String
  
  Repeat
    Delay(1000)
    i + 1
    *buffer = AquireBuffer()
    *buffer\s = "Message " + Str(i)
    PostEvent(#MESSAGE, #WINDOW, #PB_Ignore, #PB_Ignore, *buffer)
  ForEver
EndProcedure


If OpenWindow(#WINDOW, 100, 100, 400, 300, "PureBasic Thread Event Data Example", #PB_Window_SystemMenu | #PB_Window_ScreenCentered )
  TextGadget(#TEXT, 10, 10, 380, 280, "", #PB_Text_Border)
  
  mutex = CreateMutex()
  CreateThread(@AddMessagesThread(), 0)
  
  Define msg.s, *buffer.String
  Repeat
    Select WaitWindowEvent(100)
      Case #PB_Event_CloseWindow
        Break
        
      Case #MESSAGE
        *buffer = EventData();
        msg = GetGadgetText(#TEXT)
        msg + *buffer\s + #CRLF$
        SetGadgetText(#TEXT, msg)
        ReleaseBuffer(*buffer)
    EndSelect
    
  ForEver
  
  FreeMutex(mutex)
  FreeList(*bufferPool())
  FreeList(bufferMemory())
EndIf

Hope it helps.

[HeX0R]

For your second example you need no lists and no Mutex.
In the thread allocate a buffer, poke a string into it and send it by PostEvent() to the window queue.
In the window queue pick the buffer, peek the string and free the memory.
Nothing can collide here

[ss3e55]

Thank you all for the very helpful suggestions...some I knew and one or two added to my understanding...in the end the fault was with the serial port reading section of the same thread (after the queues and semaphores - which I have used in FreeRTOS and other systems with great affect) so it was really what I should have guess in the first place...Although I had a reasonably complex thread, queue, semaphore structure - that worked - the fault was in how I processed serial input characters (duh!)

again thank you so much everyone...I love PB for quick multiplatform test programs - its idea for that and more of course

[mk-soft]

Simple ... (for me)

Code: Select all

;-TOP

EnableExplicit

CompilerIf Not #PB_Compiler_Thread
  CompilerError "Use Compiler-Option ThreadSafe!"
CompilerEndIf

Enumeration CustomEvent #PB_Event_FirstCustomValue
  #MyEvent_ThreadText
EndEnumeration

Structure sThreadData
  ThreadID.i
  Cancel.i
EndStructure

Procedure AllocateString(String.s) ; Result = Pointer
  Protected *mem.string = AllocateStructure(String)
  If *mem
    *mem\s = String
  EndIf
  ProcedureReturn *mem
EndProcedure

Procedure.s FreeString(*mem.string) ; Result String
  Protected r1.s
  If *mem
    r1 = *mem\s
    FreeStructure(*mem)
  EndIf
  ProcedureReturn r1
EndProcedure

Procedure MyThread(*Data.sThreadData)
  Protected text.s, i
  
  PostEvent(#MyEvent_ThreadText, 0, 0, 0, AllocateString("Start ..."))
  Delay(100)
  For i = 1 To 100
    If *Data\Cancel
      Break
    EndIf
    text = "Message Number " + i
    PostEvent(#MyEvent_ThreadText, 0, 0, 0, AllocateString(text))
    Delay(250)
  Next
  PostEvent(#MyEvent_ThreadText, 0, 0, 0, AllocateString("Done."))
  
EndProcedure

Global MyThreadData.sThreadData

;-TOP Window

Procedure UpdateWindow()
  Protected dx, dy
  dx = WindowWidth(0)
  dy = WindowHeight(0) - StatusBarHeight(0) - MenuHeight()
  ; Resize Gadgets
  ResizeGadget(0, 5, 5, dx -10, dy - 10)
EndProcedure

Procedure Main()
  Protected dx, dy
  
  #WinStyle = #PB_Window_SystemMenu | #PB_Window_SizeGadget | #PB_Window_MaximizeGadget | #PB_Window_MinimizeGadget
  
  If OpenWindow(0, #PB_Ignore, #PB_Ignore, 600, 400, "Test Window", #WinStyle)
    ; MenuBar
    CreateMenu(0, WindowID(0))
    MenuTitle("&File")
    MenuItem(99, "E&xit")
    
    ; StatusBar
    CreateStatusBar(0, WindowID(0))
    AddStatusBarField(#PB_Ignore)
    
    ; Gadgets
    dx = WindowWidth(0)
    dy = WindowHeight(0) - StatusBarHeight(0) - MenuHeight()
    ListViewGadget(0, 5, 5, dx -10, dy - 10)
    
    ; Bind Events
    BindEvent(#PB_Event_SizeWindow, @UpdateWindow(), 0)
    
    ; Start Thread
    MyThreadData\ThreadID = CreateThread(@MyThread(), MyThreadData)
    
    ; Main Loop
    Repeat
      Select WaitWindowEvent()
        Case #PB_Event_CloseWindow
          Select EventWindow()
            Case 0
              MyThreadData\Cancel = #True
              If WaitThread(MyThreadData\ThreadID, 2000) = 0
                KillThread(MyThreadData\ThreadID)
              EndIf
              Break
          EndSelect
          
        Case #PB_Event_Menu
          Select EventMenu()
            Case 99
              PostEvent(#PB_Event_CloseWindow, 0, 0)
              
          EndSelect
          
        Case #PB_Event_Gadget
          Select EventGadget()
              
          EndSelect
          
        Case #MyEvent_ThreadText
          AddGadgetItem(0, -1, FreeString(EventData()))
          SetGadgetState(0, CountGadgetItems(0) - 1)
          SetGadgetState(0, -1)
          
      EndSelect
    ForEver
    
  EndIf
  
EndProcedure : Main()

[tored]

For your second example you need no lists and no Mutex.
In the thread allocate a buffer, poke a string into it and send it by PostEvent() to the window queue.
In the window queue pick the buffer, peek the string and free the memory.
Nothing can collide here

Correct. I just wanted to avoid multiple memory allocations, perhaps a bit too much of overengineering.

[tored]

For completeness, without the overengineering

Code: Select all

EnableExplicit

Enumeration
  #WINDOW
  #TEXT
EndEnumeration

Enumeration #PB_Event_FirstCustomValue
  #MESSAGE
EndEnumeration  

Procedure AddMessagesThread(*notused)
  Protected i, *buffer.String
  
  Repeat
    Delay(100)
    i + 1
    *buffer = AllocateStructure(String)
    *buffer\s = "Message " + Str(i)
    PostEvent(#MESSAGE, #WINDOW, #PB_Ignore, #PB_Ignore, *buffer)
  ForEver
EndProcedure


If OpenWindow(#WINDOW, 100, 100, 400, 300, "PureBasic Thread Event Data Simple Example", #PB_Window_SystemMenu | #PB_Window_ScreenCentered )
  TextGadget(#TEXT, 10, 10, 380, 280, "", #PB_Text_Border)
  
  CreateThread(@AddMessagesThread(), 0)
  
  Define msg.s, *buffer.String
  Repeat
    Select WaitWindowEvent(100)
      Case #PB_Event_CloseWindow
        Break
        
      Case #MESSAGE
        *buffer = EventData();
        msg = GetGadgetText(#TEXT)
        msg + *buffer\s + #CRLF$
        SetGadgetText(#TEXT, msg)
        FreeStructure(*buffer)
    EndSelect
    
  ForEver
EndIf

[mk-soft]

For completeness, without the overengineering

Code: Select all

EnableExplicit

Enumeration
  #WINDOW
  #TEXT
EndEnumeration

Enumeration #PB_Event_FirstCustomValue
  #MESSAGE
EndEnumeration 
...

The same as in my example