Asynchronous calls

Winter brings Java’s concurrency model to PHP: ExecutorService, Future, CompletableFuture and the @Async annotation. The names, signatures and semantics are taken from java.util.concurrent deliberately — it is a vocabulary proven over decades and there is no need to invent it again.

What this is and why

An asynchronous call is a method call that returns control without waiting for the work to finish.

The problem. A handler nearly always contains work the client should not have to wait for. A user registered — a record has to be created, an email sent, a CRM poked, an event pushed into analytics. Only the first matters; the rest add seconds to the response and, when an external service is unavailable, take down a request that was essentially successful.

The other side of the same problem is the reverse. A page needs data from three services, and they do not depend on one another. Sequential calls give you the sum of three latencies where the largest of them would do.

The solution. Mark the method with an attribute and the calling code continues immediately. When the result is needed after all, the method returns a Future you can ask for it later.

Ancestry

If you have written Java, this section is already familiar — not only the ideas match but the names:

The signatures match too: future.get(timeout), future.cancel(mayInterruptIfRunning), executor.submit(...), executor.invokeAll(...), CompletableFuture.allOf(...).

Where Winter differs from Java

Three differences worth knowing in advance.

Coroutines instead of threads. A task runs not in a separate OS thread but in a Swoole coroutine. This is cooperative multitasking: two tasks never execute code simultaneously, they take turns when one runs into a wait — a database, the network, a file, a pause. Hence the rule: asynchrony speeds up waiting, not computation. Two calls to external APIs will go in parallel; two heavy computations will not.

There is no composition. thenApply, thenCompose and the rest of the stage pipeline were deliberately not ported: in PHP they would degenerate into a tangle of closures. There is whenComplete() for a callback and allOf() for awaiting a set — which covers the overwhelming majority of cases.

Calling a neighbouring method works. In Spring, @Async does not fire when the method is called through this inside the same bean — there is a proxy wrapper there. Winter substitutes a subclass for the class, the object stays one, so $this->asyncMethod() remains asynchronous. The only condition is that the method must not be private.

Quick start

Two steps: enable proxying and mark the method.

use Flytachi\Winter\Kernel\App\Attribute\EnableAsync;

#[EnableWeb]
#[EnableAsync]
final class Application extends WinterApplication { /* ... */ }

<?php

namespace Main;

use Flytachi\Winter\DI\Attribute\Autowired;
use Flytachi\Winter\Kernel\Concurrent\Async\Async;

class NotificationService
{
  #[Autowired] private Mailer $mailer;

  #[Async]
  public function sendWelcome(User $user): void
  {
      $this->mailer->send($user->email, 'welcome');
  }
}

The call looks no different from an ordinary one — which is what makes it convenient:

#[PostMapping]
public function register(#[RequestJson, Valid] RegisterDto $dto): ResponseEntity
{
  $user = $this->users->create($dto);

  $this->notifications->sendWelcome($user);   // will not delay the response

  return ResponseEntity::created($user);
}

The result of the work

A method declares one of two things — and nothing else.

void — fire and forget

The result is unavailable and an exception inside goes only to the log.

#[Async]
public function trackEvent(string $name, array $payload): void
{
  $this->analytics->push($name, $payload);
}

Suitable for anything whose outcome does not affect the response: metrics, notifications, cache warming.

Future — the result will be needed later

use Flytachi\Winter\Kernel\Concurrent\{CompletableFuture, Future};

#[Async]
public function fetchProfile(int $id): Future
{
  return CompletableFuture::completedFuture($this->api->profile($id));
}

That is what gives the second kind of win — launching independent operations at once:

$profile = $this->users->fetchProfile($id);      // off it goes
$orders  = $this->orders->fetchRecent($id);      // off it goes
$balance = $this->billing->fetchBalance($id);    // off it goes

return ResponseEntity::ok([                       // we wait here
  'profile' => $profile->get(),
  'orders'  => $orders->get(),
  'balance' => $balance->get(),
]);

Three calls to external services go at once, and the request takes as long as the slowest rather than the sum of the three.

Future and CompletableFuture

Future is the same promise as in Java: a contract for a value that does not exist yet.

get() is the only way to obtain the value, and it is also where errors surface:

try {
  $data = $future->get(timeout: 2.0);
} catch (TimeoutException) {
  $data = $this->cache->stale($id);      // did not arrive — serve what we have
} catch (ExecutionException $e) {
  $this->logger->error('fetch failed', ['cause' => $e->getPrevious()?->getMessage()]);
  $data = null;
}

CompletableFuture is the implementation you can drive by hand:

Awaiting a set:

CompletableFuture::allOf($profile, $orders, $balance)->get(timeout: 5.0);

The timeout here covers the whole set.

Executor pools

By default tasks go to the common executor. When you need control — to cap the number of concurrent calls to an external API, to separate one kind of work from another — create a pool of your own and name it in the attribute.

use Flytachi\Winter\Kernel\App\Attribute\{Bean, Configuration};
use Flytachi\Winter\Kernel\Concurrent\{ExecutorService, Executors, RejectPolicy};

#[Configuration]
final class AppConfig
{
  #[Bean(name: 'pool.sms')]
  public function smsPool(): ExecutorService
  {
      return Executors::newFixedExecutor(
          concurrency: 5,                       // no more than 5 at a time
          queue:       100,                     // and 100 more waiting
          onReject:    RejectPolicy::CALLER_RUNS,
      );
  }
}

#[Async('pool.sms')]
public function sendSms(string $phone, string $text): void { /* ... */ }

The name in the attribute is a key in the container, so the pool can be any object implementing ExecutorService.

There is also Executors::newDeferredExecutor() — an executor that runs tasks after the response has been sent. It belongs to running under PHP-FPM and is not needed in a Swoole application.

A bounded pool can report on itself — handy for metrics:

What to do on overflow

When both the concurrency and the queue are full, the reject policy fires — the same three ThreadPoolExecutor has in Java:

CALLER_RUNS is usually the sanest choice for the web: under overload the endpoint gets slower but does not start losing work and does not start returning errors.

The pool directly, without the attribute

#[Async] is a convenient wrapper, but the pool can be used on its own: it is the same ExecutorService as in Java, with the same methods.

This fits where the number of tasks is not known in advance and there is no point in a method per task:

use Flytachi\Winter\Kernel\Concurrent\Executors;

$executor = Executors::common();

$futures = [];
foreach ($regions as $region) {
  $futures[$region] = $executor->submit(fn() => $this->api->stats($region));
}

$stats = [];
foreach ($futures as $region => $future) {
  $stats[$region] = $future->get(timeout: 5.0);
}

invokeAll() does the same thing more briefly when you need every result at once:

$futures = $executor->invokeAll(
  array_map(fn($r) => fn() => $this->api->stats($r), $regions),
  timeout: 5.0,
);

What about shared state

The question everyone who has written multithreaded code asks first: are locks needed?

Inside a worker — no. Coroutines execute one at a time in a single thread, so two tasks physically cannot modify one variable simultaneously. Memory races in the sense threads have them do not exist here, and there is nothing for synchronized to correspond to.

But switching does happen. It happens at waiting points — a database call, the network, a file, a pause. Which means another task can run between two lines of your code, if there is such a call between them:

$balance = $this->repo->balance($id);      // ← another task can cut in here
$this->repo->setBalance($id, $balance - $amount);   // and read the old value

This is not a property of #[Async] but of any concurrent handling: the same is true of two simultaneous HTTP requests. The cure is what it has always been — an atomic operation in the database or a lock on its side, not something PHP provides.

State is not shared between workers at all. Every worker is a separate process with its own memory, so a static property, a counter in an object or a local cache is visible only inside it. Anything common to the whole application lives outside: in the database, in Redis, in storage.

Requirements for the class and the method

The substitution works by inheritance: the framework generates a descendant of your class and overrides the marked methods in it. Every limitation follows from that — they are exactly the ones ordinary PHP inheritance has.

For the class:

For the method:

For the return type — exactly two allowed options, and both must be stated explicitly:

#[Async] public function send(): void { }              // ✓ fire and forget
#[Async] public function fetch(): Future { }           // ✓ the result is collected later

#[Async] public function send() { }                    // ✗ no return type
#[Async] public function fetch(): ?Future { }          // ✗ nullable is not accepted
#[Async] public function count(): int { }              // ✗ there is nowhere to return a value from

A missing type is as much an error as a wrong one: the method has no return type. The habit of leaving out : void will not work here.

The quiet trap — new

Asynchrony is provided by substituting a descendant for the class, and that substitution can only be obtained from the container.

// ✓ asynchronous — the object came from the container
#[Autowired] private NotificationService $notifications;

// ✗ synchronous — an ordinary class, the attribute ignored
$notifications = new NotificationService();

The insidious part is that nothing breaks: the same type, the same result, no error and no warning. The work simply runs synchronously and the call waits.

The framework finds such places statically:

php call di build

Besides building the proxies, the command shows where a class with #[Async] is created with new. The check is heuristic — it does not see dynamic creation (new $class, factories) — but it catches the typical case.

Examples

Side effects after registration

<?php

namespace Main;

use Flytachi\Winter\DI\Attribute\Autowired;
use Flytachi\Winter\Kernel\Concurrent\Async\Async;
use Psr\Log\LoggerInterface;

class RegistrationService
{
  #[Autowired] private Mailer $mailer;
  #[Autowired] private CrmClient $crm;
  #[Autowired] private LoggerInterface $logger;

  public function register(RegisterDto $dto): User
  {
      $user = $this->users->create($dto);

      $this->afterRegistration($user);   // asynchronous, the response does not wait

      return $user;
  }

  #[Async]
  protected function afterRegistration(User $user): void
  {
      try {
          $this->mailer->send($user->email, 'welcome');
          $this->crm->createLead($user);
      } catch (\Throwable $e) {
          $this->logger->error('post-registration failed', [
              'user'  => $user->id,
              'error' => $e->getMessage(),
          ]);
      }
  }
}

The method is declared protected rather than private — otherwise the substitution would not fire. And the try/catch inside is mandatory: a void method has nobody to return an error to.

Gathering data in parallel

<?php

namespace Main;

use Flytachi\Winter\DI\Attribute\Autowired;
use Flytachi\Winter\Kernel\Concurrent\Async\Async;
use Flytachi\Winter\Kernel\Concurrent\{CompletableFuture, Future, TimeoutException};

class DashboardService
{
  #[Autowired] private BillingApi $billing;
  #[Autowired] private StatsApi $stats;

  #[Async]
  public function balance(int $userId): Future
  {
      return CompletableFuture::completedFuture($this->billing->balance($userId));
  }

  #[Async]
  public function usage(int $userId): Future
  {
      return CompletableFuture::completedFuture($this->stats->monthly($userId));
  }

  public function build(int $userId): array
  {
      $balance = $this->balance($userId);
      $usage   = $this->usage($userId);

      try {
          return [
              'balance' => $balance->get(timeout: 3.0),
              'usage'   => $usage->get(timeout: 3.0),
          ];
      } catch (TimeoutException) {
          return ['balance' => null, 'usage' => null, 'degraded' => true];
      }
  }
}

The two calls go at once, and the timeout keeps the page from hanging if one of the services answers badly.

Bounded access to an external gateway

When a partner API holds a limit on concurrent connections, it must not be exceeded however many requests arrive.

#[Configuration]
final class GatewayConfig
{
  #[Bean(name: 'pool.gateway')]
  public function gatewayPool(): ExecutorService
  {
      return Executors::newFixedExecutor(
          concurrency: 3,
          queue:       50,
          onReject:    RejectPolicy::CALLER_RUNS,
      );
  }
}

class PaymentService
{
  #[Async('pool.gateway')]
  public function notifyGateway(Payment $payment): void { /* ... */ }
}

The pool is shared across the worker: however many requests use it, no more than three go to the gateway at a time. Overflowing the queue slows the caller down rather than causing a refusal.

Driving a promise by hand

A CompletableFuture can be completed by you — convenient when the result arrives not from a call but from outside: from a callback, from a subscription, from another mechanism.

$future = new CompletableFuture();

$this->bus->subscribe('payment.confirmed', function (Payment $p) use ($future) {
  $future->complete($p);
});

try {
  $payment = $future->get(timeout: 30.0);
} catch (TimeoutException) {
  throw new ResponseException('Payment confirmation timed out', HttpCode::GATEWAY_TIMEOUT);
}

Together with the scheduler

#[Async] can be placed on a method already marked #[Scheduled] — the combination works. But it removes the scheduler’s main guarantee: the task stops being protected from overlapping itself, because the call returns immediately and the run counts as finished the moment it is dispatched.

The analysis, with the rules and the choice of reject policy, is in the Together with #[Async] section of the scheduler page.

Next

• Application components — how #[Async] differs from a component
• Scheduler — combining it with a schedule
• Processes — when the work must outlive the request
• Daemons — when there is a lot of it as well
• Runtimes — the Swoole coroutines all of this runs on
• Dependency injection — why the object must come from the container