conn.c

   1 /* conn.c - network connection state */
   2
   3 /* Copyright (C) 2007 Keith Rarick and Philotic Inc.
   4
   5  * This program is free software: you can redistribute it and/or modify
   6  * it under the terms of the GNU General Public License as published by
   7  * the Free Software Foundation, either version 3 of the License, or
   8  * (at your option) any later version.
   9
  10  * This program is distributed in the hope that it will be useful,
  11  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13  * GNU General Public License for more details.
  14
  15  * You should have received a copy of the GNU General Public License
  16  * along with this program.  If not, see <http://www.gnu.org/licenses/>.
  17  */
  18
  19 #include <stdint.h>
  20 #include <stdlib.h>
  21 #include <string.h>
  22 #include <errno.h>
  23 #include <limits.h>
  24 #include <unistd.h>
  25 #include "dat.h"
  26
  27 #define SAFETY_MARGIN (1000000000) /* 1 second */
  28
  29 /* Doubly-linked list of free connections. */
  30 static struct conn pool = { &pool, &pool };
  31
  32 static int cur_conn_ct = 0, cur_worker_ct = 0, cur_producer_ct = 0;
  33 static uint tot_conn_ct = 0;
  34
  35 static conn
  36 conn_alloc()
  37 {
  38     conn c;
  39
  40     c = conn_remove(pool.next);
  41     if (!c) {
  42         c = malloc(sizeof(struct conn));
  43     }
  44
  45     return memset(c, 0, sizeof *c);
  46 }
  47
  48 static void
  49 conn_free(conn c)
  50 {
  51     c->sock.fd = 0;
  52     conn_insert(&pool, c);
  53 }
  54
  55 static void
  56 on_watch(ms a, tube t, size_t i)
  57 {
  58     tube_iref(t);
  59     t->watching_ct++;
  60 }
  61
  62 static void
  63 on_ignore(ms a, tube t, size_t i)
  64 {
  65     t->watching_ct--;
  66     tube_dref(t);
  67 }
  68
  69 conn
  70 make_conn(int fd, char start_state, tube use, tube watch)
  71 {
  72     job j;
  73     conn c;
  74
  75     c = conn_alloc();
  76     if (!c) return twarn("OOM"), (conn) 0;
  77
  78     ms_init(&c->watch, (ms_event_fn) on_watch, (ms_event_fn) on_ignore);
  79     if (!ms_append(&c->watch, watch)) {
  80         conn_free(c);
  81         return twarn("OOM"), (conn) 0;
  82     }
  83
  84     c->use = NULL; /* initialize */
  85     TUBE_ASSIGN(c->use, use);
  86     use->using_ct++;
  87
  88     c->sock.fd = fd;
  89     c->state = start_state;
  90     c->type = 0;
  91     c->cmd_read = 0;
  92     c->pending_timeout = -1;
  93     c->soonest_job = NULL;
  94     c->in_job = c->out_job = NULL;
  95     c->in_job_read = c->out_job_sent = 0;
  96     c->prev = c->next = c; /* must be out of a linked list right now */
  97     j = &c->reserved_jobs;
  98     j->prev = j->next = j;
  99
 100     /* stats */
 101     cur_conn_ct++;
 102     tot_conn_ct++;
 103
 104     return c;
 105 }
 106
 107 void
 108 conn_set_producer(conn c)
 109 {
 110     if (c->type & CONN_TYPE_PRODUCER) return;
 111     c->type |= CONN_TYPE_PRODUCER;
 112     cur_producer_ct++; /* stats */
 113 }
 114
 115 void
 116 conn_set_worker(conn c)
 117 {
 118     if (c->type & CONN_TYPE_WORKER) return;
 119     c->type |= CONN_TYPE_WORKER;
 120     cur_worker_ct++; /* stats */
 121 }
 122
 123 int
 124 count_cur_conns()
 125 {
 126     return cur_conn_ct;
 127 }
 128
 129 uint
 130 count_tot_conns()
 131 {
 132     return tot_conn_ct;
 133 }
 134
 135 int
 136 count_cur_producers()
 137 {
 138     return cur_producer_ct;
 139 }
 140
 141 int
 142 count_cur_workers()
 143 {
 144     return cur_worker_ct;
 145 }
 146
 147 static int
 148 has_reserved_job(conn c)
 149 {
 150     return job_list_any_p(&c->reserved_jobs);
 151 }
 152
 153
 154 static int64
 155 conntickat(conn c)
 156 {
 157     int margin = 0, should_timeout = 0;
 158     int64 t = INT64_MAX;
 159
 160     if (conn_waiting(c)) {
 161         margin = SAFETY_MARGIN;
 162     }
 163
 164     if (has_reserved_job(c)) {
 165         t = soonest_job(c)->deadline_at - nanoseconds() - margin;
 166         should_timeout = 1;
 167     }
 168     if (c->pending_timeout >= 0) {
 169         t = min(t, ((int64)c->pending_timeout) * 1000000000);
 170         should_timeout = 1;
 171     }
 172
 173     if (should_timeout) {
 174         return nanoseconds() + t;
 175     }
 176     return 0;
 177 }
 178
 179
 180 void
 181 connwant(conn c, int rw, conn list)
 182 {
 183     c->rw = rw;
 184     conn_insert(list, c);
 185     connsched(c);
 186 }
 187
 188
 189 void
 190 connsched(conn c)
 191 {
 192     c->tickat = conntickat(c);
 193     srvschedconn(c->srv, c);
 194 }
 195
 196
 197 static int
 198 conn_list_any_p(conn head)
 199 {
 200     return head->next != head || head->prev != head;
 201 }
 202
 203 conn
 204 conn_remove(conn c)
 205 {
 206     if (!conn_list_any_p(c)) return NULL; /* not in a doubly-linked list */
 207
 208     c->next->prev = c->prev;
 209     c->prev->next = c->next;
 210
 211     c->prev = c->next = c;
 212     return c;
 213 }
 214
 215 void
 216 conn_insert(conn head, conn c)
 217 {
 218     if (conn_list_any_p(c)) return; /* already in a linked list */
 219
 220     c->prev = head->prev;
 221     c->next = head;
 222     head->prev->next = c;
 223     head->prev = c;
 224 }
 225
 226 /* return the reserved job with the earliest deadline,
 227  * or NULL if there's no reserved job */
 228 job
 229 soonest_job(conn c)
 230 {
 231     job j = NULL;
 232     job soonest = c->soonest_job;
 233
 234     if (soonest == NULL) {
 235         for (j = c->reserved_jobs.next; j != &c->reserved_jobs; j = j->next) {
 236             if (j->deadline_at <= (soonest ? : j)->deadline_at) soonest = j;
 237         }
 238     }
 239     c->soonest_job = soonest;
 240     return soonest;
 241 }
 242
 243 int
 244 has_reserved_this_job(conn c, job j)
 245 {
 246     return j && j->state == JOB_STATE_RESERVED && j->reserver == c;
 247 }
 248
 249 /* return true if c has a reserved job with less than one second until its
 250  * deadline */
 251 int
 252 conn_has_close_deadline(conn c)
 253 {
 254     int64 t = nanoseconds();
 255     job j = soonest_job(c);
 256
 257     return j && t >= j->deadline_at - SAFETY_MARGIN;
 258 }
 259
 260 int
 261 conn_ready(conn c)
 262 {
 263     size_t i;
 264
 265     for (i = 0; i < c->watch.used; i++) {
 266         if (((tube) c->watch.items[i])->ready.len) return 1;
 267     }
 268     return 0;
 269 }
 270
 271
 272 int
 273 conncmp(conn a, conn b)
 274 {
 275     if (a->tickat > b->tickat) return 1;
 276     if (a->tickat < b->tickat) return -1;
 277     return 0;
 278 }
 279
 280
 281 void
 282 connrec(conn c, int i)
 283 {
 284     c->tickpos = i;
 285 }
 286
 287
 288 void
 289 conn_close(conn c)
 290 {
 291     sockwant(&c->sock, 0);
 292     close(c->sock.fd);
 293
 294     job_free(c->in_job);
 295
 296     /* was this a peek or stats command? */
 297     if (c->out_job && !c->out_job->id) job_free(c->out_job);
 298
 299     c->in_job = c->out_job = NULL;
 300     c->in_job_read = 0;
 301
 302     if (c->type & CONN_TYPE_PRODUCER) cur_producer_ct--; /* stats */
 303     if (c->type & CONN_TYPE_WORKER) cur_worker_ct--; /* stats */
 304
 305     cur_conn_ct--; /* stats */
 306
 307     remove_waiting_conn(c);
 308     conn_remove(c);
 309     if (has_reserved_job(c)) enqueue_reserved_jobs(c);
 310
 311     ms_clear(&c->watch);
 312     c->use->using_ct--;
 313     TUBE_ASSIGN(c->use, NULL);
 314
 315     conn_free(c);
 316 }