Interleave - чередование

MRDS предоставляет средства разработки приложений для роботов.

Необходимость многопоточности

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

Для организации взаимодействия MRDS основан на двух компонентах: Среда координации параллельных задач (Concurrency and Coordination Runtime, CCR) и Распределенные программные службы (Decentralized Software Services, DSS). CCR используется для работы с многопоточностью и внутризадачной синхронизации, в то время как DSS - для постороения приложения, основанного на модели слабосвязанных сервисов.

Сервисы как основная структурная единица

Основной структурной единицей в MRDS является сервис. Сервисы позволяют объединять CCR-приложения. Причем, если CCR работает с многопоточностью на одной машине, то DSS позволяет запускать на разных машинах сервисы, которые будут общаться по сети.

Взаимодействие сервисов

Одной из основных задач DSS является объединение сервисов и предоставление среды передачи сообщения между ними.

• Асинхронная модель, основанная на передаче сообщений. Сообщение - это элемент класса данных, который при передаче сериализуется в xml.
• Сообщения передаются через порты. Тип принимаемых портом сообщений указывается в его объявлении.
• Отправка (posting) сообщения осуществляется путем его размещения в порт. Сообщение остается в очереди до извлечения получателем (receiver). Получателю могут быть заданы условия активации, такие как объединение (join, можно рассматривать как логическое «И») или выбор (choise, можно рассматривать как логическое или).
• Определение условий активации - роль арбитра. Таким образом арбитры реализуют синхронизацию и управление задачами (task). Как только условие получателя выполнено, сообщение из очереди сообщений попадает в очередь диспетчера, а оттуда - диспетчеру для исполнения.
• Когда сообщение запланировано для запуска, оно передается обработчику (handler) для исполнения. Обработчик - блок кода работающий в многопоточной среде. Он может быть запущен самостоятельно, но чаще запускается при обработке сообщения, принятого получателем (receiver).
• Обработку сбоев (failure) осуществляет встроенный механизм причинности (causalities), который можно сравнить с блоком try/catch

Как видно на схеме архитектуры CCR, диспетчер назначает задачи, передавая их потокам из пула. Задача наследуют интерфейс ITask и содержит ссылку на обработчик, который должен быть выполнен.

Пример простого обработчика

void SimpleHandler() {
  Console.WriteLine("Начало простого обработчика");
  for (int i = 0; i < 5; i++) {
    Wait(100);
    Console.Write(i + " ");
  }
  Console.WriteLine("Окончание простого обработчика");
}

Класс DsspServiceBase позволяет выполнить обработчика как задачу посредством адаптера (wrapper) Spawn:

Spawn(SimpleHandler);

Количество потоков в пуле диспетчера определяется тремя способома:

• по умолчанию зависит от количества ядер процессора;
• может быть переопределено атрибутом ActivationSettings класса сервиса;

[ActivationSettings(SharedDispatcher=false, ExecutionUnitsPerDispatcher=6)]
class dssService : DsspServiceBase

• другой способ задать количество потоков - указать явно при создании своего диспетчера;

void RunTaskFromHander() {
  Dispatcher disp = new Dispatcher(4, "Тестовый Пул");
  DispatcherQueue queue = new dispatcherQueue("Тестовая Очередь", disp);
  
  // Активируем три задачи на пуле из четырех потоков
  Arbiter.Activate(queue,
    Arbiter.FromHandler(SimpleHandler),
    Arbiter.FromHandler(SimpleHandler),    
    Arbiter.FromHandler(SimpleHandler));
}

Часто задача представляет из себя всего пару строк кода. В таком случае ее удобно представить в виде делегата (delegate), который можно сравнить с анонимным методом или лексическим выражением (closure). Следующий пример поясняет использование делегата.

Activate(
  Arbiter.Receive(true, intPort,
    delegate(int n) { Console.WriteLine("Делегат"); }
  )
);

Итератор объявляется как метод, возращающий тип IEnumerator<ITask>, т.е. он итерируется по задачам (task). Он позволяет выполнять последовательные действия, не блокируя исполняющий поток при ожидании сообщения. Этот подход использует операторы yield return, yield break и называется возобновление (continuation). На этих операторах исполнение приостанавливается и через некоторое время продолжается со следующего выражения. На каждой итерации возвращается значение (yield return) или итератор прерывается (yield break).

private static IEnumerator<ITask> IteratorExample() {
  Port<int> p1 = new Port<int>();
  Port<int> p2 = new Port<int>();
  p1.Post(0);
  bool done = false;
  while (!done) {
    yield return Arbiter.Receive(false, p1,
      delegate(int i) {
        Console.WriteLine(“P1 Thread {0}: {1}”,
        Thread.CurrentThread.ManagedThreadId, i);
        p2.Post(i + 1);
      }
    );
    
    yield return Arbiter.Receive(false, p2,
      delegate(int i) {
        Console.WriteLine(“P2 Thread {0}: {1}”, Thread.CurrentThread.ManagedThreadId, i);
        if (i >= 10)
          done = true;
        else
          p1.Post(i + 1);
      }
    );
  }
  yield break;
}

