Как управлять мьютексом в асинхронном методе

Я перенес свой старый код HttpHandler (.ashx) TwitterFeed в приложение WebAPI. Ядро кода использует отличный пакет Linq2Twitter (https://linqtotwitter.codeplex.com/). Часть переноса включала обновление этого компонента с версии 2 до версии 3, которая теперь предоставляет ряд асинхронных вызовов методов, которые для меня являются новыми. Вот основной контроллер:

public async Task<IEnumerable<Status>> 
GetTweets(int count, bool includeRetweets, bool excludeReplies)
{
   var auth = new SingleUserAuthorizer
   {
      CredentialStore = new SingleUserInMemoryCredentialStore
      {
         ConsumerKey       = ConfigurationManager.AppSettings["twitterConsumerKey"],
         ConsumerSecret    = ConfigurationManager.AppSettings["twitterConsumerKeySecret"],
         AccessToken       = ConfigurationManager.AppSettings["twitterAccessToken"],
         AccessTokenSecret = ConfigurationManager.AppSettings["twitterAccessTokenSecret"]
      }
   };

   var ctx = new TwitterContext(auth);

   var tweets =
      await
      (from tweet in ctx.Status
         where (
            (tweet.Type == StatusType.Home)
            && (tweet.ExcludeReplies == excludeReplies)
            && (tweet.IncludeMyRetweet == includeRetweets)
            && (tweet.Count == count)
         )
      select tweet)
      .ToListAsync();

   return tweets;
}

Это работает нормально, но ранее я кэшировал результаты, чтобы избежать «чрезмерного вызова» API Twitter. Именно здесь я столкнулся с проблемой (больше связанной с моим непониманием асинхронного протокола, чем с чем-то еще, что я подозреваю).

В целом, я хочу сначала проверить кеш, если данные не существуют, затем повторно увлажнить кеш и вернуть данные вызывающей стороне (веб-странице). Вот моя попытка кода

public class TwitterController : ApiController {

   private const string CacheKey = "TwitterFeed";

   public async Task<IEnumerable<Status>>
   GetTweets(int count, bool includeRetweets, bool excludeReplies)
   {
      var context = System.Web.HttpContext.Current;
      var tweets = await GetTweetData(context, count, includeRetweets, excludeReplies);
      return tweets;
   }

   private async Task<IEnumerable<Status>>
   GetTweetData(HttpContext context, int count, bool includeRetweets, bool excludeReplies)
   {
      var cache = context.Cache;
      Mutex mutex = null;
      bool iOwnMutex = false;
      IEnumerable<Status> data = (IEnumerable<Status>)cache[CacheKey];

      // Start check to see if available on cache
      if (data == null)
      {
         try
         {
            // Lock base on resource key
            mutex = new Mutex(true, CacheKey);

            // Wait until it is safe to enter (someone else might already be
            // doing this), but also add 30 seconds max.
            iOwnMutex = mutex.WaitOne(30000);

            // Now let's see if some one else has added it...
            data = (IEnumerable<Status>)cache[CacheKey];

            // They did, so send it...
            if (data != null)
            {
               return data;
            }

            if (iOwnMutex)
            {
               // Still not there, so now is the time to look for it!
               data = await CallTwitterApi(count, includeRetweets, excludeReplies);

               cache.Remove(CacheKey);
               cache.Add(CacheKey, data, null, GetTwitterExpiryDate(),
                  TimeSpan.Zero, CacheItemPriority.Normal, null);
            }
         }
         finally
         {
            // Release the Mutex.
            if ((mutex != null) && (iOwnMutex))
            {
               // The following line throws the error:
               // Object synchronization method was called from an
               // unsynchronized block of code.
               mutex.ReleaseMutex();
            }
         }
      }

      return data;
   }

   private DateTime GetTwitterExpiryDate()
   {
      string szExpiry = ConfigurationManager.AppSettings["twitterCacheExpiry"];
      int expiry = Int32.Parse(szExpiry);
      return DateTime.Now.AddMinutes(expiry);
   }

   private async Task<IEnumerable<Status>>
   CallTwitterApi(int count, bool includeRetweets, bool excludeReplies)
   {
      var auth = new SingleUserAuthorizer
      {
         CredentialStore = new SingleUserInMemoryCredentialStore
         {
            ConsumerKey = ConfigurationManager.AppSettings["twitterConsumerKey"],
            ConsumerSecret = ConfigurationManager.AppSettings["twitterConsumerKeySecret"],
            AccessToken = ConfigurationManager.AppSettings["twitterAccessToken"],
            AccessTokenSecret = ConfigurationManager.AppSettings["twitterAccessTokenSecret"]
         }
      };

      var ctx = new TwitterContext(auth);

      var tweets =
         await
         (from tweet in ctx.Status
          where (
             (tweet.Type == StatusType.Home)
             && (tweet.ExcludeReplies == excludeReplies)
             && (tweet.IncludeMyRetweet == includeRetweets)
             && (tweet.Count == count)
             && (tweet.RetweetCount < 1)
          )
          select tweet)
         .ToListAsync();

      return tweets;
   }

}

Проблема возникает в блоке кода finally, где освобождается мьютекс (хотя у меня есть опасения по поводу общего шаблона и подхода к методу GetTweetData()):

if ((mutex != null) && (iOwnMutex))
{
    // The following line throws the error:
    // Object synchronization method was called from an
    // unsynchronized block of code.
    mutex.ReleaseMutex();
}

Если я закомментирую строку, код работает правильно, но (я предполагаю) я должен освободить Mutex, создав его. Из того, что я выяснил, эта проблема связана с изменением потока между созданием и освобождением мьютекса.

Из-за отсутствия у меня общих знаний об асинхронном кодировании я не уверен: а) жизнеспособен ли используемый мной шаблон и б) если да, то как решить проблему.

