Синхронизация множества узлов в Windows Azure

Облако отражает крупные перемены в развитии технологий, и многие отраслевые эксперты отмечают, что перемены такого масштаба происходят не чаще, чем каждые 12 лет или около того. Такая шумиха едва ли удивительна, если подумать, какие преимущества обещает перенос ПО в облако: значительное сокращение эксплуатационных расходов, высокую доступность и почти бесконечную масштабируемость — и это лишь первое, что приходит в голову.

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

К счастью, в феврале Microsoft выпустила платформу Windows Azure, которая содержит множество компонентов для поддержки создания приложений, способных обслуживать колоссальные количества пользователей и в то же время сохранять высокую доступность. Однако ответственность за то, чтобы приложение могло реализовать весь свой потенциал при развертывании в облаке и чтобы оно использовало преимущества самой важной особенности облака — эластичности, лежит на разработчиках системы. эластичность.

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

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

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

Сценарий 1: создание номеров заказов

Недавно мне повезло на деле проверить эту концепцию, перенося существующее веб-приложение в облако на платформе Windows Azure.

Учитывая характер используемых этим приложением данных (они легко секционируются), они были лучшим кандидатом на размещение в Windows Azure Table Storage. Этот простой, но высокопроизводительный механизм хранения — с его почти бесконечной масштабируемостью — был идеальным выбором всего с одним значимым недостатком, связанным с проблемой уникальных идентификаторов.

Целевое приложение позволяло клиентам размещать заказы и получать их номера. Используя SQL Server или SQL Azure, сгенерировать простой числовой уникальный идентификатор было бы совсем нетрудно, но Windows Azure Table Storage не поддерживает основные ключи с автоматическим приращением значений по порядку. Вместо этого разработчики, имеющие дело с Windows Azure Table Storage, могут создавать GUID и использовать его в качестве «ключа» в таблице:

505EAB78-6976-4721-97E4-314C76A8E47E

Проблема с этими GUID в том, что людям сложно работать с ними. Вообразите, как вы сообщаете оператору по телефону номер своего заказа в виде GUID. Конечно, GUID должны быть уникальны в любом контексте, и именно поэтому они столь сложны. С другой стороны, порядковый номер заказа должен быть уникален лишь в пределах таблицы Orders.

Создание простого уникального идентификатора в Windows Azure

Было рассмотрено несколько сравнительно простых решений проблемы с GUID.

1. Использование SQL Azure для генерации уникальных идентификаторов. По ряду причин предпочтение было отдано не SQL Azure, а Windows Azure Table Storage — главным образом из-за необходимости горизонтального масштабирования системы на множество узлов, на каждом из которых выполняется множество потоков, использующих данных.
2. Использование Blob Storage для управления значением приращения. Хранение централизованного счетчика в Windows Azure Blob Storage. Узлы могли бы считывать и обновлять текущий номер заказа, что позволило бы легко формировать последовательность порядковых номеров для заказов и использовать их во множестве узлов. Однако конкуренция между потоками в этой точке при высокой нагрузке на систему, когда требуется выдача большого количества новых номеров в секунду, скорее всего существенно повредила бы масштабируемости системы.
3. Разделение уникальных идентификаторов между каждым узлом. Создание облегченного счетчика в памяти, генерирующего уникальные порядковые числа. Чтобы гарантировать уникальность в пределах всех узлов, каждому узлу выделяется свой диапазон порядковых чисел, как показано в табл. 1.

Табл. 1. Выделение диапазона порядковых номеров каждому узлу, гарантирующее уникальность идентификаторов

Однако этот подход вызывает ряд вопросов. Что будет, если узел исчерпает свой диапазон? Что произойдет, когда в систему будут добавлены сотни узлов одновременно? Как быть, если узел выходит из строя и заменяется другим узлом исполняющей средой Windows Azure? Администраторам придется внимательно отслеживать эти диапазоны, чтобы не произошло повреждения данных.

Вместо этого нужен гораздо более элегантный подход — решение, не требующее конфигурирования каждого узла, допускающего лишь минимальную конкуренцию и гарантирующее уникальность значений в любых ситуациях. Для этого я скомбинировал второй и третий варианты.