Ключевые моменты при работе с итераторами:

• при выполнении оператора yield return или yield break исполняемый поток возвращается в пул до получения сообщения;
• получатели не сохраняют свое состояние (nonpersistent);
• итератор должен исполняться как задача, в противном случае он будет просто проигнорирован;
• yield нельзя использовать внутри делегата или блока try, имеющего catch; допустимо использование только внутри итератора, в противном случае возникнет ошибка компиляции.

Порт, который по сути является FIFO-очередью (First-In-First-Out) сообщений, реализован одноименным классом Port. При определении порта указывается тип принимаемых им сообщений. Cообщение - это объект заданного типа, т.е. элемент заданного класса языка C#. Класс может быть выбран произвольно взависимости от сообщения, которое вы хотите передать. Это может быть ваш собственный класс.

Рассмотрим основные методы и поля класса Port:

• port.ToString() - выводит в стандартный поток вывода содержимое порта. Обычно используется в отладочных целях.
• port.itemCount - это поле содержит значение, соответствующее количеству сообщений, которые находятся в очереди.
• port.Test(out msg) - при наличии сообщений в очереди объект msg будет содержать следующее сообщение, в противном случае - null. Полученный объект автоматически удаляется из порта. Метод поточно-ориентирован, поэтому исключено получение одного сообщения двумя потоками.
• port.Clear() - очищает порт от всех сообщений.

Отметим, что обычно вам не понадобится считывать сообщения напрямую из порта: для автоматического получения сообщений используются ресиверы.

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

На каждый тип сообщения может назначаться свой обработчик. А для активации обработчика может задаваться правило на получение сообщений определенного набора типов.

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

Работая с наборами портов обратите внимания на следующие ограничения:

• в группе портов для desktop CLR может находится максимум 20 типов;
• в группе портов для .NET Compact Framework (CF) environment - максимум 8.

Указанные ограничения можно обойти путем создания набора портов программно: используя typeof. Однако при таком подходе не осуществляется проверка типа на уровне компиляции. Это устраняется перегрузкой метода post для каждого типа сообщения.

public class PSExampleOperations : PortSet {
  public PSExampleOperations()
    : base (
      typeof(DsspDefaultLookup),
      typeof(DsspDefaultDrop),
      typeof(Get)
    )
  { }
  
  public void Post(DsspDefaultLookup msg) {
    base.PostUnknownType(msg);
  }
  
  public void Post(DsspDefaultDrop msg) {
    base.PostUnknownType(msg);
  }
  
  public void Post(Get msg) {
    base.PostUnknownType(msg);
  }
}

Для получения и обработки сообщений у порта есть список ресиверов (receiver). Ресиверы можно рассматривать как функции обратного вызова, т.е. блок кода, который передается в качестве параметра функции. Создаем получателя и прикрепляем его к порту при помощи адаптера dss или ccr:

Activate(
  Arbiter.Receive(false, intPort,
    delegate(int n)
      { Console.WriteLine(“Receiver 1: “ + n.ToString()); }
  )
);

Продемострированный метод Activate определен в классе DsspServiceBase и расширяет Arbiter.Activate. В этом случае используется очередь Environment.TaskQueue - очередь по умолчанию DSS-деспетчера. Можно задать собственного диспетчера.

Функция Arbiter.Receive создает ресивер. Рассмотрим ее параметры:

• флаг постоянства (persistent flag) - остается ли ресивер в списке после получения сообщения;
• порт из которого осуществляется чтение;
• исполняемый делегат.

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

Choice-арбитр, который можно рассматривать как логическое «ИЛИ», ожидает запуска одного из двух ресиверов, при этом второй ресивер закрывается. Типичное применение - обработка ответа от сервера: либо получены запрошенные данные, либо ответ содержит сообщение об ошибке.

void Choice() {
  // Создаем порт-сет, который принимает два типа сообщений
  PortSet<bool, int> portSet = new PortSet<bool, int>();

  // Создаем Choice-арбитра и активируем его
  Activate(
    Arbiter.Choice<bool, int>(portSet,
      // Делегат для типов из группы портов
      delegate(bool b)
        { Console.WriteLine("Received: " + b.ToString()); },
      delegate(int i)
        { Console.WriteLine("Received: " + i.ToString()); }
    )
  );
}

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

Поддерживается вложенность арбитров: как join-, так и choice-арбитры могут содержать в себе другие join- и choice-арбитры, что позволяет реализовывать логику произвольной сложности.

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

