Parallele Datenverarbeitung ohne "lock"

Hallo, ich habe hin und wieder große Datenmengen die evtl. mit Zusatzdaten zusammengefügt werden müssen. Bei der parallelen Verarbeitung ist da oft auch das Problem, dass die Daten wegen DataRace gegeneinander mit "lock" gesichert werden muss. Das kann aber auch manchmal bremsen.

Ich hatte nun gerade einen einfachen Fall, den ich nun mit einer anderen Sichtweise lösen wollte. Leider bekomme ich eine Exception, bei der parallelen Verarbeitung der Task's.

Hier mal meine Idee (mal ganz egal ob das auch noch anders lösbar ist, die Frage ist wie bekomme ich die Exception weg, da weiss ich nicht weiter).

Ich lese zwei Datentabellen ein: Tabelle "Quelldaten", Tabelle "Infodaten".

Die "Quelldaten" müssen mit den "Infodaten" abgeglichen werden. In einer normalen Schleife dauert das sehr lange. Nun die Idee die Quelldaten Prozessorkernabhängig aufzuteilen und je einen Task erstellen, dann die Tasks parallel starten.

Soweit funktioniert das auch schon, aber bei

 Task.WaitAll(lstTsk.ToArray(), new TimeSpan(0,0,30));

Eventuell kann mir jemand einen Tipp geben.

bekomme ich eine Exception.

Hie rmal der Code:

			//// INFO: 23.02.2017: Hier die Testroutine für parallele Ausführung testen
            // ############################################ TEST ###################################################################

            //// Daten vom SQL-Server einlesen
            DataTable tblPersonenDaten = GetDataSql("Personendaten");
            DataTable tblPersonenZusatzdaten = GetDataSql("PersonenZusatzdaten"); 

			//// List<> erstellen, wo die Datensatzaufteilung Prozessorkernanzahlig aufgeteilt wird 
            List <DataTable> lstTblPersonendaten = new List<DataTable>();
            int anzProzkerne = Environment.ProcessorCount;
            for (int i = 0; i < anzProzkerne; i++)
                lstTblPersonendaten.Add(tblPersonendaten.Clone());

			//// Die Tabellendatensätze dann aufsplitten
            int zaehler = anzProzkerne;
            for (int m = 0; m < tblPersonendaten.Rows.Count; m++)
            {
                if (zaehler == anzProzkerne)
                    zaehler = 0;
                lstTblPersonendaten[zaehler].Rows.Add(lstTblPersonendaten[zaehler].NewRow().ItemArray = tblPersonendaten.Rows[m].ItemArray);
                zaehler++;
            }

            
			//// Nun die parallelen Tasks erstellen und die Daten übergeben
            //// Da die Daten vereinzelt sind, muss kein "lock" verwendet werden, so die Idee
            List<Task> lstTsk = new List<Task>();
            for (int n = 0; n < anzProzkerne; n++)
            {
                lstTsk.Add(Task.Factory.StartNew(() => DatensaetzeAbgleichen(lstTblPersonendaten[n], tblPersonenZusatzdaten.Copy())));
            }
            try
            {
				//// Aufruf der parallelen Abarbeitung (das ist die Zeile 1942 in meinem Code)
				Task.WaitAll(lstTsk.ToArray(), new TimeSpan(0,0,30));
            }
            catch (Exception ex)
            {
                Allgemein.ExceptionMeldungAusgeben(ex, "Fehler bei TPL", "Fehler");
            }

            
            // #####################################################################################################################
			
			
			
			
			
			
			
 /* ========================================================================================================================================================================*/

        /// <summary>
        /// Gleicht die Daten der Tabelle "tblSource" mit den von Tabelle "tblInfos" ab.
        /// </summary>
        /// <param name="tblSource">Quelldaten die mit den Infos befüllt werden.</param>
        /// <param name="tblInfos">Tabelle mit den Info-Daten, welche in die entsprechenden Quelldatensatzspalten eingefügt werden.</param>

        private void DatensaetzeAbgleichen(DataTable tblSource, DataTable tblInfos)
        {
            try
            {
				foreach (DataRow rowSource in tblSource.Rows)
                {
                    DataRow rowScheibenInfo = tblInfos.AsEnumerable().Where(x => x["SapArtikel"] != DBNull.Value && x["SapArtikel"].ToString().Trim() == rowSource["SapArtikel"].ToString().Trim() || x["Systemcode"] != DBNull.Value && x["Systemcode"].ToString().Trim() == rowSource["Zeichnungsnummer"].ToString().Trim()).FirstOrDefault();

                    if(rowScheibenInfo != null)
                    {
                        rowSource["ScheibenInfo"] = rowScheibenInfo["ScheibenInfo"];
                        rowSource["SwInfo"] = rowScheibenInfo["SwInfo"];
                        rowSource["ZusatzInfo"] = rowScheibenInfo["ZusatzInfo"];
                    }
                }
            }
            catch (Exception ex)
            {
                Allgemein.ExceptionMeldungAusgeben(ex, "Fehler in DatensatzAbgleichen() ", "Fehler:");
            }
        }

/* ========================================================================================================================================================================*/

InnerException:
{"Index was out of range. Must be non-negative and less than the size of the collection.\r\nParameter name: index"}

Bin im Moment etwas ratlos.

InnerException:
{"Index was out of range. Must be non-negative and less than the size of the collection.\r\nParameter name: index"}

Das ist eine OutOfRange Exception, du wirst da schauen müssen wo das Passiert .

Meine Vermutung (nach dem was du beschreibst) ist, das es hier Passiert.

  List<Task> lstTsk = new List<Task>();
            for (int n = 0; n < anzProzkerne; n++)
            {
                lstTsk.Add(Task.Factory.StartNew(() => DatensaetzeAbgleichen(lstTblPersonendaten[n], tblPersonenZusatzdaten.Copy())));

        }

Hi,

wenn ich mich nicht täusche wird eine Zählervariable "n" gecaptured, weshalb du diesen nicht nachvollziehbaren Fehler hast.

Das ganze nennt sich "Closure" - im Endeffekt führt das bei dir dazu, dass "n" zwar normal hochgezählt wird - die Tasks jedoch jedoch nicht den erwarteten Wert bekommen.

Mein Tipp:
Definiere mal im Loop eine Kopie von n, damit hier nichts mehr gecaptured wird.

LG

Edit: Tipp bezieht sich auf den von Palin genannten Code-Auszug.

Hi,

s. [Artikel] Delegaten, anonyme Methoden, Lambda-Ausdrücke & Co. unter Closures.

Parallel.ForEach könnte auch eine Lösung für Dein Problem sein.
Wenn Quell und InfoDaten beide unveränderlich sind, dann sollte das einfach und elegant sein.
Meistens bringt es aber mehr, daß Du Dir saubere indizierte Datenstrukturen überlegst anstatt mit Schleifen zu arbeiten.