Концепция была относительно простой: последний на данный момент порядковый номер хранится в небольшом текстовом файле, помещенном в хранилище двоичных объектов (blob storage). Когда запрашивается новый номер, узел может обратиться к этому файлу, увеличить значение на единицу и записать его обратно в хранилище. Конечно, есть вероятность, что в процессе этого чтения-увеличения-записи другой узел обратится к тому же файлу с тем же намерением. Без управления параллельной обработкой порядковые номера не были бы уникальными, а данные могли бы быть повреждены. По традиции я было подумал создать механизм блокировки, который предотвращает одновременную работу нескольких узлов с одним файлом. Однако блокировки обходятся очень дорого, и, если в реализации главными факторами являются высокие пропускная способность и масштабируемость, блокировок следует избегать.

Вместо блокировок предпочтительнее подход с оптимистичной параллельной обработкой. В этом случае мы разрешаем взаимодействие с ресурсом множеству акторов. Когда ресурс обрабатывается актором, последний выдает маркер, указывающий версию ресурса. При обновлении можно включать маркер, чтобы указывать, какая версия ресурса модифицируется. Если ресурс уже модифицирован другим актором, обновление заканчивается неудачей, и первоначальный актор может получить последнюю версию ресурса и попробовать снова обновить его. Оптимистичная параллельная обработка неплохо функционирует при условии, что вероятность конкуренции при обновлениях невысока. Тем самым мы избегаем издержек и сложности работы с блокировками и в то же время защищаем ресурс от повреждения.

Представьте, что в пиковые моменты система выдает примерно 100 новых номеров заказов в секунду. Это означает, что ежесекундно поступает 100 запросов на обновление файла, создавая крайне высокую вероятность конкуренции, вследствие чего потребуется много повторных попыток и ситуация ухудшится. Поэтому, чтобы уменьшить вероятность такого развития событий, я решил выделять порядковые номера диапазонами.

Для инкапсуляции этого поведения был создан класс UniqueIdGenerator. Он удаляет диапазон порядковых номеров из хранилища двоичных объектов, увеличивая значение с настраиваемым шагом. Если каждый UniqueIdGenerator должен резервировать 1000 номеров заказов одновременно, обновление хранилища будет происходить лишь 10 раз в секунду в среднем, что резко уменьшает вероятность конкуренции. Каждый UniqueIdGenerator свободен в раздаче зарезервированных за ним 1000 номеров и может быть уверен, что ни один другой экземпляр класса, ссылающийся на тот же ресурс, не выдаст тот же номер заказа.

Чтобы сделать этот новый компонент тестируемым, был определен интерфейс IOptimisticSyncStore, который отделяет UniqueIdGenerator от конкретного механизма хранения. Это дает дополнительное преимущество: в будущем этот компонент сможет использовать другой тип хранилища, если это понадобится. Вот как выглядит этот интерфейс:

public interface IOptimisticSyncStore
{
  string GetData();
  bool TryOptimisticWrite(string data);
}

Как видите, это весьма простой интерфейс всего с двумя методами: один из них извлекает данные, а другой обновляет их, причем последний возвращает булево значение, где false указывает на провал оптимистичной параллельной обработки и то, что процесс следует повторить заново.

Реализация IOptimisticSyncStore, использующая хранилище двоичных объектов, доступна в исходном коде, который можно скачать для этой статьи. По большей части эта реализация проста, но на метод TryOptimisticWrite стоит посмотреть повнимательнее, чтобы понять, как реализована оптимистичная параллельная обработка.

Оптимистичной параллельной обработкой несложно пользоваться при обновлении ресурсов в Windows Azure Blob Storage благодаря Preconditions and Entity Tags (ETags). Предусловие (precondition) — это выражение, которое, как назначает разработчик, должно быть true, чтобы HTTP-запрос завершился успешно. Если веб-сервер оценивает выражение как false, он должен ответить HTTP-кодом 412 (предусловие не выполнено). ETags также является частью спецификации HTTP и идентифицирует конкретную версию ресурса, например двоичного объекта. Если этот объект изменяется, его ETag также должен измениться, как показано ниже:

try
{

  _blobReference.UploadText(

    data,

    Encoding.Default,

    new BlobRequestOptions {

    AccessCondition = AccessCondition.IfMatch(
    _blobReference.Properties.ETag) });

}

Чтобы указать предусловие в коде, мы используем тип BlobRequestOptions и задаем свойство AccessCondition. Если это условие доступа не выполнено (например, другой узел обновил двоичный объект через мгновение после того, как тот был считан первым узлом), то ETag не совпадут и будет сгенерировано исключение StorageClientException:

catch (StorageClientException exc)
{
  if (exc.StatusCode == HttpStatusCode.PreconditionFailed)
  {
    return false;
  }
  else
  {
    throw;
  }
}
return true;

В этом экземпляре реализация проверяет наличие исключения для кода состояния PreconditionFailed и возвращает false. Все остальные типы исключений – серьезные ошибки, создаваемые повторно для обработки и дальнейшей регистрации. Отсутствие исключения означает выполнение обновления; метод возвращает true. Полный список класса UniqueIdGenerator показан на рис. 2.

Рис. 2. Полный исходный код класса UniqueIdGenerator

public class UniqueIdGenerator
{
    private readonly object _padLock = new object();
    private Int64 _lastId;
    private Int64 _upperLimit;
    private readonly int _rangeSize;
    private readonly int _maxRetries;
    private readonly IOptimisticSyncStore _optimisticSyncStore;

    public UniqueIdGenerator(
      IOptimisticSyncStore optimisticSyncStore,
      int rangeSize = 1000,
      int maxRetries = 25)
    {
      _rangeSize = rangeSize;
      _maxRetries = maxRetries;
      _optimisticSyncStore = optimisticSyncStore;

      UpdateFromSyncStore();
    }

    public Int64 NextId()
    {
      lock (_padLock)
      {
        if (_lastId == _upperLimit)
        {
          UpdateFromSyncStore();
        }
        return _lastId++;
      }
    }

    private void UpdateFromSyncStore()
    {
      int retryCount = 0;
      // maxRetries + 1 because the first run isn't a 're'try.
      while (retryCount < _maxRetries + 1)
      {
        string data = _optimisticSyncStore.GetData();

        if (!Int64.TryParse(data, out _lastId))
        {
          throw new Exception(string.Format(
            "Data '{0}' in storage was corrupt and " +
            "could not be parsed as an Int64", data));
        }

        _upperLimit = _lastId + _rangeSize;

        if (_optimisticSyncStore.TryOptimisticWrite(
          _upperLimit.ToString()))
        {
          return;
        }

        retryCount++;
        // update failed, go back around the loop
      }

      throw new Exception(string.Format(
        "Failed to update the OptimisticSyncStore after {0} attempts",
        retryCount));
    }
}

Конструктор принимает три параметра. Первый — это реализация IOptimisticSyncStore, например наш BlobOptimisticSyncStore, обсуждавшийся ранее. Второй — rangeSize — целое значение, которое указывает, насколько большим должен быть выделяемый в хранилище двоичных объектов диапазон номеров. Чем больше диапазон, тем ниже вероятность конкуренции (но и больше номеров будет потеряно при крахе данного узла). Последний параметр — maxRetries целочисленного типа; он задает, сколько раз генератор должен пытаться обновлять хранилище в случае неудачи оптимистичной параллельной обработки. Если и последняя попытка окажется неудачной, будет сгенерировано исключение.

NextId — единственный открытый метод класса UniqueIdGenerator, и он используется для выборки следующего уникального номера. Тело этого метода синхронизируется, чтобы гарантировать безопасность любого экземпляра этого класса в многопоточной среде и возможность его совместного использования всеми потоками, выполняющими веб-приложение. Выражение if проверяет, не достиг ли генератор верхнего лимита выделенного ему диапазона, и, если да, вызывает UpdateFromSyncStore для получения нового диапазона из хранилища двоичных объектов.

