Multithreading using Task.Factory and Entity Framework?

Question

I have an application that is notifying each subscriber in our SMS database of an update to our system. It uses the entity framework to select each record and then creates a new Task to send a message to that person. Theoretically, this should be a fast process. I'm doing something wrong because it is only getting one complete response every second or two.

I think that the problem has to do with the way I'm setting up the Tasks in Task.Factory.StartNew(). It is acting like it is running synchronously, but I want it to run asynchronously.

If I am completely off-base with how I am using Tasks, please let me know. I got my inspiration from this post.

Here's my code:

class Program
{
static List<MessageToSend> Messages = new List<MessageToSend>();
static Entities oDatabase = new Entities();
static SMS.API oAPI = new SMS.API();

const string sAuthToken = "*****";
const string sNotificationMessage = "*****";

static void Main(string[] args)
{
    foreach (var subscriber in oDatabase.SMS_Subscribers.Where(x => x.GlobalOptOut == false))
    {
        MessageToSend oMessage = new MessageToSend();
        oMessage.ID = subscriber.ID;
        oMessage.MobileNumber = subscriber.MobileNumber;

        var recentlySentMessage = oDatabase.SMS_OutgoingMessages.Where(x => x.Message == sNotificationMessage && x.MobileNumber == oMessage.MobileNumber && x.Sent > new DateTime(2014, 3, 12)).FirstOrDefault();
        if (recentlySentMessage != null)
        {
            oMessage.Completed = true;
            continue;
        }

        Task t = Task.Factory.StartNew(() =>
        {
            try{
                var keywordID = oDatabase.SMS_SubscribersKeywords.Where(x => x.SubscriberID == oMessage.ID).First().KeywordID;
                var keyword = oDatabase.SMS_Keywords.Where(x => x.ID == keywordID).First();
                oMessage.DemographicID = keyword.DemographicID;
                oMessage.Keyword = keyword.Keyword;

                SendNotificationMessage(oMessage);
            }
            catch (Exception oEx){ //Write exception to console}
        });

        Thread.Sleep(15);
    }

    while (Messages.ToList().Any(x => !x.Completed)){ //wait till all are completed}
}

public static void SendNotificationMessage(object message)
{
    MessageToSend oMessage = (MessageToSend)message;
    try
    {
        SMS.APIResponse oResponse = oAPI.SendMessage(sAuthToken, oMessage.DemographicID, oMessage.Keyword, oMessage.MobileNumber, sNotificationMessage);

        if (oResponse.Success){ //Write success to console }
        else{ //Write failure to console }
    }
    catch (Exception oEx){ //Write Exception to console }

    oMessage.Completed = true;
}
}

class MessageToSend
{
public long ID { get; set; }
public long DemographicID {get;set;}
public string MobileNumber { get; set; }
public bool Completed { get; set; }
public string Keyword { get; set; }

public MessageToSend(){ Completed = false; }
}

EDIT: The inside of the foreach block now looks like this:

        MessageToSend oMessage = new MessageToSend();
        oMessage.ID = subscriber.ID;
        oMessage.MobileNumber = subscriber.MobileNumber;

        int keywordID = 0;
        SMSShortcodeMover.SMS_Keywords keyword;

        var recentlySentMessage = oDatabase.SMS_OutgoingMessages.Where(x => x.Message == sNotificationMessage && x.MobileNumber == oMessage.MobileNumber && x.Sent > new DateTime(2014, 3, 12)).FirstOrDefault();
        if (recentlySentMessage != null)
        {
            oMessage.Completed = true;
            continue;
        }

        try
        {
            keywordID = (int)oDatabase.SMS_SubscribersKeywords.Where(x => x.SubscriberID == oMessage.ID).First().KeywordID;
            keyword = oDatabase.SMS_Keywords.Where(x => x.ID == keywordID).First();
        } catch (Exception oEx){ //write exception to console, then continue; }

        Task t = Task.Factory.StartNew(() =>
        {
            oMessage.DemographicID = keyword.DemographicID;
            oMessage.Keyword = keyword.Keyword;

            SendNotificationMessage(oMessage);
        });

        Thread.Sleep(15);
    }

EDIT 2: I updated my code again, now I gather all of my data before I go into the send. It still is hanging somewhere, but it gets all 52,000 rows of data in about 5 seconds now. The code looks like this:

var query =
(from subscriber in oDatabase.SMS_Subscribers
where subscriber.GlobalOptOut == false
where !(from x in oDatabase.SMS_OutgoingMessages
        where x.Message == sNotificationMessage
        where x.MobileNumber == subscriber.MobileNumber
        where x.Sent > new DateTime(2014, 3, 12)
        select x).Any()
join sk in oDatabase.SMS_SubscribersKeywords
    on subscriber.ID equals sk.SubscriberID
join k in oDatabase.SMS_Keywords on sk.KeywordID equals k.ID into ks
from k2 in ks.Take(1)
select new MessageToSend()
 {
     ID = subscriber.ID,
     MobileNumber = subscriber.MobileNumber,
     DemographicID = k2.DemographicID,
     Keyword = k2.Keyword
 }).ToList();

foreach( var q in query){
    Task t = Task.Factory.StartNew(() => SendNotificationMessage(q));
    Tasks.Add(t);
    Thread.Sleep(80);
}

Task.WaitAll(Tasks.ToArray());

To begin with, try taking the `oDatabase` queries outside of the task. Also why does your code have a `Thread.Sleep(15)`? — Enigmativity, Mar 13 '14 at 20:25
I had a `Thread.Sleep(15)` to limit the speed of the looping. There is a limit to how much traffic we are allowed to generate to our SMS provider. I had the `oDatabase` queries outside of the task initially but I put them inside of it to try to keep them from blocking the rest of the code. It doesn't seem to be helping, though. — ijb109, Mar 13 '14 at 20:31

Enigmativity · Accepted Answer · 2014-03-14T02:04:09.370

1

If I were you I would try to execute all of the database calls at once, before trying to sen your messages.

Try doing this:

var query =
    from subscriber in oDatabase.SMS_Subscribers
    where subscriber.GlobalOptOut == false
    where !(from x in oDatabase.SMS_OutgoingMessages
        where x.Message == sNotificationMessage
        where x.MobileNumber == subscriber.MobileNumber
        where x.Sent > new DateTime(2014, 3, 12)
        select x
    ).Any()
    join sk in oDatabase.SMS_SubscribersKeywords
        on subscriber.ID equals sk.SubscriberID
    join k in oDatabase.SMS_Keywords on sk.KeywordID equals k.ID into ks
    from k2 in ks.Take(1)
    select new
    {
            ID = subscriber.ID,
            MobileNumber = subscriber.MobileNumber,
            DemographicID = k2.DemographicID,
            Keyword = k2.Keyword
    };

var tasks =
    from x in query.ToArray()
    let message = new MessageToSend()
    {
        ID = x.ID,
        MobileNumber = x.MobileNumber,
        DemographicID = x.DemographicID,
        Keyword = x.Keyword
    }
    select Task.Factory.StartNew(() => SendNotificationMessage(message));

Task.WaitAll(tasks.ToArray());

I don't have your database, so I can't test this, but something like this should work if this isn't exactly right.

edited Mar 14 '14 at 02:04

answered Mar 13 '14 at 21:54

Enigmativity

113,464
11
89
172

There are a few bad entries in here, which is why I had all of the try/catch blocks. Will this query avoid those? Also, thank you so much. I would never have thought to do it that way. – ijb109 Mar 14 '14 at 00:36
This is a great direction, I'm still working it out. I got an error that looks like this one in this post and I'm trying to figure out how to adjust the select at the end: http://stackoverflow.com/questions/7259567/linq-to-entities-does-not-recognize-the-method – ijb109 Mar 14 '14 at 00:49
@ijb109 - I've modified the query to get rid of the issue with the error you mentioned. Basically I separated out the database query from the building of the objects you need. Note the `query.ToArray()` call that force the execution of the database query. It should be a bit closer now to working. – Enigmativity Mar 14 '14 at 02:05
After getting the data in a more efficient manner, it seems like the problem is somewhere in the task. It feels like it is only using two or three threads OR all of the threads are hanging on the web request. Thanks for the help in getting the data sorted out! – ijb109 Mar 14 '14 at 02:53
@ijb109 - It could be that your `SMS.API` might not be thread safe. Also the TPL does limit itself to a few threads. – Enigmativity Mar 14 '14 at 03:05

score 0 · Answer 2 · answered Mar 13 '14 at 20:54

It doesn't surprise me that it's taking 1-2 seconds for each iteration of the for each loop because you have 3 separate database calls that are executing synchronously. A database call is pretty slow in the grand scheme of things. One way to approach this would be to have a method that has everything in the foreach block except the Task code and then use task to call it. Just need to be careful that nothing inside the Task Method ends up blocking.

I.e.

var tasks = new List<Task>();
foreach (var subscriber in oDatabase.SMS_Subscribers.Where(x => x.GlobalOptOut == false))
{
   tasks.Add(Task.Factory.StartNew(() => SendNotificationTask(subscriber));
   Thread.Sleep(15);
}
//Might want to to use Task.WhenAll instead of WaitAll.  Just need to debug it and see what happens.
Task.WaitAll(tasks.ToArray());



public void SendNotificationTask(SomeType subscriber)
{
    MessageToSend oMessage = new MessageToSend();
    oMessage.ID = subscriber.ID;
    oMessage.MobileNumber = subscriber.MobileNumber;

    int keywordID = 0;
    SMSShortcodeMover.SMS_Keywords keyword;

    ////Database call 1
    var recentlySentMessage = oDatabase.SMS_OutgoingMessages.Where(x => x.Message == sNotificationMessage && x.MobileNumber == oMessage.MobileNumber && x.Sent > new DateTime(2014, 3, 12)).FirstOrDefault();
    if (recentlySentMessage != null)
    {
        oMessage.Completed = true;
    }
    else
    {
      try
      {
        ////Database call 2
        keywordID = (int)oDatabase.SMS_SubscribersKeywords.Where(x => x.SubscriberID == oMessage.ID).First().KeywordID;

        ////Database call 3
        keyword = oDatabase.SMS_Keywords.Where(x => x.ID == keywordID).First();
    } catch (Exception oEx){ //write exception to console, then continue; }

    oMessage.DemographicID = keyword.DemographicID;
    oMessage.Keyword = keyword.Keyword;

    SendNotificationMessage(oMessage);

   }

}

Multithreading using Task.Factory and Entity Framework?

2 Answers2