[C#] Wiele wątków (nawet do 200, dynamicznie) - jak to opanować?

Tytułem wprowadzenia:

Odczytuję sobie strumień danych. Wyciągam z niego kilka informacji, niestety pojawił się problem. Podczas odczytu może zdarzyć się, że strumień jest uszkodzony, ale jedynym sposobem, żeby to określić jest oczekiwanie na timeout paczki danych ze strumienia. Takich "paczek" może być nawet i 200, a optymalny timeout to 5s. Jeśli mam wyjątkowego pecha, to paczek uszkodzonych może być kilka i w efekcie odczyt przeciąga się, bo czytana jest paczka za paczką (pętla for). Dlatego w mojej głowie pojawiła się kwestia zaciągnięcia do pracy wątków - każdy odczyt jednej paczki byłby jednym wątkiem i wtedy (jeśli dobrze to obmyśliłem) byłbym w stanie czytać wszystko jednocześnie i cały proces w najgorszym przypadku trwałby 5s, jeśli akurat w strumieniu trafi się cokolwiek uszkodzonego, niezależnie ile tego będzie.

Problem:

Nie wiem ile tych wątków może być, ich ilość jest dynamiczna w zależności od badanego strumienia, może być 1, a może i 200. Dowiaduję się tego dopiero, gdy odczytam inny fragment strumienia (co robię w kodzie nieco wcześniej).

• muszę stworzyć dynamicznie wątek, wykonać w nim odczyt paczki i zaraz po wykonaniu tej operacji go zabić
• cała procedura musi czekać na zamknięcie wszystkich wątków

Jak się za to zabrać? Czego powinienem użyć i jakich przykładów szukać w sieci? Czy Parallel.For rozwiązałby problem?

A do tego nie służy czasem ThreadPool?

Męczę temat dalej, poczytałem sporo zarówno o Parallel.ForEach, jak i ThreadPool i stanęło na próbie implementacji Parallel.ForEach (automatyzuje wiele), jednak utknąłem. Ktoś w tym siedzi może?

Wydaje mi się, że problem u mnie sprowadza się do wymuszenia wybrania elementu z ArrayList przetworzonego do IEnumerable (???). Spróbuję w skrócie przedstawić zarys kodu (nie wkleję 1:1, bo jest tam sporo rzeczy zaciemniających sytuację).

Pierwotnie było tak:

foreach(własny_typ_danych zmienna_lokalna in lista_array)

Zmieniłem to na:

Parallel.ForEach<własny_typ_danych>(lista_array.Cast<własny_typ_danych>(), (zmienna_lokalna) =>
{
// wykonuję operacje, m.in. wykorzystując dodatkowe wartości udostępniane przez własny_typ_danych, tj. zmienna_lokalna.cośtam, rzutując na object nie miałbym dostępu do tych danych
}

Wygląda jednak na to, że zmienna_lokalna jest pusta (w pewnym momencie program zwraca ArgumentOutOfRangeException, po informacjach z okna błędu wskazuje na linię, która ma wykorzystać wynik z operacji wykonywanej na zmienna_lokalna.cośtam; tu mała uwaga - z przyczyn technicznych nie mogę debugować - program działa z hardware, którego na kompie z Visual Studio nie mam, nie odpali się, biegam z tym na inny PC, odpalam i patrzę jakimi błędami sypie). Nie wiem, czy dobrze w tym fragmencie wpisałem dwa razy własny_typ_danych, ale jest mi to potrzebne, żeby zmienna_lokalna posiadała dodatkowe dane wewnątrz (zmienna_lokalna.cośtam).

Albo stosuję złe podejście? Zmienna_lokalna powinna mieć przy każdej iteracji inną wartość, kolejną z lista_array, tu w MSDN opisują bardziej rozbudowaną wersję Parallel.ForEach: http://msdn.microsoft.com/en-us/library/dd997393.aspx

Ostatecznie może akurat tego, co chcę zrobić nie da się wepchnąć w wątki - da się to jakoś zbadać, czy dana operacja może być wykonywana w wątkach?

Mam nadzieję, że to dobrze opisałem. Ma ktoś może jakąś koncepcję? Czy mogę coś dopisać co mogłoby pomóc w namierzeniu problemu?

ArgumentOutOfRangeException nie oznacza, że obiekt jest pusty. Nie wiem co to znaczy "w pewnym momencie". Na podstawie informacji z wyrzuconego wyjątku możesz się dowiedzieć, która to linijka i wkleić tutaj więcej kodu.
Czemu używasz ArrayList zamiast List<T>?
Jedna rzecz mnie nurtuje - chcesz jeden strumień czytać w wielu wątkach? Nie wiem czy tak się da... A jeśli da, to zapewne z jakąś synchronizacją, więc dodatkowym narzutem, czyli i spowolnieniem.

Skoro jest to strumień, to jak chcesz czytać z niego równolegle? Przecież nadejście następnej paczki nie może nastąpić w przypadku gdy czekasz jeszcze na dane poprzedniej.

To bardziej złożony temat - dokładnie jest to strumień danych z satelity, jest on podzielony na tablice i zaraz po uzyskaniu sygnału (co następuje dużo wcześniej) filtr od Microsoftu buforuje sobie dane. Od strony teorii wydaje mi się, że jednoczesny odczyt elementów powinien być możliwy - do danych da się dobrać w innej kolejności niż były oryginalnie przesyłane. GetPSITable to funkcja bazująca na GetTable z DirectShow, wyciągająca zgromadzone już informacje.

Obawiam się, że bez debuggera tego nie pociągnę, bo nawet nie wiem, czy zmienne są zapełnione, jeśli tak, to czym. Tak to zabawa po omacku :/ Nie wiem, czy coś Wam powyższy kod mówi. W każdym razie dzięki za dotychczasową pomoc :)

EDIT: Nie będę ciapać dodatkowo, bo nie ma sensu, tylko dopiszę rozwiązanie znalezione przy kolejnej wolnej chwili przeznaczonej na programowanie ;)

Otóż trzeba wcześniej zadeklarować Capacity dla listy, bo inaczej przy podejściu równoległym zabraknie miejsca, jest to związane z tego co wyczytałem w sieci z właściwościami list - ich pojemność jest zwiększana w razie potrzeby sukcesywnie wraz ze wzrostem ilości itemów i coś tu zawodzi przy dodawaniu równolegle.