gdk_logger.mx

来自「这个是内存数据库中的一个管理工具」· MX 代码 · 共 1,528 行 · 第 1/3 页

@' The contents of this file are subject to the MonetDB Public License
@' Version 1.1 (the "License"); you may not use this file except in
@' compliance with the License. You may obtain a copy of the License at
@' http://monetdb.cwi.nl/Legal/MonetDBLicense-1.1.html
@'
@' Software distributed under the License is distributed on an "AS IS"
@' basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See the
@' License for the specific language governing rights and limitations
@' under the License.
@'
@' The Original Code is the MonetDB Database System.
@'
@' The Initial Developer of the Original Code is CWI.
@' Portions created by CWI are Copyright (C) 1997-2007 CWI.
@' All Rights Reserved.

@f gdk_logger
@t Transactions
@a N. J. Nes
@v 2.0
@T
@* Introduction
In the philosophy of MonetDB, transaction management overhead should only
be paid when necessary. Transaction management is for this purpose 
implemented as a separate module and applications are required to
obey the transaction policy, e.g. obtaining/releasing locks. 

This module is designed to support efficient logging of the SQL database.
Once loaded, the SQL compiler will insert the proper calls at
transaction commit to include the changes in the log file.

The logger uses a directory to store its log files. One master log file
stores information about the version of the logger and the transaction 
log files. This file is a simple ascii file with the following format: 
 	6DIGIT-VERSION\n[log file number \n]*]*
The transaction log files have a binary format, which stores fixed size 
logformat headers (flag,nr,bid), where the flag is the type of update logged.
The nr field indicates how many changes there were (in case of inserts/deletes).
The bid stores the bid identifier.

The key decision to be made by the user is the location of the log file.
Ideally, it should be stored in fail-safe environment, or at least
the log and databases should be on separate disk columns.

This file system may reside on the same hardware as the database server
and therefore the writes are done to the same disk, but could also 
reside on another system and then the changes are flushed through the network.  
The logger works under the assumption that it is called to safeguard
updates on the database when it has an exclusive lock on
the latest version. This lock should be guaranteed by the calling
transaction manager first.

Finding the updates applied to a BAT is relatively easy, because each
BAT contains a delta structure. On commit these changes are
written to the log file and the delta management is reset. Since each 
commit is written to the same log file, the beginning and end are 
marked by a log identifier.

A server restart should only (re)process blocks which are completely 
written to disk. A log replay therefore ends in a commit or abort on
the changed bats. Once all logs have been read, the changes to
the bats are made persistent, i.e. a bbp sub-commit is done.

