#include "jscan.h"
#include <dirent.h>
static void jalign(struct jfile *jf, enum jdirection direction);
static int jreadbuf(struct jfile *jf, enum jdirection direction,
void *buf, int bytes);
struct jfile *
jopen_fd(int fd)
{
struct jfile *jf;
jf = malloc(sizeof(struct jfile));
bzero(jf, sizeof(struct jfile));
jf->jf_fd = fd;
jf->jf_write_fd = -1;
jf->jf_open_flags = O_RDONLY;
jf->jf_pos = 0;
return(jf);
}
struct jfile *
jopen_prefix(const char *prefix, int rw)
{
struct jfile *jf;
struct jdata *jd;
unsigned int seq_beg = -1;
unsigned int seq_end = -1;
unsigned int seq;
struct stat st;
const char *dirname;
struct dirent *den;
DIR *dir;
char *basename;
char *data;
char *ptr;
int hastransid;
int baselen;
int fd;
dirname = data = strdup(prefix);
if ((basename = strrchr(dirname, '/')) != NULL) {
*basename++ = 0;
} else {
basename = data;
dirname = "./";
}
baselen = strlen(basename);
if ((dir = opendir(dirname)) != NULL) {
while ((den = readdir(dir)) != NULL) {
if (strncmp(den->d_name, basename, baselen) == 0 &&
den->d_name[baselen] == '.'
) {
seq = strtoul(den->d_name + baselen + 1, &ptr, 16);
if (*ptr == 0 && seq != UINT_MAX) {
if (seq_beg == (unsigned int)-1 || seq_beg > seq)
seq_beg = seq;
if (seq_end == (unsigned int)-1 || seq_end < seq)
seq_end = seq;
}
}
}
closedir(dir);
}
free(data);
hastransid = 0;
asprintf(&data, "%s.transid", prefix);
if (stat(data, &st) == 0)
hastransid = 1;
free(data);
if (seq_beg != (unsigned int)-1 || hastransid) {
if (seq_beg == (unsigned int)-1) {
seq_beg = 0;
seq_end = 0;
if (rw) {
asprintf(&data, "%s.%08x", prefix, 0);
if ((fd = open(data, O_RDWR|O_CREAT, 0666)) >= 0)
close(fd);
free(data);
}
}
jf = malloc(sizeof(struct jfile));
bzero(jf, sizeof(struct jfile));
jf->jf_fd = -1;
jf->jf_write_fd = -1;
jf->jf_prefix = strdup(prefix);
jf->jf_seq = seq_beg;
jf->jf_seq_beg = seq_beg;
jf->jf_seq_end = seq_end;
jf->jf_open_flags = rw ? (O_RDWR|O_CREAT) : O_RDONLY;
if (verbose_opt)
fprintf(stderr, "Open prefix set %08x-%08x\n", seq_beg, seq_end);
if ((jd = jread(jf, NULL, JD_BACKWARDS)) != NULL) {
jf->jf_last_transid = jd->jd_transid;
jfree(jf, jd);
}
} else {
jf = NULL;
}
return(jf);
}
int
jrecord_init(const char *prefix)
{
struct jfile *jf;
struct stat st;
char *data;
int hasseqspace;
int fd;
jf = jopen_prefix(prefix, 0);
hasseqspace = 0;
if (jf) {
if (jf->jf_seq_beg != (unsigned int)-1)
hasseqspace = 1;
jclose(jf);
}
asprintf(&data, "%s.transid", prefix);
if (hasseqspace) {
fd = open(data, O_RDWR, 0666);
} else {
fd = open(data, O_RDWR|O_CREAT, 0666);
}
free(data);
if (fd < 0)
return(-1);
if (fstat(fd, &st) == 0 && st.st_size == 0)
write(fd, "0000000000000000\n", 17);
close(fd);
return(0);
}
void
jclose(struct jfile *jf)
{
if (jf->jf_fd >= 0) {
close(jf->jf_fd);
jf->jf_fd = -1;
}
if (jf->jf_write_fd >= 0) {
close(jf->jf_write_fd);
jf->jf_write_fd = -1;
}
free(jf);
}
struct jdata *
jread(struct jfile *jf, struct jdata *jd, enum jdirection direction)
{
struct journal_rawrecbeg head;
struct journal_rawrecbeg *headp;
struct journal_rawrecend tail;
struct journal_rawrecend *tailp;
struct stat st;
char *filename;
int allocsize;
int recsize;
int search;
int error;
int n;
if (jd) {
assert(direction != JD_SEQFIRST && direction != JD_SEQLAST);
if (jf->jf_prefix) {
if (jf->jf_fd >= 0 && jf->jf_seq != jd->jd_seq) {
close(jf->jf_fd);
jf->jf_fd = -1;
}
jf->jf_seq = jd->jd_seq;
if (jf->jf_fd < 0) {
asprintf(&filename, "%s.%08x", jf->jf_prefix, jf->jf_seq);
jf->jf_fd = open(filename, O_RDONLY);
if (verbose_opt > 1)
fprintf(stderr, "Open %s fd %d\n", filename, jf->jf_fd);
free(filename);
}
}
if ((jmodes & JMODEF_INPUT_PIPE) == 0) {
if (direction == JD_FORWARDS) {
jf->jf_pos = jd->jd_pos + jd->jd_size;
lseek(jf->jf_fd, jf->jf_pos, 0);
} else {
jf->jf_pos = jd->jd_pos;
}
} else {
assert(direction == JD_FORWARDS && jf->jf_prefix == NULL);
assert(jf->jf_pos == jd->jd_pos + jd->jd_size);
}
jfree(jf, jd);
} else {
if (jf->jf_prefix) {
if (jf->jf_fd >= 0) {
close(jf->jf_fd);
jf->jf_fd = -1;
}
switch(direction) {
case JD_FORWARDS:
jf->jf_seq = jf->jf_seq_beg;
break;
case JD_BACKWARDS:
jf->jf_seq = jf->jf_seq_end;
break;
case JD_SEQFIRST:
direction = JD_FORWARDS;
break;
case JD_SEQLAST:
direction = JD_BACKWARDS;
break;
}
} else if ((jmodes & JMODEF_INPUT_PIPE) == 0) {
switch(direction) {
case JD_SEQFIRST:
direction = JD_FORWARDS;
case JD_FORWARDS:
jf->jf_pos = lseek(jf->jf_fd, 0L, SEEK_SET);
break;
case JD_SEQLAST:
direction = JD_BACKWARDS;
case JD_BACKWARDS:
jf->jf_pos = lseek(jf->jf_fd, 0L, SEEK_END);
break;
}
} else {
if (direction == JD_SEQFIRST)
direction = JD_FORWARDS;
assert(jf->jf_pos == 0 && direction == JD_FORWARDS);
}
}
top:
if (jf->jf_fd < 0) {
asprintf(&filename, "%s.%08x", jf->jf_prefix, jf->jf_seq);
jf->jf_fd = open(filename, O_RDONLY);
if (verbose_opt > 1)
fprintf(stderr, "Open %s fd %d\n", filename, jf->jf_fd);
free(filename);
if (direction == JD_FORWARDS)
jf->jf_pos = lseek(jf->jf_fd, 0L, SEEK_SET);
else
jf->jf_pos = lseek(jf->jf_fd, 0L, SEEK_END);
}
if (jf->jf_pos & 15) {
jf_warn(jf, "realigning bad offset and entering search mode");
jalign(jf, direction);
search = 1;
} else {
search = 0;
}
error = 0;
if (direction == JD_FORWARDS) {
while ((error = jreadbuf(jf, direction, &head, sizeof(head))) == sizeof(head)) {
if (head.begmagic != JREC_BEGMAGIC) {
if (search == 0)
jf_warn(jf, "bad beginmagic, searching for new record");
search = 1;
jalign(jf, direction);
continue;
}
recsize = (head.recsize + 15) & ~15;
if (recsize < JREC_MINRECSIZE || recsize > JREC_MAXRECSIZE) {
if (search == 0)
jf_warn(jf, "bad recordsize: %d\n", recsize);
search = 1;
jalign(jf, direction);
continue;
}
allocsize = offsetof(struct jdata, jd_data[recsize]);
allocsize = (allocsize + 255) & ~255;
jd = malloc(allocsize);
bzero(jd, offsetof(struct jdata, jd_data[0]));
bcopy(&head, jd->jd_data, sizeof(head));
n = jreadbuf(jf, direction, jd->jd_data + sizeof(head),
recsize - sizeof(head));
if (n != (int)(recsize - sizeof(head))) {
if (search == 0)
jf_warn(jf, "Incomplete stream record\n");
search = 1;
jalign(jf, direction);
free(jd);
continue;
}
tailp = (void *)(jd->jd_data + recsize - sizeof(*tailp));
if (tailp->endmagic != JREC_ENDMAGIC) {
if (search == 0)
jf_warn(jf, "bad endmagic, searching for new record");
search = 1;
jalign(jf, direction);
free(jd);
continue;
}
if (head.streamid == JREC_STREAMID_PAD) {
free(jd);
continue;
}
jd->jd_transid = head.transid;
jd->jd_alloc = allocsize;
jd->jd_size = recsize;
jd->jd_seq = jf->jf_seq;
jd->jd_pos = jf->jf_pos - recsize;
jd->jd_refs = 1;
return(jd);
}
} else {
while ((error = jreadbuf(jf, direction, &tail, sizeof(tail))) == sizeof(tail)) {
if (tail.endmagic != JREC_ENDMAGIC) {
if (search == 0)
jf_warn(jf, "bad endmagic, searching for new record");
search = 1;
jalign(jf, direction);
continue;
}
recsize = (tail.recsize + 15) & ~15;
if (recsize < JREC_MINRECSIZE || recsize > JREC_MAXRECSIZE) {
if (search == 0)
jf_warn(jf, "bad recordsize: %d\n", recsize);
search = 1;
jalign(jf, direction);
continue;
}
allocsize = offsetof(struct jdata, jd_data[recsize]);
allocsize = (allocsize + 255) & ~255;
jd = malloc(allocsize);
bzero(jd, offsetof(struct jdata, jd_data[0]));
bcopy(&tail, jd->jd_data + recsize - sizeof(tail), sizeof(tail));
n = jreadbuf(jf, direction, jd->jd_data, recsize - sizeof(tail));
if (n != (int)(recsize - sizeof(tail))) {
if (search == 0)
jf_warn(jf, "Incomplete stream record\n");
search = 1;
jalign(jf, direction);
free(jd);
continue;
}
headp = (void *)jd->jd_data;
if (headp->begmagic != JREC_BEGMAGIC) {
if (search == 0)
jf_warn(jf, "bad begmagic, searching for new record");
search = 1;
jalign(jf, direction);
free(jd);
continue;
}
if (head.streamid == JREC_STREAMID_PAD) {
free(jd);
continue;
}
jd->jd_transid = headp->transid;
jd->jd_alloc = allocsize;
jd->jd_size = recsize;
jd->jd_seq = jf->jf_seq;
jd->jd_pos = jf->jf_pos;
jd->jd_refs = 1;
return(jd);
}
}
if (error == 0 && jf->jf_prefix) {
if (direction == JD_FORWARDS) {
if (jf->jf_seq < jf->jf_seq_end) {
++jf->jf_seq;
if (verbose_opt)
fprintf(stderr, "jread: roll to seq %08x\n", jf->jf_seq);
if (jf->jf_fd >= 0) {
close(jf->jf_fd);
jf->jf_fd = -1;
}
goto top;
}
if (jmodes & JMODEF_LOOP_FOREVER) {
asprintf(&filename, "%s.%08x", jf->jf_prefix, jf->jf_seq + 1);
if (stat(filename, &st) == 0) {
++jf->jf_seq_end;
if (verbose_opt)
fprintf(stderr, "jread: roll seq_end to %08x\n",
jf->jf_seq_end);
} else {
sleep(5);
}
goto top;
}
} else {
if (jf->jf_seq > jf->jf_seq_beg) {
--jf->jf_seq;
if (verbose_opt)
fprintf(stderr, "jread: roll to seq %08x\n", jf->jf_seq);
if (jf->jf_fd >= 0) {
close(jf->jf_fd);
jf->jf_fd = -1;
}
goto top;
}
}
}
if (error == 0 && jf->jf_prefix == NULL &&
(jmodes & JMODEF_LOOP_FOREVER) &&
!(jmodes & JMODEF_INPUT_PIPE) &&
direction == JD_FORWARDS
) {
sleep(5);
goto top;
}
return(NULL);
}
void
jwrite(struct jfile *jf, struct jdata *jd)
{
struct stat st;
char *path;
int n;
assert(jf->jf_prefix);
again:
if (jf->jf_write_fd < 0) {
asprintf(&path, "%s.%08x", jf->jf_prefix, jf->jf_seq_end);
jf->jf_write_fd = open(path, O_RDWR|O_CREAT, 0666);
if (jf->jf_write_fd < 0 || fstat(jf->jf_write_fd, &st) != 0) {
fprintf(stderr, "Unable to open/create %s\n", path);
exit(1);
}
jf->jf_write_pos = st.st_size;
lseek(jf->jf_write_fd, jf->jf_write_pos, 0);
free(path);
}
if (jf->jf_write_pos > 0 &&
jf->jf_write_pos + jd->jd_size > prefix_file_size
) {
close(jf->jf_write_fd);
jf->jf_write_fd = -1;
++jf->jf_seq_end;
goto again;
}
n = write(jf->jf_write_fd, jd->jd_data, jd->jd_size);
if (n != jd->jd_size) {
ftruncate(jf->jf_write_fd, jf->jf_write_pos);
fprintf(stderr, "jwrite: failed %s\n", strerror(errno));
exit(1);
}
jf->jf_write_pos += n;
jf->jf_last_transid = jd->jd_transid;
}
struct jdata *
jseek(struct jfile *jf, int64_t transid, enum jdirection direction)
{
unsigned int seq;
struct jdata *jd = NULL;
if (jmodes & JMODEF_INPUT_PIPE) {
assert(direction == JD_FORWARDS);
return (jread(jf, NULL, direction));
}
if (jf->jf_prefix) {
if (verbose_opt > 2) {
fprintf(stderr, "jseek prefix set %s %08x-%08x\n", jf->jf_prefix,
jf->jf_seq_beg, jf->jf_seq_end);
}
jd = NULL;
for (seq = jf->jf_seq_end; seq != jf->jf_seq_beg - 1; --seq) {
if (verbose_opt > 2)
fprintf(stderr, "try seq %08x\n", seq);
jf->jf_seq = seq;
if ((jd = jread(jf, NULL, JD_SEQFIRST)) != NULL) {
if (jd->jd_transid == transid)
return(jd);
if (jd->jd_transid < transid) {
jfree(jf, jd);
break;
}
jfree(jf, jd);
}
}
if (seq == jf->jf_seq_beg - 1) {
if (direction == JD_BACKWARDS)
return(NULL);
else
return(jread(jf, NULL, JD_FORWARDS));
}
if (verbose_opt > 1)
fprintf(stderr, "jseek input prefix set to seq %08x\n", seq);
}
jd = jread(jf, NULL, JD_SEQLAST);
while (jd != NULL) {
if (jd->jd_transid <= transid) {
if (jd->jd_transid < transid) {
if (direction == JD_FORWARDS)
jd =jread(jf, jd, JD_FORWARDS);
}
if (verbose_opt > 1) {
fprintf(stderr, "jseek returning seq %08x offset 0x%08jx\n",
jd->jd_seq, (uintmax_t)jd->jd_pos);
}
return(jd);
}
jd = jread(jf, jd, JD_BACKWARDS);
}
if (direction == JD_BACKWARDS)
return(NULL);
else
return(jread(jf, NULL, JD_FORWARDS));
}
struct jdata *
jref(struct jdata *jd)
{
++jd->jd_refs;
return(jd);
}
void
jfree(struct jfile *jf __unused, struct jdata *jd)
{
if (--jd->jd_refs == 0)
free(jd);
}
static void
jalign(struct jfile *jf, enum jdirection direction)
{
char dummy[16];
int bytes;
if ((int)jf->jf_pos & 15) {
if (direction == JD_FORWARDS) {
bytes = 16 - ((int)jf->jf_pos & 15);
jreadbuf(jf, direction, dummy, bytes);
} else {
jf->jf_pos = jf->jf_pos & ~(off_t)15;
}
}
}
static int
jreadbuf(struct jfile *jf, enum jdirection direction, void *buf, int bytes)
{
int ttl = 0;
int n;
if (jf->jf_fd < 0)
return(0);
if (direction == JD_FORWARDS) {
while (ttl != bytes) {
n = read(jf->jf_fd, (char *)buf + ttl, bytes - ttl);
if (n <= 0) {
if (n < 0 && ttl == 0)
ttl = -errno;
break;
}
ttl += n;
jf->jf_pos += n;
}
} else {
if (jf->jf_pos >= bytes) {
jf->jf_pos -= bytes;
lseek(jf->jf_fd, jf->jf_pos, 0);
while (ttl != bytes) {
n = read(jf->jf_fd, (char *)buf + ttl, bytes - ttl);
if (n <= 0) {
if (n < 0 && ttl == 0)
ttl = -errno;
break;
}
ttl += n;
}
}
}
return(ttl);
}