Rewrite using opaque record data.

Records can carry arbitrary data, whose format is defined using
teplate strings similar to those of Perl's pack() function.  The
teplate string is stored in the logfile header and is used by
binlogcat to render a human-readable representation.

* src/.gitignore: New file.
* pack.c: New file.
* pack.h: New file.
* src/Makefile.am: Add pack.c, pack.h
* src/binlog.c: Rewrite.  Use pack routines to construct each
record.
* src/binlogcat.c: Likewise.  Use packout to unpack each record.
* src/vmod-binlog.h (struct binlog_record): Remove nid,aid.
(struct binlog_file_header): Add hdrsize.
(BINLOG_HEADER_SIZE,union binlog_header): Remove
(binlog_recnum): Remove.
(binlog_size): Rewrite.
* src/vmod.vcc (append,sappend): Remove.
(init): Change signature.
(start,commit,pack): New functions.
* tests/test01.at: Rewrite.
* tests/test02.at: Rewrite.

1 files changed, 125 insertions, 36 deletions

diff --git a/src/binlog.c b/src/binlog.c
index be32e10..6f438a8 100644
--- a/src/binlog.c
+++ b/src/binlog.c
@@ -24,2 +24,3 @@
 #include <syslog.h>
+#include <stddef.h>
 #include <stdlib.h>
@@ -31,2 +32,3 @@
 #include "vmod-binlog.h"
+#include "pack.h"
 
@@ -38,2 +40,8 @@
 
