Thread question

[Steve B.]

Hi,

In continuation of my previous post, I'm trying to build an application that
use a determined number of worker threads in order to parrallelize actions.

I've a queue of actions, each actions have an execute method and in this
execute method can enqueue new items.

Here is a simpliefied version of my classes :

using System;
using System.Collections.Generic;
using System.Text;
using System.Threading;
namespace ConsoleApplication1
{
class Program
{
static void Main(string[] args)
{
ThreadDispatcher<Action> td = new ThreadDispatcher<Action>();
td.Dispatch(
delegate(Action a)
{
a.Execute();
},
10
);
for (int i = 0; i < 1000; i++)
{
td.Append(new Action(i));
Thread.Sleep(i);
}
Console.ReadLine();
}
}

public class Action
{
private static Random rnd = new Random();
private int m_value;
public int Value
{
get { return m_value; }
}
public Action(int value)
{
this.m_value = value;
}
internal void Execute()
{
Thread.Sleep(rnd.Next(1000));
Console.WriteLine(m_value.ToString("0000") + "Executed from tread " +
Thread.CurrentThread.Name);
}
public override string ToString()
{
return m_value.ToString();
}
}
public class ThreadDispatcher<T>
{
private List<T> m_items = new List<T>();
public void Append(T item)
{
//Console.WriteLine(item.ToString());
lock (m_items)
{
m_items.Add(item);
if (m_items.Count == 1)
{
Monitor.Pulse(m_items);
}
}
}
public void Dispatch(DispatchDelegate<T> executeMethod, int numberOfThreads)
{
for (int i = 0; i < numberOfThreads; i++)
{
Action<DispatchDelegate<T>> a = new Action<DispatchDelegate<T>>(
delegate(DispatchDelegate<T> delegatedExecuteMethod)
{
InternalExecute(delegatedExecuteMethod, i);
}
);
a.BeginInvoke(executeMethod, null, null);
}
}
private void InternalExecute(
ThreadDispatcher<T>.DispatchDelegate<T> executeMethod,
int threadNumber
)
{
Thread.CurrentThread.Name = "Thread Nø" + threadNumber.ToString("00");
while (true)
{
T item;
lock (m_items)
{
if (m_items.Count == 0)
{
Console.WriteLine("Wainting in Thread Nø" + threadNumber.ToString("00"));
Monitor.Wait(m_items);
}
item = m_items[0];
m_items.RemoveAt(0);
}
executeMethod(item);
}
}
public delegate void DispatchDelegate<T>(T item);
}


When I run this program, only the 10th thread is executing actions...

So what have I to do to have several threads executing actions ?
How do I have to properly implement Stop() method ?

Thanks in advance,
Steve

 

[Jon Skeet [C# MVP]]

When I run this program, only the 10th thread is executing actions...

So what have I to do to have several threads executing actions ?

You already have different threads executing methods, but the "i"
variable in your Dispatch method is a captured variable, so when the
delegate executes, it all ways calls InternalExecute with 10 as the
second parameter.

If you change the loop to the following, you'll see the difference:
for (int i = 0; i < numberOfThreads; i++)
{
int x = i;
Action<DispatchDelegate<T>> a = new Action<DispatchDelegate<T>>(
delegate(DispatchDelegate<T> delegatedExecuteMethod)
{
InternalExecute(delegatedExecuteMethod, x);
}
);
a.BeginInvoke(executeMethod, null, null);
}

How do I have to properly implement Stop() method ?

See http://pobox.com/~skeet/csharp/threads/shutdown.shtml

 

[Steve B.]

Thanks, the x = i solved the problem.
However, I'm having another question.

How can I "detect" the whole process is finished ?
In fact, process is finished when no action is in the queue and no action is
executed and then can add new actions.

thanks,
Steve

 

[Jon Skeet [C# MVP]]

Thanks, the x = i solved the problem.
However, I'm having another question.

How can I "detect" the whole process is finished ?
In fact, process is finished when no action is in the queue and no action is
executed and then can add new actions.

It's not entirely clear what you mean, but one idea I've used before in
producer/consumer queues is to have a special entry which means "I've
finished as the producer". Add it as many times as you have consumers,
and make each consumer understand that when it receives that entry, it
should terminate.