conn.c

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