+enum binlog_state {
+	state_init,
+	state_start,
+	state_pack
+};
+
 struct binlog_config {
@@ -47,6 +55,6 @@ struct binlog_config {
 	int fd;              /* current file descriptor */
-	union binlog_header *base;     /* mmap base */
-	struct binlog_record *recbase; /* record base */
+	struct binlog_file_header *base;     /* mmap base */
+	char *recbase;       /* record base */
 	size_t recnum;       /* number of records in recbase */
-	size_t recidx;       /* index of the next free entry in recbase */
+	size_t recsize;      /* record size */
 	time_t stoptime;     /* when to rotate the current file */
@@ -55,2 +63,9 @@ struct binlog_config {
 	int flags;
+
+	char *dataspec;
+	struct packinst *inst_head;
+	struct packinst *inst_cur;
+	struct packenv *env;
+	enum binlog_state state;
+	time_t timestamp;
 };
@@ -160,3 +175,4 @@ getinterval(char *p, char **endp)
 void
-vmod_init(struct sess *sp, struct vmod_priv *priv, const char *param)
+vmod_init(struct sess *sp, struct vmod_priv *priv,
+	  const char *dir, const char *dataspec, const char *param)
 {
@@ -164,3 +180,3 @@ vmod_init(struct sess *sp, struct vmod_priv *priv, const char *param)
 	struct stat st;
-	char *dir, *p, *q;
+	char *p, *q;
 	unsigned long n;
@@ -172,8 +188,3 @@ vmod_init(struct sess *sp, struct vmod_priv *priv, const char *param)
 	}
-	
-	dir = findparam(param, "dir");
-	if (!dir) {
-		binlog_error("parameter \"dir\" not set");
-		abort();
-	}
+
 	if (stat(dir, &st)) {
@@ -196,4 +207,16 @@ vmod_init(struct sess *sp, struct vmod_priv *priv, const char *param)
 	}
-	conf->dir = dir;
+	conf->dir = strdup(dir);
+	AN(conf->dir);
 
+	conf->inst_head = packcomp(dataspec, &p);
+	if (*p) {
+		binlog_error("cannot compile data format near %s", p);
+		abort();
+	}
+	conf->recsize = packsize(conf->inst_head);
+	conf->env = packenv_create(conf->recsize);
+	conf->recsize += offsetof(struct binlog_record,data);
+	conf->dataspec = strdup(dataspec);
+	AN(conf->dataspec);
+		
 	p = findparam(param, "pattern");
@@ -369,3 +392,2 @@ reset(struct binlog_config *conf)
 	conf->recnum = 0;
-	conf->recidx = 0;
 }
@@ -381,2 +403,5 @@ setstoptime(struct binlog_config *conf)
 
+#define binlog_recnum(conf) \
+	(((conf)->size - (conf)->base->hdrsize) / (conf)->base->recsize)
+
 static int
@@ -386,2 +411,3 @@ newfile(struct sess *sp, struct binlog_config *conf)
 	void *base;	
+	size_t n;
 	
@@ -415,10 +441,14 @@ newfile(struct sess *sp, struct binlog_config *conf)
 	conf->base = base;
-	memcpy(conf->base->hdr.magic, BINLOG_MAGIC_STR, BINLOG_MAGIC_LEN);
-	conf->base->hdr.version = BINLOG_VERSION;
-	conf->base->hdr.recsize = sizeof(struct binlog_record);
-	conf->base->hdr.recnum = 0;
+	memcpy(conf->base->magic, BINLOG_MAGIC_STR, BINLOG_MAGIC_LEN);
+	conf->base->version = BINLOG_VERSION;
+	conf->base->recsize = conf->recsize;
+	conf->base->recnum = 0;
+	strcpy((char*)(conf->base + 1), conf->dataspec);
 
-	conf->recbase = (struct binlog_record *) (conf->base + 1);
+	n = (sizeof(struct binlog_file_header) + strlen(conf->dataspec) +
+	     conf->recsize - 1) / conf->recsize;
+	conf->base->hdrsize = n * conf->recsize;
+
+	conf->recbase = (char *) conf->base + conf->base->hdrsize;
 	conf->recnum = binlog_recnum(conf);
-	conf->recidx = 0;
 
@@ -431,2 +461,4 @@ closefile(struct sess *sp, struct binlog_config *conf)
 {
+	size_t size;
+	
 	if (conf->fd == -1)
@@ -434,4 +466,5 @@ closefile(struct sess *sp, struct binlog_config *conf)
 	debug(conf,1,("closing log file %s",conf->fname));
+	size = binlog_size(conf->base);
 	munmap(conf->base, conf->size);
-	if (ftruncate(conf->fd, binlog_size(conf)))
+	if (ftruncate(conf->fd, size))
 		binlog_error("error truncating \"%s/%s\": %s",
@@ -442,6 +475,5 @@ closefile(struct sess *sp, struct binlog_config *conf)
 }
-	
 
 void
-vmod_append(struct sess *sp, struct vmod_priv *priv, int nid, int aid)
+vmod_start(struct sess *sp, struct vmod_priv *priv)
 {
@@ -461,2 +493,50 @@ vmod_append(struct sess *sp, struct vmod_priv *priv, int nid, int aid)
 	}
+
+	packenv_init(conf->env);
+	conf->state = state_start;
+	conf->inst_cur = conf->inst_head;
+	conf->timestamp = ts;
+}
+
+void
+vmod_pack(struct sess *sp, struct vmod_priv *priv, const char *str)
+{
+	struct binlog_config *conf = priv->priv;
+	char *argv[2];
+	
+	if (!conf)
+		return;
+
+	switch (conf->state) {
+	case state_start:
+	case state_pack:
+		break;
+	default:
+		binlog_error("pack called in wrong state (%d)", conf->state);
+		return;
+	}
+
+	if (!conf->inst_cur) {
+		binlog_error("format spec exhausted");
+		return;
+	}
+	
+	argv[0] = (char*) str;
+	argv[1] = NULL;
+	conf->env->argv = argv;
+	conf->env->argc = 2;
+	conf->env->argi = 0;
+
+	conf->inst_cur = packinnext(conf->inst_cur, conf->env);
+	
+	conf->state = state_pack;
+}
+
+void
+vmod_commit(struct sess *sp, struct vmod_priv *priv)
+{
+	struct binlog_config *conf = priv->priv;
+
+	if (!conf)
+		return;
 	if (conf->fd == -1)
@@ -464,20 +544,29 @@ vmod_append(struct sess *sp, struct vmod_priv *priv, int nid, int aid)
 
+	switch (conf->state) {
+	case state_start:
+		binlog_error("committing empty binlog record");
+		break;
+	case state_pack:
+		if (conf->inst_cur)
+			binlog_error("committing incomplete binlog record");
+		break;
+	default:
+		binlog_error("pack called in wrong state (%d)", conf->state);
+		return;
+	}
+
 	AZ(pthread_mutex_lock(&conf->mutex));
-	if (conf->recidx == conf->recnum) {
+	if (conf->base->recnum == conf->recnum) {
 		binlog_error("overflow of %s/%s", conf->dir, conf->fname);
 	} else {
-		struct binlog_record *p = conf->recbase + conf->recidx++;
-		p->nid = nid;
-		p->aid = aid;
-		p->ts = ts;
-		conf->base->hdr.recnum++;
+		struct binlog_record *p =
+			(struct binlog_record *)(conf->recbase +
+						 conf->base->recnum *
+						 conf->recsize);
+		p->ts = conf->timestamp;
+		memcpy(p->data, conf->env->buf_base, conf->env->buf_size);
+		conf->base->recnum++;
 	}
 	AZ(pthread_mutex_unlock(&conf->mutex));
-}
-
-void
-vmod_sappend(struct sess *sp, struct vmod_priv *priv, const char *nid,
-	     const char *aid)
-{
-	vmod_append(sp, priv, nid ? atoi(nid): 0, aid ? atoi(aid) : 0);
+	conf->state = state_init;
 }
@@ -494,3 +583,3 @@ vmod_sync(struct sess *sp, struct vmod_priv *priv)
 	if (conf->base)
-		msync(conf->base, binlog_size(conf), 0);
+		msync(conf->base, binlog_size(conf->base), 0);
 	AZ(pthread_mutex_unlock(&conf->mutex));