transaction.c

samba最新软件

 /* 
   Unix SMB/CIFS implementation.

   trivial database library

   Copyright (C) Andrew Tridgell              2005

     ** NOTE! The following LGPL license applies to the tdb
     ** library. This does NOT imply that all of Samba is released
     ** under the LGPL
   
   This library is free software; you can redistribute it and/or
   modify it under the terms of the GNU Lesser General Public
   License as published by the Free Software Foundation; either
   version 3 of the License, or (at your option) any later version.

   This library is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
   Lesser General Public License for more details.

   You should have received a copy of the GNU Lesser General Public
   License along with this library; if not, see <http://www.gnu.org/licenses/>.
*/

#include "tdb_private.h"

/*
  transaction design:

  - only allow a single transaction at a time per database. This makes
    using the transaction API simpler, as otherwise the caller would
    have to cope with temporary failures in transactions that conflict
    with other current transactions

  - keep the transaction recovery information in the same file as the
    database, using a special 'transaction recovery' record pointed at
    by the header. This removes the need for extra journal files as
    used by some other databases

  - dynamically allocated the transaction recover record, re-using it
    for subsequent transactions. If a larger record is needed then
    tdb_free() the old record to place it on the normal tdb freelist
    before allocating the new record

  - during transactions, keep a linked list of writes all that have
    been performed by intercepting all tdb_write() calls. The hooked
    transaction versions of tdb_read() and tdb_write() check this
    linked list and try to use the elements of the list in preference
    to the real database.

  - don't allow any locks to be held when a transaction starts,
    otherwise we can end up with deadlock (plus lack of lock nesting
    in posix locks would mean the lock is lost)

  - if the caller gains a lock during the transaction but doesn't
    release it then fail the commit

  - allow for nested calls to tdb_transaction_start(), re-using the
    existing transaction record. If the inner transaction is cancelled
    then a subsequent commit will fail
 
  - keep a mirrored copy of the tdb hash chain heads to allow for the
    fast hash heads scan on traverse, updating the mirrored copy in
    the transaction version of tdb_write

  - allow callers to mix transaction and non-transaction use of tdb,
    although once a transaction is started then an exclusive lock is
    gained until the transaction is committed or cancelled

  - the commit stategy involves first saving away all modified data
    into a linearised buffer in the transaction recovery area, then
    marking the transaction recovery area with a magic value to
    indicate a valid recovery record. In total 4 fsync/msync calls are
    needed per commit to prevent race conditions. It might be possible
    to reduce this to 3 or even 2 with some more work.

  - check for a valid recovery record on open of the tdb, while the
    global lock is held. Automatically recover from the transaction
    recovery area if needed, then continue with the open as
    usual. This allows for smooth crash recovery with no administrator
    intervention.

  - if TDB_NOSYNC is passed to flags in tdb_open then transactions are
    still available, but no transaction recovery area is used and no
    fsync/msync calls are made.

*/


/*
  hold the context of any current transaction
*/
struct tdb_transaction {
	/* we keep a mirrored copy of the tdb hash heads here so
	   tdb_next_hash_chain() can operate efficiently */
	uint32_t *hash_heads;

	/* the original io methods - used to do IOs to the real db */
	const struct tdb_methods *io_methods;

	/* the list of transaction blocks. When a block is first
	   written to, it gets created in this list */
	uint8_t **blocks;
	uint32_t num_blocks;
	uint32_t block_size;      /* bytes in each block */
	uint32_t last_block_size; /* number of valid bytes in the last block */

	/* non-zero when an internal transaction error has
	   occurred. All write operations will then fail until the
	   transaction is ended */
	int transaction_error;

	/* when inside a transaction we need to keep track of any
	   nested tdb_transaction_start() calls, as these are allowed,
	   but don't create a new transaction */
	int nesting;

	/* old file size before transaction */
	tdb_len_t old_map_size;
};


