conn.c

   1 /* conn.c - network connection state */
   2
   3 #include <stdlib.h>
   4 #include <stdio.h>
   5 #include <time.h>
   6 #include <errno.h>
   7
   8 #include "conn.h"
   9 #include "net.h"
  10 #include "util.h"
  11 #include "prot.h"
  12 #include "reserve.h"
  13
  14 /* Doubly-linked list of free connections. */
  15 static struct conn pool = { &pool, &pool, 0 };
  16
  17 static int cur_conn_ct = 0, cur_worker_ct = 0, cur_producer_ct = 0;
  18 static unsigned int tot_conn_ct = 0;
  19
  20 static conn
  21 conn_alloc()
  22 {
  23     return conn_remove(pool.next) ? : malloc(sizeof(struct conn));
  24 }
  25
  26 static void
  27 conn_free(conn c)
  28 {
  29     c->fd = 0;
  30     conn_insert(&pool, c);
  31 }
  32
  33 conn
  34 make_conn(int fd, char start_state)
  35 {
  36     job j;
  37     conn c;
  38
  39     c = conn_alloc();
  40     if (!c) return twarn("OOM"), NULL;
  41
  42     c->fd = fd;
  43     c->state = start_state;
  44     c->type = 0;
  45     c->cmd_read = 0;
  46     c->in_job = c->out_job = NULL;
  47     c->in_job_read = c->out_job_sent = 0;
  48     c->prev = c->next = c; /* must be out of a linked list right now */
  49     j = &c->reserved_jobs;
  50     j->prev = j->next = j;
  51
  52     /* stats */
  53     cur_conn_ct++;
  54     tot_conn_ct++;
  55
  56     return c;
  57 }
  58
  59 void
  60 conn_set_producer(conn c)
  61 {
  62     if (c->type & CONN_TYPE_PRODUCER) return;
  63     c->type |= CONN_TYPE_PRODUCER;
  64     cur_producer_ct++; /* stats */
  65 }
  66
  67 void
  68 conn_set_worker(conn c)
  69 {
  70     if (c->type & CONN_TYPE_WORKER) return;
  71     c->type |= CONN_TYPE_WORKER;
  72     cur_worker_ct++; /* stats */
  73 }
  74
  75 int
  76 count_cur_conns()
  77 {
  78     return cur_conn_ct;
  79 }
  80
  81 unsigned int
  82 count_tot_conns()
  83 {
  84     return tot_conn_ct;
  85 }
  86
  87 int
  88 count_cur_producers()
  89 {
  90     return cur_producer_ct;
  91 }
  92
  93 int
  94 count_cur_workers()
  95 {
  96     return cur_worker_ct;
  97 }
  98
  99 int
 100 conn_set_evq(conn c, const int events, evh handler)
 101 {
 102     int r;
 103     struct timeval tv = {0, 0};
 104
 105     event_set(&c->evq, c->fd, events, handler, c);
 106
 107     if (has_reserved_job(c)) tv.tv_sec = soonest_job(c)->deadline - time(NULL);
 108
 109     r = event_add(&c->evq, has_reserved_job(c) ? &tv : NULL);
 110     if (r == -1) return twarn("event_add() err %d", errno), -1;
 111
 112     return 0;
 113 }
 114
 115 int
 116 conn_update_evq(conn c, const int events)
 117 {
 118     int r;
 119
 120     if (!c) return -1;
 121
 122     /* If it's been added, try to delete it first */
 123     if (c->evq.ev_base) {
 124         r = event_del(&c->evq);
 125         if (r == -1) return -1;
 126     }
 127
 128     return conn_set_evq(c, events, c->evq.ev_callback);
 129 }
 130
 131 int
 132 conn_list_any_p(conn head)
 133 {
 134     return head->next != head || head->prev != head;
 135 }
 136
 137 conn
 138 conn_remove(conn c)
 139 {
 140     if (!conn_list_any_p(c)) return NULL; /* not in a doubly-linked list */
 141
 142     c->next->prev = c->prev;
 143     c->prev->next = c->next;
 144
 145     c->prev = c->next = c;
 146     return c;
 147 }
 148
 149 void
 150 conn_insert(conn head, conn c)
 151 {
 152     if (conn_list_any_p(c)) return; /* already in a linked list */
 153
 154     c->prev = head->prev;
 155     c->next = head;
 156     head->prev->next = c;
 157     head->prev = c;
 158 }
 159
 160 void
 161 conn_close(conn c)
 162 {
 163     event_del(&c->evq);
 164
 165     close(c->fd);
 166
 167     free(c->in_job);
 168
 169     /* was this a peek or stats command? */
 170     if (!has_reserved_this_job(c, c->out_job)) free(c->out_job);
 171
 172     c->in_job = c->out_job = NULL;
 173
 174     if (c->type & CONN_TYPE_PRODUCER) cur_producer_ct--; /* stats */
 175     if (c->type & CONN_TYPE_WORKER) cur_worker_ct--; /* stats */
 176
 177     cur_conn_ct--; /* stats */
 178
 179     unbrake(NULL);
 180     remove_waiting_conn(c);
 181     conn_remove(c);
 182     if (has_reserved_job(c)) enqueue_reserved_jobs(c);
 183
 184     conn_free(c);
 185 }