• получение набора элементов (multiple item receive) - указывается количество сообщений для активации ресивера;
• получение на наборе портов (multiple port receive) - аналогично предыдущему пункту, только прием сообщений осуществляется не одним, а группой портов (PortSet).

Приведем пример кода активации арбитра.

// multiple item receive
Port<int> port = new Port<int>();

Arbiter.Activate(Environment.TaskQueue,
  Arbiter.MultipleItemReceive(
    true, port, 6,
    delegate(int[] array) {
      Console.WriteLine("Количество элементов: {0}", array.Length);
    }  
  )
);

//multiple port receive
PortSet<int, bool> portSet = new Port<int, bool>();

Arbiter.Activate(Environment.TaskQueue,
  Arbiter.MultipleItemReceive(
    portSet, 6,
    delegate(int[] arrayInt, bool[] arrayBool) {
      Console.WriteLine("Количество элементов: {0}, {1}", arrayInt.Length, arrayBool.Length);
    }  
  )
);

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

Установка нескольких ресиверов - задача interleave, который для разграничения ресурсов предоставляет 3 группы ресиверов:

• TearDown - группа содержит ресиверы, которые вызываются перед завершением работы Interleave;
• Concurrent - ресиверы, реализующие основной функционал сервиса. Ресиверы этой группы могут работать параллельно;
• Exclusive - одновременная работа ресиверов этой группы не допускается. Используется, например, для изменения состояния сервиса.

В созданный Interleave можно добавить ресиверы при помощи метода MainPortInterleav.CombineWith():

MainPortInterleave.CombineWith( new Interleave(
  new TeardownReceiverGroup(),
  new ExclusiveReceiverGroup(),
  new ConcurrentReceiverGroup()
  )
);

Диспетчеры и очереди диспетчеров позволяют создавать необходимое количество пулов потоков, назначать им приоритеты и политики планирования.
При создании DSS-сервиса автоматически создается диспетчер и очередь диспетчера по умолчанию. Количество потоков может быть задано при создании диспетчера, по умолчанию количество потоков соответствует количеству ядер центрального процессора, но не меньше двух.

Все потоки одного диспетчера имеют одинаковый приоритет. У диспетчера есть 5 уровней приоритета: Lowest, BelowNormal, Normal, AboveNormal, Highest.

Укажем некоторые часто используемые атрибуты диспетчера:

• WorkerThreadCount - количество потоков в пуле;
• ProcessedTaskCount - количество обработанных задач с момента создания диспетчера;
• PendingTaskCount - количество задач, находящихся в очереди.

Задачи добавляются в очередь диспетчера. Каждой очереди можно назначить одну из политик планирования, которые задаются перечислением TaskExecutionPolicy. Посмотрим на их краткое описание:

• Unconstrained - все задачи добавляются в очередь без ограничений (по умолчанию);
• ConstrainQueueDepthDiscardTask - игнорировать новые задачи после заполнения очереди;
• ConstrainQueueDepthThrottleExecution - потоки, добавляющие, задачи блокируются до тех пор, пока кол-во задач не уменьшится до указанного порога;
• ConstrainSchedulingRateDiscardTasks - игнорировать новые задачи, пока скорость их добавления превышает указанный порог;
• ConstrainSchedulingRateThrottleExecution - потоки, добавляющие задачи, блокируются, пока скорость добавления задач не снизится до допустимого значения.

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

При помощи итераторов достаточно легко реализовать исполнение последовательности операций. Однако иногда требуется запустить отдельное задание и дождаться его завершения. В таком случае можно использовать два подхода:

• создать порт, принимающий сообщение о завершении, и отправлять на него сообщение по завершению задания;
• использовать арбитр ExecuteToCompletion

Приведем пример кода для второго подхода.

private IEnumerator<ITask> Parent() {

  // сделать что-то

  Console.WriteLine(“Вызов дочернего задания”);
  
  yield return Arbiter.ExecuteToCompletion(
    taskQueue,
    new IterativeTask<double>(5.43, Child)
  );

  Console.WriteLine(“Дочернее задание завершено”);

  // продолжить исполнение

}

private IEnumerator<ITask> Child(double number) {

  // сделать что-то
  yield return Arbiter.xxx();

  // сделать еще что-то
  yield return Arbiter.xxx();
}

Термин scatter/gather обозначает массовую рассылку сообщений и затем ожидание получения всех ответов. С рассылкой сообщений вопросов возникнуть не должно, а для организации ожидания всех ответов используем получение набора элементов (multiple item receive) или получение на наборе портов (multiple port receive).

Иногда сообщения поступают гораздо быстрее чем их можно или нужно обрабатывать.

Стандартный подход для реализации задежки, то есть при помощи функции System.Threading.Thread.Sleep, в среде CCR использовать не рекомендуется. Вместо этого используется TimeoutPort().

