Hello Przemek
[Xbase++]
Here is the complete implementation detail how various
threads coordinate with each other :
Controlling threads using signals
------------------------------
There is a possibility for coordinating threads which does not require a
thread to terminate before another thread resumes. However, this requires
the usage of an object of the Signal class. The Signal object must be
visible in two threads at the same time. With the aid of a Signal object,
one thread can tell one or more other threads to leave their wait state and
to resume program execution:
Thread A Thread B
running running // simultaneous execution
| | // of program code
| oSignal:wait() // thread B stops
|
| wait state // thread A executes code
oSignal:signal()
| running // thread B resumes
| |
Whenever a thread executes the :wait() method of a Signal object, it enters
a wait state and stops program execution. The thread leaves its wait state
only after another thread calls the :signal() method of the same Signal
object. In this way a communication between threads is realized. One thread
tells other threads to leave their wait state and to resume program
execution.
Mutual exclusion of threads
--------------------------
As long as multiple threads execute different program code at the same time,
the coordination of threads is possible using wait states as achieved with
:synchronize() or ThreadWait(). However, wait states are not possible if the
same program code is running simultaneously in multiple threads. A common
example for this situation is adding/deleting elements to/from a globally
visible array:
PUBLIC aQueue := {}
FOR i:=1 TO 10000 // This loop cannot
Add( "Test" ) // be executed in
Del() // multiple threads
NEXT
**********************
FUNCTION Add( xValue )
RETURN AAdd( aQueue, xValue )
**************
FUNCTION Del()
LOCAL xValue
IF Len(aQueue) > 1
xValue := aQueue[1]
ADel ( aQueue, 1 )
ASize( aQueue, Len(aQueue)-1 )
ENDIF
RETURN xValue
In this example, the array aQueue is used to temporarily store arbitrary
values. The values are retrieved from the array according to the FIFO
principle (First In First Out). Function Add() adds an element to the end of
the array, while function Del() reads the first element and shrinks the
array by one element (note: this kind of data management is called a Queue).
When the functions Add() and Del() are executed in different threads, the
PUBLIC array aQueue is accessed simultaneously by multiple threads. This
leads to a critical situation in function Del() when the array has only one
element. In this case, a runtime error can occur:
Thread A Thread B
LOCAL xValue
IF Len(aQueue) > 1
Thread is Thread executes function completely
interrupted by the
operating system LOCAL xValue
IF Len(aQueue) > 1
xValue := aQueue[1]
ADel ( aQueue, 1 )
ASize( aQueue, Len(aQueue)-1 )
ENDIF
RETURN xValue
Thread resumes
xValue := aQueue[1]
Runtime error:
Meanwhile, the array is empty
The operating system can interrupt a thread at any time in order to give
another thread access to the CPU. If threads A and B execute function Del()
at the same time, it is possible that thread A is interrupted immediately
after the IF statement. Thread B may then run the function to completion
before thread A is scheduled again for program execution. In this case, a
runtime error can occur because the function Del() is not completely
executed in one thread before another thread executes the same function.
The example function Del() represents those situations in multi-threading
which require muliple operations to be executed in one particular thread
before another thread may execute the same operations. This can be resolved
when thread B is stopped while thread A executes the Del() function. Only
after thread A has run this function to completion may thread B begin with
executing the same function. Such a situation is called "mutual exclusion"
because one thread excludes all other threads from executing the same code
at the same time.
Mutual exclusion of threads is achieved in Xbase++ not on the
PROCEDURE/FUNCTION level but with the aid of SYNC methods. The SYNC
attribute for methods guarantees that the method code is executed by only
one thread at any time. However, this is restricted to one and the same
object. If one object is visible in multiple threads and the method is
called simultaneously with that object, the execution of the method is
serialized between the threads. In contrast, if two objects of the same
class are used in two threads and the same method is called with both
objects, the program code of the method runs parallel in both threads. As a
result, mutual exclusion is only possible if two threads attempt to execute
the same method with the same object. The object must be an instance of a
user-defined class that implements SYNC methods. A SYNC method is executed
entirely in one thread. All other threads are automatically stopped when
they attempt to execute the same method with the same object.
The example with the beforementioned PUBLIC array aQueue must be programmed
as a user-defined class in order to safely access the array from multiple
threads:
PUBLIC oQueue := Queue():new()
FOR i:=1 TO 10000 // This loop can run
oQueue:add( "Test" ) // simultaneously in
oQueue:del() // multiple threads
NEXT
***********
CLASS Queue // Class for managing
PROTECTED: // a queue
VAR aQueue
EXPORTED:
INLINE METHOD init
::aQueue := {} // Initialize array
RETURN self
SYNC METHOD add, del // Synchronized methods
ENDCLASS
METHOD Queue:add( xValue ) // Add an element
RETURN AAdd( ::aQueue, xValue )
METHOD Queue:del // Remove first element
LOCAL xValue // and shrink array
IF Len(::aQueue) > 1
xValue := aQueue[1]
ADel ( ::aQueue, 1 )
ASize( ::aQueue, Len(::aQueue)-1 )
ENDIF
RETURN xValue
In this example, the Queue class is used for managing a dynamic array that
may be accessed and changed from multiple threads simultaneously. The array
is referenced in the instance variable :aQueue and it is accessed within the
SYNC methods :add() and :del() . The Queue object which contains the array
is globally visible. The execution of the :del() method is automatically
serialized between multiple threads:
Thread A Thread B
| |
oQueue:del() |
| oQueue:del()
<...>
ADel( ::aQueue, 1 ) thread is stopped
<...>
RETURN xValue thread resumes
| <...>
| ADel( ::aQueue, 1 )
| <...>
| RETURN xValue
| |
When thread B wants to execute the :del() method while this method is
executed by thread A, thread B is stopped because it wants to execute the
method with the same Queue object. Therefore, SYNC methods are used whenever
multiple operations must be guaranteed to be executed in one thread before
another thread executes the same operations. A SYNC method can be executed
with the same object only in one thread at any time (Note: if a class method
is declared with the SYNC attribute, its execution is serialized for all
objects of the class).
Signal() - Class function of the Signal class.
------------------------------------------
Return
The function Signal() returns the class object of the Signal class.
Description
Signal objects are used to control program flow within one or more threads
from one other thread. To achieve this, it is necessary that one and the
same Signal object is visible in at least two threads. Therefore, Signal
objects can only be used in conjunction with Thread objects. In other words,
at least one additional Thread object must exist in an application for the
Signal class to become useful (note: the Main procedure is executed by an
implicit Thread object).
There are two methods in the Signal class: :signal() and :wait() . The
method :signal() triggers a signal, while the :wait() method causes a thread
to enter a wait state. If a thread executes the :wait() method, it waits for
the signal and suspends program execution. The thread resumes as soon as a
second thread executes the :signal() method with the same Signal object.
With the aid of Signal objects it becomes possible for the current thread to
control program flow in other threads. Each thread which executes the
:wait() method enters a wait state until another thread executes the
:signal() method.
Caution: Multiple Signal objects can be used to control program execution in
multiple threads. Since a Signal object limits program control in threads to
a wait state, so-called "dead lock" situations can occur when multiple
Signal objects are used. A dead lock is reached when thread A waits for a
signal from thread B, while, at the same time, thread B waits for a signal
from thread A.
Class methods
:new()
Creates an instance of the Signal class.
Instance variables
:cargo Instance variable for ad-hoc use.
Attribute: EXPORTED
Datatype: All data types
Methods
:signal() --> self
Triggers a signal for other threads.
:wait( [<nTimeOut>] ) --> lSignalled
Waits until another thread triggers the signal.
Example - 1
// Simple screen saver
// A simple screen saver for text mode applications is programmed
// in this example. The code for the screen saver runs in
// a separate thread which waits to be signalled for a maximum of
// 5 seconds. If no signal is triggered, the screen saver is
// activated. The thread is signalled from the Main procedure
// each time an event (keyboard, mouse) is taken from the
// event queue.
#include "AppEvent.ch"
PROCEDURE Main
LOCAL nEvent, oThread, oSignal
? "Press a key. ESC quits"
oThread := Thread():new()
oSignal := Signal():new()
oThread:start( "ScreenSaver", oSignal, 5 )
DO WHILE nEvent <> xbeK_ESC
nEvent := AppEvent(,,,0)
oSignal:signal()
? "Event =", nEvent
ENDDO
RETURN
** Procedure runs in separate thread
PROCEDURE ScreenSaver( oSignal, nSeconds )
LOCAL cScreen, cTemp
DO WHILE .T.
IF .NOT. oSignal:wait( nSeconds * 100 )
// This code is executed if no signal is triggered
// for more than <nSeconds> seconds
cScreen := SaveScreen( 0,0, MaxRow(), MaxCol() )
cTemp := SubStr( cScreen, 2 ) + Left( cScreen, 1 )
// Change screen until thread is signalled
DO WHILE .NOT. oSignal:wait(10)
cTemp := SubStr( cTemp, 3 ) + Left( cTemp, 2 )
RestScreen( 0,0, MaxRow(), MaxCol(), cTemp )
ENDDO
RestScreen( 0,0, MaxRow(), MaxCol(), cScreen )
cScreen := ;
cTemp := NIL
ENDIF
ENDDO
RETURN
Example - 2
// Controlling Append in two threads using signals
// A new database is created in this example and records from
// an existing database are appended. An array is used as a
// read/write buffer to transfer data from the existing to the
// new database. Reading and writing data runs simultaneously
// in two threads. Two Signal objects coordinate which thread
// may read and which one may write. A third signal causes
// the second thread to terminate.
PROCEDURE Main
LOCAL oReadDone, oWriteDone, oDoExit, oThread, aValues
USE LargeDB
DbCreate( "Test", DbStruct() )
oThread := Thread():new()
aValues := Array( FCount() )
oReadDone := Signal():new()
oWriteDone := Signal():new()
oDoExit := Signal():new()
oThread:start( "AppendTo", "Test" , aValues , ;
oReadDone, oWriteDone, oDoExit )
oWriteDone:wait() // Wait until 2nd database
// is open
DO WHILE .T. // Transfer record to array
Aeval( aValues, {|x,i| x:=FieldGet(i) },,, .T. )
IF ! Eof()
oReadDone:signal() // 1st signal: record is read
SKIP
ELSE
oDoExit:signal() // 2nd signal: terminate thread #2
EXIT
ENDIF
oWriteDone:wait() // Wait until data is written
ENDDO // to new database
oThread:synchronize(0)
USE
RETURN
** Procedure runs in separate thread
PROCEDURE AppendTo( cDbFile, aValues, ;
oReadDone, oWriteDone, oDoExit )
USE (cDbFile) EXCLUSIVE
oWriteDone:signal() // Signal: database is open
DO WHILE .T.
IF oReadDone:wait(10) // Check if record is read
DbAppend() // Write data
AEval( aValues, {|x,i| FieldPut(i,x) } )
oWriteDone:signal() // Signal: Data is written
ELSEIF oDoExit:wait(10) // Got an Exit signal?
EXIT
ENDIF
ENDDO
USE
RETURN
//-------------------------------------------------//
More to follow.
Regards
Pritpal Bedi
--
View this message in context:
http://www.nabble.com/MT-and-Thread-Control---Signals-tp19523545p19523545.html
Sent from the Harbour - Dev mailing list archive at Nabble.com.
_______________________________________________
Harbour mailing list
Harbour@harbour-project.org
http://lists.harbour-project.org/mailman/listinfo/harbour
