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

Question

У меня есть служба, которая выполняет итерации над X-актерами, прося их о своем состоянии, используя ActorProxy.

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

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

class Actor : IMyActor{ 

public Task<MyState> GetState() => StateManager.GetAsync("key") 
} 

альтернативный вариант.

Каков надлежащий способ формирования сервиса для звонка, и если он не отвечает в течение 5 секунд, просто скройте.

var proxy = ActorProxy.Create<IMyActor(); 
var state = await proxy.GetState(); // this will wait until the actor is ready to send back the state. 

Answer 1

Можно прочитать состояние актера даже для Актеров, которые в настоящее время выполняют метод блокировки. Актеры хранят свое состояние, используя IActorStateManager, который, в свою очередь, использует IActorStateProvider. IActorStateProvider создается один раз за ActorService. Каждый раздел создает экземпляр ActorService, который отвечает за хостинг и запуск участников. Актерская служба находится в ядре StatefulService (или, скорее, StatefulServiceBase, который является базовым классом, который использует постоянный сервис с сохранением состояния). Имея это в виду, мы можем работать с ActorService, который подходит нашим Актерам так же, как мы будем работать с регулярной услугой, то есть с сервисным интерфейсом на основе IService.

IActorStateProvider (Осуществляется KvsActorStateProvider, если вы используете сохранялось состояние) имеет два метода, которые мы можем использовать:

Task<T> LoadStateAsync<T>(ActorId actorId, string stateName, CancellationToken cancellationToken = null); 
Task<PagedResult<ActorId>> GetActorsAsync(int numItemsToReturn, ContinuationToken continuationToken, CancellationToken cancellationToken); 

Вызовы этих методов не влияет на замки актеров, которые имеет смысл, так как они предназначенный для поддержки всех участников раздела.

Пример:

Создание пользовательских ActorService и использовать это один для размещения своих актеров:

public interface IManyfoldActorService : IService 
{ 
    Task<IDictionary<long, int>> GetCountsAsync(CancellationToken cancellationToken); 
} 

public class ManyfoldActorService : ActorService, IManyfoldActorService 
{ 
    ... 
} 

Регистрация нового ActorService в Program.Main:

ActorRuntime.RegisterActorAsync<ManyfoldActor>(
    (context, actorType) => new ManyfoldActorService(context, actorType)).GetAwaiter().GetResult(); 

Предположим, что мы имеем простой Актер со следующим методом:

Task IManyfoldActor.SetCountAsync(int count, CancellationToken cancellationToken) 
    { 
     Task.Delay(TimeSpan.FromSeconds(30), cancellationToken).GetAwaiter().GetResult(); 
     var task = this.StateManager.SetStateAsync("count", count, cancellationToken); 
     ActorEventSource.Current.ActorMessage(this, $"Finished set {count} on {this.Id.GetLongId()}"); 
     return task; 
    } 

Он ждет в течение 30 секунд (для имитации длинного хода, блокирования, вызовов методов), а затем установить значение состояния "count" к int.

В отдельную услугу теперь мы можем назвать SetCountAsync для актеров, чтобы сгенерировать данные состояния:

protected override async Task RunAsync(CancellationToken cancellationToken) 
    { 
     var actorProxyFactory = new ActorProxyFactory(); 
     long iterations = 0; 
     while (true) 
     { 
      cancellationToken.ThrowIfCancellationRequested(); 
      iterations += 1; 
      var actorId = iterations % 10; 
      var count = Environment.TickCount % 100; 
      var manyfoldActor = actorProxyFactory.CreateActorProxy<IManyfoldActor>(new ActorId(actorId)); 
      manyfoldActor.SetCountAsync(count, cancellationToken).ConfigureAwait(false); 
      ServiceEventSource.Current.ServiceMessage(this.Context, $"Set count {count} on {actorId} @ {iterations}"); 
      await Task.Delay(TimeSpan.FromSeconds(3), cancellationToken); 
     } 
    } 

Этот метод просто петли бесконечно изменяя значения актеров. (Обратите внимание на корреляцию между 10 участниками, задержкой на 3 секунды и задержкой 30 секунд в актере. Просто разработан таким образом, чтобы предотвратить бесконечное нарастание вызовов Actor, ожидающих блокировки). Каждый вызов также выполняется как «огонь-и-забыть», поэтому мы можем продолжить обновление состояния следующего актера до его возвращения. Это глупая часть кода, она просто разработана таким образом, чтобы доказать теорию.

В настоящее время в службе актера мы можем реализовать метод GetCountsAsync так:

public async Task<IDictionary<long, int>> GetCountsAsync(CancellationToken cancellationToken) 
    { 
     ContinuationToken continuationToken = null; 
     var actors = new Dictionary<long, int>(); 

     do 
     { 
      var page = await this.StateProvider.GetActorsAsync(100, continuationToken, cancellationToken); 

      foreach (var actor in page.Items) 
      { 
       var count = await this.StateProvider.LoadStateAsync<int>(actor, "count", cancellationToken); 
       actors.Add(actor.GetLongId(), count); 
      } 

      continuationToken = page.ContinuationToken; 
     } 
     while (continuationToken != null); 

     return actors; 
    } 

Это использует основной ActorStateProvider для запроса всех известных актеров (для этого раздела), а затем непосредственно считывает состояние для каждого это путь «в обход» Актера и не блокируется исполнением действия актера.

Окончательный кусок, какой-то способ, который может позвонить в наш ActorService и вызвать GetCountsAsync по всем разделам:

public IDictionary<long, int> Get() 
    { 
     var applicationName = FabricRuntime.GetActivationContext().ApplicationName; 
     var actorServiceName = $"{typeof(IManyfoldActorService).Name.Substring(1)}"; 
     var actorServiceUri = new Uri($"{applicationName}/{actorServiceName}"); 

     var fabricClient = new FabricClient(); 
     var partitions = new List<long>(); 
     var servicePartitionList = fabricClient.QueryManager.GetPartitionListAsync(actorServiceUri).GetAwaiter().GetResult(); 
     foreach (var servicePartition in servicePartitionList) 
     { 
      var partitionInformation = servicePartition.PartitionInformation as Int64RangePartitionInformation; 
      partitions.Add(partitionInformation.LowKey); 
     } 

     var serviceProxyFactory = new ServiceProxyFactory(); 

     var actors = new Dictionary<long, int>(); 
     foreach (var partition in partitions) 
     { 
      var actorService = serviceProxyFactory.CreateServiceProxy<IManyfoldActorService>(actorServiceUri, new ServicePartitionKey(partition)); 

      var counts = actorService.GetCountsAsync(CancellationToken.None).GetAwaiter().GetResult(); 
      foreach (var count in counts) 
      { 
       actors.Add(count.Key, count.Value); 
      } 
     } 
     return actors; 
    } 

Идущие этот код теперь будет давать нам 10 актеров, что каждый 33: d второй получает ее состояние обновляется и где каждый актер занят в течение 30 секунд каждый раз. Служба Actor видит обновленное состояние, как только возвращается каждый метод актера.

В этом примере некоторые вещи опущены, например, при загрузке состояния в службе актера мы, вероятно, должны защищать от тайм-аутов.

Answer 2

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