/*
  read while in a transaction. We need to check first if the data is in our list
  of transaction elements, then if not do a real read
*/
static int transaction_read(struct tdb_context *tdb, tdb_off_t off, void *buf, 
			    tdb_len_t len, int cv)
{
	uint32_t blk;

	/* break it down into block sized ops */
	while (len + (off % tdb->transaction->block_size) > tdb->transaction->block_size) {
		tdb_len_t len2 = tdb->transaction->block_size - (off % tdb->transaction->block_size);
		if (transaction_read(tdb, off, buf, len2, cv) != 0) {
			return -1;
		}
		len -= len2;
		off += len2;
		buf = (void *)(len2 + (char *)buf);
	}

	if (len == 0) {
		return 0;
	}

	blk = off / tdb->transaction->block_size;

	/* see if we have it in the block list */
	if (tdb->transaction->num_blocks <= blk ||
	    tdb->transaction->blocks[blk] == NULL) {
		/* nope, do a real read */
		if (tdb->transaction->io_methods->tdb_read(tdb, off, buf, len, cv) != 0) {
			goto fail;
		}
		return 0;
	}

	/* it is in the block list. Now check for the last block */
	if (blk == tdb->transaction->num_blocks-1) {
		if (len > tdb->transaction->last_block_size) {
			goto fail;
		}
	}
	
	/* now copy it out of this block */
	memcpy(buf, tdb->transaction->blocks[blk] + (off % tdb->transaction->block_size), len);
	if (cv) {
		tdb_convert(buf, len);
	}
	return 0;

fail:
	TDB_LOG((tdb, TDB_DEBUG_FATAL, "transaction_read: failed at off=%d len=%d\n", off, len));
	tdb->ecode = TDB_ERR_IO;
	tdb->transaction->transaction_error = 1;
	return -1;
}


/*
  write while in a transaction
*/
static int transaction_write(struct tdb_context *tdb, tdb_off_t off, 
			     const void *buf, tdb_len_t len)
{
	uint32_t blk;

	/* if the write is to a hash head, then update the transaction
	   hash heads */
	if (len == sizeof(tdb_off_t) && off >= FREELIST_TOP &&
	    off < FREELIST_TOP+TDB_HASHTABLE_SIZE(tdb)) {
		uint32_t chain = (off-FREELIST_TOP) / sizeof(tdb_off_t);
		memcpy(&tdb->transaction->hash_heads[chain], buf, len);
	}

	/* break it up into block sized chunks */
	while (len + (off % tdb->transaction->block_size) > tdb->transaction->block_size) {
		tdb_len_t len2 = tdb->transaction->block_size - (off % tdb->transaction->block_size);
		if (transaction_write(tdb, off, buf, len2) != 0) {
			return -1;
		}
		len -= len2;
		off += len2;
		if (buf != NULL) {
			buf = (const void *)(len2 + (const char *)buf);
		}
	}

	if (len == 0) {
		return 0;
	}

	blk = off / tdb->transaction->block_size;
	off = off % tdb->transaction->block_size;

	if (tdb->transaction->num_blocks <= blk) {
		uint8_t **new_blocks;
		/* expand the blocks array */
		if (tdb->transaction->blocks == NULL) {
			new_blocks = (uint8_t **)malloc(
				(blk+1)*sizeof(uint8_t *));
		} else {
			new_blocks = (uint8_t **)realloc(
				tdb->transaction->blocks,
				(blk+1)*sizeof(uint8_t *));
		}
		if (new_blocks == NULL) {
			tdb->ecode = TDB_ERR_OOM;
			goto fail;
		}
		memset(&new_blocks[tdb->transaction->num_blocks], 0, 
		       (1+(blk - tdb->transaction->num_blocks))*sizeof(uint8_t *));
		tdb->transaction->blocks = new_blocks;
		tdb->transaction->num_blocks = blk+1;
		tdb->transaction->last_block_size = 0;
	}

	/* allocate and fill a block? */
	if (tdb->transaction->blocks[blk] == NULL) {
		tdb->transaction->blocks[blk] = (uint8_t *)calloc(tdb->transaction->block_size, 1);
		if (tdb->transaction->blocks[blk] == NULL) {
			tdb->ecode = TDB_ERR_OOM;
			tdb->transaction->transaction_error = 1;
			return -1;			
		}
		if (tdb->transaction->old_map_size > blk * tdb->transaction->block_size) {
			tdb_len_t len2 = tdb->transaction->block_size;
			if (len2 + (blk * tdb->transaction->block_size) > tdb->transaction->old_map_size) {
				len2 = tdb->transaction->old_map_size - (blk * tdb->transaction->block_size);
			}
			if (tdb->transaction->io_methods->tdb_read(tdb, blk * tdb->transaction->block_size, 
								   tdb->transaction->blocks[blk], 
								   len2, 0) != 0) {
				SAFE_FREE(tdb->transaction->blocks[blk]);				
				tdb->ecode = TDB_ERR_IO;
				goto fail;
			}
			if (blk == tdb->transaction->num_blocks-1) {
				tdb->transaction->last_block_size = len2;
			}			
		}
	}
	
	/* overwrite part of an existing block */
	if (buf == NULL) {
		memset(tdb->transaction->blocks[blk] + off, 0, len);
	} else {
		memcpy(tdb->transaction->blocks[blk] + off, buf, len);
	}
	if (blk == tdb->transaction->num_blocks-1) {
		if (len + off > tdb->transaction->last_block_size) {
			tdb->transaction->last_block_size = len + off;
		}
	}

	return 0;

fail:
	TDB_LOG((tdb, TDB_DEBUG_FATAL, "transaction_write: failed at off=%d len=%d\n", 
		 (blk*tdb->transaction->block_size) + off, len));
	tdb->transaction->transaction_error = 1;
	return -1;
}


