How to ensure unique local state among threads for email queue processing

Question

THE SCENARIO:

I want to ask this question regarding Parallel.For(or any other multithreading approach in C#.Net). I have to build a MultiThreaded Mailer Windows service that will send mails to all the recipients as fast as it can. I get the serialized rows from the database that contains the email message and SmtpDetails and then deSerialize them in code.

An emails may have 1000 reciepients and so on a Dual Core machine ( development machine) at least 2 threads can run simultaneously. So i use parallel.For in order ro do this. I have read about the LocalInit delegate that runs once for every thread.

THE CODE:

int itemCount = serMailObj.ReceipientList.Count;

                    Parallel.For(0, itemCount, () =>
                    {
                        return new ThreadLocalStateCache()
                        {
                            Receipient = serMailObj.ReceipientList.Dequeue(),
                            mail = serMailObj.Email,
                            SerializableSmtpDetails = serSmtpObj
                        };
                    }
     , doWork, (x) => { });

 private static ThreadLocalStateCache doWork(int instance, ParallelLoopState state, ThreadLocalStateCache threadInstance)
        {
            KeyValuePair<string, string> kvp = threadInstance.Receipient;
            SerializableSmtpDetails serSmtpObj = threadInstance.SerializableSmtpDetails;
            MailMessage email = threadInstance.mail;

                email.To.Add(new MailAddress(kvp.Key, kvp.Value));

                SmtpClient client = new SmtpClient();
                client.Credentials = new System.Net.NetworkCredential(serSmtpObj.UserName, serSmtpObj.Password);
                client.Host = serSmtpObj.Host;
                client.Port = serSmtpObj.Port;
                client.EnableSsl = serSmtpObj.EnableSSL;

                    try 
                    {           
                        client.Send(email);
                        Console.WriteLine("sending mail....");
                    }
                    catch (Exception)
                    {

                        throw;
                    }


            return null;
        }

public class ThreadLocalStateCache
    {
        public KeyValuePair<string, string> Receipient { get; set; }

        public MailMessage mail { get; set; }

        public SerializableSmtpDetails SerializableSmtpDetails { get; set; }
    }

The above code is pretty straight forward. The localInit delegate constructs a local object foreach thread. and then the doWork tries to process the queue.

THE PROBLEMS:

1. I am getting multiple mails for each recipient. seems as if the email object is being shared among threads.

2. getting failure sending mail sometimes.

Kindly explain as to how i can isolate the mail and smtpclient objects in each thread. and process the queue.

EDIT 1: If the multithreading gurus would help me please tell that is there any way for every thread to have a unique copy of its local variables and not shared ones. Since the MailMessage object is not immutable i cannot create a clone of it also. apart from deseralizing it in each thread(which would ensure a new object is created) is there any magic way to achieve this?

Answer 1

below might be the issue:

serMailObj.ReceipientList.Dequeue()

Try using ConcurrentQueue (.NET 4) or put locks around so that one thread at a time can Dequeue it.

also make to sure Concurrent classes or locks anywhere yo uhave access to shared resource from threads.

A Queue <(Of <(T >)>) can support multiple readers concurrently, as long as the collection is not modified. Even so, enumerating through a collection is intrinsically not a thread-safe procedure. To guarantee thread safety during enumeration, you can lock the collection during the entire enumeration. To allow the collection to be accessed by multiple threads for reading and writing, you must implement your own synchronization.

Answer 2

There might be problems due to doWork() returning null. As I learned when answering your recent comment here, the thread local object should be passed between subsequent invocations of the Parallel.For body at the same thread, because it is supposed to work as an accumulator; see the usage example at MSDN. It's unclear what happens when you return null, but I would fix that and see whether it makes difference.