void Wait(int millisec) {
AutoResetEvent signal = new AutoResetEvent(false);

  Activate(
    Arbiter.Receive(
      false,
      TimeoutPort(millisec),
      delegate(DateTime timeout) {
        // сигнал окончания таймаута
        signal.Set();
      }
    )
  );
  
  signal.WaitOne();
}

Для обработки ошибок можно воспользоваться стандартной стуктурой try/catch/finally, но в многопоточной среде может возникнуть ситуация, когда исключение, возникшее в одном потоке, должно быть передано обратно в родительский поток для обработки. Если во время обработки сообщения возникает исключение, которое не контролируется блоком try/catch, то DSS автоматически конвертирует это исключение в объект Fault и пересылает его обратно в качестве ответа. Для вывода информации о возникшей ошибке можно использовать метод LogError.

Обосновать использование causality вместо try\catch\finally - ошибка возникла в другом потоке, а обработать надо в месте вызова неперехваченный exception преобразуется в fault и отправляется обратно в ответе

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

private void Causality() {
  using (Dispatcher d = new Dispatcher()) {
    using (DispatcherQueue taskQ = new DispatcherQueue(“Causality Queue”, d)) {
      Port<Exception> ep = new Port<Exception>();
      Port<int> p = new Port<int>();

      // установка причинности для данного потока
      Dispatcher.AddCausality(new Causality(“Test”, ep));
      Console.WriteLine(“Main thread: {0}”,
      Thread.CurrentThread.ManagedThreadId);
      // установка ресивера, который будет генерировать исключение
      Arbiter.Activate(taskQ, Arbiter.Receive(false, p, CausalityExample));

      p.Post(2);
      Wait(500);
      Exception e;
      while (ep.Test(out e))
        Console.WriteLine(“Exception: “ + e.Message);
    }
  }
}
// ресивер с установленной причинностью
private void CausalityExample(int num) {
  Console.WriteLine(“CausalityExample thread: {0}”,
  Thread.CurrentThread.ManagedThreadId);
  Port<int> p = new Port<int>();
  // Set up another receiver
  Arbiter.Activate(Environment.TaskQueue,
  Arbiter.Receive(false, p,
  // And now an anonymous delegate
    delegate(int n) {
      Console.WriteLine(“Anonymous method thread: {0}”,
      Thread.CurrentThread.ManagedThreadId);
      int i = 0; n = n / i;
    }
  ));
  // отправить сообщение для активации ресивера
  p.Post(num);
}

Примеров еще недостаточно? Может вставить примеры в scatter/gather и «последнее сообщение» и на этом остановиться?

Децентрализованные програмные сервисы (DSS) отвечают за запуск и завершение служб и за управление потоком сообщений между службами через переадресованные порты служб. Сам DSS является набором служб: каталог запущенных служб, контроль доступа к файла, служба конфигураций и прочие.
DSS использует DSSP протокол, основанный на REST модели.
В этом разделе вы познакомитесь с основными понятиями DSS и научитесь создавать DSS-сервисы.

Составной элемент в MRDS - служба. Приложение создается объединением служб: так называемым взаимодействием или оркестровкой (orchestration). Создать среду слабо связанных служб - это и есть одна из основных задач DSS.

Составные части сервиса:

• Контракт - идентификатор службы + список принимаемых сообщений;
• Внутреннее состояние;
• Поведения - набор операций производимых службой, которые реализованы обработчиками;
• Контекст исполнения - взаимодействие с другими сервисами.

Сервисы могут создаваться и завершаться динамически. Однако, обычно они указываются в манифесте при запуске. Манифест - xml-файл конфигураций со списком сервисов и взаимодействий между ними.

Контракт (contract identifier) используется для однозначной идентификации различных сервисов. Каждый сервис должен включать в себя класс Contract, который в свою очередь должен содержать поле Identifier. В качестве значения поля Identifier MRDS использует URI следующего формата: http://host.domain/path/year/month/servicename.html. Каждый такой URI должен быть уникальным.
Программный интерфейс сервиса определяется как часть контакта через группу классов, которые описывают типы сообщений, принимаемы сервисом. Задача заключается в том, чтобы поддерживать этот интерфейс исключительно сообщениями без использования методов. В частности, должен быть хотябы одна публичная группа портов (PortSet), которая содержит порты для всех доступных типов сообщений.
Для того, чтобы указать, что класс является портом операций, используется атрибут [ServicePort()].

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

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

Контекст сервиса включает в себя всех партнеров. Партнером сервиса называется другой сервис, необходимый для работы первого. Ограничений на количество партнеров нет. Кроме того, партнерство может изменяться динамически.
Хотя зачастую партнерство задается декларативно через использование атрибутов в коде, в MRDS есть понятие обобщенного контакта. Этот тип контракта определяет операции, которые должен поддерживать конформный сервис. Во время исполнения, любой сервис удовлетворяющий обобщенному контракту может быть использован в качестве партнера. Связь между действительным и обобщенным сервисами указывается в манифесте - xml-файл определенного формата.

