Use -O not -O3: reduces ccan/tdb test time from 24 to 18 seconds.
[ccan] / ccan / tdb / tdb.c
1  /* 
2    Unix SMB/CIFS implementation.
3
4    trivial database library
5
6    Copyright (C) Andrew Tridgell              1999-2005
7    Copyright (C) Paul `Rusty' Russell              2000
8    Copyright (C) Jeremy Allison                    2000-2003
9    
10      ** NOTE! The following LGPL license applies to the tdb
11      ** library. This does NOT imply that all of Samba is released
12      ** under the LGPL
13    
14    This library is free software; you can redistribute it and/or
15    modify it under the terms of the GNU Lesser General Public
16    License as published by the Free Software Foundation; either
17    version 3 of the License, or (at your option) any later version.
18
19    This library is distributed in the hope that it will be useful,
20    but WITHOUT ANY WARRANTY; without even the implied warranty of
21    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
22    Lesser General Public License for more details.
23
24    You should have received a copy of the GNU Lesser General Public
25    License along with this library; if not, see <http://www.gnu.org/licenses/>.
26 */
27
28 #include "tdb_private.h"
29
30 TDB_DATA tdb_null;
31
32 /*
33   non-blocking increment of the tdb sequence number if the tdb has been opened using
34   the TDB_SEQNUM flag
35 */
36 void tdb_increment_seqnum_nonblock(struct tdb_context *tdb)
37 {
38         tdb_off_t seqnum=0;
39         
40         if (!(tdb->flags & TDB_SEQNUM)) {
41                 return;
42         }
43
44         /* we ignore errors from this, as we have no sane way of
45            dealing with them.
46         */
47         tdb_ofs_read(tdb, TDB_SEQNUM_OFS, &seqnum);
48         seqnum++;
49         tdb_ofs_write(tdb, TDB_SEQNUM_OFS, &seqnum);
50 }
51
52 /*
53   increment the tdb sequence number if the tdb has been opened using
54   the TDB_SEQNUM flag
55 */
56 static void tdb_increment_seqnum(struct tdb_context *tdb)
57 {
58         if (!(tdb->flags & TDB_SEQNUM)) {
59                 return;
60         }
61
62         if (tdb_brlock(tdb, F_WRLCK, TDB_SEQNUM_OFS, 1,
63                        TDB_LOCK_WAIT|TDB_LOCK_PROBE) != 0) {
64                 return;
65         }
66
67         tdb_increment_seqnum_nonblock(tdb);
68
69         tdb_brunlock(tdb, F_WRLCK, TDB_SEQNUM_OFS, 1);
70 }
71
72 static int tdb_key_compare(TDB_DATA key, TDB_DATA data, void *private_data)
73 {
74         return memcmp(data.dptr, key.dptr, data.dsize);
75 }
76
77 /* Returns 0 on fail.  On success, return offset of record, and fills
78    in rec */
79 static tdb_off_t tdb_find(struct tdb_context *tdb, TDB_DATA key, uint32_t hash,
80                         struct list_struct *r)
81 {
82         tdb_off_t rec_ptr;
83         
84         /* read in the hash top */
85         if (tdb_ofs_read(tdb, TDB_HASH_TOP(hash), &rec_ptr) == -1)
86                 return 0;
87
88         /* keep looking until we find the right record */
89         while (rec_ptr) {
90                 if (tdb_rec_read(tdb, rec_ptr, r) == -1)
91                         return 0;
92
93                 if (!TDB_DEAD(r) && hash==r->full_hash
94                     && key.dsize==r->key_len
95                     && tdb_parse_data(tdb, key, rec_ptr + sizeof(*r),
96                                       r->key_len, tdb_key_compare,
97                                       NULL) == 0) {
98                         return rec_ptr;
99                 }
100                 /* detect tight infinite loop */
101                 if (rec_ptr == r->next) {
102                         tdb->ecode = TDB_ERR_CORRUPT;
103                         TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_find: loop detected.\n"));
104                         return 0;
105                 }
106                 rec_ptr = r->next;
107         }
108         tdb->ecode = TDB_ERR_NOEXIST;
109         return 0;
110 }
111
112 /* As tdb_find, but if you succeed, keep the lock */
113 tdb_off_t tdb_find_lock_hash(struct tdb_context *tdb, TDB_DATA key, uint32_t hash, int locktype,
114                            struct list_struct *rec)
115 {
116         uint32_t rec_ptr;
117
118         if (tdb_lock(tdb, BUCKET(hash), locktype) == -1)
119                 return 0;
120         if (!(rec_ptr = tdb_find(tdb, key, hash, rec)))
121                 tdb_unlock(tdb, BUCKET(hash), locktype);
122         return rec_ptr;
123 }
124
125
126 /* update an entry in place - this only works if the new data size
127    is <= the old data size and the key exists.
128    on failure return -1.
129 */
130 static int tdb_update_hash(struct tdb_context *tdb, TDB_DATA key, uint32_t hash, TDB_DATA dbuf)
131 {
132         struct list_struct rec;
133         tdb_off_t rec_ptr;
134
135         /* find entry */
136         if (!(rec_ptr = tdb_find(tdb, key, hash, &rec)))
137                 return -1;
138
139         /* must be long enough key, data and tailer */
140         if (rec.rec_len < key.dsize + dbuf.dsize + sizeof(tdb_off_t)) {
141                 tdb->ecode = TDB_SUCCESS; /* Not really an error */
142                 return -1;
143         }
144
145         if (tdb->methods->tdb_write(tdb, rec_ptr + sizeof(rec) + rec.key_len,
146                       dbuf.dptr, dbuf.dsize) == -1)
147                 return -1;
148
149         if (dbuf.dsize != rec.data_len) {
150                 /* update size */
151                 rec.data_len = dbuf.dsize;
152                 return tdb_rec_write(tdb, rec_ptr, &rec);
153         }
154  
155         return 0;
156 }
157
158 /* find an entry in the database given a key */
159 /* If an entry doesn't exist tdb_err will be set to
160  * TDB_ERR_NOEXIST. If a key has no data attached
161  * then the TDB_DATA will have zero length but
162  * a non-zero pointer
163  */
164 static TDB_DATA _tdb_fetch(struct tdb_context *tdb, TDB_DATA key)
165 {
166         tdb_off_t rec_ptr;
167         struct list_struct rec;
168         TDB_DATA ret;
169         uint32_t hash;
170
171         /* find which hash bucket it is in */
172         hash = tdb->hash_fn(&key);
173         if (!(rec_ptr = tdb_find_lock_hash(tdb,key,hash,F_RDLCK,&rec))) {
174                 return tdb_null;
175         }
176         ret.dptr = tdb_alloc_read(tdb, rec_ptr + sizeof(rec) + rec.key_len,
177                                   rec.data_len);
178         ret.dsize = rec.data_len;
179         tdb_unlock(tdb, BUCKET(rec.full_hash), F_RDLCK);
180         return ret;
181 }
182
183 TDB_DATA tdb_fetch(struct tdb_context *tdb, TDB_DATA key)
184 {
185         TDB_DATA ret = _tdb_fetch(tdb, key);
186
187         tdb_trace_1rec_retrec(tdb, "tdb_fetch", key, ret);
188         return ret;
189 }
190
191 /*
192  * Find an entry in the database and hand the record's data to a parsing
193  * function. The parsing function is executed under the chain read lock, so it
194  * should be fast and should not block on other syscalls.
195  *
196  * DONT CALL OTHER TDB CALLS FROM THE PARSER, THIS MIGHT LEAD TO SEGFAULTS.
197  *
198  * For mmapped tdb's that do not have a transaction open it points the parsing
199  * function directly at the mmap area, it avoids the malloc/memcpy in this
200  * case. If a transaction is open or no mmap is available, it has to do
201  * malloc/read/parse/free.
202  *
203  * This is interesting for all readers of potentially large data structures in
204  * the tdb records, ldb indexes being one example.
205  */
206
207 int tdb_parse_record(struct tdb_context *tdb, TDB_DATA key,
208                      int (*parser)(TDB_DATA key, TDB_DATA data,
209                                    void *private_data),
210                      void *private_data)
211 {
212         tdb_off_t rec_ptr;
213         struct list_struct rec;
214         int ret;
215         uint32_t hash;
216
217         /* find which hash bucket it is in */
218         hash = tdb->hash_fn(&key);
219
220         if (!(rec_ptr = tdb_find_lock_hash(tdb,key,hash,F_RDLCK,&rec))) {
221                 tdb_trace_1rec_ret(tdb, "tdb_parse_record", key,
222                                    -TDB_ERR_NOEXIST);
223                 tdb->ecode = TDB_ERR_NOEXIST;
224                 return 0;
225         }
226         tdb_trace_1rec_ret(tdb, "tdb_parse_record", key, 0);
227
228         ret = tdb_parse_data(tdb, key, rec_ptr + sizeof(rec) + rec.key_len,
229                              rec.data_len, parser, private_data);
230
231         tdb_unlock(tdb, BUCKET(rec.full_hash), F_RDLCK);
232
233         return ret;
234 }
235
236 /* check if an entry in the database exists 
237
238    note that 1 is returned if the key is found and 0 is returned if not found
239    this doesn't match the conventions in the rest of this module, but is
240    compatible with gdbm
241 */
242 static int tdb_exists_hash(struct tdb_context *tdb, TDB_DATA key, uint32_t hash)
243 {
244         struct list_struct rec;
245         
246         if (tdb_find_lock_hash(tdb, key, hash, F_RDLCK, &rec) == 0)
247                 return 0;
248         tdb_unlock(tdb, BUCKET(rec.full_hash), F_RDLCK);
249         return 1;
250 }
251
252 int tdb_exists(struct tdb_context *tdb, TDB_DATA key)
253 {
254         uint32_t hash = tdb->hash_fn(&key);
255         int ret;
256
257         ret = tdb_exists_hash(tdb, key, hash);
258         tdb_trace_1rec_ret(tdb, "tdb_exists", key, ret);
259         return ret;
260 }
261
262 /* actually delete an entry in the database given the offset */
263 int tdb_do_delete(struct tdb_context *tdb, tdb_off_t rec_ptr, struct list_struct *rec)
264 {
265         tdb_off_t last_ptr, i;
266         struct list_struct lastrec;
267
268         if (tdb->read_only || tdb->traverse_read) return -1;
269
270         if (tdb->traverse_write != 0 || 
271             tdb_write_lock_record(tdb, rec_ptr) == -1) {
272                 /* Someone traversing here: mark it as dead */
273                 rec->magic = TDB_DEAD_MAGIC;
274                 return tdb_rec_write(tdb, rec_ptr, rec);
275         }
276         if (tdb_write_unlock_record(tdb, rec_ptr) != 0)
277                 return -1;
278
279         /* find previous record in hash chain */
280         if (tdb_ofs_read(tdb, TDB_HASH_TOP(rec->full_hash), &i) == -1)
281                 return -1;
282         for (last_ptr = 0; i != rec_ptr; last_ptr = i, i = lastrec.next)
283                 if (tdb_rec_read(tdb, i, &lastrec) == -1)
284                         return -1;
285
286         /* unlink it: next ptr is at start of record. */
287         if (last_ptr == 0)
288                 last_ptr = TDB_HASH_TOP(rec->full_hash);
289         if (tdb_ofs_write(tdb, last_ptr, &rec->next) == -1)
290                 return -1;
291
292         /* recover the space */
293         if (tdb_free(tdb, rec_ptr, rec) == -1)
294                 return -1;
295         return 0;
296 }
297
298 static int tdb_count_dead(struct tdb_context *tdb, uint32_t hash)
299 {
300         int res = 0;
301         tdb_off_t rec_ptr;
302         struct list_struct rec;
303         
304         /* read in the hash top */
305         if (tdb_ofs_read(tdb, TDB_HASH_TOP(hash), &rec_ptr) == -1)
306                 return 0;
307
308         while (rec_ptr) {
309                 if (tdb_rec_read(tdb, rec_ptr, &rec) == -1)
310                         return 0;
311
312                 if (rec.magic == TDB_DEAD_MAGIC) {
313                         res += 1;
314                 }
315                 rec_ptr = rec.next;
316         }
317         return res;
318 }
319
320 /*
321  * Purge all DEAD records from a hash chain
322  */
323 static int tdb_purge_dead(struct tdb_context *tdb, uint32_t hash)
324 {
325         int res = -1;
326         struct list_struct rec;
327         tdb_off_t rec_ptr;
328
329         if (tdb_lock(tdb, -1, F_WRLCK) == -1) {
330                 return -1;
331         }
332         
333         /* read in the hash top */
334         if (tdb_ofs_read(tdb, TDB_HASH_TOP(hash), &rec_ptr) == -1)
335                 goto fail;
336
337         while (rec_ptr) {
338                 tdb_off_t next;
339
340                 if (tdb_rec_read(tdb, rec_ptr, &rec) == -1) {
341                         goto fail;
342                 }
343
344                 next = rec.next;
345
346                 if (rec.magic == TDB_DEAD_MAGIC
347                     && tdb_do_delete(tdb, rec_ptr, &rec) == -1) {
348                         goto fail;
349                 }
350                 rec_ptr = next;
351         }
352         res = 0;
353  fail:
354         tdb_unlock(tdb, -1, F_WRLCK);
355         return res;
356 }
357
358 /* delete an entry in the database given a key */
359 static int tdb_delete_hash(struct tdb_context *tdb, TDB_DATA key, uint32_t hash)
360 {
361         tdb_off_t rec_ptr;
362         struct list_struct rec;
363         int ret;
364
365         if (tdb->max_dead_records != 0) {
366
367                 /*
368                  * Allow for some dead records per hash chain, mainly for
369                  * tdb's with a very high create/delete rate like locking.tdb.
370                  */
371
372                 if (tdb_lock(tdb, BUCKET(hash), F_WRLCK) == -1)
373                         return -1;
374
375                 if (tdb_count_dead(tdb, hash) >= tdb->max_dead_records) {
376                         /*
377                          * Don't let the per-chain freelist grow too large,
378                          * delete all existing dead records
379                          */
380                         tdb_purge_dead(tdb, hash);
381                 }
382
383                 if (!(rec_ptr = tdb_find(tdb, key, hash, &rec))) {
384                         tdb_unlock(tdb, BUCKET(hash), F_WRLCK);
385                         return -1;
386                 }
387
388                 /*
389                  * Just mark the record as dead.
390                  */
391                 rec.magic = TDB_DEAD_MAGIC;
392                 ret = tdb_rec_write(tdb, rec_ptr, &rec);
393         }
394         else {
395                 if (!(rec_ptr = tdb_find_lock_hash(tdb, key, hash, F_WRLCK,
396                                                    &rec)))
397                         return -1;
398
399                 ret = tdb_do_delete(tdb, rec_ptr, &rec);
400         }
401
402         if (ret == 0) {
403                 tdb_increment_seqnum(tdb);
404         }
405
406         if (tdb_unlock(tdb, BUCKET(rec.full_hash), F_WRLCK) != 0)
407                 TDB_LOG((tdb, TDB_DEBUG_WARNING, "tdb_delete: WARNING tdb_unlock failed!\n"));
408         return ret;
409 }
410
411 int tdb_delete(struct tdb_context *tdb, TDB_DATA key)
412 {
413         uint32_t hash = tdb->hash_fn(&key);
414         int ret;
415
416         ret = tdb_delete_hash(tdb, key, hash);
417         tdb_trace_1rec_ret(tdb, "tdb_delete", key, ret);
418         return ret;
419 }
420
421 /*
422  * See if we have a dead record around with enough space
423  */
424 static tdb_off_t tdb_find_dead(struct tdb_context *tdb, uint32_t hash,
425                                struct list_struct *r, tdb_len_t length)
426 {
427         tdb_off_t rec_ptr;
428         
429         /* read in the hash top */
430         if (tdb_ofs_read(tdb, TDB_HASH_TOP(hash), &rec_ptr) == -1)
431                 return 0;
432
433         /* keep looking until we find the right record */
434         while (rec_ptr) {
435                 if (tdb_rec_read(tdb, rec_ptr, r) == -1)
436                         return 0;
437
438                 if (TDB_DEAD(r) && r->rec_len >= length) {
439                         /*
440                          * First fit for simple coding, TODO: change to best
441                          * fit
442                          */
443                         return rec_ptr;
444                 }
445                 rec_ptr = r->next;
446         }
447         return 0;
448 }
449
450 static int _tdb_store(struct tdb_context *tdb, TDB_DATA key,
451                       TDB_DATA dbuf, int flag, uint32_t hash)
452 {
453         struct list_struct rec;
454         tdb_off_t rec_ptr;
455         char *p = NULL;
456         int ret = -1;
457
458         /* check for it existing, on insert. */
459         if (flag == TDB_INSERT) {
460                 if (tdb_exists_hash(tdb, key, hash)) {
461                         tdb->ecode = TDB_ERR_EXISTS;
462                         goto fail;
463                 }
464         } else {
465                 /* first try in-place update, on modify or replace. */
466                 if (tdb_update_hash(tdb, key, hash, dbuf) == 0) {
467                         goto done;
468                 }
469                 if (tdb->ecode == TDB_ERR_NOEXIST &&
470                     flag == TDB_MODIFY) {
471                         /* if the record doesn't exist and we are in TDB_MODIFY mode then
472                          we should fail the store */
473                         goto fail;
474                 }
475         }
476         /* reset the error code potentially set by the tdb_update() */
477         tdb->ecode = TDB_SUCCESS;
478
479         /* delete any existing record - if it doesn't exist we don't
480            care.  Doing this first reduces fragmentation, and avoids
481            coalescing with `allocated' block before it's updated. */
482         if (flag != TDB_INSERT)
483                 tdb_delete_hash(tdb, key, hash);
484
485         /* Copy key+value *before* allocating free space in case malloc
486            fails and we are left with a dead spot in the tdb. */
487
488         if (!(p = (char *)malloc(key.dsize + dbuf.dsize))) {
489                 tdb->ecode = TDB_ERR_OOM;
490                 goto fail;
491         }
492
493         memcpy(p, key.dptr, key.dsize);
494         if (dbuf.dsize)
495                 memcpy(p+key.dsize, dbuf.dptr, dbuf.dsize);
496
497         if (tdb->max_dead_records != 0) {
498                 /*
499                  * Allow for some dead records per hash chain, look if we can
500                  * find one that can hold the new record. We need enough space
501                  * for key, data and tailer. If we find one, we don't have to
502                  * consult the central freelist.
503                  */
504                 rec_ptr = tdb_find_dead(
505                         tdb, hash, &rec,
506                         key.dsize + dbuf.dsize + sizeof(tdb_off_t));
507
508                 if (rec_ptr != 0) {
509                         rec.key_len = key.dsize;
510                         rec.data_len = dbuf.dsize;
511                         rec.full_hash = hash;
512                         rec.magic = TDB_MAGIC;
513                         if (tdb_rec_write(tdb, rec_ptr, &rec) == -1
514                             || tdb->methods->tdb_write(
515                                     tdb, rec_ptr + sizeof(rec),
516                                     p, key.dsize + dbuf.dsize) == -1) {
517                                 goto fail;
518                         }
519                         goto done;
520                 }
521         }
522
523         /*
524          * We have to allocate some space from the freelist, so this means we
525          * have to lock it. Use the chance to purge all the DEAD records from
526          * the hash chain under the freelist lock.
527          */
528
529         if (tdb_lock(tdb, -1, F_WRLCK) == -1) {
530                 goto fail;
531         }
532
533         if ((tdb->max_dead_records != 0)
534             && (tdb_purge_dead(tdb, hash) == -1)) {
535                 tdb_unlock(tdb, -1, F_WRLCK);
536                 goto fail;
537         }
538
539         /* we have to allocate some space */
540         rec_ptr = tdb_allocate(tdb, key.dsize + dbuf.dsize, &rec);
541
542         tdb_unlock(tdb, -1, F_WRLCK);
543
544         if (rec_ptr == 0) {
545                 goto fail;
546         }
547
548         /* Read hash top into next ptr */
549         if (tdb_ofs_read(tdb, TDB_HASH_TOP(hash), &rec.next) == -1)
550                 goto fail;
551
552         rec.key_len = key.dsize;
553         rec.data_len = dbuf.dsize;
554         rec.full_hash = hash;
555         rec.magic = TDB_MAGIC;
556
557         /* write out and point the top of the hash chain at it */
558         if (tdb_rec_write(tdb, rec_ptr, &rec) == -1
559             || tdb->methods->tdb_write(tdb, rec_ptr+sizeof(rec), p, key.dsize+dbuf.dsize)==-1
560             || tdb_ofs_write(tdb, TDB_HASH_TOP(hash), &rec_ptr) == -1) {
561                 /* Need to tdb_unallocate() here */
562                 goto fail;
563         }
564
565  done:
566         ret = 0;
567  fail:
568         if (ret == 0) {
569                 tdb_increment_seqnum(tdb);
570         }
571
572         SAFE_FREE(p); 
573         return ret;
574 }
575
576 /* store an element in the database, replacing any existing element
577    with the same key 
578
579    return 0 on success, -1 on failure
580 */
581 int tdb_store(struct tdb_context *tdb, TDB_DATA key, TDB_DATA dbuf, int flag)
582 {
583         uint32_t hash;
584         int ret;
585
586         if (tdb->read_only || tdb->traverse_read) {
587                 tdb->ecode = TDB_ERR_RDONLY;
588                 tdb_trace_2rec_flag_ret(tdb, "tdb_store", key, dbuf, flag,
589                                         -TDB_ERR_RDONLY);
590                 return -1;
591         }
592
593         /* find which hash bucket it is in */
594         hash = tdb->hash_fn(&key);
595         if (tdb_lock(tdb, BUCKET(hash), F_WRLCK) == -1)
596                 return -1;
597
598         ret = _tdb_store(tdb, key, dbuf, flag, hash);
599         tdb_trace_2rec_flag_ret(tdb, "tdb_store", key, dbuf, flag, ret);
600         tdb_unlock(tdb, BUCKET(hash), F_WRLCK);
601         return ret;
602 }
603
604
605 /* Append to an entry. Create if not exist. */
606 int tdb_append(struct tdb_context *tdb, TDB_DATA key, TDB_DATA new_dbuf)
607 {
608         uint32_t hash;
609         TDB_DATA dbuf;
610         int ret = -1;
611
612         /* find which hash bucket it is in */
613         hash = tdb->hash_fn(&key);
614         if (tdb_lock(tdb, BUCKET(hash), F_WRLCK) == -1)
615                 return -1;
616
617         dbuf = _tdb_fetch(tdb, key);
618
619         if (dbuf.dptr == NULL) {
620                 dbuf.dptr = (unsigned char *)malloc(new_dbuf.dsize);
621         } else {
622                 unsigned int new_len = dbuf.dsize + new_dbuf.dsize;
623                 unsigned char *new_dptr;
624
625                 /* realloc '0' is special: don't do that. */
626                 if (new_len == 0)
627                         new_len = 1;
628                 new_dptr = (unsigned char *)realloc(dbuf.dptr, new_len);
629                 if (new_dptr == NULL) {
630                         free(dbuf.dptr);
631                 }
632                 dbuf.dptr = new_dptr;
633         }
634
635         if (dbuf.dptr == NULL) {
636                 tdb->ecode = TDB_ERR_OOM;
637                 goto failed;
638         }
639
640         memcpy(dbuf.dptr + dbuf.dsize, new_dbuf.dptr, new_dbuf.dsize);
641         dbuf.dsize += new_dbuf.dsize;
642
643         ret = _tdb_store(tdb, key, dbuf, 0, hash);
644         tdb_trace_2rec_retrec(tdb, "tdb_append", key, new_dbuf, dbuf);
645         
646 failed:
647         tdb_unlock(tdb, BUCKET(hash), F_WRLCK);
648         SAFE_FREE(dbuf.dptr);
649         return ret;
650 }
651
652
653 /*
654   return the name of the current tdb file
655   useful for external logging functions
656 */
657 const char *tdb_name(struct tdb_context *tdb)
658 {
659         return tdb->name;
660 }
661
662 /*
663   return the underlying file descriptor being used by tdb, or -1
664   useful for external routines that want to check the device/inode
665   of the fd
666 */
667 int tdb_fd(struct tdb_context *tdb)
668 {
669         return tdb->fd;
670 }
671
672 /*
673   return the current logging function
674   useful for external tdb routines that wish to log tdb errors
675 */
676 tdb_log_func tdb_log_fn(struct tdb_context *tdb)
677 {
678         return tdb->log.log_fn;
679 }
680
681
682 /*
683   get the tdb sequence number. Only makes sense if the writers opened
684   with TDB_SEQNUM set. Note that this sequence number will wrap quite
685   quickly, so it should only be used for a 'has something changed'
686   test, not for code that relies on the count of the number of changes
687   made. If you want a counter then use a tdb record.
688
689   The aim of this sequence number is to allow for a very lightweight
690   test of a possible tdb change.
691 */
692 int tdb_get_seqnum(struct tdb_context *tdb)
693 {
694         tdb_off_t seqnum=0;
695
696         tdb_ofs_read(tdb, TDB_SEQNUM_OFS, &seqnum);
697         tdb_trace_ret(tdb, "tdb_get_seqnum", seqnum);
698         return seqnum;
699 }
700
701 int tdb_hash_size(struct tdb_context *tdb)
702 {
703         return tdb->header.hash_size;
704 }
705
706 size_t tdb_map_size(struct tdb_context *tdb)
707 {
708         return tdb->map_size;
709 }
710
711 int tdb_get_flags(struct tdb_context *tdb)
712 {
713         return tdb->flags;
714 }
715
716 void tdb_add_flags(struct tdb_context *tdb, unsigned flags)
717 {
718         tdb->flags |= flags;
719 }
720
721 void tdb_remove_flags(struct tdb_context *tdb, unsigned flags)
722 {
723         tdb->flags &= ~flags;
724 }
725
726
727 /*
728   enable sequence number handling on an open tdb
729 */
730 void tdb_enable_seqnum(struct tdb_context *tdb)
731 {
732         tdb->flags |= TDB_SEQNUM;
733 }
734
735
736 /*
737   add a region of the file to the freelist. Length is the size of the region in bytes, 
738   which includes the free list header that needs to be added
739  */
740 static int tdb_free_region(struct tdb_context *tdb, tdb_off_t offset, ssize_t length)
741 {
742         struct list_struct rec;
743         if (length <= sizeof(rec)) {
744                 /* the region is not worth adding */
745                 return 0;
746         }
747         if (length + offset > tdb->map_size) {
748                 TDB_LOG((tdb, TDB_DEBUG_FATAL,"tdb_free_region: adding region beyond end of file\n"));
749                 return -1;              
750         }
751         memset(&rec,'\0',sizeof(rec));
752         rec.rec_len = length - sizeof(rec);
753         if (tdb_free(tdb, offset, &rec) == -1) {
754                 TDB_LOG((tdb, TDB_DEBUG_FATAL,"tdb_free_region: failed to add free record\n"));
755                 return -1;
756         }
757         return 0;
758 }
759
760 /*
761   wipe the entire database, deleting all records. This can be done
762   very fast by using a global lock. The entire data portion of the
763   file becomes a single entry in the freelist.
764
765   This code carefully steps around the recovery area, leaving it alone
766  */
767 int tdb_wipe_all(struct tdb_context *tdb)
768 {
769         int i;
770         tdb_off_t offset = 0;
771         ssize_t data_len;
772         tdb_off_t recovery_head;
773         tdb_len_t recovery_size = 0;
774
775         if (tdb_lockall(tdb) != 0) {
776                 return -1;
777         }
778
779         tdb_trace(tdb, "tdb_wipe_all");
780
781         /* see if the tdb has a recovery area, and remember its size
782            if so. We don't want to lose this as otherwise each
783            tdb_wipe_all() in a transaction will increase the size of
784            the tdb by the size of the recovery area */
785         if (tdb_ofs_read(tdb, TDB_RECOVERY_HEAD, &recovery_head) == -1) {
786                 TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_wipe_all: failed to read recovery head\n"));
787                 goto failed;
788         }
789
790         if (recovery_head != 0) {
791                 struct list_struct rec;
792                 if (tdb->methods->tdb_read(tdb, recovery_head, &rec, sizeof(rec), DOCONV()) == -1) {
793                         TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_wipe_all: failed to read recovery record\n"));
794                         return -1;
795                 }       
796                 recovery_size = rec.rec_len + sizeof(rec);
797         }
798
799         /* wipe the hashes */
800         for (i=0;i<tdb->header.hash_size;i++) {
801                 if (tdb_ofs_write(tdb, TDB_HASH_TOP(i), &offset) == -1) {
802                         TDB_LOG((tdb, TDB_DEBUG_FATAL,"tdb_wipe_all: failed to write hash %d\n", i));
803                         goto failed;
804                 }
805         }
806
807         /* wipe the freelist */
808         if (tdb_ofs_write(tdb, FREELIST_TOP, &offset) == -1) {
809                 TDB_LOG((tdb, TDB_DEBUG_FATAL,"tdb_wipe_all: failed to write freelist\n"));
810                 goto failed;
811         }
812
813         /* add all the rest of the file to the freelist, possibly leaving a gap 
814            for the recovery area */
815         if (recovery_size == 0) {
816                 /* the simple case - the whole file can be used as a freelist */
817                 data_len = (tdb->map_size - TDB_DATA_START(tdb->header.hash_size));
818                 if (tdb_free_region(tdb, TDB_DATA_START(tdb->header.hash_size), data_len) != 0) {
819                         goto failed;
820                 }
821         } else {
822                 /* we need to add two freelist entries - one on either
823                    side of the recovery area 
824
825                    Note that we cannot shift the recovery area during
826                    this operation. Only the transaction.c code may
827                    move the recovery area or we risk subtle data
828                    corruption
829                 */
830                 data_len = (recovery_head - TDB_DATA_START(tdb->header.hash_size));
831                 if (tdb_free_region(tdb, TDB_DATA_START(tdb->header.hash_size), data_len) != 0) {
832                         goto failed;
833                 }
834                 /* and the 2nd free list entry after the recovery area - if any */
835                 data_len = tdb->map_size - (recovery_head+recovery_size);
836                 if (tdb_free_region(tdb, recovery_head+recovery_size, data_len) != 0) {
837                         goto failed;
838                 }
839         }
840
841         if (tdb_unlockall(tdb) != 0) {
842                 TDB_LOG((tdb, TDB_DEBUG_FATAL,"tdb_wipe_all: failed to unlock\n"));
843                 goto failed;
844         }
845
846         return 0;
847
848 failed:
849         tdb_unlockall(tdb);
850         return -1;
851 }
852
853
854 struct traverse_state {
855         bool error;
856         struct tdb_context *dest_db;
857 };
858
859 /*
860   traverse function for repacking
861  */
862 static int repack_traverse(struct tdb_context *tdb, TDB_DATA key, TDB_DATA data, void *private)
863 {
864         struct traverse_state *state = (struct traverse_state *)private;
865         if (tdb_store(state->dest_db, key, data, TDB_INSERT) != 0) {
866                 state->error = true;
867                 return -1;
868         }
869         return 0;
870 }
871
872 /*
873   repack a tdb
874  */
875 int tdb_repack(struct tdb_context *tdb)
876 {
877         struct tdb_context *tmp_db;
878         struct traverse_state state;
879
880         tdb_trace(tdb, "tdb_repack");
881
882         if (tdb_transaction_start(tdb) != 0) {
883                 TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_repack: Failed to start transaction\n"));
884                 return -1;
885         }
886
887         tmp_db = tdb_open("tmpdb", tdb_hash_size(tdb), TDB_INTERNAL, O_RDWR|O_CREAT, 0);
888         if (tmp_db == NULL) {
889                 TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_repack: Failed to create tmp_db\n"));
890                 tdb_transaction_cancel(tdb);
891                 return -1;
892         }
893
894         state.error = false;
895         state.dest_db = tmp_db;
896
897         if (tdb_traverse_read(tdb, repack_traverse, &state) == -1) {
898                 TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_repack: Failed to traverse copying out\n"));
899                 tdb_transaction_cancel(tdb);
900                 tdb_close(tmp_db);
901                 return -1;              
902         }
903
904         if (state.error) {
905                 TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_repack: Error during traversal\n"));
906                 tdb_transaction_cancel(tdb);
907                 tdb_close(tmp_db);
908                 return -1;
909         }
910
911         if (tdb_wipe_all(tdb) != 0) {
912                 TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_repack: Failed to wipe database\n"));
913                 tdb_transaction_cancel(tdb);
914                 tdb_close(tmp_db);
915                 return -1;
916         }
917
918         state.error = false;
919         state.dest_db = tdb;
920
921         if (tdb_traverse_read(tmp_db, repack_traverse, &state) == -1) {
922                 TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_repack: Failed to traverse copying back\n"));
923                 tdb_transaction_cancel(tdb);
924                 tdb_close(tmp_db);
925                 return -1;              
926         }
927
928         if (state.error) {
929                 TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_repack: Error during second traversal\n"));
930                 tdb_transaction_cancel(tdb);
931                 tdb_close(tmp_db);
932                 return -1;
933         }
934
935         tdb_close(tmp_db);
936
937         if (tdb_transaction_commit(tdb) != 0) {
938                 TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_repack: Failed to commit\n"));
939                 return -1;
940         }
941
942         return 0;
943 }
944
945 #ifdef TDB_TRACE
946 static void tdb_trace_write(struct tdb_context *tdb, const char *str)
947 {
948         if (write(tdb->tracefd, str, strlen(str)) != strlen(str)) {
949                 close(tdb->tracefd);
950                 tdb->tracefd = -1;
951         }
952 }
953
954 static void tdb_trace_start(struct tdb_context *tdb)
955 {
956         tdb_off_t seqnum=0;
957         char msg[sizeof(tdb_off_t) * 4];
958
959         tdb_ofs_read(tdb, TDB_SEQNUM_OFS, &seqnum);
960         sprintf(msg, "%u ", seqnum);
961         tdb_trace_write(tdb, msg);
962 }
963
964 static void tdb_trace_end(struct tdb_context *tdb)
965 {
966         tdb_trace_write(tdb, "\n");
967 }
968
969 static void tdb_trace_end_ret(struct tdb_context *tdb, int ret)
970 {
971         char msg[sizeof(ret) * 4];
972         sprintf(msg, " = %i\n", ret);
973         tdb_trace_write(tdb, msg);
974 }
975
976 static void tdb_trace_record(struct tdb_context *tdb, TDB_DATA rec)
977 {
978         char msg[20 + rec.dsize*2], *p;
979         unsigned int i;
980
981         /* We differentiate zero-length records from non-existent ones. */
982         if (rec.dptr == NULL) {
983                 tdb_trace_write(tdb, " NULL");
984                 return;
985         }
986
987         p = msg;
988         p += sprintf(p, " %zu:", rec.dsize);
989         for (i = 0; i < rec.dsize; i++)
990                 p += sprintf(p, "%02x", rec.dptr[i]);
991
992         tdb_trace_write(tdb, msg);
993 }
994
995 void tdb_trace(struct tdb_context *tdb, const char *op)
996 {
997         tdb_trace_start(tdb);
998         tdb_trace_write(tdb, op);
999         tdb_trace_end(tdb);
1000 }
1001
1002 void tdb_trace_seqnum(struct tdb_context *tdb, uint32_t seqnum, const char *op)
1003 {
1004         char msg[sizeof(tdb_off_t) * 4];
1005
1006         sprintf(msg, "%u ", seqnum);
1007         tdb_trace_write(tdb, msg);
1008         tdb_trace_write(tdb, op);
1009         tdb_trace_end(tdb);
1010 }
1011
1012 void tdb_trace_open(struct tdb_context *tdb, const char *op,
1013                     unsigned hash_size, unsigned tdb_flags, unsigned open_flags)
1014 {
1015         char msg[128];
1016
1017         sprintf(msg, "%s %u %#x %#x", op, hash_size, tdb_flags, open_flags);
1018         tdb_trace_start(tdb);
1019         tdb_trace_write(tdb, msg);
1020         tdb_trace_end(tdb);
1021 }
1022
1023 void tdb_trace_ret(struct tdb_context *tdb, const char *op, int ret)
1024 {
1025         tdb_trace_start(tdb);
1026         tdb_trace_write(tdb, op);
1027         tdb_trace_end_ret(tdb, ret);
1028 }
1029
1030 void tdb_trace_retrec(struct tdb_context *tdb, const char *op, TDB_DATA ret)
1031 {
1032         tdb_trace_start(tdb);
1033         tdb_trace_write(tdb, op);
1034         tdb_trace_write(tdb, " =");
1035         tdb_trace_record(tdb, ret);
1036         tdb_trace_end(tdb);
1037 }
1038
1039 void tdb_trace_1rec(struct tdb_context *tdb, const char *op,
1040                     TDB_DATA rec)
1041 {
1042         tdb_trace_start(tdb);
1043         tdb_trace_write(tdb, op);
1044         tdb_trace_record(tdb, rec);
1045         tdb_trace_end(tdb);
1046 }
1047
1048 void tdb_trace_1rec_ret(struct tdb_context *tdb, const char *op,
1049                         TDB_DATA rec, int ret)
1050 {
1051         tdb_trace_start(tdb);
1052         tdb_trace_write(tdb, op);
1053         tdb_trace_record(tdb, rec);
1054         tdb_trace_end_ret(tdb, ret);
1055 }
1056
1057 void tdb_trace_1rec_retrec(struct tdb_context *tdb, const char *op,
1058                            TDB_DATA rec, TDB_DATA ret)
1059 {
1060         tdb_trace_start(tdb);
1061         tdb_trace_write(tdb, op);
1062         tdb_trace_record(tdb, rec);
1063         tdb_trace_write(tdb, " =");
1064         tdb_trace_record(tdb, ret);
1065         tdb_trace_end(tdb);
1066 }
1067
1068 void tdb_trace_2rec_flag_ret(struct tdb_context *tdb, const char *op,
1069                              TDB_DATA rec1, TDB_DATA rec2, unsigned flag,
1070                              int ret)
1071 {
1072         char msg[sizeof(ret) * 4];
1073
1074         sprintf(msg, " %#x", flag); 
1075         tdb_trace_start(tdb);
1076         tdb_trace_write(tdb, op);
1077         tdb_trace_record(tdb, rec1);
1078         tdb_trace_record(tdb, rec2);
1079         tdb_trace_write(tdb, msg);
1080         tdb_trace_end_ret(tdb, ret);
1081 }
1082
1083 void tdb_trace_2rec_retrec(struct tdb_context *tdb, const char *op,
1084                            TDB_DATA rec1, TDB_DATA rec2, TDB_DATA ret)
1085 {
1086         tdb_trace_start(tdb);
1087         tdb_trace_write(tdb, op);
1088         tdb_trace_record(tdb, rec1);
1089         tdb_trace_record(tdb, rec2);
1090         tdb_trace_write(tdb, " =");
1091         tdb_trace_record(tdb, ret);
1092         tdb_trace_end(tdb);
1093 }
1094 #endif