Метод UpdateFromSyncStore — последняя и самая интересная часть класса. Для получения верхнего лимита ранее выделенного диапазона используется реализация IOptimisticSyncStore. Это значение увеличивается на размер диапазона для генератора и записывается обратно в хранилище. В конце тела метода находится простой цикл while, обеспечивающий должное количество повторных попыток, если TryOptimisticWrite вернет false.

В следующем фрагменте кода конструируется экземпляр UniqueIdGenerator и используется BlobOptimisticSyncStore с файлом ordernumber.dat в контейнере uniqueids (заметьте, что имена контейнеров в хранилище двоичных объектов должны состоять из букв нижнего регистра):

IOptimisticSyncStore storage = new BlobOptimisticSyncStore(
  CloudStorageAccount.DevelopmentStorageAccount,
  "uniqueids",
  "ordernumber.dat");
UniqueIdGenerator
  generator = new UniqueIdGenerator(storage, 1000, 10);

Этот экземпляр удаляет 1000 идентификаторов из диапазона в централизованном хранилище и будет выполнять 10 попыток при неудаче оптимистичной параллельной обработки перед генерацией исключения.

Пользоваться UniqueIdGenerator еще проще. Когда вам понадобится новый уникальный номер заказа, просто вызовите NextId:

Int64 orderId = generator.NextId();

В примере кода показан веб-роль Windows Azure, которая использует множество потоков для быстрого выделения уникальных номеров заказов и их записи в базу данных SQL. В данном случае SQL применяется просто для того, чтобы доказать, что каждый порядковый номер уникален; любое нарушение этого правила привело бы к нарушению корректности основного ключа и вызвало бы исключение.

Преимущество такого подхода, если не считать создания двоичного объекта и присваивания ему нулевого значения в момент начала жизненного цикла приложения, заключается в том, что от системного администратора не требуется никакого вмешательства. UniqueIdGenerator сам управляет выделением идентификаторов на основе ваших настроек, корректно восстанавливается в случае неудачи и без проблем масштабируется в самых эластичных средах.

Сценарий 2: спускайте собак!

Другое интересное требование, предъявляемое приложением, заключалось в необходимости быстро обрабатывать большие объемы данных вслед за определенным событием, которое происходило в примерно известное время. Из-за природы этой обработки никакую работу с любыми данными после этого события начинать было нельзя.

В этом сценарии очевидным выбором являются рабочие роли, и было бы возможным просто указать Windows Azure предоставлять нужное количество рабочих ролей в ответ на вышеупомянутое событие. Однако подготовка новых ролей может занимать до 30 минут, а в данном сценарии все решала скорость. Поэтому было решено, что роли будут готовиться заранее, но оставаться в приостановленном состоянии, пока не будут запущены администратором, — я назвал это действо «спускайте собак!». Рассматривались два возможных подхода, и я по очереди расскажу о каждом из них.

Обратите внимание:так как оплата за рабочие роли Windows Azure взимается с момента их развертывания (а не с момента активного использования ими процессоров), этот вариант будет стоить дороже по сравнению с простым созданием рабочих ролей в ответ на событие. Однако заказчик дал ясно понять, что эти расходы стоят того:для него было очень важно, чтобы обработка начиналась как можно быстрее.

Подход 1: опрос

При первом подходе, показанном на рис. 2, каждый узел периодически опрашивает флаг состояния в централизованном хранилище (он вновь хранится в Windows Azure Blob Storage), чтобы определить, можно ли уже начать обработку.

Рис. 3. Узлы опрашивают флаг состояния в централизованном хранилище

Чтобы «снять узлы с паузы», клиентское приложение должно просто установить этот флаг в true, и при следующем опросе каждый узел будет запущен. Основной недостаток этого подхода — задержка, которая потенциально может достигать длительности интервала опроса. С другой стороны, это очень простой и надежный механизм.

Этот вариант демонстрирует класс PollingRelease, доступный в примерах кода. Для поддержки тестируемости механизм хранения флага был абстрагирован и скрыт за интерфейсом во многом по аналогии с интерфейсом для класса UniqueIdGenerator. Интерфейс IGlobalFlag и сопутствующая реализация хранилища показаны на рис. 4.

