Code archives/Miscellaneous/WorkingQueue

This code has been declared by its author to be Public Domain code.

Download source code

WorkingQueue by Kurator
(Posted 18 years ago)
A Threadool creates a defined number of Threads. These Threads do not die after their work is done, so there is no syscall overhead when they are needed again. They rest in a pool, used simply when they are needed and some work has to be done.

The work is organised in so called "Tasks"

You can create 1000's of different Tasks, and add them to an Workingqueue - The queue is simply First in - First out.

The queue simply looks for unused threads in its threadspool, and assigns them some work. 
SuperStrict

' *** THREADPOOL CLASSES START ***

Type TWorkQueue

	Field nThreads:Int
	Field threads:TThread[]
	Field queue:TList
	
	Field mut_List:Int
	
	Function Create:TWorkQueue(nThreads:Int = 10)
		Local t:TWorkQueue = New TWorkQueue
		t.nThreads = nThreads
		t.threads = New TThread[nThreads]
		t.queue = New TList
		t.mut_List = CreateMutex()
		
		For Local i:Int = 0 To nThreads-1
			t.threads[i] = TThread.Create( QueueWrapper, Object(t) )
		Next
		
		Return t
	EndFunction
	
	Function QueueWrapper:Object( data:Object)
		
		Local myTask:TWorkQueue = TWorkQueue(data)
		myTask.Execute()
	
	EndFunction
	
	Method AddTask(obj:TTask)
		LockMutex(mut_List)
		  queue.AddLast(obj)
		UnlockMutex(mut_List)		
	EndMethod
	
	Method DoWork:Int()
		If Not queue.IsEmpty()
			LockMutex(mut_List)
			  Local t:TTask = TTask(queue.First())
			UnlockMutex(mut_List)
			
			Local i:Int = 0

			While i < nThreads
				If threads[i].GetStatus() = False
					threads[i].SetFunction(t.func, t.data)
					LockMutex(mut_List)
					  queue.RemoveFirst
					UnlockMutex(mut_List)
					threads[i].Start()
					Exit
				Else
					i :+ 1
				EndIf
			Wend

			Return True
		Else
			Return False
		EndIf
	EndMethod
	
	Method Execute()
	
		' Dummy
		
	EndMethod

EndType


Type TTask

	Field func:Object( data:Object )
	Field data:Object
	
	Method Run()
		If func Then func(data)
	EndMethod
	
	Function Create:TTask(func:Object( data:Object ), data:Object )
		Local t:TTask = New TTask
		t.func = func
		t.data = data
		Return t

	EndFunction
	
EndType

Type TThread
	
	Field id:Int
	Field mutex:Int
	
	Field func:Object( data:Object )
	Field data:Object
	
	Field working:Int

	
	Function Create:TThread(func:Object( data:Object ), data:Object)
		Local t:TThread = New TThread
		t.id    = CreateThread( ThreadWrapper, Object(t) )
		t.mutex = CreateMutex()
		t.func = func
		t.data = data
		t.working = False
		
		LockMutex(t.mutex)

		Return t

	EndFunction
	
	Function ThreadWrapper:Object( data:Object)
		
		Local myTask:TThread = TThread(data)
		myTask.Execute()
	
	EndFunction
	
	Method SetFunction(func:Object( data:Object ), data:Object)
		Self.func = func
		Self.data = data
		Self.working = False
	EndMethod

	Method GetStatus:Int()
		Return working
	EndMethod
	
	
	Method Destroy()
		DetachThread(id)
		id = 0
		CloseMutex(mutex)
		mutex = 0
		working = False
		func = Null
		data = Null
	EndMethod
	
	Method Execute()

		Repeat
			
			If Self.working = True 
				working = Self.DoWork()
			EndIf
			
			If Self.working = False
				LockMutex(mutex)
			EndIf
	
		Forever
	
	EndMethod
	
	Method DoWork:Int()
		If func Then func(data)
		Return False
	EndMethod
	
	Method Start()
		UnlockMutex(mutex)
		Self.working = True
	EndMethod
	
	Method Suspend()
		Self.working = False
	EndMethod
	
EndType

' *** THREADPOOL CLASSES END ***


' ***** EXAMPLE START ****
' Demo Functions, just generate some heavy Workload :)


Function fTest1:Object(data:Object)
	Local superCalcResult:Double
	For Local x:Int = 0 To 1000000
		superCalcResult = x*Cos(x)+Tan(x)*x/x
	Next
EndFunction

Function fTest2:Object(data:Object)
	Local superCalcResult:Double
	For Local x:Int = 0 To 1000000
		superCalcResult = x*Cos(x)+Tan(x)*x/x
	Next
EndFunction


'Create a WorkQueue with 8 preinitalised Threads waiting for work
Local q:TWorkQueue = TWorkQueue.Create(8)


'Create a LOT of Tasks to be done

Local test1:TTask[] = New TTask[200]
Local test2:TTask[] = New TTask[200]


'Assign the Functions you want to be done to the Tasks
'I'm pretty lazy - using, just the same functions :)
For Local i:Int = 0 To 199
	test1[i] =  TTask.Create(fTest1, String(i))
	test2[i] =  TTask.Create(fTest2, String(i))
Next

Print "Initialized..."
' To prove that ther is no CPU time wasted with 8 Threads waiting for work (look at the Taskmanager)
Delay 1000
Print "Starting..."


'Assign the 400 Tasks to the WorkQeue

For Local i:Int = 0 To 199
	q.AddTask(test1[i])
	q.AddTask(test2[i])
Next

Local starttime:Int = MilliSecs()

While Not KeyHit(KEY_ESCAPE)
	'Update the WorkQueue
	If Not q.DoWork() Exit
	'Print q.queue.Count()
	'it should also be possible to but this loop in an extra thread, maybe in future :)
Wend

Print "Duration "+String(MilliSecs()-starttime)+" ms"