Access List elements in multiple threads and report progress

Question

Inside my application I have list of persons from my database. For every person I must call 5 (for now) services to search for some informations. If service returns info I' adding it to that person (list of orders for specific person)
Because services work independent I thought I could try to run them parallel. I've created my code as so:

using System;
using System.Collections.Generic;
using System.Threading;

namespace Testy
{
    internal class Program
    {
        internal class Person
        {
            public int Id { get; set; }
            public string Name { get; set; }
            public List<string> Orders { get; private set; }

            public Person()
            {
                // thanks for tip @juharr
                Orders = new List<string>();
            }

            public void AddOrder(string order)
            {
                lock (Orders) //access across threads
                {
                    Orders.Add(order);
                }
            }
        }

        internal class Service
        {
            public int Id { get; private set; }

            public Service(int id)
            {
                Id = id;
            }

            //I get error when I use IList instead of List
            public void Search(ref List<Person> list) 
            {
                foreach (Person p in list)
                {
                    lock (p) //should I lock Person here? and like this???
                    {
                        Search(p);
                    }
                }
            }
            private void Search(Person p)
            {
                Thread.Sleep(50);
                p.AddOrder(string.Format("test order from {0,2}",
                                      Thread.CurrentThread.ManagedThreadId));
                Thread.Sleep(100);
            }
        }

        private static void Main()
        {
            //here I load my services from external dll's
            var services = new List<Service>();
            for (int i = 1; i <= 5; i++)
            {
                services.Add(new Service(i));
            }

            //sample data load from db    
            var persons = new List<Person>();

            for (int i = 1; i <= 10; i++)
            {
                persons.Add(
                    new Person {Id = i, 
                    Name = string.Format("Test {0}", i)});
            }

            Console.WriteLine("Number of services: {0}", services.Count);
            Console.WriteLine("Number of persons: {0}", persons.Count);

            ManualResetEvent resetEvent = new ManualResetEvent(false);
            int toProcess = services.Count;

            foreach (Service service in services)
            {
                new Thread(() =>
                    {
                        service.Search(ref persons);
                        if (Interlocked.Decrement(ref toProcess) == 0)
                            resetEvent.Set();
                    }
                    ).Start();
            }

            // Wait for workers.
            resetEvent.WaitOne();

            foreach (Person p in persons)
            {
                Console.WriteLine("{0,2} Person name: {1}",p.Id,p.Name);
                if (null != p.Orders)
                {
                    Console.WriteLine("    Orders:");
                    foreach (string order in p.Orders)
                    {
                        Console.WriteLine("    Order: {0}", order);
                    }
                }
                else
                {
                    Console.WriteLine("    No orders!");
                }
            }
            Console.ReadLine();
        }
    }
}

I have 2 problems with my code:

When I run my app I should get list of 10 persons and for every person 5 orders, but from time to time (ones for 3-5 runs) for first person I get only 4 orders. How I can prevent such behaviour?
solved! thanks to @juharr
How can I report progress from my threads? What I would like to get is one Function from my Program class that will be called every time order is added from service - I need that to show some kind of progress for every report.
I was trying solution described here: https://stackoverflow.com/a/3874184/965722, but I'm wondering if there is an easier way. Ideally I would like to add delegate to Service class and place all Thread code there.
How should I add event and delegate to Service class and how to subscribe to it in Main method?

I'm using .NET 3.5

I've added this code to be able to get progress reports:

using System;
using System.Collections.Generic;
using System.Threading;

namespace Testy
{
    internal class Program
    {
        public class ServiceEventArgs : EventArgs
        {
            public ServiceEventArgs(int sId, int progress)
            {
                SId = sId;
                Progress = progress;
            }

            public int SId { get; private set; }
            public int Progress { get; private set; }
        }

        internal class Person
        {
            private static readonly object ordersLock = new object();

            public int Id { get; set; }
            public string Name { get; set; }
            public List<string> Orders { get; private set; }

            public Person()
            {
                Orders = new List<string>();
            }

            public void AddOrder(string order)
            {
                lock (ordersLock) //access across threads
                {
                    Orders.Add(order);
                }
            }
        }

        internal class Service
        {
            public event EventHandler<ServiceEventArgs> ReportProgress;

            public int Id { get; private set; }
            public string Name { get; private set; }

            private int counter;

            public Service(int id, string name)
            {
                Id = id;
                Name = name;
            }

            public void Search(List<Person> list) //I get error when I use IList instead of List
            {
                counter = 0;
                foreach (Person p in list)
                {
                    counter++;
                    Search(p);
                    Thread.Sleep(3000);
                }
            }

