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