Provide stats count of current watchers per tube.

conn.c

@@ -47,14 +47,16 @@ conn_free(conn c)
 }
 
 static void
-inc(ms a, tube t, size_t i)
+on_watch(ms a, tube t, size_t i)
 {
  tube_iref(t);
+ t->watching_ct++;
 }
 
 static void
-dec(ms a, tube t, size_t i)
+on_ignore(ms a, tube t, size_t i)
 {
+ t->watching_ct--;
  tube_dref(t);
 }
 
@@ -67,7 +69,7 @@ make_conn(int fd, char start_state, tube use, tube watch)
  c = conn_alloc();
  if (!c) return twarn("OOM"), NULL;
 
- ms_init(&c->watch, (ms_event_fn) inc, (ms_event_fn) dec);
+ ms_init(&c->watch, (ms_event_fn) on_watch, (ms_event_fn) on_ignore);
  if (!ms_append(&c->watch, watch)) {
  conn_free(c);
  return twarn("OOM"), NULL;

prot.c

@@ -149,6 +149,7 @@
  "current-jobs-reserved: %u\n" \
  "current-jobs-buried: %u\n" \
  "total-jobs: %llu\n" \
+ "current-watching: %u\n" \
  "current-waiting: %u\n" \
  "\r\n"
 
@@ -891,6 +892,7 @@ fmt_stats_tube(char *buf, size_t size, tube t)
  t->stat.reserved_ct,
  t->stat.buried_ct,
  t->stat.total_jobs_ct,
+ t->watching_ct,
  t->stat.waiting_ct);
 }
 

tube.c

@@ -43,6 +43,7 @@ make_tube(const char *name)
  ms_init(&t->waiting, NULL, NULL);
 
  t->stat = (struct stats) {0, 0, 0, 0};
+ t->watching_ct = 0;
 
  return t;
 }

tube.h

@@ -35,6 +35,7 @@ struct tube {
  struct job buried;
  struct ms waiting; /* set of conns */
  struct stats stat;
+ unsigned int watching_ct;
 };
 
 tube make_tube(const char *name);