Contributors mailing list archives
contributors@odoo-community.org
Browse archives
queue_job commit
by "Martin Fraňo" <waky007@gmail.com> - 09/06/2022 10:05:40Hello,
I would like to commit fix for queue_job but am not able to do that with git. Is it enough if I send it to you in email so I don't break something in the repo itself? The description and code itself is below.
Thanks!
queue_job: split autovacuum execution
Sometimes autovacuum method reach thread max execution time and fail (for us it is 120 seconds). Therefore stop its execution if takes long time and run it again as a new job.
Sometimes autovacuum method reach thread max execution time and fail (for us it is 120 seconds). Therefore stop its execution if takes long time and run it again as a new job.
@@ -282,13 +282,19 @@ class QueueJob(models.Model):
"""Delete all jobs done based on the removal interval defined on the
channel
Called from a cron.
"""
+ start = datetime.now()
+ time_exceeded = False
for channel in self.env["queue.job.channel"].search([]):
deadline = datetime.now() - timedelta(days=int(channel.removal_interval))
while True:
+ if (datetime.now() - start).total_seconds() > 60:
+ # we don't want to have thread max execution time exceeded exception
+ time_exceeded = True
+ break
jobs = self.search(
[
("date_done", "<=", deadline),
("channel", "=", channel.complete_name),
],
@@ -296,10 +302,15 @@ class QueueJob(models.Model):
)
if jobs:
jobs.unlink()
else:
break
+ if time_exceeded:
+ break
+ if time_exceeded:
+ # there is probably still something to cleanup so run it again
+ self.env.ref("queue_job.ir_cron_autovacuum_queue_jobs").with_delay().method_direct_trigger()
return True
def requeue_stuck_jobs(self, enqueued_delta=5, started_delta=0):
"""Fix jobs that are in a bad states
"""Delete all jobs done based on the removal interval defined on the
channel
Called from a cron.
"""
+ start = datetime.now()
+ time_exceeded = False
for channel in self.env["queue.job.channel"].search([]):
deadline = datetime.now() - timedelta(days=int(channel.removal_interval))
while True:
+ if (datetime.now() - start).total_seconds() > 60:
+ # we don't want to have thread max execution time exceeded exception
+ time_exceeded = True
+ break
jobs = self.search(
[
("date_done", "<=", deadline),
("channel", "=", channel.complete_name),
],
@@ -296,10 +302,15 @@ class QueueJob(models.Model):
)
if jobs:
jobs.unlink()
else:
break
+ if time_exceeded:
+ break
+ if time_exceeded:
+ # there is probably still something to cleanup so run it again
+ self.env.ref("queue_job.ir_cron_autovacuum_queue_jobs").with_delay().method_direct_trigger()
return True
def requeue_stuck_jobs(self, enqueued_delta=5, started_delta=0):
"""Fix jobs that are in a bad states
Code of the autovacuum method:
def autovacuum(self):
"""Delete all jobs done based on the removal interval defined on the
channel
Called from a cron.
"""
start = datetime.now()
time_exceeded = False
for channel in self.env["queue.job.channel"].search([]):
deadline = datetime.now() - timedelta(days=int(channel.removal_interval))
while True:
if (datetime.now() - start).total_seconds() > 60:
# we don't want to have thread max execution time exceeded exception
time_exceeded = True
break
jobs = self.search(
[
("date_done", "<=", deadline),
("channel", "=", channel.complete_name),
],
limit=1000,
)
if jobs:
jobs.unlink()
else:
break
if time_exceeded:
break
if time_exceeded:
# there is probably still something to cleanup so run it again
self.env.ref("queue_job.ir_cron_autovacuum_queue_jobs").with_delay().method_direct_trigger()
return True
"""Delete all jobs done based on the removal interval defined on the
channel
Called from a cron.
"""
start = datetime.now()
time_exceeded = False
for channel in self.env["queue.job.channel"].search([]):
deadline = datetime.now() - timedelta(days=int(channel.removal_interval))
while True:
if (datetime.now() - start).total_seconds() > 60:
# we don't want to have thread max execution time exceeded exception
time_exceeded = True
break
jobs = self.search(
[
("date_done", "<=", deadline),
("channel", "=", channel.complete_name),
],
limit=1000,
)
if jobs:
jobs.unlink()
else:
break
if time_exceeded:
break
if time_exceeded:
# there is probably still something to cleanup so run it again
self.env.ref("queue_job.ir_cron_autovacuum_queue_jobs").with_delay().method_direct_trigger()
return True
Kind regards,
Martin
