Skip to content

Locked jobs - #137

Merged
sylvinus merged 13 commits into
masterfrom
locked_jobs
Dec 9, 2016
Merged

Locked jobs#137
sylvinus merged 13 commits into
masterfrom
locked_jobs

Conversation

@ggueret

@ggueret ggueret commented Dec 1, 2016

Copy link
Copy Markdown
Contributor
  • Added 'expired' status for the jobs who were unable to lock
  • Added 'locked_job' bool option to mrq.task.Task, to lock a job execution by his task path.
  • Mark mrq.basetasks.cleaning.[RequeueInterruptedJobs, RequeueRetryJobs, RequeueStartedJobs, RequeueRedisStartedJobs, RequeueLostJobs, MigrateKnownQueues, CleanKnownQueues, ] as locked jobs.

@sylvinus

sylvinus commented Dec 2, 2016

Copy link
Copy Markdown
Contributor
  • locked_job is probably not the best name
  • lock_timeout is not explicit enough (this lock should have a specific name to avoid confusion)
  • The locking code is not safe enough
  • I'm not (yet) convinced we need to add a new status expired just for this feature.

Comment thread mrq/basetasks/cleaning.py

""" Requeue jobs that were marked as status=interrupt when a worker got a SIGTERM. """

max_concurrency = 1

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

"maxconcurrency" serait plus facile à grepper ?

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

c'est par respect de la sémantique existante, avec par exemple le Job.max_retries

Comment thread mrq/config.py Outdated
'to dequeue them in command-line order.')

parser.add_argument(
'--lock_timeout',

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

--maxconcurrency_timeout

Comment thread mrq/config.py Outdated
default=1.,
type=float,
action='store',
help='Max seconds to wait for a lock before changing the job status to expired.')

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

text to update

Comment thread mrq/job.py Outdated
pipeline = context.connections.redis.pipeline()
pipeline.incr(self.redis_concurrency_key)
pipeline.expireat(self.redis_concurrency_key, int(time.time()) + self.timeout)
current = pipeline.execute()[0]

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this is not safe enough - with maxconcurrency=2, if a worker dies here, the count will stay at a minimum at 1, and if there are still tasks happening, the key will never expire.

@ggueret

ggueret commented Dec 7, 2016

Copy link
Copy Markdown
Contributor Author
  • Added 'maxconcurrency' status for the jobs who reach the max_concurrency limit.
  • Replaced 'locked_job' by 'max_concurrency' who now take an integer value.
  • Mark mrq.basetasks.cleaning.[RequeueInterruptedJobs, RequeueRetryJobs, RequeueStartedJobs, RequeueRedisStartedJobs, RequeueLostJobs, MigrateKnownQueues, CleanKnownQueues] with a concurrency of 1 per Job.timeout.

@sylvinus
sylvinus merged commit 3f587f4 into master Dec 9, 2016
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants