Collect most stats. This is for bug beanstalkd#595.

beanstalkd.c

@@ -17,6 +17,9 @@
 /* job data cannot be greater than this */
 #define JOB_DATA_SIZE_LIMIT ((1 << 16) - 1)
 
+static unsigned long long int put_ct = 0, peek_ct = 0, reserve_ct = 0,
+ delete_ct = 0, stats_ct = 0;
+
 static void
 drop_root()
 {
@@ -194,6 +197,8 @@ dispatch_cmd(conn c)
  /* don't allow trailing garbage */
  if (end_buf[0] != '\0') return conn_close(c);
 
+ put_ct++; /* stats */
+
  c->in_job = make_job(pri, data_size + 2);
 
  fill_extra_data(c);
@@ -207,6 +212,8 @@ dispatch_cmd(conn c)
  id = strtoull(c->cmd + CMD_PEEK_LEN, &end_buf, 10);
  if (errno) return conn_close(c);
 
+ peek_ct++; /* stats */
+
  /* So, peek is annoying, because some other connection might free the
  * job while we are still trying to write it out. So we copy it and
  * then free the copy when it's done sending. */
@@ -220,6 +227,8 @@ dispatch_cmd(conn c)
  /* don't allow trailing garbage */
  if (c->cmd_len != CMD_RESERVE_LEN + 2) return conn_close(c);
 
+ reserve_ct++; /* stats */
+
  /* does this conn already have a job reserved? */
  if (c->reserved_job) return reply_job(c, c->reserved_job, MSG_RESERVED);
 
@@ -232,6 +241,8 @@ dispatch_cmd(conn c)
  id = strtoull(c->cmd + CMD_DELETE_LEN, &end_buf, 10);
  if (errno) return conn_close(c);
 
+ delete_ct++; /* stats */
+
  if (!c->reserved_job || id != c->reserved_job->id) {
  reply(c, MSG_NOTFOUND, MSG_NOTFOUND_LEN, STATE_SENDWORD);
  break;
@@ -245,10 +256,12 @@ dispatch_cmd(conn c)
  case OP_STATS:
  /* don't allow trailing garbage */
  if (c->cmd_len != CMD_STATS_LEN + 2) return conn_close(c);
+ stats_ct++; /* stats */
  warn("got stats command");
  return conn_close(c);
  break;
  case OP_JOBSTATS:
+ stats_ct++; /* stats */
  warn("got job stats command");
  return conn_close(c);
  break;

conn.c

@@ -11,6 +11,8 @@
 /* Singly-linked list of free connections (the prev pointer isn't used here). */
 static conn pool_front = NULL, pool_rear = NULL;
 
+int cur_conn_ct = 0, cur_worker_ct = 0, cur_producer_ct = 0;
+
 static int
 pool_conn_p()
 {
@@ -55,13 +57,50 @@ make_conn(int fd, char start_state)
 
  c->fd = fd;
  c->state = start_state;
+ c->type = 0;
  c->cmd_read = 0;
  c->in_job = c->out_job = c->reserved_job = NULL;
  c->in_job_read = c->out_job_sent = 0;
 
+ cur_conn_ct++; /* stats */
+
  return c;
 }
 
+void
+conn_set_producer(conn c)
+{
+ if (c->type & CONN_TYPE_PRODUCER) return;
+ c->type |= CONN_TYPE_PRODUCER;
+ cur_producer_ct++; /* stats */
+}
+
+void
+conn_set_worker(conn c)
+{
+ if (c->type & CONN_TYPE_WORKER) return;
+ c->type |= CONN_TYPE_WORKER;
+ cur_worker_ct++; /* stats */
+}
+
+int
+count_cur_conns()
+{
+ return cur_conn_ct;
+}
+
+int
+count_cur_producers()
+{
+ return cur_producer_ct;
+}
+
+int
+count_cur_workers()
+{
+ return cur_worker_ct;
+}
+
 int
 conn_set_evq(conn c, const int events, evh handler)
 {
@@ -127,6 +166,10 @@ conn_close(conn c)
  if (c->out_job != c->reserved_job) free(c->out_job); /* peek command? */
  c->in_job = c->out_job = c->reserved_job = NULL;
 
+
+ if (c->type & CONN_TYPE_PRODUCER) cur_producer_ct--; /* stats */
+ if (c->type & CONN_TYPE_WORKER) cur_worker_ct--; /* stats */
+
  conn_remove(c);
  conn_free(c);
 }

conn.h

@@ -27,11 +27,16 @@
 #define OP_STATS 4
 #define OP_JOBSTATS 5
 
+/* CONN_TYPE_* are bit masks */
+#define CONN_TYPE_PRODUCER 1
+#define CONN_TYPE_WORKER 2
+
 typedef struct conn *conn;
 
 struct conn {
  int fd;
  char state;
+ char type;
  struct event evq;
 
  /* we cannot share this buffer with the reply line because we might read in
@@ -64,4 +69,11 @@ void conn_close(conn c);
 void conn_remove(conn c);
 void conn_insert(conn head, conn c);
 
+int count_cur_conns();
+int count_cur_producers();
+int count_cur_workers();
+
+void conn_set_producer(conn c);
+void conn_set_worker(conn c);
+
 #endif /*conn_h*/