2 Unix SMB/CIFS implementation.
4 trivial database library
6 Copyright (C) Andrew Tridgell 2005
8 ** NOTE! The following LGPL license applies to the tdb
9 ** library. This does NOT imply that all of Samba is released
12 This library is free software; you can redistribute it and/or
13 modify it under the terms of the GNU Lesser General Public
14 License as published by the Free Software Foundation; either
15 version 3 of the License, or (at your option) any later version.
17 This library is distributed in the hope that it will be useful,
18 but WITHOUT ANY WARRANTY; without even the implied warranty of
19 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
20 Lesser General Public License for more details.
22 You should have received a copy of the GNU Lesser General Public
23 License along with this library; if not, see <http://www.gnu.org/licenses/>.
26 #include "tdb_private.h"
31 - only allow a single transaction at a time per database. This makes
32 using the transaction API simpler, as otherwise the caller would
33 have to cope with temporary failures in transactions that conflict
34 with other current transactions
36 - keep the transaction recovery information in the same file as the
37 database, using a special 'transaction recovery' record pointed at
38 by the header. This removes the need for extra journal files as
39 used by some other databases
41 - dynamically allocated the transaction recover record, re-using it
42 for subsequent transactions. If a larger record is needed then
43 tdb_free() the old record to place it on the normal tdb freelist
44 before allocating the new record
46 - during transactions, keep a linked list of writes all that have
47 been performed by intercepting all tdb_write() calls. The hooked
48 transaction versions of tdb_read() and tdb_write() check this
49 linked list and try to use the elements of the list in preference
52 - don't allow any locks to be held when a transaction starts,
53 otherwise we can end up with deadlock (plus lack of lock nesting
54 in posix locks would mean the lock is lost)
56 - if the caller gains a lock during the transaction but doesn't
57 release it then fail the commit
59 - allow for nested calls to tdb_transaction_start(), re-using the
60 existing transaction record. If the inner transaction is cancelled
61 then a subsequent commit will fail
63 - keep a mirrored copy of the tdb hash chain heads to allow for the
64 fast hash heads scan on traverse, updating the mirrored copy in
65 the transaction version of tdb_write
67 - allow callers to mix transaction and non-transaction use of tdb,
68 although once a transaction is started then an exclusive lock is
69 gained until the transaction is committed or cancelled
71 - the commit stategy involves first saving away all modified data
72 into a linearised buffer in the transaction recovery area, then
73 marking the transaction recovery area with a magic value to
74 indicate a valid recovery record. In total 4 fsync/msync calls are
75 needed per commit to prevent race conditions. It might be possible
76 to reduce this to 3 or even 2 with some more work.
78 - check for a valid recovery record on open of the tdb, while the
79 global lock is held. Automatically recover from the transaction
80 recovery area if needed, then continue with the open as
81 usual. This allows for smooth crash recovery with no administrator
84 - if TDB_NOSYNC is passed to flags in tdb_open then transactions are
85 still available, but no transaction recovery area is used and no
86 fsync/msync calls are made.
88 - if TDB_ALLOW_NESTING is passed to flags in tdb open, or added using
89 tdb_add_flags() transaction is enabled.
90 The default is that transaction nesting is not allowed and an attempt
91 to create a nested transaction will fail with TDB_ERR_NESTING.
93 Beware. when transactions are nested a transaction successfully
94 completed with tdb_transaction_commit() can be silently unrolled later.
99 hold the context of any current transaction
101 struct tdb_transaction {
102 /* we keep a mirrored copy of the tdb hash heads here so
103 tdb_next_hash_chain() can operate efficiently */
104 uint32_t *hash_heads;
106 /* the original io methods - used to do IOs to the real db */
107 const struct tdb_methods *io_methods;
109 /* the list of transaction blocks. When a block is first
110 written to, it gets created in this list */
113 uint32_t block_size; /* bytes in each block */
114 uint32_t last_block_size; /* number of valid bytes in the last block */
116 /* non-zero when an internal transaction error has
117 occurred. All write operations will then fail until the
118 transaction is ended */
119 int transaction_error;
121 /* when inside a transaction we need to keep track of any
122 nested tdb_transaction_start() calls, as these are allowed,
123 but don't create a new transaction */
126 /* set when a prepare has already occurred */
128 tdb_off_t magic_offset;
130 /* old file size before transaction */
131 tdb_len_t old_map_size;
133 /* we should re-pack on commit */
139 read while in a transaction. We need to check first if the data is in our list
140 of transaction elements, then if not do a real read
142 static int transaction_read(struct tdb_context *tdb, tdb_off_t off, void *buf,
143 tdb_len_t len, int cv)
147 /* break it down into block sized ops */
148 while (len + (off % tdb->transaction->block_size) > tdb->transaction->block_size) {
149 tdb_len_t len2 = tdb->transaction->block_size - (off % tdb->transaction->block_size);
150 if (transaction_read(tdb, off, buf, len2, cv) != 0) {
155 buf = (void *)(len2 + (char *)buf);
162 blk = off / tdb->transaction->block_size;
164 /* see if we have it in the block list */
165 if (tdb->transaction->num_blocks <= blk ||
166 tdb->transaction->blocks[blk] == NULL) {
167 /* nope, do a real read */
168 if (tdb->transaction->io_methods->tdb_read(tdb, off, buf, len, cv) != 0) {
174 /* it is in the block list. Now check for the last block */
175 if (blk == tdb->transaction->num_blocks-1) {
176 if (len > tdb->transaction->last_block_size) {
181 /* now copy it out of this block */
182 memcpy(buf, tdb->transaction->blocks[blk] + (off % tdb->transaction->block_size), len);
184 tdb_convert(buf, len);
189 TDB_LOG((tdb, TDB_DEBUG_FATAL, "transaction_read: failed at off=%d len=%d\n", off, len));
190 tdb->ecode = TDB_ERR_IO;
191 tdb->transaction->transaction_error = 1;
197 write while in a transaction
199 static int transaction_write(struct tdb_context *tdb, tdb_off_t off,
200 const void *buf, tdb_len_t len)
204 /* Only a commit is allowed on a prepared transaction */
205 if (tdb->transaction->prepared) {
206 tdb->ecode = TDB_ERR_EINVAL;
207 TDB_LOG((tdb, TDB_DEBUG_FATAL, "transaction_write: transaction already prepared, write not allowed\n"));
208 tdb->transaction->transaction_error = 1;
212 /* if the write is to a hash head, then update the transaction
214 if (len == sizeof(tdb_off_t) && off >= FREELIST_TOP &&
215 off < FREELIST_TOP+TDB_HASHTABLE_SIZE(tdb)) {
216 uint32_t chain = (off-FREELIST_TOP) / sizeof(tdb_off_t);
217 memcpy(&tdb->transaction->hash_heads[chain], buf, len);
220 /* break it up into block sized chunks */
221 while (len + (off % tdb->transaction->block_size) > tdb->transaction->block_size) {
222 tdb_len_t len2 = tdb->transaction->block_size - (off % tdb->transaction->block_size);
223 if (transaction_write(tdb, off, buf, len2) != 0) {
229 buf = (const void *)(len2 + (const char *)buf);
237 blk = off / tdb->transaction->block_size;
238 off = off % tdb->transaction->block_size;
240 if (tdb->transaction->num_blocks <= blk) {
241 uint8_t **new_blocks;
242 /* expand the blocks array */
243 if (tdb->transaction->blocks == NULL) {
244 new_blocks = (uint8_t **)malloc(
245 (blk+1)*sizeof(uint8_t *));
247 new_blocks = (uint8_t **)realloc(
248 tdb->transaction->blocks,
249 (blk+1)*sizeof(uint8_t *));
251 if (new_blocks == NULL) {
252 tdb->ecode = TDB_ERR_OOM;
255 memset(&new_blocks[tdb->transaction->num_blocks], 0,
256 (1+(blk - tdb->transaction->num_blocks))*sizeof(uint8_t *));
257 tdb->transaction->blocks = new_blocks;
258 tdb->transaction->num_blocks = blk+1;
259 tdb->transaction->last_block_size = 0;
262 /* allocate and fill a block? */
263 if (tdb->transaction->blocks[blk] == NULL) {
264 tdb->transaction->blocks[blk] = (uint8_t *)calloc(tdb->transaction->block_size, 1);
265 if (tdb->transaction->blocks[blk] == NULL) {
266 tdb->ecode = TDB_ERR_OOM;
267 tdb->transaction->transaction_error = 1;
270 if (tdb->transaction->old_map_size > blk * tdb->transaction->block_size) {
271 tdb_len_t len2 = tdb->transaction->block_size;
272 if (len2 + (blk * tdb->transaction->block_size) > tdb->transaction->old_map_size) {
273 len2 = tdb->transaction->old_map_size - (blk * tdb->transaction->block_size);
275 if (tdb->transaction->io_methods->tdb_read(tdb, blk * tdb->transaction->block_size,
276 tdb->transaction->blocks[blk],
278 SAFE_FREE(tdb->transaction->blocks[blk]);
279 tdb->ecode = TDB_ERR_IO;
282 if (blk == tdb->transaction->num_blocks-1) {
283 tdb->transaction->last_block_size = len2;
288 /* overwrite part of an existing block */
290 memset(tdb->transaction->blocks[blk] + off, 0, len);
292 memcpy(tdb->transaction->blocks[blk] + off, buf, len);
294 if (blk == tdb->transaction->num_blocks-1) {
295 if (len + off > tdb->transaction->last_block_size) {
296 tdb->transaction->last_block_size = len + off;
303 TDB_LOG((tdb, TDB_DEBUG_FATAL, "transaction_write: failed at off=%d len=%d\n",
304 (blk*tdb->transaction->block_size) + off, len));
305 tdb->transaction->transaction_error = 1;
311 write while in a transaction - this varient never expands the transaction blocks, it only
312 updates existing blocks. This means it cannot change the recovery size
314 static int transaction_write_existing(struct tdb_context *tdb, tdb_off_t off,
315 const void *buf, tdb_len_t len)
319 /* break it up into block sized chunks */
320 while (len + (off % tdb->transaction->block_size) > tdb->transaction->block_size) {
321 tdb_len_t len2 = tdb->transaction->block_size - (off % tdb->transaction->block_size);
322 if (transaction_write_existing(tdb, off, buf, len2) != 0) {
328 buf = (const void *)(len2 + (const char *)buf);
336 blk = off / tdb->transaction->block_size;
337 off = off % tdb->transaction->block_size;
339 if (tdb->transaction->num_blocks <= blk ||
340 tdb->transaction->blocks[blk] == NULL) {
344 if (blk == tdb->transaction->num_blocks-1 &&
345 off + len > tdb->transaction->last_block_size) {
346 if (off >= tdb->transaction->last_block_size) {
349 len = tdb->transaction->last_block_size - off;
352 /* overwrite part of an existing block */
353 memcpy(tdb->transaction->blocks[blk] + off, buf, len);
360 accelerated hash chain head search, using the cached hash heads
362 static void transaction_next_hash_chain(struct tdb_context *tdb, uint32_t *chain)
365 for (;h < tdb->header.hash_size;h++) {
366 /* the +1 takes account of the freelist */
367 if (0 != tdb->transaction->hash_heads[h+1]) {
375 out of bounds check during a transaction
377 static int transaction_oob(struct tdb_context *tdb, tdb_off_t len, int probe)
379 if (len <= tdb->map_size) {
382 tdb->ecode = TDB_ERR_IO;
387 transaction version of tdb_expand().
389 static int transaction_expand_file(struct tdb_context *tdb, tdb_off_t size,
392 /* add a write to the transaction elements, so subsequent
393 reads see the zero data */
394 if (transaction_write(tdb, size, NULL, addition) != 0) {
398 tdb->transaction->need_repack = true;
404 brlock during a transaction - ignore them
406 static int transaction_brlock(struct tdb_context *tdb,
407 int rw_type, tdb_off_t offset, size_t len,
408 enum tdb_lock_flags flags)
413 static int transaction_brunlock(struct tdb_context *tdb,
414 int rw_type, tdb_off_t offset, size_t len)
419 static const struct tdb_methods transaction_methods = {
422 transaction_next_hash_chain,
424 transaction_expand_file,
432 static int transaction_sync(struct tdb_context *tdb, tdb_off_t offset, tdb_len_t length)
434 if (tdb->flags & TDB_NOSYNC) {
438 if (fsync(tdb->fd) != 0) {
439 tdb->ecode = TDB_ERR_IO;
440 TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction: fsync failed\n"));
445 tdb_off_t moffset = offset & ~(tdb->page_size-1);
446 if (msync(moffset + (char *)tdb->map_ptr,
447 length + (offset - moffset), MS_SYNC) != 0) {
448 tdb->ecode = TDB_ERR_IO;
449 TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction: msync failed - %s\n",
458 /* ltype is F_WRLCK after prepare. */
459 int _tdb_transaction_cancel(struct tdb_context *tdb, int ltype)
463 if (tdb->transaction == NULL) {
464 TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_transaction_cancel: no transaction\n"));
468 if (tdb->transaction->nesting != 0) {
469 tdb->transaction->transaction_error = 1;
470 tdb->transaction->nesting--;
474 tdb->map_size = tdb->transaction->old_map_size;
476 /* free all the transaction blocks */
477 for (i=0;i<tdb->transaction->num_blocks;i++) {
478 if (tdb->transaction->blocks[i] != NULL) {
479 free(tdb->transaction->blocks[i]);
482 SAFE_FREE(tdb->transaction->blocks);
484 if (tdb->transaction->magic_offset) {
485 const struct tdb_methods *methods = tdb->transaction->io_methods;
488 /* remove the recovery marker */
489 if (methods->tdb_write(tdb, tdb->transaction->magic_offset, &zero, 4) == -1 ||
490 transaction_sync(tdb, tdb->transaction->magic_offset, 4) == -1) {
491 TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction_cancel: failed to remove recovery magic\n"));
496 /* remove any global lock created during the transaction */
497 if (tdb->global_lock.count != 0) {
498 tdb_brunlock(tdb, tdb->global_lock.ltype,
499 FREELIST_TOP, 4*tdb->header.hash_size);
500 tdb->global_lock.count = 0;
503 /* remove any locks created during the transaction */
504 if (tdb->num_locks != 0) {
505 for (i=0;i<tdb->num_lockrecs;i++) {
506 tdb_brunlock(tdb, tdb->lockrecs[i].ltype,
507 FREELIST_TOP+4*tdb->lockrecs[i].list, 1);
510 tdb->num_lockrecs = 0;
511 SAFE_FREE(tdb->lockrecs);
514 /* restore the normal io methods */
515 tdb->methods = tdb->transaction->io_methods;
517 tdb_brunlock(tdb, ltype, FREELIST_TOP, 0);
518 tdb_transaction_unlock(tdb, F_WRLCK);
519 SAFE_FREE(tdb->transaction->hash_heads);
520 SAFE_FREE(tdb->transaction);
526 start a tdb transaction. No token is returned, as only a single
527 transaction is allowed to be pending per tdb_context
529 int tdb_transaction_start(struct tdb_context *tdb)
531 /* some sanity checks */
532 if (tdb->read_only || (tdb->flags & TDB_INTERNAL) || tdb->traverse_read) {
533 TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_transaction_start: cannot start a transaction on a read-only or internal db\n"));
534 tdb->ecode = TDB_ERR_EINVAL;
538 /* cope with nested tdb_transaction_start() calls */
539 if (tdb->transaction != NULL) {
540 if (!(tdb->flags & TDB_ALLOW_NESTING)) {
541 tdb->ecode = TDB_ERR_NESTING;
544 tdb_trace(tdb, "tdb_transaction_start");
545 tdb->transaction->nesting++;
546 TDB_LOG((tdb, TDB_DEBUG_TRACE, "tdb_transaction_start: nesting %d\n",
547 tdb->transaction->nesting));
551 if (tdb->num_locks != 0 || tdb->global_lock.count) {
552 /* the caller must not have any locks when starting a
553 transaction as otherwise we'll be screwed by lack
554 of nested locks in posix */
555 TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_transaction_start: cannot start a transaction with locks held\n"));
556 tdb->ecode = TDB_ERR_LOCK;
560 if (tdb->travlocks.next != NULL) {
561 /* you cannot use transactions inside a traverse (although you can use
562 traverse inside a transaction) as otherwise you can end up with
564 TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_transaction_start: cannot start a transaction within a traverse\n"));
565 tdb->ecode = TDB_ERR_LOCK;
569 tdb->transaction = (struct tdb_transaction *)
570 calloc(sizeof(struct tdb_transaction), 1);
571 if (tdb->transaction == NULL) {
572 tdb->ecode = TDB_ERR_OOM;
576 /* a page at a time seems like a reasonable compromise between compactness and efficiency */
577 tdb->transaction->block_size = tdb->page_size;
579 /* get the transaction write lock. This is a blocking lock. As
580 discussed with Volker, there are a number of ways we could
581 make this async, which we will probably do in the future */
582 if (tdb_transaction_lock(tdb, F_WRLCK) == -1) {
583 SAFE_FREE(tdb->transaction->blocks);
584 SAFE_FREE(tdb->transaction);
588 /* get a read lock from the freelist to the end of file. This
589 is upgraded to a write lock during the commit */
590 if (tdb_brlock(tdb, F_RDLCK, FREELIST_TOP, 0, TDB_LOCK_WAIT) == -1) {
591 TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_transaction_start: failed to get hash locks\n"));
592 tdb->ecode = TDB_ERR_LOCK;
596 /* setup a copy of the hash table heads so the hash scan in
597 traverse can be fast */
598 tdb->transaction->hash_heads = (uint32_t *)
599 calloc(tdb->header.hash_size+1, sizeof(uint32_t));
600 if (tdb->transaction->hash_heads == NULL) {
601 tdb->ecode = TDB_ERR_OOM;
604 if (tdb->methods->tdb_read(tdb, FREELIST_TOP, tdb->transaction->hash_heads,
605 TDB_HASHTABLE_SIZE(tdb), 0) != 0) {
606 TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction_start: failed to read hash heads\n"));
607 tdb->ecode = TDB_ERR_IO;
611 /* make sure we know about any file expansions already done by
613 tdb->methods->tdb_oob(tdb, tdb->map_size + 1, 1);
614 tdb->transaction->old_map_size = tdb->map_size;
616 /* finally hook the io methods, replacing them with
617 transaction specific methods */
618 tdb->transaction->io_methods = tdb->methods;
619 tdb->methods = &transaction_methods;
621 /* Trace at the end, so we get sequence number correct. */
622 tdb_trace(tdb, "tdb_transaction_start");
626 tdb_brunlock(tdb, F_RDLCK, FREELIST_TOP, 0);
627 tdb_transaction_unlock(tdb, F_WRLCK);
628 SAFE_FREE(tdb->transaction->blocks);
629 SAFE_FREE(tdb->transaction->hash_heads);
630 SAFE_FREE(tdb->transaction);
636 cancel the current transaction
638 int tdb_transaction_cancel(struct tdb_context *tdb)
641 tdb_trace(tdb, "tdb_transaction_cancel");
642 if (tdb->transaction && tdb->transaction->prepared)
644 return _tdb_transaction_cancel(tdb, ltype);
648 work out how much space the linearised recovery data will consume
650 static tdb_len_t tdb_recovery_size(struct tdb_context *tdb)
652 tdb_len_t recovery_size = 0;
655 recovery_size = sizeof(uint32_t);
656 for (i=0;i<tdb->transaction->num_blocks;i++) {
657 if (i * tdb->transaction->block_size >= tdb->transaction->old_map_size) {
660 if (tdb->transaction->blocks[i] == NULL) {
663 recovery_size += 2*sizeof(tdb_off_t);
664 if (i == tdb->transaction->num_blocks-1) {
665 recovery_size += tdb->transaction->last_block_size;
667 recovery_size += tdb->transaction->block_size;
671 return recovery_size;
675 allocate the recovery area, or use an existing recovery area if it is
678 static int tdb_recovery_allocate(struct tdb_context *tdb,
679 tdb_len_t *recovery_size,
680 tdb_off_t *recovery_offset,
681 tdb_len_t *recovery_max_size)
683 struct list_struct rec;
684 const struct tdb_methods *methods = tdb->transaction->io_methods;
685 tdb_off_t recovery_head;
687 if (tdb_ofs_read(tdb, TDB_RECOVERY_HEAD, &recovery_head) == -1) {
688 TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_recovery_allocate: failed to read recovery head\n"));
694 if (recovery_head != 0 &&
695 methods->tdb_read(tdb, recovery_head, &rec, sizeof(rec), DOCONV()) == -1) {
696 TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_recovery_allocate: failed to read recovery record\n"));
700 *recovery_size = tdb_recovery_size(tdb);
702 if (recovery_head != 0 && *recovery_size <= rec.rec_len) {
703 /* it fits in the existing area */
704 *recovery_max_size = rec.rec_len;
705 *recovery_offset = recovery_head;
709 /* we need to free up the old recovery area, then allocate a
710 new one at the end of the file. Note that we cannot use
711 tdb_allocate() to allocate the new one as that might return
712 us an area that is being currently used (as of the start of
714 if (recovery_head != 0) {
715 if (tdb_free(tdb, recovery_head, &rec) == -1) {
716 TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_recovery_allocate: failed to free previous recovery area\n"));
721 /* the tdb_free() call might have increased the recovery size */
722 *recovery_size = tdb_recovery_size(tdb);
724 /* round up to a multiple of page size */
725 *recovery_max_size = TDB_ALIGN(sizeof(rec) + *recovery_size, tdb->page_size) - sizeof(rec);
726 *recovery_offset = tdb->map_size;
727 recovery_head = *recovery_offset;
729 if (methods->tdb_expand_file(tdb, tdb->transaction->old_map_size,
730 (tdb->map_size - tdb->transaction->old_map_size) +
731 sizeof(rec) + *recovery_max_size) == -1) {
732 TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_recovery_allocate: failed to create recovery area\n"));
736 /* remap the file (if using mmap) */
737 methods->tdb_oob(tdb, tdb->map_size + 1, 1);
739 /* we have to reset the old map size so that we don't try to expand the file
740 again in the transaction commit, which would destroy the recovery area */
741 tdb->transaction->old_map_size = tdb->map_size;
743 /* write the recovery header offset and sync - we can sync without a race here
744 as the magic ptr in the recovery record has not been set */
745 CONVERT(recovery_head);
746 if (methods->tdb_write(tdb, TDB_RECOVERY_HEAD,
747 &recovery_head, sizeof(tdb_off_t)) == -1) {
748 TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_recovery_allocate: failed to write recovery head\n"));
751 if (transaction_write_existing(tdb, TDB_RECOVERY_HEAD, &recovery_head, sizeof(tdb_off_t)) == -1) {
752 TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_recovery_allocate: failed to write recovery head\n"));
761 setup the recovery data that will be used on a crash during commit
763 static int transaction_setup_recovery(struct tdb_context *tdb,
764 tdb_off_t *magic_offset)
766 tdb_len_t recovery_size;
767 unsigned char *data, *p;
768 const struct tdb_methods *methods = tdb->transaction->io_methods;
769 struct list_struct *rec;
770 tdb_off_t recovery_offset, recovery_max_size;
771 tdb_off_t old_map_size = tdb->transaction->old_map_size;
772 uint32_t magic, tailer;
776 check that the recovery area has enough space
778 if (tdb_recovery_allocate(tdb, &recovery_size,
779 &recovery_offset, &recovery_max_size) == -1) {
783 data = (unsigned char *)malloc(recovery_size + sizeof(*rec));
785 tdb->ecode = TDB_ERR_OOM;
789 rec = (struct list_struct *)data;
790 memset(rec, 0, sizeof(*rec));
793 rec->data_len = recovery_size;
794 rec->rec_len = recovery_max_size;
795 rec->key_len = old_map_size;
798 /* build the recovery data into a single blob to allow us to do a single
799 large write, which should be more efficient */
800 p = data + sizeof(*rec);
801 for (i=0;i<tdb->transaction->num_blocks;i++) {
805 if (tdb->transaction->blocks[i] == NULL) {
809 offset = i * tdb->transaction->block_size;
810 length = tdb->transaction->block_size;
811 if (i == tdb->transaction->num_blocks-1) {
812 length = tdb->transaction->last_block_size;
815 if (offset >= old_map_size) {
818 if (offset + length > tdb->transaction->old_map_size) {
819 TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction_setup_recovery: transaction data over new region boundary\n"));
821 tdb->ecode = TDB_ERR_CORRUPT;
824 memcpy(p, &offset, 4);
825 memcpy(p+4, &length, 4);
829 /* the recovery area contains the old data, not the
830 new data, so we have to call the original tdb_read
832 if (methods->tdb_read(tdb, offset, p + 8, length, 0) != 0) {
834 tdb->ecode = TDB_ERR_IO;
841 tailer = sizeof(*rec) + recovery_max_size;
842 memcpy(p, &tailer, 4);
845 /* write the recovery data to the recovery area */
846 if (methods->tdb_write(tdb, recovery_offset, data, sizeof(*rec) + recovery_size) == -1) {
847 TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction_setup_recovery: failed to write recovery data\n"));
849 tdb->ecode = TDB_ERR_IO;
852 if (transaction_write_existing(tdb, recovery_offset, data, sizeof(*rec) + recovery_size) == -1) {
853 TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction_setup_recovery: failed to write secondary recovery data\n"));
855 tdb->ecode = TDB_ERR_IO;
859 /* as we don't have ordered writes, we have to sync the recovery
860 data before we update the magic to indicate that the recovery
862 if (transaction_sync(tdb, recovery_offset, sizeof(*rec) + recovery_size) == -1) {
869 magic = TDB_RECOVERY_MAGIC;
872 *magic_offset = recovery_offset + offsetof(struct list_struct, magic);
874 if (methods->tdb_write(tdb, *magic_offset, &magic, sizeof(magic)) == -1) {
875 TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction_setup_recovery: failed to write recovery magic\n"));
876 tdb->ecode = TDB_ERR_IO;
879 if (transaction_write_existing(tdb, *magic_offset, &magic, sizeof(magic)) == -1) {
880 TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction_setup_recovery: failed to write secondary recovery magic\n"));
881 tdb->ecode = TDB_ERR_IO;
885 /* ensure the recovery magic marker is on disk */
886 if (transaction_sync(tdb, *magic_offset, sizeof(magic)) == -1) {
893 static int _tdb_transaction_prepare_commit(struct tdb_context *tdb)
895 const struct tdb_methods *methods;
897 if (tdb->transaction == NULL) {
898 TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_transaction_prepare_commit: no transaction\n"));
902 if (tdb->transaction->prepared) {
903 tdb->ecode = TDB_ERR_EINVAL;
904 _tdb_transaction_cancel(tdb, F_WRLCK);
905 TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_transaction_prepare_commit: transaction already prepared\n"));
909 if (tdb->transaction->transaction_error) {
910 tdb->ecode = TDB_ERR_IO;
911 _tdb_transaction_cancel(tdb, F_RDLCK);
912 TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_transaction_prepare_commit: transaction error pending\n"));
917 if (tdb->transaction->nesting != 0) {
918 tdb->transaction->nesting--;
923 /* store seqnum now, before reading becomes illegal. */
924 tdb_ofs_read(tdb, TDB_SEQNUM_OFS, &tdb->transaction_prepare_seqnum);
927 /* check for a null transaction */
928 if (tdb->transaction->blocks == NULL) {
932 methods = tdb->transaction->io_methods;
934 /* if there are any locks pending then the caller has not
935 nested their locks properly, so fail the transaction */
936 if (tdb->num_locks || tdb->global_lock.count) {
937 tdb->ecode = TDB_ERR_LOCK;
938 TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_transaction_prepare_commit: locks pending on commit\n"));
939 _tdb_transaction_cancel(tdb, F_RDLCK);
943 /* upgrade the main transaction lock region to a write lock */
944 if (tdb_brlock_upgrade(tdb, FREELIST_TOP, 0) == -1) {
945 TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_transaction_prepare_commit: failed to upgrade hash locks\n"));
946 tdb->ecode = TDB_ERR_LOCK;
947 _tdb_transaction_cancel(tdb, F_RDLCK);
951 /* get the global lock - this prevents new users attaching to the database
953 if (tdb_brlock(tdb, F_WRLCK, GLOBAL_LOCK, 1, TDB_LOCK_WAIT) == -1) {
954 TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_transaction_prepare_commit: failed to get global lock\n"));
955 tdb->ecode = TDB_ERR_LOCK;
956 _tdb_transaction_cancel(tdb, F_WRLCK);
960 if (!(tdb->flags & TDB_NOSYNC)) {
961 /* write the recovery data to the end of the file */
962 if (transaction_setup_recovery(tdb, &tdb->transaction->magic_offset) == -1) {
963 TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction_prepare_commit: failed to setup recovery data\n"));
964 tdb_brunlock(tdb, F_WRLCK, GLOBAL_LOCK, 1);
965 _tdb_transaction_cancel(tdb, F_WRLCK);
970 tdb->transaction->prepared = true;
972 /* expand the file to the new size if needed */
973 if (tdb->map_size != tdb->transaction->old_map_size) {
974 if (methods->tdb_expand_file(tdb, tdb->transaction->old_map_size,
976 tdb->transaction->old_map_size) == -1) {
977 tdb->ecode = TDB_ERR_IO;
978 TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction_prepare_commit: expansion failed\n"));
979 tdb_brunlock(tdb, F_WRLCK, GLOBAL_LOCK, 1);
980 _tdb_transaction_cancel(tdb, F_WRLCK);
983 tdb->map_size = tdb->transaction->old_map_size;
984 methods->tdb_oob(tdb, tdb->map_size + 1, 1);
987 /* Keep the global lock until the actual commit */
993 prepare to commit the current transaction
995 int tdb_transaction_prepare_commit(struct tdb_context *tdb)
997 tdb_trace(tdb, "tdb_transaction_prepare_commit");
998 return _tdb_transaction_prepare_commit(tdb);
1002 commit the current transaction
1004 int tdb_transaction_commit(struct tdb_context *tdb)
1006 const struct tdb_methods *methods;
1010 if (tdb->transaction == NULL) {
1011 TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_transaction_commit: no transaction\n"));
1015 /* If we've prepared, can't read seqnum. */
1016 if (tdb->transaction->prepared) {
1017 tdb_trace_seqnum(tdb, tdb->transaction_prepare_seqnum,
1018 "tdb_transaction_commit");
1020 tdb_trace(tdb, "tdb_transaction_commit");
1023 if (tdb->transaction->transaction_error) {
1024 tdb->ecode = TDB_ERR_IO;
1025 tdb_transaction_cancel(tdb);
1026 TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_transaction_commit: transaction error pending\n"));
1031 if (tdb->transaction->nesting != 0) {
1032 tdb->transaction->nesting--;
1036 /* check for a null transaction */
1037 if (tdb->transaction->blocks == NULL) {
1038 _tdb_transaction_cancel(tdb, F_RDLCK);
1042 if (!tdb->transaction->prepared) {
1043 int ret = _tdb_transaction_prepare_commit(tdb);
1048 methods = tdb->transaction->io_methods;
1050 /* perform all the writes */
1051 for (i=0;i<tdb->transaction->num_blocks;i++) {
1055 if (tdb->transaction->blocks[i] == NULL) {
1059 offset = i * tdb->transaction->block_size;
1060 length = tdb->transaction->block_size;
1061 if (i == tdb->transaction->num_blocks-1) {
1062 length = tdb->transaction->last_block_size;
1065 if (methods->tdb_write(tdb, offset, tdb->transaction->blocks[i], length) == -1) {
1066 TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction_commit: write failed during commit\n"));
1068 /* we've overwritten part of the data and
1069 possibly expanded the file, so we need to
1070 run the crash recovery code */
1071 tdb->methods = methods;
1072 tdb_transaction_recover(tdb);
1074 _tdb_transaction_cancel(tdb, F_WRLCK);
1075 tdb_brunlock(tdb, F_WRLCK, GLOBAL_LOCK, 1);
1077 TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction_commit: write failed\n"));
1080 SAFE_FREE(tdb->transaction->blocks[i]);
1083 SAFE_FREE(tdb->transaction->blocks);
1084 tdb->transaction->num_blocks = 0;
1086 /* ensure the new data is on disk */
1087 if (transaction_sync(tdb, 0, tdb->map_size) == -1) {
1091 tdb_brunlock(tdb, F_WRLCK, GLOBAL_LOCK, 1);
1094 TODO: maybe write to some dummy hdr field, or write to magic
1095 offset without mmap, before the last sync, instead of the
1099 /* on some systems (like Linux 2.6.x) changes via mmap/msync
1100 don't change the mtime of the file, this means the file may
1101 not be backed up (as tdb rounding to block sizes means that
1102 file size changes are quite rare too). The following forces
1103 mtime changes when a transaction completes */
1105 utime(tdb->name, NULL);
1108 need_repack = tdb->transaction->need_repack;
1110 /* use a transaction cancel to free memory and remove the
1111 transaction locks */
1112 _tdb_transaction_cancel(tdb, F_WRLCK);
1115 return tdb_repack(tdb);
1123 recover from an aborted transaction. Must be called with exclusive
1124 database write access already established (including the global
1125 lock to prevent new processes attaching)
1127 int tdb_transaction_recover(struct tdb_context *tdb)
1129 tdb_off_t recovery_head, recovery_eof;
1130 unsigned char *data, *p;
1132 struct list_struct rec;
1134 /* find the recovery area */
1135 if (tdb_ofs_read(tdb, TDB_RECOVERY_HEAD, &recovery_head) == -1) {
1136 TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction_recover: failed to read recovery head\n"));
1137 tdb->ecode = TDB_ERR_IO;
1141 if (recovery_head == 0) {
1142 /* we have never allocated a recovery record */
1146 /* read the recovery record */
1147 if (tdb->methods->tdb_read(tdb, recovery_head, &rec,
1148 sizeof(rec), DOCONV()) == -1) {
1149 TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction_recover: failed to read recovery record\n"));
1150 tdb->ecode = TDB_ERR_IO;
1154 if (rec.magic != TDB_RECOVERY_MAGIC) {
1155 /* there is no valid recovery data */
1159 if (tdb->read_only) {
1160 TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction_recover: attempt to recover read only database\n"));
1161 tdb->ecode = TDB_ERR_CORRUPT;
1165 recovery_eof = rec.key_len;
1167 data = (unsigned char *)malloc(rec.data_len);
1169 TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction_recover: failed to allocate recovery data\n"));
1170 tdb->ecode = TDB_ERR_OOM;
1174 /* read the full recovery data */
1175 if (tdb->methods->tdb_read(tdb, recovery_head + sizeof(rec), data,
1176 rec.data_len, 0) == -1) {
1177 TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction_recover: failed to read recovery data\n"));
1178 tdb->ecode = TDB_ERR_IO;
1182 /* recover the file data */
1184 while (p+8 < data + rec.data_len) {
1190 memcpy(&len, p+4, 4);
1192 if (tdb->methods->tdb_write(tdb, ofs, p+8, len) == -1) {
1194 TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction_recover: failed to recover %d bytes at offset %d\n", len, ofs));
1195 tdb->ecode = TDB_ERR_IO;
1203 if (transaction_sync(tdb, 0, tdb->map_size) == -1) {
1204 TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction_recover: failed to sync recovery\n"));
1205 tdb->ecode = TDB_ERR_IO;
1209 /* if the recovery area is after the recovered eof then remove it */
1210 if (recovery_eof <= recovery_head) {
1211 if (tdb_ofs_write(tdb, TDB_RECOVERY_HEAD, &zero) == -1) {
1212 TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction_recover: failed to remove recovery head\n"));
1213 tdb->ecode = TDB_ERR_IO;
1218 /* remove the recovery magic */
1219 if (tdb_ofs_write(tdb, recovery_head + offsetof(struct list_struct, magic),
1221 TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction_recover: failed to remove recovery magic\n"));
1222 tdb->ecode = TDB_ERR_IO;
1226 /* reduce the file size to the old size */
1228 if (ftruncate(tdb->fd, recovery_eof) != 0) {
1229 TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction_recover: failed to reduce to recovery size\n"));
1230 tdb->ecode = TDB_ERR_IO;
1233 tdb->map_size = recovery_eof;
1236 if (transaction_sync(tdb, 0, recovery_eof) == -1) {
1237 TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_transaction_recover: failed to sync2 recovery\n"));
1238 tdb->ecode = TDB_ERR_IO;
1242 TDB_LOG((tdb, TDB_DEBUG_TRACE, "tdb_transaction_recover: recovered %d byte database\n",