Kue: Stop Processing Jobs

[danpalmer]

There does not seem to be the ability to stop processing jobs, it would be nice to have this.

[cmawhorter]

I second this. There doesn't seem to be a way to shift the next single job off the queue and return it to an http request, for example.

Being able to do something like:

jobs.process(function(job, done) {
  done();
  return false; // stop at this one execution
});

And here's some code to get someone started with a proper pull request. If I get some time I may do it myself:

Worker.js:

Worker.prototype.process = function(job, fn){
  var self = this
    , start = new Date
    , result;
  job.active();
  result = fn(job, function(err) { // store result of kue.process callback
    if (err) return self.failed(job, err, fn);
    job.complete();
    job.set('duration', job.duration = new Date - start);
    self.emit('job complete', job);
    events.emit(job.id, 'complete');
    process.nextTick(function() { 
      // any non-false value results in the next job being processed (continuously)
      // return false stops processing
      if (false !== result) self.start(fn); 
    });
  });
  return this;
};

[juniorplenty]

Submitted a pull request for @cmawhorter code above. One issue I've noticed: it seems as though, when running concurrent processing (60 in my case,) is that even after I start returning false from the process function, I keep getting more jobs queued, and then eventually it drops off and stops processing. I have a SIGTERM handler that causes my processor to return true and trigger this code path, but it keeps processing jobs for quite a while before falling off. E.g., w/60 concurrent jobs, ~260 jobs complete before it finally exits.

So, a couple of other suggested fixes:

• have the done() handler accept a boolean on whether or not to queue another job. This coule be implemented in the same patch as above
• better (maybe), have kue set up a SIGTERM handler (or SIGUSRx) to stop queueing new jobs (maybe a config param or something if there's a conflict)

[cmawhorter]

Closed without comment?

[behrad]

As of 0.7.x, kue has the ability to pause job processing within an specific worker.

queue.process( 'video-convert', function( job, done, ctx ){
   done();
   ctx.pause( function(){
      console.log( "worker stopped within 3 secs" );
   }, 3000 );
});

I'll update docs for sure.

[knutole]

Still not possible to cancel jobs?

kue.Job.get(job_id, function (err, job) {
    job.failed().error(new Error('whatever')); // doesn't work
    job.failed(callback); // doesn't work
});

How can I manually cancel a running job? I'm doing image processing, and sometimes I need to cancel heavy duty tasks. Thanks

[behrad]

Kue currently doesn't support canceling a running job, the only way to do this is using done to mark current job as failed/completed.

[ineedthekeyboard]

I've spent several days now trying without success to cancel heavy duty jobs. Is there any update to this thread, is Kue still unable to cancel a running job?

I have tried several other methods recommended by @Behrad to no avail (such as spoken about in issue #942 #130 and others)

Even if I do mark the job as done..or listen for a completed message on the job, Kue takes no action until the event look it freed up by the job.

Am looking to see if anyone has any ideas on how to cancel a event-loop blocking job midway through?

[cptrodgers]

I've spent several days now trying without success to cancel heavy duty jobs. Is there any update to this thread, is Kue still unable to cancel a running job?

I have tried several other methods recommended by @behrad to no avail (such as spoken about in issue #942 #130 and others)

Even if I do mark the job as done..or listen for a completed message on the job, Kue takes no action until the event look it freed up by the job.

Am looking to see if anyone has any ideas on how to cancel a event-loop blocking job midway through?

We can use semaphore style for mark job dont use resource and call next_job. I don't read your source code. But with many concurrent processing. I think we need semaphore.

In short: User must use semaphore for mark resource done. Store in redis or update to job by job_id.