How to Re-Invoke the work when one task completed within .WithAny

Question

Hi All,

I've a certain code as mentioned below

            int[] intCol = new int[] { 2, 5, 6, 8 };
            Random r = new Random();

            var actPtr = new ActionBlock<int>
                (intInpVal =>
                {
                    int intNew = r.Next(1, 9);
                    Console.WriteLine(string.Format("Waiting {1} milliseconds for .. {0} .. to complete ", intInpVal.ToString(), intNew));
                    Thread.Sleep(intNew * 100);
                    Console.WriteLine(intInpVal);
                });
            foreach (int intPtr in intCol)
            {
                actPtr.Post(intPtr);
            }

You get an output as sequenctial. But now that I need to convert this into parallel execution rather as sequential.

Note a point here towards the "Thread.Sleep(intNew * 100)" is just to keep one thread sleep for a long time so that i have to switch to multiple tasks as parallel

Thanks for reading

---

Every thing is perfect as long as you share

Answer 1

Hi Chakravarthy,

What is it that you want running in parallel, the processing of all of the intInpVal values being fed into the ActionBlock?  If so, when you construct the ActionBlock, you can give it an ExecutionDataflowBlockOptions that has a MaxDegreeOfParallelism property set to a value higher than 1.  Of course, when you do so, that delegate will end up potentially running in parallel with itself, which means it needs to be thread-safe.  So, for example, you'd need to protect usage of 'r', which would then be potentially accessed from multiple threads concurrently.

I hope that helps.

Answer 2

Stephen,

Thanks for  your quick reply. I might have not conveyed the requirement properly.

Let me rephrase my requirement. I have a long running method that has to be executed for multiple values. You could see the below code that is currently working. But we are not aware of the following points

1. Which task is long running
2. While a task is under process, the main is “DoParallelWork()” invoked again. Thus, we don’t have a hook /
   handle for the already running thread, so that we can cancel the task and start with the new values
3. As we are using the  new Thread mechanism, we are interested to execute this either async or parallel way.

        public void DoParallelWork()
        {
            DataTable objTasks = GetDataFromDB();
            foreach (DataRow objDataRow in objTasks.Select("CurVal > 0"))
            {
                StartLongRunningThread(objDataRow.DId, objDataRow.ShortCode, objDataRow.NotificationGroup);
                //Stop for 1 minute
                Thread.Sleep(60 * 1000);
            }
        }

        public void StartLongRunningThread(int intUniqueID, Int32 intShortCode, string strNotifyGroup)
        {
            // long running code
            Thread _notificationThread = new Thread(GenerateNotifications);
            _notificationThread.Start();
        }


        public void GenerateNotifications()
        {
            // The actual data intensive and CPU intensive work which is taking long time
        }

What do you recommend?

---

Every thing is perfect as long as you share

Answer 3

Hi Chakravarthy-

I'm still not clear on what you're trying to achieve.  Are you saying that you want to:

- Launch a piece of work asynchronously, and this work may take some time to run

- If another piece of work comes in while that first piece is processing, you want to cancel the first piece, and then once it stops (due to the cancellation), start processing the new piece of work.

Is that it?  If so, you could do something like:

private Task m_currentTask = Task.FromResult(0);
private CancellationTokenSource m_canceler = new CancellationTokenSource();
internal void DoWork()
{
    m_canceler.Cancel();
    var cts = new CancellationTokenSource();
    m_canceler = cts;
    m_currentTask = m_currentTask.ContinueWith(delegate
    {
        ... // do your processing here, monitoring cts.Token as necessary to know when cancellation was requested
    }, cts.Token, TaskContinuationOptions.None, TaskScheduler.Default);
}

Answer 4

Hi Stephen,

Thanks for taking time in writing back to me.

To give you a clue about what I'm planning to do is, mix the async mechanism with parallel programming and get the result from the async method to store in the calling context.

While planning for this, I'm not ready for the wait state of the caller thread. Thus, the caller thread would invoke the work as a paralle activity or concurrent activity.

Hope I'm clear..

---

Every thing is perfect as long as you share

Answer 5

Hi Chakravarthy-

I still don't understand.  What do you mean by mixing async with parallel programming?  What do you mean by storing the result into the calling context?  What do you mean by not being ready for the wait state of the caller thread?

Are we still talking about TPL Dataflow, or is this something else entirely?  It might help to start from scratch here and describe in explicit detail what you're trying to accomplish.