Рис. 4. Интерфейс IGlobalFlag и реализация для Blob Storage

public interface IGlobalFlag
{
  bool GetFlag();
  void SetFlag(bool status);
}

public class BlobGlobalFlag : IGlobalFlag
{
  private readonly string _token = "Set";
  private readonly CloudBlob _blobReference;
  public BlobGlobalFlag(CloudStorageAccount account, string container,
    string address)
  {
    var blobClient = account.CreateCloudBlobClient();
    var blobContainer =
      blobClient.GetContainerReference(container.ToLower());
    _blobReference = blobContainer.GetBlobReference(address);
  }

  public void SetFlag(bool status)
  {
    if (status)
   {
      _blobReference.UploadText(_token);
    }
    else
    {
      _blobReference.DeleteIfExists();
    }
  }

  public bool GetFlag()
  {
    try
    {
      _blobReference.DownloadText();
      return true;
    }
    catch (StorageClientException exc)
    {
      if (exc.StatusCode == System.Net.HttpStatusCode.NotFound)
      {
        return false;
      }
      throw;
    }
  }
}

Обратите внимание, что в этом примере простое наличие файла в хранилище двоичных объектов указывает на true независимо от содержимого этого файла.

Сам по себе класс PollingRelease довольно прост (рис. 4) и содержит всего один открытый метод Wait.

Рис. 5. Класс PollingRelease

public class PollingRelease
{
  private readonly IGlobalFlag _globalFlag;
  private readonly int _intervalMilliseconds;

  public PollingRelease(IGlobalFlag globalFlag,
    int intervalMilliseconds)
  {
    _globalFlag = globalFlag;
    _intervalMilliseconds = intervalMilliseconds;
  }

  public void Wait()
  {
    while (!_globalFlag.GetFlag())
    {
      Thread.Sleep(_intervalMilliseconds);
    }
  }
}

Этот метод блокирует любой вызвавший код до тех пор, пока реализация IGlobalFlag указывает, что ее статус равен false. В следующем фрагменте кода показано, как пользоваться классом PollingRelease:

BlobGlobalFlag globalFlag = new BlobGlobalFlag(
  CloudStorageAccount.DevelopmentStorageAccount,
  "globalflags",
  "start-order-processing.dat");
PollingRelease pollingRelease = new PollingRelease(globalFlag, 2500);
pollingRelease.Wait();

Создаваемый экземпляр BlobGlobalFlag указывает на контейнер globalflags. Через каждые 2,5 секунды класс PollingRelease будет проверять наличие файла start-order-processing.dat; любой вызов метода Wait блокируется до тех пор, пока в хранилище не появится этот файл.

Подход 2: прослушивание

Во втором подходе используется Windows Azure AppFabric Service Bus для одновременного взаимодействия напрямую со всеми рабочими ролями и из запуска (рис. 6).

Рис. 6. Одновременное взаимодействие со всеми рабочими ролями через Windows Azure AppFabric Service Bus

Service Bus — крупномасштабный сервис обмена сообщениями и поддержки соединений, также построенный на Windows Azure. Он обеспечивает безопасное взаимодействие между различными компонентами распределенного приложения. Service Bus — идеальный способ соединения двух приложений, которым иначе было бы трудно взаимодействовать, например, из-за их размещения за границами NAT (Network Address Translation) или из-за частой смены IP-адресов. Подробный обзор Windows Azure AppFabric Service Bus не вписывается в рамки этой статьи, но вы найдете превосходное учебное пособие по этой шине сервисов на сайте MSDN по ссылке msdn.microsoft.com/library/ee706736.

Чтобы продемонстрировать этот подход, был создан класс ListeningRelease, который подобно PollingRelease имеет один открытый метод Wait. Этот метод обеспечивает соединение с Service Bus и использует ManualResetEvent для блокировки потока, пока не будет принято оповещение:

public void Wait()
{
  using (ConnectToServiceBus())
  {
    _manualResetEvent.WaitOne();
  }
}

