score:1

Accepted answer

You can't make your own ParallelQuery<T> based classes unfortunately due to the fact that while ParallelQuery<T> is public it does not have any public constructors.

What you can do is use the existing PLINQ infrastructure to do what you want. All you really are wanting to do is do a Where with a Contains being the predicate... So do that.

public static ParallelQuery<TSource> WhereContains<TSource, TKey>(
    this ParallelQuery<TSource> source,
    IEnumerable<TKey> values,
    Func<TSource, TKey> keySelector)
{
    HashSet<TKey> elements = new HashSet<TKey>(values);

    return source.Where(item => elements.Contains(keySelector(item)));
}

This performs the Where clause in parallel, and (while not documented) Contains is thread safe as long as you are not performing any write operations, and because you are making a local HashSet to perform the lookup you don't need to worry about writes happening.

Here is a example project that prints out to the console what thread and item it is processing, you can see that it is using multiple threads.

class Program
{
    static void Main(string[] args)
    {
        List<int> items = new List<int>(Enumerable.Range(0,100));

        int[] values = {5, 12, 25, 17, 0};

        Console.WriteLine("thread: {0}", Environment.CurrentManagedThreadId);

        var result = items.AsParallel().WhereContains(values, x=>x).ToList();

        Console.WriteLine("Done");
        Console.ReadLine();
    }
}

static class Extensions
{
    public static ParallelQuery<TSource> WhereContains<TSource, TKey>(
        this ParallelQuery<TSource> source,
        IEnumerable<TKey> values,
        Func<TSource, TKey> keySelector)
    {
        HashSet<TKey> elements = new HashSet<TKey>(values);

        return source.Where(item =>
        {
            Console.WriteLine("item:{0} thread: {1}", item, Environment.CurrentManagedThreadId);
            return elements.Contains(keySelector(item));
        });
    }
}

score:1

Could you not just do this?

public static ParallelQuery<TSource> Where<TSource>(
    this ParallelQuery<TSource> source, 
    Func<TSource, bool> predicate)
{
    return
        source
            .SelectMany(x =>
                predicate(x)
                ? new TSource[] { x } 
                : Enumerable.Empty<TSource>());
}

Related Articles