How to Implement a Custom WeighedBlockingCollection in C#

Introduction to BlockingCollection and Memory Management In C#, the BlockingCollection class is a powerful tool for handling producer/consumer scenarios in a thread-safe way. However, when dealing with large objects that can vary in size significantly—like the ones you mentioned ranging from 10 MB to 700 MB—it's crucial to manage memory effectively to avoid OutOfMemoryExceptions. To achieve this, we need a customized collection that extends the functionality of BlockingCollection by allowing us to limit the total memory used by the items stored in it. Understanding the WeighedBlockingCollection The goal here is to create a custom collection called WeighedBlockingCollection, which will not only mimic the behavior of BlockingCollection but also impose a weight restriction on the total items stored. This collection will maintain the FIFO (First In, First Out) principle, ensuring that items are processed in the order they are added, regardless of size. Key Features of WeighedBlockingCollection Limit on Total Weight: The collection will enforce a maximum total memory weight. Thread Safety: It will be designed to work seamlessly with multiple producers and consumers. Delayed Item Creation: Items will only be created when there is enough space in the collection, using a factory function. Implementation Details Let's delve into the implementation details of WeighedBlockingCollection. Here's how we can construct our class: Step 1: Class Definition We'll start by defining our class with appropriate fields and constructors. using System; using System.Collections.Concurrent; using System.Collections.Generic; using System.Threading; public class WeighedBlockingCollection { private readonly BlockingCollection _collection; private long _maximumTotalWeight; private long _currentTotalWeight; private readonly object _lock = new object(); private bool _isAddingCompleted = false; public WeighedBlockingCollection(long maximumTotalWeight) { if (maximumTotalWeight

May 11, 2025 - 10:01

How to Implement a Custom WeighedBlockingCollection in C#

Introduction to BlockingCollection and Memory Management

In C#, the BlockingCollection class is a powerful tool for handling producer/consumer scenarios in a thread-safe way. However, when dealing with large objects that can vary in size significantly—like the ones you mentioned ranging from 10 MB to 700 MB—it's crucial to manage memory effectively to avoid OutOfMemoryExceptions. To achieve this, we need a customized collection that extends the functionality of BlockingCollection by allowing us to limit the total memory used by the items stored in it.

Understanding the WeighedBlockingCollection

The goal here is to create a custom collection called WeighedBlockingCollection, which will not only mimic the behavior of BlockingCollection but also impose a weight restriction on the total items stored. This collection will maintain the FIFO (First In, First Out) principle, ensuring that items are processed in the order they are added, regardless of size.

Key Features of WeighedBlockingCollection

Limit on Total Weight: The collection will enforce a maximum total memory weight.
Thread Safety: It will be designed to work seamlessly with multiple producers and consumers.
Delayed Item Creation: Items will only be created when there is enough space in the collection, using a factory function.

Implementation Details

Let's delve into the implementation details of WeighedBlockingCollection. Here's how we can construct our class:

Step 1: Class Definition

We'll start by defining our class with appropriate fields and constructors.

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Threading;

public class WeighedBlockingCollection
{
    private readonly BlockingCollection _collection;
    private long _maximumTotalWeight;
    private long _currentTotalWeight;
    private readonly object _lock = new object();
    private bool _isAddingCompleted = false;

    public WeighedBlockingCollection(long maximumTotalWeight)
    {
        if (maximumTotalWeight <= 0)
            throw new ArgumentOutOfRangeException(nameof(maximumTotalWeight), "Weight must be greater than zero.");

        _maximumTotalWeight = maximumTotalWeight;
        _collection = new BlockingCollection();
    }
}

Step 2: Method to Add Items

Next, we’ll implement the Add method. This method will manage memory constraints and handle item creation using a factory function.

    public void Add(long itemWeight, Func itemFactory)
    {
        if (itemWeight <= 0)
            throw new ArgumentException("Item weight must be positive.");
        if (itemWeight > _maximumTotalWeight)
            throw new InvalidOperationException("Item weight exceeds the maximum total weight allowed.");

        T item = null;
        lock (_lock)
        {
            // Wait until there's enough space
            while (_currentTotalWeight + itemWeight > _maximumTotalWeight)
            {
                Monitor.Wait(_lock);
            }
            // Create the item using the factory
            item = itemFactory();
            _currentTotalWeight += itemWeight;
            _collection.Add(item);
        }
        Monitor.PulseAll(_lock);
    }

Step 3: Completing the Addition

To signal that no more items will be added, we implement CompleteAdding.

    public void CompleteAdding()
    {
        _isAddingCompleted = true;
        _collection.CompleteAdding();
    }

Step 4: Consuming Items

We'll also need the method for consumers to get items while maintaining the same behavior as BlockingCollection. This method will return an enumerable of items and ensure thread safety.

    public IEnumerable GetConsumingEnumerable()
    {
        foreach (var item in _collection.GetConsumingEnumerable())
        {
            lock (_lock)
            {
                // Here you would need to remove the item's weight accordingly
                // Assume we have a way to know the weight of item
                // For demonstration purposes let's assume itemWeight is provided somehow
                long itemWeight = GetWeight(item);  // You need to implement this
                _currentTotalWeight -= itemWeight;
            }
            yield return item;
        }
    }

    private long GetWeight(T item)
    {
        // Implement a way to get the weight of the item
        // This will depend on your specific use case
        return 0; // placeholder
    }
}

Conclusion

By implementing the WeighedBlockingCollection as shown, you can effectively manage large objects in a producer/consumer scenario without the risk of exceeding memory limits. This customized collection extends the functionality of BlockingCollection while enforcing weight restrictions and ensuring thread safety for multiple producers and consumers.

Feel free to adapt the GetWeight method based on how you define the weight of your objects. This implementation allows you to maintain efficient memory usage while leveraging the power of C#’s concurrent collections.

Frequently Asked Questions

1. What happens if I try to add an item that exceeds the max weight?

An InvalidOperationException will be thrown to indicate that the item weight exceeds the maximum allowed limit.

2. Is this implementation thread-safe?

Yes, the WeighedBlockingCollection has been designed to support concurrent access from multiple threads safely.

3. Can I change the maximum weight limit after the collection is created?

No, the maximum weight limit is set during construction and cannot be changed dynamically. You would need to create a new instance for a different limit.