Cách sử dụng lại các chủ đề trong .NET 3.5

Question

Tôi có một chương trình con xử lý các khối thông tin lớn. Để tận dụng toàn bộ CPU, nó chia công việc thành các luồng riêng biệt. Sau khi tất cả các chủ đề đã hoàn thành, nó kết thúc. Tôi đọc rằng việc tạo và hủy các chủ đề sử dụng rất nhiều chi phí, vì vậy tôi đã thử sử dụng threadpool, nhưng điều đó thực sự chạy chậm hơn so với việc tạo chủ đề của riêng tôi. Làm thế nào tôi có thể tạo chủ đề của riêng mình khi chương trình chạy và sau đó tiếp tục sử dụng lại chúng? Tôi đã thấy một số người nói rằng nó không thể được thực hiện, nhưng threadpool hiện nó vì vậy nó phải có thể, phải không?

Đây là một phần của mã mà ra mắt chủ đề mới/sử dụng threadpool:

//initialization for threads 
Thread[] AltThread = null; 
if (NumThreads > 1) 
    AltThread = new Thread[pub.NumThreads - 1]; 

do 
{ 
    if (NumThreads > 1) 
    { //split the matrix up into NumThreads number of even-sized blocks and execute on separate threads 
     int ThreadWidth = DataWidth/NumThreads; 
     if (UseThreadPool) //use threadpool threads 
     { 
      for (int i = 0; i < NumThreads - 1; i++) 
      { 
       ThreadPool.QueueUserWorkItem(ComputePartialDataOnThread, 
        new object[] { AltEngine[i], ThreadWidth * (i + 1), ThreadWidth * (i + 2) }); 
      } 
      //get number of threads available after queue 
      System.Threading.Thread.Sleep(0); 
      int StartThreads, empty, EndThreads; 
      ThreadPool.GetAvailableThreads(out StartThreads, out empty); 
      ComputePartialData(ThisEngine, 0, ThreadWidth); 

      //wait for all threads to finish 
      do 
      { 
       ThreadPool.GetAvailableThreads(out EndThreads, out empty); 
       System.Threading.Thread.Sleep(1); 
      } while (StartThreads - EndThreads > 0); 
     } 
     else //create new threads each time (can we reuse these?) 
     { 
      for (int i = 0; i < NumThreads - 1; i++) 
      { 
       AltThread[i] = new Thread(ComputePartialDataOnThread); 
       AltThread[i].Start(new object[] { AltEngine[i], ThreadWidth * (i + 1), ThreadWidth * (i + 2) }); 
      } 
      ComputePartialData(ThisEngine, 0, ThreadWidth); 

      //wait for all threads to finish 
      foreach (Thread t in AltThread) 
       t.Join(1000); 
      foreach (Thread t in AltThread) 
       if (t.IsAlive) t.Abort(); 
     } 
    } 
} 

ComputePartialDataOnThread đơn giản unpackages thông tin và kêu gọi ComputePartialData. Dữ liệu sẽ được xử lý được chia sẻ giữa các luồng (chúng không cố gắng đọc/ghi cùng một vị trí). AltEngine [] là một công cụ tính toán riêng biệt cho mỗi luồng.

Thao tác này chạy khoảng 10-20% khi sử dụng dải chỉ.

Answer 1

này nghe có vẻ giống như một yêu cầu khá phổ biến mà có thể được giải quyết bằng một hàng đợi đa luồng sản xuất-tiêu dùng. Các chủ đề được giữ 'sống' và được báo hiệu để làm việc khi công việc mới được thêm vào hàng đợi. Tác phẩm được đại diện bởi một đại biểu (trong trường hợp của bạn là ComputePartialDataOnThread) và dữ liệu được truyền cho đại biểu là những gì được xếp hàng đợi (trong trường hợp của bạn là params to ComputePartialDataOnThread). Các tính năng hữu ích là việc thực hiện quản lý các chủ đề công nhân và các thuật toán thực tế là riêng biệt. Đây là hàng đợi p-c:

public class SuperQueue<T> : IDisposable where T : class 
{ 
    readonly object _locker = new object(); 
    readonly List<Thread> _workers; 
    readonly Queue<T> _taskQueue = new Queue<T>(); 
    readonly Action<T> _dequeueAction; 

    /// <summary> 
    /// Initializes a new instance of the <see cref="SuperQueue{T}"/> class. 
    /// </summary> 
    /// <param name="workerCount">The worker count.</param> 
    /// <param name="dequeueAction">The dequeue action.</param> 
    public SuperQueue(int workerCount, Action<T> dequeueAction) 
    { 
     _dequeueAction = dequeueAction; 
     _workers = new List<Thread>(workerCount); 

     // Create and start a separate thread for each worker 
     for (int i = 0; i < workerCount; i++) 
     { 
      Thread t = new Thread(Consume) { IsBackground = true, Name = string.Format("SuperQueue worker {0}",i)}; 
      _workers.Add(t); 
      t.Start(); 
     } 
    } 

    /// <summary> 
    /// Enqueues the task. 
    /// </summary> 
    /// <param name="task">The task.</param> 
    public void EnqueueTask(T task) 
    { 
     lock (_locker) 
     { 
      _taskQueue.Enqueue(task); 
      Monitor.PulseAll(_locker); 
     } 
    } 

    /// <summary> 
    /// Consumes this instance. 
    /// </summary> 
    void Consume() 
    { 
     while (true) 
     { 
      T item; 
      lock (_locker) 
      { 
       while (_taskQueue.Count == 0) Monitor.Wait(_locker); 
       item = _taskQueue.Dequeue(); 
      } 
      if (item == null) return; 

      // run actual method 
      _dequeueAction(item); 
     } 
    } 

    /// <summary> 
    /// Performs application-defined tasks associated with freeing, releasing, or resetting unmanaged resources. 
    /// </summary> 
    public void Dispose() 
    { 
     // Enqueue one null task per worker to make each exit. 
     _workers.ForEach(thread => EnqueueTask(null)); 

     _workers.ForEach(thread => thread.Join()); 

    } 
} 

Như poster trước đó đã nói, có rất nhiều được xây dựng trong các cấu trúc (nhìn vào TPL), trong đó sử dụng ThreadPool, mà bạn có thể muốn xem xét trước khi thực hiện hàng đợi của riêng bạn.

Answer 2

Vì vậy, theo cách thông thường người ta sẽ làm điều này là để có entrypoint của mỗi chủ đề cơ bản làm điều gì đó tương tự như (điều này chỉ là một thuật toán, không phải C# mã, xin lỗi):

1. Kiểm tra để xem nếu bạn có công việc làm
2. làm việc nếu thấy
3. Chờ trên một tín hiệu

ở phía bên kia bất cứ khi nào bạn có làm việc nhiều hơn cho chủ đề của bạn thêm nó vào danh sách các việc phải làm và sau đó chủ đề của bạn trong essen ce đang được tái sử dụng. Điều này khá giống với cách một người sẽ tự mình thực hiện một nhóm luồng (nếu bạn đang trong thời gian chạy, bạn có thể làm một số việc khác để giúp bạn, nhưng nó không phải là một vấn đề lớn).

Answer 3

Dưới đây là một chủ đề nói về điều này rất: A custom thread-pool/queue class.