Class to encapsulate and manage multiple background web crawlers

Question

I need to crawl web contents from some websites and then do some processing. Note that this is a small application, so the dataset is relatively small (need to crawl about 30,000 pages every time, once a week). The problem is that I can't start too many threads to crawl the pages at the same time. Otherwise my IP will be recognized unusual and will be blocked.

So, I create a class, called CrawlingService. It's designed to encapsulate these things:

• Start some threads to crawl the web content
• Control the waiting time after a page is crawled (a thread need to "have a rest" after a page is crawled to prevent the app from being blocked by the server)
• Notify other classes that a webpage is crawled
• Automatically retry N times when fail to crawl a page (most are "Timeout" errors)
• When an error occurs (99% is timeout), all threads need to pause for a while. Because the "Timeout" is mostly caused by server busy.

The following is my implementation.

• The main class is CrawlingService which is already mentioned above.
• ITaskRestStrategy.Duration() method is used to return the information about how long a thread need to wait for after a page is crawled.
• The AbstractHttpClient is used to make HTTP requests, not important.
• Let's ignore argument null checking

What I want to know

• Is the multi-threading implementation correct? (Not good at this)
• Can I improve the multi-threading implementation by using a better approach?
• Can I improve the design of these classes?
• Better naming for the classes/methods/variables? (Not quite good at English)

WebResource.cs

/// <summary>
/// Represents a webpage to be crawled.
/// </summary>
public class WebResource
{
    public string Url { get; private set; }

    public Encoding Encoding { get; private set; }

    public string Content { get; set; }

    // Ignore the constructor
}

CrawlingEventArgs.cs

public class CrawlingEventArgs : EventArgs
{
    public WebResource Resource { get; private set; }

    public Exception Exception { get; private set; }

    // Ignore the constructor
}

CrawlingService.cs

public class CrawlingService
{
    static readonly Logger _log = LogManager.GetCurrentClassLogger();

    private ConcurrentQueue<QueueItem> _queue = new ConcurrentQueue<QueueItem>();

    private AbstractHttpClient _httpClient;

    // the strategy for "having rest" after a page is crawled
    private ITaskRestStrategy _itemRestStrategy;
    // the strategy for "having rest" after an error occurs
    private ITaskRestStrategy _errorRestStrategy;

    // this is used to control the "pause" of all threads
    private ManualResetEventSlim _continueEvent;
    private readonly object _startLock = new object();

    // this is used for the Wait method called by client code
    private int _totalWorkingThreads;
    private ManualResetEventSlim _exitEvent;

    public event EventHandler<CralwingEventArgs> ItemSucceeded;

    public event EventHandler<CralwingEventArgs> ItemFailed;

    public bool IsRunning { get; private set; }

    /// <summary>
    /// The maximum number of threads can be run in parallel.
    /// </summary>
    public int MaxDegreeOfParallelism { get; private set; }

    /// <summary>
    /// The maximum allowed retries when failed to crawl a page.
    /// </summary>
    public int MaxRetriesForEachItem { get; private set; }

    public CralwingService()
        : this(RandomTaskRestStrategy.FromSeconds(1, 2, 3, 4, 5), new SimpleTaskRestStrategy(TimeSpan.FromSeconds(15)))
    {
    }

    public CralwingService(
        ITaskRestStrategy itemRestStrategy,
        ITaskRestStrategy errorRestStrategy)
        : this(itemRestStrategy, errorRestStrategy, 2, 3, new DefaultHttpClient())
    {
    }

    public CralwingService(
        ITaskRestStrategy itemRestStrategy,
        ITaskRestStrategy errorRestStrategy,
        int maxDegreeOfParallelism,
        int maxRetriesForEachItem,
        AbstractHttpClient httpClient)
    {
        _httpClient = httpClient;
        _itemRestStrategy = itemRestStrategy;
        _errorRestStrategy = errorRestStrategy;
        MaxDegreeOfParallelism = maxDegreeOfParallelism;
        MaxRetriesForEachItem = maxRetriesForEachItem;
    }

    /// <summary>
    /// Add webpages to the crawling queue.
    /// </summary>
    public void Add(IEnumerable<WebResource> resouces)
    {
        lock (_startLock)
        {
            if (IsRunning)
                throw new InvalidOperationException("Cannot add new items after the service is started.");

            foreach (var info in resouces)
            {
                _queue.Enqueue(new QueueItem { ResourceInfo = info });
            }
        }
    }

    /// <summary>
    /// Starts the background crawling threads.
    /// </summary>
    public bool Start()
    {
        if (IsRunning)
        {
            return false;
        }

        lock (_startLock)
        {
            if (IsRunning)
            {
                return false;
            }

            IsRunning = true;

            _continueEvent = new ManualResetEventSlim(true);
            _totalWorkingThreads = MaxDegreeOfParallelism;

            for (var i = 0; i < MaxDegreeOfParallelism; i++)
            {
                StartProcessingNextItem();
            }
        }

        return true;
    }

    /// <summary>
    /// Wait until all pages are crawled.
    /// </summary>
    public void Wait()
    {
        if (_exitEvent == null)
        {
            _exitEvent = new ManualResetEventSlim();
        }

        _exitEvent.Wait();

        OnExit();
    }

