Using Task.Run to start a continuously running background job with a BlockingCollection

Question

I am using Task.Run to start a continuous background processing job on a BlockingCollection. It works but I don't think it's the ideal way to structure the code.

This issue concerns the line: public async Task ProcessQueue(). I get a warning:

This async method lacks 'await' operators and will run synchronously.

If I remove async there is an error:

'MessageProcessor.ProcessQueue()': not all code paths return a value

This is reasonable because the compiler doesn't see that getConsumingEnumerable blocks. Should I use the async version and ignore this warning or restructure my code in a different way?

Note: Imagine I have 100s of these MessageProcessor's, I don't think creating a separate thread for each of them would be an efficient use of memory hence my use of Task.Run.

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

public class Message
{
    public int id { get; set; }
    public int value { get; set; }
}

public class MessageProcessor
{
    private CancellationToken token;
    private BlockingCollection<Message> messageQueue;
    private List<Message> processedMessageList;

    public MessageProcessor(CancellationToken token)
    {
        this.token = token;
        messageQueue = new BlockingCollection<Message>(new ConcurrentQueue<Message>());
        processedMessageList = new List<Message>(10);
    }

    public void Add(Message m)
    {
        if (!messageQueue.IsAddingCompleted)
        {
            messageQueue.Add(m);
        }
    }

    public void CompleteAdding()
    {
        messageQueue.CompleteAdding();
    }

    public async Task ProcessQueue()
    {
        foreach (Message m in messageQueue.GetConsumingEnumerable())
        {
            Console.WriteLine($"Processing. id: {m.id}, value: {m.value}");
            processedMessageList.Add(m);

        }
    }
}

class Program
{
    static void Main(string[] args)
    {
        Random random = new Random();
        CancellationTokenSource tokenSource = new CancellationTokenSource();
        CancellationToken token = tokenSource.Token;
        MessageProcessor messageProcessor = new MessageProcessor(token);

        var messageProcessorResult = Task.Run(() => messageProcessor.ProcessQueue());

        for (int i = 0; i < 10; i++)
        {
            Message m = new Message();
            m.id = i;
            m.value = random.Next(-2, 2);

            Console.WriteLine($"Producing. id: {m.id}, value: {m.value}");
            messageProcessor.Add(m);
        }

        messageProcessor.CompleteAdding();
        messageProcessorResult.Wait();
    }
}

@HansPassant async void is a bad practice, what a shame to see this advice by someone with 800k points — Ozkan, Feb 17 '19 at 14:25
Pretty sure he meant without the `async` which is part of what the OP tried — pinkfloydx33, Feb 17 '19 at 20:28
Removing `async Task` and making it `void` returning is the right option here. `Task.CompletedTask` would make sense for example if this was implementing an interface that requires a `Task`, but here the method signature is in our control. — Stuart, Feb 17 '19 at 22:16
@Evk I saw one of your comments about how "BlockingCollection with ConcurrentQueue ... will block when removing" (https://stackoverflow.com/a/41271450/10984827). In that case will my ProcessQueue method block here? Do you recommend an alternative way to structure this code? Thanks — Mike Ekim, Mar 04 '19 at 04:06

score 1 · Answer 1 · answered Feb 17 '19 at 15:57

1

What you need is an action block.

answered Feb 17 '19 at 15:57

Paulo Morgado

14,111
3
31
59

score 0 · Answer 2 · answered Feb 17 '19 at 14:29

0

Remove async and write the following at the end of the ProcessQueue method:

return Task.CompletedTask;

This way we solve the not all code paths return a value error.

answered Feb 17 '19 at 14:29

Ozkan

3,880
9
47
78

1

The compiler already told him to delete the async keyword. Declaring methods to be async when you don't need await is pointless. – Hans Passant Feb 17 '19 at 14:35

Using Task.Run to start a continuously running background job with a BlockingCollection

2 Answers2