Hello,

I've been looking for a celery alternative for quite a while, thanks for taking the time to make this.

A suggestion i'd like to make is to add the ability of defining class-based actor where grouping common functionality among different "tasks" or "actors" is possible.

Thank you.

Issues

GitHub issues are for bugs. If you have questions, please ask them on the discussion board.

Checklist

• [x] Does your title concisely summarize the problem?
• [x] Did you include a minimal, reproducible example?
• [x] What OS are you using?
• [x] What version of Dramatiq are you using?
• [x] What did you do?
• [x] What did you expect would happen?
• [x] What happened?

What OS are you using?

CentOs7

What version of Dramatiq are you using?

1.5.0

What did you do?

called function completes and returns a list successfully ( I have chekd this with print statement, and it prints correctly whenever I call the function asynchronously ), but when I try to retrieve the retruned result with tacotron.send(model).get_result(backend = result_backend, block=True), I get:

Traceback (most recent call last):
  File "actor.py", line 55, in <module>
    ret = tacotron.send().get_result(backend = result_backend, block=True)
  File "/home/msl/.virtualenvs/venv_tts/lib/python3.5/site-packages/dramatiq-1.5.0-py3.5.egg/dramatiq/message.py", line 147, in get_result
    return backend.get_result(self, block=block, timeout=timeout)
  File "/home/msl/.virtualenvs/venv_tts/lib/python3.5/site-packages/dramatiq-1.5.0-py3.5.egg/dramatiq/results/backends/redis.py", line 81, in get_result
    raise ResultTimeout(message)
dramatiq.results.errors.ResultTimeout: tacotron()

this is how I defined tacotron(model):

@dramatiq.actor
def tacotron(model):
    return model.GetMel("President Trump met with other leaders at the Group of 20 conference.", 0)

I have put print statement inside GetMel of model, and it prints correctly -- i.e. right before it returns, I check the content of the list that is being returned, and they are fine. It prints within 1 second of the request.

Result backend: Redis I'm not sure if this is correct. How I installed and started Redis:

sudo pip install redis
sudo yum install redis
sudo systemctl start redis

Broker: RabbitMQ

What did you expect would happen?

for tacotron.send(model).get_result(backend = result_backend, block=True) to return successfully

What happened?

get_result(block=True) hung then errored

Traceback (most recent call last):
  File "actor.py", line 55, in <module>
    ret = tacotron.send().get_result(backend = result_backend, block=True)
  File "/home/msl/.virtualenvs/venv_tts/lib/python3.5/site-packages/dramatiq-1.5.0-py3.5.egg/dramatiq/message.py", line 147, in get_result
    return backend.get_result(self, block=block, timeout=timeout)
  File "/home/msl/.virtualenvs/venv_tts/lib/python3.5/site-packages/dramatiq-1.5.0-py3.5.egg/dramatiq/results/backends/redis.py", line 81, in get_result
    raise ResultTimeout(message)
dramatiq.results.errors.ResultTimeout: tacotron()

This fixes #480

I'm not sure how to test this, since the race condition is not easy to trigger. @Bogdanp do you have any suggestions?

Also looking at the stub broker it seems that it may have a similar race condition, but I've never triggered a similar issue with it, so I'm not sure if the change (swapping the order or the delay queue and normal one) is needed https://github.com/Bogdanp/dramatiq/blob/2d0ca2519a0289946337b308e450a915cf5fd869/dramatiq/brokers/stub.py#L153-L164

What OS are you using?

Debian Stretch inside Docker

What version of Dramatiq are you using?

1.5.0

What did you do?

Run dramatiq not.existing; echo $? in a Docker container.

A simple docker run --rm -ti python:3.7-stretch /bin/bash followed by pip install dramatiq is the fastest way to get a repro.

What did you expect would happen?

The worker fails to boot and complains with something along the lines of ModuleNotFoundError: No module named 'not', the return code is always 2.

When running inside an iterm2 Mac zsh shell this is exactly what happens.

