Repeating a task that may fail (in parallel)

Question

Almost like this question here, I'm playing with some methods that may fail, but in my case, the method failing is not a exceptional case; it just means I need to try running it again.

The method I'm writing is inherently stochastic, so running it multiple times yields different results. Some are valid, some are invalid.

Like in the other question, I created a class that represents a successful / unsuccessful run.

public readonly struct Maybe<T> {

    public readonly bool ContainsValue;
    public readonly T Value;

    [MethodImpl(MethodImplOptions.AggressiveInlining)]
    public Maybe(bool containsValue, T value) {
        ContainsValue = containsValue;
        Value = value;
    }

    [MethodImpl(MethodImplOptions.AggressiveInlining)]
    public static Maybe<T> Just(T value) {
        return new Maybe<T>(
            containsValue: true,
            value: value);
    }

    public static Maybe<T> Empty { get; } = new Maybe<T>(
        containsValue: false,
        value: default
        );

    [MethodImpl(MethodImplOptions.AggressiveInlining)]
    public static implicit operator Maybe<T>(T value) => Maybe.Just(value);
}

// This class is used just to make the syntax of Maybe<T> cleaner
public static class Maybe {

    [MethodImpl(MethodImplOptions.AggressiveInlining)]
    public static Maybe<T> Just<T>(T value) => Maybe<T>.Just(value);
}

In the end of my method I can just check if the result is valid and return Maybe.Just(Result) or Maybe.Empty() if it's not.

Review questions about Maybe<T> and Maybe:

• Is this struct design ok? I opted for a struct to reduce allocations. I do not intend to store references to it. Just return it, check it's results and forget about it.
• Is the creation of a non-generic class just to enable a prettier syntax legitimate? I prefer writing Maybe.Just(whatever) than Maybe<LongType>.Just(whatever).

I then noticed that I tend to wrap my "try methods" inside a loop... So I "hoisted" the code to a different class:

public static class ProcessRetrier {

    public static Maybe<T> Run<T>(int maximumRetries, Func<Maybe<T>> func) {
        if (maximumRetries < 0)
            throw new ArgumentOutOfRangeException(nameof(maximumRetries) + " must be >= 0");
        if (func == null)
            throw new ArgumentNullException(nameof(func));

        for (int i = 0; i <= maximumRetries; i++) {
            var result = func();

            if (result.ContainsValue)
                return result;
        }

        return Maybe<T>.Empty;
    }
}

Review question about ProcessRetrier:

• Is this class design legit?
• Do you have any suggestions?

Finally, I decided that it would be cool if I could run the different attempts in different threads, so I created this monstrosity here:

public static Maybe<T> RunParallel<T>(int maximumRetries, Func<Maybe<T>> func) {
    if (maximumRetries < 0)
        throw new ArgumentOutOfRangeException(nameof(maximumRetries) + " must be >= 0");
    if (func == null)
        throw new ArgumentNullException(nameof(func));

    var retries = 0;

    var tasks = new Task<Maybe<T>>[Environment.ProcessorCount];
    var finished = 0;

    for (int i = 0; i < tasks.Length; i++) {
        tasks[i] = Task.Run(() => {
            while (true) {
                if (retries >= maximumRetries || finished > 0)
                    return Maybe<T>.Empty;

                var attempt = func();
                if (attempt.ContainsValue) {
                    Interlocked.Increment(ref finished);
                    return attempt;
                } else {
                    Interlocked.Increment(ref retries);
                }
            }
        });
    }

    Task.WaitAny(tasks);

    foreach (var t in tasks) {
        if (t.Result.ContainsValue)
            return t.Result;
    }

    return Maybe<T>.Empty;
}

And now is the meat of the problem:

• Is this parallel implementation correct?
• When the lambda "swallows" retries and finished, are the different threads seeing the same values? Or each task will try maximumRetries?
• If each thread task is reading / modifying it's own copy of retries, how can I make them all use the same value?

Edit: I tested the "RunParallel" logic with the snippet bellow, it looks like that each threading is running roughly N / number_of_processors attempts. But just because it's works now doesn't mean the implementation is correct; hence the review question.

var retries = 0;
var maximumRetries = 10 * 1000 * 1000;

var tasks = new Task<int>[Environment.ProcessorCount];
var finished = 0;

for (int i = 0; i < tasks.Length; i++) {
    tasks[i] = Task.Run(() => {
        var taskAttempts = 0;
        while (true) {
            if (retries >= maximumRetries || finished > 0)
                return taskAttempts;

            Interlocked.Increment(ref retries);
            taskAttempts += 1;
        }
    });
}

Task.WaitAll(tasks);
var attempts = tasks.Select(t => t.Result).ToArray();
Console.WriteLine(attempts.Sum());

Answer 1

Concurrency

Is this parallel implementation correct?

I would expect a method called RunParallel<T> to use processor affinity (Parallel.For) specially since you take the processor count into account var tasks = new Task<Maybe<T>>[Environment.ProcessorCount];. But instead, you use thread affinity tasks[i] = Task.Run(() => ..

Either refactor RunParallel<T> to actually be parallel, or keep it asynchronous and rename it to something in the likes of Any<T>. You might also want to use the async/await pattern when choosing the latter.

---

Syntax

Is the creation of a non-generic class just to enable a prettier syntax legitimate? I prefer writing Maybe.Just(whatever) than Maybe<LongType>.Just(whatever).

Have you considered an extension method:

public static class Maybe {

    public static Maybe<T> Just<T>(this T value) => Maybe<T>.Just(value);
}

And usage:

int myValue = 5;
var maybe = myValue.Just(); // Maybe<int>();