Полный исходный код метода ConnectToServiceBus приведен на рис. 7. Он использует типы из сборок System.ServiceModel и Microsoft.ServiceBus, чтобы предоставлять класс UnleashService облаку через Windows Azure AppFabric Service Bus рис. 8.

Рис. 7. Метод ConnectToServiceBus

private IDisposable ConnectToServiceBus()
{
  Uri address = ServiceBusEnvironment.CreateServiceUri("sb",
    _serviceNamespace, _servicePath);
  TransportClientEndpointBehavior sharedSecretServiceBusCredential =
    new TransportClientEndpointBehavior();
  sharedSecretServiceBusCredential.CredentialType =
    TransportClientCredentialType.SharedSecret;
  sharedSecretServiceBusCredential.Credentials.SharedSecret.
    IssuerName = _issuerName;
  sharedSecretServiceBusCredential.Credentials.SharedSecret.
    IssuerSecret = _issuerSecret;

  // Create the single instance service, which raises an event
  // when the signal is received.
  UnleashService unleashService = new UnleashService();
  unleashService.Unleashed += new
    EventHandler(unleashService_Unleashed);

  // Create the service host reading the configuration.
  ServiceHost host = new ServiceHost(unleashService, address);

  IEndpointBehavior serviceRegistrySettings =
    new ServiceRegistrySettings(DiscoveryType.Public);

  foreach (ServiceEndpoint endpoint in host.Description.Endpoints)
  {
    endpoint.Behaviors.Add(serviceRegistrySettings);
    endpoint.Behaviors.Add(sharedSecretServiceBusCredential);
  }

  host.Open();

  return host;
}

Рис. 8. Класс UnleashService

[ServiceBehavior(InstanceContextMode= InstanceContextMode.Single)]
public class UnleashService : IUnleashContract
{
  public void Unleash()
  {
    OnUnleashed();
  }

  protected virtual void OnUnleashed()
  {
    EventHandler temp = Unleashed;
    if (temp != null)
    {
      temp(this, EventArgs.Empty);
    }
  }

  public event EventHandler Unleashed;
}

Хостинг UnleashService осуществляется Windows Communication Foundation (WCF) как единственного экземпляра; он реализует контракт IUnleashService, в котором только один метод: Unleash. Класс ListeningRelease ожидает вызова этого метода через показанное ранее событие Unleashed. Когда класс ListeningRelease обнаруживает это событие, ManualResetEvent, который до сих пор блокировал любые вызовы Wait, сбрасывается и все блокированные потоки освобождаются.

В конфигурации сервиса я использовал привязку NetEventRelayBinding, которая поддерживает групповую рассылку (multicasting) через Service Bus, позволяя любому количеству издателей и подписчиков взаимодействовать через одну конечную точку. Природа взаимодействия черех групповую рассылку требует, чтобы все операции были односторонними way, как демонстрируется интерфейсом IUnleashContract:

[ServiceContract]
public interface IUnleashContract
{
  [OperationContract(IsOneWay=true)]
  void Unleash();
}

Конечная точка защищается общим секретом (именем пользователя и сложным паролем). Зная эти детали, любой клиент с доступом в Интернет мог бы запустить метод Unleash, в том числе, например, консоль администратора, включенная в примеры кода (рис. 8).

Рис. 9. Консоль администратора

Хотя подход с ListeningRelease действительно избавляет от задержек, присущих классу PollingRelease, есть и другие задержки, с которыми тоже надо как-то бороться. Однако главный недостаток подхода с прослушиванием заключается в том, что он не поддерживает состояния, из-за чего любые подготовленные узлы после передачи оповещения о запуске не увидели бы это событие и остались бы в приостановленном состоянии. Конечно, очевидным решением могло бы быть комбинированное использование как Service Bus, так и глобального флага в Blob Storage, но это я оставлю читателям в качестве упражнения.

Примеры кода

Сопутствующий пример решения доступен по ссылке code.msdn.microsoft.com/mag201011Sync и включает файл ReadMe, в котором перечисляются требования к аппаратно-программному обеспечению и содержатся инструкции по установке и настройке. Этот пример использует ListeningRelease, PollingRelease и UniqueIdGenerator в единственной рабочей роли.