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