2

I am trying to parse data from several websites continuously. I would like this action to be preformed individually in a loop in an asynchronous manner until the program is closed. I am not sure what the structure should be for this kind of logic.

Right now I am following this pattern.

async public void ParseAll(List<Site> SiteList)
{
    List<Task> TaskList = new List<Task>();

    foreach(Site s in SiteList)
    {
        TaskList.Add(s.ParseData);
    }

    await Task.WhenAll(TaskList)
}

The issue is that if I construct a Loop around this method then the sites that are updated first will have to wait until the whole list is finished before the method can run again. Theoretically, what I would like to do is just put each site back on the bottom of the TaskList when it finished its ParseData method but I am not sure if thats possible, or if thats the best way.

5
  • 2
    Do not use async void methods unless they are event handlers. Use async Task for async methods that do not return results. Commented Jun 3, 2014 at 17:10
  • How many sites are there and how often do you want to update? Commented Jun 3, 2014 at 17:11
  • @DanielMann Considering the method is designed to never ever return, getting a Task that will never complete isn't really any more useful than a void method. Commented Jun 3, 2014 at 17:28
  • 2
    AFAIK that's not true as exception behavior is different as async void will go up the call stack if an exception occured in a continuation. Commented Jun 3, 2014 at 17:36
  • 1
    @Servy Linky is correct. Exceptions are handled differently when they occur in async void methods. Hit up your search engine of choice and take a look -- unhandled exceptions in async void methods are rethrown on the thread pool and cause all sorts of nasty behavior. Commented Jun 3, 2014 at 17:46

4 Answers 4

3

Theoretically, what I would like to do is just put each site back on the bottom of the TaskList when it finished its ParseData

Looks like you need to maintain a queue of sites to be processed. Below is my take on this, using SemaphoreSlim. This way you can also limit the number of concurrent tasks to be less than the actual number of sites, or add new sites on-the-fly. A CancellationToken is used to stop the processing from outside. The use of async void is justified here IMO, QueueSiteAsync keeps track of the tasks it starts.

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

namespace AsyncLoop
{
    class Program
    {
        public class Site
        {
            public string Url { get; set; }
            public async Task ParseDataAsync(CancellationToken token)
            {
                // simulate download and parse
                int delay = new Random(Environment.TickCount).Next(100, 1000);
                await Task.Delay(delay, token);
                Console.WriteLine("Processed: #{0}, delay: {1}", this.Url, delay);
            }
        }

        object _lock = new Object();
        HashSet<Task> _pending = new HashSet<Task>(); // sites in progress
        SemaphoreSlim _semaphore;

        async void QueueSiteAsync(Site site, CancellationToken token)
        {
            Func<Task> processSiteAsync = async () =>
            {
                await _semaphore.WaitAsync(token).ConfigureAwait(false);
                try 
                {           
                    await site.ParseDataAsync(token);
                    QueueSiteAsync(site, token);
                }
                finally
                {
                    _semaphore.Release();
                }
            };

            var task = processSiteAsync();
            lock (_lock)
                _pending.Add(task);
            try
            {
                await task;
                lock (_lock)
                    _pending.Remove(task);
            }
            catch
            {
                if (!task.IsCanceled && !task.IsFaulted)
                    throw; // non-task error, re-throw

                // leave the faulted task in the pending list and exit
                // ProcessAllSites will pick it up
            }
        }

        public async Task ProcessAllSites(
            Site[] sites, int maxParallel, CancellationToken token)
        {
            _semaphore = new SemaphoreSlim(Math.Min(sites.Length, maxParallel));

            // start all sites
            foreach (var site in sites)
                QueueSiteAsync(site, token);

            // wait for cancellation
            try
            {
                await Task.Delay(Timeout.Infinite, token);
            }
            catch (OperationCanceledException)
            {
            }

            // wait for pending tasks
            Task[] tasks;
            lock (_lock)
                tasks = _pending.ToArray();
            await Task.WhenAll(tasks);
        }

        // testing
        static void Main(string[] args)
        {
            // cancel processing in 10s
            var cts = new CancellationTokenSource(millisecondsDelay: 10000); 
            var sites = Enumerable.Range(0, count: 10).Select(i => 
                new Site { Url = i.ToString() });
            try
            {
                new Program().ProcessAllSites(
                    sites.ToArray(), 
                    maxParallel: 5, 
                    token: cts.Token).Wait();
            }
            catch (AggregateException ex)
            {
                foreach (var innerEx in ex.InnerExceptions)
                    Console.WriteLine(innerEx.Message);
            }
        }
    }
}

You may also want to separate download and parsing into separate pipelines, check this for more details.

Sign up to request clarification or add additional context in comments.

Comments

1

It's easy enough to create a method to loop continuously and parse a single site over and over again. Once you have that method, you can call it once on each site in the list:

private async void ParseSite(Site s)
{
    while (true)
    {
        await s.ParseData();
    }
}

public void ParseAll(List<Site> siteList)
{
    foreach (var site in siteList)
    {
        ParseSite(site);
    }
}

10 Comments

Is there a way to extend this method to have a concurrency?
@YevgenyYurgenson It's already going to do the work concurrently. No extension is necessary.
@Servy: Doesn't this depend on the SynchronizationContext? AFAIK, when running from a UI Thread the synchronous code (which can take some time while parsing) will run sequentially (although "interlocked")
@Linky Any CPU bound work would need to run sequentially, sure. But the whole idea here is that the work is asynchronous, so while two site's parser can't be checking their loop condition at the same time, one request can be doing its small amount of CPU bound processing while others are waiting asynchronously.
@Linky Yes, because you're making the inherent assumption that the majority of the work is CPU bound work, despite the fact that there is no such indication that it is the case. There is no need to use multiple threads to create parallelism. Doing actions asynchronously, and not waiting on them before starting other actions, creates parallelism..
|
0

If you want to visit the site again as soon as it is complete, you probably want to use Task.WhenAny and integrate your outer loop with your inner loop, something like this (assuming the ParseData function will return the Site it is parsing for):

async public void ParseAll(List<Site> SiteList)
{
    while (true)
    {
        List<Task<Site>> TaskList = new List<Task<Site>>();

        foreach(Site s in SiteList)
        {
            TaskList.Add(s.ParseData());
        }

        await Task.WhenAny(TaskList);
        TaskList = TaskList.Select(t => t.IsCompleted ? t.Result.ParseData() : t).ToList();
    }
}

Comments

-1

Did you tried the PLinq lib?

Plinq allows you to execute linq querys async.

In your case it would look like:

SiteList.AsParallel().ForEach(s => s.ParseData);

1 Comment

He doesn't want to call this on each site once, he wants to have each site continue to loop, parsing the data each time it finishes.

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Start asking to get answers

Find the answer to your question by asking.

Ask question

Explore related questions

See similar questions with these tags.