    private void StartProcessingNextItem()
    {
        _continueEvent.Wait();

        QueueItem item = null;

        if (_queue.TryDequeue(out item))
        {
            var resource = item.ResourceInfo;
            var task = _httpClient.GetAsync(resource.Url, resource.Encoding)
                                    .ContinueWith(t =>
                                    {
                                        if (t.Exception != null)
                                        {
                                            // If error occuors, all threads need to pause and "have a rest"
                                            _continueEvent.Reset();

                                            // Add the failed item back to the crawling  queue if it's still retryable
                                            if (item.TotalRetries < MaxRetriesForEachItem)
                                            {
                                                item.TotalRetries++;
                                                _queue.Enqueue(item);
                                            }
                                            else
                                            {
                                                OnItemFailed(resource, t.Exception);
                                            }

                                            Thread.Sleep(_errorRestStrategy.Duration());

                                            // Notify all threads to continue after the "rest"
                                            _continueEvent.Set();
                                        }
                                        else
                                        {
                                            OnItemSucceeded(resource, t.Result);

                                            // Have a rest also after an item is processed successfully
                                            Thread.Sleep(_itemRestStrategy.Duration());
                                        }

                                        // Finish processing one page, so now can start processing next page
                                        StartProcessingNextItem();
                                    });
        }
        else
        {
            var totalWorkingThreads = Interlocked.Decrement(ref _totalWorkingThreads);

            if (totalWorkingThreads == 0)
            {
                // Now I'm the only thread still executing
                if (_exitEvent != null)
                {
                    _exitEvent.Set();
                }
                else
                {
                    OnExit();
                }
            }
        }
    }

    private void OnItemFailed(WebResource resource, Exception exception)
    {
        if (ItemFailed != null)
        {
            SafeExecuteAsync(() =>
            {
                ItemFailed(this, new CralwingEventArgs(resource, exception));
            }, "Error invoking ItemFailed event handlers.");
        }
    }

    private void OnItemSucceeded(WebResource resource, string content)
    {
        if (ItemSucceeded != null)
        {
            resource.Content = content;

            SafeExecuteAsync(() =>
            {
                ItemSucceeded(this, new CralwingEventArgs(resource));
            }, "Error invoking ItemSucceeded event handlers.");
        }
    }

    private void SafeExecuteAsync(Action action, string errorMessage)
    {
        Task.Factory.StartNew(() =>
        {
            try
            {
                action();
            }
            catch (Exception ex)
            {
                _log.ErrorException(UserReference.System(), ex, errorMessage);
            }
        });
    }

    // Cleanup resources
    private void OnExit()
    {
        _continueEvent.Dispose();
        _continueEvent = null;

        if (_exitEvent != null)
        {
            _exitEvent.Dispose();
            _exitEvent = null;
        }

        IsRunning = false;
    }

    class QueueItem
    {
        public WebResource ResourceInfo = null;

        public int TotalRetries = 0;
    }
}

public interface ITaskRestStrategy
{
    TimeSpan Duration();
}

public class SimpleTaskRestStrategy : ITaskRestStrategy
{
    private TimeSpan _duration;

    public SimpleTaskRestStrategy(TimeSpan duration)
    {
        _duration = duration;
    }

    public TimeSpan Duration()
    {
        return _duration;
    }
}

public class RandomTaskRestStrategy : ITaskRestStrategy
{
    static readonly Random _random = new Random();

    private TimeSpan[] _durations;

    public RandomTaskRestStrategy(TimeSpan[] durations)
    {
        Require.NotNull(durations, "durations");
        Require.That(durations.Length > 0, "'durations' must have one item at least.");

        _durations = durations;
    }

    public static RandomTaskRestStrategy FromSeconds(params int[] seconds)
    {
        var durations = new TimeSpan[seconds.Length];

        for (var i = 0; i < durations.Length; i++)
        {
            durations[i] = TimeSpan.FromSeconds(seconds[i]);
        }

        return new RandomTaskRestStrategy(durations);
    }

    public TimeSpan Duration()
    {
        var index = _random.Next(0, _durations.Length);
        return _durations[index];
    }
}

Found Issues

1. The "Wait" method call might block forever:

   service.Start();
   // Short tasks might all complete here.
   // In this case, calling Wait will block forever
   service.Wait();
   

2. The _startLock is useless

3. System.Random is not thread safe

Answer 1

Generally, very well written code. I just have a few suggestions:

1. In the Start method, I understand why, but I don't like that there are two calls to check if the service is running. It feels kind of redundant, and I don't think you'll get a speed increase by short-circuiting the lock.

2. I would rather create and throw customized Exception classes rather than using the built-in one.

   throw new ServiceIsAlreadyRunningException()
   

   is less confusing than:

   InvalidOperationException("Cannot add new items after the service is started.")
   

3. I would change the IsRunning property to use an enum. This will allow the addition of more states in the future (Starting, Running, Stopped, ShuttingDown, ...)
4. I'm not sure what version of .Net you are using, but look into the await keyword. I found it makes my code flow so much better and makes it easier to follow when using threads.

5. In your constructors, check the injected classes for null. This will save you problems when trying to use them later:

   public CralwingService(
       ITaskRestStrategy itemRestStrategy,
       ITaskRestStrategy errorRestStrategy,
       int maxDegreeOfParallelism,
       int maxRetriesForEachItem,
       AbstractHttpClient httpClient)
   {
   
       if (itemRestStrategy == null) throw new ArgumentNullException("itemRestStrategy");
       // same for all other reference types injected
   
       // rest of constructor code
   }
   

Overall, this was easy code to read, and very well written, which I like. The suggestions I've pointed out will take it from good code to excellent code, in my opinion.