Command Dispatch

El sistema de dispatch enruta comandos desde procesos a handlers. Los procesos hacen yield de comandos con tags de correlación, los handlers ejecutan trabajo asíncrono, y los resultados fluyen de vuelta vía colas de eventos.

Flujo

sequenceDiagram
    participant P as Process
    participant W as Worker
    participant R as Registry
    participant H as Handler

    P->>W: yield(command, tag)
    W->>R: getHandler(cmdID)
    R-->>W: handler
    W->>H: Handle(cmd, tag, receiver)
    H-->>H: trabajo asíncrono
    H->>W: CompleteYield(tag, result)
    W->>P: encolar evento, despertar
    P->>P: resume con resultado

Registry de Comandos

El registry almacena handlers en una estructura híbrida:

type Registry struct {
    handlers [256]Handler         // Comandos de sistema: índice O(1)
    extended map[CommandID]Handler // Comandos extendidos: lookup en mapa
    frozen   atomic.Bool          // Sin lock después de boot
}

Comandos de sistema (0-255) usan indexación de array. Comandos extendidos usan lookup en mapa. Después de Freeze(), todos los lookups son sin lock.

Rangos de Command ID

El registro ocurre durante boot vía MustRegisterCommands(). Las colisiones causan panic en startup.

Definir Comandos

Los comandos son estructuras de datos con un CommandID único:

const MyCommand dispatcher.CommandID = 200

type MyCmd struct {
    Input  string
    Option int
}

var myCmdPool = sync.Pool{New: func() any { return &MyCmd{} }}

func (c *MyCmd) CmdID() dispatcher.CommandID { return MyCommand }

func (c *MyCmd) Release() {
    c.Input = ""
    c.Option = 0
    myCmdPool.Put(c)
}

La reutilización de pool elimina asignaciones en rutas críticas. Registre en package init:

func init() {
    dispatcher.MustRegisterCommands("myservice", MyCommand)
}

Dispatchers

Un dispatcher agrupa handlers relacionados. Implementa RegisterAll para registrar handlers y métodos de ciclo de vida para setup/teardown:

type Handler interface {
    Handle(ctx context.Context, cmd Command, tag uint64, receiver ResultReceiver) error
}

type ResultReceiver interface {
    CompleteYield(tag uint64, data any, err error)
}

type Dispatcher struct {
    // estado del servicio
}

func (d *Dispatcher) RegisterAll(register func(id dispatcher.CommandID, h dispatcher.Handler)) {
    register(myapi.MyCommand, dispatcher.HandlerFunc(d.handleMyCommand))
}

func (d *Dispatcher) handleMyCommand(ctx context.Context, cmd Command, tag uint64, receiver ResultReceiver) error {
    c := cmd.(*myapi.MyCmd)
    go func() {
        result := doWork(c)
        if ctx.Err() == nil {
            receiver.CompleteYield(tag, result, nil)
        }
    }()
    return nil
}

Registre como componente de boot:

func MyDispatcher() boot.Component {
    return boot.New(boot.P{
        Name:      "dispatcher.myservice",
        DependsOn: []boot.Name{DispatcherName},
        Load: func(ctx context.Context) (context.Context, error) {
            reg := dispatcher.GetRegistrar(ctx)
            svc := myservice.NewDispatcher()
            svc.RegisterAll(reg.Register)
            return ctx, nil
        },
    })
}

Yields y Correlación

Cuando un proceso necesita trabajo asíncrono, hace yield de un comando con un tag de correlación:

type Yield struct {
    Cmd Command
    Tag uint64    // Contador local al proceso para correlación
}

El worker extrae yields de StepOutput después de cada step y los despacha a handlers. Cada tag identifica únicamente la solicitud para que los resultados puedan matchearse de vuelta.

Ver También

• Scheduler - Ejecución de procesos
• Módulos - Integración de módulos Lua
• Process Model - Conceptos de alto nivel