Перемещение не двоичной древовидной структуры из TwinCat с несколькими потоками в C#

Question

Я пытаюсь оптимизировать алгоритм поиска, который я использую, чтобы найти отмеченные символы в TwinCat 3 через интерфейс ADS. Вопрос не в TwinCat, так что пока не испугайтесь.

Проблемы: Символы не загружаются сразу. Я думаю, что библиотека TwinCatAds использует ленивую загрузку. Символы имеют древовидную структуру не двоичного не сбалансированного дерева.

Решение: Вы можете открыть несколько потоков для ADS. И обрабатывайте потоки в нескольких потоках.

Вопрос в том, что я разделяю первый уровень символов на количество процессорных ядер. Так как дерево не сбалансировано, некоторые из потоков заканчиваются быстрее, чем другие. Из-за этого мне нужно лучшее решение, как разделить работу между потоками.

PS: Я не могу использовать Parallel.ForEach(). Из-за потоков это приводит к тому же или большему количеству времени, что и однопоточное решение.

Мой тестовый код выглядит так, он просто подсчитывает все символы огромного проекта.

using TwinCAT.Ads; 
using System.Threading; 
using System.IO; 
using System.Diagnostics; 
using System.Collections; 


namespace MultipleStreamsTest 
{ 
class Program 
{ 
    static int numberOfThreads = Environment.ProcessorCount; 
    static TcAdsClient client; 
    static TcAdsSymbolInfoLoader symbolLoader; 
    static TcAdsSymbolInfoCollection[] collection = new TcAdsSymbolInfoCollection[numberOfThreads]; 
    static int[] portionResult = new int[numberOfThreads]; 
    static int[] portionStart = new int[numberOfThreads]; 
    static int[] portionStop = new int[numberOfThreads]; 

    static void Connect() 
    { 
     client = new TcAdsClient(); 
     client.Connect(851); 
     Console.WriteLine("Conected "); 
    } 
    static void Main(string[] args) 
    { 
     Connect(); 
     symbolLoader = client.CreateSymbolInfoLoader(); 
     CountAllOneThread(); 
     CountWithMultipleThreads(); 
     Console.ReadKey(); 
    }   
    static public void CountAllOneThread() 
    { 
     Stopwatch stopwatch = new Stopwatch(); 
     int index = 0; 
     stopwatch.Start(); 
     Console.WriteLine("Counting with one thread..."); 
     //Count all symbols 
     foreach (TcAdsSymbolInfo symbol in symbolLoader) 
     {     
      index++; 
     } 
     stopwatch.Stop(); 
     //Output 
     Console.WriteLine("Counted with one thred " + index + " symbols in " + stopwatch.Elapsed); 
    } 
    static public int countRecursive(TcAdsSymbolInfo symbol) 
    { 
     int i = 0; 
     TcAdsSymbolInfo subSymbol = symbol.FirstSubSymbol; 
     while (subSymbol != null) 
     { 
      i = i + countRecursive(subSymbol); 
      subSymbol = subSymbol.NextSymbol; 
      i++; 
     } 
     return i; 
    } 
    static public void countRecursiveMultiThread(object portionNum) 
    { 
     int portionNumAsInt = (int)portionNum; 
     for (int i = portionStart[portionNumAsInt]; i <= portionStop[portionNumAsInt]; i++) 
     { 
       portionResult[portionNumAsInt] += countRecursive(collection[portionNumAsInt][i]);//Collection Teil 
     } 
    } 
    static public void CountWithMultipleThreads() 
    { 
     Stopwatch stopwatch = new Stopwatch(); 
     int sum = 0; 
     stopwatch.Start(); 
     Console.WriteLine("Counting with multiple thread..."); 
     for (int i = 0; i < numberOfThreads; i++) 
     { 
      collection[i] = symbolLoader.GetSymbols(true); 
     } 
     int size = (int)(collection[0].Count/numberOfThreads); 
     int rest = collection[0].Count % numberOfThreads; 
     int m = 0; 
     for (; m < numberOfThreads; m++) 
     { 
      portionStart[m] = m * size; 
      portionStop[m] = portionStart[m] + size - 1; 
     } 
     portionStop[m - 1] += rest; 

     Thread[] threads = new Thread[numberOfThreads]; 
     for (int i = 0; i < numberOfThreads; i++) 
     { 
      threads[i] = new Thread(countRecursiveMultiThread); 
      threads[i].Start(i); 
      Console.WriteLine("Thread #" + threads[i].ManagedThreadId + " started, fieldIndex: " + i); 
     } 
     //Check when threads finishing: 
     int threadsFinished = 0; 
     bool[] threadFinished = new bool[numberOfThreads]; 
     int x = 0; 
     while (true) 
     { 
      if (threads[x].Join(10) && !threadFinished[x]) 
      { 
       Console.WriteLine("Thread #" + threads[x].ManagedThreadId + " finished ~ at: " + stopwatch.Elapsed); 
       threadsFinished++; 
       threadFinished[x] = true;      
      } 
      x++; 
      x = x % numberOfThreads; 
      if (threadsFinished == numberOfThreads) break; 
      Thread.Sleep(50); 
     }    
     foreach (int n in portionResult) 
     { 
      sum += n; 
     } 
     sum += collection[0].Count; 
     stopwatch.Stop(); 
     //Output 
     Console.WriteLine("Counted with multiple threds in Collection " + sum + " symbols " + " in " + stopwatch.Elapsed); 
     for (int i = 0; i < numberOfThreads; i++) 
     { 
      Console.WriteLine("#" + i + ": " + portionResult[i]); 
     } 
    } 
} 
} 

The console output:

Если вы пытаетесь запустить использование кода TwinCat.Ads версии 4.0.17.0 (что я использую). Они сломали что-то в новой версии, доступной с NuGet.

Answer 1

Создайте пул потоков и отслеживайте состояние потоков и состояние холостого хода. В каждой ветке проверяйте, есть ли потоки холостого хода, если есть выделение потока для подвешенной ветви.