Hành vi không mong muốn - Tách dòng các mục trong khi TriggerBatch đang thực thi

Question

Khi bạn tạo một kho chặn có dung lượng giới hạn và gọi triggerBatch trong khi (song song với) đăng một mục mới - đăng mục mới sẽ không thành công trong thời gian thực thi lô kích hoạt.

Lô kích hoạt cuộc gọi (mỗi lần X) được thực hiện để đảm bảo dữ liệu không bị trì hoãn quá lâu trong khối, trong trường hợp luồng dữ liệu đến bị tạm dừng hoặc bị chậm lại.

Đoạn mã sau sẽ xuất ra một số sự kiện "đăng sự cố". Ví dụ:

public static void Main(string[] args) 
    { 
     var batchBlock = new BatchBlock<int>(10, new GroupingDataflowBlockOptions() { BoundedCapacity = 10000000 }); 
     var actionBlock = new ActionBlock<int[]>(x => ProcessBatch(x), new ExecutionDataflowBlockOptions() { MaxDegreeOfParallelism = 1 }); 
     batchBlock.LinkTo(actionBlock); 

     var producerTask = Task.Factory.StartNew(() => 
     { 
      //Post 10K Items 
      for (int i = 0; i < 10000; i++) 
      { 
       var postResult = batchBlock.Post(i); 
       if (!postResult) 
        Console.WriteLine("Failed to Post"); 
      } 
     }); 

     var triggerBatchTask = Task.Factory.StartNew(() => 
      {      
       //Trigger Batch.. 
       for (int i = 0; i < 1000000; i++) 
        batchBlock.TriggerBatch(); 
      }); 

     producerTask.Wait(); 
     triggerBatchTask.Wait(); 
    } 

    public static void ProcessBatch(int[] batch) 
    { 
     Console.WriteLine("{0} - {1}", batch.First(), batch.Last()); 
    } 

* Lưu ý rằng trường hợp này chỉ có thể lặp lại khi lô khóa được Bounded.

Tôi có thiếu gì đó hay là vấn đề với batchBlock?

Answer 1

BatchBlock không thực sự từ chối mục, nó cố gắng trì hoãn mục đó. Ngoại trừ trường hợp của Post(), việc trì hoãn không phải là một lựa chọn. Một cách đơn giản để khắc phục điều này là sử dụng await batchBlock.SendAsync(i) thay vì batchBlock.Post(i) (điều này cũng có nghĩa là bạn cần thay đổi Task.Factory.StartNew(() => thành Task.Run(async() =>).

Tại sao điều này lại xảy ra? Theo the source code, nếu BatchBlock bị chặn, TriggerBatch() được xử lý không đồng bộ và trong khi được xử lý, sẽ không có mục mới nào được chấp nhận.

Trong mọi trường hợp, bạn không nên mong đợi true trên khối bị chặn, nếu khối đầy, Post() cũng sẽ trả về false.