Problem mit Threads

[Lord]

Hallo!

Ich habe hier folgenden Code:
ThreadX() sind die aufgerufenen Threads
DistrX() sind die verteilenden Prozeduren
In der LinkedList werden in \Thread der aufgerufene Thread vermerkt
und in \Status der "Bearbeitungsstand" der aktuelle Zeile, wobei
0=nicht bearbeiten, 1=soll bearbeitet werden und -1=ist bearbeitet
bedeutete.

Code: Alles auswählen

#TestMutex=#False

CompilerIf #TestMutex
  Global Mutex=CreateMutex()
CompilerEndIf

Structure Daten
  Thread.i
  Status.i
  Pos.i
  Text.s
EndStructure

Enumeration #PB_Event_FirstCustomValue
  #myEvent1
  #myEvent2
  #myEvent3
EndEnumeration

Global NewList Rows.Daten()

Procedure Thread1(*P)
  Static Thread1
  Thread1+1
  Debug "Start Thread1 ["+Thread1+"]"
  Delay(Random(2000)); Payload
  PostEvent(#myEvent1)
  Debug "Thread1 ["+Thread1+"] PostEvent myEvent1"
EndProcedure
Procedure Thread2(*P)
  Static Thread2
  Thread2+1
  Debug #TAB$+#TAB$+"Start Thread2 ["+Thread2+"]"
  Delay(Random(2000)); Payload
PostEvent(#myEvent2)  
  Debug #TAB$+#TAB$+"Thread2 ["+Thread2+"] PostEvent myEvent2"
EndProcedure
Procedure Thread3(*P)
  Static Thread3
  Thread3+1
  Debug #TAB$+#TAB$+#TAB$+#TAB$+"Start Thread3 ["+Thread3+"]"
  Delay(Random(2000)); Payload
PostEvent(#myEvent3)  
  Debug #TAB$+#TAB$+#TAB$+#TAB$+"Thread3 ["+Thread3+"] PostEvent myEvent3"
EndProcedure
Procedure Distr1()
  Static Distr1
  Distr1+1
  Debug "Enter Distr1("+Distr1+")"
  Anz=CountGadgetItems(1)
  i=0
  CompilerIf #TestMutex
    ULM=#True
    LockMutex(Mutex)
    Debug "Mutex locked 1"
  CompilerEndIf
  
  Repeat
    ChangeCurrentElement(Rows(), GetGadgetItemData(1, i))
    If Rows()\Status=1; ist selektiert
      If Rows()\Thread=1; Thread 1 
        Rows()\Status=-1:Debug "Thread1: mark finished "+i
      EndIf
      Break
    EndIf
    i+1
  Until i=Anz
  
  i=0
  Repeat
    ChangeCurrentElement(Rows(), GetGadgetItemData(1, i))
    If Rows()\Status=1
      If Rows()\Thread=0
        Rows()\Thread=1:Debug "Thread1: load line "+i
        *P=GetGadgetItemData(1, i)
        CompilerIf #TestMutex
          UnlockMutex(Mutex)
          ULM=#False
          Debug "Mutex unlocked 1"
        CompilerEndIf
        CreateThread(@Thread1(), *P)
        Break
      EndIf
    EndIf
    i+1
  Until i=Anz
  CompilerIf #TestMutex
    If ULM=#True
      UnlockMutex(Mutex)
      ULM=#False
      Debug "Mutex unlocked 1 (Anz=i)"
    EndIf
  CompilerEndIf
  ForEach Rows()
    St.s+"["+Rows()\Status+"]"
    Th.s+"["+Rows()\Thread+"]"
  Next
  Debug St
  Debug Th
  Debug "Leave Distr1("+Distr1+")"
EndProcedure
Procedure Distr2()
  Static Distr2
  Distr2+1
  Debug #TAB$+#TAB$+"Enter Distr2("+Distr2+")"
  Anz=CountGadgetItems(1)
  i=0
  CompilerIf #TestMutex
    ULM=#True
    LockMutex(Mutex)
    Debug #TAB$+#TAB$+"Mutex locked 2"
  CompilerEndIf
  Repeat
    ChangeCurrentElement(Rows(), GetGadgetItemData(1, i))
    If Rows()\Status=1; ist selektiert
      If Rows()\Thread=2; Thread 1
        Rows()\Status=-1:Debug #TAB$+#TAB$+"Thread2: mark finished "+i
      EndIf
      Break
    EndIf
    i+1
  Until i=Anz
  
  i=0
  Repeat
    ChangeCurrentElement(Rows(), GetGadgetItemData(1, i))
    If Rows()\Status=1
      If Rows()\Thread=0
        Rows()\Thread=2:Debug #TAB$+#TAB$+"Thread2: load line "+i
        *P=GetGadgetItemData(1, i)
        CompilerIf #TestMutex
          UnlockMutex(Mutex)
          ULM=#False
          Debug #TAB$+#TAB$+"Mutex unlocked 2"
        CompilerEndIf
        CreateThread(@Thread2(), *P)
        Break
      EndIf
    EndIf
    i+1
  Until i=Anz
  CompilerIf #TestMutex
    If ULM=#True
      UnlockMutex(Mutex)
      Debug #TAB$+#TAB$+"Mutex unlocked 2 (Anz=i)"
    EndIf
  CompilerEndIf 
  ForEach Rows()
    St.s+"["+Rows()\Status+"]"
    Th.s+"["+Rows()\Thread+"]"
  Next
  Debug St
  Debug Th
 Debug #TAB$+#TAB$+"Leave Distr2("+Distr2+")" 
EndProcedure
Procedure Distr3()
  Static Distr3
  Distr3+1
  Debug #TAB$+#TAB$+#TAB$+#TAB$+"Enter Distr3("+Distr3+")"
  Anz=CountGadgetItems(1)
  i=0
  CompilerIf #TestMutex
    ULM=#True
    LockMutex(Mutex)
        Debug #TAB$+#TAB$+#TAB$+#TAB$+"Mutex locked 3"
  CompilerEndIf
  Repeat
    ChangeCurrentElement(Rows(), GetGadgetItemData(1, i))
    If Rows()\Status=1; ist selektiert
      If Rows()\Thread=3; Thread 1
        Rows()\Status=-1:Debug #TAB$+#TAB$+#TAB$+#TAB$+"Thread3: mark finished "+i
      EndIf 
      Break
    EndIf
    i+1
  Until i=Anz
  i=0
  Repeat
    ChangeCurrentElement(Rows(), GetGadgetItemData(1, i))
    If Rows()\Status=1
      If Rows()\Thread=0
        Rows()\Thread=3:Debug #TAB$+#TAB$+#TAB$+#TAB$+"Thread3: load line "+i
        CompilerIf #TestMutex
          UnlockMutex(Mutex)
          ULM=#False
    Debug #TAB$+#TAB$+#TAB$+#TAB$+"Mutex unlocked 3"
        CompilerEndIf
        *P=GetGadgetItemData(1, i)
        CreateThread(@Thread3(), *P)
        Break
      EndIf
    EndIf
    i+1
  Until i=Anz
  CompilerIf #TestMutex
    If ULM=#True
      UnlockMutex(Mutex)
    Debug #TAB$+#TAB$+#TAB$+#TAB$+"Mutex unlocked 3 (Anz=i)"
    EndIf     
  CompilerEndIf
  ForEach Rows()
    St.s+"["+Rows()\Status+"]"
    Th.s+"["+Rows()\Thread+"]"
  Next
  Debug St
  Debug Th
  Debug #TAB$+#TAB$+#TAB$+#TAB$+"Leave Distr3("+Distr3+")" 
EndProcedure
Procedure Start()
  Debug "Enter Start()"
  Anz=CountGadgetItems(1)
    Repeat
      If GetGadgetItemState(1, i)&#PB_ListIcon_Checked
        ChangeCurrentElement(Rows(), GetGadgetItemData(1, i))
        Rows()\Status=1
        count+1
      EndIf
      St.s+"["+Rows()\Status+"]"
      i+1
    Until i=Anz
    Debug count
    Debug St
    Debug "---------------"
    Distr1()
    Distr2()
    Distr3()
    Debug "Leave Start()"
EndProcedure
Procedure FillList(Anz)
  For i = 1 To Anz
    AddGadgetItem(1, -1, RSet(Str(i), 4, "0")+Chr(10)+"Row "+Str(i))
    *P=AddElement(Rows())
    SetGadgetItemState(1, i-1, #PB_ListIcon_Checked)
    Rows()\Text="Col "+Str(i)
    Rows()\Pos=i
    SetGadgetItemData(1, i-1, *P)
  Next
EndProcedure

OpenWindow(1, 10, 10 ,640, 480, "")
ListIconGadget(1, 0, 0, WindowWidth(1), WindowHeight(1)-30, "Col 1", 120, #PB_ListIcon_CheckBoxes)
AddGadgetColumn(1, 1, "Col 2", 120)
AddGadgetColumn(1, 2, "Col 3", 120)
AddGadgetColumn(1, 3, "Col 4", 120)

ButtonGadget(11, 10, GadgetHeight(1)+2, 64, 24, "Start")
BindGadgetEvent(11, @Start())

BindEvent(#myEvent1, @Distr1())
BindEvent(#myEvent2, @Distr2())
BindEvent(#myEvent3, @Distr3())

FillList(20)


quit=#False
Repeat
  Event=WaitWindowEvent()
  Select Event
    Case #PB_Event_CloseWindow
      quit=#True
  EndSelect
Until quit=#True

Es sollen 3 Threads die Liste nacheinander "abarbeiten".

Mit nur einem Thread funktionierte es ohne Probleme:
in der Procedure Start() die aufrufe Distr2() und Distr3() auskommentieren:
alle Einträge Rows()Status werden auf -1 gesetzt.

Kommen weitere Threads hinzu, werden diese auch aktiviert und durch-
laufen, aber nach etwa der 11. Zeile wird in der Liste Rows()\Status die
bearbeitete Zeile nicht mehr als bearbeitet (-1) markiert.
Daß die Threads gestartet wurden, wird in Rows()\Thread vermerkt.
Wie kann das sein?
Irgendwo muß ich hier doch einen Denkfehler gemacht haben.

[PureLust]

Nach den Programmstart sind bei mir Spalte 1 & 2 gefüllt, Spalte 3 & 4 leer.

Nach klick auf 'Start' werden einige Debug-Informationen angezeigt - in der Liste passiert nichts.

Was genau soll denn passieren bzw. wie soll's denn aussehen?

[Lord]

Hallo PureLust!

In der Liste soll auch nichts passieren (jedenfalls jetzt noch nichts).

Der Debug-Output ist der Ort des Geschehens:

Die Ausgabe der Infos zu den einzelnen Threads ist in einzelne
Spalten geschrieben. Ereignisse betreffend Thread 1 in Spalte 1,
Ereignisse von Thread 2 in Spalte 2 usw. des Debug-Outputs.

Desweiteren wird der Zustand der Liste angezeigt:

z.B.:
[-1][-1][-1][1][1][1][1][1][1][1][1][1][1][1][1][1][1][1][1][1]
[1][2][3][2][2][1][3][2][2][1][3][1][1][0][0][0][0][0][0][0]

Die obere Reihe kennzeichnet, von links nach rechts, den Status
der Reihen. Im Beispiel sind Reihe 1 bis 3 bereits bearbeitet, die
Restlichen Reihen sind noch nicht abgearbeitet.
Die unterer Reihe zeigt an, welcher Thread die jeweilige Reihe be-
arbeitet/bearbeitet hat.

Das Problem, das ich habe ist, daß nach einem kompletten Durch-
lauf der Reihen nicht alle Reihen als bearbeitet markiert sind.
Zu erkennen ist dies daran, daß nicht alle Reihen mit -1 gekenn-
zeichnet sind, obwohl alle Reihen von einem Thread bearbeitet
wurden:

Code: Alles auswählen

[-1][-1][-1][-1][-1][-1][-1][1][1][1][1][1][1][1][1][1][1][1][1][1]
[1][2][3][2][2][1][3][2][2][1][3][1][1][2][1][3][1][2][3][3]

Irgendwie bricht das Markieren der Reihen ab und ich weiß nicht
warum. Wie oben geschrieben, passiert das mit einem Thread nicht.

[PureLust]

Hallo Lord,

sorry, ich blicke immer noch nicht ganz dahinter was Du da genau machst bzw. versuchst.

Aber was ich so bislang herauslesen konnte, gehst Du die Sache mit den Threads etwas falsch an.

Mal generell etwas dazu:

- Du kannst ein-und-die-selbe Thread-Routine mehrfach starten (Du brauchst also nicht 3 Routinen für 3 Threads).
- Wenn Du innerhalb eines Threads auf globale Variablen oder Listen zugreifst, solltest Du vorher einen Mutex setzen um den anderen Threads den Zugriff auf diese Variablen und Listen zu verbieten.
Wenn Du mit der Bearbeitung der Daten fertig bist, kannst Du den Mutex wieder entfernen.

Ich mach Dir gleich mal ein Beispiel, aber zuerst ruft leider das Real-Life.

[NicTheQuick]

Schau mal nach ConcurrentQueue hier im Forum, auf meiner Webseite oder meinem YouTube-Kanal. Sowas brauchst du.

[Lord]

...
- Wenn Du innerhalb eines Threads auf globale Variablen oder Listen zugreifst, solltest Du vorher einen Mutex setzen um den anderen Threads den Zugriff auf diese Variablen und Listen zu verbieten.
...

Das hatte ich in meinem Beispiel auch versucht. Selbes Ergebnis.
Setze mal in der ersten Zeile die Konstante #TestMutex auf #True.

[Lord]

Schau mal nach ConcurrentQueue hier im Forum, auf meiner Webseite oder meinem YouTube-Kanal. Sowas brauchst du.

Werde ich mir noch ansehen.
Aber zuerst möchte ich verstehen, was an meinem Code falsch ist.
Theoretisch dürfte das nicht passieren, oder?

[NicTheQuick]

Ich hab ihn mir nicht angeschaut, da ich ohne Laptop gerade im Zug unterwegs bin. Vielleicht hab ich dafür später Zeit.

[PureLust]

Wenn Du in Threads mit Listen arbeitest, musst Du noch beachten, dass Du die ListenPosition zwischenspeicherst bevor Du sie veränderst und sie nachher wieder zurück setzt.
Sonst verändert der eine Thread die ListenPosition des anderen Threads oder des Hauptprogramms.

Wie's funktioniert kannst Du Dir hier am Beispiel mal anschauen:

Code: Alles auswählen

EnableExplicit

Global	Fertig_Zaehler = 0				; Zähler für die fertig bearbeiteten Elemente
Global	myMutex =	CreateMutex()		; Mutex, um parallele Thread-Zugriffe zu verhindern
Define	n

; Test-Liste mit unbearbeiteten Einträgen erstellen

Global NewList ToDo()
For n = 1 To 40
	AddElement(ToDo())
	ToDo() = 1						; Status:  1 = unbearbeitet, 0 = in Bearbeitung,  -1 = fertig bearbeitet
Next

Procedure	myThread(*Nr)
	
	Protected	myListPos.i, ActListPos.i		; WICHTIG:  Zwischenspeicher für ListenPositionen
	Protected	Fertig
	
	Repeat
		
		LockMutex(myMutex)
		
		ActListPos = ListIndex(ToDo())			; WICHTIG:  Bevor die Listenposition in einem Thread verändert wird,
															; diese zwischenspeichern und anschließend wieder zurück setzen
		Fertig		= #True
		
		ForEach ToDo()
			If ToDo() = 1
				ToDo() = 0		; Eintrag auf "in Bearbetung" setzen
				myListPos = ListIndex(ToDo())		; eigene Listenposition zwischenspeichern
				Fertig = #False
				Break
			EndIf
		Next
		
		If Not Fertig		; Es ist noch ein Eintrag zu bearbeiten
			
			Debug	"Tread "+Str(*Nr)+":  Bearbeite Element - "+Str(ListIndex(ToDo())+1)
			
			SelectElement(ToDo(), ActListPos)		; vor Freigabe des Mutex, die ListenPosition wieder zurück setzen
			UnlockMutex(myMutex)
			
			; Dummy Bearbeitung
			
			Delay(1000+Random(1000))
			
			; Eintrag als bearbeitet markieren
			
			
			LockMutex(myMutex)
			
			ActListPos = ListIndex(ToDo())			; Listenposition wieder zwischenspeichern
			
			SelectElement(ToDo(), myListPos)			; ListenPosition wieder auf die eigene Position setzen
			
			Debug	"Tread "+Str(*Nr)+":  Element fertig - "+Str(ListIndex(ToDo())+1)
			
			ToDo() = -1										; Eintrag als fertig bearbeitet markieren
			Fertig_Zaehler + 1
			
			SelectElement(ToDo(), ActListPos)		; vor Freigabe des Mutex, die ListenPosition wieder zurück setzen
			UnlockMutex(myMutex)
			
		Else		; kein zu bearbeitender Einrag mehr ... also beende thread
			
			SelectElement(ToDo(), ActListPos)		; vor Freigabe des Mutex, die ListenPosition wieder zurück setzen
			UnlockMutex(myMutex)
			Debug	"Tread "+Str(*Nr)+":  wird beendet."
			Break
			
		EndIf
	Until #False
	
EndProcedure

Define	Thread1 = CreateThread(@myThread(), 1)
Define	Thread2 = CreateThread(@myThread(), 2)
Define	Thread3 = CreateThread(@myThread(), 3)
; Define	Thread4 = CreateThread(@myThread(), 4)
; Define	Thread5 = CreateThread(@myThread(), 5)
; Define	Thread6 = CreateThread(@myThread(), 6)
; Define	Thread7 = CreateThread(@myThread(), 7)

; Wärend die Threads laufen, kann irgend etwas gemacht werden

While Fertig_Zaehler < ListSize(ToDo())
	Debug	"Dumdideldumm ... "+Str(Fertig_Zaehler)+" Einträge wurden bisher bearbeitet."
	Delay(1000)
Wend

WaitThread(Thread1)
WaitThread(Thread2)
WaitThread(Thread3)
; WaitThread(Thread4)
; WaitThread(Thread5)
; WaitThread(Thread6)
; WaitThread(Thread7)

Viel Erfolg.

[Lord]

Wenn Du in Threads mit Listen arbeitest, musst Du noch beachten, dass Du die ListenPosition zwischenspeicherst bevor Du sie veränderst und sie nachher wieder zurück setzt.
Sonst verändert der eine Thread die ListenPosition des anderen Threads oder des Hauptprogramms.
...

Ich greife jeweils immer auf das richtige Element der Liste zu:

Code: Alles auswählen

...
ChangeCurrentElement(Rows(), GetGadgetItemData(1, i))
...

Der Zeiger wurde vorher in der Prozedur FillList() gespeichert:

Code: Alles auswählen

...
*P=AddElement(Rows())
...
SetGadgetItemData(1, i-1, *P)
...

Ich werde mal versuchen, Deinen Ansatz bei mir umzusetzen, da er
problemlos durchläuft.
Trotzdem möchte ich gerne verstehen, warum in meinem Code der
Status nicht durchgängig auf -1 gesetzt wird.