Do task beanstalkd#586 - job timeouts.

beanstalkd.c

@@ -326,7 +326,7 @@ reset_conn(conn c)
 }
 
 static void
-handle_connection(conn c)
+h_conn_data(conn c)
 {
  int r, free_reply_buf = 0;
  job j;
@@ -413,6 +413,19 @@ handle_connection(conn c)
  }
 }
 
+/* if we get a timeout, it means that a job has been reserved for too long, so
+ * we should put it back in the queue */
+static void
+h_conn_timeout(conn c)
+{
+ job j = c->reserved_job;
+
+ if (!j) return;
+
+ enqueue_job(j);
+ c->reserved_job = NULL;
+}
+
 #define want_command(c) ((c)->fd && ((c)->state == STATE_WANTCOMMAND))
 #define cmd_data_ready(c) (want_command(c) && (c)->cmd_read)
 
@@ -425,7 +438,16 @@ h_conn(const int fd, const short which, conn c)
  return conn_close(c);
  }
 
- handle_connection(c);
+ switch (which) {
+ case EV_TIMEOUT:
+ h_conn_timeout(c);
+ break;
+ case EV_READ:
+ /* fall through... */
+ case EV_WRITE:
+ h_conn_data(c);
+ }
+
  while (cmd_data_ready(c) && (c->cmd_len = cmd_len(c))) do_cmd(c);
 }
 

conn.c

@@ -2,6 +2,7 @@
 
 #include <stdlib.h>
 #include <stdio.h>
+#include <time.h>
 
 #include "conn.h"
 #include "net.h"
@@ -99,10 +100,13 @@ int
 conn_set_evq(conn c, const int events, evh handler)
 {
  int r;
+ struct timeval tv = {0, 0};
 
  event_set(&c->evq, c->fd, events, handler, c);
 
- r = event_add(&c->evq, NULL);
+ if (c->reserved_job) tv.tv_sec = c->reserved_job->deadline - time(NULL);
+
+ r = event_add(&c->evq, c->reserved_job ? &tv : NULL);
  if (r == -1) return -1;
 
  return 0;

job.h

@@ -3,10 +3,13 @@
 #ifndef job_h
 #define job_h
 
+#include <time.h>
+
 typedef struct job {
  unsigned long long int id;
  unsigned int pri;
  int data_size;
+ time_t deadline;
  unsigned char data[];
 } *job;
 

prot.c

@@ -58,6 +58,7 @@ reply_job(conn c, job j, const char *word)
 static void
 reserve_job(conn c, job j)
 {
+ j->deadline = time(NULL) + RESERVATION_TIMEOUT;
  c->reserved_job = j;
  conn_insert(&running_list, c);
  return reply_job(c, j, MSG_RESERVED);

prot.h

@@ -9,6 +9,9 @@
 /* space for 16 Mi jobs */
 #define HEAP_SIZE 16 * 1024 * 1024
 
+/* measured in seconds */
+#define RESERVATION_TIMEOUT 120
+
 #define MSG_RESERVED "RESERVED"
 
 void prot_init();