Upgrade to Pro — share decks privately, control downloads, hide ads and more …

Phluxorでアクターモデルを 理解・体験しよう / toolkit-for-flexibl...

Phluxorでアクターモデルを 理解・体験しよう / toolkit-for-flexible-actor-models-in-php-phluxor

A toolkit for flexible actor models in PHP, empowering the PHP ecosystem.
https://github.com/ytake/phluxor

yuuki takezawa

June 24, 2024
Tweet

More Decks by yuuki takezawa

Other Decks in Technology

Transcript

  1. Pro fi le • ஛ᖒ ༗و a.k.a ytake • ઍגࣜձࣾ

    CTO / ٕज़ސ໰ ωοτϓϩςΫγϣϯζ΄͔ • Go / Scala • ΞΫλʔϞσϧେ޷͖
  2. Phluxor • ScalaͳͲͰ͓ͳ͡ΈͷAkka / Pekko • Go΍.NETͰ࢖ΘΕ͍ͯΔ Proto.Actor • ͜ΕΒͱಉ֓͡೦ʹج͖ͮ

    PHPͰΞΫλʔϞσϧΛ࣮ݱ͢ΔͨΊͷ΋ͷ͕Phluxor • ΫϥελɾҰ෦ϧʔλʔػೳ͸ݱࡏ࣮૷தʢ਺ϲ݄Ҏ಺ʹఏڙʣ
  3. Phluxor͕αϙʔτ͍ͯ͠Δ΋ͷ • ΞΫλʔͷجຊཁૉશൠ • EventStream • Supervision • Future (Async/AwaitΈ͍ͨͳ΋ͷ)

    • Behavior • Persistence • Router • DeadLetter • Middleware • Reentrancy
  4. readonly class BootAppActor { public function __construct( private ContainerInterface $container

    ) { } public function __invoke(WorkerStartEvent $event): void { $system = ActorSystem::create(); $spawned = $system->root()->spawnNamed( Props::fromProducer(fn() => new BoxOffice()), AppActor::NAME ); if ($this->container instanceof ServiceManager) { $this->container->setService(AppActor::class, new AppActor($system->root(), $spawned->getRef())); return; } throw new RuntimeException('Container is not a ServiceManager'); } }
  5. readonly class BootAppActor { public function __construct( private ContainerInterface $container

    ) { } public function __invoke(WorkerStartEvent $event): void { $system = ActorSystem::create(); $spawned = $system->root()->spawnNamed( Props::fromProducer(fn() => new BoxOffice()), AppActor::NAME ); if ($this->container instanceof ServiceManager) { $this->container->setService(AppActor::class, new AppActor($system->root(), $spawned->getRef())); return; } throw new RuntimeException('Container is not a ServiceManager'); } } NF[[JPTXPPMFΛτϦΨʔʹ ΞΫλʔγεςϜىಈ ϦΫΤετͷͨͼʹΞΫλʔγεςϜΛ ىಈ͢Δͱɺ ߴίετͰঢ়ଶҡ͕࣋Ͱ͖ͳ͍ͨΊ
  6. readonly class BootAppActor { public function __construct( private ContainerInterface $container

    ) { } public function __invoke(WorkerStartEvent $event): void { $system = ActorSystem::create(); $spawned = $system->root()->spawnNamed( Props::fromProducer(fn() => new BoxOffice()), AppActor::NAME ); if ($this->container instanceof ServiceManager) { $this->container->setService(AppActor::class, new AppActor($system->root(), $spawned->getRef())); return; } throw new RuntimeException('Container is not a ServiceManager'); } } SPPU͸ΞΫλʔγεςϜτοϓϨϕϧ ઃܭʹΑͬͯ͸ τοϓϨϕϧΞΫλʔΛෳ਺࡞ͬͯ΋0,
  7. readonly class BootAppActor { public function __construct( private ContainerInterface $container

    ) { } public function __invoke(WorkerStartEvent $event): void { $system = ActorSystem::create(); $spawned = $system->root()->spawnNamed( Props::fromProducer(fn() => new BoxOffice()), AppActor::NAME ); if ($this->container instanceof ServiceManager) { $this->container->setService(AppActor::class, new AppActor($system->root(), $spawned->getRef())); return; } throw new RuntimeException('Container is not a ServiceManager'); } } ΞΫλʔੜ੒࣌ͷઃఆ ʢͳʹ΋ͳ͍ͱ͖͸ͨͩͷΞΫλʔʣ
  8. readonly class BootAppActor { public function __construct( private ContainerInterface $container

    ) { } public function __invoke(WorkerStartEvent $event): void { $system = ActorSystem::create(); $spawned = $system->root()->spawnNamed( Props::fromProducer(fn() => new BoxOffice()), AppActor::NAME ); if ($this->container instanceof ServiceManager) { $this->container->setService(AppActor::class, new AppActor($system->root(), $spawned->getRef())); return; } throw new RuntimeException('Container is not a ServiceManager'); } } ΞΫλʔੜ੒࣌ʹ໊લΛ͚ͭΔ 4QBXO/BNFE
  9. readonly class BootAppActor { public function __construct( private ContainerInterface $container

    ) { } public function __invoke(WorkerStartEvent $event): void { $system = ActorSystem::create(); $spawned = $system->root()->spawnNamed( Props::fromProducer(fn() => new BoxOffice()), AppActor::NAME ); if ($this->container instanceof ServiceManager) { $this->container->setService(AppActor::class, new AppActor($system->root(), $spawned->getRef())); return; } throw new RuntimeException('Container is not a ServiceManager'); } } #PY0 ff i DFͷΞΫλʔΛ CPY@P ff i DFͱ͍͏໊લͰ τοϓϨϕϧʹ഑ஔ
  10. public function handle(ServerRequestInterface $request): ResponseInterface { $actor = $this->appActor; $post

    = $request->getParsedBody(); if (! isset($post['tickets'])) { return new JsonResponse(['error' => 'tickets is required'], 400); } $eventName = $request->getAttribute('name'); $future = $actor->root->requestFuture( $actor->actorRef, new EventDescription($eventName, (int) $post['tickets']), 2000 ); $fr = $future->result(); if ($fr->error() !== null) { return new JsonResponse(['error' => $fr->error()], 400); } $v = $fr->value(); return match (true) { $v instanceof EventCreated => new JsonResponse(['message' => 'event created', 'event' => $eventName]), $v instanceof EventExists => new JsonResponse(['message' => 'event exists', 'event' => $eventName], 409), default => new JsonResponse(['message' => 'unknown'], 400), }; }
  11. public function handle(ServerRequestInterface $request): ResponseInterface { $actor = $this->appActor; $post

    = $request->getParsedBody(); if (! isset($post['tickets'])) { return new JsonResponse(['error' => 'tickets is required'], 400); } $eventName = $request->getAttribute('name'); $future = $actor->root->requestFuture( $actor->actorRef, new EventDescription($eventName, (int) $post['tickets']), 2000 ); $fr = $future->result(); if ($fr->error() !== null) { return new JsonResponse(['error' => $fr->error()], 400); } $v = $fr->value(); return match (true) { $v instanceof EventCreated => new JsonResponse(['message' => 'event created', 'event' => $eventName]), $v instanceof EventExists => new JsonResponse(['message' => 'event exists', 'event' => $eventName], 409), default => new JsonResponse(['message' => 'unknown'], 400), }; } τοϓϨϕϧΞΫλʔʹ νέοτ࡞੒Λґཔ
  12. public function handle(ServerRequestInterface $request): ResponseInterface { $actor = $this->appActor; $post

    = $request->getParsedBody(); if (! isset($post['tickets'])) { return new JsonResponse(['error' => 'tickets is required'], 400); } $eventName = $request->getAttribute('name'); $future = $actor->root->requestFuture( $actor->actorRef, new EventDescription($eventName, (int) $post['tickets']), 2000 ); $fr = $future->result(); if ($fr->error() !== null) { return new JsonResponse(['error' => $fr->error()], 400); } $v = $fr->value(); return match (true) { $v instanceof EventCreated => new JsonResponse(['message' => 'event created', 'event' => $eventName]), $v instanceof EventExists => new JsonResponse(['message' => 'event exists', 'event' => $eventName], 409), default => new JsonResponse(['message' => 'unknown'], 400), }; } SFRVFTU'VUVSF͸ BTZODBXBJUͷΑ͏ʹ ޙʹૹΒΕͯ͘Δ஋͕ར༻Ͱ͖Δ
  13. public function handle(ServerRequestInterface $request): ResponseInterface { $actor = $this->appActor; $post

    = $request->getParsedBody(); if (! isset($post['tickets'])) { return new JsonResponse(['error' => 'tickets is required'], 400); } $eventName = $request->getAttribute('name'); $future = $actor->root->requestFuture( $actor->actorRef, new EventDescription($eventName, (int) $post['tickets']), 2000 ); $fr = $future->result(); if ($fr->error() !== null) { return new JsonResponse(['error' => $fr->error()], 400); } $v = $fr->value(); return match (true) { $v instanceof EventCreated => new JsonResponse(['message' => 'event created', 'event' => $eventName]), $v instanceof EventExists => new JsonResponse(['message' => 'event exists', 'event' => $eventName], 409), default => new JsonResponse(['message' => 'unknown'], 400), }; } GVUVSFͷ݁Ռ
  14. public function receive(ContextInterface $context): void { $msg = $context->message(); switch

    (true) { case $msg instanceof EventDescription: $result = $context->spawnNamed( Props::fromProducer(fn() => new TicketSeller()), $msg->name ); if ($result->isError() instanceof NameExistsException) { // ΞΫλʔ͕ੜ੒ࡁΈͷ৔߹͸ੜ੒ࡁΈͰ͋Δ͜ͱΛ௨஌͠·͢ $context->respond(new EventExists()); break; } $context->send($result->getRef(), new Add($msg->name, $msg->tickets, $context->sender())); break;
  15. public function receive(ContextInterface $context): void { $msg = $context->message(); switch

    (true) { case $msg instanceof EventDescription: $result = $context->spawnNamed( Props::fromProducer(fn() => new TicketSeller()), $msg->name ); if ($result->isError() instanceof NameExistsException) { // ΞΫλʔ͕ੜ੒ࡁΈͷ৔߹͸ੜ੒ࡁΈͰ͋Δ͜ͱΛ௨஌͠·͢ $context->respond(new EventExists()); break; } $context->send($result->getRef(), new Add($msg->name, $msg->tickets, $context->sender())); break; #PY0 ff i DFΞΫλʔ SFDFJWFϝιουͷΈ࣮૷
  16. public function receive(ContextInterface $context): void { $msg = $context->message(); switch

    (true) { case $msg instanceof EventDescription: $result = $context->spawnNamed( Props::fromProducer(fn() => new TicketSeller()), $msg->name ); if ($result->isError() instanceof NameExistsException) { // ΞΫλʔ͕ੜ੒ࡁΈͷ৔߹͸ੜ੒ࡁΈͰ͋Δ͜ͱΛ௨஌͠·͢ $context->respond(new EventExists()); break; } $context->send($result->getRef(), new Add($msg->name, $msg->tickets, $context->sender())); break; ྲྀΕͯ͘Δϝοηʔδ͸ DPOUFYUNFTTBHF Ͱऔಘ
  17. public function receive(ContextInterface $context): void { $msg = $context->message(); switch

    (true) { case $msg instanceof EventDescription: $result = $context->spawnNamed( Props::fromProducer(fn() => new TicketSeller()), $msg->name ); if ($result->isError() instanceof NameExistsException) { // ΞΫλʔ͕ੜ੒ࡁΈͷ৔߹͸ੜ੒ࡁΈͰ͋Δ͜ͱΛ௨஌͠·͢ $context->respond(new EventExists()); break; } $context->send($result->getRef(), new Add($msg->name, $msg->tickets, $context->sender())); break; &WFOU%FTDSJQUJPOϝοηʔδ΁ͷ ରԠΛهड़ NBUDI౳Ͱ΋0,
  18. public function receive(ContextInterface $context): void { $msg = $context->message(); switch

    (true) { case $msg instanceof EventDescription: $result = $context->spawnNamed( Props::fromProducer(fn() => new TicketSeller()), $msg->name ); if ($result->isError() instanceof NameExistsException) { // ΞΫλʔ͕ੜ੒ࡁΈͷ৔߹͸ੜ੒ࡁΈͰ͋Δ͜ͱΛ௨஌͠·͢ $context->respond(new EventExists()); break; } $context->send($result->getRef(), new Add($msg->name, $msg->tickets, $context->sender())); break; ࢓ࣄΛґཔ͢ΔࢠΞΫλʔΛੜ੒
  19. public function receive(ContextInterface $context): void { $msg = $context->message(); switch

    (true) { case $msg instanceof EventDescription: $result = $context->spawnNamed( Props::fromProducer(fn() => new TicketSeller()), $msg->name ); if ($result->isError() instanceof NameExistsException) { // ΞΫλʔ͕ੜ੒ࡁΈͷ৔߹͸ੜ੒ࡁΈͰ͋Δ͜ͱΛ௨஌͠·͢ $context->respond(new EventExists()); break; } $context->send($result->getRef(), new Add($msg->name, $msg->tickets, $context->sender())); break; ΠϕϯτνέοτΛ୲౰͢Δ ΞΫλʔ ࣝผͰ͖Δ໊લΛ͚ͭΔྫ
  20. public function receive(ContextInterface $context): void { $msg = $context->message(); switch

    (true) { case $msg instanceof EventDescription: $result = $context->spawnNamed( Props::fromProducer(fn() => new TicketSeller()), $msg->name ); if ($result->isError() instanceof NameExistsException) { // ΞΫλʔ͕ੜ੒ࡁΈͷ৔߹͸ੜ੒ࡁΈͰ͋Δ͜ͱΛ௨஌͠·͢ $context->respond(new EventExists()); break; } $context->send($result->getRef(), new Add($msg->name, $msg->tickets, $context->sender())); break; ࣝผՄೳͳΞΫλʔ͕ઌʹ ىಈ͍ͯ͠Δঢ়ଶ͕͢Ͱʹ͋Δ ͜ͷͨΊ࡞੒͠ͳ͍
  21. public function receive(ContextInterface $context): void { $msg = $context->message(); switch

    (true) { case $msg instanceof EventDescription: $result = $context->spawnNamed( Props::fromProducer(fn() => new TicketSeller()), $msg->name ); if ($result->isError() instanceof NameExistsException) { // ΞΫλʔ͕ੜ੒ࡁΈͷ৔߹͸ੜ੒ࡁΈͰ͋Δ͜ͱΛ௨஌͠·͢ $context->respond(new EventExists()); break; } $context->send($result->getRef(), new Add($msg->name, $msg->tickets, $context->sender())); break; νέοτൢചΞΫλʔѼʹ "EEϝοηʔδΛૹ৴
  22. BoxO ffi ce ΞΫλʔ TicketSeller ΞΫλʔ TicketSeller ΞΫλʔ TicketSeller ΞΫλʔ

    TicketSeller ΞΫλʔ TicketSeller ΞΫλʔ TicketSeller ΞΫλʔ
  23. <?php declare(strict_types=1); namespace App\Message; use Phluxor\ActorSystem\Ref; class Add { public

    function __construct( readonly public string $name, readonly public int $tickets, readonly public Ref $replyTo ) { } }
  24. <?php declare(strict_types=1); namespace App\Message; use Phluxor\ActorSystem\Ref; class Add { public

    function __construct( readonly public string $name, readonly public int $tickets, readonly public Ref $replyTo ) { } } Πϕϯτ໊ νέοτຕ਺ νέοτ࡞੒݁Ռૹ৴ઌ
  25. class TicketSeller implements ActorInterface { private int $tickets = 0;

    private string $name = ''; private string $id = ''; public function receive(ContextInterface $context): void { $msg = $context->message(); switch (true) { case $msg instanceof Add: // Ұ෦ൈਮ $this->name = $msg->name; $this->tickets = $msg->tickets; $this->id = (string)$context->self(); $context->send($msg->replyTo, new EventCreated($msg->name, $msg->tickets)); break; } } }
  26. class TicketSeller implements ActorInterface { private int $tickets = 0;

    private string $name = ''; private string $id = ''; public function receive(ContextInterface $context): void { $msg = $context->message(); switch (true) { case $msg instanceof Add: // Ұ෦ൈਮ $this->name = $msg->name; $this->tickets = $msg->tickets; $this->id = (string)$context->self(); $context->send($msg->replyTo, new EventCreated($msg->name, $msg->tickets)); break; } } } 5JDLFU4FMMFS͕ "EEϝοηʔδΛड͚औͬͨΒ
  27. class TicketSeller implements ActorInterface { private int $tickets = 0;

    private string $name = ''; private string $id = ''; public function receive(ContextInterface $context): void { $msg = $context->message(); switch (true) { case $msg instanceof Add: // Ұ෦ൈਮ $this->name = $msg->name; $this->tickets = $msg->tickets; $this->id = (string)$context->self(); $context->send($msg->replyTo, new EventCreated($msg->name, $msg->tickets)); break; } } } 5JDLFU4FMMFSʹঢ়ଶΛอଘ 1FSTJTUFODFར༻࣌͸ɺ ঢ়ଶมߋຖʹอଘ͞ΕΔ
  28. class TicketSeller implements ActorInterface { private int $tickets = 0;

    private string $name = ''; private string $id = ''; public function receive(ContextInterface $context): void { $msg = $context->message(); switch (true) { case $msg instanceof Add: // Ұ෦ൈਮ $this->name = $msg->name; $this->tickets = $msg->tickets; $this->id = (string)$context->self(); $context->send($msg->replyTo, new EventCreated($msg->name, $msg->tickets)); break; } } } νέοτൢചΠϕϯτͷ ঢ়ଶ͕อଘ͞ΕͨΒ &WFOU$SFBUFEΠϕϯτΛฦ٫ ʢSFQMZ5Pʹࢦఆ͞ΕͨΞΫλʔѼʣ
  29. private function fetchEvents(ContextInterface $context): array { $wg = new WaitGroup();

    $events = []; foreach ($context->children() as $child) { $wg->add(); $future = $context->requestFuture($child, new GetEvent(), 2000); $fr = $future->result(); if ($fr->error() !== null) { $wg->done(); continue; } if ($fr->value() instanceof Event) { $events = array_merge($events, [$fr->value()]); $wg->done(); } } $wg->wait(); return $events; }
  30. private function fetchEvents(ContextInterface $context): array { $wg = new WaitGroup();

    $events = []; foreach ($context->children() as $child) { $wg->add(); $future = $context->requestFuture($child, new GetEvent(), 2000); $fr = $future->result(); if ($fr->error() !== null) { $wg->done(); continue; } if ($fr->value() instanceof Event) { $events = array_merge($events, [$fr->value()]); $wg->done(); } } $wg->wait(); return $events; } ࣗ਎͕ੜ੒ͨ͠ࢠΞΫλʔ ʢ֤ΠϕϯτνέοτൢചΞΫλʔʣ ʹݱࡏͷঢ়ଶΛਘͶΔ
  31. private function fetchEvents(ContextInterface $context): array { $wg = new WaitGroup();

    $events = []; foreach ($context->children() as $child) { $wg->add(); $future = $context->requestFuture($child, new GetEvent(), 2000); $fr = $future->result(); if ($fr->error() !== null) { $wg->done(); continue; } if ($fr->value() instanceof Event) { $events = array_merge($events, [$fr->value()]); $wg->done(); } } $wg->wait(); return $events; } ͢΂ͯͷνέοτൢചঢ়گΛϚʔδ