job.c

   1 /* job.c - a job in the queue */
   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 "dat.h"
  23
  24 static uint64 next_id = 1;
  25
  26 static int cur_prime = 0;
  27
  28 static job all_jobs_init[12289] = {0};
  29 static job *all_jobs = all_jobs_init;
  30 static size_t all_jobs_cap = 12289; /* == primes[0] */
  31 static size_t all_jobs_used = 0;
  32
  33 static int hash_table_was_oom = 0;
  34
  35 static void rehash();
  36
  37 static int
  38 _get_job_hash_index(uint64 job_id)
  39 {
  40     return job_id % all_jobs_cap;
  41 }
  42
  43 static void
  44 store_job(job j)
  45 {
  46     int index = 0;
  47
  48     index = _get_job_hash_index(j->id);
  49
  50     j->ht_next = all_jobs[index];
  51     all_jobs[index] = j;
  52     all_jobs_used++;
  53
  54     /* accept a load factor of 4 */
  55     if (all_jobs_used > (all_jobs_cap << 2)) rehash();
  56 }
  57
  58 static void
  59 rehash()
  60 {
  61     job *old = all_jobs;
  62     size_t old_cap = all_jobs_cap, old_used = all_jobs_used, i;
  63
  64     if (cur_prime >= NUM_PRIMES) return;
  65     if (hash_table_was_oom) return;
  66
  67     all_jobs_cap = primes[++cur_prime];
  68     all_jobs = calloc(all_jobs_cap, sizeof(job));
  69     if (!all_jobs) {
  70         twarnx("Failed to allocate %d new hash buckets", all_jobs_cap);
  71         hash_table_was_oom = 1;
  72         --cur_prime;
  73         all_jobs = old;
  74         all_jobs_cap = old_cap;
  75         all_jobs_used = old_used;
  76         return;
  77     }
  78     all_jobs_used = 0;
  79
  80     for (i = 0; i < old_cap; i++) {
  81         while (old[i]) {
  82             job j = old[i];
  83             old[i] = j->ht_next;
  84             j->ht_next = NULL;
  85             store_job(j);
  86         }
  87     }
  88     if (old != all_jobs_init) {
  89         free(old);
  90     }
  91 }
  92
  93 job
  94 job_find(uint64 job_id)
  95 {
  96     job jh = NULL;
  97     int index = _get_job_hash_index(job_id);
  98
  99     for (jh = all_jobs[index]; jh && jh->id != job_id; jh = jh->ht_next);
 100
 101     return jh;
 102 }
 103
 104 job
 105 allocate_job(int body_size)
 106 {
 107     job j;
 108
 109     j = malloc(sizeof(struct job) + body_size);
 110     if (!j) return twarnx("OOM"), (job) 0;
 111
 112     j->id = 0;
 113     j->state = JOB_STATE_INVALID;
 114     j->created_at = nanoseconds();
 115     j->reserve_ct = j->timeout_ct = j->release_ct = j->bury_ct = j->kick_ct = 0;
 116     j->body_size = body_size;
 117     j->next = j->prev = j; /* not in a linked list */
 118     j->ht_next = NULL;
 119     j->tube = NULL;
 120     j->binlog = NULL;
 121     j->heap_index = 0;
 122     j->reserved_binlog_space = 0;
 123
 124     return j;
 125 }
 126
 127 job
 128 make_job_with_id(uint pri, int64 delay, int64 ttr,
 129                  int body_size, tube tube, uint64 id)
 130 {
 131     job j;
 132
 133     j = allocate_job(body_size);
 134     if (!j) return twarnx("OOM"), (job) 0;
 135
 136     if (id) {
 137         j->id = id;
 138         if (id >= next_id) next_id = id + 1;
 139     } else {
 140         j->id = next_id++;
 141     }
 142     j->pri = pri;
 143     j->delay = delay;
 144     j->ttr = ttr;
 145
 146     store_job(j);
 147
 148     TUBE_ASSIGN(j->tube, tube);
 149
 150     return j;
 151 }
 152
 153 static void
 154 job_hash_free(job j)
 155 {
 156     job *slot;
 157
 158     slot = &all_jobs[_get_job_hash_index(j->id)];
 159     while (*slot && *slot != j) slot = &(*slot)->ht_next;
 160     if (*slot) {
 161         *slot = (*slot)->ht_next;
 162         --all_jobs_used;
 163     }
 164 }
 165
 166 void
 167 job_free(job j)
 168 {
 169     if (j) {
 170         TUBE_ASSIGN(j->tube, NULL);
 171         if (j->state != JOB_STATE_COPY) job_hash_free(j);
 172     }
 173
 174     free(j);
 175 }
 176
 177 void
 178 job_setheappos(void *j, int pos)
 179 {
 180     ((job)j)->heap_index = pos;
 181 }
 182
 183 /* We can't substrct any of these values because there are too many bits */
 184 int
 185 job_pri_cmp(void *ax, void *bx)
 186 {
 187     job a = ax, b = bx;
 188     if (a->pri > b->pri) return 1;
 189     if (a->pri < b->pri) return -1;
 190     if (a->id > b->id) return 1;
 191     if (a->id < b->id) return -1;
 192     return 0;
 193 }
 194
 195 /* We can't substrct any of these values because there are too many bits */
 196 int
 197 job_delay_cmp(void *ax, void *bx)
 198 {
 199     job a = ax, b = bx;
 200     if (a->deadline_at > b->deadline_at) return 1;
 201     if (a->deadline_at < b->deadline_at) return -1;
 202     if (a->id > b->id) return 1;
 203     if (a->id < b->id) return -1;
 204     return 0;
 205 }
 206
 207 job
 208 job_copy(job j)
 209 {
 210     job n;
 211
 212     if (!j) return NULL;
 213
 214     n = malloc(sizeof(struct job) + j->body_size);
 215     if (!n) return twarnx("OOM"), (job) 0;
 216
 217     memcpy(n, j, sizeof(struct job) + j->body_size);
 218     n->next = n->prev = n; /* not in a linked list */
 219
 220     n->binlog = NULL; /* copies do not have refcnt on the binlog */
 221
 222     n->tube = 0; /* Don't use memcpy for the tube, which we must refcount. */
 223     TUBE_ASSIGN(n->tube, j->tube);
 224
 225     /* Mark this job as a copy so it can be appropriately freed later on */
 226     n->state = JOB_STATE_COPY;
 227
 228     return n;
 229 }
 230
 231 const char *
 232 job_state(job j)
 233 {
 234     if (j->state == JOB_STATE_READY) return "ready";
 235     if (j->state == JOB_STATE_RESERVED) return "reserved";
 236     if (j->state == JOB_STATE_BURIED) return "buried";
 237     if (j->state == JOB_STATE_DELAYED) return "delayed";
 238     return "invalid";
 239 }
 240
 241 int
 242 job_list_any_p(job head)
 243 {
 244     return head->next != head || head->prev != head;
 245 }
 246
 247 job
 248 job_remove(job j)
 249 {
 250     if (!j) return NULL;
 251     if (!job_list_any_p(j)) return NULL; /* not in a doubly-linked list */
 252
 253     j->next->prev = j->prev;
 254     j->prev->next = j->next;
 255
 256     j->prev = j->next = j;
 257
 258     return j;
 259 }
 260
 261 void
 262 job_insert(job head, job j)
 263 {
 264     if (job_list_any_p(j)) return; /* already in a linked list */
 265
 266     j->prev = head->prev;
 267     j->next = head;
 268     head->prev->next = j;
 269     head->prev = j;
 270 }
 271
 272 uint64
 273 total_jobs()
 274 {
 275     return next_id - 1;
 276 }
 277
 278 /* for unit tests */
 279 size_t
 280 get_all_jobs_used()
 281 {
 282     return all_jobs_used;
 283 }
 284
 285 void
 286 job_init()
 287 {
 288     all_jobs = calloc(all_jobs_cap, sizeof(job));
 289     if (!all_jobs) {
 290         twarnx("Failed to allocate %d hash buckets", all_jobs_cap);
 291     }
 292 }