Multithreading mit Threadpool - Warte auf letzten lesenden Thread

  • VB.NET

Es gibt 51 Antworten in diesem Thema. Der letzte Beitrag () ist von StGo.

    Multithreading mit Threadpool - Warte auf letzten lesenden Thread

    Hallo zusammen,

    ich habe eine Frage. Ich habe mir einen Code geschrieben mit dem ich parallel Excel Dateien auslesen kann. Diese Werte schreibe ich in ein Array und will dieses später in eine Excel Datei schreiben. Ich habe leider keine Idee wie ich am Ende des Threadpools auf den letzten noch lesenden Thread warte.

    Hier mein Code

    Quellcode

    1. Imports System.Threading
    2. Imports Microsoft.Office.Interop.Excel
    3. Imports Microsoft.VisualBasic.FileIO
    4. Imports Microsoft.VisualBasic.FileIO.FileSystem
    6. Module Module1
    7. '##################################################################################################################################################
    8. 'Alle Anpassungen der Variablen hier:
    9. Dim Dateipfad As String = "\\Eigene Dokumente\xls Test Dateien"
    10. Dim DateipfadAusgabe As String = "\\Eigene Dokumente\Test_File.xls"
    11. Dim DokumentenfeldAuslesen As String = "B1"
    12. Dim WorksheetNr As Integer = 1
    13. Dim Stackgroesse As Integer = 100
    14. '##################################################################################################################################################
    16. 'Deklaration Excel Mappe
    17. Dim oExcel As Microsoft.Office.Interop.Excel.Application
    18. Dim oBook As Microsoft.Office.Interop.Excel.Workbook
    19. Dim oSheet As Microsoft.Office.Interop.Excel.Worksheet
    21. 'Initialisiert eine neue Instanz der Semaphore-Klasse, gibt die maximale Anzahl gleichzeitiger Einträge an und reserviert optional einige Einträge.
    22. 'Semaphore(initialCount, maximumCount)
    23. 'initialCount: Die anfängliche Anzahl von Anforderungen für das Semaphor, die gleichzeitig gewährt werden können (Int32).
    24. 'maximumCount: Die maximale Anzahl von Anforderungen für das Semaphor, die gleichzeitig gewährt werden können (Int32).
    25. Dim sem As New Semaphore(1, 1)
    27. 'Stack erstellen
    28. Dim meinStack As New Stack(Stackgroesse)
    30. 'Variablen Deklaration incl. Dateipfad
    31. Dim Dokumentennummer As Integer
    32. Dim DateiCounter As System.Collections.ObjectModel.ReadOnlyCollection(Of String) = My.Computer.FileSystem.GetFiles(Dateipfad)
    33. Dim meinArray(DateiCounter.Count - 1, 1)
    34. Dim AnzahlThreads As Integer = 0
    36. 'Hautroutine
    37. 'Verzeichnis wird durchsucht und Dateipfade werden aufden Stack geschrieben
    38. Sub Main()
    39. For Each File As String In System.IO.Directory.GetFiles(Dateipfad)
    40. meinStack.Push(File)
    41. Next
    42. Console.WriteLine("Stack ist mit {0} Dokumentenpfaden gefüllt", meinStack.Count)
    43. 'Verbrauchroutine starten
    44. Verbraucher()
    46. End Sub
    48. 'In Verbraucher Sub Thread-Pool erzeugen
    49. Sub Verbraucher()
    50. While meinStack.Count <> 0
    52. ThreadPool.QueueUserWorkItem(AddressOf LeseDateiAus, AnzahlThreads)
    53. End While
    55. oExcel = New Microsoft.Office.Interop.Excel.Application
    56. oExcel.Visible = False
    57. oBook = oExcel.Workbooks.Add
    58. oSheet = oBook.Worksheets(1)
    59. oSheet.Range("A1").Value = "Dokumentennummer"
    60. oSheet.Range("B1").Value = "Pfad der Datei"
    61. oSheet.Range("A2").Resize(DateiCounter, 2).Value = meinArray
    62. oBook.SaveAs(DateipfadAusgabe)
    63. oBook.Close()
    64. oExcel.Quit()
    65. End Sub
    67. 'Threadsichere Sub
    68. Sub LeseDateiAus(ByVal state As Object)
    69. Console.WriteLine("Thread {0} tritt in das Semaphore ein", Thread.CurrentThread.ManagedThreadId)
    70. sem.WaitOne()
    71. Try
    72. oExcel = New Microsoft.Office.Interop.Excel.Application
    73. oExcel.Visible = False
    74. If meinStack.Count = 0 Then Exit Try
    75. oBook = oExcel.Workbooks.Open(meinStack.Pop, , True)
    76. oSheet = oBook.Worksheets(WorksheetNr)
    77. Dokumentennummer = oSheet.Range(DokumentenfeldAuslesen).Value
    78. oBook.Close()
    79. oExcel.Quit()
    80. Console.WriteLine(Dokumentennummer)
    82. meinArray(meinStack.Count, 0) = Dokumentennummer
    83. meinArray(meinStack.Count, 1) = meinStack.Pop
    84. Finally
    85. sem.Release()
    86. End Try
    87. Console.WriteLine("Thread {0} tritt aus dem Semaphore aus", Thread.CurrentThread.ManagedThreadId)
    88. End Sub
    89. End Module

    @StGo
    Willkommen im Forum. :thumbup:
    1. mach Dir eine EinForm-Anwendung, keine Console.
    2. mach Dir ein Flag-Array mit sovielen Elementen, wie Threads laufen.
    3. Lass jeden Event bei Beendigung ein Event senden.
      Im Eventhandler wird das kommunizierende Flag geändert.
    4. Überprüfe bei jedem Event, ob alle Flags geändert sind. Wenn ja, war es der letzte Thread.
    Falls Du diesen Code kopierst, achte auf die C&P-Bremse.
    Jede einzelne Zeile Deines Programms, die Du nicht explizit getestet hast, ist falsch :!:
    Ein guter .NET-Snippetkonverter (der ist verfügbar).
    Programmierfragen über PN / Konversation werden ignoriert!

    Danke für die schnelle Antwort. Aber ich kann im Threadpool doch nicht wissen wieviele Threads laufen. Das wird doch vom Scheduling bestimmt oder irre ich mich. Davon mal abgesehen ist eine Windows Forms Anwendung keine Option. Meine Vorgabe ist eine Konolenanwendung. Warum kann ich nicht beantworten.

    Gruß

    StGo schrieb:

    Aber ich kann im Threadpool doch nicht wissen wieviele Threads laufen.

    Brauchst du auch nicht. Du musst das Array(oder die Liste) nur mit so vielen Flags austatten, wie du Threads erstellst. Und beim Eventhandling eben einen weiteren Eintrag auf True setzen und prüfen, ob alle schon True sind.

    PS: Welches Framework nutzt du? Wenn du 4.5 nutzt kannst du(wenn erlaubt) auch mit Tasks arbeiten. Geht glaube ich auch schon mit 4.0, aber ohne Async/Await, sondern mit Task.WaitAll(...)(was sich wohl sowieso anbieten würde)...

    StGo schrieb:

    Aber ich kann im Threadpool doch nicht wissen
    Soeben hast Du gelernt, dass solch völlig unwichtige Information doch nicht ganz unwichtig ist und beim nächsten Problem unbedingt in den Eröffnungspost gehört. Die Lösung hat Dir @Higlav gegeben.
    Falls der erste Thread fertig ist, bevor der zweite anfängt, mach Dir noch ein Zusatzflag, das erst dann die beendeten Threads abgefragt werden, wenn der letzte Thread gestartet wurde.
    Sonst bist Du fertig, bevor Du angefangen hast.
    Falls Du diesen Code kopierst, achte auf die C&P-Bremse.
    Jede einzelne Zeile Deines Programms, die Du nicht explizit getestet hast, ist falsch :!:
    Ein guter .NET-Snippetkonverter (der ist verfügbar).
    Programmierfragen über PN / Konversation werden ignoriert!

    Ich frage mich gerade ob ich auf dem Schlauch stehe!?

    Ich weiss nicht wieviele Threads der Threadpool erzeugt ergo kann sie auch nicht in eine Liste packen und mit einem Flag versehen. ImSemaphore gebe ich nur an wieviele Startthreads es gibt und wie die Max Anzahl im Semaphore ist.

    Ich suche eien Art "Thrad Join" oder "WaitAll". Meine Hoffnung lag daran das sich jemand besser mit dem Thema auskennt als ich und eventuell eine erklärung parat hat. Was auch der Grund dafür ist das ich kein weiteres Wissen zum Threadpool im ersten Post habe.

    Gruß

    PS: .NET 4.0.30319

    Ich gebe zu, mit Semaphores oder Mutexes habe ich bis jetzt noch nicht gearbeitet(war auch nie nötig, genausowenig wie Threadsynchronisierung), aber bei .NET 4.0 funktioniert untenstehender Code:

    VB.NET-Quellcode

    1. Dim t1 = Task.Factory.StartNew(Function() "TestString")
    2. Dim t2 = Task.Factory.StartNew(Sub() Console.WriteLine("Hällöü"))
    3. Task.WaitAll(t1, t2)
    4. Console.WriteLine(t1.Result)

    Ich weiss nicht, wie, oder ob du das einbauen kannst, das wäre dein Part.

    StGo schrieb:

    ergo
    solltest Du fragen, wie solch geacht wird. X(
    In Deinem MainWindow / MainThread legst Du Dir eine Private MyThreadList As New List(Of IRGENDWAS) an. Struktue mit Thread-Kennung oder so.
    Irgendwo in Deinem Programm wird einer Deiner Threads aus dem MainWindow / MainThread heraus gestartet.
    Exakt da befüllst Du diese Liste mit einem speziell für diesen Thread generierten Item:

    VB.NET-Quellcode

    1. MyThreadList.Add(New (IRGENDWAS(paramX))
    und feddich.
    Falls Du diesen Code kopierst, achte auf die C&P-Bremse.
    Jede einzelne Zeile Deines Programms, die Du nicht explizit getestet hast, ist falsch :!:
    Ein guter .NET-Snippetkonverter (der ist verfügbar).
    Programmierfragen über PN / Konversation werden ignoriert!

    Es geht einfacher. Beispiel:

    VB.NET-Quellcode

    1. Dim threadCount As Int32
    2. Dim are As AutoResetEvent
    3. Dim lock As New Object
    5. Sub StartMyThreads()
    6. threadCount = 5
    7. are = New AutoResetEvent(False)
    9. For i = 1 To threadCount
    10. Console.WriteLine("Starting Thread {0}", i)
    11. ThreadPool.QueueUserWorkItem(AddressOf Work, i)
    12. Next
    14. While True
    15. SyncLock lock
    16. If threadCount = 0 Then Exit While
    17. End SyncLock
    18. Console.WriteLine("Waiting for {0} remaining threads", threadCount)
    19. are.WaitOne()
    20. End While
    22. Console.WriteLine("All threads ended")
    23. End Sub
    25. Sub Work(state As Object)
    26. 'Do work
    27. Thread.Sleep(2000)
    29. SyncLock lock
    30. threadCount -= 1
    31. Console.WriteLine("Release from Thread {0}", CInt(state))
    32. are.Set()
    33. End SyncLock
    34. End Sub

    Die Bedingung in Zeile 16 prüft die Anzahl der Threads, die gestartet wurden. Falls das an verschiedenen Stellen passiert, muss die Zählvariable VOR dem Aufruf an QueueUserWorkItem() in SyncLock erhöht werden. Diese Vorgehensweise ist aber nicht empfehlenswert - die Anzahl sollte im Voraus bekannt sein.

    Edit: In deinem Fall ist threadCount = meinStack.Count. Wenn dieser Wert jedoch nicht fest ist, hast du ein erstklassiges Producer-Consumer-Problem, vgl. en.wikipedia.org/wiki/Producer-consumer_problem
    Gruß
    hal2000

    Dieser Beitrag wurde bereits 2 mal editiert, zuletzt von „hal2000“ ()

    @hal2000null
    interessante Lösung. Was mich gerade etwas verunsichert ist das du das Producer Consumer Problem ansprichts. Eigentlich ist der Sinn meines Programms besteht darin das Dokumente (bzw. deren Pfade) auf dem Stack nicht doppelt ausgelesen werden. Demnach ist der Stack die geteilte Resscource die Gesperrt ist. Den Threadpool nutze ich um so viel Prozesse wie möglich und so wenig wie nötig zu beanspruchen.

    Meine Idee Stammt aus dem Open Book von der Galilio Press:
    openbook.galileocomputing.de/v…585f3701860b60859e305def1


    Ressourcenbegrenzung mit Semaphore
    Wenn eine Ressource nicht exklusiv gesperrt werden soll, sondern nur ihre
    Nutzung nicht unbeschränkt sein soll, kommt die Klasse Semaphore (Signalmast) ins Spiel. In einem internen Zähler
    hält ein solches Objekt die Anzahl Threads fest, die es beanspruchen.


    Ich dachte das man an diesen Zähler herankommt und ihn abfragen kann.

    @Higlav
    Ja soetwas meinte ich. Nur bekomme ich das mit dem Semaphore nicht hin.

    StGo schrieb:

    [...] Producer Consumer Problem ansprichts. Eigentlich ist der Sinn meines Programms besteht darin das Dokumente (bzw. deren Pfade) auf dem Stack nicht doppelt ausgelesen werden.
    Das eine hat mit dem anderen nichts zutun. Wenn der Producer eine Datei nur einmal zum Verbrauchen in die Queue einstellt, wird sie auch nur einmal bearbeitet. Das PC-Problem zielt auf die Synchronisierung ab, damit a) nicht aus einer leeren Queue gelesen wird (Producer zu langsam / Consumer zu schnell) und b) die Queue nicht überläuft (Producer zu schnell / Consumer zu langsam). Die Daten in der Queue sind unerheblich - in der Theorie werden die nur als "Schlüssel" bezeichnet (einfach eine ID auf ein beliebiges Objekt).

    StGo schrieb:

    Ich dachte das man an diesen Zähler herankommt und ihn abfragen kann.
    Kann man nicht, und das ist Absicht. Der interne Zähler speichert nur die Anzahl an Anfragen, die vom Semaphor angefordert wurden. Darüber wird ermittelt, wann das Semaphor blockiert.
    Gruß
    hal2000

    @ErfinderDesRades

    Dein Link ist interessant. Leider kann ich nicht behaupten das ich den so ohne weiteres komplett verstehe. Mich würde aber interessieren wie ich mein Grundproblem besser lösen kann wenn meine Lösung veraltet ist.

    Kurze Beschreibung:
    1. Ich lese ein verzeichnis aus um speichere die Dateipfade in einem Stack.
    2. Ich starte mehrere Threads (Threadpool) und lasse sie per wechselseitigem Ausschluss auf den Stack zugreifen wo sie einen Pfad auslesen und das Dokument öffnen. Damit Kollisionen vermieden werden sperre ich den Stack und lasse immer nur einen Thread den Stack auslesen
    3. Wenn alle Informationen ausgelesen sind un din einem Array gespeichert sind trage ich diese Informationen in einem Neuen Dokument zusammen.

    Ich finde aber auch keine passende Literatur zum Nachschlagen. Meine Idee stammt aus einer Vorlesung zur parallel Programmierung.

    Danke

    Gruß

    StGo schrieb:

    Ich starte mehrere Threads (Threadpool) und lasse sie per wechselseitigem Ausschluss auf den Stack zugreifen wo sie einen Pfad auslesen und das Dokument öffnen.
    Das ist glaub zwecklos.
    Die Festplatte hat nur einen Lesekopf, und es gibt ühaupt keinen Gewinn, wenn mehrere Threads gleichzeitig den ansteuern.

    @StGo
    Der Link von EDR bedeutet in Kurzform (ist übrigens tatsächlich die elegantere Lösung):

    VB.NET-Quellcode

    1. Sub StartMyThreads()
    2. Dim threadCount As Int32 = 5
    3. Dim asyncResults As New List(Of IAsyncResult)
    5. 'Zeiger auf Work erstellen
    6. Dim callDelegate As New Action(AddressOf Work)
    8. 'mehrfach asynchron aufrufen (BeginInvoke verwendet den Threadpool)
    9. For i = 0 To threadCount
    10. asyncResults.Add(callDelegate.BeginInvoke(Nothing, Nothing))
    11. Next
    13. 'EndInvoke blockiert nur, wenn der Thread in Work noch nicht fertig ist
    14. For Each ar In asyncResults
    15. callDelegate.EndInvoke(ar)
    16. Next
    18. End Sub
    20. Sub Work()
    21. 'DoWork
    22. End Sub

    Falls Work Parameter hat, musst du einen passenden Action(Of T...)-Delegaten verwenden. Wenn es noch einen Rückgabewert gibt, benötigst du stattdessen Func(Of T). Die Rückgabewerte kannst du bei EndInvoke() einsammeln, wie es im Link beschrieben wird.
    Gruß
    hal2000


    Das ist glaub zwecklos.
    Die Festplatte hat nur einen Lesekopf, und es gibt ühaupt keinen Gewinn, wenn mehrere Threads gleichzeitig den ansteuern.


    Auch wenn ich diese tatsache nicht bedacht habe ist es nur ein Prototyp den ich schreibe. Der später die Files aus verschiedenen Verzeichnissen auf verschiedenen Servern ausliest und eine Übersicht schreibt. Das Threading soll später die Ladezeiten sinnvoll nutzen.

    @hal2000
    Der Code sieht gut aus. Morgenfrüh nach dem kaffee setzte ich mich damit auseinander.

    Danke dür die Antworten.

    StGo schrieb:

    ist es nur ein Prototyp den ich schreibe.
    Auch den sollte man anständig proggen.
    Falls Du diesen Code kopierst, achte auf die C&P-Bremse.
    Jede einzelne Zeile Deines Programms, die Du nicht explizit getestet hast, ist falsch :!:
    Ein guter .NET-Snippetkonverter (der ist verfügbar).
    Programmierfragen über PN / Konversation werden ignoriert!

Facebook
Telefonspende