worker dynamic pickup task with some constraint

[big91987]

one more question，，，，how can I make some worker that can dynamic pickup task with some pattern （no matches task will be taken by other suitable workers）？

eg： some Specific work will pick up those tasks matches the pattern：{device: CPU, dirver_version: 1.2.0, etc ...），

well,, I think these can be done by making lots of queues like:
Queues: mapa[string]{
"cpu_driver_NA_os_ubuntu_queue": 1,
"cpu_driver_NA_os_centos_queue": 1,
"gpu_driver_370_cuda_9_queue": 1,
"gpu_driver_375_cuda_11_queue": 1,
"npu_driver_1_1_queue": 1,
"npu_driver_1_2_queue": 1,
}

but it seems not a smart solution（When the latter factors become more and more, complicated reconstruction is required），, is there some more reasonable and flexible implementation ?

PS：I search some solution found kafka may impl that by Confirm offset manually

These requirements came from A distributed model computing service, which need to distinguish hardware, os, driver and other factors to evaluate the impact of these factors on computing

[hibiken]

@big91987 Thank you for opening another discussion thread :)

I have an open feature request #250, and I'm planning to add a feature to match queue names based on the prefix (currently queue names in Config.Queues are looking for exact match.

But is it necessary for you to create this many queues? Can you use only the three queues cpu, gpu, and npu and put tasks in these queues. You can enqueue different types of tasks in a queue, and let Handler decide how to handle each task type.

Let me know if that makes sense. I'm happy to help :)

[big91987]

well.. I'm not sure the Post processing behavior when a worker reject a task ( when it found it's tag not Satisfy the constraints from the task)，is the rejected task will lost ？Did I need to add some re-queue logic to the code?

such as：

func HandleCalculateTask(ctx context.Context, t *asynq.Task) error {
var p EmailDeliveryPayload
if err := json.Unmarshal(t.Payload(), &p); err != nil {
return fmt.Errorf("json.Unmarshal failed: %v: %w", err, asynq.SkipRetry)
}

// get tag of this worker ...
// get the  constraints from the task payload
if not match // found not match
{      
        // reject the task
        // requeue the task
        //  sleep 3s ...
} else {
        // get job done ...
}
return nil

}

[hibiken]

Sorry for the delay here. What do you mean by "rejecting" a task. Do you mean if the Handler returns a non-nil error?

Those tasks will be retried automatically up to configured maximum retry.
Note: if you return asynq.SkipRetry, it'll skip retry and the task will be put in the archive (and will stay there for some time until archive gets full or the task is older than 90d).

Either way, you don't need to perform requeue logic yourself.