            private void Search(Person p)
            {
                p.AddOrder(string.Format("Order from {0,2}", Thread.CurrentThread.ManagedThreadId));

                EventHandler<ServiceEventArgs> handler = ReportProgress;
                if (handler != null)
                {
                    var e = new ServiceEventArgs(Id, counter);
                    handler(this, e);
                }
            }
        }

        private static void Main()
        {
            const int count = 5;
            var services = new List<Service>();
            for (int i = 1; i <= count; i++)
            {
                services.Add(new Service(i, "Service " + i));
            }

            var persons = new List<Person>();

            for (int i = 1; i <= 10; i++)
            {
                persons.Add(new Person {Id = i, Name = string.Format("Test {0}", i)});
            }

            Console.WriteLine("Number of services: {0}", services.Count);
            Console.WriteLine("Number of persons: {0}", persons.Count);
            Console.WriteLine("Press ENTER to start...");
            Console.ReadLine();

            ManualResetEvent resetEvent = new ManualResetEvent(false);
            int toProcess = services.Count;

            foreach (Service service in services)
            {
                new Thread(() =>
                    {
                        service.ReportProgress += service_ReportProgress;
                        service.Search(persons);
                        if (Interlocked.Decrement(ref toProcess) == 0)
                            resetEvent.Set();
                    }
                    ).Start();
            }

            // Wait for workers.
            resetEvent.WaitOne();

            foreach (Person p in persons)
            {
                if (p.Orders.Count != count)
                    Console.WriteLine("{0,2} Person name: {1}, orders: {2}", p.Id, p.Name, p.Orders.Count);
            }
            Console.WriteLine("END :)");
            Console.ReadLine();
        }

        private static void service_ReportProgress(object sender, ServiceEventArgs e)
        {
            Console.CursorLeft = 0;
            Console.CursorTop = e.SId;
            Console.WriteLine("Id: {0,2}, Name: {1,2} - Progress: {2,2}", e.SId, ((Service) sender).Name, e.Progress);
        }
    }
}

I've added custom EventArgs, event for Service class. In this configuration I should have 5 services running, but only 3 of them report progress.
I imagined that if I have 5 services I should have 5 events (5 lines showing progress).
This is probably because of threads, but I have no ideas how to solve this.

Sample output now looks like this:

Number of services: 5
Number of persons: 10
Press ENTER to start...
Id:  3, Name: Service 3 - Progress: 10
Id:  4, Name: Service 4 - Progress: 10
Id:  5, Name: Service 5 - Progress: 19
END :)

It should look like this:

Number of services: 5
Number of persons: 10
Press ENTER to start...
Id:  1, Name: Service 1 - Progress: 10
Id:  2, Name: Service 2 - Progress: 10
Id:  3, Name: Service 3 - Progress: 10
Id:  4, Name: Service 4 - Progress: 10
Id:  5, Name: Service 5 - Progress: 10
END :)

Last edit
I've moved all my thread creation to separate class ServiceManager now my code looks like so:

using System;
using System.Collections.Generic;
using System.Threading;

namespace Testy
{
    internal class Program
    {
        public class ServiceEventArgs : EventArgs
        {
            public ServiceEventArgs(int sId, int progress)
            {
                SId = sId;
                Progress = progress;
            }

            public int SId { get; private set; } // service id
            public int Progress { get; private set; }
        }

        internal class Person
        {
            private static readonly object ordersLock = new object();

            public int Id { get; set; }
            public string Name { get; set; }
            public List<string> Orders { get; private set; }

            public Person()
            {
                Orders = new List<string>();
            }

            public void AddOrder(string order)
            {
                lock (ordersLock) //access across threads
                {
                    Orders.Add(order);
                }
            }
        }

        internal class Service
        {
            public event EventHandler<ServiceEventArgs> ReportProgress;

            public int Id { get; private set; }
            public string Name { get; private set; }

            public Service(int id, string name)
            {
                Id = id;
                Name = name;
            }

            public void Search(List<Person> list)
            {
                int counter = 0;
                foreach (Person p in list)
                {
                    counter++;
                    Search(p);
                    var e = new ServiceEventArgs(Id, counter);
                    OnReportProgress(e);
                }
            }

            private void Search(Person p)
            {
                p.AddOrder(string.Format("Order from {0,2}", Thread.CurrentThread.ManagedThreadId));
                Thread.Sleep(50*Id);
            }

            protected virtual void OnReportProgress(ServiceEventArgs e)
            {
                var handler = ReportProgress;
                if (handler != null)
                {
                    handler(this, e);
                }
            }
        }

        internal static class ServiceManager
        {
            private static IList<Service> _services;

            public static IList<Service> Services
            {
                get
                {
                    if (null == _services)
                        Reload();
                    return _services;
                }
            }