What happened?

When running below docker most of the times the output is completely swallowed. Sometimes only [dramatiq.MainProcess] [INFO] Dramatiq '1.5.0' is booting up. is printed.

Every 10th or so time the complete expected output can the seen.

The exit code is oscillating between 0 and 2.

I did not have the time yet to investigate but it seems to be related to the logging pipe of the workers. If I tweak setup_worker_logging not to pass the pipe the errors are reported as expected.

I was running into this problem when adding Dramatiq in a Kubernetes setup. I cannot say yet if there are additional problems when the module can be successfully imported.

As someone new to dramatiq, the retry middleware may cause task failures during testing to appear as if the test worker is hanging indefinitely or is otherwise failing to execute tasks. What actually happened:

• Some of my application task code failed during testing, causing the retry middleware to kick in.
• The retry middleware defaults to a 7-day maximum backoff.
• pytest tends to capture stdout/stderr, and terminating the test run causes that output to be lost.

Given the lack of output and apparent hang, my initial reaction was confusion and to assume that I had somehow misconfigured the test worker. The reality was that everything was working correctly, and pytest was just eating the output.

The unit testing guide should probably mention that apparent hangs may be the result of the retry middleware, and that they should try temporarily setting max_retries=0 or consider disabling the retry middleware during testing.

The memory leak takes place only when an exception is thrown.

The problem has been traced down to the exception handler of process_message and more specifically to how the exception object is handled and stored in order to be referred back to down the road, eg. for debugging purposes or by the results' middleware. When message.stuff_exception(e) is called, a cyclic reference is created, which causes objects to not be garbage-collected (at least not until the cyclic GC kicks in).

In general, a cyclic reference is created when the exception object is stored in a variable x, which is part of the stack frame. The exception object references the stack frame, thus the variable x, via its __traceback__ attribute. But now x references the exception object, too, thus the stack frame. So, a cyclic reference is created!

In this particular case, the reference cycle created is:

message -> message._exception -> e -> stackframe -> message

In general, it is not recommended to store exception objects in local vars. However, if that's necessary, such variables should always be (explicitly) cleared in order to break the cyclic reference - usually in a try/finally statement.

In that sense, it comes in handy that a try/finally statement is already being used. By setting message._exception to None at the very end the reference cycle is broken and objects are garbage-collected soon enough.

The following piece of code can be used to re-produce the problem:

import dramatiq


@dramatiq.actor
def foo(*args, **kwargs):
    a = tuple(range(5000000))                                        
    raise Exception('bar')                                                      

After running for _ in range(5): foo.send() the memory leak is evident:

When the fix is applied:

Hello, @Bogdanp, Thanks for dramatiq!

As described at Raven README.md Raven SDK is being phased out for Sentry-Python.

It will be cool to update docs(raven example) and add official integration to SDK.

https://getsentry.github.io/sentry-python\

discuss topic

First of all - Big thanks @Bogdanp for creating a truly pythonic task processor! Celery has always given me a kind of bitter after taste..