Модель сервисов DSSP определяет множество классов операций. Большинство из них вам вряд ли понадобится. Полная спецификация DSSP доступна по адресу: http://download.microsoft.com/download/5/6/B/56B49917-65E8-494A-BB8C-3D49850DAAC1/DSSP.pdf
Приведем полный список DSSP операций.

• Create - создает новый сервис;
• Delete - удаляет (часть) состояния сервиса;
• Drop - терминирует сервис: отправляет сервису запрос на терминацию;
• Get - получает копию состояния сервиса;
• Insert - добавляет информацию в состояние сервиса;
• Lookup - получает информацию о сервисе и его контексте;
• Query - аналогично Get, но с дополнительными параметрами, позволяющими структурированные запросы;
• Replace - полностью заменить состояние сервиса;
• Subscribe - запрашивает уведомление обо всех изменениях состояния;
• Submit - особая форма Update, которая может не изменять состояние;
• Update - получает информацию о сервисе и его контексте;
• Upsert - производит Update если информация о состоянии существует, в противном случае вызывается Insert.

Согласно спецификации DSSP только Lookup является обязательной операцией.

Среда исполнения DSS предоставляет упаковщик CreateService для операции Create.

Все операции основаны на общем классе DsspOperation:

public class DsspOperation<TBody, Tresponse>

Чтобы задать операцию сервиса, унаследуйте вашим классом один из стандартных классов DSSP-операций, предоставте тип запроса и группу портов (PortSet) для ответов. Приведем пример операции получения состояния ИК-сенсора.

[DisplayName(“GetSensors”)]
[Description(“Gets the state of the infrared sensors”)]
public class QueryInfraRed : Query<SensorsRequest, PortSet<Sensors, Fault>> { }

[Description(“Requests sensor data”)]
[DataContract]
public class SensorsRequest { }

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

[ServiceHandler {ServiceHandlerBehavior.Concurrent}]
public virtual IEnumerator<ITask> QueryInfraRedHandler(QueryInfraRed query) {
  query.ResponsePort.Post(_state.Sensors);
  yield break;
}

Атрибут [ServiceHandeler] указывает на то, что метод является обработчиком. В этом же атрубуте указывается тип interleave-группы: Concurrent. Параметр QueryInfraRed указывает, какой тип операций принимается обработчиком.

За предоставление среды исполнения для MRDS отвечает программа DssHost.exe. DssHost включает в себя реализацию веб-сервера, к которому вы можете обратиться через веб-браузер. Это позволяет через браузер проверять и управлять состояниям сервисов. При запуске DSS-узла некоторые сервисы стартуют автоматически. Вот некоторые из них:

• Console Output - принимает и фильтрует информационные сообщения и сообщнеия об ошибках;
• Constructor - создает новые экземпляры сервисов;
• Control Panel - предоставляет интерфейс для запуска и остановки сервисов вручную;
• Manifest Loader - интерпретатор манифеста;
• Mount Service - предоставляет доступ к локальной файловой системе;
• Embedded Resource Manager - отображает иконки, растровые изображения, XSLT-файлы и пр.;
• Security Manager - управляет безопасностью DSS узла;
• Service Directory - поддерживает список доступных на узле сервисов;
• State Partner Service - считывает файлы конфигураций при запуске сервисов.

Расположение директорий DSS-узла следующее:

• Bin
• Store
  • Logs
  • Media
  • Styles
  • Transforms

Это минимальная раскладка директорий, которая создается при развертывании среды исполнения MRDS. При помощи браузера в сервисе Service Directory вы можете узнать путь установки MRDS, который называется корневой директорией или точкой монтирования. Сервис точки монтирования (mountpoint service) может быть использован в вашей программе для доступа к локальным файлам. Однако при этом доступны будут только файлы в корневой дериктории MRDS. В программе путь к локальной установке DSS-узла можно получить при помощи статического класса LayoutPaths.

string logdir = LayoutPaths.RootDir + LayoutPaths.LogDir;

Откройте командную строку MRDS: Start → All Programs → Microsoft Robotics Studio → Command Prompt.
Для вывода доступных опций наберите:

> DssHost /?

Запуск DssHost без сервисов:

> DssHost /p:50000 /t:50001

• /p - HTTP порт (port), используемый для взаимодействия пользователя через браузер.
• /t - SOAP порт (tcpport); сервис принимает запросы (SOAP сообщения) через этот порт.

После запуска DssHost откройте браузер и перейдите по адресу http://localhost:50000. Вы должны оказаться на главной странице веб-интерфейса DSS.

