fork/django-q2
1
0
Fork 0
mirror of https://github.com/django-q2/django-q2.git synced 2026-10-06 15:48:12 +08:00
Code Issues Packages Projects Releases Wiki Activity

Lock the Schedule objects to ensure tasks are created once

Browse Source 
Tested this with PostgreSQL and MySQL
This commit is contained in:
Martijn Jacobs
2019-08-12 14:28:26 +02:00
parent abaa0ba39b
commit 924ff928b0
1 changed files with 63 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files
django_q/cluster.py
+63 -64
View File
@@ -491,70 +491,69 @@ def scheduler(broker=None):
broker = get_broker() broker = get_broker()
db.close_old_connections() db.close_old_connections()
try: try:
for s in Schedule.objects.exclude(repeats=0).filter(next_run__lt=timezone.now()): with db.transaction.atomic():
args = () for s in Schedule.objects.select_for_update().exclude(repeats=0).filter(next_run__lt=timezone.now()):
kwargs = {} args = ()
# get args, kwargs and hook kwargs = {}
if s.kwargs: # get args, kwargs and hook
try: if s.kwargs:
# eval should be safe here because dict() try:
kwargs = eval('dict({})'.format(s.kwargs)) # eval should be safe here because dict()
except SyntaxError: kwargs = eval('dict({})'.format(s.kwargs))
kwargs = {} except SyntaxError:
if s.args: kwargs = {}
args = ast.literal_eval(s.args) if s.args:
# single value won't eval to tuple, so: args = ast.literal_eval(s.args)
if type(args) != tuple: # single value won't eval to tuple, so:
args = (args,) if type(args) != tuple:
q_options = kwargs.get('q_options', {}) args = (args,)
if s.hook: q_options = kwargs.get('q_options', {})
q_options['hook'] = s.hook if s.hook:
# set up the next run time q_options['hook'] = s.hook
if not s.schedule_type == s.ONCE: # set up the next run time
next_run = arrow.get(s.next_run) if not s.schedule_type == s.ONCE:
while True: next_run = arrow.get(s.next_run)
if s.schedule_type == s.MINUTES: while True:
next_run = next_run.shift(minutes=+(s.minutes or 1)) if s.schedule_type == s.MINUTES:
elif s.schedule_type == s.HOURLY: next_run = next_run.shift(minutes=+(s.minutes or 1))
next_run = next_run.shift(hours=+1) elif s.schedule_type == s.HOURLY:
elif s.schedule_type == s.DAILY: next_run = next_run.shift(hours=+1)
next_run = next_run.shift(days=+1) elif s.schedule_type == s.DAILY:
elif s.schedule_type == s.WEEKLY: next_run = next_run.shift(days=+1)
next_run = next_run.shift(weeks=+1) elif s.schedule_type == s.WEEKLY:
elif s.schedule_type == s.MONTHLY: next_run = next_run.shift(weeks=+1)
next_run = next_run.shift(months=+1) elif s.schedule_type == s.MONTHLY:
elif s.schedule_type == s.QUARTERLY: next_run = next_run.shift(months=+1)
next_run = next_run.shift(months=+3) elif s.schedule_type == s.QUARTERLY:
elif s.schedule_type == s.YEARLY: next_run = next_run.shift(months=+3)
next_run = next_run.shift(years=+1) elif s.schedule_type == s.YEARLY:
if Conf.CATCH_UP or next_run > arrow.utcnow(): next_run = next_run.shift(years=+1)
break if Conf.CATCH_UP or next_run > arrow.utcnow():
# arrow always returns a tz aware datetime, and we don't want break
# this when we explicitly configured django with USE_TZ=False s.next_run = next_run.datetime
s.next_run = next_run.datetime if settings.USE_TZ else next_run.datetime.replace(tzinfo=None) s.repeats += -1
s.repeats += -1 # send it to the cluster
# send it to the cluster q_options['broker'] = broker
q_options['broker'] = broker q_options['group'] = q_options.get('group', s.name or s.id)
q_options['group'] = q_options.get('group', s.name or s.id) kwargs['q_options'] = q_options
kwargs['q_options'] = q_options s.task = django_q.tasks.async_task(s.func, *args, **kwargs)
s.task = django_q.tasks.async_task(s.func, *args, **kwargs) # log it
# log it if not s.task:
if not s.task: logger.error(
logger.error( _('{} failed to create a task from schedule [{}]').format(current_process().name,
_('{} failed to create a task from schedule [{}]').format(current_process().name, s.name or s.id))
s.name or s.id)) else:
else: logger.info(
logger.info( _('{} created a task from schedule [{}]').format(current_process().name, s.name or s.id))
_('{} created a task from schedule [{}]').format(current_process().name, s.name or s.id)) # default behavior is to delete a ONCE schedule
# default behavior is to delete a ONCE schedule if s.schedule_type == s.ONCE:
if s.schedule_type == s.ONCE: if s.repeats < 0:
if s.repeats < 0: s.delete()
s.delete() continue
continue # but not if it has a positive repeats
# but not if it has a positive repeats s.repeats = 0
s.repeats = 0 # save the schedule
# save the schedule s.save()
s.save()
except Exception as e: except Exception as e:
logger.error(e) logger.error(e)