I'm working towards a backend system that can't handle concurrent writes to the same database table. On top of that the task writing to this system might fail half-way and then we have no way to know where to resume when the task is retried (unless I keep state in the app which I'd like to avoid to keep things simple), so I've passed max_retries=0 to dramatic.actor and made the task notify me of any failures.

So my thought was that dramatiq.rate_limits.ConcurrentRateLimiter would be a perfect fit here. In the example in the cookbook though you're using it inside the task function. But I guess this wouldn't work in my scenario as the task will never be retried... Is there any way to have the mutex apply to the task itself so that it never even starts if another instance of that task is already executing?

In my head an API like this would do it for my scenario:

import dramatiq
import time

from dramatiq.rate_limits import ConcurrentRateLimiter
from dramatiq.rate_limits.backends import RedisBackend

backend = RedisBackend()
DISTRIBUTED_MUTEX = ConcurrentRateLimiter(backend, "distributed-mutex", limit=1)

@dramatiq.actor(rate_limiter=DISTRIBUTED_MUTEX, max_retries=0)
def one_at_a_time():
  print("I'll never run concurrently! And never twice!")

Or maybe I'm going all wrong with this?

Any input would be greatly appreciated!

What OS are you using?

Running in default Python 3.7 Docker container (Debian)

What version of Dramatiq are you using?

1.4.3

What did you do?

Created an actor using the decorator. Then called that actor using send(). This was called from the request context in a Flask application.

What did you expect would happen?

Expected message to be queued in rabbitmq

What happened?

FileNotFoundError. Doesn't consistently happen, but periodically it throws this error.

FileNotFoundError: [Errno 2] No such file or directory
  File "flask/app.py", line 2309, in __call__
    return self.wsgi_app(environ, start_response)
  File "appoptics_apm/middleware.py", line 142, in __call__
    result = self.wrapped_app(environ, wrapped_start_response)
  File "werkzeug/contrib/fixers.py", line 152, in __call__
    return self.app(environ, start_response)
  File "raven/middleware.py", line 100, in __call__
    iterable = self.application(environ, start_response)
  File "flask/app.py", line 2295, in wsgi_app
    response = self.handle_exception(e)
  File "flask_restful/__init__.py", line 273, in error_router
    return original_handler(e)
  File "flask/app.py", line 1741, in handle_exception
    reraise(exc_type, exc_value, tb)
  File "flask/_compat.py", line 35, in reraise
    raise value
  File "flask/app.py", line 2292, in wsgi_app
    response = self.full_dispatch_request()
  File "flask/app.py", line 1815, in full_dispatch_request
    rv = self.handle_user_exception(e)
  File "flask_restful/__init__.py", line 273, in error_router
    return original_handler(e)
  File "flask/app.py", line 1718, in handle_user_exception
    reraise(exc_type, exc_value, tb)
  File "flask/_compat.py", line 35, in reraise
    raise value
  File "flask/app.py", line 1813, in full_dispatch_request
    rv = self.dispatch_request()
  File "flask/app.py", line 1799, in dispatch_request
    return self.view_functions[rule.endpoint](**req.view_args)
  File "flask/views.py", line 88, in view
    return self.dispatch_request(*args, **kwargs)
  File "flask/views.py", line 158, in dispatch_request
    return meth(*args, **kwargs)
  File "flask_login/utils.py", line 261, in decorated_view
    return func(*args, **kwargs)
  File "sylo/permissions/access.py", line 96, in wrapped
    return f(*args, **kwargs)
  File "webargs/core.py", line 447, in wrapper
    return func(*new_args, **kwargs)
  File "<my-file>", line 47, in wrapper
    result = func(*args, **kwargs)
  File "<my-file>", line 486, in func
    <my_actor>.send(one_id, two_id)
  File "dramatiq/actor.py", line 179, in send
    return self.send_with_options(args=args, kwargs=kwargs)
  File "dramatiq/actor.py", line 198, in send_with_options
    return self.broker.enqueue(message, delay=delay)
  File "dramatiq/brokers/rabbitmq.py", line 250, in enqueue
    properties=properties,
  File "pika/adapters/blocking_connection.py", line 2206, in publish
    immediate=immediate)
  File "pika/channel.py", line 425, in basic_publish
    (properties, body))
  File "pika/channel.py", line 1328, in _send_method
    self.connection._send_method(self.channel_number, method, content)
  File "pika/connection.py", line 2341, in _send_method
    self._send_message(channel_number, method, content)
  File "pika/connection.py", line 2355, in _send_message
    self._send_frame(frame.Method(channel_number, method_frame))
  File "pika/connection.py", line 2327, in _send_frame
    self._flush_outbound()
  File "pika/adapters/base_connection.py", line 330, in _flush_outbound
    self._manage_event_state()
  File "pika/adapters/base_connection.py", line 536, in _manage_event_state
    self.event_state)
  File "pika/adapters/select_connection.py", line 446, in update_handler
    self._poller.update_handler(fileno, events)
  File "pika/adapters/select_connection.py", line 639, in update_handler
    events_to_set=events_set)
  File "pika/adapters/select_connection.py", line 1152, in _modify_fd_events
    self._poll.modify(fileno, events)