/*
  write while in a transaction - this varient never expands the transaction blocks, it only
  updates existing blocks. This means it cannot change the recovery size
*/
static int transaction_write_existing(struct tdb_context *tdb, tdb_off_t off, 
				      const void *buf, tdb_len_t len)
{
	uint32_t blk;

	/* break it up into block sized chunks */
	while (len + (off % tdb->transaction->block_size) > tdb->transaction->block_size) {
		tdb_len_t len2 = tdb->transaction->block_size - (off % tdb->transaction->block_size);
		if (transaction_write_existing(tdb, off, buf, len2) != 0) {
			return -1;
		}
		len -= len2;
		off += len2;
		if (buf != NULL) {
			buf = (const void *)(len2 + (const char *)buf);
		}
	}

	if (len == 0) {
		return 0;
	}

	blk = off / tdb->transaction->block_size;
	off = off % tdb->transaction->block_size;

	if (tdb->transaction->num_blocks <= blk ||
	    tdb->transaction->blocks[blk] == NULL) {
		return 0;
	}

	if (blk == tdb->transaction->num_blocks-1 &&
	    off + len > tdb->transaction->last_block_size) {
		if (off >= tdb->transaction->last_block_size) {
			return 0;
		}
		len = tdb->transaction->last_block_size - off;
	}

	/* overwrite part of an existing block */
	memcpy(tdb->transaction->blocks[blk] + off, buf, len);

	return 0;
}


/*
  accelerated hash chain head search, using the cached hash heads
*/
static void transaction_next_hash_chain(struct tdb_context *tdb, uint32_t *chain)
{
	uint32_t h = *chain;
	for (;h < tdb->header.hash_size;h++) {
		/* the +1 takes account of the freelist */
		if (0 != tdb->transaction->hash_heads[h+1]) {
			break;
		}
	}
	(*chain) = h;
}

/*
  out of bounds check during a transaction
*/
static int transaction_oob(struct tdb_context *tdb, tdb_off_t len, int probe)
{
	if (len <= tdb->map_size) {
		return 0;
	}
	return TDB_ERRCODE(TDB_ERR_IO, -1);
}

/*
  transaction version of tdb_expand().
*/
static int transaction_expand_file(struct tdb_context *tdb, tdb_off_t size, 
				   tdb_off_t addition)
{
	/* add a write to the transaction elements, so subsequent
	   reads see the zero data */
	if (transaction_write(tdb, size, NULL, addition) != 0) {
		return -1;
	}