rusty-celery / rusty-celery Goto Github PK

View Code? Open in Web Editor NEW

721.0 17.0 81.0 2.23 MB

🦀 Rust implementation of Celery for producing and consuming background tasks

Home Page: https://rusty-celery.github.io

License: Apache License 2.0

Makefile 0.44% Rust 98.12% Shell 0.55% Python 0.89%

rust celery celery-workers rabbitmq redis

rusty-celery's Introduction

A Rust implementation of Celery for producing and consuming asynchronous tasks with a distributed message queue.

We welcome contributions from everyone regardless of your experience level with Rust. For complete beginners, see HACKING_QUICKSTART.md.

If you already know the basics of Rust but are new to Celery, check out the Rusty Celery Book or the original Python Celery Project.

Quick start

Define tasks by decorating functions with the task attribute.

use celery::prelude::*;

#[celery::task]
fn add(x: i32, y: i32) -> TaskResult<i32> {
    Ok(x + y)
}

Create an app with the app macro and register your tasks with it:

let my_app = celery::app!(
    broker = AMQPBroker { std::env::var("AMQP_ADDR").unwrap() },
    tasks = [add],
    task_routes = [
        "*" => "celery",
    ],
).await?;

Then send tasks to a queue with

my_app.send_task(add::new(1, 2)).await?;

And consume tasks as a worker from a queue with

my_app.consume().await?;

Examples

The examples/ directory contains:

a simple Celery app implemented in Rust using an AMQP broker (examples/celery_app.rs),
the same Celery app implemented in Python (examples/celery_app.py),
and a Beat app implemented in Rust (examples/beat_app.rs).

Prerequisites

If you already have an AMQP broker running you can set the environment variable AMQP_ADDR to your broker's URL (e.g., amqp://localhost:5672//, where the second slash at the end is the name of the default vhost). Otherwise simply run the helper script:

./scripts/brokers/amqp.sh

This will download and run the official RabbitMQ image (RabbitMQ is a popular AMQP broker).

Run the examples

Run Rust Celery app

You can consume tasks with:

cargo run --example celery_app consume

And you can produce tasks with:

cargo run --example celery_app produce [task_name]

Current supported tasks for this example are: add, buggy_task, long_running_task and bound_task

Run Python Celery app

Similarly, you can consume or produce tasks from Python by running

python examples/celery_app.py consume [task_name]

python examples/celery_app.py produce

You'll need to have Python 3 installed, along with the requirements listed in the requirements.txt file. You'll also have to provide a task name. This example implements 4 tasks: add, buggy_task, long_running_task and bound_task

Run Rust Beat app

You can start the Rust beat with:

cargo run --example beat_app

And then you can consume tasks from Rust or Python as explained above.

Road map and current state

✅ = Supported and mostly stable, although there may be a few incomplete features.
⚠️ = Partially implemented and under active development.
🔴 = Not supported yet but on-deck to be implemented soon.

Core

	Status	Tracking
Protocol	⚠️
Producers	✅
Consumers	✅
Brokers	✅
Beat	✅
Backends	🔴
Baskets	🔴

Brokers

	Status	Tracking
AMQP	✅
Redis	✅

Backends

	Status	Tracking
RPC	🔴
Redis	🔴

rusty-celery's People

Contributors

Stargazers

Watchers

rusty-celery's Issues

Support workflow primitives

https://docs.celeryproject.org/en/latest/getting-started/next-steps.html#the-primitives

Document protocol

Support countdown and ETA for tasks

Just make thread sleep until ETA is reached

Speed up CI

https://github.com/actions/cache/blob/master/examples.md#rust---cargo

ErrorKind::BrokerError needs to be more general

Right now it takes a lapin Error type, which of course only applies to the AMQP broker.

Implement the `arg_names` Task function within the macro

Document Task

Add contribution guideline

Document Broker

Implement an attribute macro for quickly defining tasks

https://doc.rust-lang.org/book/ch19-06-macros.html?highlight=macros#attribute-like-macros

https://github.com/kureuil/batch-rs/tree/master/batch-codegen/src

Populate origin field of MessageHeaders

This comes from https://docs.celeryproject.org/en/latest/internals/reference/celery.utils.nodenames.html#celery-utils-nodenames

Implement Redis backend

Switch to new tokio::select! in consume loop after next release

tokio-rs/tokio#2152

This means we wouldn't have to .fuse() the streams.

Handle heartbeat / make configurable

Set ETA when retrying task

Use exponential backoff, bounded by (min|max)_retry_delay

Add an `type Error` assoicated type to `Task` trait so tasks can return any type of error

Document app::Celery

Do we need to handle CPU bound tasks differently?

Is the tokio multi-threaded work-stealing runtime enough?