            public static void RunAll(List<Person> persons)
            {
                ManualResetEvent resetEvent = new ManualResetEvent(false);
                int toProcess = _services.Count;

                foreach (Service service in _services)
                {
                    var local = service;
                    local.ReportProgress += ServiceReportProgress;
                    new Thread(() =>
                        {
                            local.Search(persons);
                            if (Interlocked.Decrement(ref toProcess) == 0)
                                resetEvent.Set();
                        }
                        ).Start();
                }
                // Wait for workers.
                resetEvent.WaitOne();
            }

            private static readonly object consoleLock = new object();

            private static void ServiceReportProgress(object sender, ServiceEventArgs e)
            {
                lock (consoleLock)
                {
                    Console.CursorTop = 1 + (e.SId - 1)*2;
                    int progress = (100*e.Progress)/100;
                    RenderConsoleProgress(progress, '■', ConsoleColor.Cyan, String.Format("{0} - {1,3}%", ((Service) sender).Name, progress));
                }
            }

            private static void ConsoleMessage(string message)
            {
                Console.CursorLeft = 0;
                int maxCharacterWidth = Console.WindowWidth - 1;
                if (message.Length > maxCharacterWidth)
                {
                    message = message.Substring(0, maxCharacterWidth - 3) + "...";
                }
                message = message + new string(' ', maxCharacterWidth - message.Length);
                Console.Write(message);
            }

            private static void RenderConsoleProgress(int percentage, char progressBarCharacter,
                                                      ConsoleColor color, string message)
            {
                ConsoleColor originalColor = Console.ForegroundColor;
                Console.ForegroundColor = color;
                Console.CursorLeft = 0;
                int width = Console.WindowWidth - 1;
                var newWidth = (int) ((width*percentage)/100d);
                string progBar = new string(progressBarCharacter, newWidth) + new string(' ', width - newWidth);
                Console.Write(progBar);
                if (!String.IsNullOrEmpty(message))
                {
                    Console.CursorTop++;
                    ConsoleMessage(message);
                    Console.CursorTop--;
                }
                Console.ForegroundColor = originalColor;
            }

            private static void Reload()
            {
                if (null == _services)
                    _services = new List<Service>();
                else
                    _services.Clear();

                for (int i = 1; i <= 5; i++)
                {
                    _services.Add(new Service(i, "Service " + i));
                }
            }
        }

        private static void Main()
        {
            var services = ServiceManager.Services;
            int count = services.Count;

            var persons = new List<Person>();

            for (int i = 1; i <= 100; i++)
            {
                persons.Add(new Person {Id = i, Name = string.Format("Test {0}", i)});
            }

            Console.WriteLine("Services: {0}, Persons: {1}", services.Count, persons.Count);
            Console.WriteLine("Press ENTER to start...");
            Console.ReadLine();
            Console.Clear();
            Console.CursorVisible = false;

            ServiceManager.RunAll(persons);

            foreach (Person p in persons)
            {
                if (p.Orders.Count != count)
                    Console.WriteLine("{0,2} Person name: {1}, orders: {2}", p.Id, p.Name, p.Orders.Count);
            }
            Console.CursorTop = 12;
            Console.CursorLeft = 0;
            Console.WriteLine("END :)");
            Console.CursorVisible = true;
            Console.ReadLine();
        }
    }
}

Your null check and creation of the Order list should be inside the lock, or better yet inside a Person constructor. — juharr, Jan 28 '13 at 13:16
@juharr - thanks for that tip :) I'll modify and check my code right away! — Misiu, Jan 28 '13 at 13:22

juharr · Accepted Answer · 2013-01-28T16:12:23.417

Basically you have a race condition with the create of the Orders. Imagine the following execution of two threads.

Thread 1 checks if Orders is null and it is.
Thread 2 checks if Orders is null and it is.
Thread 1 sets Orders to a new list.
Thread 1 gets the lock.
Thread 1 adds to the Orders list.
Thread 2 sets Order to a new list. (you just lost what Thread 1 added)

You need to include the creation of the Orders inside the lock.

public void AddOrder(string order)
{
    lock (Orders) //access across threads
    {
        if (null == Orders)
            Orders = new List<string>();
        Orders.Add(order);
    }
}

Or you really should create the Order list in a Person constructor

public Person()
{
    Orders = new List<Order>();
}

Also you should really create a separate object for locking.

private object ordersLock = new object();


public void AddOrder(string order)
{
    lock (ordersLock) //access across threads
    {
        Orders.Add(order);
    }
}

EDIT:

In your foreach where you create the threads you need to create a local copy of the service to use inside the lambda expression. This is because the foreach will update the service variable and the thread can end up capturing the wrong variable. So something like this.