does dramatiq has anything like celery about task routing.

I didn't find any api to send task with queue_name, but I can build message myself to send task to specific queue.

 Message(
            queue_name="only_i_can_receive",
            actor_name="count_words",
            args=("https://www.google.com", ),
        )
broker.enqueue(message, delay=delay)

but i can't start worker only consumer for specific queue like celery.

celery worker tasks -Q tasks.to.nodes.1

Hi,

This is not a bug but a wish, I would use the decorator actor on a method or a classmethod have you a solution ?

exemple:

class foo:
    @dramatiq.actor()
    def count_worlds(url):
          pass

Issues

GitHub issues are for bugs. If you have questions, please ask them on the discussion board.

Checklist

• [X] Does your title concisely summarize the problem?
• [X] Did you include a minimal, reproducible example?
• [X] What OS are you using?
• [X] What version of Dramatiq are you using?
• [X] What did you do?
• [X] What did you expect would happen?
• [X] What happened?

What OS are you using?

Happens both on Ubuntu 16.04 or macOS 10.13.3

What version of Dramatiq are you using?

1.13.0

What did you do?

Tried to create 65000+ threads and send one message per thread

What did you expect would happen?

Messages to be dispatched without any error

What happened?

Got an exception - OSError98/99

Situation

The default behavior of Dramatiq when multiple queues are present (either all of them or a subset), is to fetch data from them using the consumer threads. This in practice means that all queues are drained equally (assuming they all have messages in them).

In contrast, Resque, grabs messages in the order the queues are set when starting the workers. For example, given the queues low, high and default, one could start a worker with --queues=high,default,low and the queues would be drained in order. This ofc could lead to the low queue being starved if high or default always had a constant influx of new messages. To overcome this, often there would be multiple workers with different settings like --queues=low,default,high to ensure that all queues had at least one workers dedicated to them.

When compared to dramatiq, resque allows creating dedicated workers for a given queue, that when idle will try to process jobs from other queues. Dramatiq can have dedicated workers for a queue, however, they will stay idle if that queue is empty.

Potential Solutions

Middleware

Since dramatiq has good middleware support, one could write a new middleware to implement this behavior. However, it seems that there aren't any available hooks that could be used to handle this in an efficient way.

Ideally, no tasks should be waiting in a worker, while a potential another worker could be handling it. We should also avoid taking tasks off a queue, and then adding it back, repeatedly.

Builtin

Changing how dramatiq works probably isn't the best approach, since it could break setups that are functional with the current behavior. We could have a setting to toggle this behavior. Something like --drain-queues-in-order. IMO this solution is less preferred, since it adds more complexity and a somewhat opinionated way of how things should work.

Questions

0. Is anyone else interested in this? Is this even useful or desired?
1. Can any of the current hooks provided by the middleware architecture allow this to be implemented in a reasonable manner? Preferentially without changing dramatiq itself
2. If not, what do we need in order do implement it the "middleware way"?

I am trying to get an MVP working using the middleware approach, but so far nothing. Still getting familiar with dramatiq's internals

Note: I am starting the discussion here because the reddit seems to be abandoned

As discussed in #425 it is sometimes useful to do some special handling using the on_failure callback after the last retry. However, at the moment, the on_failure callback does not have any way to dynamically determine if the current retry is the last.

In other words, I think it could be useful to either pass the max_retries information in the message options or add another callback on_last_failure.

Unless I'm missing something, I think the first option should be fairly straightforward to implement by adding a before_enqueue method to the Retries middleware

class Retries(Middleware):
    ...
    def before_enqueue(self, broker, message, delay):
        actor = broker.get_actor(message.actor_name)
        max_retries = message.options.get("max_retries") or actor.options.get(
            "max_retries", self.max_retries
        )
        message.options["max_retries"] = self.max_retries

