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 <stdlib.h>
  20 #include <stdio.h>
  21 #include <time.h>
  22 #include <errno.h>
  23
  24 #include "conn.h"
  25 #include "net.h"
  26 #include "util.h"
  27 #include "prot.h"
  28
  29 #define SAFETY_MARGIN 1 /* seconds */
  30
  31 /* Doubly-linked list of free connections. */
  32 static struct conn pool = { &pool, &pool, 0 };
  33
  34 static int cur_conn_ct = 0, cur_worker_ct = 0, cur_producer_ct = 0;
  35 static unsigned int tot_conn_ct = 0;
  36
  37 static conn
  38 conn_alloc()
  39 {
  40     return conn_remove(pool.next) ? : malloc(sizeof(struct conn));
  41 }
  42
  43 static void
  44 conn_free(conn c)
  45 {
  46     c->fd = 0;
  47     conn_insert(&pool, c);
  48 }
  49
  50 static void
  51 on_watch(ms a, tube t, size_t i)
  52 {
  53     tube_iref(t);
  54     t->watching_ct++;
  55 }
  56
  57 static void
  58 on_ignore(ms a, tube t, size_t i)
  59 {
  60     t->watching_ct--;
  61     tube_dref(t);
  62 }
  63
  64 conn
  65 make_conn(int fd, char start_state, tube use, tube watch)
  66 {
  67     job j;
  68     conn c;
  69
  70     c = conn_alloc();
  71     if (!c) return twarn("OOM"), NULL;
  72
  73     ms_init(&c->watch, (ms_event_fn) on_watch, (ms_event_fn) on_ignore);
  74     if (!ms_append(&c->watch, watch)) {
  75         conn_free(c);
  76         return twarn("OOM"), NULL;
  77     }
  78
  79     c->use = NULL; /* initialize */
  80     TUBE_ASSIGN(c->use, use);
  81     use->using_ct++;
  82
  83     c->fd = fd;
  84     c->state = start_state;
  85     c->type = 0;
  86     c->cmd_read = 0;
  87     c->in_job = c->out_job = NULL;
  88     c->in_job_read = c->out_job_sent = 0;
  89     c->prev = c->next = c; /* must be out of a linked list right now */
  90     j = &c->reserved_jobs;
  91     j->prev = j->next = j;
  92
  93     /* stats */
  94     cur_conn_ct++;
  95     tot_conn_ct++;
  96
  97     return c;
  98 }
  99
 100 void
 101 conn_set_producer(conn c)
 102 {
 103     if (c->type & CONN_TYPE_PRODUCER) return;
 104     c->type |= CONN_TYPE_PRODUCER;
 105     cur_producer_ct++; /* stats */
 106 }
 107
 108 void
 109 conn_set_worker(conn c)
 110 {
 111     if (c->type & CONN_TYPE_WORKER) return;
 112     c->type |= CONN_TYPE_WORKER;
 113     cur_worker_ct++; /* stats */
 114 }
 115
 116 int
 117 count_cur_conns()
 118 {
 119     return cur_conn_ct;
 120 }
 121
 122 unsigned int
 123 count_tot_conns()
 124 {
 125     return tot_conn_ct;
 126 }
 127
 128 int
 129 count_cur_producers()
 130 {
 131     return cur_producer_ct;
 132 }
 133
 134 int
 135 count_cur_workers()
 136 {
 137     return cur_worker_ct;
 138 }
 139
 140 static int
 141 has_reserved_job(conn c)
 142 {
 143     return job_list_any_p(&c->reserved_jobs);
 144 }
 145
 146 int
 147 conn_set_evq(conn c, const int events, evh handler)
 148 {
 149     int r, margin = 0;
 150     struct timeval tv = {0, 0};
 151
 152     event_set(&c->evq, c->fd, events, handler, c);
 153
 154     if (conn_waiting(c)) margin = 1;
 155     if (has_reserved_job(c)) {
 156         time_t t = soonest_job(c)->deadline - time(NULL) - margin;
 157         tv.tv_sec = t > 0 ? t : 0;
 158     }
 159
 160     r = event_add(&c->evq, has_reserved_job(c) ? &tv : NULL);
 161     if (r == -1) return twarn("event_add() err %d", errno), -1;
 162
 163     return 0;
 164 }
 165
 166 int
 167 conn_update_evq(conn c, const int events)
 168 {
 169     int r;
 170
 171     if (!c) return twarnx("c is NULL"), -1;
 172
 173     /* If it's been added, try to delete it first */
 174     if (c->evq.ev_base) {
 175         r = event_del(&c->evq);
 176         if (r == -1) return twarn("event_del() err %d", errno), -1;
 177     }
 178
 179     return conn_set_evq(c, events, c->evq.ev_callback);
 180 }
 181
 182 static int
 183 conn_list_any_p(conn head)
 184 {
 185     return head->next != head || head->prev != head;
 186 }
 187
 188 conn
 189 conn_remove(conn c)
 190 {
 191     if (!conn_list_any_p(c)) return NULL; /* not in a doubly-linked list */
 192
 193     c->next->prev = c->prev;
 194     c->prev->next = c->next;
 195
 196     c->prev = c->next = c;
 197     return c;
 198 }
 199
 200 void
 201 conn_insert(conn head, conn c)
 202 {
 203     if (conn_list_any_p(c)) return; /* already in a linked list */
 204
 205     c->prev = head->prev;
 206     c->next = head;
 207     head->prev->next = c;
 208     head->prev = c;
 209 }
 210
 211 /* return the reserved job with the earliest deadline,
 212  * or NULL if there's no reserved job */
 213 job
 214 soonest_job(conn c)
 215 {
 216     job j, soonest = NULL;
 217
 218     for (j = c->reserved_jobs.next; j != &c->reserved_jobs; j = j->next) {
 219         if (j->deadline <= (soonest ? : j)->deadline) soonest = j;
 220     }
 221     return soonest;
 222 }
 223
 224 int
 225 has_reserved_this_job(conn c, job needle)
 226 {
 227     job j;
 228
 229     for (j = c->reserved_jobs.next; j != &c->reserved_jobs; j = j->next) {
 230         if (needle == j) return 1;
 231     }
 232     return 0;
 233 }
 234
 235 /* return true if c has a reserved job with less than one second until its
 236  * deadline */
 237 int
 238 conn_has_close_deadline(conn c)
 239 {
 240     time_t t = time(NULL);
 241     job j = soonest_job(c);
 242
 243     return j && t >= j->deadline - SAFETY_MARGIN;
 244 }
 245
 246 int
 247 conn_ready(conn c)
 248 {
 249     size_t i;
 250
 251     for (i = 0; i < c->watch.used; i++) {
 252         if (((tube) c->watch.items[i])->ready.used) return 1;
 253     }
 254     return 0;
 255 }
 256
 257 void
 258 conn_close(conn c)
 259 {
 260     event_del(&c->evq);
 261
 262     close(c->fd);
 263
 264     job_free(c->in_job);
 265
 266     /* was this a peek or stats command? */
 267     if (!has_reserved_this_job(c, c->out_job)) job_free(c->out_job);
 268
 269     c->in_job = c->out_job = NULL;
 270
 271     if (c->type & CONN_TYPE_PRODUCER) cur_producer_ct--; /* stats */
 272     if (c->type & CONN_TYPE_WORKER) cur_worker_ct--; /* stats */
 273
 274     cur_conn_ct--; /* stats */
 275
 276     unbrake(NULL);
 277     remove_waiting_conn(c);
 278     conn_remove(c);
 279     if (has_reserved_job(c)) enqueue_reserved_jobs(c);
 280
 281     ms_clear(&c->watch);
 282     c->use->using_ct--;
 283     TUBE_ASSIGN(c->use, NULL);
 284
 285     conn_free(c);
 286 }