foreach (Service service in services)
{
    Service local = service;
    local.ReportProgress += service_ReportProgress;
    new Thread(() =>
        {
            local.Search(persons);
            if (Interlocked.Decrement(ref toProcess) == 0)
                resetEvent.Set();
        }
    ).Start();
}

Note the subscription doesn't need to be inside the thread.

Alternatively you could move the creation of the thread inside the Search method of your Service class.

Additionally you might want to create a OnReportProgress method in the Service class like so:

protected virtual void OnReportProgress(ServiceEventArgs e)
{
    EventHandler<ServiceEventArgs> handler = ReportProgress;
    if (handler != null)
    {
        handler(this, e);
    }
}

Then call that inside your Search method. Personally I'd call it in the public Search method and make the counter a local variable as well to allow for reuse of the Service object on another list.

Finally you will need an additional lock in the event handler when writing to the console to make sure one thread doesn't change the cursor position before another one writes it's output.

private static object consoleLock = new object();

private static void service_ReportProgress(object sender, ServiceEventArgs e)
{
    lock (consoleLock)
    {
        Console.CursorLeft = 0;
        Console.CursorTop = e.SId;
        Console.WriteLine("Id: {0}, Name: {1} - Progress: {2}", e.SId, ((Service)sender).Name, e.Progress);
    }
}

Also you might want to use Console.Clear() in the following locaiton:

...
Console.WriteLine("Number of services: {0}", services.Count);
Console.WriteLine("Number of persons: {0}", persons.Count);
Console.WriteLine("Press ENTER to start...");
Console.Clear();
Console.ReadLine();
...

And you'll need to update the cursor position before your write out your end statement.

Console.CursorTop = 6;
Console.WriteLine("END :)");

thanks for tips :) I've changed my `person` class, but I'm wondering about that locker. How will I be able to use it in my services if they will be in separate dll's? should I create one object in my main application and pass it using ref to services? — Misiu, Jan 28 '13 at 13:30
The lock object would be inside your `Person` class. So each `Person` would have a lock object that is `private` for use in your `AddOrder` method. — juharr, Jan 28 '13 at 13:32
So I don't need another lock inside `Search` method in `Service` class? Please see my edited `Service` class code in my question — Misiu, Jan 28 '13 at 13:34
And can `ordersLock` be readonly? like this: `private static readonly object ordersLock = new object();`? I get this hint from intellisence. — Misiu, Jan 28 '13 at 13:36
No you don't need the lock inside the `Search` method because you are not modifying the `Person` list. However a lock there would solve the problem because it limits one `Service` to update a `Person` at a time. Typically you'd want the locking at the lowest level though. And yes the lock object can be `readonly`. — juharr, Jan 28 '13 at 13:40
Thanks :) I'll leave lock inside Person class. Could You advice me something about second question? Right now I start all threads from my main class, but I think it would be better to write one class to put all code responsible for threads there or put it inside `Service` class. But how should I rise events from threads (I would like to show some kind of progress inside my application for each service)? — Misiu, Jan 28 '13 at 13:45
You could could create an event in the Service class then subscribe to it from the Main before you start searching. Then in the Search method call the event with the info you need in a custom event argument class. Here's a good article about writing events http://msdn.microsoft.com/en-us/library/w369ty8x(v=vs.80).aspx — juharr, Jan 28 '13 at 14:03
I've tried solution with events, but I get weird behaviour - not all services are reporting progress. I've added code to my question at the bottom. — Misiu, Jan 28 '13 at 14:54
AWESOME!!! I'll modify my code right away and try it. Sorry for all those silly questions, but I want to create as best code as possible at the beginning to avoid rewriting whole thing later. — Misiu, Jan 28 '13 at 19:44
I've managed to get some things running, but I'm stuck when trying to move creation of thread to Service class. I've created separate class `ServiceManager` that will handle service loading/reloading maybe I could move thread creation logic there? Could You please look at last edit of my question? Can I improve it more? I'll add interface to Service class to improve it more and BaseService class with all methods that will be the same for all services, so in future adding new service will be as easy as possible- just override private Search in Service. — Misiu, Jan 29 '13 at 08:29

score 0 · Answer 2 · answered Jan 28 '13 at 13:27

0

This might not totally answer your question (but I think you might have a race condition), when you start dealing with threads you need to implement proper synchronization when updating objects from different threads. You should make sure only one thread is able to update an instance of the person class at any given time. The p.AddOrder( should have mutex that ensures only one thread is updating the Person object.

answered Jan 28 '13 at 13:27

TYY

2,702
1
13
14

can't I use `lock` instead of `mutex`? – Misiu Jan 28 '13 at 13:31

Access List elements in multiple threads and report progress

2 Answers2