Winter DTCE: Distributed Task Computing Engine

Winter DTCE (Distributed Task Computing Engine) runs tasks across one or more nodes. It accepts tasks and jobs, queues them, and processes them with workers that you write. This page explains how DTCE works, which backends it supports, and how to run your first task.

A task is a single unit of work. A job is a collection of similar tasks.

DTCE provides the following capabilities:

• Run a single task, synchronously or asynchronously
• Run a large job (many tasks), synchronously or asynchronously
• Keep task input and output in a storage backend, so work survives restarts and failures (with a persistent backend)

Setup

Install suvera/winter-modules as described in Installation, then enable DTCE in application.yml:

application.yml

modules:
  - module: dev\winterframework\dtce\DtceModule
    enabled: true
    configFile: dtce-config.yml

PHP extension prerequisites

• swoole extension is required.
• redis extension is needed if Redis is used as a backend store.
• rdkafka extension is needed if Kafka is used as a queue.

Architecture

There are three main components.

1. Task Queue

Any executable task is placed in the queue, then processed by a worker. Any queue implementation must implement TaskQueueHandler. DTCE includes four implementations.

Queue class	Kind	Notes
dev\winterframework\dtce\task\storage\TaskQueueShared	Local	Uses the local queue store — set winter.queue.port in application.yml. Shared by all processes on one host.
dev\winterframework\dtce\task\storage\TaskQueueRedis	Distributed	Requires the Redis module.
dev\winterframework\dtce\task\storage\TaskQueueKafka	Distributed	Requires the Kafka module.
dev\winterframework\dtce\task\storage\TaskQueuePdbc	Distributed	Uses a database-backed queue.

2. Task Store

Input data to a worker and output data from a worker are stored in the Task Store. Any store implementation must implement TaskIOStorageHandler. DTCE includes four implementations.

Store class	Kind	Notes
dev\winterframework\dtce\task\storage\TaskIOStorageDisk	Local disk	Local disks are not distributed; they only work on a single node. For multiple nodes, use NFS, HDFS, GlusterFS, or another shared file system.
dev\winterframework\dtce\task\storage\TaskIOStorageKvStore	Local memory KV	Uses the local KV store — set winter.kv.port in application.yml. Shared by all processes on one host.
dev\winterframework\dtce\task\storage\TaskIOStorageRedis	Distributed	Requires the Redis module.
dev\winterframework\dtce\task\storage\TaskIOStoragePdbc	Distributed	Uses a database-backed store.

3. Task Worker

This is the actual implementation of your work. A task worker must implement TaskWorker.

Example worker and execution

The following WordCounter worker counts the number of words in a string:

class WordCounter extends AbstractTaskWorker {

    public function work(mixed $input): TaskOutput {
        if (!is_scalar($input)) {
            return new NumericOutput(0);
        }

        $input = '' . $input;

        return new NumericOutput(str_word_count($input));
    }
}

Execute it through TaskExecutionServiceFactory:

#[Autowired]
private TaskExecutionServiceFactory $factory;

// Name of the task "wordCount", provided in dtce-config.yml.
$executor = $this->factory->executionService("wordCount");

$result = $executor->executeTask("Some text here. Every application has some number of tasks that run in the background.");

if ($result->isSuccess()) {
    print "Task SUCCESS: " . $result->getResult() . " words.\n";
} else {
    print "Task FAILED!\n";
}

DTCE configuration (dtce-config.yml)

A minimal dtce-config.yml assigns a queue handler, a storage handler, and a worker class to a named task:

config/dtce-config.yml

tasks:
  - name: wordCount
    storage:
      handler: dev\winterframework\dtce\task\storage\TaskIOStorageDisk
      path: /tmp
    worker:
      total: 4
      class: com\company\winter\task\word\WordCounter
    queue:
      handler: dev\winterframework\dtce\task\storage\TaskQueueRedis

  # - name: another-task
  #   ...

Full configuration

tasks:
  - name: name-of-the-task

    # TASK STORAGE configuration
    storage:
      handler: dev\winterframework\dtce\task\storage\TaskIOStorage***

      # TaskIOStorageDisk handler specific
      path: /folder/path

      # TaskIOStoragePdbc handler specific
      pdbc:
        bean:      # optional, unless you have your own PdbcTemplate implementation
        entity:    # optional, defaults to "TaskIoTable"; you need the table created beforehand
        ttl:       # optional, in seconds; time to keep stored records, defaults to 4 hours

      # TaskIOStorageRedis handler specific
      redis:
        bean:      # optional, unless you have a redis bean name; see the Redis module for details
        ttl:       # optional, in seconds; time to keep stored records, defaults to 4 hours

    # TASK WORKER configuration
    worker:
      total: 10   # Number of workers to spawn. This is the concurrency setting.
      class: com\company\winter\task\word\WordCounter

    # TASK QUEUE configuration
    queue:
      handler: dev\winterframework\dtce\task\storage\TaskQueue**

      readTimeoutMs:   # milliseconds to wait while reading data from queue
      writeTimeoutMs:  # milliseconds to wait while writing data to queue

      # TaskQueuePdbc specific
      pdbc:
        bean:      # optional, unless you have your own PdbcTemplate implementation
        entity:    # optional, defaults to "TaskQueueTable"; you need the table created beforehand

      # TaskQueueRedis specific
      redis:
        bean:      # optional, unless you have a redis bean name; see the Redis module for details
        key:       # optional, if you want to use a specific list key

      # TaskQueueKafka specific
      kafka:
        bootstrap.servers: 127.0.0.1  # Kafka nodes
        topic: topic1               # Kafka topic name
        consumer:
          - auto.commit.interval.ms: 100
            auto.commit.enable: true
            # as many rdkafka settings as needed; see https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md
        producer:
          - message.max.bytes: 10485760
            retries: 5
            # as many rdkafka settings as needed; see https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md

Tip

DTCE complements the built-in framework primitives. For single-process background work see Async Tasks and Scheduled Tasks; use DTCE when you need distributed, persistent task processing across nodes. Redis-backed queues use the Redis module; Kafka-backed queues use the Kafka module.