Làm cách nào để lấy một mục từ bất kỳ hai BlockingCollections nào có ưu tiên cho bộ sưu tập đầu tiên?

Question

Tôi có hai đối tượng BlockingCollection<T>, collection1 và collection2. Tôi muốn tiêu thụ các mặt hàng từ các bộ sưu tập này ưu tiên cho các mục trong số collection1. Tức là, nếu cả hai bộ sưu tập đều có các mục, tôi muốn lấy các mục từ số collection1 trước. Nếu không có mục nào trong số chúng có mục, tôi muốn chờ một mục có sẵn.

Tôi có đoạn mã sau:

public static T Take<T>(
    BlockingCollection<T> collection1, 
    BlockingCollection<T> collection2) where T:class 
{ 
    if (collection1.TryTake(out var item1)) 
    { 
     return item1; 
    } 

    T item2; 

    try 
    { 
     BlockingCollection<T>.TakeFromAny(
      new[] { collection1, collection2 }, 
      out item2); 
    } 
    catch (ArgumentException) 
    { 
     return null; 
    } 

    return item2; 
} 

Mã này được dự kiến ​​sẽ trở null khi CompleteAdding được gọi trên cả hai bộ sưu tập và cả hai đều là những sản phẩm nào.

vấn đề chính của tôi với mã này là tài liệu cho các phương pháp TakeFromAny xác định rằng TakeFromAny sẽ ném một ArgumentException nếu CompleteAdding được gọi vào "bộ sưu tập":

ArgumentException

Đối số bộ sưu tập là một mảng có độ dài 0 hoặc chứa một phần tử null hoặc CompleteAdding() đã được gọi trên bộ sưu tập.

Nó có ném nếu CompleteAdding được gọi trên bộ sưu tập nào không? hoặc cả hai bộ sưu tập?

Điều gì xảy ra nếu CompleteAdding được gọi và bộ sưu tập vẫn có một số mục, nó có bị ném không?

Tôi cần một cách đáng tin cậy để thực hiện việc này.

Trong mã này tôi đang cố gắng để có được từ collection1 đầu tiên bởi vì tài liệu của TakeFromAny không đưa ra bất cứ đảm bảo về trật tự bộ sưu tập mà từ đó để có những item nếu hai bộ sưu tập có các mục.

Điều này cũng có nghĩa là nếu cả hai bộ sưu tập đều trống và sau đó chúng nhận được các mục cùng một lúc sau đó, thì tôi có thể nhận được một mặt hàng từ collection2 trước.

EDIT:

Lý do tôi thêm các mục vào hai bộ sưu tập (và không chỉ đơn giản là một bộ sưu tập duy nhất) là bộ sưu tập đầu tiên không có một bộ sưu tập trên bên bị ràng buộc, và lần thứ hai không.

Thêm các chi tiết cho những ai quan tâm đến lý do tại sao tôi cần điều này:

Tôi đang sử dụng này trong một dự án mã nguồn mở được gọi là ProceduralDataflow. Xem tại đây để biết thêm chi tiết https://github.com/ymassad/ProceduralDataflow

Mỗi nút xử lý trong hệ thống dataflow có hai bộ sưu tập, một bộ sưu tập sẽ chứa các mục lần đầu tiên (vì vậy tôi cần làm chậm nhà sản xuất nếu cần) và bộ sưu tập khác sẽ chứa các mục đến lần thứ hai (hoặc thứ ba, ..) (do kết quả của vòng lặp trong luồng dữ liệu).

Lý do tại sao một bộ sưu tập không có giới hạn trên là tôi không muốn có deadlocks do kết quả của vòng lặp trong luồng dữ liệu.

Answer 1

Đầu tiên, câu trả lời ngắn cho câu hỏi cụ thể của bạn.

Nó có ném nếu CompleteAdding được gọi trên bộ sưu tập nào không? hoặc cả hai bộ sưu tập?

Cả hai (tất cả) - nhưng chỉ khi không có phần tử có sẵn trong bất kỳ bộ sưu tập nào.

Điều gì sẽ xảy ra nếu CompleteAdding được gọi và bộ sưu tập vẫn có một số mục?

Không. Nếu có phần tử trong bộ sưu tập, phần tử đó sẽ bị xóa khỏi bộ sưu tập và trả lại người gọi.

Kết luận

Rõ ràng tài liệu không rõ ràng. Phần

hoặc CompleteAdding() đã kêu gọi các bộ sưu tập

nên đã được xây dựng cách khác nhau - một cái gì đó giống như

hoặc không có yếu tố có sẵn trong bất kỳ của bộ sưu tập và CompleteAdding() đã được gọi trên tất cả bộ sưu tập

Lý

Vâng, tôi biết dựa vào việc thực hiện không phải là một thói quen tốt, nhưng khi tài liệu hướng dẫn không rõ ràng, việc thực hiện chỉ là nguồn đáng tin cậy và chính thức tôi có thể nghĩ đến. Vì vậy, lấy reference source, cả hai TakeFromAny và TryTakeFromAny gọi phương thức riêng TryTakeFromAnyCore. Nó bắt đầu với những điều sau đây:

ValidateCollectionsArray(collections, false); 

false đây là một cuộc tranh luận bool gọi isAddOperation và được sử dụng bên trong ValidateCollectionsArray như sau:

if (isAddOperation && collections[i].IsAddingCompleted) 
{ 
    throw new ArgumentException(
     SR.GetString(SR.BlockingCollection_CantAddAnyWhenCompleted), "collections"); 
} 

đó là một trong những nơi có thể ném ArgumentException cho các bộ sưu tập với CompleteAdding() được gọi là. Và như chúng ta có thể thấy, đây không phải là trường hợp (câu hỏi # 1).

Sau đó, việc thực hiện tiếp tục với những điều sau "con đường nhanh":

//try the fast path first 
for (int i = 0; i < collections.Length; i++) 
{ 
    // Check if the collection is not completed, and potentially has at least one element by checking the semaphore count 
    if (!collections[i].IsCompleted && collections[i].m_occupiedNodes.CurrentCount > 0 && collections[i].TryTake(out item)) 
     return i; 
} 

Điều này chứng tỏ câu trả lời cho câu hỏi # 2.

Cuối cùng, nếu không có yếu tố có sẵn trong bất kỳ bộ sưu tập, việc thực hiện lấy "đường chậm" bằng cách gọi một phương pháp riêng TryTakeFromAnyCoreSlow, với những nhận xét sau đây là lời giải thích cần thiết về hành vi thực hiện:

//Loop until one of these conditions is met: 
// 1- The operation is succeeded 
// 2- The timeout expired for try* versions 
// 3- The external token is cancelled, throw 
// 4- The operation is TryTake and all collections are marked as completed, return false 
// 5- The operation is Take and all collection are marked as completed, throw 

Câu trả lời cho cả hai câu hỏi của chúng tôi là trong trường hợp # 1 và trường hợp số 5 (lưu ý từ tất cả). Btw, nó cũng cho thấy sự khác biệt duy nhất giữa TakeFromAny và TryTakeFromAny - trường hợp # 4 so với # 5, tức là throw so với return -1.