Wielowątkowość: przykład modelu aktor

W ostatnim wpisie przedstawiłem zasadę działania modelu aktor. Zachęcam do przeczytania poprzedniego wpisu ponieważ dzisiaj skupię się na przykładzie, a nie podstawach teoretycznych. Jeśli poprzedni wpis nie był do końca zrozumiały, zachęcam do przeanalizowania przykładu z tego wpisu i potem powrócenia do poprzedniego postu – wtedy myślę, że wiele zagadnień będzie prostsze w zrozumieniu.

Poniższe przykłady należy stanowić jako pseudokod. Stanowią one szkic wzorca aktor, a nie jego implementację. Do implementacji w kolejnych wpisach będę używał akka.net, ale moim zdaniem najważniejsze jest zrozumienie zasad, a nie nauczenie się kolejnego framework’a. Z tego względu, bardziej będę skupiał się na rozwiązywaniu różnych problemów wielowątkowych (np.problem ucztujących filozofów), a nie dokumentacji API.

Załóżmy, że chcemy rozwiązać klasyczny problem przelewu pieniędzy z jednego konta na drugie. W najprostszej postaci, będziemy mieli następującą klasę:

     class BankAccount
     {
         private int _Balance;


         public void Deposit(int amount)
         {
             _Balance += amount;
         }

         public void Withdraw(int amount)
         {
             if (amount <= _Balance)
                 _Balance -= amount;
             else
                 throw new ArgumentOutOfRangeException();
         }
     }

Oczywiście powyższy kod nie jest thread-safe, dlatego należy użyć blokady:

     class BankAccount
     {
         private int _Balance;
         private object _sync=new object();

         public void Deposit(int amount)
         {
             lock(_sync)
             {
               _Balance += amount;
             }
         }

         public void Withdraw(int amount)
         {
             lock(_sync)
             {
                 if (amount <= _Balance)
                    _Balance -= amount;
                 else
                    throw new ArgumentOutOfRangeException();
             } 
         }
     }

Kolejne zadanie to przetransferowanie pieniędzy z jednego konta na drugie. Klasyczne, błędne rozwiązanie to:

lock(accountA)
{
   lock(accountB)
   {
        accountA.Withdraw(5);
        accountB.Deposit(5);
   }
}

Oczywiście powyższy kod zakończy się deadlock, jeśli w tym samym czasie będziemy chcieli przelać pieniądze z konta A do B oraz z B do A.  Prawidłowe rozwiązanie to np. sortowanie blokad, przedstawione tutaj.

Wróćmy jednak do wzorca aktor. Stanowi on po prostu wyższy poziom abstrakcji dla wątków. Z poprzedniego wpisu wiemy, że aktorzy komunikują się za pomocą wiadomości, tak jak np. instancje w systemie kolejkowym. Zdefiniujmy zatem dwie wiadomości, dla depozytu i wycofywania środków:

     public class DepositMessage
     {
         public int Amount { get; }

         public DepositMessage(int amount)
         {
             Amount = amount;
         }
     }

     public class WithdrawMessage
     {
         public int Amount { get; }

         public WithdrawMessage(int amount)
         {
             Amount = amount;
         }
     }

Proszę zauważyć, że są one immutable – zawsze chcemy uniknąć współdzielenia stanu między różnymi aktorami. Następnie aktor, będzie obsługiwał wiadomości w sposób asynchroniczny:

     class BankAccountActor
     {
         private int _balance;

         public void OnReceive(object message)
         {
             if (message is WithdrawMessage)
             {
                 var withdrawMessage = ((WithdrawMessage)message);
                 if (withdrawMessage.Amount <= _balance)
                     _balance -= withdrawMessage.Amount;
             }

             if (message is DepositMessage)
             {
                 _balance += ((DepositMessage)message).Amount;
             }
         }
     }

Metoda OnReceive będzie wywoływana przez framework, w momencie otrzymania konkretnej wiadomości. Jak wspomniałem, przypomina to klasyczny system kolejkowy, ale OnReceive zawsze MUSI być wykonywane jedno po drugim. Jeśli dwie wiadomości przyjdą w tym samym czasie, mamy zagwarantowane, że OnReceive nie będzie wykonywane równocześnie z dwóch różnych wątków. Obsługa zatem może wyglądać następująco:

while(true)
{
    var message = blockingCollection.Dequeue();
    actor.OnReceive(message);
}

Z tego względu, nie musimy umieszczać w tych metodach żadnych blokad (brak współdzielonego stanu).
Następnie chcemy mieć możliwość transferu środków z jednego konta do drugiego. Zdefiniujmy zatem kolejną wiadomość:

     class TransferMessage
     {
         public string From { get; }
         public string To { get; }
         public int Amount { get; }

         public TransferMessage(string from, string to, int amount)
         {
             From = @from;
             To = to;
             Amount = amount;
         }
     }

