Issue 1:
Checklist

I have verified that the issue exists against the master branch of Celery.

This has already been asked to the discussions forum first.

I have read the relevant section in the

contribution guide

on reporting bugs.

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).

Mandatory Debugging Information

I have included the output of celery -A proj report in the issue.

(if you are not able to do this, then at least specify the Celery

version affected).

I have verified that the issue exists against the master branch of Celery.

I have included the contents of pip freeze in the issue.

I have included all the versions of all the external dependencies required

to reproduce this bug.

Optional Debugging Information

I have tried reproducing the issue on more than one Python version

and/or implementation.

I have tried reproducing the issue on more than one message broker and/or

result backend.

I have tried reproducing the issue on more than one version of the message

broker and/or result backend.

I have tried reproducing the issue on more than one operating system.

I have tried reproducing the issue on more than one workers pool.

I have tried reproducing the issue with autoscaling, retries,

ETA/Countdown & rate limits disabled.

I have tried reproducing the issue after downgrading

and/or upgrading Celery and its dependencies.

Related Issues and Possible Duplicates

Related Issues

None

Possible Duplicates

None

Environment & Settings

4.4.7 (cliffs)

Celery version:

celery report Output:

software -> celery:4.4.7 (cliffs) kombu:4.6.11 py:3.6.12
billiard:3.6.3.0 redis:3.5.3
platform -> system:Linux arch:64bit
kernel version:4.4.0-1098-aws imp:CPython
loader -> celery.loaders.app.AppLoader
settings -> transport:redis results:disabled

Steps to Reproduce

Required Dependencies

Minimal Python Version: N/A or Unknown

Minimal Celery Version: 4.0.0

Minimal Kombu Version: N/A or Unknown

Minimal Broker Version: N/A or Unknown

Minimal Result Backend Version: N/A or Unknown

Minimal OS and/or Kernel Version: N/A or Unknown

Minimal Broker Client Version: N/A or Unknown

Minimal Result Backend Client Version: N/A or Unknown

Python Packages

pip freeze Output:

amqp==2.6.1
anyjson==0.3.3
apns2==0.3.0
appdirs==1.4.4
args==0.1.0
asn1crypto==1.4.0
attrs==21.4.0
backports.shutil-get-terminal-size==1.0.0
beautifulsoup4==4.6.0
billiard==3.6.3.0
boto==2.48.0
cached-property==1.5.2
cairocffi==1.1.0
CairoSVG==2.4.0
celery==4.4.7
certifi==2018.1.18
cffi==1.14.3
chardet==3.0.4
clint==0.5.1
cryptography==2.9.2
cssselect==1.0.0
cssselect2==0.2.2
cssutils==1.0.2
decorator==4.2.1
defusedxml==0.6.0
diff-match-patch==20121119
Django==1.11.11
django-admin-bootstrapped-rubgomez93==3.0.1
django-ajax-selects==1.7.0
django-bootstrap3-datetimepicker-rubgomez93==3.0.2
django-boto==0.3.12
django-constance==2.1.0
django-cookies-samesite==0.5.1
django-cors-headers==2.2.0
django-countries==5.5
django-dbconn-retry==0.1.5
django-debug-toolbar==1.9.1
django-filter==1.1.0
django-form-admin==0.5.1
django-import-export==1.0.0
django-localflavor==2.0
django-money==0.12.3
django-multiselectfield==0.1.8
django-p3p==1.2.0
django-picklefield==1.0.0
django-proxy==1.1.0
django-push-notifications-rubgomez93==1.6.4
django-redis==4.11.0
django-session-timeout-joinup==1.0.0
django-storages==1.6.5
django-su==0.6.0
django-tables2==2.3.1
django-timezone-field==2.1
djangorestframework==3.6.4
djangorestframework-gis==0.11.2
djangorestframework-jwt==1.11.0
djangorestframework-xml==1.3.0
ecdsa==0.13
eight==0.4.2
endesive==1.5.12
enum34==1.1.6
et-xmlfile==1.0.1
forex-python==1.3
funcsigs==1.0.2
future==0.16.0
geopy==1.11.0
google-api-python-client==1.6.5
gunicorn==19.7.1
h2==2.6.2
hpack==3.0.0
html5lib==1.0.1
http-ece==1.0.5
httplib2==0.18.1
hyper==0.7.0
hyperframe==3.2.0
idna==2.6
importlib-metadata==4.8.3
importlib-resources==5.4.0
ipaddress==1.0.19
ipython-genutils==0.2.0
iso3166==2.0.2
isodate==0.6.1
jdcal==1.3
jsonfield-joinup==2.1.2
kombu==4.6.11
lxml==4.1.1
mock==2.0.0
numpy==1.19.5
oauth2client==4.1.2
odfpy==1.3.6
olefile==0.45.1
openpyxl==2.4.9
oscrypto==1.2.1
paramiko==2.1.1
pathlib2==2.3.0
pbr==3.1.1
pdfrw==0.4
pexpect==4.4.0
PGPy==0.5.3
phonenumbers==8.9.0
pickleshare==0.7.4
Pillow==4.1.1
polyline==1.4.0
prompt-toolkit==2.0.10
psutil==5.6.1
psycopg2==2.7.4
psycopg2-binary==2.7.4
ptyprocess==0.5.2
py-moneyed==0.7.0
py-vapid==1.3.0
pyasn1==0.4.2
pyasn1-modules==0.2.1
pycairo==1.16.3
pycparser==2.20
Pygments==2.2.0
PyJWT==1.6.0
PyKCS11==1.5.9
pymongo==3.10.1
pynliner==0.8.0
pyOpenSSL==17.5.0
Pyphen==0.9.4
pysftp==0.2.9
PySocks==1.7.1
python-dateutil==2.6.1
pytz==2018.3
pywebpush==1.6.0
PyYAML==3.12
redis==3.5.3
requests==2.22.0
requests-toolbelt==0.9.1
rsa==3.4.2
scandir==1.7
schwifty==2020.2.2
signxml==2.5.2
simplegeneric==0.8.1
simplejson==3.16.0
six==1.15.0
smsutil==1.1.3
sqlparse==0.2.4
static3==0.7.0
stripe==2.32.1
tablib==0.12.1
timezonefinder==3.4.2
tinycss==0.4
tinycss2==1.0.0
traitlets==4.3.2
typing-extensions==4.0.1
unicodecsv==0.14.1
uritemplate==3.0.0
urllib3==1.22
vine==1.3.0
wcwidth==0.1.7
WeasyPrint==51
webencodings==0.5.1
whitenoise==3.3.1
xlrd==1.1.0
XlsxWriter==1.0.2
xlwt==1.3.0
zeep==3.4.0
zipp==3.6.0

