1. tube定义
在beanstalk中，tube只是一个消息队列的名字，本身只是作为job的容器，因此不具有复杂的操作。
由于job具有优先级，这里采用最小堆保存job
job具有ready和delay两种状态，因此每个tube采用两个最小堆对job进行管理。
struct tube { uint refs; char name[MAX_TUBE_NAME_LEN]; Heap ready; Heap delay; struct ms waiting; /* set of conns */ struct stats stat; uint using_ct; uint watching_ct; int64 pause; int64 deadline_at; struct job buried; };
2.tube管理
tube的操作定义在tube.c中，
采用一个全局变量tubes管理所有的tube，而每个tube采用引用计数的方式进行管理。
tubes定义为一个集合，采用数组的方式实现。
tube的查找采用的遍历操作的方式，因此性能并不高。在实际使用时每个实例不要包含过多的tube。
不过，如果use和watch操作比较少时，性能影响并不大，因为tubes的遍历只有在use、watch、state、pause的时候才会发生。其他时候使用连接自己管理的tube。
struct ms tubes; tube make_tube(const char *name) { tube t; t = new(struct tube); if (!t) return NULL; t->name[MAX_TUBE_NAME_LEN - 1] = '\0'; strncpy(t->name, name, MAX_TUBE_NAME_LEN - 1); if (t->name[MAX_TUBE_NAME_LEN - 1] != '\0') twarnx("truncating tube name"); t->ready.less = job_pri_less; t->delay.less = job_delay_less; t->ready.rec = job_setheappos; t->delay.rec = job_setheappos; t->buried = (struct job) { }; t->buried.prev = t->buried.next = &t->buried; ms_init(&t->waiting, NULL, NULL); return t; } static void tube_free(tube t) { prot_remove_tube(t); free(t->ready.data); free(t->delay.data); ms_clear(&t->waiting); free(t); } void tube_dref(tube t) { if (!t) return; if (t->refs < 1) return twarnx("refs is zero for tube: %s", t->name); --t->refs; if (t->refs < 1) tube_free(t); } void tube_iref(tube t) { if (!t) return; ++t->refs; } static tube make_and_insert_tube(const char *name) { int r; tube t = NULL; t = make_tube(name); if (!t) return NULL; /* We want this global tube list to behave like "weak" refs, so don't * increment the ref count. */ r = ms_append(&tubes, t); if (!r) return tube_dref(t), (tube) 0; return t; } tube tube_find(const char *name) { tube t; size_t i; for (i = 0; i < tubes.used; i++) { t = tubes.items[i]; if (strncmp(t->name, name, MAX_TUBE_NAME_LEN) == 0) return t; } return NULL; } tube tube_find_or_make(const char *name) { return tube_find(name) ? : make_and_insert_tube(name); }
3. job定义
job属于消息队列中的一个消息，因此必须同时注重内存管理和binlog生成。
// if you modify this struct, you must increment Walver above struct Jobrec { uint64 id; uint32 pri; int64 delay; int64 ttr; int32 body_size; int64 created_at; int64 deadline_at; uint32 reserve_ct; uint32 timeout_ct; uint32 release_ct; uint32 bury_ct; uint32 kick_ct; byte state; }; struct job { Jobrec r; // persistent fields; these get written to the wal /* bookeeping fields; these are in-memory only */ char pad[6]; tube tube; job prev, next; /* linked list of jobs */ job ht_next; /* Next job in a hash table list */ size_t heap_index; /* where is this job in its current heap */ File *file; job fnext; job fprev; void *reserver; int walresv; int walused; char body[]; // written separately to the wal };
4. job管理
为了保存job，整个实例采用一个全局数组保存所有的job。
因此这里存在一个问题就是如何找到一个job？通过hash的方式将job映射到数组元素中。
job的管理主要是增删查复制操作。
static int _get_job_hash_index(uint64 job_id) { return job_id % all_jobs_cap; } static void store_job(job j) { int index = 0; index = _get_job_hash_index(j->r.id); j->ht_next = all_jobs[index]; all_jobs[index] = j; all_jobs_used++; /* accept a load factor of 4 */ if (all_jobs_used > (all_jobs_cap << 2)) rehash(); } static void rehash() { job *old = all_jobs; size_t old_cap = all_jobs_cap, old_used = all_jobs_used, i; if (cur_prime >= NUM_PRIMES) return; if (hash_table_was_oom) return; all_jobs_cap = primes[++cur_prime]; all_jobs = calloc(all_jobs_cap, sizeof(job)); if (!all_jobs) { twarnx("Failed to allocate %zu new hash buckets", all_jobs_cap); hash_table_was_oom = 1; --cur_prime; all_jobs = old; all_jobs_cap = old_cap; all_jobs_used = old_used; return; } all_jobs_used = 0; for (i = 0; i < old_cap; i++) { while (old[i]) { job j = old[i]; old[i] = j->ht_next; j->ht_next = NULL; store_job(j); } } if (old != all_jobs_init) { free(old); } } job job_find(uint64 job_id) { job jh = NULL; int index = _get_job_hash_index(job_id); for (jh = all_jobs[index]; jh && jh->r.id != job_id; jh = jh->ht_next); return jh; } job allocate_job(int body_size) { job j; j = malloc(sizeof(struct job) + body_size); if (!j) return twarnx("OOM"), (job) 0; memset(j, 0, sizeof(struct job)); j->r.created_at = nanoseconds(); j->r.body_size = body_size; j->next = j->prev = j; /* not in a linked list */ return j; } job make_job_with_id(uint pri, int64 delay, int64 ttr, int body_size, tube tube, uint64 id) { job j; j = allocate_job(body_size); if (!j) return twarnx("OOM"), (job) 0; if (id) { j->r.id = id; if (id >= next_id) next_id = id + 1; } else { j->r.id = next_id++; } j->r.pri = pri; j->r.delay = delay; j->r.ttr = ttr; store_job(j); TUBE_ASSIGN(j->tube, tube); return j; } static void job_hash_free(job j) { job *slot; slot = &all_jobs[_get_job_hash_index(j->r.id)]; while (*slot && *slot != j) slot = &(*slot)->ht_next; if (*slot) { *slot = (*slot)->ht_next; --all_jobs_used; } } void job_free(job j) { if (j) { TUBE_ASSIGN(j->tube, NULL); if (j->r.state != Copy) job_hash_free(j); } free(j); } void job_setheappos(void *j, int pos) { ((job)j)->heap_index = pos; } int job_pri_less(void *ax, void *bx) { job a = ax, b = bx; if (a->r.pri < b->r.pri) return 1; if (a->r.pri > b->r.pri) return 0; return a->r.id < b->r.id; } int job_delay_less(void *ax, void *bx) { job a = ax, b = bx; if (a->r.deadline_at < b->r.deadline_at) return 1; if (a->r.deadline_at > b->r.deadline_at) return 0; return a->r.id < b->r.id; } job job_copy(job j) { job n; if (!j) return NULL; n = malloc(sizeof(struct job) + j->r.body_size); if (!n) return twarnx("OOM"), (job) 0; memcpy(n, j, sizeof(struct job) + j->r.body_size); n->next = n->prev = n; /* not in a linked list */ n->file = NULL; /* copies do not have refcnt on the wal */ n->tube = 0; /* Don't use memcpy for the tube, which we must refcount. */ TUBE_ASSIGN(n->tube, j->tube); /* Mark this job as a copy so it can be appropriately freed later on */ n->r.state = Copy; return n; } const char * job_state(job j) { if (j->r.state == Ready) return "ready"; if (j->r.state == Reserved) return "reserved"; if (j->r.state == Buried) return "buried"; if (j->r.state == Delayed) return "delayed"; return "invalid"; } int job_list_any_p(job head) { return head->next != head || head->prev != head; } job job_remove(job j) { if (!j) return NULL; if (!job_list_any_p(j)) return NULL; /* not in a doubly-linked list */ j->next->prev = j->prev; j->prev->next = j->next; j->prev = j->next = j; return j; } void job_insert(job head, job j) { if (job_list_any_p(j)) return; /* already in a linked list */ j->prev = head->prev; j->next = head; head->prev->next = j; head->prev = j; } uint64 total_jobs() { return next_id - 1; } /* for unit tests */ size_t get_all_jobs_used() { return all_jobs_used; }