@@ -255,6 +255,7 @@ main(int argc, char **argv) { int r; struct event_base *ev_base; + struct job binlog_jobs = {}; progname = argv[0]; opts(argc, argv); @@ -278,8 +279,11 @@ main(int argc, char **argv) nudge_fd_limit(); unbrake((evh) h_accept); - prot_replay_binlog(); + + binlog_jobs.prev = binlog_jobs.next = &binlog_jobs; + binlog_read(&binlog_jobs); binlog_init(); + prot_replay_binlog(&binlog_jobs); if (detach) { daemonize(); beanstalkd.c @@ -40,12 +40,14 @@ size_t binlog_size_limit = 10 << 20; char *binlog_dir = NULL; static int binlog_index = 0; -static int binlog_fd = -1; -static int binlog_version = 1; +static int binlog_fd = -1, next_binlog_fd = -1; +static int binlog_version = 2; +static size_t binlog_space = 0, next_binlog_space = 0; +static size_t binlog_reserved = 0, next_binlog_reserved = 0; static size_t bytes_written; static int lock_fd; -static binlog first_binlog = NULL, last_binlog = NULL; +static binlog first_binlog = NULL, last_binlog = NULL, next_binlog = NULL; static int binlog_scan_dir() @@ -110,12 +112,19 @@ binlog_dref(binlog b) } } +/* static void -binlog_warn(const char *msg, int fd, const char* path) +binlog_warn(int fd, const char* path, const char *msg) { warnx("WARNING, %s at %s:%u.\n%s", msg, path, lseek(fd, 0, SEEK_CUR), - " Continuing with next file. You may be missing data."); + " Continuing. You may be missing data."); } +*/ + +#define binlog_warn(fd, path, fmt, args...) \ + warnx("WARNING, " fmt " at %s:%u. %s: ", \ + ##args, path, lseek(fd, 0, SEEK_CUR), \ + "Continuing. You may be missing data.") static void binlog_read_one(int fd, job binlog_jobs, const char *path) @@ -131,7 +140,7 @@ binlog_read_one(int fd, job binlog_jobs, const char *path) r = read(fd, &version, sizeof(version)); if (r == -1) return twarn("read()"); if (r < sizeof(version)) { - return binlog_warn("version record is too short", fd, path); + return binlog_warn(fd, path, "EOF while reading version record"); } if (version != binlog_version) { @@ -145,7 +154,7 @@ binlog_read_one(int fd, job binlog_jobs, const char *path) if (r == -1) return twarn("read()"); if (r < namelen) { lseek(fd, SEEK_CUR, 0); - return binlog_warn("tube name is too short", fd, path); + return binlog_warn(fd, path, "EOF while reading tube name"); } } @@ -153,9 +162,11 @@ binlog_read_one(int fd, job binlog_jobs, const char *path) r = read(fd, &js, sizeof(struct job)); if (r == -1) return twarn("read()"); if (r < sizeof(struct job)) { - return binlog_warn("job record is too short", fd, path); + return binlog_warn(fd, path, "EOF while reading job record"); } + if (!js.id) break; + j = job_find(js.id); switch (js.state) { case JOB_STATE_INVALID: @@ -175,6 +186,16 @@ binlog_read_one(int fd, job binlog_jobs, const char *path) j->next = j->prev = j; j->creation = js.creation; job_insert(binlog_jobs, j); + } + if (js.body_size) { + if (js.body_size > j->body_size) { + warnx("job size increased from %zu to %zu", j->body_size, + js.body_size); + job_remove(j); + binlog_dref(j->binlog); + job_free(j); + return binlog_warn(fd, path, "EOF while reading job body"); + } r = read(fd, j->body, js.body_size); if (r == -1) return twarn("read()"); if (r < js.body_size) { @@ -182,7 +203,7 @@ binlog_read_one(int fd, job binlog_jobs, const char *path) job_remove(j); binlog_dref(j->binlog); job_free(j); - return binlog_warn("job body is too short", fd, path); + return binlog_warn(fd, path, "EOF while reading job body"); } } break; @@ -207,8 +228,8 @@ binlog_read_one(int fd, job binlog_jobs, const char *path) } } -void -binlog_close() +static void +binlog_close_last() { if (binlog_fd < 0) return; close(binlog_fd); @@ -217,15 +238,35 @@ binlog_close() } static binlog -add_binlog(char *path) +make_binlog(char *path) { binlog b; - b = (binlog)malloc(sizeof(struct binlog) + strlen(path) + 1); + b = (binlog) malloc(sizeof(struct binlog) + strlen(path) + 1); if (!b) return twarnx("OOM"), NULL; strcpy(b->path, path); b->refs = 0; b->next = NULL; + return b; +} + +static binlog +make_next_binlog() +{ + int r; + char path[PATH_MAX]; + + if (!binlog_dir) return NULL; + + r = snprintf(path, PATH_MAX, "%s/binlog.%d", binlog_dir, ++binlog_index); + if (r > PATH_MAX) return twarnx("path too long: %s", binlog_dir), NULL; + + return make_binlog(path); +} + +static binlog +add_binlog(binlog b) +{ if (last_binlog) last_binlog->next = b; last_binlog = b; if (!first_binlog) first_binlog = b; @@ -234,59 +275,82 @@ add_binlog(char *path) } static int -binlog_open() +binlog_open(binlog log) { - char path[PATH_MAX]; - binlog b; int fd, r; - if (!binlog_dir) return -1; - r = snprintf(path, PATH_MAX, "%s/binlog.%d", binlog_dir, ++binlog_index); - if (r > PATH_MAX) return twarnx("path too long: %s", binlog_dir), -1; + if (!binlog_iref(log)) return -1; - if (!binlog_iref(add_binlog(path))) return -1; - fd = open(path, O_WRONLY | O_CREAT, 0400); + fd = open(log->path, O_WRONLY | O_CREAT, 0400); - if (fd < 0) { - twarn("Cannot open binlog %s", path); - return -1; - } + if (fd < 0) return twarn("Cannot open binlog %s", log->path), -1; + r = posix_fallocate(fd, 0, binlog_size_limit); + if (r) { + close(fd); + binlog_dref(log); + errno = r; + return twarn("Cannot allocate space for binlog %s", log->path), -1; + } bytes_written = write(fd, &binlog_version, sizeof(int)); if (bytes_written < sizeof(int)) { twarn("Cannot write to binlog"); close(fd); - binlog_dref(last_binlog); + binlog_dref(log); return -1; } return fd; } -static void -binlog_open_next() +/* returns 1 on success, 0 on error. */ +static int +binlog_use_next() { - if (binlog_fd < 0) return; - close(binlog_fd); - binlog_dref(last_binlog); - binlog_fd = binlog_open(); + if (binlog_fd < 0) return 0; + if (next_binlog_fd < 0) return 0; + if (binlog_reserved > next_binlog_space) return twarnx("overextended"), 0; + + binlog_close_last(); + + binlog_fd = next_binlog_fd; + add_binlog(next_binlog); + + next_binlog = NULL; + next_binlog_fd = -1; + + binlog_space = next_binlog_space - binlog_reserved; + binlog_reserved = next_binlog_reserved + binlog_reserved; + + next_binlog_reserved = next_binlog_space = 0; + return 1; } void +binlog_close() +{ + binlog_use_next(); + binlog_close_last(); +} + +/* Returns the number of jobs successfully written (either 0 or 1). */ +/* If we are not using the binlog at all (binlog_fd < 0), then we pretend to + have made a successful write and return 1. */ +int binlog_write_job(job j) { - size_t tube_namelen, to_write; + ssize_t written; + size_t tube_namelen, to_write = 0; struct iovec vec[4], *vptr; - int vcnt = 3; + int vcnt = 3, r; - if (binlog_fd < 0) return; + if (binlog_fd < 0) return 1; tube_namelen = 0; vec[0].iov_base = (char *) &tube_namelen; - vec[0].iov_len = sizeof(size_t); - to_write = sizeof(size_t); + to_write += vec[0].iov_len = sizeof(size_t); vec[1].iov_base = j->tube->name; vec[1].iov_len = 0; @@ -294,38 +358,42 @@ binlog_write_job(job j) /* we could save some bytes in the binlog file by only saving some parts of * the job struct */ vec[2].iov_base = (char *) j; - vec[2].iov_len = sizeof(struct job); - to_write += sizeof(struct job); + to_write += vec[2].iov_len = sizeof(struct job); + + printf("writing job %lld state %d\n", j->id, j->state); if (j->state == JOB_STATE_READY || j->state == JOB_STATE_DELAYED) { if (!j->binlog) { tube_namelen = strlen(j->tube->name); - vec[1].iov_len = tube_namelen; - to_write += tube_namelen; + to_write += vec[1].iov_len = tube_namelen; vcnt = 4; vec[3].iov_base = j->body; - vec[3].iov_len = j->body_size; - to_write += j->body_size; + to_write += vec[3].iov_len = j->body_size; } } else if (j->state == JOB_STATE_INVALID) { if (j->binlog) binlog_dref(j->binlog); j->binlog = NULL; } else { - return twarnx("unserializable job state: %d", j->state); + return twarnx("unserializable job state: %d", j->state), 0; } - if ((bytes_written + to_write) > binlog_size_limit) binlog_open_next(); - if (binlog_fd < 0) return; + if (to_write > binlog_reserved) { + r = binlog_use_next(); + if (!r) return twarnx("failed to use next binlog"), 0; + } if (j->state && !j->binlog) j->binlog = binlog_iref(last_binlog); while (to_write > 0) { - size_t written = writev(binlog_fd, vec, vcnt); + written = writev(binlog_fd, vec, vcnt); if (written < 0) { - twarn("Cannot write to binlog"); - binlog_close(); - return; + if (errno == EAGAIN) continue; + if (errno == EINTR) continue; + + twarn("writev"); + binlog_close_last(); + return 0; } bytes_written += written; @@ -339,8 +407,74 @@ binlog_write_job(job j) vptr->iov_len -= written; } } + + return 1; } +/* Returns the number of bytes successfully reserved: either 0 or n. */ +static size_t +binlog_reserve_space(size_t n) +{ + /* This value must be nonzero but is otherwise ignored. */ + if (binlog_fd < 0) return 1; + + if (n <= binlog_space) { + binlog_space -= n; + binlog_reserved += n; + return n; + } + + if (n <= next_binlog_space) { + next_binlog_space -= n; + next_binlog_reserved += n; + return n; + } + + /* The next binlog is already allocated and it is full. */ + if (next_binlog_fd >= 0) return 0; + + /* open a new binlog with more space to reserve */ + next_binlog = make_next_binlog(); + if (!next_binlog) return twarnx("error making next binlog"), 0; + next_binlog_fd = binlog_open(next_binlog); + + /* open failed, so we can't reserve any space */ + if (next_binlog_fd < 0) return 0; + + next_binlog_space = binlog_size_limit - bytes_written - n; + next_binlog_reserved = n; + + return n; +} + +/* Returns the number of bytes reserved. */ +size_t +binlog_reserve_space_put(job j) +{ + size_t z = 0; + + /* reserve space for the initial job record */ + z += sizeof(size_t); + z += strlen(j->tube->name); + z += sizeof(struct job); + z += j->body_size; + + /* plus space for a delete to come later */ + z += sizeof(size_t); + z += sizeof(struct job); + + return binlog_reserve_space(z); +} + +size_t +binlog_reserve_space_update(job j) +{ + size_t z = 0; + + z += sizeof(size_t); + z += sizeof(struct job); + return binlog_reserve_space(z); +} void binlog_read(job binlog_jobs) @@ -372,7 +506,7 @@ binlog_read(job binlog_jobs) if (fd < 0) { twarn("%s", path); } else { - b = binlog_iref(add_binlog(path)); + b = binlog_iref(add_binlog(make_binlog(path))); binlog_read_one(fd, binlog_jobs, path); close(fd); binlog_dref(b); @@ -408,7 +542,14 @@ binlog_lock() void binlog_init() { - binlog_fd = binlog_open(); + binlog log; + + if (!binlog_dir) return; + + log = make_next_binlog(); + if (!log) return twarnx("error making first binlog"); + binlog_fd = binlog_open(log); + if (binlog_fd >= 0) add_binlog(log); } const char * binlog.c @@ -32,10 +32,16 @@ struct binlog { extern char *binlog_dir; extern size_t binlog_size_limit; +void binlog_init(); + /* Return the number of locks acquired: either 0 or 1. */ int binlog_lock(); -void binlog_write_job(job j); +/* Returns the number of jobs successfully written (either 0 or 1). */ +int binlog_write_job(job j); +size_t binlog_reserve_space_put(job j); +size_t binlog_reserve_space_update(job j); + void binlog_read(job binlog_jobs); void binlog_close(); const char *binlog_oldest_index(); binlog.h @@ -118,6 +118,7 @@ allocate_job(int body_size) j->tube = NULL; j->binlog = NULL; j->heap_index = 0; + j->reserved_binlog_space = 0; return j; } job.c @@ -32,8 +32,11 @@ typedef int(*job_cmp_fn)(job, job); #define JOB_STATE_BURIED 3 #define JOB_STATE_DELAYED 4 +/* If you modify this struct, you MUST increment binlog format version in + * binlog.c. */ struct job { - job prev, next; /* linked list of jobs */ + + /* persistent fields; these get written to the binlog */ unsigned long long int id; unsigned int pri; unsigned int delay; @@ -45,12 +48,19 @@ struct job { unsigned int release_ct; unsigned int bury_ct; unsigned int kick_ct; - tube tube; - void *reserver; - void *binlog; char state; + + /* 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 a heap */ + size_t heap_index; /* where is this job in its current heap */ + void *binlog; + void *reserver; + size_t reserved_binlog_space; + + /* variable-size job data; written separately to the binlog */ char body[]; }; job.h @@ -426,7 +426,7 @@ set_main_delay_timeout() } static int -enqueue_job(job j, unsigned int delay) +enqueue_job(job j, unsigned int delay, char update_store) { int r; @@ -447,21 +447,37 @@ enqueue_job(job j, unsigned int delay) j->tube->stat.urgent_ct++; } } - binlog_write_job(j); + + if (update_store) { + r = binlog_write_job(j); + if (!r) return -1; + } + process_queue(); return 1; } -static void -bury_job(job j) +static int +bury_job(job j, char update_store) { + size_t z; + + if (update_store) { + z = binlog_reserve_space_update(j); + if (!z) return 0; + j->reserved_binlog_space += z; + } + job_insert(&j->tube->buried, j); global_stat.buried_ct++; j->tube->stat.buried_ct++; j->state = JOB_STATE_BURIED; j->reserver = NULL; j->bury_ct++; - binlog_write_job(j); + + if (update_store) return binlog_write_job(j); + + return 1; } void @@ -472,8 +488,8 @@ enqueue_reserved_jobs(conn c) while (job_list_any_p(&c->reserved_jobs)) { j = job_remove(c->reserved_jobs.next); - r = enqueue_job(j, 0); - if (!r) bury_job(j); + r = enqueue_job(j, 0, 0); + if (r < 1) bury_job(j, 0); global_stat.reserved_ct--; j->tube->stat.reserved_ct--; c->soonest_job = NULL; @@ -493,15 +509,21 @@ kick_buried_job(tube t) { int r; job j; + size_t z; if (!buried_job_p(t)) return 0; j = remove_buried_job(t->buried.next); + + z = binlog_reserve_space_update(j); + if (!z) return pq_give(&t->delay, j), 0; /* put it back */ + j->reserved_binlog_space += z; + j->kick_ct++; - r = enqueue_job(j, 0); - if (r) return 1; + r = enqueue_job(j, 0, 1); + if (r == 1) return 1; /* ready queue is full, so bury it */ - bury_job(j); + bury_job(j, 0); return 0; } @@ -524,19 +546,25 @@ kick_delayed_job(tube t) { int r; job j; + size_t z; j = pq_take(&t->delay); if (!j) return 0; + + z = binlog_reserve_space_update(j); + if (!z) return pq_give(&t->delay, j), 0; /* put it back */ + j->reserved_binlog_space += z; + j->kick_ct++; - r = enqueue_job(j, 0); - if (r) return 1; + r = enqueue_job(j, 0, 1); + if (r == 1) return 1; /* ready queue is full, so delay it again */ - r = enqueue_job(j, j->delay); - if (r) return 0; + r = enqueue_job(j, j->delay, 0); + if (r == 1) return 0; /* last resort */ - bury_job(j); + bury_job(j, 0); return 0; } @@ -745,16 +773,22 @@ enqueue_incoming_job(conn c) return reply_serr(c, MSG_DRAINING); } + if (j->reserved_binlog_space) return reply_serr(c, MSG_INTERNAL_ERROR); + j->reserved_binlog_space = binlog_reserve_space_put(j); + if (!j->reserved_binlog_space) return reply_serr(c, MSG_OUT_OF_MEMORY); + /* we have a complete job, so let's stick it in the pqueue */ - r = enqueue_job(j, j->delay); + r = enqueue_job(j, j->delay, 1); + if (r < 0) return reply_serr(c, MSG_INTERNAL_ERROR); + op_ct[OP_PUT]++; /* stats */ global_stat.total_jobs_ct++; j->tube->stat.total_jobs_ct++; - if (r) return reply_line(c, STATE_SENDWORD, MSG_INSERTED_FMT, j->id); + if (r == 1) return reply_line(c, STATE_SENDWORD, MSG_INSERTED_FMT, j->id); /* out of memory trying to grow the queue, so it gets buried */ - bury_job(j); + bury_job(j, 0); reply_line(c, STATE_SENDWORD, MSG_BURIED_FMT, j->id); } @@ -1019,6 +1053,7 @@ static void dispatch_cmd(conn c) { int r, i, timeout = -1; + size_t z; unsigned int count; job j; unsigned char type; @@ -1174,9 +1209,11 @@ dispatch_cmd(conn c) if (!j) return reply(c, MSG_NOTFOUND, MSG_NOTFOUND_LEN, STATE_SENDWORD); j->state = JOB_STATE_INVALID; - binlog_write_job(j); + r = binlog_write_job(j); job_free(j); + if (!r) return reply_serr(c, MSG_INTERNAL_ERROR); + reply(c, MSG_DELETED, MSG_DELETED_LEN, STATE_SENDWORD); break; case OP_RELEASE: @@ -1195,14 +1232,26 @@ dispatch_cmd(conn c) if (!j) return reply(c, MSG_NOTFOUND, MSG_NOTFOUND_LEN, STATE_SENDWORD); + /* We want to update the delay deadline on disk, so reserve space for + * that. */ + if (delay) { + z = binlog_reserve_space_update(j); + if (!z) return reply_serr(c, MSG_OUT_OF_MEMORY); + j->reserved_binlog_space += z; + } + j->pri = pri; j->delay = delay; j->release_ct++; - r = enqueue_job(j, delay); - if (r) return reply(c, MSG_RELEASED, MSG_RELEASED_LEN, STATE_SENDWORD); + + r = enqueue_job(j, delay, !!delay); + if (r < 0) return reply_serr(c, MSG_INTERNAL_ERROR); + if (r == 1) { + return reply(c, MSG_RELEASED, MSG_RELEASED_LEN, STATE_SENDWORD); + } /* out of memory trying to grow the queue, so it gets buried */ - bury_job(j); + bury_job(j, 0); reply(c, MSG_BURIED, MSG_BURIED_LEN, STATE_SENDWORD); break; case OP_BURY: @@ -1219,7 +1268,8 @@ dispatch_cmd(conn c) if (!j) return reply(c, MSG_NOTFOUND, MSG_NOTFOUND_LEN, STATE_SENDWORD); j->pri = pri; - bury_job(j); + r = bury_job(j, 1); + if (!r) return reply_serr(c, MSG_INTERNAL_ERROR); reply(c, MSG_BURIED, MSG_BURIED_LEN, STATE_SENDWORD); break; case OP_KICK: @@ -1399,8 +1449,8 @@ h_conn_timeout(conn c) timeout_ct++; /* stats */ j->timeout_ct++; - r = enqueue_job(remove_this_reserved_job(c, j), 0); - if (!r) bury_job(j); /* there was no room in the queue, so bury it */ + r = enqueue_job(remove_this_reserved_job(c, j), 0, 0); + if (r < 1) bury_job(j, 0); /* out of memory, so bury it */ r = conn_update_evq(c, c->evq.ev_events); if (r == -1) return twarnx("conn_update_evq() failed"), conn_close(c); } @@ -1590,8 +1640,8 @@ h_delay() while ((j = delay_q_peek())) { if (j->deadline > t) break; j = delay_q_take(); - r = enqueue_job(j, 0); - if (!r) bury_job(j); /* there was no room in the queue, so bury it */ + r = enqueue_job(j, 0, 0); + if (r < 1) bury_job(j, 0); /* out of memory, so bury it */ } set_main_delay_timeout(); @@ -1642,28 +1692,27 @@ prot_init() } void -prot_replay_binlog() +prot_replay_binlog(job binlog_jobs) { - struct job binlog_jobs; job j, nj; unsigned int delay; + int r; - binlog_jobs.prev = binlog_jobs.next = &binlog_jobs; - binlog_read(&binlog_jobs); - - for (j = binlog_jobs.next ; j != &binlog_jobs ; j = nj) { + for (j = binlog_jobs->next ; j != binlog_jobs ; j = nj) { nj = j->next; job_remove(j); + binlog_reserve_space_update(j); /* reserve space for a delete */ delay = 0; switch (j->state) { case JOB_STATE_BURIED: - bury_job(j); + bury_job(j, 0); break; case JOB_STATE_DELAYED: if (start_time < j->deadline) delay = j->deadline - start_time; /* fall through */ default: - enqueue_job(j,delay); + r = enqueue_job(j, delay, 0); + if (r < 1) twarnx("error processing binlog job %llu", j->id); } } } prot.c @@ -35,6 +35,6 @@ void enqueue_reserved_jobs(conn c); void enter_drain_mode(int sig); void h_accept(const int fd, const short which, struct event *ev); void prot_remove_tube(tube t); -void prot_replay_binlog(); +void prot_replay_binlog(job binlog_jobs); #endif /*prot_h*/ prot.h @@ -43,7 +43,8 @@ make_tube(const char *name) pq_init(&t->ready, job_pri_cmp); pq_init(&t->delay, job_delay_cmp); - t->buried = (struct job) { &t->buried, &t->buried, 0 }; + t->buried = (struct job) { }; + t->buried.prev = t->buried.next = &t->buried; ms_init(&t->waiting, NULL, NULL); t->stat = (struct stats) {0, 0, 0, 0}; tube.c 8313430cdc94eebc9eb66b95c41d46a49fff0df9 Keith Rarick kr@xph.us http://github.com/kr/beanstalkd/commit/12299e05b4a0e472c97fe73073d5e2d5bcecd5a7 12299e05b4a0e472c97fe73073d5e2d5bcecd5a7 2009-05-05T16:55:06-07:00 2009-01-17T10:56:10-08:00 Allocate disk space more carefully. Closes gh-1. c9e9f28c41d21b7dbd38f7ca25b1fc4917f21001 Keith Rarick kr@xph.us