Issue 1:
Checklist

I have checked the issues list

for similar or identical enhancement to an existing feature.

I have checked the pull requests list

for existing proposed enhancements.

I have checked the commit log

to find out if the if the same enhancement was already implemented in the

master branch.

I have included all related issues and possible duplicate issues in this issue

(If there are none, check this box anyway).

Related Issues and Possible Duplicates

Related Issues

#5974

Possible Duplicates

None

Brief Summary

When it's used asynchronous backend as Redis with pubsub, result of short task (0.1s length, for example) available immediate.

Once i add gevent, result of the same short task became available after 1s, 10 times slower!

I will clarify that in both cases I used identical separate worker runed in docker with standard perfork pool.

celery -A tasks worker --prefetch-multiplier=1 --concurrency=10

This behavior relate with implementation features of GH-5974

AsyncResult.get() in a gevent context has at least 1 second latency, because

I've updated the code to have a specific implementation for gevent and eventlet that will cause wait_for to only return every "timeout" # of seconds, rather than returning immediately

I would like the task result to be available as soon as possible after the worker has submitted it. I think that wait_for have to return immediate after socket operation result_consumer.drain_events occurs, rather than every "timeout" # of seconds.

It won't cause overhead, because socket operation still not block event loop, so it let other greenlets executing.

Design

Now in wait_for it wait (with timeout 1 second) when greenletDrainer.run will finish, of course it's never finish, so it return after 1 second timeout.

If in wait_for wait the end of socket operation _pubsub.get_message instead greenletDrainer.run it will less latency and still not block event loop, so it let other greenlets executing.

I can prepare PR with changes in backends.asynchronous greenletDrainer, eventletDrainer, geventDrainer, that every time after result_consumer.drain_events (in greenletDrainer run loop) will send notification for wakeup all greenlets waiting results in wait_for.

Architectural Considerations

None

Proposed Behavior

Asynchronous backends (for example Redis) will be same efficient under gevent context as without it.

Proposed UI/UX

Diagrams

N/A

Alternatives

Another workaround may be to pass interval argument into BaseResultConsumer._wait_for_pending for pass it into drain_events_until and than into wait_for.

In this case it will be possible define own timeout for wait_for interation when execute AsyncResult.get(interval=0.05).

Howto reproduce

run docker with redis and rabbitmq

docker run -d -p 6379:6379 --name redis redis
docker run -d -p 5672:5672 --name rabbit rabbitmq:3

create tasks.py with your ip address for broker and backend

from celery import Celery
import time

app = Celery('tasks', broker='pyamqp://guest@192.168.1.40//', backend='redis://192.168.1.40')

@app.task(bind=True)
def sleep(self, timeout):
print(f'task {self.request.id} sleep({timeout}) started')
time.sleep(timeout)
print(f'task {self.request.id} sleep({timeout}) finished')
return f'task {self.request.id} sleep({timeout}) finished'

create Dockerfile

FROM python:3.8

RUN pip install celery redis

WORKDIR /root
CMD bash

build and run worker

docker build . -t worker
docker run -it --entrypoint celery -v $(pwd):/root worker -A tasks worker --prefetch-multiplier=1 --concurrency=10 --loglevel DEBUG

create gevent_example.py

import gevent
from gevent import monkey; monkey.patch_all()
from datetime import datetime
from tasks import sleep

st = datetime.now()

def test(timeout):
print(f'{datetime.now() - st} sleep.delay({timeout})')
a = sleep.delay(timeout)
gevent.sleep(0)
res = a.get()
print(f'{datetime.now()-st} {res}')
return str(res)

jobs = [gevent.spawn(test, timeout) for timeout in [0.1, 0.1, 0.1, 0.1, 0.1, 0.1]]
gevent.sleep(0)
gevent.joinall(jobs)
print(f'{datetime.now()-st} finish')

now run gevent_example.py