If you think this could be a useful feature and that the proposed solution makes sense, I'd be happy to open a PR with the change.

As a workaround I'm currently retrieving the actor in the on_failure handler

@dramatiq.actor
def my_callback(message_data, exception_data):
    actor = broker.get_actor(message_data["actor_name"])
    max_retries = actor.options.get("max_retries", 0)

    retries = message_data["options"]["retries"]
    if retries > max_retries:
        print("handling the error")

Hi, @Bogdanp! Maybe it's time to release a new version of dramatiq? Python 3.11 was released not too long ago -- it would be great to have a version compatible with modern python 😎 Of course, we can install a package with reference to a specific commit, but it's still better and more convinient to get dramatiq from PyPi, for example, v1.14.0.

Issues

Note -- I'm running Django-dramatiq

I need to use dramatiq to execute higher-priority tasks before lower-priority tasks, and I have a fairly long queue of tasks (maybe a few thousand running on only a 6 to 8 workers). However, I have very few high-priority tasks, and I'd expect those to always be executed first. But, the behavior I observe is that dramatiq will keep choosing to run enqueued actors with lower priorities (sometimes).

As far as I understood from reading a bunch of issues what I'd actually need here is "queue priority" or some such, which is only supported by RabbitMQ. But I'd be nice to double-check and see I'm not doing something wrong, and also, if priority is indeed limited to a specific situation (e.g. priority of resource allocation to already running tasks) I'd be nice if the documentation was more specific about this.

Checklist

• [x] Does your title concisely summarize the problem?
• [ ] Did you include a minimal, reproducible example?
• [x] What OS are you using?
• [x] What version of Dramatiq are you using?
• [x] What did you do?
• [x] What did you expect would happen?
• [x] What happened?

What OS are you using?

Ubuntu 22.04 LTS

What version of Dramatiq are you using?

1.13.0

What did you do?

I need to use dramatiq to execute higher-priority tasks before lower-priority tasks, and I have a fairly long queue of tasks (maybe a few thousand running on only a 6 to 8 workers). However, I only enqueued a few high-priority tasks

What did you expect would happen?

I have very few high-priority tasks, and I'd expect those to always be executed first.

What happened?

The behavior I observe is that dramatiq will keep choosing to run enqueued actors with lower priorities (sometimes).

Issues

GitHub issues are for bugs. If you have questions, please ask them on the discussion board.

Checklist

• [x] Does your title concisely summarize the problem?
• [x] Did you include a minimal, reproducible example?
• [x] What OS are you using?
• [x] What version of Dramatiq are you using?
• [x] What did you do?
• [x] What did you expect would happen?
• [x] What happened?

What OS are you using?

Ubuntu 20.04 and MacOS 12.6

What version of Dramatiq are you using?

Dramatiq 1.1.13

What did you do?

Enqueue messages over a long running process.

What did you expect would happen?

Messages should be published reliably.

What happened?

We've recently migrated to a RabbitMQ deployment on AmazonMQ and connect over SSL (where as previously it was plain text). We're seeing the following error occasionally:

Socket EOF; <ssl.SSLSocket fd=32, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=6, laddr=('x.x.x.x', 40536)>

We have a Django and Dramatiq deployment where both the Django and Dramatiq workers can publish messages. We've seen StreamLostErrors appear in our logs on long running processes, but have filtered them out assuming that Dramatiq's underlying retry mechanism will reopen the connection based on this Github Issue: https://github.com/Bogdanp/django_dramatiq/issues/85

I believe this is where pika is raising the SSL error: https://github.com/pika/pika/blob/main/pika/adapters/utils/io_services_utils.py#L794-L797

Browsing through the code, it looks like Dramatiq doesn't handle the SSL error, only subclasses of AMQPConnectionError or AMQPChannelError: https://github.com/Bogdanp/dramatiq/blob/master/dramatiq/brokers/rabbitmq.py#L341-L354