job.c

   1 #include <stdint.h>
   2 #include <stdlib.h>
   3 #include <string.h>
   4 #include "dat.h"
   5
   6 static uint64 next_id = 1;
   7
   8 static int cur_prime = 0;
   9
  10 static job all_jobs_init[12289] = {0};
  11 static job *all_jobs = all_jobs_init;
  12 static size_t all_jobs_cap = 12289; /* == primes[0] */
  13 static size_t all_jobs_used = 0;
  14
  15 static int hash_table_was_oom = 0;
  16
  17 static void rehash();
  18
  19 static int
  20 _get_job_hash_index(uint64 job_id)
  21 {
  22     return job_id % all_jobs_cap;
  23 }
  24
  25 static void
  26 store_job(job j)
  27 {
  28     int index = 0;
  29
  30     index = _get_job_hash_index(j->r.id);
  31
  32     j->ht_next = all_jobs[index];
  33     all_jobs[index] = j;
  34     all_jobs_used++;
  35
  36     /* accept a load factor of 4 */
  37     if (all_jobs_used > (all_jobs_cap << 2)) rehash(1);
  38 }
  39
  40 static void
  41 rehash(int is_upscaling)
  42 {
  43     job *old = all_jobs;
  44     size_t old_cap = all_jobs_cap, old_used = all_jobs_used, i;
  45     int old_prime = cur_prime;
  46     int d = is_upscaling ? 1 : -1;
  47
  48     if (cur_prime + d >= NUM_PRIMES) return;
  49     if (cur_prime + d < 0) return;
  50     if (is_upscaling && hash_table_was_oom) return;
  51
  52     cur_prime += d;
  53
  54     all_jobs_cap = primes[cur_prime];
  55     all_jobs = calloc(all_jobs_cap, sizeof(job));
  56     if (!all_jobs) {
  57         twarnx("Failed to allocate %zu new hash buckets", all_jobs_cap);
  58         hash_table_was_oom = 1;
  59         cur_prime = old_prime;
  60         all_jobs = old;
  61         all_jobs_cap = old_cap;
  62         all_jobs_used = old_used;
  63         return;
  64     }
  65     all_jobs_used = 0;
  66     hash_table_was_oom = 0;
  67
  68     for (i = 0; i < old_cap; i++) {
  69         while (old[i]) {
  70             job j = old[i];
  71             old[i] = j->ht_next;
  72             j->ht_next = NULL;
  73             store_job(j);
  74         }
  75     }
  76     if (old != all_jobs_init) {
  77         free(old);
  78     }
  79 }
  80
  81 job
  82 job_find(uint64 job_id)
  83 {
  84     job jh = NULL;
  85     int index = _get_job_hash_index(job_id);
  86
  87     for (jh = all_jobs[index]; jh && jh->r.id != job_id; jh = jh->ht_next);
  88
  89     return jh;
  90 }
  91
  92 job
  93 allocate_job(int body_size)
  94 {
  95     job j;
  96
  97     j = malloc(sizeof(struct job) + body_size);
  98     if (!j) return twarnx("OOM"), (job) 0;
  99
 100     memset(j, 0, sizeof(struct job));
 101     j->r.created_at = nanoseconds();
 102     j->r.body_size = body_size;
 103     j->next = j->prev = j; /* not in a linked list */
 104     return j;
 105 }
 106
 107 job
 108 make_job_with_id(uint pri, int64 delay, int64 ttr,
 109                  int body_size, tube tube, uint64 id)
 110 {
 111     job j;
 112
 113     j = allocate_job(body_size);
 114     if (!j) return twarnx("OOM"), (job) 0;
 115
 116     if (id) {
 117         j->r.id = id;
 118         if (id >= next_id) next_id = id + 1;
 119     } else {
 120         j->r.id = next_id++;
 121     }
 122     j->r.pri = pri;
 123     j->r.delay = delay;
 124     j->r.ttr = ttr;
 125
 126     store_job(j);
 127
 128     TUBE_ASSIGN(j->tube, tube);
 129
 130     return j;
 131 }
 132
 133 static void
 134 job_hash_free(job j)
 135 {
 136     job *slot;
 137
 138     slot = &all_jobs[_get_job_hash_index(j->r.id)];
 139     while (*slot && *slot != j) slot = &(*slot)->ht_next;
 140     if (*slot) {
 141         *slot = (*slot)->ht_next;
 142         --all_jobs_used;
 143     }
 144
 145     // Downscale when the hashmap is too sparse
 146     if (all_jobs_used < (all_jobs_cap >> 4)) rehash(0);
 147 }
 148
 149 void
 150 job_free(job j)
 151 {
 152     if (j) {
 153         TUBE_ASSIGN(j->tube, NULL);
 154         if (j->r.state != Copy) job_hash_free(j);
 155     }
 156
 157     free(j);
 158 }
 159
 160 void
 161 job_setheappos(void *j, int pos)
 162 {
 163     ((job)j)->heap_index = pos;
 164 }
 165
 166 int
 167 job_pri_less(void *ax, void *bx)
 168 {
 169     job a = ax, b = bx;
 170     if (a->r.pri < b->r.pri) return 1;
 171     if (a->r.pri > b->r.pri) return 0;
 172     return a->r.id < b->r.id;
 173 }
 174
 175 int
 176 job_delay_less(void *ax, void *bx)
 177 {
 178     job a = ax, b = bx;
 179     if (a->r.deadline_at < b->r.deadline_at) return 1;
 180     if (a->r.deadline_at > b->r.deadline_at) return 0;
 181     return a->r.id < b->r.id;
 182 }
 183
 184 job
 185 job_copy(job j)
 186 {
 187     job n;
 188
 189     if (!j) return NULL;
 190
 191     n = malloc(sizeof(struct job) + j->r.body_size);
 192     if (!n) return twarnx("OOM"), (job) 0;
 193
 194     memcpy(n, j, sizeof(struct job) + j->r.body_size);
 195     n->next = n->prev = n; /* not in a linked list */
 196
 197     n->file = NULL; /* copies do not have refcnt on the wal */
 198
 199     n->tube = 0; /* Don't use memcpy for the tube, which we must refcount. */
 200     TUBE_ASSIGN(n->tube, j->tube);
 201
 202     /* Mark this job as a copy so it can be appropriately freed later on */
 203     n->r.state = Copy;
 204
 205     return n;
 206 }
 207
 208 const char *
 209 job_state(job j)
 210 {
 211     if (j->r.state == Ready) return "ready";
 212     if (j->r.state == Reserved) return "reserved";
 213     if (j->r.state == Buried) return "buried";
 214     if (j->r.state == Delayed) return "delayed";
 215     return "invalid";
 216 }
 217
 218 int
 219 job_list_any_p(job head)
 220 {
 221     return head->next != head || head->prev != head;
 222 }
 223
 224 job
 225 job_remove(job j)
 226 {
 227     if (!j) return NULL;
 228     if (!job_list_any_p(j)) return NULL; /* not in a doubly-linked list */
 229
 230     j->next->prev = j->prev;
 231     j->prev->next = j->next;
 232
 233     j->prev = j->next = j;
 234
 235     return j;
 236 }
 237
 238 void
 239 job_insert(job head, job j)
 240 {
 241     if (job_list_any_p(j)) return; /* already in a linked list */
 242
 243     j->prev = head->prev;
 244     j->next = head;
 245     head->prev->next = j;
 246     head->prev = j;
 247 }
 248
 249 uint64
 250 total_jobs()
 251 {
 252     return next_id - 1;
 253 }
 254
 255 /* for unit tests */
 256 size_t
 257 get_all_jobs_used()
 258 {
 259     return all_jobs_used;
 260 }