Other Dependencies

N/A

Minimally Reproducible Test Case

I have added a unit test (test_pending_left_10_milliseconds_tick) in my fork. This test fails with current master branch, but this does not fails with my forked master branch

Expected Behavior

Celery beat executes tasks periodically and sleeps until the next task is dued. So we expect celery beat periodically executes the corresponding tasks without the process sleeping more than the necessary time between task executions.

For our case:

The task that we scheduled with the lowest value is 13 seconds.

Celery beat should not sleep more than 13 seconds in any case.

Actual Behavior

Sometimes celery beat calculates wrong that the time to sleep is 300 seconds (default value for max interval). In our case, it pass kind of one time per day.

How it is possible?:

If the next task to run will due in 0.01 seconds, so this tasks is not dued still, the tick function will calculate the time that celery beat process should sleep.

To calculate it, tick function calls to the function adjust, this subtracts 0.01 to next_time_to_run variable. The result of this operation is 0.

The return value of the tick function will be maximum between:

0 or max interval => max interval

max interval

So, celery beat will sleep wrongly for max interval (300 seconds)

I have added comments to these functions with the tag [ISSUE_COMMENT] to describe this case

def adjust(self, n, drift=-0.010):
if n and n > 0:
# [ISSUE_COMMENT] if n is 0.01 this function returns 0.01 + (-0.010) = 0.0
return n + drift
return n

def tick(self, event_t=event_t, min=min,
heappop=heapq.heappop, heappush=heapq.heappush,
heapify=heapq.heapify, mktime=time.mktime):

...

event = H[0]
entry = event[2]
# [ISSUE_COMMENT] when exactly lefts 0.01 seconds to run the next task => is_due = False ; next_time_to_run = 0.01
is_due, next_time_to_run = self.is_due(entry)
if is_due:
verify = heappop(H)

...

# [ISSUE_COMMENT] if next_time_to_run = 0.01, adjust(next_time_to_run) = 0 => 0 or max_interval = max_interval ; max(max_interval, max_interval) = max_interval
# [ISSUE_COMMENT] max_interval by default is 300 seconds
return min(adjust(next_time_to_run) or max_interval, max_interval)

Issue 2:
I am currently running celery 4.0.2 with a single worker like this:

celery.py:

app = Celery('project',
broker='amqp://jimmy:jimmy123@localhost/jimmy_vhost',
backend='rpc://',
include=['project.tasks'])

if __name__ == '__main__':
app.start()
app.name

tasks.py:

from .celery import app
from celery.schedules import schedule
from time import sleep, strftime

app.conf.beat_schedule = {
'planner_1': {
'task': 'project.tasks.call_orders',
'schedule': 1800,
},
'planner_2': {
'task': 'project.tasks.call_inventory',
'schedule': 900,
},
}

I used the following command to run with beat:

celery -A project worker -l info --concurrency=3 --beat -E

Right now it is only a single queue with only one worker running.

My question is how to run celery with multiple workers and single queue so that tasks are executed in parallel using multiprocessing without duplication?

I looked up on the internet, how to run celery with multiprocessing. According to this article:

celery worker -l info -P processes -c 16 will result in a single message consumer delegating work to 16 OS-level pool processes. Each OS-level process can be assigned to different CPU in a multicore environment, and as such it will process tasks in parallel, but it will not consume messages in parallel.

Can using the -p processes argument solve my problem? Also but what is meant by, "it will process tasks in parallel, but it will not consume messages in parallel"?