One option is to handle CPU vs IO bound tasks differently by using spawn_blocking or another thread pool instead of spawn for CPU bound tasks.

Something to think about though is that our timeout functionality doesn't work with blocking tasks, i.e. a blocking task could run forever. This is a problem even if we were to use spawn_blocking, because even though the timeout would return an error after the duration was exceeded, the thread that was spawned could still run forever in the background.

A second option is to completely switch over to async-std: https://async.rs/blog/stop-worrying-about-blocking-the-new-async-std-runtime/ (NOTE: this isn't shipped yet as of the current version 1.4)? We would still have the same issue with timeouts though.

Implement RPC backend

pass additional context to task on_failure / on_success callbacks

Right now the task callback methods on_failure and on_success only take the error and returned value, respectively, as arguments. However it might be useful to provide additional information to these callbacks such as the task ID. We could pass this information in a context struct like this:

async fn on_failure(&mut self, ctx: &Context, err: &Error);

and

async fun on_success(&mut self, ctx: &Context, returned: &Self::Returns);

Implement celery beat

Handle task failure with appropriate non-ack and retry

Add support for backends

Add call for participation

https://users.rust-lang.org/t/twir-call-for-participation/4821

Add logging that matches celery logging

Use interior mutability pattern for Celery struct so that `registert_task` doesn't have to take `&mut self`

Improve send_task method to support other options

Implement Redis broker

Add support for initializing app from a config file that matches a Python Celery config

Support args in addition to kwargs

Graceful shutdown

"Warm shutdown". Finish currently executing tasks.

Add runnable examples

Supports other serializers

Currently only JSON supported

Should acks_late=false be the default?

See https://docs.celeryproject.org/en/latest/userguide/tasks.html and https://docs.celeryproject.org/en/latest/faq.html#faq-acks-late-vs-retry

Check 'expires' task property before executing

Add tests for conversion between Message and AMQP Properties / Delivery

Support acks_late configuration

Right now acks are always late

Add logo to docs

Provide better mechanism for making task parameters optional / providing default values

Currently there is no way to do this through the #[task] macro

Support task routing configuration

Handle connection errors

Add app autoregistration and default name to task definition macro

Add a required(?) positional argument to the task macro that is the app the task needs to be registered on. Also make the name of the task inferred from the function name if not explicitly defined like how python celery does.

Bug: tasks received with ETA will count towards AMQP prefetch count

Provide macro for building celery app

This would wrap the lazy static macro and the builder

Add integration tests with local RabbitMQ broker instance or a mock broker

Support a distributed deployment of celery beat

Would need to use some form of a distributed lock, like redbeat does.

Implement SQS broker

Document best practices

For example,

Tasks should never panic. Return an error with ? instead.
Task timeouts only work when task is non-blocking. If your wrote a task with an infinite loop, the whole worker would freeze indefinitely.
Error handling should be done by returned either ErrorKind::ExpectedError or ErrorKind::UnexpectedError
Use non blocking functions for IO, such as those from tokio
Tuning prefetch_count: For CPU-bound tasks, set prefetch_count to num CPUs. For IO-bound tasks, set much, much higher. Generally try to separate CPU bound and IO bound tasks to different queues.
Each task should do one thing and do it well. Ideally there should only be at most one possible point of failure in a task.

Support "baskets": long term scheduler backends

The current system (and essentially how Python Celery does it) for handling tasks with a far-off ETA is to have workers consume them as usual but use the delay_for async function to delay execution until ETA is reached, which could be a while.

In the meantime that task is still taking up resources on a worker and more importantly is overriding the backpressure mechanism - the prefetch_count configuration setting - because in order for the worker to continue consuming other tasks while it is holding onto delayed tasks it needs to tell the broker to increase its channel's prefetch_count behind the scenes.

If it didn't do this and the worker kept receiving tasks with a far-off ETA, the initial prefetch_count would soon be reached and so the broker would stop sending tasks to this worker. Absent of more workers being spun up, this would cause the broker to pile up with messages since it has no where to send them. So no new tasks (even those without a future ETA) could be executed until the worker executes some of the tasks it is holding onto.

So we choose the lesser of the two evils: increasing the prefetch_count. In other words, the worker says "hey, thanks for this task, but I can't do anything with it right now so want to just give me another one?". And that's all fine unless there are a ton of tasks with far-off ETA, in which case the worker will keep taking in more of these until it runs out of memory.

The solution to this is to offload those tasks - tasks with a far-off ETA - somewhere else. Someplace where the chance of running out of memory or storage space is a lot lower, and where the cost of additional memory or storage space is a lot cheaper.

Any traditional database works well for this as long as you can index by the ETA, which you should be able to do with pretty much any database since you can represent the ETA by an integer or float. Then workers (or a dedicated worker solely for this purpose) just need to occasionally poll the database for tasks that are due soon.