Панель управления (control panel) отображает все доступные сервисы. Вы можете вручную запустить сервисы из этого списка. При большом количестве сервисов для удобства воспользуйтесь поиском. Перед запуском можно выбрать файл манифеста либо задать запуск без манифеста.

Каждый DSS-узел содержит каталок служб (service directory), который содержит список запущенных на данный момент сервисов. Для просмотра подробной информации по конкретному сервисы, нажмите на его ссылку - вы перейдете на страницу его описания.

На странце отладки выводятся информационные сообщения и сообщения об ошибках. Не путайте, сообщения, которые вы выводите при помощи Console.WriteLine не отображаются на этой странице. В разделе фильтров (Filters) вы можете задать необходимый уровень фильтрации сообщений - это аргумент в пользу LogInfo. Уровень фильтрации по умолчанию задается в файле конфигураций DSS: bin/dsshost.exe.config.

Эта страница указывает размещение сервисов на локальном жестком диске.

Менеджер безопасности (security manager) отображает содержимое файла store\SecuritySettings.xml, который при желании можно отредактировать. Путь к этому файлу может быть изменен. Указывается он в настройках DssHost, в файле bin\dsshost.exe.config.

<!-- Comment the line below to disable security -->
<add key=”Security” value=”..\store\SecuritySettings.xml”/>

Если файл SecuritySettings.xml отстутствует, то использоваться будут настройки по умолчанию.

Страница диагностики ресурсов (resource diagnostics) содержит информацию о диспетчерах (и их очередях) запущенных на DSS-узле. Эта информация полезна при диагностике проблем.

BoeBot (Running a Robot Service)

Базовой единицей вашего MRDS приложения является сервис. Давайте создадим сервис с нуля.
В папке mrds создаем (если еще не создана) папку projects - тут должны быть все Ваши проекты MRDS. В Visual Studio создаем новый проект через меню File → New → Project:

• выбираем шаблон «Simple DSS Service»;
• проверяем расположение проекта (…/mrds/pojects);
• указываем имя проекта (например ServiceEx).

В только что созданном проекте вы увидите:

• AssemblyInfo.cs - описание собираемого dll;
• ServiceEx.cs - основной исходный файл сервиса;
• ServiceEx.manifest.xml - файл манифеста, используемый DSS для загрузки зервиса;
• ServiceExTypes.cs - классы (или типы) используемые для взаимодействия с сервисом;
• остальные файлы можно оставить за кадром.

Основные свойства проекта по вкладкам:

• Application: Тип приложения любого сервиса должен быть задан как библиотека классов (DLL).
• Build: проверьте, что «output path» = ..\..\bin. Все сервисы MRDS помещаются в папку bin корневой папки MRDS.
• Build Events: комманды выполняемые до и после сборки.
• Reference Paths: в путях должны быть указаны 2 директории. Первая указывает на место расположения DLL-файлов сервисов и составляющих компонентов. Вторая директория расположение .NET Framework.

Так же, новый сервис можно создать через инструмент командной строки DssNewService. Идинственный параметр - название сервиса.

В этом разделе подробнее рассмотрим два файла исходного кода, которые генерируются при создании проекта: ServiceExTypes.cs, ServiceEx.cs. Чуть позже будет описан файл-манифест.

Начинаться файл будет примерно так:

namespace Robotics.ServiceEx {
  public sealed class Contract {
    public const String Identifier = "http://cs.mipt.ru/2010/12/serviceex.html";
  }

Пространство имен основано на имени сервиса. Класс Contract является неотъемлемой частью контракта, и он должен содержать строковую переменную Identifier.
Далее идет состояние сервиса с атрибутом [DataContract()]. Он необходим для того, чтобы класс был скопирован в Proxy DLL при компиляции.

[DataContract()]
public class ServiceExState { }

Следующий фрагмент кода описывает основной порт операций (PortSet) с обязательным атрибутом [ServicePort].

[ServicePort()]
public class ServiceExOperations : PortSet<DsspDefaultLookup, DsspDefaultDrop, Get> { }

Оканчивается файл определением типов запросов операций.

public class Get : Get<GetRequestType, PortSet<ServiceExState, Fault>> {
  public Get() { }
  public Get(Microsoft.Dss.ServiceModel.Dssp.GetRequestType body) : base(body) { }
  public Get(Microsoft.Dss.ServiceModel.Dssp.GetRequestType body,
    Microsoft.Ccr.Core.PortSet<ServiceExState,W3C.Soap.Fault> responsePort) :
    base(body, responsePort) { }
}

Для операций DsspDefaultLookup и DsspDefaultDrop определения в этом файле нет, поскольку они обрабатываются классом DsspServiceBase. При необходимости эти операции можно переопределить.
Обратите внимание, что Get использует два обобщенных типа: GetrequestType для типов запросов и PortSet<ServiceExState, Fault> для ответов.

Взглянем на основной файл реализации сервиса (ServiceEx.cs). Он содержит инициализацию и операции сервиса (или поведения).

[DisplayName(“ServiceEx”)]
[Description(“The ServiceEx Service”)]
[Contract(Contract.Identifier)]
public class ServiceExService : DsspServiceBase {

Назначение всех трех атрибутов понятно из названия: отображаемое имя (DisplayName), описание (Description), контракт (Contract).
Каждый сервис должен обладать экземпляром своего состояния (даже если оно пустое) и основной порт операций.

private ServiceExState _state = new ServiceExState();

[ServicePort(“/serviceex”, AllowMultipleInstances=false)]
private ServiceExOperations _mainPort = new ServiceExOperations();

Рекомендуем использовать общепринятые имена для этих переменных: _state, _mainPort.
Затем идут конструктор класса и метод Start, который инициализирует сервис при его запуске.

public ServiceExService(DsspServiceCreationPort creationPort) :
  base(creationPort) { }

protected override void Start() {
  base.Start();
  // инициализация сервиса
}

Следом идет обработчик, который должен быть помечен атрибутом [ServiceHandler()].

[ServiceHandler(ServiceHandlerBehavior.Concurrent)]
public virtual IEnumerator<ITask> GetHandler(Get get) {
  get.ResponsePort.Post(_state);
  yield break;
}

Перед компилирования сервиса вам необходимо задать ссылки на другие используемые сервисы: выбирается из списка dll-файлов (панель Solution Explorer → References).

Сервисы завязаны не на рабочие сервисы а на dll. Прокси-dll отвечают за сериализазию и десереализацию сообщений, передающихся по сети.

Для запуска сервиса в Visual Studio используйте кнопку F5 или Debug → Start Debugging.
При первом запуске нового сервиса перестраивается кэш директории контрактов. Кэш состоит из 2-х файлов в папке store: contractDirectory.state.xml, contractDirectoryCache.xml. Вы можете принудительно обновить кэш удалив эти два файла.
Затем загружается файл манифеста сервиса. Если при сосатвлении манифеста была допущена ошибка, то информация о ней появится в этот момент.

Понятие состояния сервиса подразумевает, что состояние должно содержать всю необходимую информацию, которую надо сохранить, чтобы после перезапуска сервиса иметь возможность продолжать работу с того же места. В простом варианте состояние представляется в виде xml-файла.
Задается состояние сервиса в классе ServiceEx.State файла ServiceExTypes.cs.

[DataContract()]
public class ServiceExState {
  [DataMember]
  public int field1;
  [DataMember]
  public int field2;
}

Обращение к состоянию сервиса внутри самого сервиса осуществляется через переменную _state: _state.field1.

Протокол DSSP определяет несколько способов изменения состояния сервиса, но широко используются только два из них: Replace и Update.

Replace поставляет цельное описание нового состояния. Необходимо определить класс Replace в файле типов сервиса.

public class Replace : Replace<xxxState, PortSet<DefaultReplaceResponseType, Fault>> {
  public Replace() { }
  public Replace(xxxState body) : base(body) { }
}

Пример обработчика Replace:

[ServiceHandler(ServiceHandlerBehavior.Exclusive)]
public IEnumerator<ITask> ReplaceHandler(Replace replace) {
  _state = replace.Body;
    replace.ResponsePort.Post(DefaultReplaceResponseType.Instance);
    yield break;
}

Во избежание возможных конфликтов обработчик следует объявлять в группе Exclusive. Когда обработчик заканчивает работу, он отправляет уведомление на порт ResponsePort. Для этого используется предопределенный тип DefaultReplaceResponseType.Instance - вам не надо создавать тип для ответа.

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

Определите класс для обновления состояния в файле ServiceExType.cs:

public class SetInterval: Update<SetIntervalRequest, PortSet<DefaultUpdateResponseType, Fault>> {
  public SetInterval() { }
}

[DataContract]
[DataMemberConstructor]
public class SetIntervalRequest {
  [DataMember, DataMemberConstructor]
  public int Interval;
}

Класс SetInterval расширяет класс Update, использует тип запроса SetIntervalRequest и в случае успеха возвращает сообщение типа DefaultUpdateResponseType. Обратите внимание на атрибуты DataContract и DataMemberConstructor, благодаря которым можно вызывать эту операцию через другой сервис: _serviceExPort.SetInterval(2000). Предполагается, что _serviceExPort указывает на ServiceEx.

Чтобы установить обработчик на операцию SetInterval, добавим соответствующий код в файл ServiceEx.cs.

[ServiceHandler(ServiceHandlerBehavior.Exclusive)]
public virtual IEnumerator<ITask> SetIntervalHandler(SetInterval request) {
  public virtual IEnumerator<ITask> SetIntervalHandler(SetInterval request) {
    if (_state == null) {
      request.ResponsePort.Post(new Fault());
    } else {
      _state.Interval = request.Body.Interval;
      request.ResponsePort.Post(DefaultUpdateResponseType.Instance);
    }
  }
  yield break;
}

Этот обработчик обновляет _state.Interval из тела запроса и отправляет оповещение используя один из предопределенных типов ответов.
Также GetHandler сервиса ServiceEx должен корректно обрабатывать случай, когда переменная state не определена.
В случае ошибок необходимо отправить Fault через порт ответа. Можно строкой указать сообщение об ошибке:

Fault fault = Fault.FromCodeSubcodeReason(FaultCodes.Receiver, 
    DsspFaultCodes.OperationFailed, "Тут сообщение об ошибке");

Каждый сервис содержит метод base.Start, который делает следующее:

• активирует обработчиков DSSP операций
• добавляет севис в каталог сервисов
• выполняет LogInfo с URI сервиса

Переопределите метод Start и добавьте в него код инициализации.

protected override void Start() {
  base.Start();
  // Тут добавьте свой код инициализации.
}

Нет необходимости при инициализации указывать обработчиков: DSS находит и добавляет их автоматически при помощи атрибута [ServiceHandler] и сигнатуры метода. Если понадобится добавить обработчиков в основной interleave после запуска сервиса, используйте метод Arbiter.CombineWith.

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

• порт переадресации сервиса
• использование порта завершения для синхронизации

В *.cs файле сервиса добавьте обработчик запроса на остановку:

[ServiceHandler(ServiceHandlerBehavior.Teardown)]
public virtual IEnumerator<ITask> DropHandler(DsspDefaultDrop drop) {
  _shutdown = true;
  Console.WriteLine("Остановка сервиса");
  base.DefaultDropHandler(drop);
  yield break;
}

Теперь другой сервис может послать запрос на остановку этого сервиса следующим образом:

  _serviceExPort.DsspDefaultDrop();

Часто удобнее не создавать сервис явно, а указать сервис-партнера декларативно, используя атрибут [Parter]. Отредактируйте *.cs файл сервиса, которому хотите добавить партнера:

// Партнерство с ServiceEx
[Partner("ServiceEx", Contract = serviceEx.Contract.Identifier,
  CreationPolicy = PartnerCreationPolicy.CreateAlways, Optional = false)]
  privatge serviceEx.ServiceExOperations _serviceExPort = 
    new serviceEx.ServiceExOperations();

Обязательно надо укать атрибут в объявлении _serviceExPort, в котором аргумент «ServiceEx» является обязательным.
Доступны четыре политики создания сервиса:

• CreateAlways (всегда создавать)
• UseExistingOrCreate (использовать существующий или создать)
• UseExisting (использовать существующий)
• UsePartnerListEntry (использовать элемент списка партнера)

FindPartner

Манифест представляет из себя xml-документ, который описывает запускаемый сервис с его партнертнерствами. При запуске манифест передается в DssHost.
Простой пример манифеста:

<?xml version="1.0"?>
<Manifest
  xmlns="http://schemas.microsoft.com/xw/2004/10/manifest.html"
  xmlns:dssp="http://schemas.microsoft.com/xw/2004/10/dssp.html">

  <CreateServiceList>
    <ServiceRecordType>
      <dssp:Contract>
        http://schemas.tempuri.org/2010/12/serviceex.html
      </dssp:Contract>
    </ServiceRecordType>
  </CreateServiceList>
</Manifest>

Редактировать манифест можно и вручную, но рекомендуется использовать для этого редактор манифеста: DSS Manifest Editor.
При создании нового сервиса манифест создается автоматически. Рассмотрим подробнее некоторые его фрагменты.

Можно скопировать ServiceRecord из манифеста партнера и вставить в манифест основного сервиса. Теперь можно использовать политику UseExisting, т.к. сервис-партнер будет запущен при старте dss-узла.
Обратите внимание, что идентификаторы контрактов чувствительны к регистру.

Можно использовать политику явного указания партнеров (UsePartnerListEntry). В таком случае в манифесте в теге PartnerList необходимо перечислить партнеров.

Два способа получения данных:

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

Сервисы позволяющие подписку должны реализовать операцию Subscribe. Подписка осуществляется следующим образом:

• Установите партнерство (декларативно или динамически)
• Отправьте сообщение Subscribe, указав необходимые параметры
• Установите получателей и соответствующих обработчиков для обработки оповещений

(подробнее о каждом пункте с кусками кода)

Для отписки отправьте сообщение Shutdown на порт, который Вы указали при подписке. Дождитесь ответа, чтобы убедиться, что отписка прошла успешно. (пример кода) ?! Не понял куда этот метод вставлять. Кто такой TeleOperation ?!