@{
@* Implementation Code 
@h
#ifndef _LOGGER_H_
#define _LOGGER_H_

#define LOG_OK 0
#define LOG_ERR (-1)

#include <gdk.h>
#include <stream.h>

typedef struct logaction {
	int type;   /* type of change */
	int nr;
	int ht;     /* vid(-1),void etc */
	int tt;
	lng id;
	char *name; /* optional */
	BAT  *b;    /* temporary bat with changes */
} logaction;

/* during the recover process a number of transactions could be active */
typedef struct trans {
	int tid;  /* transaction id */
	int sz;	  /* sz of the changes array */
	int nr;	  /* nr of changes */

	logaction *changes; 

	struct trans *tr;
} trans;

typedef struct logger {
	int debug;
	int changes;
	int version;
	lng id;
	int tid;
	char *fn;
	char *dir;
	stream *log;
	/* Store log_bids (int) to circumvent trouble with reference counting */
	BAT *catalog;	/* int, str */
	BAT *seqs; 	/* int, lng */
	BAT *snapshots;	/* int, int the bid and tid of snapshot bat */
} logger;

#define BATSIZE 0

typedef int log_bid;

@+ Sequence numbers
The logger also keeps sequence numbers. A sequence needs to store each
requested (block) of sequence numbers. This is done using the log_sequence
function. The logger_sequence function can be used to return the
last logged sequence number. Sequences identifiers should be unique, and
2 are already used. The first LOG_SID is used internally for the log files
sequence. The second OBJ_SID is for frontend objects, for example the sql
objects have a global sequence counter such that each table, trigger, sequence
etc. has a unique number. 
@h
/* the sequence identifier for the sequence of log files */
#define LOG_SID	0
/* the sequence identifier for frontend objects */
#define OBJ_SID	1

gdk_export logger *logger_create(int debug, char *fn, char *logdir, char *dbname, int version);
gdk_export void logger_destroy(logger *lg);
gdk_export int logger_exit(logger *lg);
gdk_export int logger_restart(logger *lg);
gdk_export int logger_cleanup(logger *lg);
gdk_export int logger_changes(logger *lg);
gdk_export int logger_sequence(logger *lg, int seq, lng *id);

/* todo pass the transaction id */
gdk_export int log_bat(logger *lg, BAT *b, char *n);
gdk_export int log_bat_clear(logger *lg, char *n);
gdk_export int log_bat_persists(logger *lg, BAT *b, char *n);
gdk_export int log_bat_transient(logger *lg, char *n);
gdk_export int log_delta(logger *lg, BAT *b, char *n);

gdk_export int log_tstart(logger *lg);  /* TODO return transaction id */
gdk_export int log_tend(logger *lg);
gdk_export int log_abort(logger *lg);

gdk_export int log_sequence(logger *lg, int seq, lng id);

gdk_export log_bid logger_add_bat(logger *lg, BAT *b, char *name);
gdk_export void logger_del_bat(logger *lg, log_bid bid);
gdk_export log_bid logger_find_bat(logger *lg, char *name);

#endif /*_LOGGER_H_*/
@c
#include "monetdb_config.h"
#include "gdk_logger.h"
#include <string.h>


@-
The log record encoding is geared at reduced storage space, but
at the expense of readability. A user can not easily inspect the
log a posteriori to check what has happened.

@c
#define LOG_START	1
#define LOG_END		2
#define LOG_INSERT	3
#define LOG_DELETE	4
#define LOG_UPDATE	5
#define LOG_CREATE	6
#define LOG_DESTROY	7
#define LOG_USE		8
#define LOG_CLEAR	9
#define LOG_SEQ		10

typedef struct logformat_t {
	char flag;
	int tid;
	int nr;
} logformat;

#define LOGFILE "log"

static int bm_commit(logger *lg);
static void tr_grow(trans *tr);

static void
logbat_destroy(BAT *b)
{
	if (b)
		BBPunfix(b->batCacheid);
}

static BAT *
logbat_new(int ht, int tt, size_t size)
{
	BAT *nb = BATnew(ht, tt, size);

	if (ht == TYPE_void)
		BATseqbase(nb, 0);
	nb->batDirty |= 2;
	return nb;
}

static int
log_read_format(logger *l, logformat *data)
{
	int res = 1;

	if (stream_read(l->log, &data->flag, 1, 1) != 1)
		return 0;
	res = stream_readInt(l->log, &data->nr);
	if (res)
		res = stream_readInt(l->log, &data->tid);
	return res;
}

static int
log_write_format(logger *l, logformat *data)
{
	if (stream_write(l->log, &data->flag, 1, 1) != 1 || !stream_writeInt(l->log, data->nr) || !stream_writeInt(l->log, data->tid))
		return LOG_ERR;
	return LOG_OK;
}

static char *
log_read_string(logger *l)
{
	int len;
	ssize_t nr;
	char *buf;

	if (!stream_readInt(l->log, &len))
		return NULL;
	if (len == 0)
		return NULL;
	buf = (char*)GDKmalloc(len);

	if ((nr = stream_read(l->log, buf, 1, len)) != (ssize_t) len) {
		buf[len-1] = 0;
		printf("!ERROR: couldn't read name (%s) " SSZFMT "\n", buf, nr);
		GDKfree(buf);
		return NULL;
	}
	buf[len-1] = 0;
	return buf;
}

static int
log_write_string(logger *l, char *n)
{
	int len = (int) strlen(n) + 1;	/* log including EOS */

	assert(len > 1);
	if (!stream_writeInt(l->log, len) ||
	    stream_write(l->log, n, 1, len) != (ssize_t) len)
		return LOG_ERR;
	return LOG_OK;
}

static void
log_read_clear(logger *lg, trans *tr, char *name)
{
	if (lg->debug & 1)
		fprintf(stderr, "logger found log_read_clear %s\n", name);

	tr_grow(tr);
	tr->changes[tr->nr].type = LOG_CLEAR;
	tr->changes[tr->nr].name = GDKstrdup(name);
	tr->nr++;
}

static void
la_bat_clear(logger *lg, logaction *la)
{
	log_bid bid = logger_find_bat(lg, la->name);
	BAT *b;

	/* do we need to skip these old updates */
	if (BATcount(lg->snapshots)) {
		BUN p = BUNfnd(lg->snapshots, &bid);
		if (p) {
			int tid = *(int*)BUNtloc(lg->snapshots, p);

			if (lg->tid <= tid) 
				return ;
		}
	}

 	b = BATdescriptor(bid);
	if (b) {
		BATclear(b);
		logbat_destroy(b);
	}
}

static int
log_read_seq(logger *lg, logformat *l)
{
	int seq = l->nr;
	lng id;

	if (!stream_readLng(lg->log, &id))
		return LOG_ERR;

	if (BUNfnd(lg->seqs, &seq)) {
		BUNdelHead(lg->seqs, &seq, FALSE);
	}
	BUNins(lg->seqs, &seq, &id, FALSE);
	return LOG_OK;
}

static int
log_read_updates(logger *lg, trans *tr, logformat *l, char *name)
{
	log_bid bid = logger_find_bat(lg, name);
	BAT *b = BATdescriptor(bid);
	int res = LOG_OK;
	int ht = -1, tt = -1, hseq = 0, tseq = 0;

	if (lg->debug & 1)
		fprintf(stderr, "logger found log_read_updates %s %s %d\n", name, l->flag == LOG_INSERT ? "insert" : l->flag == LOG_DELETE ? "delete" : "update", l->nr);

	if (b) { 
		ht = b->htype;
		if (ht == TYPE_void && b->hseqbase != oid_nil)
			hseq = 1;
		tt = b->ttype;
		if (tt == TYPE_void && b->tseqbase != oid_nil)
			tseq = 1;
	} else { /* search trans action for create statement */
		int i;
		for (i=0; i<tr->nr; i++) {
			if (tr->changes[i].type == LOG_CREATE &&
			    strcmp(tr->changes[i].name, name) == 0) {
				ht = tr->changes[i].ht;
				if (ht < 0) {
					hseq = 1;
					ht = TYPE_void;
				}
				tt = tr->changes[i].tt;
				if (tt < 0) {
					tseq = 1;
					tt = TYPE_void;
				}
				break;
			}
		}
	}
	if (ht>=0 && tt>=0) {
		BAT *r;
		void *(*rt) (ptr, stream *, size_t) = BATatoms[tt].atomRead;
		void *tv = ATOMnil(tt);

		r = BATnew(ht, tt, l->nr);

		if (hseq)
			BATseqbase(r, 0);
		if (tseq)
			BATseqbase(BATmirror(r), 0);

		if (ht == TYPE_void && l->flag == LOG_INSERT) { 
			for (; l->nr > 0; l->nr--) {
				void *t = rt(tv, lg->log, 1);

				if (!t) {
					res = LOG_ERR;
					break;
				}
				if (l->flag == LOG_INSERT)
					BUNappend(r, t, TRUE);
				if (t != tv)
					GDKfree(t);
			}
		} else {
			void *(*rh) (ptr, stream *, size_t) = ht == TYPE_void ? BATatoms[TYPE_oid].atomRead : BATatoms[ht].atomRead;
			void *hv = ATOMnil(ht);

			for (; l->nr > 0; l->nr--) {
				void *h = rh(hv, lg->log, 1);
				void *t = rt(tv, lg->log, 1);

				if (!h || !t) {
					res = LOG_ERR;
					break;
				}
				BUNins(r, h, t, TRUE);
				if (h != hv)
					GDKfree(h);
				if (t != tv)
					GDKfree(t);
			}
			GDKfree(hv);
		}
		GDKfree(tv);
		logbat_destroy(b);
		
		tr_grow(tr);
		tr->changes[tr->nr].type = l->flag;
		tr->changes[tr->nr].nr = l->nr;
		tr->changes[tr->nr].ht = ht;
		tr->changes[tr->nr].tt = tt;
		tr->changes[tr->nr].name = GDKstrdup(name);
		tr->changes[tr->nr].b = r;
		tr->nr++;
	} else { /* bat missing ERROR or ignore ? currently error. */
		res = LOG_ERR;
	}
	return res;
}

static void
la_bat_updates(logger *lg, logaction *la)
{
	log_bid bid = logger_find_bat(lg, la->name);
	BAT *b;
        if (bid == 0) return; /* ignore bats no longer in the catalog */

	/* do we need to skip these old updates */
	if (BATcount(lg->snapshots)) {
		BUN p = BUNfnd(lg->snapshots, &bid);
		if (p) {
			int tid = *(int*)BUNtloc(lg->snapshots, p);

			if (lg->tid <= tid) 
				return ;
		}
	}

	b = BATdescriptor(bid);
	assert(b);
	if (b) {
		if (b->htype == TYPE_void && la->type == LOG_INSERT) { 
			BATappend(b, la->b, TRUE);
		} else {
			if (la->type == LOG_INSERT)
				BATins(b, la->b, TRUE);
			else if (la->type == LOG_DELETE)
				BATdel(b, la->b, TRUE);
			else if (la->type == LOG_UPDATE) { 
				BUN p, q;
				BATloop(la->b, p, q) {
					ptr h = BUNhead(la->b, p);
					ptr t = BUNtail(la->b, p);
					if (BUNfnd(b, h) == NULL) {
						/* if value doesn't exist, insert it
					   	   if b void headed, maintain that by inserting nils */
						if (b->htype == TYPE_void) {
							if (b->batCount == 0 && * (oid *) h != oid_nil)
								b->hseqbase = * (oid *) h;
							if (b->hseqbase != oid_nil && * (oid *) h != oid_nil) {
								void *tv = ATOMnilptr(b->ttype);
								while (b->hseqbase + b->batCount < * (oid *) h)
									BUNappend(b, tv, TRUE);
							}
							BUNappend(b, t, TRUE);
						} else {
							BUNins(b, h, t, TRUE);
						}
					} else {
						BUNreplace(b, h, t, TRUE);
					}
				}
			}
		}
		logbat_destroy(b);
	}
}

static void
log_read_destroy(logger *lg, trans *tr, char *name)
{
	(void)lg;
	tr_grow(tr);
	tr->changes[tr->nr].type = LOG_DESTROY;
	tr->changes[tr->nr].name = GDKstrdup(name);
	tr->nr++;
}

static void
la_bat_destroy(logger *lg, logaction *la)
{
	log_bid bid = logger_find_bat(lg, la->name);

	if (bid)
		logger_del_bat(lg, bid);
}

static int
log_read_create(logger *lg, trans *tr, char *name)
{
	char *buf = log_read_string(lg);

	if (lg->debug & 1)
		fprintf(stderr, "log_read_create %s\n", name);

	if (!buf) {
		return LOG_ERR;
	} else {
		int ht, tt;
		char *ha = buf, *ta = strchr(buf, ',');

		if (!ta) 
			return LOG_ERR;
		*ta = 0;
		ta++;		/* skip over , */