X-Git-Url: http://git.ozlabs.org/?p=ccan;a=blobdiff_plain;f=ccan%2Ftdb2%2Ftransaction.c;h=805ea0948aad0e0ddfa11289aadc26334cfc8a44;hp=c29c46510cffea170ea9d6f8079433902fb5f29d;hb=71d8cfb69339f3587ded2693cbab845c916c5c7e;hpb=5eaf46e9a0d38c371b50d5dd3a433fc721c1c4dc diff --git a/ccan/tdb2/transaction.c b/ccan/tdb2/transaction.c index c29c4651..805ea094 100644 --- a/ccan/tdb2/transaction.c +++ b/ccan/tdb2/transaction.c @@ -53,13 +53,13 @@ - 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) + 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 + existing transaction record. If the inner transaction is canceled then a subsequent commit will fail - keep a mirrored copy of the tdb hash chain heads to allow for the @@ -68,7 +68,7 @@ - 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 + gained until the transaction is committed or canceled - the commit stategy involves first saving away all modified data into a linearised buffer in the transaction recovery area, then @@ -88,7 +88,6 @@ fsync/msync calls are made. */ - /* hold the context of any current transaction */ @@ -110,7 +109,7 @@ struct tdb_transaction { /* 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; + unsigned int nesting; /* set when a prepare has already occurred */ bool prepared; @@ -120,21 +119,25 @@ struct tdb_transaction { tdb_len_t old_map_size; }; +/* This doesn't really need to be pagesize, but we use it for similar reasons. */ +#define PAGESIZE 65536 /* 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) +static enum TDB_ERROR transaction_read(struct tdb_context *tdb, tdb_off_t off, + void *buf, tdb_len_t len) { size_t blk; + enum TDB_ERROR ecode; /* break it down into block sized ops */ - while (len + (off % getpagesize()) > getpagesize()) { - tdb_len_t len2 = getpagesize() - (off % getpagesize()); - if (transaction_read(tdb, off, buf, len2) != 0) { - return -1; + while (len + (off % PAGESIZE) > PAGESIZE) { + tdb_len_t len2 = PAGESIZE - (off % PAGESIZE); + ecode = transaction_read(tdb, off, buf, len2); + if (ecode != TDB_SUCCESS) { + return ecode; } len -= len2; off += len2; @@ -142,16 +145,17 @@ static int transaction_read(struct tdb_context *tdb, tdb_off_t off, void *buf, } if (len == 0) { - return 0; + return TDB_SUCCESS; } - blk = off / getpagesize(); + blk = off / PAGESIZE; /* 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->read(tdb, off, buf, len) != 0) { + ecode = tdb->transaction->io_methods->tread(tdb, off, buf, len); + if (ecode != TDB_SUCCESS) { goto fail; } return 0; @@ -160,43 +164,45 @@ static int transaction_read(struct tdb_context *tdb, tdb_off_t off, void *buf, /* 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) { + ecode = TDB_ERR_IO; goto fail; } } /* now copy it out of this block */ - memcpy(buf, tdb->transaction->blocks[blk] + (off % getpagesize()), len); - return 0; + memcpy(buf, tdb->transaction->blocks[blk] + (off % PAGESIZE), len); + return TDB_SUCCESS; fail: - tdb_logerr(tdb, TDB_ERR_IO, TDB_DEBUG_FATAL, - "transaction_read: failed at off=%zu len=%zu", - (size_t)off, (size_t)len); tdb->transaction->transaction_error = 1; - return -1; + return tdb_logerr(tdb, ecode, TDB_LOG_ERROR, + "transaction_read: failed at off=%zu len=%zu", + (size_t)off, (size_t)len); } /* write while in a transaction */ -static int transaction_write(struct tdb_context *tdb, tdb_off_t off, - const void *buf, tdb_len_t len) +static enum TDB_ERROR transaction_write(struct tdb_context *tdb, tdb_off_t off, + const void *buf, tdb_len_t len) { size_t blk; + enum TDB_ERROR ecode; /* Only a commit is allowed on a prepared transaction */ if (tdb->transaction->prepared) { - tdb_logerr(tdb, TDB_ERR_EINVAL, TDB_DEBUG_FATAL, - "transaction_write: transaction already prepared," - " write not allowed"); + ecode = tdb_logerr(tdb, TDB_ERR_EINVAL, TDB_LOG_ERROR, + "transaction_write: transaction already" + " prepared, write not allowed"); goto fail; } /* break it up into block sized chunks */ - while (len + (off % getpagesize()) > getpagesize()) { - tdb_len_t len2 = getpagesize() - (off % getpagesize()); - if (transaction_write(tdb, off, buf, len2) != 0) { + while (len + (off % PAGESIZE) > PAGESIZE) { + tdb_len_t len2 = PAGESIZE - (off % PAGESIZE); + ecode = transaction_write(tdb, off, buf, len2); + if (ecode != TDB_SUCCESS) { return -1; } len -= len2; @@ -207,11 +213,11 @@ static int transaction_write(struct tdb_context *tdb, tdb_off_t off, } if (len == 0) { - return 0; + return TDB_SUCCESS; } - blk = off / getpagesize(); - off = off % getpagesize(); + blk = off / PAGESIZE; + off = off % PAGESIZE; if (tdb->transaction->num_blocks <= blk) { uint8_t **new_blocks; @@ -225,8 +231,9 @@ static int transaction_write(struct tdb_context *tdb, tdb_off_t off, (blk+1)*sizeof(uint8_t *)); } if (new_blocks == NULL) { - tdb_logerr(tdb, TDB_ERR_OOM, TDB_DEBUG_FATAL, - "transaction_write: failed to allocate"); + ecode = tdb_logerr(tdb, TDB_ERR_OOM, TDB_LOG_ERROR, + "transaction_write:" + " failed to allocate"); goto fail; } memset(&new_blocks[tdb->transaction->num_blocks], 0, @@ -238,24 +245,29 @@ static int transaction_write(struct tdb_context *tdb, tdb_off_t off, /* allocate and fill a block? */ if (tdb->transaction->blocks[blk] == NULL) { - tdb->transaction->blocks[blk] = (uint8_t *)calloc(getpagesize(), 1); + tdb->transaction->blocks[blk] = (uint8_t *)calloc(PAGESIZE, 1); if (tdb->transaction->blocks[blk] == NULL) { - tdb_logerr(tdb, TDB_ERR_OOM, TDB_DEBUG_FATAL, - "transaction_write: failed to allocate"); + ecode = tdb_logerr(tdb, TDB_ERR_OOM, TDB_LOG_ERROR, + "transaction_write:" + " failed to allocate"); goto fail; } - if (tdb->transaction->old_map_size > blk * getpagesize()) { - tdb_len_t len2 = getpagesize(); - if (len2 + (blk * getpagesize()) > tdb->transaction->old_map_size) { - len2 = tdb->transaction->old_map_size - (blk * getpagesize()); + if (tdb->transaction->old_map_size > blk * PAGESIZE) { + tdb_len_t len2 = PAGESIZE; + if (len2 + (blk * PAGESIZE) > tdb->transaction->old_map_size) { + len2 = tdb->transaction->old_map_size - (blk * PAGESIZE); } - if (tdb->transaction->io_methods->read(tdb, blk * getpagesize(), - tdb->transaction->blocks[blk], - len2) != 0) { - tdb_logerr(tdb, TDB_ERR_OOM, TDB_DEBUG_FATAL, - "transaction_write: failed to" - " read old block: %s", - strerror(errno)); + ecode = tdb->transaction->io_methods->tread(tdb, + blk * PAGESIZE, + tdb->transaction->blocks[blk], + len2); + if (ecode != TDB_SUCCESS) { + ecode = tdb_logerr(tdb, ecode, + TDB_LOG_ERROR, + "transaction_write:" + " failed to" + " read old block: %s", + strerror(errno)); SAFE_FREE(tdb->transaction->blocks[blk]); goto fail; } @@ -277,16 +289,16 @@ static int transaction_write(struct tdb_context *tdb, tdb_off_t off, } } - return 0; + return TDB_SUCCESS; fail: tdb->transaction->transaction_error = 1; - return -1; + return ecode; } /* - write while in a transaction - this varient never expands the transaction blocks, it only + write while in a transaction - this variant never expands the transaction blocks, it only updates existing blocks. This means it cannot change the recovery size */ static void transaction_write_existing(struct tdb_context *tdb, tdb_off_t off, @@ -295,8 +307,8 @@ static void transaction_write_existing(struct tdb_context *tdb, tdb_off_t off, size_t blk; /* break it up into block sized chunks */ - while (len + (off % getpagesize()) > getpagesize()) { - tdb_len_t len2 = getpagesize() - (off % getpagesize()); + while (len + (off % PAGESIZE) > PAGESIZE) { + tdb_len_t len2 = PAGESIZE - (off % PAGESIZE); transaction_write_existing(tdb, off, buf, len2); len -= len2; off += len2; @@ -309,8 +321,8 @@ static void transaction_write_existing(struct tdb_context *tdb, tdb_off_t off, return; } - blk = off / getpagesize(); - off = off % getpagesize(); + blk = off / PAGESIZE; + off = off % PAGESIZE; if (tdb->transaction->num_blocks <= blk || tdb->transaction->blocks[blk] == NULL) { @@ -333,42 +345,45 @@ static void transaction_write_existing(struct tdb_context *tdb, tdb_off_t off, /* out of bounds check during a transaction */ -static int transaction_oob(struct tdb_context *tdb, tdb_off_t len, bool probe) +static enum TDB_ERROR transaction_oob(struct tdb_context *tdb, tdb_off_t len, + bool probe) { - if (len <= tdb->map_size) { - return 0; + if (len <= tdb->file->map_size) { + return TDB_SUCCESS; } - tdb->ecode = TDB_ERR_IO; if (!probe) { - tdb_logerr(tdb, TDB_ERR_IO, TDB_DEBUG_FATAL, + tdb_logerr(tdb, TDB_ERR_IO, TDB_LOG_ERROR, "tdb_oob len %lld beyond transaction size %lld", (long long)len, - (long long)tdb->map_size); + (long long)tdb->file->map_size); } - return -1; + return TDB_ERR_IO; } /* transaction version of tdb_expand(). */ -static int transaction_expand_file(struct tdb_context *tdb, tdb_off_t addition) +static enum TDB_ERROR transaction_expand_file(struct tdb_context *tdb, + tdb_off_t addition) { + enum TDB_ERROR ecode; + /* add a write to the transaction elements, so subsequent reads see the zero data */ - if (transaction_write(tdb, tdb->map_size, NULL, addition) != 0) { - return -1; + ecode = transaction_write(tdb, tdb->file->map_size, NULL, addition); + if (ecode == TDB_SUCCESS) { + tdb->file->map_size += addition; } - tdb->map_size += addition; - return 0; + return ecode; } static void *transaction_direct(struct tdb_context *tdb, tdb_off_t off, size_t len, bool write_mode) { - size_t blk = off / getpagesize(), end_blk; + size_t blk = off / PAGESIZE, end_blk; /* This is wrong for zero-length blocks, but will fail gracefully */ - end_blk = (off + len - 1) / getpagesize(); + end_blk = (off + len - 1) / PAGESIZE; /* Can only do direct if in single block and we've already copied. */ if (write_mode) { @@ -378,17 +393,17 @@ static void *transaction_direct(struct tdb_context *tdb, tdb_off_t off, return NULL; if (tdb->transaction->blocks[blk] == NULL) return NULL; - return tdb->transaction->blocks[blk] + off % getpagesize(); + return tdb->transaction->blocks[blk] + off % PAGESIZE; } /* Single which we have copied? */ if (blk == end_blk && blk < tdb->transaction->num_blocks && tdb->transaction->blocks[blk]) - return tdb->transaction->blocks[blk] + off % getpagesize(); + return tdb->transaction->blocks[blk] + off % PAGESIZE; /* Otherwise must be all not copied. */ - while (blk < end_blk) { + while (blk <= end_blk) { if (blk >= tdb->transaction->num_blocks) break; if (tdb->transaction->blocks[blk]) @@ -409,40 +424,40 @@ static const struct tdb_methods transaction_methods = { /* sync to disk */ -static int transaction_sync(struct tdb_context *tdb, tdb_off_t offset, tdb_len_t length) +static enum TDB_ERROR transaction_sync(struct tdb_context *tdb, + tdb_off_t offset, tdb_len_t length) { if (tdb->flags & TDB_NOSYNC) { - return 0; + return TDB_SUCCESS; } - if (fsync(tdb->fd) != 0) { - tdb_logerr(tdb, TDB_ERR_IO, TDB_DEBUG_FATAL, - "tdb_transaction: fsync failed: %s", - strerror(errno)); - return -1; + if (fsync(tdb->file->fd) != 0) { + return tdb_logerr(tdb, TDB_ERR_IO, TDB_LOG_ERROR, + "tdb_transaction: fsync failed: %s", + strerror(errno)); } #ifdef MS_SYNC - if (tdb->map_ptr) { - tdb_off_t moffset = offset & ~(getpagesize()-1); - if (msync(moffset + (char *)tdb->map_ptr, + if (tdb->file->map_ptr) { + tdb_off_t moffset = offset & ~(PAGESIZE-1); + if (msync(moffset + (char *)tdb->file->map_ptr, length + (offset - moffset), MS_SYNC) != 0) { - tdb_logerr(tdb, TDB_ERR_IO, TDB_DEBUG_FATAL, - "tdb_transaction: msync failed: %s", - strerror(errno)); - return -1; + return tdb_logerr(tdb, TDB_ERR_IO, TDB_LOG_ERROR, + "tdb_transaction: msync failed: %s", + strerror(errno)); } } #endif - return 0; + return TDB_SUCCESS; } static void _tdb_transaction_cancel(struct tdb_context *tdb) { int i; + enum TDB_ERROR ecode; if (tdb->transaction == NULL) { - tdb_logerr(tdb, TDB_ERR_EINVAL, TDB_DEBUG_ERROR, + tdb_logerr(tdb, TDB_ERR_EINVAL, TDB_LOG_USE_ERROR, "tdb_transaction_cancel: no transaction"); return; } @@ -453,7 +468,7 @@ static void _tdb_transaction_cancel(struct tdb_context *tdb) return; } - tdb->map_size = tdb->transaction->old_map_size; + tdb->file->map_size = tdb->transaction->old_map_size; /* free all the transaction blocks */ for (i=0;itransaction->num_blocks;i++) { @@ -468,18 +483,21 @@ static void _tdb_transaction_cancel(struct tdb_context *tdb) uint64_t invalid = TDB_RECOVERY_INVALID_MAGIC; /* remove the recovery marker */ - if (methods->write(tdb, tdb->transaction->magic_offset, - &invalid, sizeof(invalid)) == -1 || - transaction_sync(tdb, tdb->transaction->magic_offset, - sizeof(invalid)) == -1) { - tdb_logerr(tdb, tdb->ecode, TDB_DEBUG_FATAL, + ecode = methods->twrite(tdb, tdb->transaction->magic_offset, + &invalid, sizeof(invalid)); + if (ecode == TDB_SUCCESS) + ecode = transaction_sync(tdb, + tdb->transaction->magic_offset, + sizeof(invalid)); + if (ecode != TDB_SUCCESS) { + tdb_logerr(tdb, ecode, TDB_LOG_ERROR, "tdb_transaction_cancel: failed to remove" " recovery magic"); } } - if (tdb->allrecord_lock.count) - tdb_allrecord_unlock(tdb, tdb->allrecord_lock.ltype); + if (tdb->file->allrecord_lock.count) + tdb_allrecord_unlock(tdb, tdb->file->allrecord_lock.ltype); /* restore the normal io methods */ tdb->methods = tdb->transaction->io_methods; @@ -496,73 +514,87 @@ static void _tdb_transaction_cancel(struct tdb_context *tdb) start a tdb transaction. No token is returned, as only a single transaction is allowed to be pending per tdb_context */ -int tdb_transaction_start(struct tdb_context *tdb) +enum TDB_ERROR tdb_transaction_start(struct tdb_context *tdb) { + enum TDB_ERROR ecode; + /* some sanity checks */ if (tdb->read_only || (tdb->flags & TDB_INTERNAL)) { - tdb_logerr(tdb, TDB_ERR_EINVAL, TDB_DEBUG_ERROR, - "tdb_transaction_start: cannot start a transaction" - " on a read-only or internal db"); - return -1; + return tdb->last_error = tdb_logerr(tdb, TDB_ERR_EINVAL, + TDB_LOG_USE_ERROR, + "tdb_transaction_start:" + " cannot start a" + " transaction on a " + "read-only or internal db"); } /* cope with nested tdb_transaction_start() calls */ if (tdb->transaction != NULL) { - tdb_logerr(tdb, TDB_ERR_NESTING, TDB_DEBUG_ERROR, - "tdb_transaction_start:" - " already inside transaction"); - return -1; + if (!(tdb->flags & TDB_ALLOW_NESTING)) { + return tdb->last_error + = tdb_logerr(tdb, TDB_ERR_IO, + TDB_LOG_USE_ERROR, + "tdb_transaction_start:" + " already inside transaction"); + } + tdb->transaction->nesting++; + return 0; } if (tdb_has_hash_locks(tdb)) { /* the caller must not have any locks when starting a transaction as otherwise we'll be screwed by lack - of nested locks in posix */ - tdb_logerr(tdb, TDB_ERR_LOCK, TDB_DEBUG_ERROR, - "tdb_transaction_start: cannot start a transaction" - " with locks held"); - return -1; + of nested locks in POSIX */ + return tdb->last_error = tdb_logerr(tdb, TDB_ERR_LOCK, + TDB_LOG_USE_ERROR, + "tdb_transaction_start:" + " cannot start a" + " transaction with locks" + " held"); } tdb->transaction = (struct tdb_transaction *) calloc(sizeof(struct tdb_transaction), 1); if (tdb->transaction == NULL) { - tdb_logerr(tdb, TDB_ERR_OOM, TDB_DEBUG_ERROR, - "tdb_transaction_start: cannot allocate"); - return -1; + return tdb->last_error = tdb_logerr(tdb, TDB_ERR_OOM, + TDB_LOG_ERROR, + "tdb_transaction_start:" + " cannot allocate"); } /* get the transaction write lock. This is a blocking lock. As discussed with Volker, there are a number of ways we could make this async, which we will probably do in the future */ - if (tdb_transaction_lock(tdb, F_WRLCK) == -1) { + ecode = tdb_transaction_lock(tdb, F_WRLCK); + if (ecode != TDB_SUCCESS) { SAFE_FREE(tdb->transaction->blocks); SAFE_FREE(tdb->transaction); - return -1; + return tdb->last_error = ecode; } /* get a read lock over entire file. This is upgraded to a write lock during the commit */ - if (tdb_allrecord_lock(tdb, F_RDLCK, TDB_LOCK_WAIT, true) == -1) { + ecode = tdb_allrecord_lock(tdb, F_RDLCK, TDB_LOCK_WAIT, true); + if (ecode != TDB_SUCCESS) { goto fail_allrecord_lock; } /* make sure we know about any file expansions already done by anyone else */ - tdb->methods->oob(tdb, tdb->map_size + 1, true); - tdb->transaction->old_map_size = tdb->map_size; + tdb->methods->oob(tdb, tdb->file->map_size + 1, true); + tdb->transaction->old_map_size = tdb->file->map_size; /* finally hook the io methods, replacing them with transaction specific methods */ tdb->transaction->io_methods = tdb->methods; tdb->methods = &transaction_methods; - return 0; + return tdb->last_error = TDB_SUCCESS; fail_allrecord_lock: tdb_transaction_unlock(tdb, F_WRLCK); SAFE_FREE(tdb->transaction->blocks); SAFE_FREE(tdb->transaction); - return -1; + return tdb->last_error = ecode; } @@ -575,16 +607,16 @@ void tdb_transaction_cancel(struct tdb_context *tdb) } /* - work out how much space the linearised recovery data will consume + work out how much space the linearised recovery data will consume (worst case) */ static tdb_len_t tdb_recovery_size(struct tdb_context *tdb) { tdb_len_t recovery_size = 0; int i; - recovery_size = sizeof(tdb_len_t); + recovery_size = 0; for (i=0;itransaction->num_blocks;i++) { - if (i * getpagesize() >= tdb->transaction->old_map_size) { + if (i * PAGESIZE >= tdb->transaction->old_map_size) { break; } if (tdb->transaction->blocks[i] == NULL) { @@ -594,356 +626,415 @@ static tdb_len_t tdb_recovery_size(struct tdb_context *tdb) if (i == tdb->transaction->num_blocks-1) { recovery_size += tdb->transaction->last_block_size; } else { - recovery_size += getpagesize(); + recovery_size += PAGESIZE; } } return recovery_size; } -/* - allocate the recovery area, or use an existing recovery area if it is - large enough -*/ -static int tdb_recovery_allocate(struct tdb_context *tdb, - tdb_len_t *recovery_size, - tdb_off_t *recovery_offset, - tdb_len_t *recovery_max_size) +static enum TDB_ERROR tdb_recovery_area(struct tdb_context *tdb, + const struct tdb_methods *methods, + tdb_off_t *recovery_offset, + struct tdb_recovery_record *rec) { - struct tdb_recovery_record rec; - const struct tdb_methods *methods = tdb->transaction->io_methods; - tdb_off_t recovery_head; - size_t addition; - - recovery_head = tdb_read_off(tdb, offsetof(struct tdb_header,recovery)); - if (recovery_head == TDB_OFF_ERR) { - tdb_logerr(tdb, tdb->ecode, TDB_DEBUG_FATAL, - "tdb_recovery_allocate:" - " failed to read recovery head"); - return -1; - } - - if (recovery_head != 0) { - if (methods->read(tdb, recovery_head, &rec, sizeof(rec))) { - tdb_logerr(tdb, tdb->ecode, TDB_DEBUG_FATAL, - "tdb_recovery_allocate:" - " failed to read recovery record"); - return -1; - } - tdb_convert(tdb, &rec, sizeof(rec)); - /* ignore invalid recovery regions: can happen in crash */ - if (rec.magic != TDB_RECOVERY_MAGIC && - rec.magic != TDB_RECOVERY_INVALID_MAGIC) { - recovery_head = 0; - } - } - - *recovery_size = tdb_recovery_size(tdb); + enum TDB_ERROR ecode; - if (recovery_head != 0 && *recovery_size <= rec.max_len) { - /* it fits in the existing area */ - *recovery_max_size = rec.max_len; - *recovery_offset = recovery_head; - return 0; + *recovery_offset = tdb_read_off(tdb, + offsetof(struct tdb_header, recovery)); + if (TDB_OFF_IS_ERR(*recovery_offset)) { + return *recovery_offset; } - /* we need to free up the old recovery area, then allocate a - new one at the end of the file. Note that we cannot use - normal allocation to allocate the new one as that might return - us an area that is being currently used (as of the start of - the transaction) */ - if (recovery_head != 0) { - add_stat(tdb, frees, 1); - if (add_free_record(tdb, recovery_head, - sizeof(rec) + rec.max_len) != 0) { - tdb_logerr(tdb, tdb->ecode, TDB_DEBUG_FATAL, - "tdb_recovery_allocate:" - " failed to free previous recovery area"); - return -1; - } + if (*recovery_offset == 0) { + rec->max_len = 0; + return TDB_SUCCESS; } - /* the tdb_free() call might have increased the recovery size */ - *recovery_size = tdb_recovery_size(tdb); - - /* round up to a multiple of page size */ - *recovery_max_size - = (((sizeof(rec) + *recovery_size) + getpagesize()-1) - & ~(getpagesize()-1)) - - sizeof(rec); - *recovery_offset = tdb->map_size; - recovery_head = *recovery_offset; + ecode = methods->tread(tdb, *recovery_offset, rec, sizeof(*rec)); + if (ecode != TDB_SUCCESS) + return ecode; - /* Restore ->map_size before calling underlying expand_file. - Also so that we don't try to expand the file again in the - transaction commit, which would destroy the recovery - area */ - addition = (tdb->map_size - tdb->transaction->old_map_size) + - sizeof(rec) + *recovery_max_size; - tdb->map_size = tdb->transaction->old_map_size; - if (methods->expand_file(tdb, addition) == -1) { - tdb_logerr(tdb, tdb->ecode, TDB_DEBUG_FATAL, - "tdb_recovery_allocate:" - " failed to create recovery area"); - return -1; + tdb_convert(tdb, rec, sizeof(*rec)); + /* ignore invalid recovery regions: can happen in crash */ + if (rec->magic != TDB_RECOVERY_MAGIC && + rec->magic != TDB_RECOVERY_INVALID_MAGIC) { + *recovery_offset = 0; + rec->max_len = 0; } + return TDB_SUCCESS; +} - /* we have to reset the old map size so that we don't try to - expand the file again in the transaction commit, which - would destroy the recovery area */ - tdb->transaction->old_map_size = tdb->map_size; +static unsigned int same(const unsigned char *new, + const unsigned char *old, + unsigned int length) +{ + unsigned int i; - /* write the recovery header offset and sync - we can sync without a race here - as the magic ptr in the recovery record has not been set */ - tdb_convert(tdb, &recovery_head, sizeof(recovery_head)); - if (methods->write(tdb, offsetof(struct tdb_header, recovery), - &recovery_head, sizeof(tdb_off_t)) == -1) { - tdb_logerr(tdb, tdb->ecode, TDB_DEBUG_FATAL, - "tdb_recovery_allocate:" - " failed to write recovery head"); - return -1; + for (i = 0; i < length; i++) { + if (new[i] != old[i]) + break; } - transaction_write_existing(tdb, offsetof(struct tdb_header, recovery), - &recovery_head, - sizeof(tdb_off_t)); - return 0; + return i; } -/* Set up header for the recovery record. */ -static void set_recovery_header(struct tdb_recovery_record *rec, - uint64_t magic, - uint64_t datalen, uint64_t actuallen, - uint64_t oldsize) +static unsigned int different(const unsigned char *new, + const unsigned char *old, + unsigned int length, + unsigned int min_same, + unsigned int *samelen) { - rec->magic = magic; - rec->max_len = actuallen; - rec->len = datalen; - rec->eof = oldsize; + unsigned int i; + + *samelen = 0; + for (i = 0; i < length; i++) { + if (new[i] == old[i]) { + (*samelen)++; + } else { + if (*samelen >= min_same) { + return i - *samelen; + } + *samelen = 0; + } + } + + if (*samelen < min_same) + *samelen = 0; + return length - *samelen; } -/* - setup the recovery data that will be used on a crash during commit -*/ -static int transaction_setup_recovery(struct tdb_context *tdb, - tdb_off_t *magic_offset) +/* Allocates recovery blob, without tdb_recovery_record at head set up. */ +static struct tdb_recovery_record *alloc_recovery(struct tdb_context *tdb, + tdb_len_t *len) { - tdb_len_t recovery_size; - unsigned char *data, *p; - const struct tdb_methods *methods = tdb->transaction->io_methods; struct tdb_recovery_record *rec; - tdb_off_t recovery_offset, recovery_max_size; - tdb_off_t old_map_size = tdb->transaction->old_map_size; - uint64_t magic, tailer; - int i; + size_t i; + enum TDB_ERROR ecode; + unsigned char *p; + const struct tdb_methods *old_methods = tdb->methods; - /* - check that the recovery area has enough space - */ - if (tdb_recovery_allocate(tdb, &recovery_size, - &recovery_offset, &recovery_max_size) == -1) { - return -1; + rec = malloc(sizeof(*rec) + tdb_recovery_size(tdb)); + if (!rec) { + tdb_logerr(tdb, TDB_ERR_OOM, TDB_LOG_ERROR, + "transaction_setup_recovery:" + " cannot allocate"); + return TDB_ERR_PTR(TDB_ERR_OOM); } - data = (unsigned char *)malloc(recovery_size + sizeof(*rec)); - if (data == NULL) { - tdb_logerr(tdb, TDB_ERR_OOM, TDB_DEBUG_FATAL, - "transaction_setup_recovery: cannot allocate"); - return -1; - } - - rec = (struct tdb_recovery_record *)data; - set_recovery_header(rec, TDB_RECOVERY_INVALID_MAGIC, - recovery_size, recovery_max_size, old_map_size); - tdb_convert(tdb, rec, sizeof(*rec)); + /* We temporarily revert to the old I/O methods, so we can use + * tdb_access_read */ + tdb->methods = tdb->transaction->io_methods; /* build the recovery data into a single blob to allow us to do a single large write, which should be more efficient */ - p = data + sizeof(*rec); + p = (unsigned char *)(rec + 1); for (i=0;itransaction->num_blocks;i++) { tdb_off_t offset; tdb_len_t length; + unsigned int off; + const unsigned char *buffer; if (tdb->transaction->blocks[i] == NULL) { continue; } - offset = i * getpagesize(); - length = getpagesize(); + offset = i * PAGESIZE; + length = PAGESIZE; if (i == tdb->transaction->num_blocks-1) { length = tdb->transaction->last_block_size; } - if (offset >= old_map_size) { + if (offset >= tdb->transaction->old_map_size) { continue; } - if (offset + length > tdb->map_size) { - tdb_logerr(tdb, TDB_ERR_CORRUPT, TDB_DEBUG_FATAL, - "tdb_transaction_setup_recovery:" - " transaction data over new region boundary"); - free(data); - return -1; + + if (offset + length > tdb->file->map_size) { + ecode = tdb_logerr(tdb, TDB_ERR_CORRUPT, TDB_LOG_ERROR, + "tdb_transaction_setup_recovery:" + " transaction data over new region" + " boundary"); + goto fail; } - memcpy(p, &offset, sizeof(offset)); - memcpy(p + sizeof(offset), &length, sizeof(length)); - tdb_convert(tdb, p, sizeof(offset) + sizeof(length)); - - /* the recovery area contains the old data, not the - new data, so we have to call the original tdb_read - method to get it */ - if (methods->read(tdb, offset, - p + sizeof(offset) + sizeof(length), - length) != 0) { - free(data); - return -1; + if (offset + length > tdb->transaction->old_map_size) { + /* Short read at EOF. */ + length = tdb->transaction->old_map_size - offset; + } + buffer = tdb_access_read(tdb, offset, length, false); + if (TDB_PTR_IS_ERR(buffer)) { + ecode = TDB_PTR_ERR(buffer); + goto fail; + } + + /* Skip over anything the same at the start. */ + off = same(tdb->transaction->blocks[i], buffer, length); + offset += off; + + while (off < length) { + tdb_len_t len; + unsigned int samelen; + + len = different(tdb->transaction->blocks[i] + off, + buffer + off, length - off, + sizeof(offset) + sizeof(len) + 1, + &samelen); + + memcpy(p, &offset, sizeof(offset)); + memcpy(p + sizeof(offset), &len, sizeof(len)); + tdb_convert(tdb, p, sizeof(offset) + sizeof(len)); + p += sizeof(offset) + sizeof(len); + memcpy(p, buffer + off, len); + p += len; + off += len + samelen; + offset += len + samelen; } - p += sizeof(offset) + sizeof(length) + length; + tdb_access_release(tdb, buffer); + } + + *len = p - (unsigned char *)(rec + 1); + tdb->methods = old_methods; + return rec; + +fail: + free(rec); + tdb->methods = old_methods; + return TDB_ERR_PTR(ecode); +} + +static tdb_off_t create_recovery_area(struct tdb_context *tdb, + tdb_len_t rec_length, + struct tdb_recovery_record *rec) +{ + tdb_off_t off, recovery_off; + tdb_len_t addition; + enum TDB_ERROR ecode; + const struct tdb_methods *methods = tdb->transaction->io_methods; + + /* round up to a multiple of page size. Overallocate, since each + * such allocation forces us to expand the file. */ + rec->max_len + = (((sizeof(*rec) + rec_length + rec_length / 2) + + PAGESIZE-1) & ~(PAGESIZE-1)) + - sizeof(*rec); + off = tdb->file->map_size; + + /* Restore ->map_size before calling underlying expand_file. + Also so that we don't try to expand the file again in the + transaction commit, which would destroy the recovery + area */ + addition = (tdb->file->map_size - tdb->transaction->old_map_size) + + sizeof(*rec) + rec->max_len; + tdb->file->map_size = tdb->transaction->old_map_size; + ecode = methods->expand_file(tdb, addition); + if (ecode != TDB_SUCCESS) { + return tdb_logerr(tdb, ecode, TDB_LOG_ERROR, + "tdb_recovery_allocate:" + " failed to create recovery area"); + } + + /* we have to reset the old map size so that we don't try to + expand the file again in the transaction commit, which + would destroy the recovery area */ + tdb->transaction->old_map_size = tdb->file->map_size; + + /* write the recovery header offset and sync - we can sync without a race here + as the magic ptr in the recovery record has not been set */ + recovery_off = off; + tdb_convert(tdb, &recovery_off, sizeof(recovery_off)); + ecode = methods->twrite(tdb, offsetof(struct tdb_header, recovery), + &recovery_off, sizeof(tdb_off_t)); + if (ecode != TDB_SUCCESS) { + return tdb_logerr(tdb, ecode, TDB_LOG_ERROR, + "tdb_recovery_allocate:" + " failed to write recovery head"); } + transaction_write_existing(tdb, offsetof(struct tdb_header, recovery), + &recovery_off, + sizeof(tdb_off_t)); + return off; +} + +/* + setup the recovery data that will be used on a crash during commit +*/ +static enum TDB_ERROR transaction_setup_recovery(struct tdb_context *tdb) +{ + tdb_len_t recovery_size = 0; + tdb_off_t recovery_off = 0; + tdb_off_t old_map_size = tdb->transaction->old_map_size; + struct tdb_recovery_record *recovery; + const struct tdb_methods *methods = tdb->transaction->io_methods; + uint64_t magic; + enum TDB_ERROR ecode; + + recovery = alloc_recovery(tdb, &recovery_size); + if (TDB_PTR_IS_ERR(recovery)) + return TDB_PTR_ERR(recovery); + + ecode = tdb_recovery_area(tdb, methods, &recovery_off, recovery); + if (ecode) { + free(recovery); + return ecode; + } + + if (recovery->max_len < recovery_size) { + /* Not large enough. Free up old recovery area. */ + if (recovery_off) { + tdb->stats.frees++; + ecode = add_free_record(tdb, recovery_off, + sizeof(*recovery) + + recovery->max_len, + TDB_LOCK_WAIT, true); + free(recovery); + if (ecode != TDB_SUCCESS) { + return tdb_logerr(tdb, ecode, TDB_LOG_ERROR, + "tdb_recovery_allocate:" + " failed to free previous" + " recovery area"); + } - /* and the tailer */ - tailer = sizeof(*rec) + recovery_max_size; - memcpy(p, &tailer, sizeof(tailer)); - tdb_convert(tdb, p, sizeof(tailer)); + /* Refresh recovery after add_free_record above. */ + recovery = alloc_recovery(tdb, &recovery_size); + if (TDB_PTR_IS_ERR(recovery)) + return TDB_PTR_ERR(recovery); + } + + recovery_off = create_recovery_area(tdb, recovery_size, + recovery); + if (TDB_OFF_IS_ERR(recovery_off)) { + free(recovery); + return recovery_off; + } + } + + /* Now we know size, convert rec header. */ + recovery->magic = TDB_RECOVERY_INVALID_MAGIC; + recovery->len = recovery_size; + recovery->eof = old_map_size; + tdb_convert(tdb, recovery, sizeof(*recovery)); /* write the recovery data to the recovery area */ - if (methods->write(tdb, recovery_offset, data, - sizeof(*rec) + recovery_size) == -1) { - tdb_logerr(tdb, tdb->ecode, TDB_DEBUG_FATAL, - "tdb_transaction_setup_recovery:" - " failed to write recovery data"); - free(data); - return -1; + ecode = methods->twrite(tdb, recovery_off, recovery, recovery_size); + if (ecode != TDB_SUCCESS) { + free(recovery); + return tdb_logerr(tdb, ecode, TDB_LOG_ERROR, + "tdb_transaction_setup_recovery:" + " failed to write recovery data"); } - transaction_write_existing(tdb, recovery_offset, data, - sizeof(*rec) + recovery_size); + transaction_write_existing(tdb, recovery_off, recovery, recovery_size); + + free(recovery); /* as we don't have ordered writes, we have to sync the recovery data before we update the magic to indicate that the recovery data is present */ - if (transaction_sync(tdb, recovery_offset, - sizeof(*rec) + recovery_size) == -1) { - free(data); - return -1; - } - - free(data); + ecode = transaction_sync(tdb, recovery_off, recovery_size); + if (ecode != TDB_SUCCESS) + return ecode; magic = TDB_RECOVERY_MAGIC; tdb_convert(tdb, &magic, sizeof(magic)); - *magic_offset = recovery_offset + offsetof(struct tdb_recovery_record, - magic); + tdb->transaction->magic_offset + = recovery_off + offsetof(struct tdb_recovery_record, magic); - if (methods->write(tdb, *magic_offset, &magic, sizeof(magic)) == -1) { - tdb_logerr(tdb, tdb->ecode, TDB_DEBUG_FATAL, - "tdb_transaction_setup_recovery:" - " failed to write recovery magic"); - return -1; + ecode = methods->twrite(tdb, tdb->transaction->magic_offset, + &magic, sizeof(magic)); + if (ecode != TDB_SUCCESS) { + return tdb_logerr(tdb, ecode, TDB_LOG_ERROR, + "tdb_transaction_setup_recovery:" + " failed to write recovery magic"); } - transaction_write_existing(tdb, *magic_offset, &magic, sizeof(magic)); + transaction_write_existing(tdb, tdb->transaction->magic_offset, + &magic, sizeof(magic)); /* ensure the recovery magic marker is on disk */ - if (transaction_sync(tdb, *magic_offset, sizeof(magic)) == -1) { - return -1; - } - - return 0; + return transaction_sync(tdb, tdb->transaction->magic_offset, + sizeof(magic)); } -static int _tdb_transaction_prepare_commit(struct tdb_context *tdb) +static enum TDB_ERROR _tdb_transaction_prepare_commit(struct tdb_context *tdb) { const struct tdb_methods *methods; + enum TDB_ERROR ecode; if (tdb->transaction == NULL) { - tdb_logerr(tdb, TDB_ERR_EINVAL, TDB_DEBUG_ERROR, - "tdb_transaction_prepare_commit: no transaction"); - return -1; + return tdb_logerr(tdb, TDB_ERR_EINVAL, TDB_LOG_USE_ERROR, + "tdb_transaction_prepare_commit:" + " no transaction"); } if (tdb->transaction->prepared) { _tdb_transaction_cancel(tdb); - tdb_logerr(tdb, TDB_ERR_EINVAL, TDB_DEBUG_ERROR, - "tdb_transaction_prepare_commit:" - " transaction already prepared"); - return -1; + return tdb_logerr(tdb, TDB_ERR_EINVAL, TDB_LOG_USE_ERROR, + "tdb_transaction_prepare_commit:" + " transaction already prepared"); } if (tdb->transaction->transaction_error) { _tdb_transaction_cancel(tdb); - tdb_logerr(tdb, TDB_ERR_EINVAL, TDB_DEBUG_ERROR, - "tdb_transaction_prepare_commit:" - " transaction error pending"); - return -1; + return tdb_logerr(tdb, TDB_ERR_EINVAL, TDB_LOG_ERROR, + "tdb_transaction_prepare_commit:" + " transaction error pending"); } if (tdb->transaction->nesting != 0) { - tdb->transaction->nesting--; - return 0; + return TDB_SUCCESS; } /* check for a null transaction */ if (tdb->transaction->blocks == NULL) { - return 0; + return TDB_SUCCESS; } methods = tdb->transaction->io_methods; /* upgrade the main transaction lock region to a write lock */ - if (tdb_allrecord_upgrade(tdb) == -1) { - tdb_logerr(tdb, tdb->ecode, TDB_DEBUG_ERROR, - "tdb_transaction_prepare_commit:" - " failed to upgrade hash locks"); - _tdb_transaction_cancel(tdb); - return -1; + ecode = tdb_allrecord_upgrade(tdb); + if (ecode != TDB_SUCCESS) { + return ecode; } /* get the open lock - this prevents new users attaching to the database during the commit */ - if (tdb_lock_open(tdb, TDB_LOCK_WAIT|TDB_LOCK_NOCHECK) == -1) { - tdb_logerr(tdb, tdb->ecode, TDB_DEBUG_ERROR, - "tdb_transaction_prepare_commit:" - " failed to get open lock"); - _tdb_transaction_cancel(tdb); - return -1; + ecode = tdb_lock_open(tdb, TDB_LOCK_WAIT|TDB_LOCK_NOCHECK); + if (ecode != TDB_SUCCESS) { + return ecode; } /* Since we have whole db locked, we don't need the expansion lock. */ if (!(tdb->flags & TDB_NOSYNC)) { - /* write the recovery data to the end of the file */ - if (transaction_setup_recovery(tdb, &tdb->transaction->magic_offset) == -1) { - tdb_logerr(tdb, tdb->ecode, TDB_DEBUG_FATAL, - "tdb_transaction_prepare_commit:" - " failed to setup recovery data"); - _tdb_transaction_cancel(tdb); - return -1; + /* Sets up tdb->transaction->recovery and + * tdb->transaction->magic_offset. */ + ecode = transaction_setup_recovery(tdb); + if (ecode != TDB_SUCCESS) { + return ecode; } } tdb->transaction->prepared = true; /* expand the file to the new size if needed */ - if (tdb->map_size != tdb->transaction->old_map_size) { - tdb_len_t add = tdb->map_size - tdb->transaction->old_map_size; + if (tdb->file->map_size != tdb->transaction->old_map_size) { + tdb_len_t add; + + add = tdb->file->map_size - tdb->transaction->old_map_size; /* Restore original map size for tdb_expand_file */ - tdb->map_size = tdb->transaction->old_map_size; - if (methods->expand_file(tdb, add) == -1) { - tdb_logerr(tdb, tdb->ecode, TDB_DEBUG_ERROR, - "tdb_transaction_prepare_commit:" - " expansion failed"); - _tdb_transaction_cancel(tdb); - return -1; + tdb->file->map_size = tdb->transaction->old_map_size; + ecode = methods->expand_file(tdb, add); + if (ecode != TDB_SUCCESS) { + return ecode; } } /* Keep the open lock until the actual commit */ - - return 0; + return TDB_SUCCESS; } /* prepare to commit the current transaction */ -int tdb_transaction_prepare_commit(struct tdb_context *tdb) +enum TDB_ERROR tdb_transaction_prepare_commit(struct tdb_context *tdb) { return _tdb_transaction_prepare_commit(tdb); } @@ -951,43 +1042,38 @@ int tdb_transaction_prepare_commit(struct tdb_context *tdb) /* commit the current transaction */ -int tdb_transaction_commit(struct tdb_context *tdb) +enum TDB_ERROR tdb_transaction_commit(struct tdb_context *tdb) { const struct tdb_methods *methods; int i; + enum TDB_ERROR ecode; if (tdb->transaction == NULL) { - tdb_logerr(tdb, TDB_ERR_EINVAL, TDB_DEBUG_ERROR, - "tdb_transaction_commit: no transaction"); - return -1; + return tdb->last_error = tdb_logerr(tdb, TDB_ERR_EINVAL, + TDB_LOG_USE_ERROR, + "tdb_transaction_commit:" + " no transaction"); } tdb_trace(tdb, "tdb_transaction_commit"); - if (tdb->transaction->transaction_error) { - tdb_transaction_cancel(tdb); - tdb_logerr(tdb, TDB_ERR_IO, TDB_DEBUG_ERROR, - "tdb_transaction_commit:" - " transaction error pending"); - return -1; - } - - if (tdb->transaction->nesting != 0) { tdb->transaction->nesting--; - return 0; + return tdb->last_error = TDB_SUCCESS; } /* check for a null transaction */ if (tdb->transaction->blocks == NULL) { _tdb_transaction_cancel(tdb); - return 0; + return tdb->last_error = TDB_SUCCESS; } if (!tdb->transaction->prepared) { - int ret = _tdb_transaction_prepare_commit(tdb); - if (ret) - return ret; + ecode = _tdb_transaction_prepare_commit(tdb); + if (ecode != TDB_SUCCESS) { + _tdb_transaction_cancel(tdb); + return tdb->last_error = ecode; + } } methods = tdb->transaction->io_methods; @@ -1001,18 +1087,15 @@ int tdb_transaction_commit(struct tdb_context *tdb) continue; } - offset = i * getpagesize(); - length = getpagesize(); + offset = i * PAGESIZE; + length = PAGESIZE; if (i == tdb->transaction->num_blocks-1) { length = tdb->transaction->last_block_size; } - if (methods->write(tdb, offset, tdb->transaction->blocks[i], - length) == -1) { - tdb_logerr(tdb, tdb->ecode, TDB_DEBUG_FATAL, - "tdb_transaction_commit:" - " write failed during commit"); - + ecode = methods->twrite(tdb, offset, + tdb->transaction->blocks[i], length); + if (ecode != TDB_SUCCESS) { /* we've overwritten part of the data and possibly expanded the file, so we need to run the crash recovery code */ @@ -1021,7 +1104,7 @@ int tdb_transaction_commit(struct tdb_context *tdb) _tdb_transaction_cancel(tdb); - return -1; + return tdb->last_error = ecode; } SAFE_FREE(tdb->transaction->blocks[i]); } @@ -1030,8 +1113,9 @@ int tdb_transaction_commit(struct tdb_context *tdb) tdb->transaction->num_blocks = 0; /* ensure the new data is on disk */ - if (transaction_sync(tdb, 0, tdb->map_size) == -1) { - return -1; + ecode = transaction_sync(tdb, 0, tdb->file->map_size); + if (ecode != TDB_SUCCESS) { + return tdb->last_error = ecode; } /* @@ -1050,10 +1134,11 @@ int tdb_transaction_commit(struct tdb_context *tdb) #endif /* use a transaction cancel to free memory and remove the - transaction locks */ + transaction locks: it "restores" map_size, too. */ + tdb->transaction->old_map_size = tdb->file->map_size; _tdb_transaction_cancel(tdb); - return 0; + return tdb->last_error = TDB_SUCCESS; } @@ -1062,63 +1147,61 @@ int tdb_transaction_commit(struct tdb_context *tdb) database write access already established (including the open lock to prevent new processes attaching) */ -int tdb_transaction_recover(struct tdb_context *tdb) +enum TDB_ERROR tdb_transaction_recover(struct tdb_context *tdb) { tdb_off_t recovery_head, recovery_eof; unsigned char *data, *p; struct tdb_recovery_record rec; + enum TDB_ERROR ecode; /* find the recovery area */ recovery_head = tdb_read_off(tdb, offsetof(struct tdb_header,recovery)); - if (recovery_head == TDB_OFF_ERR) { - tdb_logerr(tdb, tdb->ecode, TDB_DEBUG_FATAL, - "tdb_transaction_recover:" - " failed to read recovery head"); - return -1; + if (TDB_OFF_IS_ERR(recovery_head)) { + return tdb_logerr(tdb, recovery_head, TDB_LOG_ERROR, + "tdb_transaction_recover:" + " failed to read recovery head"); } if (recovery_head == 0) { /* we have never allocated a recovery record */ - return 0; + return TDB_SUCCESS; } /* read the recovery record */ - if (tdb_read_convert(tdb, recovery_head, &rec, sizeof(rec)) == -1) { - tdb_logerr(tdb, tdb->ecode, TDB_DEBUG_FATAL, - "tdb_transaction_recover:" - " failed to read recovery record"); - return -1; + ecode = tdb_read_convert(tdb, recovery_head, &rec, sizeof(rec)); + if (ecode != TDB_SUCCESS) { + return tdb_logerr(tdb, ecode, TDB_LOG_ERROR, + "tdb_transaction_recover:" + " failed to read recovery record"); } if (rec.magic != TDB_RECOVERY_MAGIC) { /* there is no valid recovery data */ - return 0; + return TDB_SUCCESS; } if (tdb->read_only) { - tdb_logerr(tdb, TDB_ERR_CORRUPT, TDB_DEBUG_FATAL, - "tdb_transaction_recover:" - " attempt to recover read only database"); - return -1; + return tdb_logerr(tdb, TDB_ERR_CORRUPT, TDB_LOG_ERROR, + "tdb_transaction_recover:" + " attempt to recover read only database"); } recovery_eof = rec.eof; data = (unsigned char *)malloc(rec.len); if (data == NULL) { - tdb_logerr(tdb, TDB_ERR_OOM, TDB_DEBUG_FATAL, - "tdb_transaction_recover:" - " failed to allocate recovery data"); - return -1; + return tdb_logerr(tdb, TDB_ERR_OOM, TDB_LOG_ERROR, + "tdb_transaction_recover:" + " failed to allocate recovery data"); } /* read the full recovery data */ - if (tdb->methods->read(tdb, recovery_head + sizeof(rec), data, - rec.len) == -1) { - tdb_logerr(tdb, tdb->ecode, TDB_DEBUG_FATAL, - "tdb_transaction_recover:" - " failed to read recovery data"); - return -1; + ecode = tdb->methods->tread(tdb, recovery_head + sizeof(rec), data, + rec.len); + if (ecode != TDB_SUCCESS) { + return tdb_logerr(tdb, ecode, TDB_LOG_ERROR, + "tdb_transaction_recover:" + " failed to read recovery data"); } /* recover the file data */ @@ -1131,71 +1214,75 @@ int tdb_transaction_recover(struct tdb_context *tdb) memcpy(&len, p + sizeof(ofs), sizeof(len)); p += sizeof(ofs) + sizeof(len); - if (tdb->methods->write(tdb, ofs, p, len) == -1) { + ecode = tdb->methods->twrite(tdb, ofs, p, len); + if (ecode != TDB_SUCCESS) { free(data); - tdb_logerr(tdb, tdb->ecode, TDB_DEBUG_FATAL, - "tdb_transaction_recover:" - " failed to recover %zu bytes at offset %zu", - (size_t)len, (size_t)ofs); - return -1; + return tdb_logerr(tdb, ecode, TDB_LOG_ERROR, + "tdb_transaction_recover:" + " failed to recover %zu bytes" + " at offset %zu", + (size_t)len, (size_t)ofs); } p += len; } free(data); - if (transaction_sync(tdb, 0, tdb->map_size) == -1) { - tdb_logerr(tdb, tdb->ecode, TDB_DEBUG_FATAL, - "tdb_transaction_recover: failed to sync recovery"); - return -1; + ecode = transaction_sync(tdb, 0, tdb->file->map_size); + if (ecode != TDB_SUCCESS) { + return tdb_logerr(tdb, ecode, TDB_LOG_ERROR, + "tdb_transaction_recover:" + " failed to sync recovery"); } /* if the recovery area is after the recovered eof then remove it */ if (recovery_eof <= recovery_head) { - if (tdb_write_off(tdb, offsetof(struct tdb_header,recovery), 0) - == -1) { - tdb_logerr(tdb, tdb->ecode, TDB_DEBUG_FATAL, - "tdb_transaction_recover:" - " failed to remove recovery head"); - return -1; + ecode = tdb_write_off(tdb, offsetof(struct tdb_header, + recovery), + 0); + if (ecode != TDB_SUCCESS) { + return tdb_logerr(tdb, ecode, TDB_LOG_ERROR, + "tdb_transaction_recover:" + " failed to remove recovery head"); } } /* remove the recovery magic */ - if (tdb_write_off(tdb, - recovery_head - + offsetof(struct tdb_recovery_record, magic), - TDB_RECOVERY_INVALID_MAGIC) == -1) { - tdb_logerr(tdb, tdb->ecode, TDB_DEBUG_FATAL, - "tdb_transaction_recover:" - " failed to remove recovery magic"); - return -1; + ecode = tdb_write_off(tdb, + recovery_head + + offsetof(struct tdb_recovery_record, magic), + TDB_RECOVERY_INVALID_MAGIC); + if (ecode != TDB_SUCCESS) { + return tdb_logerr(tdb, ecode, TDB_LOG_ERROR, + "tdb_transaction_recover:" + " failed to remove recovery magic"); } - if (transaction_sync(tdb, 0, recovery_eof) == -1) { - tdb_logerr(tdb, tdb->ecode, TDB_DEBUG_FATAL, - "tdb_transaction_recover: failed to sync2 recovery"); - return -1; + ecode = transaction_sync(tdb, 0, recovery_eof); + if (ecode != TDB_SUCCESS) { + return tdb_logerr(tdb, ecode, TDB_LOG_ERROR, + "tdb_transaction_recover:" + " failed to sync2 recovery"); } - tdb_logerr(tdb, TDB_SUCCESS, TDB_DEBUG_TRACE, + tdb_logerr(tdb, TDB_SUCCESS, TDB_LOG_WARNING, "tdb_transaction_recover: recovered %zu byte database", (size_t)recovery_eof); /* all done */ - return 0; + return TDB_SUCCESS; } -/* Any I/O failures we say "needs recovery". */ -bool tdb_needs_recovery(struct tdb_context *tdb) +tdb_bool_err tdb_needs_recovery(struct tdb_context *tdb) { tdb_off_t recovery_head; struct tdb_recovery_record rec; + enum TDB_ERROR ecode; /* find the recovery area */ recovery_head = tdb_read_off(tdb, offsetof(struct tdb_header,recovery)); - if (recovery_head == TDB_OFF_ERR) { - return true; + if (TDB_OFF_IS_ERR(recovery_head)) { + return recovery_head; } if (recovery_head == 0) { @@ -1204,8 +1291,9 @@ bool tdb_needs_recovery(struct tdb_context *tdb) } /* read the recovery record */ - if (tdb_read_convert(tdb, recovery_head, &rec, sizeof(rec)) == -1) { - return true; + ecode = tdb_read_convert(tdb, recovery_head, &rec, sizeof(rec)); + if (ecode != TDB_SUCCESS) { + return ecode; } return (rec.magic == TDB_RECOVERY_MAGIC);