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