conn.c

   1 #include <stdint.h>
   2 #include <stdlib.h>
   3 #include <string.h>
   4 #include <errno.h>
   5 #include <limits.h>
   6 #include <unistd.h>
   7 #include "dat.h"
   8
   9 #define SAFETY_MARGIN (1000000000) /* 1 second */
  10
  11 static int cur_conn_ct = 0, cur_worker_ct = 0, cur_producer_ct = 0;
  12 static uint tot_conn_ct = 0;
  13
  14 static void
  15 on_watch(ms a, tube t, size_t i)
  16 {
  17     tube_iref(t);
  18     t->watching_ct++;
  19 }
  20
  21 static void
  22 on_ignore(ms a, tube t, size_t i)
  23 {
  24     t->watching_ct--;
  25     tube_dref(t);
  26 }
  27
  28 conn
  29 make_conn(int fd, char start_state, tube use, tube watch)
  30 {
  31     job j;
  32     conn c;
  33
  34     c = new(struct conn);
  35     if (!c) return twarn("OOM"), (conn) 0;
  36
  37     ms_init(&c->watch, (ms_event_fn) on_watch, (ms_event_fn) on_ignore);
  38     if (!ms_append(&c->watch, watch)) {
  39         free(c);
  40         return twarn("OOM"), (conn) 0;
  41     }
  42
  43     c->use = NULL; /* initialize */
  44     TUBE_ASSIGN(c->use, use);
  45     use->using_ct++;
  46
  47     c->sock.fd = fd;
  48     c->state = start_state;
  49     c->type = 0;
  50     c->cmd_read = 0;
  51     c->pending_timeout = -1;
  52     c->soonest_job = NULL;
  53     c->in_job = c->out_job = NULL;
  54     c->in_job_read = c->out_job_sent = 0;
  55     c->prev = c->next = c; /* must be out of a linked list right now */
  56     j = &c->reserved_jobs;
  57     j->prev = j->next = j;
  58
  59     /* stats */
  60     cur_conn_ct++;
  61     tot_conn_ct++;
  62
  63     return c;
  64 }
  65
  66 void
  67 conn_set_producer(conn c)
  68 {
  69     if (c->type & CONN_TYPE_PRODUCER) return;
  70     c->type |= CONN_TYPE_PRODUCER;
  71     cur_producer_ct++; /* stats */
  72 }
  73
  74 void
  75 conn_set_worker(conn c)
  76 {
  77     if (c->type & CONN_TYPE_WORKER) return;
  78     c->type |= CONN_TYPE_WORKER;
  79     cur_worker_ct++; /* stats */
  80 }
  81
  82 int
  83 count_cur_conns()
  84 {
  85     return cur_conn_ct;
  86 }
  87
  88 uint
  89 count_tot_conns()
  90 {
  91     return tot_conn_ct;
  92 }
  93
  94 int
  95 count_cur_producers()
  96 {
  97     return cur_producer_ct;
  98 }
  99
 100 int
 101 count_cur_workers()
 102 {
 103     return cur_worker_ct;
 104 }
 105
 106 static int
 107 has_reserved_job(conn c)
 108 {
 109     return job_list_any_p(&c->reserved_jobs);
 110 }
 111
 112
 113 static int64
 114 conntickat(conn c)
 115 {
 116     int margin = 0, should_timeout = 0;
 117     int64 t = INT64_MAX;
 118
 119     if (conn_waiting(c)) {
 120         margin = SAFETY_MARGIN;
 121     }
 122
 123     if (has_reserved_job(c)) {
 124         t = soonest_job(c)->r.deadline_at - nanoseconds() - margin;
 125         should_timeout = 1;
 126     }
 127     if (c->pending_timeout >= 0) {
 128         t = min(t, ((int64)c->pending_timeout) * 1000000000);
 129         should_timeout = 1;
 130     }
 131
 132     if (should_timeout) {
 133         return nanoseconds() + t;
 134     }
 135     return 0;
 136 }
 137
 138
 139 void
 140 connwant(conn c, int rw, conn list)
 141 {
 142     c->rw = rw;
 143     conn_insert(list, c);
 144     connsched(c);
 145 }
 146
 147
 148 void
 149 connsched(conn c)
 150 {
 151     c->tickat = conntickat(c);
 152     srvschedconn(c->srv, c);
 153 }
 154
 155
 156 static int
 157 conn_list_any_p(conn head)
 158 {
 159     return head->next != head || head->prev != head;
 160 }
 161
 162 conn
 163 conn_remove(conn c)
 164 {
 165     if (!conn_list_any_p(c)) return NULL; /* not in a doubly-linked list */
 166
 167     c->next->prev = c->prev;
 168     c->prev->next = c->next;
 169
 170     c->prev = c->next = c;
 171     return c;
 172 }
 173
 174 void
 175 conn_insert(conn head, conn c)
 176 {
 177     if (conn_list_any_p(c)) return; /* already in a linked list */
 178
 179     c->prev = head->prev;
 180     c->next = head;
 181     head->prev->next = c;
 182     head->prev = c;
 183 }
 184
 185 /* return the reserved job with the earliest deadline,
 186  * or NULL if there's no reserved job */
 187 job
 188 soonest_job(conn c)
 189 {
 190     job j = NULL;
 191     job soonest = c->soonest_job;
 192
 193     if (soonest == NULL) {
 194         for (j = c->reserved_jobs.next; j != &c->reserved_jobs; j = j->next) {
 195             if (j->r.deadline_at <= (soonest ? : j)->r.deadline_at) soonest = j;
 196         }
 197     }
 198     c->soonest_job = soonest;
 199     return soonest;
 200 }
 201
 202 int
 203 has_reserved_this_job(conn c, job j)
 204 {
 205     return j && j->r.state == Reserved && j->reserver == c;
 206 }
 207
 208 /* return true if c has a reserved job with less than one second until its
 209  * deadline */
 210 int
 211 conn_has_close_deadline(conn c)
 212 {
 213     int64 t = nanoseconds();
 214     job j = soonest_job(c);
 215
 216     return j && t >= j->r.deadline_at - SAFETY_MARGIN;
 217 }
 218
 219 int
 220 conn_ready(conn c)
 221 {
 222     size_t i;
 223
 224     for (i = 0; i < c->watch.used; i++) {
 225         if (((tube) c->watch.items[i])->ready.len) return 1;
 226     }
 227     return 0;
 228 }
 229
 230
 231 int
 232 connless(conn a, conn b)
 233 {
 234     return a->tickat < b->tickat;
 235 }
 236
 237
 238 void
 239 connrec(conn c, int i)
 240 {
 241     c->tickpos = i;
 242 }
 243
 244
 245 void
 246 conn_close(conn c)
 247 {
 248     sockwant(&c->sock, 0);
 249     close(c->sock.fd);
 250
 251     job_free(c->in_job);
 252
 253     /* was this a peek or stats command? */
 254     if (c->out_job && !c->out_job->r.id) job_free(c->out_job);
 255
 256     c->in_job = c->out_job = NULL;
 257     c->in_job_read = 0;
 258
 259     if (c->type & CONN_TYPE_PRODUCER) cur_producer_ct--; /* stats */
 260     if (c->type & CONN_TYPE_WORKER) cur_worker_ct--; /* stats */
 261
 262     cur_conn_ct--; /* stats */
 263
 264     remove_waiting_conn(c);
 265     conn_remove(c);
 266     if (has_reserved_job(c)) enqueue_reserved_jobs(c);
 267
 268     ms_clear(&c->watch);
 269     c->use->using_ct--;
 270     TUBE_ASSIGN(c->use, NULL);
 271
 272     free(c);
 273 }