Любые советы будут высоко ценится.


person Neilski    schedule 19.03.2014    source источник


Ответы (1)


Использование такого мьютекса не сработает. Во-первых, Mutex аффинен к потоку, поэтому его нельзя использовать с кодом async.

Другие проблемы, которые я заметил:

Есть один важный совет относительно кэширования: если у вас есть только кэш в памяти, кэшируйте задачу, а не результирующие данные. Кстати, я должен задаться вопросом, является ли HttpContext.Cache лучшим кэшем для использования, но я оставлю его как есть, поскольку ваш вопрос больше касается того, как асинхронный код изменяет шаблоны кэширования.

Итак, я бы рекомендовал что-то вроде этого:

private const string CacheKey = "TwitterFeed";

public Task<IEnumerable<Status>> GetTweetsAsync(int count, bool includeRetweets, bool excludeReplies)
{
  var context = System.Web.HttpContext.Current;
  return GetTweetDataAsync(context, count, includeRetweets, excludeReplies);
}

private Task<IEnumerable<Status>> GetTweetDataAsync(HttpContext context, int count, bool includeRetweets, bool excludeReplies)
{
  var cache = context.Cache;
  Task<IEnumerable<Status>> data = cache[CacheKey] as Task<IEnumerable<Status>>;
  if (data != null)
    return data;
  data = CallTwitterApiAsync(count, includeRetweets, excludeReplies);
  cache.Insert(CacheKey, data, null, GetTwitterExpiryDate(), TimeSpan.Zero);
  return data;
}

private async Task<IEnumerable<Status>> CallTwitterApiAsync(int count, bool includeRetweets, bool excludeReplies)
{
  ...
}

Существует небольшая вероятность того, что если два разных запроса (из двух разных сеансов) запрашивают один и тот же твиттер-канал в одно и то же время, этот канал будет запрошен дважды. Но я бы не потерял сон из-за этого.

person Stephen Cleary    schedule 19.03.2014
comment
Спасибо, Степан, очень полезно. Причина, по которой я пытался внедрить Mutex, заключалась в том, чтобы избежать двух запросов Twitter из двух сеансов. Лента Twitter отображается на каждой странице веб-сайта, поэтому я рассудил, что если на возврат ленты требуется, скажем, 5 секунд, то вероятность двух одновременных запросов довольно высока, но я понимаю вашу точку зрения. Еще раз спасибо за ваше подробное объяснение и пример. - person Neilski; 20.03.2014
comment
Если вы кэшируете задачу, существует очень низкая вероятность того, что фид Twitter будет запрошен дважды. Это связано с тем, что задача добавляется в кеш сразу после начала загрузки, а не когда она завершается. - person Stephen Cleary; 20.03.2014
comment
Это на самом деле фантастически умно. Я предполагаю, что из-за того, что он кэширует задачу, даже если другой процесс делает запрос к кешу до его загрузки, он просто ждет ответа задачи? Мне трудно понять, что это вообще возможно! - person Neilski; 20.03.2014
comment
Да. Как только (незавершенная) задача будет добавлена ​​в кеш, любые другие потоки смогут прочитать эту задачу из кеша. Задачу можно ожидать любым количеством методов одновременно, и все они продолжат работу после завершения задачи. Так же после завершения задачи ожидание задачи не дает результата; вместо этого ожидание следует за быстрый путь и продолжает выполняться синхронно. Так что мне нравится хранить задачи в кеше, но это можно сделать только с кешем в памяти. - person Stephen Cleary; 20.03.2014