Received: by 10.58.132.145 with SMTP id ou17mr7598415veb.41.1352813478070;
Tue, 13 Nov 2012 05:31:18 -0800 (PST)
X-BeenThere: celery-users@googlegroups.com
Received: by 10.220.222.132 with SMTP id ig4ls444997vcb.0.gmail; Tue, 13 Nov
2012 05:31:15 -0800 (PST)
Received: by 10.52.65.33 with SMTP id u1mr4383614vds.18.1352813475567;
Tue, 13 Nov 2012 05:31:15 -0800 (PST)
Date: Tue, 13 Nov 2012 05:31:14 -0800 (PST)
From: Darryl Lickt <darryl.li...@gmail.com>
To: celery-users@googlegroups.com
Message-Id: <a9b4d01c-5eb6-489c-b05e-3988baf07e55@googlegroups.com>
In-Reply-To: <3fb6a458-c3d1-4f38-92b8-6312179f8c96@googlegroups.com>
References: <7672b096-19a1-48b1-aa4b-638eef0cd009@y8g2000yqy.googlegroups.com>
<7D3AD0B0-DE4F-41E9-BA54-D55A61CBC847@celeryproject.org>
<3fb6a458-c3d1-4f38-92b8-6312179f8c96@googlegroups.com>
Subject: Re: [celery-users] all workers reserve two tasks even with acks
late = True and prefetch multiplier = 1
MIME-Version: 1.0
Content-Type: multipart/mixed;
boundary="----=_Part_316_9555911.1352813475003"
------=_Part_316_9555911.1352813475003
Content-Type: multipart/alternative;
boundary="----=_Part_317_14688901.1352813475003"
------=_Part_317_14688901.1352813475003
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 7bit
This error was actually self inflicted. There were some other parts of my
code that were overwriting the settings I was trying to use.
On a different note however, I can't get the late acking to work correctly.
I'm thinking it might have something to do with raising SystemExit from the
task_postrun signal.
The problem I'm seeing now is that once a task completes, it never gets
ack'd again but instead gets added back into the queue in the "ready" state.
So I'll have something like this:
ready unackd total
2 0 0 - queued two tasks
0 2 0 - workers start executing both tasks (fine, they
shouldn't be ack'd until it's complete)
2 0 0 - tasks have been run, workers have shut down, but
tasks get put back in ready queue
Make sense? Or is this maybe the default retry settings doing something
weird because it caught the SystemExit exception I raised?
On Friday, November 9, 2012 8:25:28 AM UTC-6, Darryl Lickt wrote:
>
> The worker exiting is working just fine. The problem I'm getting stuck on
> is that the worker prefetches two tasks instead of only one. I've been
> monitoring this by looking at the queue for the task in the RabbitMQ
> Management web interface. If I queue up two tasks and two workers, the
> first worker to connect will prefetch both tasks. The second worker to
> connect has to wait until the first worker shuts down and releases the
> unack'd second task that it greedily fetched.
>
> In the RabbitMQ web interface, it goes something like this:
>
> ready unackd total
> 2 0 0 - queued two tasks
> 0 1 0 - the first worker connects and prefetched both
> tasks
> 1 0 0 - the first worker finished, 2nd task goes back
> to ready
> 0 0 0 - second worker consumes the 2nd task after
>
> The late ack isn't quite working the way I thought it would either. If the
> worker prefetched two tasks somehow (instead of one), I would've thought
> that both of them would have stayed in the unackd state until one of the
> tasks was finished.
>
> I at first thought that maybe the late_ack setting was getting clobbered
> somewhere, but it seems like it's getting set.
>
> Any thoughts? Thanks for your time and help
>
>
> On Friday, November 9, 2012 7:55:45 AM UTC-6, Ask Solem wrote:
>>
>>
>> On 8 Nov 2012, at 18:28, Darryl Lickt <darryl...@gmail.com> wrote:
>>
>> > I'm current using a setup where I want each worker to *ONLY* prefetch
>> > one task, execute it, and then shutdown. These tasks are potentially
>> > very long running tasks.
>> >
>> > Below is how I'm currently running the workers:
>> >
>> > from celery import Celery
>> > from celery import signals
>> >
>> > app = Celery('tasks',
>> > set_as_current = True,
>> > broker='amqp://guest@localhost//',
>> > backend="mongodb://localhost//",
>> > )
>> >
>> > app.config_from_object({
>> > "CELERY_ACKS_LATE": True,
>> > "CELERY_SEND_EVENTS": True,
>> > "CELERY_DISABLE_RATE_LIMITS": True,
>> >
>> > "CELERYD_MAX_TASKS_PER_CHILD": 1,
>> > "CELERYD_POOL": "solo",
>> > "CELERYD_CONCURRENCY": 1,
>> > "CELERYD_PREFETCH_MULTIPLIER": 1,
>> > })
>> >
>> > def shutdown_worker(**kwargs):
>> > print("SHUTTING DOWN WORKER (RAISING SystemExit)")
>> > raise SystemExit()
>> >
>> > import tasks
>> >
>> > signals.task_postrun.connect(shutdown_worker)
>> >
>> > print("STARTING WORKER")
>> > app.worker_main()
>> > print("WORKER EXITED!")
>> >
>> > Everything is working great except that each worker prefetches 2
>> > tasks, instead of one. This leaves workers sitting around, waiting
>> > until those extra fetched tasks are freed up again by the greedy
>> > workers.
>> >
>> > Based on what I've read, my understanding of the code, as well as
>> > other things I've found online (eg
>> >
>> https://groups.google.com/forum/?fromgroups=#!topic/celery-users/pGW1XF6Ipk0),
>>
>> > setting the prefetch multiplier to 1 and setting acks late to True
>> > should have done the trick.
>> >
>> > What am I missing?
>>
>>
>> I'm trying your program above, and I'm not able to reproduce.
>>
>> The worker immediately exits after the first task completes,
>> what are you seeing?
>>
>> --
>> Ask Solem
>> twitter.com/asksol | +44 (0)7713357179
>>
>>
------=_Part_317_14688901.1352813475003
Content-Type: text/html; charset=utf-8
Content-Transfer-Encoding: quoted-printable
This error was actually self inflicted. There were some other parts of my c=
ode that were overwriting the settings I was trying to use.<div><br></div><=
div>On a different note however, I can't get the late acking to work correc=
tly. I'm thinking it might have something to do with raising SystemExit fro=
m the task_postrun signal.</div><div><br></div><div>The problem I'm seeing =
now is that once a task completes, it never gets ack'd again but instead ge=
ts added back into the queue in the "ready" state.</div><div><br></div><div=
>So I'll have something like this:</div><div><br></div><div><div><font face=
=3D"courier new, monospace">ready unackd total</font></=
div><div><font face=3D"courier new, monospace"> 2  =
; 0 0 - queued two tasks</fon=
t></div><div><font face=3D"courier new, monospace"> 0 &=
nbsp; 2 0 - workers start exe=
cuting both tasks (fine, they shouldn't be ack'd until it's complete)</font=
></div><div><font face=3D"courier new, monospace"> 2 &n=
bsp; 0 0 - tasks have been ru=
n, workers have shut down, but tasks get put back in ready queue</font></di=
v><div><br></div></div><div>Make sense? Or is this maybe the default retry =
settings doing something weird because it caught the SystemExit exception I=
raised?</div><div><br></div><div><br></div><div><br>On Friday, November 9,=
2012 8:25:28 AM UTC-6, Darryl Lickt wrote:<blockquote class=3D"gmail_quote=
" style=3D"margin: 0;margin-left: 0.8ex;border-left: 1px #ccc solid;padding=
-left: 1ex;">The worker exiting is working just fine. The problem I'm getti=
ng stuck on is that the worker prefetches two tasks instead of only one. I'=
ve been monitoring this by looking at the queue for the task in the RabbitM=
Q Management web interface. If I queue up two tasks and two workers, the fi=
rst worker to connect will prefetch both tasks. The second worker to connec=
t has to wait until the first worker shuts down and releases the unack'd se=
cond task that it greedily fetched.<div><br></div><div>In the RabbitMQ web =
interface, it goes something like this:</div><div><br></div><div><font face=
=3D"courier new, monospace">ready unackd total</font></=
div><div><font face=3D"courier new, monospace"> 2  =
; 0 0 - queued two tasks</fon=
t></div><div><font face=3D"courier new, monospace"> 0 &=
nbsp; 1 0 - the first worker =
connects and prefetched both tasks</font></div><div><font face=3D"courier n=
ew, monospace"> 1 0 &=
nbsp;0 - the first worker finished, 2nd task goes back to read=
y</font></div><div><font face=3D"courier new, monospace"> 0 &n=
bsp; 0 0 - second work=
er consumes the 2nd task after</font></div><div><font face=3D"courier new, =
monospace"><br></font></div><div><font face=3D"arial, sans-serif">The late =
ack isn't quite working the way I thought it would either. If the worker pr=
efetched two tasks somehow (instead of one), I would've thought that both o=
f them would have stayed in the unackd state until one of the tasks was fin=
ished.</font></div><div><font face=3D"arial, sans-serif"><br></font></div><=
div><font face=3D"arial, sans-serif">I at first thought that maybe the late=
_ack setting was getting clobbered somewhere, but it seems like it's gettin=
g set.</font></div><div><font face=3D"arial, sans-serif"><br></font></div><=
div><font face=3D"arial, sans-serif">Any thoughts? Thanks for your time and=
help</font></div><div><font face=3D"arial, sans-serif"><br></font></div><d=
iv><br></div><div><div>On Friday, November 9, 2012 7:55:45 AM UTC-6, Ask So=
lem wrote:<blockquote class=3D"gmail_quote" style=3D"margin:0;margin-left:0=
.8ex;border-left:1px #ccc solid;padding-left:1ex">
<br>On 8 Nov 2012, at 18:28, Darryl Lickt <<a>darryl...@gmail.com</a>>=
; wrote:
<br>
<br>> I'm current using a setup where I want each worker to *ONLY* prefe=
tch
<br>> one task, execute it, and then shutdown. These tasks are potential=
ly
<br>> very long running tasks.
<br>>=20
<br>> Below is how I'm currently running the workers:
<br>>=20
<br>> from celery import=
Celery
<br>> from celery import=
signals
<br>>=20
<br>> app =3D Celery('ta=
sks',
<br>> =
set_as_current =3D True,
<br>> =
broker=3D'amqp:/<wbr>/guest@localhost//',
<br>> =
backend=3D"<wbr>mongodb://localhost//",
<br>> )
<br>>=20
<br>> app.config_from_<w=
br>object({
<br>> =
"CELERY_ACKS_<wbr>LATE": True,
<br>> =
"CELERY_SEND_<wbr>EVENTS": True,
<br>> =
"CELERY_<wbr>DISABLE_RATE_LIMITS": True,
<br>>=20
<br>> =
"CELERYD_MAX_<wbr>TASKS_PER_CHILD": 1,
<br>> =
"CELERYD_POOL"<wbr>: "solo",
<br>> =
"CELERYD_<wbr>CONCURRENCY": 1,
<br>> =
"CELERYD_<wbr>PREFETCH_MULTIPLIER": 1,
<br>> })
<br>>=20
<br>> def shutdown_worke=
r(**kwargs):
<br>> =
print("<wbr>SHUTTING DOWN WORKER (RAISING Sys=
temExit)")
<br>> =
raise SystemExit()
<br>>=20
<br>> import tasks
<br>>=20
<br>> signals.task_postr=
un.<wbr>connect(shutdown_worker)
<br>>=20
<br>> print("STARTING WO=
RKER")
<br>> app.worker_main()
<br>> print("WORKER EXIT=
ED!")
<br>>=20
<br>> Everything is working great except that each worker prefetches 2
<br>> tasks, instead of one. This leaves workers sitting around, waiting
<br>> until those extra fetched tasks are freed up again by the greedy
<br>> workers.
<br>>=20
<br>> Based on what I've read, my understanding of the code, as well as
<br>> other things I've found online (eg
<br>> <a href=3D"https://groups.google.com/forum/?fromgroups=3D#!topic/c=
elery-users/pGW1XF6Ipk0" target=3D"_blank">https://groups.google.com/<wbr>f=
orum/?fromgroups=3D#!topic/<wbr>celery-users/pGW1XF6Ipk0</a>),
<br>> setting the prefetch multiplier to 1 and setting acks late to True
<br>> should have done the trick.
<br>>=20
<br>> What am I missing?
<br>
<br>
<br>I'm trying your program above, and I'm not able to reproduce.
<br>
<br>The worker immediately exits after the first task completes,
<br>what are you seeing?
<br>
<br>--=20
<br>Ask Solem
<br><a href=3D"http://twitter.com/asksol" target=3D"_blank">twitter.com/ask=
sol</a> | +44 (0)7713357179
<br>
<br></blockquote></div></div></blockquote></div>
------=_Part_317_14688901.1352813475003--
------=_Part_316_9555911.1352813475003--
Message from discussion all workers reserve two tasks even with acks late = True and prefetch multiplier = 1
| Create a group - Google Groups - Google Home - Terms of Service - Privacy Policy |
| ©2013 Google |