0:00:00.015999 sleep.delay(0.1)
0:00:00.055000 sleep.delay(0.1)
0:00:00.055000 sleep.delay(0.1)
0:00:00.055999 sleep.delay(0.1)
0:00:00.055999 sleep.delay(0.1)
0:00:00.056997 sleep.delay(0.1)
0:00:01.226389 task 3a4074c3-294b-4bfc-9139-8ed8724ef564 sleep(0.1) finished
0:00:01.226389 task e69e3232-f925-4a9e-afbc-13add65e5806 sleep(0.1) finished
0:00:01.226389 task 23c40dd4-6083-4d04-aac9-3c23e8508912 sleep(0.1) finished
0:00:01.241475 task df35ca20-8e7d-47b0-9fe8-13fe6b1a7a65 sleep(0.1) finished
0:00:01.241475 task 76bf5b84-fe4b-4791-9686-2dd3018221b0 sleep(0.1) finished
0:00:01.241475 task 2fc0ae38-2521-4371-8caf-88d36ea53eff sleep(0.1) finished
0:00:01.241475 finish

It is seen that first task with length 0.1s has completed only after ~1.23s (0:00:01.226389)

But if comment line with monkey patching:

# from gevent import monkey; monkey.patch_all()

0:00:00.015999 sleep.delay(0.1)
0:00:00.226367 sleep.delay(0.1)
0:00:00.228763 sleep.delay(0.1)
0:00:00.231746 sleep.delay(0.1)
0:00:00.233513 sleep.delay(0.1)
0:00:00.234733 sleep.delay(0.1)
0:00:00.338647 task 38f76488-d956-4671-be14-e43df9a593aa sleep(0.1) finished
0:00:00.349064 task 5222feed-6a49-4a55-8a47-f516c3642c30 sleep(0.1) finished
0:00:00.354673 task 87529c14-50a2-4ab3-8e3f-b1f9d7241c42 sleep(0.1) finished
0:00:00.361298 task af68327d-6c37-44e4-9ffa-388ef7c8a8e7 sleep(0.1) finished
0:00:00.363136 task 390980d6-e99d-40b4-afa5-241080ade7a4 sleep(0.1) finished
0:00:00.364271 task f9115418-3175-4caa-98d2-2b39be4ace41 sleep(0.1) finished
0:00:00.364271 finish

It is seen that first task with length 0.1s has completed just after ~0.34s (0:00:00.338647)

with proposed changes in backends.asynchronous become:

0:00:00.016001 sleep.delay(0.1)
0:00:00.054004 sleep.delay(0.1)
0:00:00.055003 sleep.delay(0.1)
0:00:00.055003 sleep.delay(0.1)
0:00:00.056002 sleep.delay(0.1)
0:00:00.056002 sleep.delay(0.1)
0:00:00.322748 task 00a71b72-c12d-40fe-b7e1-7526120c25ad sleep(0.1) finished
0:00:00.325041 task fa31d127-35c9-42f8-a5a4-0d1e29172b59 sleep(0.1) finished
0:00:00.346190 task 3ecb8a4e-78e2-471d-b6ac-e84f9b2c44ac sleep(0.1) finished
0:00:00.349314 task eaa27232-37b1-42ed-87f5-e92c19e87aed sleep(0.1) finished
0:00:00.351648 task 48ed06c8-3683-4353-88e1-f35aeb870332 sleep(0.1) finished
0:00:00.353663 task 8e95e4b7-6988-4b00-b93d-d4240eb38645 sleep(0.1) finished
0:00:00.353663 finish

It is seen that first task with length 0.1s has completed just after ~0.32s (0:00:00.322748)

Issue 2:
Checklist

I have checked the issues list

for similar or identical bug reports.

I have checked the pull requests list

for existing proposed fixes.

I have checked the commit log

to find out if the bug was already fixed in the master branch.

I have included all related issues and possible duplicate issues in this issue

(If there are none, check this box anyway).

Related Issues and Possible Duplicates

Related Issues

None

Possible Duplicates

None

Description

The serialization docs include a section on msgpack:

Specifically I'm referring to

It’s very young however, and support should be considered experimental at this point.

It seems this sentence was added 12 years back in this commit, which seems to have been around version 2.2.0.

Suggestions

I think it'd be good to revise the docs. I don't think this is accurate any longer. Do you agree? 🙂