Aktorzy mogą tworzyć hierarchie, w której jeden aktor zarządza kolejnymi. W naszym przypadku będziemy mieli dwa typy aktorów: TransferMoneyActor oraz BankAccountActor. Pierwszy z nich służy do koordynowania przepływu środków.

Najpierw implementujemy obsługę wiadomości TransferMessage:

     class TransferActor
     {
         public void OnTransferMessageReceived(TransferMessage transferMessage)
         {
             ActorsSystem.GetActor(transferMessage.From).Send(new WithdrawMessage(transferMessage.Amount));
             
             Context.Become(AwaitFrom(transferMessage.From,transferMessage.To,transferMessage.Amount));
         }

W momencie otrzymania TransferMessage, zostanie wysłana wiadomość do aktora, który reprezentuje konto nadawcy. Pamiętajmy, że wszystkie operacje są asynchroniczne, zatem stanowią model “fire&forget”. TransferActor jednak musi dowiedzieć się, czy środki zostały prawidłowo zdjęte z konta nadawcy. Z tego względu, jedną z bardzo ważnych właściwości aktorów jest zmiana kontekstu. W powyższym przykładzie chcemy zmienić kontekst w tryb oczekiwania na odpowiedź od nadawcy. Służy zwykle do tego metoda “Become”. Aktor zatem staje się aktorem oczekującym na odpowiedź od nadawcy. Odpowiedź przyjdzie oczywiście w formie kolejnej wiadomości:

     class MoneyWithdrawn
     {
         public ActorRef ActorRef { get;  }
         public int Amount { get;  }

         public MoneyWithdrawn(ActorRef actorRef,int amount)
         {
             ActorRef = actorRef;
             Amount = amount;
         }
     }

Następnie w momencie potwierdzenia wycofania pieniędzy, możemy wysłać wiadomość w celu umieszczenia środków na innym koncie:

     class TransferActor
     {
         public void OnTransferMessageReceived(TransferMessage transferMessage)
         {
             ActorsSystem.GetActor(transferMessage.From).Send(new WithdrawMessage(transferMessage.Amount));
             
             Context.Become(AwaitFrom(transferMessage.From,transferMessage.To,transferMessage.Amount));
         }

         public void AwaitFrom(string from, string to, int amount)
         {
             ActorsSystem.GetActor(to).Send(new DepositMessage(amount));
             Context.Became(AwaitTo(transferMessage.From, transferMessage.To, transferMessage.Amount));
         }

Analogicznie, aktor przechodzi w kolejny stan, oczekiwania na potwierdzenie złożenia depozytu. Potwierdzenie przyjdzie w formie kolejnej wiadomości:

     class TransferActor
     {
         public void OnTransferMessageReceived(TransferMessage transferMessage)
         {
            ActorsSystem.GetActor(transferMessage.From).Send(new WithdrawMessage(transferMessage.Amount));
             
             Context.Become(AwaitFrom(transferMessage.From,transferMessage.To,transferMessage.Amount));
         }

         public void AwaitFrom(string from, string to, int amount)
         {
             ActorsSystem.GetActor(to).Send(new DepositMessage(amount));
             Context.Became(AwaitTo(transferMessage.From, transferMessage.To, transferMessage.Amount));
         }

         public void AwaitTo(string from, string to, int amount)
         {
             Context.Finished();
         }

Widzimy, że każda operacja jest atomowa (pod warunkiem, że przetwarzanie wiadomości nie jest współbieżne). To bardzo ważna cecha systemów opartych na aktorach – należy rozszerzać hierarchie o tyle poziomów, aby konkretne zadanie było łatwe w implementacji. Przez “łatwe” mam na myśli sytuację, w której nie musimy korzystać z blokad.

Model dla prostych problemów (takich jak powyższy) jest moim zdaniem zła praktyką i przykładem over-engineering’u. Dla bardziej skomplikowanych problemów, znacząco to ułatwia zapobiegnięcie zakleszczeniom. Tak jak wspomniałem, aktor to pewien poziom abstrakcji. Ta abstrakcja daje nam ogromne możliwości skalowania – od problemu rozwiązywanego współbieżnie na np. 4 procesorach do środowiska opartego na wielu komputerach połączonych w sieć. Jeśli aktor jest abstrakcyjny, nic nie stoi na przykładzie, aby umieścić go na osobnym komputerze i przesyłać wiadomości za pomocą TCP. Jak widać, można skalować rozwiązanie od jednego procesu po wiele usług webowych komunikujących się dowolnymi sposobami (HTTP, systemy kolejkowy, TCP itp.). Użycie prostej blokady jest dobre, ale nie posiada żadnej abstrakcji – ogranicza nas do jednego procesu.

Wyświetl pełny artykuł