CODEWITHSUNDEEP
c#linqplinq
David Pfeffer
David Pfeffer

Reputation: 39823

Parallel Linq - return first result that comes back

I'm using PLINQ to run a function that tests serial ports to determine if they're a GPS device.

Some serial ports immediately are found to be a valid GPS. In this case, I want the first one to complete the test to be the one returned. I don't want to wait for the rest of the results.

Can I do this with PLINQ, or do I have to schedule a batch of tasks and wait for one to return?

Upvotes: 4

Views: 3309

Answers (4)

psaxton
psaxton

Reputation: 1843

To accomplish this entirely with PLINQ in .NET 4.0:

SerialPorts.                        // Your IEnumerable of serial ports
    AsParallel().AsUnordered().     // Run as an unordered parallel query
    Where(IsGps).                   // Matching the predicate IsGps (Func<SerialPort, bool>)
    Take(1).                        // Taking the first match
    FirstOrDefault();               // And unwrap it from the IEnumerable (or null if none are found

The key is to not use an ordered evaluation like First or FirstOrDefault until you have specified that you only care to find one.

Upvotes: 0

bj0
bj0

Reputation: 8213

I have been thinking about this on and off for the past couple days and I can't find a built in PLINQ way to do this in C# 4.0. The accepted answer to this question of using FirstOrDefault does not return a value until the full PLINQ query is complete and still returns the (ordered) first result. The following extreme example shows the behavior:

var cts = new CancellationTokenSource();
var rnd = new ThreadLocal<Random>(() => new Random());

var q = Enumerable.Range(0, 11).Select(x => x).AsParallel()
    .WithCancellation(cts.Token).WithMergeOptions( ParallelMergeOptions.NotBuffered).WithDegreeOfParallelism(10).AsUnordered()
    .Where(i => i % 2 == 0 )
    .Select( i =>
    {
        if( i == 0 )
            Thread.Sleep(3000);
        else
            Thread.Sleep(rnd.Value.Next(50, 100));
        return string.Format("dat {0}", i).Dump();
    });

cts.CancelAfter(5000);

// waits until all results are in, then returns first
q.FirstOrDefault().Dump("result");

I don't see a built-in way to immediately get the first available result, but I was able to come up with two workarounds.

The first creates Tasks to do the work and returns the Task, resulting in a quickly completed PLINQ query. The resulting tasks can be passed to WaitAny to get the first result as soon as it is available:

var cts = new CancellationTokenSource();
var rnd = new ThreadLocal<Random>(() => new Random());

var q = Enumerable.Range(0, 11).Select(x => x).AsParallel()
    .WithCancellation(cts.Token).WithMergeOptions( ParallelMergeOptions.NotBuffered).WithDegreeOfParallelism(10).AsUnordered()
    .Where(i => i % 2 == 0 )
    .Select( i =>
    {
        return Task.Factory.StartNew(() =>
        {
        if( i == 0 )
            Thread.Sleep(3000);
        else
            Thread.Sleep(rnd.Value.Next(50, 100));
        return string.Format("dat {0}", i).Dump();
        });
    });

cts.CancelAfter(5000);

// returns as soon as the tasks are created
var ts = q.ToArray();

// wait till the first task finishes
var idx = Task.WaitAny( ts );
ts[idx].Result.Dump("res");

This is probably a terrible way to do it. Since the actual work of the PLINQ query is just a very fast Task.Factory.StartNew, it's pointless to use PLINQ at all. A simple .Select( i => Task.Factory.StartNew( ... on the IEnumerable is cleaner and probably faster.

The second workaround uses a queue (BlockingCollection) and just inserts results into this queue once they are computed:

var cts = new CancellationTokenSource();
var rnd = new ThreadLocal<Random>(() => new Random());

var q = Enumerable.Range(0, 11).Select(x => x).AsParallel()
    .WithCancellation(cts.Token).WithMergeOptions( ParallelMergeOptions.NotBuffered).WithDegreeOfParallelism(10).AsUnordered()
    .Where(i => i % 2 == 0 )
    .Select( i =>
    {
        if( i == 0 )
            Thread.Sleep(3000);
        else
            Thread.Sleep(rnd.Value.Next(50, 100));
        return string.Format("dat {0}", i).Dump();
    });

cts.CancelAfter(5000);

var qu = new BlockingCollection<string>();

// ForAll blocks until PLINQ query is complete
Task.Factory.StartNew(() => q.ForAll( x => qu.Add(x) ));

// get first result asap
qu.Take().Dump("result");

With this method, the work is done using PLINQ, and the BlockingCollecion's Take() will return the first result as soon as it is inserted by the PLINQ query.

While this produces the desired result, I am not sure it has any advantage over just using the simpler Tasks + WaitAny

Upvotes: 2

David Pfeffer
David Pfeffer

Reputation: 39823

Upon further review, you can apparently just use FirstOrDefault to solve this. PLINQ will not preserve ordering by default, and with an unbuffered query, will return immediately.

http://msdn.microsoft.com/en-us/library/dd460677.aspx

Upvotes: 0

Reed Copsey
Reed Copsey

Reputation: 564333

PLINQ is probably not going to suffice here. While you can use .First, in .NET 4, this will cause it to run sequentially, which defeats the purpose. (Note that this will be improved in .NET 4.5.)

The TPL, however, is most likely the right answer here. You can create a Task<Location> for each serial port, and then use Task.WaitAny to wait on the first successful operation.

This provides a simple way to schedule a bunch of "tasks" and then just use the first result.

Upvotes: 7

Related Questions