r4466: rather than defining "STANDALONE" for building tdb, ldb and talloc
[samba.git] / source4 / lib / tdb / common / tdb.c
1  /* 
2    Unix SMB/CIFS implementation.
3
4    trivial database library
5
6    Copyright (C) Andrew Tridgell              1999-2004
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 2 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, write to the Free Software
26    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
27 */
28
29
30 /* NOTE: If you use tdbs under valgrind, and in particular if you run
31  * tdbtorture, you may get spurious "uninitialized value" warnings.  I
32  * think this is because valgrind doesn't understand that the mmap'd
33  * area may be written to by other processes.  Memory can, from the
34  * point of view of the grinded process, spontaneously become
35  * initialized.
36  *
37  * I can think of a few solutions.  [mbp 20030311]
38  *
39  * 1 - Write suppressions for Valgrind so that it doesn't complain
40  * about this.  Probably the most reasonable but people need to
41  * remember to use them.
42  *
43  * 2 - Use IO not mmap when running under valgrind.  Not so nice.
44  *
45  * 3 - Use the special valgrind macros to mark memory as valid at the
46  * right time.  Probably too hard -- the process just doesn't know.
47  */ 
48
49 #ifndef _SAMBA_BUILD_
50 #if HAVE_CONFIG_H
51 #include <config.h>
52 #endif
53
54 #include <stdlib.h>
55 #include <stdio.h>
56 #include <stdint.h>
57 #include <fcntl.h>
58 #include <unistd.h>
59 #include <string.h>
60 #include <fcntl.h>
61 #include <errno.h>
62 #include <sys/mman.h>
63 #include <sys/stat.h>
64 #include <signal.h>
65 #include "tdb.h"
66 #include "spinlock.h"
67 #else
68 #include "includes.h"
69 #include "system/time.h"
70 #include "system/shmem.h"
71 #include "system/filesys.h"
72 #endif
73
74 #define TDB_MAGIC_FOOD "TDB file\n"
75 #define TDB_VERSION (0x26011967 + 6)
76 #define TDB_MAGIC (0x26011999U)
77 #define TDB_FREE_MAGIC (~TDB_MAGIC)
78 #define TDB_DEAD_MAGIC (0xFEE1DEAD)
79 #define TDB_ALIGNMENT 4
80 #define MIN_REC_SIZE (2*sizeof(struct list_struct) + TDB_ALIGNMENT)
81 #define DEFAULT_HASH_SIZE 131
82 #define TDB_PAGE_SIZE 0x2000
83 #define FREELIST_TOP (sizeof(struct tdb_header))
84 #define TDB_ALIGN(x,a) (((x) + (a)-1) & ~((a)-1))
85 #define TDB_BYTEREV(x) (((((x)&0xff)<<24)|((x)&0xFF00)<<8)|(((x)>>8)&0xFF00)|((x)>>24))
86 #define TDB_DEAD(r) ((r)->magic == TDB_DEAD_MAGIC)
87 #define TDB_BAD_MAGIC(r) ((r)->magic != TDB_MAGIC && !TDB_DEAD(r))
88 #define TDB_HASH_TOP(hash) (FREELIST_TOP + (BUCKET(hash)+1)*sizeof(tdb_off))
89 #define TDB_DATA_START(hash_size) (TDB_HASH_TOP(hash_size-1) + TDB_SPINLOCK_SIZE(hash_size))
90
91
92 /* NB assumes there is a local variable called "tdb" that is the
93  * current context, also takes doubly-parenthesized print-style
94  * argument. */
95 #define TDB_LOG(x) tdb->log_fn x
96
97 /* lock offsets */
98 #define GLOBAL_LOCK 0
99 #define ACTIVE_LOCK 4
100
101 #ifndef MAP_FILE
102 #define MAP_FILE 0
103 #endif
104
105 #ifndef MAP_FAILED
106 #define MAP_FAILED ((void *)-1)
107 #endif
108
109 /* free memory if the pointer is valid and zero the pointer */
110 #ifndef SAFE_FREE
111 #define SAFE_FREE(x) do { if ((x) != NULL) {free((x)); (x)=NULL;} } while(0)
112 #endif
113
114 #define BUCKET(hash) ((hash) % tdb->header.hash_size)
115 TDB_DATA tdb_null;
116
117 /* all contexts, to ensure no double-opens (fcntl locks don't nest!) */
118 static TDB_CONTEXT *tdbs = NULL;
119
120 static int tdb_munmap(TDB_CONTEXT *tdb)
121 {
122         if (tdb->flags & TDB_INTERNAL)
123                 return 0;
124
125 #ifdef HAVE_MMAP
126         if (tdb->map_ptr) {
127                 int ret = munmap(tdb->map_ptr, tdb->map_size);
128                 if (ret != 0)
129                         return ret;
130         }
131 #endif
132         tdb->map_ptr = NULL;
133         return 0;
134 }
135
136 static void tdb_mmap(TDB_CONTEXT *tdb)
137 {
138         if (tdb->flags & TDB_INTERNAL)
139                 return;
140
141 #ifdef HAVE_MMAP
142         if (!(tdb->flags & TDB_NOMMAP)) {
143                 tdb->map_ptr = mmap(NULL, tdb->map_size, 
144                                     PROT_READ|(tdb->read_only? 0:PROT_WRITE), 
145                                     MAP_SHARED|MAP_FILE, tdb->fd, 0);
146
147                 /*
148                  * NB. When mmap fails it returns MAP_FAILED *NOT* NULL !!!!
149                  */
150
151                 if (tdb->map_ptr == MAP_FAILED) {
152                         tdb->map_ptr = NULL;
153                         TDB_LOG((tdb, 2, "tdb_mmap failed for size %d (%s)\n", 
154                                  tdb->map_size, strerror(errno)));
155                 }
156         } else {
157                 tdb->map_ptr = NULL;
158         }
159 #else
160         tdb->map_ptr = NULL;
161 #endif
162 }
163
164 /* Endian conversion: we only ever deal with 4 byte quantities */
165 static void *convert(void *buf, u32 size)
166 {
167         u32 i, *p = buf;
168         for (i = 0; i < size / 4; i++)
169                 p[i] = TDB_BYTEREV(p[i]);
170         return buf;
171 }
172 #define DOCONV() (tdb->flags & TDB_CONVERT)
173 #define CONVERT(x) (DOCONV() ? convert(&x, sizeof(x)) : &x)
174
175 /* the body of the database is made of one list_struct for the free space
176    plus a separate data list for each hash value */
177 struct list_struct {
178         tdb_off next; /* offset of the next record in the list */
179         tdb_len rec_len; /* total byte length of record */
180         tdb_len key_len; /* byte length of key */
181         tdb_len data_len; /* byte length of data */
182         u32 full_hash; /* the full 32 bit hash of the key */
183         u32 magic;   /* try to catch errors */
184         /* the following union is implied:
185                 union {
186                         char record[rec_len];
187                         struct {
188                                 char key[key_len];
189                                 char data[data_len];
190                         }
191                         u32 totalsize; (tailer)
192                 }
193         */
194 };
195
196 /***************************************************************
197  Allow a caller to set a "alarm" flag that tdb can check to abort
198  a blocking lock on SIGALRM.
199 ***************************************************************/
200
201 static sig_atomic_t *palarm_fired;
202
203 void tdb_set_lock_alarm(sig_atomic_t *palarm)
204 {
205         palarm_fired = palarm;
206 }
207
208 /* a byte range locking function - return 0 on success
209    this functions locks/unlocks 1 byte at the specified offset.
210
211    On error, errno is also set so that errors are passed back properly
212    through tdb_open(). */
213 static int tdb_brlock(TDB_CONTEXT *tdb, tdb_off offset, 
214                       int rw_type, int lck_type, int probe)
215 {
216         struct flock fl;
217         int ret;
218
219         if (tdb->flags & TDB_NOLOCK)
220                 return 0;
221         if ((rw_type == F_WRLCK) && (tdb->read_only)) {
222                 errno = EACCES;
223                 return -1;
224         }
225
226         fl.l_type = rw_type;
227         fl.l_whence = SEEK_SET;
228         fl.l_start = offset;
229         fl.l_len = 1;
230         fl.l_pid = 0;
231
232         do {
233                 ret = fcntl(tdb->fd,lck_type,&fl);
234                 if (ret == -1 && errno == EINTR && palarm_fired && *palarm_fired)
235                         break;
236         } while (ret == -1 && errno == EINTR);
237
238         if (ret == -1) {
239                 if (!probe && lck_type != F_SETLK) {
240                         /* Ensure error code is set for log fun to examine. */
241                         if (errno == EINTR && palarm_fired && *palarm_fired)
242                                 tdb->ecode = TDB_ERR_LOCK_TIMEOUT;
243                         else
244                                 tdb->ecode = TDB_ERR_LOCK;
245                         TDB_LOG((tdb, 5,"tdb_brlock failed (fd=%d) at offset %d rw_type=%d lck_type=%d\n", 
246                                  tdb->fd, offset, rw_type, lck_type));
247                 }
248                 /* Was it an alarm timeout ? */
249                 if (errno == EINTR && palarm_fired && *palarm_fired) {
250                         TDB_LOG((tdb, 5, "tdb_brlock timed out (fd=%d) at offset %d rw_type=%d lck_type=%d\n", 
251                                  tdb->fd, offset, rw_type, lck_type));
252                         return TDB_ERRCODE(TDB_ERR_LOCK_TIMEOUT, -1);
253                 }
254                 /* Otherwise - generic lock error. errno set by fcntl.
255                  * EAGAIN is an expected return from non-blocking
256                  * locks. */
257                 if (errno != EAGAIN) {
258                 TDB_LOG((tdb, 5, "tdb_brlock failed (fd=%d) at offset %d rw_type=%d lck_type=%d: %s\n", 
259                                  tdb->fd, offset, rw_type, lck_type, 
260                                  strerror(errno)));
261                 }
262                 return TDB_ERRCODE(TDB_ERR_LOCK, -1);
263         }
264         return 0;
265 }
266
267 /* lock a list in the database. list -1 is the alloc list */
268 static int tdb_lock(TDB_CONTEXT *tdb, int list, int ltype)
269 {
270         if (list < -1 || list >= (int)tdb->header.hash_size) {
271                 TDB_LOG((tdb, 0,"tdb_lock: invalid list %d for ltype=%d\n", 
272                            list, ltype));
273                 return -1;
274         }
275         if (tdb->flags & TDB_NOLOCK)
276                 return 0;
277
278         /* Since fcntl locks don't nest, we do a lock for the first one,
279            and simply bump the count for future ones */
280         if (tdb->locked[list+1].count == 0) {
281                 if (!tdb->read_only && tdb->header.rwlocks) {
282                         if (tdb_spinlock(tdb, list, ltype)) {
283                                 TDB_LOG((tdb, 0, "tdb_lock spinlock failed on list %d ltype=%d\n", 
284                                            list, ltype));
285                                 return -1;
286                         }
287                 } else if (tdb_brlock(tdb,FREELIST_TOP+4*list,ltype,F_SETLKW, 0)) {
288                         TDB_LOG((tdb, 0,"tdb_lock failed on list %d ltype=%d (%s)\n", 
289                                            list, ltype, strerror(errno)));
290                         return -1;
291                 }
292                 tdb->locked[list+1].ltype = ltype;
293         }
294         tdb->locked[list+1].count++;
295         return 0;
296 }
297
298 /* unlock the database: returns void because it's too late for errors. */
299         /* changed to return int it may be interesting to know there
300            has been an error  --simo */
301 static int tdb_unlock(TDB_CONTEXT *tdb, int list, int ltype)
302 {
303         int ret = -1;
304
305         if (tdb->flags & TDB_NOLOCK)
306                 return 0;
307
308         /* Sanity checks */
309         if (list < -1 || list >= (int)tdb->header.hash_size) {
310                 TDB_LOG((tdb, 0, "tdb_unlock: list %d invalid (%d)\n", list, tdb->header.hash_size));
311                 return ret;
312         }
313
314         if (tdb->locked[list+1].count==0) {
315                 TDB_LOG((tdb, 0, "tdb_unlock: count is 0\n"));
316                 return ret;
317         }
318
319         if (tdb->locked[list+1].count == 1) {
320                 /* Down to last nested lock: unlock underneath */
321                 if (!tdb->read_only && tdb->header.rwlocks) {
322                         ret = tdb_spinunlock(tdb, list, ltype);
323                 } else {
324                         ret = tdb_brlock(tdb, FREELIST_TOP+4*list, F_UNLCK, F_SETLKW, 0);
325                 }
326         } else {
327                 ret = 0;
328         }
329         tdb->locked[list+1].count--;
330
331         if (ret)
332                 TDB_LOG((tdb, 0,"tdb_unlock: An error occurred unlocking!\n")); 
333         return ret;
334 }
335
336 /* This is based on the hash algorithm from gdbm */
337 static u32 default_tdb_hash(TDB_DATA *key)
338 {
339         u32 value;      /* Used to compute the hash value.  */
340         u32   i;        /* Used to cycle through random values. */
341
342         /* Set the initial value from the key size. */
343         for (value = 0x238F13AF * key->dsize, i=0; i < key->dsize; i++)
344                 value = (value + (key->dptr[i] << (i*5 % 24)));
345
346         return (1103515243 * value + 12345);  
347 }
348
349 /* check for an out of bounds access - if it is out of bounds then
350    see if the database has been expanded by someone else and expand
351    if necessary 
352    note that "len" is the minimum length needed for the db
353 */
354 static int tdb_oob(TDB_CONTEXT *tdb, tdb_off len, int probe)
355 {
356         struct stat st;
357         if (len <= tdb->map_size)
358                 return 0;
359         if (tdb->flags & TDB_INTERNAL) {
360                 if (!probe) {
361                         /* Ensure ecode is set for log fn. */
362                         tdb->ecode = TDB_ERR_IO;
363                         TDB_LOG((tdb, 0,"tdb_oob len %d beyond internal malloc size %d\n",
364                                  (int)len, (int)tdb->map_size));
365                 }
366                 return TDB_ERRCODE(TDB_ERR_IO, -1);
367         }
368
369         if (fstat(tdb->fd, &st) == -1)
370                 return TDB_ERRCODE(TDB_ERR_IO, -1);
371
372         if (st.st_size < (size_t)len) {
373                 if (!probe) {
374                         /* Ensure ecode is set for log fn. */
375                         tdb->ecode = TDB_ERR_IO;
376                         TDB_LOG((tdb, 0,"tdb_oob len %d beyond eof at %d\n",
377                                  (int)len, (int)st.st_size));
378                 }
379                 return TDB_ERRCODE(TDB_ERR_IO, -1);
380         }
381
382         /* Unmap, update size, remap */
383         if (tdb_munmap(tdb) == -1)
384                 return TDB_ERRCODE(TDB_ERR_IO, -1);
385         tdb->map_size = st.st_size;
386         tdb_mmap(tdb);
387         return 0;
388 }
389
390 /* write a lump of data at a specified offset */
391 static int tdb_write(TDB_CONTEXT *tdb, tdb_off off, void *buf, tdb_len len)
392 {
393         if (tdb_oob(tdb, off + len, 0) != 0)
394                 return -1;
395
396         if (tdb->map_ptr)
397                 memcpy(off + (char *)tdb->map_ptr, buf, len);
398 #ifdef HAVE_PWRITE
399         else if (pwrite(tdb->fd, buf, len, off) != (ssize_t)len) {
400 #else
401         else if (lseek(tdb->fd, off, SEEK_SET) != off
402                  || write(tdb->fd, buf, len) != (ssize_t)len) {
403 #endif
404                 /* Ensure ecode is set for log fn. */
405                 tdb->ecode = TDB_ERR_IO;
406                 TDB_LOG((tdb, 0,"tdb_write failed at %d len=%d (%s)\n",
407                            off, len, strerror(errno)));
408                 return TDB_ERRCODE(TDB_ERR_IO, -1);
409         }
410         return 0;
411 }
412
413 /* read a lump of data at a specified offset, maybe convert */
414 static int tdb_read(TDB_CONTEXT *tdb,tdb_off off,void *buf,tdb_len len,int cv)
415 {
416         if (tdb_oob(tdb, off + len, 0) != 0)
417                 return -1;
418
419         if (tdb->map_ptr)
420                 memcpy(buf, off + (char *)tdb->map_ptr, len);
421 #ifdef HAVE_PREAD
422         else if (pread(tdb->fd, buf, len, off) != (ssize_t)len) {
423 #else
424         else if (lseek(tdb->fd, off, SEEK_SET) != off
425                  || read(tdb->fd, buf, len) != (ssize_t)len) {
426 #endif
427                 /* Ensure ecode is set for log fn. */
428                 tdb->ecode = TDB_ERR_IO;
429                 TDB_LOG((tdb, 0,"tdb_read failed at %d len=%d (%s)\n",
430                            off, len, strerror(errno)));
431                 return TDB_ERRCODE(TDB_ERR_IO, -1);
432         }
433         if (cv)
434                 convert(buf, len);
435         return 0;
436 }
437
438 /* read a lump of data, allocating the space for it */
439 static char *tdb_alloc_read(TDB_CONTEXT *tdb, tdb_off offset, tdb_len len)
440 {
441         char *buf;
442
443         if (!(buf = malloc(len))) {
444                 /* Ensure ecode is set for log fn. */
445                 tdb->ecode = TDB_ERR_OOM;
446                 TDB_LOG((tdb, 0,"tdb_alloc_read malloc failed len=%d (%s)\n",
447                            len, strerror(errno)));
448                 return TDB_ERRCODE(TDB_ERR_OOM, buf);
449         }
450         if (tdb_read(tdb, offset, buf, len, 0) == -1) {
451                 SAFE_FREE(buf);
452                 return NULL;
453         }
454         return buf;
455 }
456
457 /* read/write a tdb_off */
458 static int ofs_read(TDB_CONTEXT *tdb, tdb_off offset, tdb_off *d)
459 {
460         return tdb_read(tdb, offset, (char*)d, sizeof(*d), DOCONV());
461 }
462 static int ofs_write(TDB_CONTEXT *tdb, tdb_off offset, tdb_off *d)
463 {
464         tdb_off off = *d;
465         return tdb_write(tdb, offset, CONVERT(off), sizeof(*d));
466 }
467
468 /* read/write a record */
469 static int rec_read(TDB_CONTEXT *tdb, tdb_off offset, struct list_struct *rec)
470 {
471         if (tdb_read(tdb, offset, rec, sizeof(*rec),DOCONV()) == -1)
472                 return -1;
473         if (TDB_BAD_MAGIC(rec)) {
474                 /* Ensure ecode is set for log fn. */
475                 tdb->ecode = TDB_ERR_CORRUPT;
476                 TDB_LOG((tdb, 0,"rec_read bad magic 0x%x at offset=%d\n", rec->magic, offset));
477                 return TDB_ERRCODE(TDB_ERR_CORRUPT, -1);
478         }
479         return tdb_oob(tdb, rec->next+sizeof(*rec), 0);
480 }
481 static int rec_write(TDB_CONTEXT *tdb, tdb_off offset, struct list_struct *rec)
482 {
483         struct list_struct r = *rec;
484         return tdb_write(tdb, offset, CONVERT(r), sizeof(r));
485 }
486
487 /* read a freelist record and check for simple errors */
488 static int rec_free_read(TDB_CONTEXT *tdb, tdb_off off, struct list_struct *rec)
489 {
490         if (tdb_read(tdb, off, rec, sizeof(*rec),DOCONV()) == -1)
491                 return -1;
492
493         if (rec->magic == TDB_MAGIC) {
494                 /* this happens when a app is showdown while deleting a record - we should
495                    not completely fail when this happens */
496                 TDB_LOG((tdb, 0,"rec_free_read non-free magic 0x%x at offset=%d - fixing\n", 
497                          rec->magic, off));
498                 rec->magic = TDB_FREE_MAGIC;
499                 if (tdb_write(tdb, off, rec, sizeof(*rec)) == -1)
500                         return -1;
501         }
502
503         if (rec->magic != TDB_FREE_MAGIC) {
504                 /* Ensure ecode is set for log fn. */
505                 tdb->ecode = TDB_ERR_CORRUPT;
506                 TDB_LOG((tdb, 0,"rec_free_read bad magic 0x%x at offset=%d\n", 
507                            rec->magic, off));
508                 return TDB_ERRCODE(TDB_ERR_CORRUPT, -1);
509         }
510         if (tdb_oob(tdb, rec->next+sizeof(*rec), 0) != 0)
511                 return -1;
512         return 0;
513 }
514
515 /* update a record tailer (must hold allocation lock) */
516 static int update_tailer(TDB_CONTEXT *tdb, tdb_off offset,
517                          const struct list_struct *rec)
518 {
519         tdb_off totalsize;
520
521         /* Offset of tailer from record header */
522         totalsize = sizeof(*rec) + rec->rec_len;
523         return ofs_write(tdb, offset + totalsize - sizeof(tdb_off),
524                          &totalsize);
525 }
526
527 static tdb_off tdb_dump_record(TDB_CONTEXT *tdb, tdb_off offset)
528 {
529         struct list_struct rec;
530         tdb_off tailer_ofs, tailer;
531
532         if (tdb_read(tdb, offset, (char *)&rec, sizeof(rec), DOCONV()) == -1) {
533                 printf("ERROR: failed to read record at %u\n", offset);
534                 return 0;
535         }
536
537         printf(" rec: offset=0x%08x next=0x%08x rec_len=%d key_len=%d data_len=%d full_hash=0x%x magic=0x%x\n",
538                offset, rec.next, rec.rec_len, rec.key_len, rec.data_len, rec.full_hash, rec.magic);
539
540         tailer_ofs = offset + sizeof(rec) + rec.rec_len - sizeof(tdb_off);
541         if (ofs_read(tdb, tailer_ofs, &tailer) == -1) {
542                 printf("ERROR: failed to read tailer at %u\n", tailer_ofs);
543                 return rec.next;
544         }
545
546         if (tailer != rec.rec_len + sizeof(rec)) {
547                 printf("ERROR: tailer does not match record! tailer=%u totalsize=%u\n",
548                                 (unsigned int)tailer, (unsigned int)(rec.rec_len + sizeof(rec)));
549         }
550         return rec.next;
551 }
552
553 static int tdb_dump_chain(TDB_CONTEXT *tdb, int i)
554 {
555         tdb_off rec_ptr, top;
556
557         top = TDB_HASH_TOP(i);
558
559         if (tdb_lock(tdb, i, F_WRLCK) != 0)
560                 return -1;
561
562         if (ofs_read(tdb, top, &rec_ptr) == -1)
563                 return tdb_unlock(tdb, i, F_WRLCK);
564
565         if (rec_ptr)
566                 printf("hash=%d\n", i);
567
568         while (rec_ptr) {
569                 rec_ptr = tdb_dump_record(tdb, rec_ptr);
570         }
571
572         return tdb_unlock(tdb, i, F_WRLCK);
573 }
574
575 void tdb_dump_all(TDB_CONTEXT *tdb)
576 {
577         int i;
578         for (i=0;i<tdb->header.hash_size;i++) {
579                 tdb_dump_chain(tdb, i);
580         }
581         printf("freelist:\n");
582         tdb_dump_chain(tdb, -1);
583 }
584
585 int tdb_printfreelist(TDB_CONTEXT *tdb)
586 {
587         int ret;
588         long total_free = 0;
589         tdb_off offset, rec_ptr;
590         struct list_struct rec;
591
592         if ((ret = tdb_lock(tdb, -1, F_WRLCK)) != 0)
593                 return ret;
594
595         offset = FREELIST_TOP;
596
597         /* read in the freelist top */
598         if (ofs_read(tdb, offset, &rec_ptr) == -1) {
599                 tdb_unlock(tdb, -1, F_WRLCK);
600                 return 0;
601         }
602
603         printf("freelist top=[0x%08x]\n", rec_ptr );
604         while (rec_ptr) {
605                 if (tdb_read(tdb, rec_ptr, (char *)&rec, sizeof(rec), DOCONV()) == -1) {
606                         tdb_unlock(tdb, -1, F_WRLCK);
607                         return -1;
608                 }
609
610                 if (rec.magic != TDB_FREE_MAGIC) {
611                         printf("bad magic 0x%08x in free list\n", rec.magic);
612                         tdb_unlock(tdb, -1, F_WRLCK);
613                         return -1;
614                 }
615
616                 printf("entry offset=[0x%08x], rec.rec_len = [0x%08x (%d)] (end = 0x%08x)\n", 
617                        rec_ptr, rec.rec_len, rec.rec_len, rec_ptr + rec.rec_len);
618                 total_free += rec.rec_len;
619
620                 /* move to the next record */
621                 rec_ptr = rec.next;
622         }
623         printf("total rec_len = [0x%08x (%d)]\n", (int)total_free, 
624                (int)total_free);
625
626         return tdb_unlock(tdb, -1, F_WRLCK);
627 }
628
629 /* Remove an element from the freelist.  Must have alloc lock. */
630 static int remove_from_freelist(TDB_CONTEXT *tdb, tdb_off off, tdb_off next)
631 {
632         tdb_off last_ptr, i;
633
634         /* read in the freelist top */
635         last_ptr = FREELIST_TOP;
636         while (ofs_read(tdb, last_ptr, &i) != -1 && i != 0) {
637                 if (i == off) {
638                         /* We've found it! */
639                         return ofs_write(tdb, last_ptr, &next);
640                 }
641                 /* Follow chain (next offset is at start of record) */
642                 last_ptr = i;
643         }
644         TDB_LOG((tdb, 0,"remove_from_freelist: not on list at off=%d\n", off));
645         return TDB_ERRCODE(TDB_ERR_CORRUPT, -1);
646 }
647
648 /* Add an element into the freelist. Merge adjacent records if
649    neccessary. */
650 static int tdb_free(TDB_CONTEXT *tdb, tdb_off offset, struct list_struct *rec)
651 {
652         tdb_off right, left;
653
654         /* Allocation and tailer lock */
655         if (tdb_lock(tdb, -1, F_WRLCK) != 0)
656                 return -1;
657
658         /* set an initial tailer, so if we fail we don't leave a bogus record */
659         if (update_tailer(tdb, offset, rec) != 0) {
660                 TDB_LOG((tdb, 0, "tdb_free: upfate_tailer failed!\n"));
661                 goto fail;
662         }
663
664         /* Look right first (I'm an Australian, dammit) */
665         right = offset + sizeof(*rec) + rec->rec_len;
666         if (right + sizeof(*rec) <= tdb->map_size) {
667                 struct list_struct r;
668
669                 if (tdb_read(tdb, right, &r, sizeof(r), DOCONV()) == -1) {
670                         TDB_LOG((tdb, 0, "tdb_free: right read failed at %u\n", right));
671                         goto left;
672                 }
673
674                 /* If it's free, expand to include it. */
675                 if (r.magic == TDB_FREE_MAGIC) {
676                         if (remove_from_freelist(tdb, right, r.next) == -1) {
677                                 TDB_LOG((tdb, 0, "tdb_free: right free failed at %u\n", right));
678                                 goto left;
679                         }
680                         rec->rec_len += sizeof(r) + r.rec_len;
681                 }
682         }
683
684 left:
685         /* Look left */
686         left = offset - sizeof(tdb_off);
687         if (left > TDB_DATA_START(tdb->header.hash_size)) {
688                 struct list_struct l;
689                 tdb_off leftsize;
690                 
691                 /* Read in tailer and jump back to header */
692                 if (ofs_read(tdb, left, &leftsize) == -1) {
693                         TDB_LOG((tdb, 0, "tdb_free: left offset read failed at %u\n", left));
694                         goto update;
695                 }
696                 left = offset - leftsize;
697
698                 /* Now read in record */
699                 if (tdb_read(tdb, left, &l, sizeof(l), DOCONV()) == -1) {
700                         TDB_LOG((tdb, 0, "tdb_free: left read failed at %u (%u)\n", left, leftsize));
701                         goto update;
702                 }
703
704                 /* If it's free, expand to include it. */
705                 if (l.magic == TDB_FREE_MAGIC) {
706                         if (remove_from_freelist(tdb, left, l.next) == -1) {
707                                 TDB_LOG((tdb, 0, "tdb_free: left free failed at %u\n", left));
708                                 goto update;
709                         } else {
710                                 offset = left;
711                                 rec->rec_len += leftsize;
712                         }
713                 }
714         }
715
716 update:
717         if (update_tailer(tdb, offset, rec) == -1) {
718                 TDB_LOG((tdb, 0, "tdb_free: update_tailer failed at %u\n", offset));
719                 goto fail;
720         }
721
722         /* Now, prepend to free list */
723         rec->magic = TDB_FREE_MAGIC;
724
725         if (ofs_read(tdb, FREELIST_TOP, &rec->next) == -1 ||
726             rec_write(tdb, offset, rec) == -1 ||
727             ofs_write(tdb, FREELIST_TOP, &offset) == -1) {
728                 TDB_LOG((tdb, 0, "tdb_free record write failed at offset=%d\n", offset));
729                 goto fail;
730         }
731
732         /* And we're done. */
733         tdb_unlock(tdb, -1, F_WRLCK);
734         return 0;
735
736  fail:
737         tdb_unlock(tdb, -1, F_WRLCK);
738         return -1;
739 }
740
741
742 /* expand a file.  we prefer to use ftruncate, as that is what posix
743   says to use for mmap expansion */
744 static int expand_file(TDB_CONTEXT *tdb, tdb_off size, tdb_off addition)
745 {
746         char buf[1024];
747 #if HAVE_FTRUNCATE_EXTEND
748         if (ftruncate(tdb->fd, size+addition) != 0) {
749                 TDB_LOG((tdb, 0, "expand_file ftruncate to %d failed (%s)\n", 
750                            size+addition, strerror(errno)));
751                 return -1;
752         }
753 #else
754         char b = 0;
755
756 #ifdef HAVE_PWRITE
757         if (pwrite(tdb->fd,  &b, 1, (size+addition) - 1) != 1) {
758 #else
759         if (lseek(tdb->fd, (size+addition) - 1, SEEK_SET) != (size+addition) - 1 || 
760             write(tdb->fd, &b, 1) != 1) {
761 #endif
762                 TDB_LOG((tdb, 0, "expand_file to %d failed (%s)\n", 
763                            size+addition, strerror(errno)));
764                 return -1;
765         }
766 #endif
767
768         /* now fill the file with something. This ensures that the file isn't sparse, which would be
769            very bad if we ran out of disk. This must be done with write, not via mmap */
770         memset(buf, 0x42, sizeof(buf));
771         while (addition) {
772                 int n = addition>sizeof(buf)?sizeof(buf):addition;
773 #ifdef HAVE_PWRITE
774                 int ret = pwrite(tdb->fd, buf, n, size);
775 #else
776                 int ret;
777                 if (lseek(tdb->fd, size, SEEK_SET) != size)
778                         return -1;
779                 ret = write(tdb->fd, buf, n);
780 #endif
781                 if (ret != n) {
782                         TDB_LOG((tdb, 0, "expand_file write of %d failed (%s)\n", 
783                                    n, strerror(errno)));
784                         return -1;
785                 }
786                 addition -= n;
787                 size += n;
788         }
789         return 0;
790 }
791
792
793 /* expand the database at least size bytes by expanding the underlying
794    file and doing the mmap again if necessary */
795 static int tdb_expand(TDB_CONTEXT *tdb, tdb_off size)
796 {
797         struct list_struct rec;
798         tdb_off offset;
799
800         if (tdb_lock(tdb, -1, F_WRLCK) == -1) {
801                 TDB_LOG((tdb, 0, "lock failed in tdb_expand\n"));
802                 return -1;
803         }
804
805         /* must know about any previous expansions by another process */
806         tdb_oob(tdb, tdb->map_size + 1, 1);
807
808         /* always make room for at least 10 more records, and round
809            the database up to a multiple of TDB_PAGE_SIZE */
810         size = TDB_ALIGN(tdb->map_size + size*10, TDB_PAGE_SIZE) - tdb->map_size;
811
812         if (!(tdb->flags & TDB_INTERNAL))
813                 tdb_munmap(tdb);
814
815         /*
816          * We must ensure the file is unmapped before doing this
817          * to ensure consistency with systems like OpenBSD where
818          * writes and mmaps are not consistent.
819          */
820
821         /* expand the file itself */
822         if (!(tdb->flags & TDB_INTERNAL)) {
823                 if (expand_file(tdb, tdb->map_size, size) != 0)
824                         goto fail;
825         }
826
827         tdb->map_size += size;
828
829         if (tdb->flags & TDB_INTERNAL)
830                 tdb->map_ptr = realloc(tdb->map_ptr, tdb->map_size);
831         else {
832                 /*
833                  * We must ensure the file is remapped before adding the space
834                  * to ensure consistency with systems like OpenBSD where
835                  * writes and mmaps are not consistent.
836                  */
837
838                 /* We're ok if the mmap fails as we'll fallback to read/write */
839                 tdb_mmap(tdb);
840         }
841
842         /* form a new freelist record */
843         memset(&rec,'\0',sizeof(rec));
844         rec.rec_len = size - sizeof(rec);
845
846         /* link it into the free list */
847         offset = tdb->map_size - size;
848         if (tdb_free(tdb, offset, &rec) == -1)
849                 goto fail;
850
851         tdb_unlock(tdb, -1, F_WRLCK);
852         return 0;
853  fail:
854         tdb_unlock(tdb, -1, F_WRLCK);
855         return -1;
856 }
857
858
859 /* 
860    the core of tdb_allocate - called when we have decided which
861    free list entry to use
862  */
863 static tdb_off tdb_allocate_ofs(TDB_CONTEXT *tdb, tdb_len length, tdb_off rec_ptr,
864                                 struct list_struct *rec, tdb_off last_ptr)
865 {
866         struct list_struct newrec;
867         tdb_off newrec_ptr;
868
869         memset(&newrec, '\0', sizeof(newrec));
870
871         /* found it - now possibly split it up  */
872         if (rec->rec_len > length + MIN_REC_SIZE) {
873                 /* Length of left piece */
874                 length = TDB_ALIGN(length, TDB_ALIGNMENT);
875                 
876                 /* Right piece to go on free list */
877                 newrec.rec_len = rec->rec_len - (sizeof(*rec) + length);
878                 newrec_ptr = rec_ptr + sizeof(*rec) + length;
879                 
880                 /* And left record is shortened */
881                 rec->rec_len = length;
882         } else {
883                 newrec_ptr = 0;
884         }
885         
886         /* Remove allocated record from the free list */
887         if (ofs_write(tdb, last_ptr, &rec->next) == -1) {
888                 return 0;
889         }
890         
891         /* Update header: do this before we drop alloc
892            lock, otherwise tdb_free() might try to
893            merge with us, thinking we're free.
894            (Thanks Jeremy Allison). */
895         rec->magic = TDB_MAGIC;
896         if (rec_write(tdb, rec_ptr, rec) == -1) {
897                 return 0;
898         }
899         
900         /* Did we create new block? */
901         if (newrec_ptr) {
902                 /* Update allocated record tailer (we
903                    shortened it). */
904                 if (update_tailer(tdb, rec_ptr, rec) == -1) {
905                         return 0;
906                 }
907                 
908                 /* Free new record */
909                 if (tdb_free(tdb, newrec_ptr, &newrec) == -1) {
910                         return 0;
911                 }
912         }
913         
914         /* all done - return the new record offset */
915         return rec_ptr;
916 }
917
918 /* allocate some space from the free list. The offset returned points
919    to a unconnected list_struct within the database with room for at
920    least length bytes of total data
921
922    0 is returned if the space could not be allocated
923  */
924 static tdb_off tdb_allocate(TDB_CONTEXT *tdb, tdb_len length,
925                             struct list_struct *rec)
926 {
927         tdb_off rec_ptr, last_ptr, newrec_ptr;
928         struct {
929                 tdb_off rec_ptr, last_ptr;
930                 tdb_len rec_len;
931         } bestfit;
932
933         if (tdb_lock(tdb, -1, F_WRLCK) == -1)
934                 return 0;
935
936         /* Extra bytes required for tailer */
937         length += sizeof(tdb_off);
938
939  again:
940         last_ptr = FREELIST_TOP;
941
942         /* read in the freelist top */
943         if (ofs_read(tdb, FREELIST_TOP, &rec_ptr) == -1)
944                 goto fail;
945
946         bestfit.rec_ptr = 0;
947
948         /* 
949            this is a best fit allocation strategy. Originally we used
950            a first fit strategy, but it suffered from massive fragmentation
951            issues when faced with a slowly increasing record size.
952          */
953         while (rec_ptr) {
954                 if (rec_free_read(tdb, rec_ptr, rec) == -1) {
955                         goto fail;
956                 }
957
958                 if (rec->rec_len >= length) {
959                         if (bestfit.rec_ptr == 0 ||
960                             rec->rec_len < bestfit.rec_len) {
961                                 bestfit.rec_len = rec->rec_len;
962                                 bestfit.rec_ptr = rec_ptr;
963                                 bestfit.last_ptr = last_ptr;
964                                 /* consider a fit to be good enough if we aren't wasting more than half the space */
965                                 if (bestfit.rec_len < 2*length) {
966                                         break;
967                                 }
968                         }
969                 }
970
971                 /* move to the next record */
972                 last_ptr = rec_ptr;
973                 rec_ptr = rec->next;
974         }
975
976         if (bestfit.rec_ptr != 0) {
977                 if (rec_free_read(tdb, bestfit.rec_ptr, rec) == -1) {
978                         goto fail;
979                 }
980
981                 newrec_ptr = tdb_allocate_ofs(tdb, length, bestfit.rec_ptr, rec, bestfit.last_ptr);
982                 tdb_unlock(tdb, -1, F_WRLCK);
983                 return newrec_ptr;
984         }
985
986         /* we didn't find enough space. See if we can expand the
987            database and if we can then try again */
988         if (tdb_expand(tdb, length + sizeof(*rec)) == 0)
989                 goto again;
990  fail:
991         tdb_unlock(tdb, -1, F_WRLCK);
992         return 0;
993 }
994
995 /* initialise a new database with a specified hash size */
996 static int tdb_new_database(TDB_CONTEXT *tdb, int hash_size)
997 {
998         struct tdb_header *newdb;
999         int size, ret = -1;
1000
1001         /* We make it up in memory, then write it out if not internal */
1002         size = sizeof(struct tdb_header) + (hash_size+1)*sizeof(tdb_off);
1003         if (!(newdb = calloc(size, 1)))
1004                 return TDB_ERRCODE(TDB_ERR_OOM, -1);
1005
1006         /* Fill in the header */
1007         newdb->version = TDB_VERSION;
1008         newdb->hash_size = hash_size;
1009 #ifdef USE_SPINLOCKS
1010         newdb->rwlocks = size;
1011 #endif
1012         if (tdb->flags & TDB_INTERNAL) {
1013                 tdb->map_size = size;
1014                 tdb->map_ptr = (char *)newdb;
1015                 memcpy(&tdb->header, newdb, sizeof(tdb->header));
1016                 /* Convert the `ondisk' version if asked. */
1017                 CONVERT(*newdb);
1018                 return 0;
1019         }
1020         if (lseek(tdb->fd, 0, SEEK_SET) == -1)
1021                 goto fail;
1022
1023         if (ftruncate(tdb->fd, 0) == -1)
1024                 goto fail;
1025
1026         /* This creates an endian-converted header, as if read from disk */
1027         CONVERT(*newdb);
1028         memcpy(&tdb->header, newdb, sizeof(tdb->header));
1029         /* Don't endian-convert the magic food! */
1030         memcpy(newdb->magic_food, TDB_MAGIC_FOOD, strlen(TDB_MAGIC_FOOD)+1);
1031         if (write(tdb->fd, newdb, size) != size)
1032                 ret = -1;
1033         else
1034                 ret = tdb_create_rwlocks(tdb->fd, hash_size);
1035
1036   fail:
1037         SAFE_FREE(newdb);
1038         return ret;
1039 }
1040
1041 /* Returns 0 on fail.  On success, return offset of record, and fills
1042    in rec */
1043 static tdb_off tdb_find(TDB_CONTEXT *tdb, TDB_DATA key, u32 hash,
1044                         struct list_struct *r)
1045 {
1046         tdb_off rec_ptr;
1047         
1048         /* read in the hash top */
1049         if (ofs_read(tdb, TDB_HASH_TOP(hash), &rec_ptr) == -1)
1050                 return 0;
1051
1052         /* keep looking until we find the right record */
1053         while (rec_ptr) {
1054                 if (rec_read(tdb, rec_ptr, r) == -1)
1055                         return 0;
1056
1057                 if (!TDB_DEAD(r) && hash==r->full_hash && key.dsize==r->key_len) {
1058                         char *k;
1059                         /* a very likely hit - read the key */
1060                         k = tdb_alloc_read(tdb, rec_ptr + sizeof(*r), 
1061                                            r->key_len);
1062                         if (!k)
1063                                 return 0;
1064
1065                         if (memcmp(key.dptr, k, key.dsize) == 0) {
1066                                 SAFE_FREE(k);
1067                                 return rec_ptr;
1068                         }
1069                         SAFE_FREE(k);
1070                 }
1071                 rec_ptr = r->next;
1072         }
1073         return TDB_ERRCODE(TDB_ERR_NOEXIST, 0);
1074 }
1075
1076 /* As tdb_find, but if you succeed, keep the lock */
1077 static tdb_off tdb_find_lock_hash(TDB_CONTEXT *tdb, TDB_DATA key, u32 hash, int locktype,
1078                              struct list_struct *rec)
1079 {
1080         u32 rec_ptr;
1081
1082         if (tdb_lock(tdb, BUCKET(hash), locktype) == -1)
1083                 return 0;
1084         if (!(rec_ptr = tdb_find(tdb, key, hash, rec)))
1085                 tdb_unlock(tdb, BUCKET(hash), locktype);
1086         return rec_ptr;
1087 }
1088
1089 enum TDB_ERROR tdb_error(TDB_CONTEXT *tdb)
1090 {
1091         return tdb->ecode;
1092 }
1093
1094 static struct tdb_errname {
1095         enum TDB_ERROR ecode; const char *estring;
1096 } emap[] = { {TDB_SUCCESS, "Success"},
1097              {TDB_ERR_CORRUPT, "Corrupt database"},
1098              {TDB_ERR_IO, "IO Error"},
1099              {TDB_ERR_LOCK, "Locking error"},
1100              {TDB_ERR_OOM, "Out of memory"},
1101              {TDB_ERR_EXISTS, "Record exists"},
1102              {TDB_ERR_NOLOCK, "Lock exists on other keys"},
1103              {TDB_ERR_NOEXIST, "Record does not exist"} };
1104
1105 /* Error string for the last tdb error */
1106 const char *tdb_errorstr(TDB_CONTEXT *tdb)
1107 {
1108         u32 i;
1109         for (i = 0; i < sizeof(emap) / sizeof(struct tdb_errname); i++)
1110                 if (tdb->ecode == emap[i].ecode)
1111                         return emap[i].estring;
1112         return "Invalid error code";
1113 }
1114
1115 /* update an entry in place - this only works if the new data size
1116    is <= the old data size and the key exists.
1117    on failure return -1.
1118 */
1119
1120 static int tdb_update_hash(TDB_CONTEXT *tdb, TDB_DATA key, u32 hash, TDB_DATA dbuf)
1121 {
1122         struct list_struct rec;
1123         tdb_off rec_ptr;
1124
1125         /* find entry */
1126         if (!(rec_ptr = tdb_find(tdb, key, hash, &rec)))
1127                 return -1;
1128
1129         /* must be long enough key, data and tailer */
1130         if (rec.rec_len < key.dsize + dbuf.dsize + sizeof(tdb_off)) {
1131                 tdb->ecode = TDB_SUCCESS; /* Not really an error */
1132                 return -1;
1133         }
1134
1135         if (tdb_write(tdb, rec_ptr + sizeof(rec) + rec.key_len,
1136                       dbuf.dptr, dbuf.dsize) == -1)
1137                 return -1;
1138
1139         if (dbuf.dsize != rec.data_len) {
1140                 /* update size */
1141                 rec.data_len = dbuf.dsize;
1142                 return rec_write(tdb, rec_ptr, &rec);
1143         }
1144  
1145         return 0;
1146 }
1147
1148 /* find an entry in the database given a key */
1149 /* If an entry doesn't exist tdb_err will be set to
1150  * TDB_ERR_NOEXIST. If a key has no data attached
1151  * tdb_err will not be set. Both will return a
1152  * zero pptr and zero dsize.
1153  */
1154
1155 TDB_DATA tdb_fetch(TDB_CONTEXT *tdb, TDB_DATA key)
1156 {
1157         tdb_off rec_ptr;
1158         struct list_struct rec;
1159         TDB_DATA ret;
1160         u32 hash;
1161
1162         /* find which hash bucket it is in */
1163         hash = tdb->hash_fn(&key);
1164         if (!(rec_ptr = tdb_find_lock_hash(tdb,key,hash,F_RDLCK,&rec)))
1165                 return tdb_null;
1166
1167         if (rec.data_len)
1168                 ret.dptr = tdb_alloc_read(tdb, rec_ptr + sizeof(rec) + rec.key_len,
1169                                           rec.data_len);
1170         else
1171                 ret.dptr = NULL;
1172         ret.dsize = rec.data_len;
1173         tdb_unlock(tdb, BUCKET(rec.full_hash), F_RDLCK);
1174         return ret;
1175 }
1176
1177 /* check if an entry in the database exists 
1178
1179    note that 1 is returned if the key is found and 0 is returned if not found
1180    this doesn't match the conventions in the rest of this module, but is
1181    compatible with gdbm
1182 */
1183 static int tdb_exists_hash(TDB_CONTEXT *tdb, TDB_DATA key, u32 hash)
1184 {
1185         struct list_struct rec;
1186         
1187         if (tdb_find_lock_hash(tdb, key, hash, F_RDLCK, &rec) == 0)
1188                 return 0;
1189         tdb_unlock(tdb, BUCKET(rec.full_hash), F_RDLCK);
1190         return 1;
1191 }
1192
1193 int tdb_exists(TDB_CONTEXT *tdb, TDB_DATA key)
1194 {
1195         u32 hash = tdb->hash_fn(&key);
1196         return tdb_exists_hash(tdb, key, hash);
1197 }
1198
1199 /* record lock stops delete underneath */
1200 static int lock_record(TDB_CONTEXT *tdb, tdb_off off)
1201 {
1202         return off ? tdb_brlock(tdb, off, F_RDLCK, F_SETLKW, 0) : 0;
1203 }
1204 /*
1205   Write locks override our own fcntl readlocks, so check it here.
1206   Note this is meant to be F_SETLK, *not* F_SETLKW, as it's not
1207   an error to fail to get the lock here.
1208 */
1209  
1210 static int write_lock_record(TDB_CONTEXT *tdb, tdb_off off)
1211 {
1212         struct tdb_traverse_lock *i;
1213         for (i = &tdb->travlocks; i; i = i->next)
1214                 if (i->off == off)
1215                         return -1;
1216         return tdb_brlock(tdb, off, F_WRLCK, F_SETLK, 1);
1217 }
1218
1219 /*
1220   Note this is meant to be F_SETLK, *not* F_SETLKW, as it's not
1221   an error to fail to get the lock here.
1222 */
1223
1224 static int write_unlock_record(TDB_CONTEXT *tdb, tdb_off off)
1225 {
1226         return tdb_brlock(tdb, off, F_UNLCK, F_SETLK, 0);
1227 }
1228 /* fcntl locks don't stack: avoid unlocking someone else's */
1229 static int unlock_record(TDB_CONTEXT *tdb, tdb_off off)
1230 {
1231         struct tdb_traverse_lock *i;
1232         u32 count = 0;
1233
1234         if (off == 0)
1235                 return 0;
1236         for (i = &tdb->travlocks; i; i = i->next)
1237                 if (i->off == off)
1238                         count++;
1239         return (count == 1 ? tdb_brlock(tdb, off, F_UNLCK, F_SETLKW, 0) : 0);
1240 }
1241
1242 /* actually delete an entry in the database given the offset */
1243 static int do_delete(TDB_CONTEXT *tdb, tdb_off rec_ptr, struct list_struct*rec)
1244 {
1245         tdb_off last_ptr, i;
1246         struct list_struct lastrec;
1247
1248         if (tdb->read_only) return -1;
1249
1250         if (write_lock_record(tdb, rec_ptr) == -1) {
1251                 /* Someone traversing here: mark it as dead */
1252                 rec->magic = TDB_DEAD_MAGIC;
1253                 return rec_write(tdb, rec_ptr, rec);
1254         }
1255         if (write_unlock_record(tdb, rec_ptr) != 0)
1256                 return -1;
1257
1258         /* find previous record in hash chain */
1259         if (ofs_read(tdb, TDB_HASH_TOP(rec->full_hash), &i) == -1)
1260                 return -1;
1261         for (last_ptr = 0; i != rec_ptr; last_ptr = i, i = lastrec.next)
1262                 if (rec_read(tdb, i, &lastrec) == -1)
1263                         return -1;
1264
1265         /* unlink it: next ptr is at start of record. */
1266         if (last_ptr == 0)
1267                 last_ptr = TDB_HASH_TOP(rec->full_hash);
1268         if (ofs_write(tdb, last_ptr, &rec->next) == -1)
1269                 return -1;
1270
1271         /* recover the space */
1272         if (tdb_free(tdb, rec_ptr, rec) == -1)
1273                 return -1;
1274         return 0;
1275 }
1276
1277 /* Uses traverse lock: 0 = finish, -1 = error, other = record offset */
1278 static int tdb_next_lock(TDB_CONTEXT *tdb, struct tdb_traverse_lock *tlock,
1279                          struct list_struct *rec)
1280 {
1281         int want_next = (tlock->off != 0);
1282
1283         /* Lock each chain from the start one. */
1284         for (; tlock->hash < tdb->header.hash_size; tlock->hash++) {
1285                 if (tdb_lock(tdb, tlock->hash, F_WRLCK) == -1)
1286                         return -1;
1287
1288                 /* No previous record?  Start at top of chain. */
1289                 if (!tlock->off) {
1290                         if (ofs_read(tdb, TDB_HASH_TOP(tlock->hash),
1291                                      &tlock->off) == -1)
1292                                 goto fail;
1293                 } else {
1294                         /* Otherwise unlock the previous record. */
1295                         if (unlock_record(tdb, tlock->off) != 0)
1296                                 goto fail;
1297                 }
1298
1299                 if (want_next) {
1300                         /* We have offset of old record: grab next */
1301                         if (rec_read(tdb, tlock->off, rec) == -1)
1302                                 goto fail;
1303                         tlock->off = rec->next;
1304                 }
1305
1306                 /* Iterate through chain */
1307                 while( tlock->off) {
1308                         tdb_off current;
1309                         if (rec_read(tdb, tlock->off, rec) == -1)
1310                                 goto fail;
1311                         if (!TDB_DEAD(rec)) {
1312                                 /* Woohoo: we found one! */
1313                                 if (lock_record(tdb, tlock->off) != 0)
1314                                         goto fail;
1315                                 return tlock->off;
1316                         }
1317                         /* Try to clean dead ones from old traverses */
1318                         current = tlock->off;
1319                         tlock->off = rec->next;
1320                         if (!tdb->read_only && 
1321                             do_delete(tdb, current, rec) != 0)
1322                                 goto fail;
1323                 }
1324                 tdb_unlock(tdb, tlock->hash, F_WRLCK);
1325                 want_next = 0;
1326         }
1327         /* We finished iteration without finding anything */
1328         return TDB_ERRCODE(TDB_SUCCESS, 0);
1329
1330  fail:
1331         tlock->off = 0;
1332         if (tdb_unlock(tdb, tlock->hash, F_WRLCK) != 0)
1333                 TDB_LOG((tdb, 0, "tdb_next_lock: On error unlock failed!\n"));
1334         return -1;
1335 }
1336
1337 /* traverse the entire database - calling fn(tdb, key, data) on each element.
1338    return -1 on error or the record count traversed
1339    if fn is NULL then it is not called
1340    a non-zero return value from fn() indicates that the traversal should stop
1341   */
1342 int tdb_traverse(TDB_CONTEXT *tdb, tdb_traverse_func fn, void *private)
1343 {
1344         TDB_DATA key, dbuf;
1345         struct list_struct rec;
1346         struct tdb_traverse_lock tl = { NULL, 0, 0 };
1347         int ret, count = 0;
1348
1349         /* This was in the initializaton, above, but the IRIX compiler
1350          * did not like it.  crh
1351          */
1352         tl.next = tdb->travlocks.next;
1353
1354         /* fcntl locks don't stack: beware traverse inside traverse */
1355         tdb->travlocks.next = &tl;
1356
1357         /* tdb_next_lock places locks on the record returned, and its chain */
1358         while ((ret = tdb_next_lock(tdb, &tl, &rec)) > 0) {
1359                 count++;
1360                 /* now read the full record */
1361                 key.dptr = tdb_alloc_read(tdb, tl.off + sizeof(rec), 
1362                                           rec.key_len + rec.data_len);
1363                 if (!key.dptr) {
1364                         ret = -1;
1365                         if (tdb_unlock(tdb, tl.hash, F_WRLCK) != 0)
1366                                 goto out;
1367                         if (unlock_record(tdb, tl.off) != 0)
1368                                 TDB_LOG((tdb, 0, "tdb_traverse: key.dptr == NULL and unlock_record failed!\n"));
1369                         goto out;
1370                 }
1371                 key.dsize = rec.key_len;
1372                 dbuf.dptr = key.dptr + rec.key_len;
1373                 dbuf.dsize = rec.data_len;
1374
1375                 /* Drop chain lock, call out */
1376                 if (tdb_unlock(tdb, tl.hash, F_WRLCK) != 0) {
1377                         ret = -1;
1378                         goto out;
1379                 }
1380                 if (fn && fn(tdb, key, dbuf, private)) {
1381                         /* They want us to terminate traversal */
1382                         ret = count;
1383                         if (unlock_record(tdb, tl.off) != 0) {
1384                                 TDB_LOG((tdb, 0, "tdb_traverse: unlock_record failed!\n"));;
1385                                 ret = -1;
1386                         }
1387                         tdb->travlocks.next = tl.next;
1388                         SAFE_FREE(key.dptr);
1389                         return count;
1390                 }
1391                 SAFE_FREE(key.dptr);
1392         }
1393 out:
1394         tdb->travlocks.next = tl.next;
1395         if (ret < 0)
1396                 return -1;
1397         else
1398                 return count;
1399 }
1400
1401 /* find the first entry in the database and return its key */
1402 TDB_DATA tdb_firstkey(TDB_CONTEXT *tdb)
1403 {
1404         TDB_DATA key;
1405         struct list_struct rec;
1406
1407         /* release any old lock */
1408         if (unlock_record(tdb, tdb->travlocks.off) != 0)
1409                 return tdb_null;
1410         tdb->travlocks.off = tdb->travlocks.hash = 0;
1411
1412         if (tdb_next_lock(tdb, &tdb->travlocks, &rec) <= 0)
1413                 return tdb_null;
1414         /* now read the key */
1415         key.dsize = rec.key_len;
1416         key.dptr =tdb_alloc_read(tdb,tdb->travlocks.off+sizeof(rec),key.dsize);
1417         if (tdb_unlock(tdb, BUCKET(tdb->travlocks.hash), F_WRLCK) != 0)
1418                 TDB_LOG((tdb, 0, "tdb_firstkey: error occurred while tdb_unlocking!\n"));
1419         return key;
1420 }
1421
1422 /* find the next entry in the database, returning its key */
1423 TDB_DATA tdb_nextkey(TDB_CONTEXT *tdb, TDB_DATA oldkey)
1424 {
1425         u32 oldhash;
1426         TDB_DATA key = tdb_null;
1427         struct list_struct rec;
1428         char *k = NULL;
1429
1430         /* Is locked key the old key?  If so, traverse will be reliable. */
1431         if (tdb->travlocks.off) {
1432                 if (tdb_lock(tdb,tdb->travlocks.hash,F_WRLCK))
1433                         return tdb_null;
1434                 if (rec_read(tdb, tdb->travlocks.off, &rec) == -1
1435                     || !(k = tdb_alloc_read(tdb,tdb->travlocks.off+sizeof(rec),
1436                                             rec.key_len))
1437                     || memcmp(k, oldkey.dptr, oldkey.dsize) != 0) {
1438                         /* No, it wasn't: unlock it and start from scratch */
1439                         if (unlock_record(tdb, tdb->travlocks.off) != 0)
1440                                 return tdb_null;
1441                         if (tdb_unlock(tdb, tdb->travlocks.hash, F_WRLCK) != 0)
1442                                 return tdb_null;
1443                         tdb->travlocks.off = 0;
1444                 }
1445
1446                 SAFE_FREE(k);
1447         }
1448
1449         if (!tdb->travlocks.off) {
1450                 /* No previous element: do normal find, and lock record */
1451                 tdb->travlocks.off = tdb_find_lock_hash(tdb, oldkey, tdb->hash_fn(&oldkey), F_WRLCK, &rec);
1452                 if (!tdb->travlocks.off)
1453                         return tdb_null;
1454                 tdb->travlocks.hash = BUCKET(rec.full_hash);
1455                 if (lock_record(tdb, tdb->travlocks.off) != 0) {
1456                         TDB_LOG((tdb, 0, "tdb_nextkey: lock_record failed (%s)!\n", strerror(errno)));
1457                         return tdb_null;
1458                 }
1459         }
1460         oldhash = tdb->travlocks.hash;
1461
1462         /* Grab next record: locks chain and returned record,
1463            unlocks old record */
1464         if (tdb_next_lock(tdb, &tdb->travlocks, &rec) > 0) {
1465                 key.dsize = rec.key_len;
1466                 key.dptr = tdb_alloc_read(tdb, tdb->travlocks.off+sizeof(rec),
1467                                           key.dsize);
1468                 /* Unlock the chain of this new record */
1469                 if (tdb_unlock(tdb, tdb->travlocks.hash, F_WRLCK) != 0)
1470                         TDB_LOG((tdb, 0, "tdb_nextkey: WARNING tdb_unlock failed!\n"));
1471         }
1472         /* Unlock the chain of old record */
1473         if (tdb_unlock(tdb, BUCKET(oldhash), F_WRLCK) != 0)
1474                 TDB_LOG((tdb, 0, "tdb_nextkey: WARNING tdb_unlock failed!\n"));
1475         return key;
1476 }
1477
1478 /* delete an entry in the database given a key */
1479 static int tdb_delete_hash(TDB_CONTEXT *tdb, TDB_DATA key, u32 hash)
1480 {
1481         tdb_off rec_ptr;
1482         struct list_struct rec;
1483         int ret;
1484
1485         if (!(rec_ptr = tdb_find_lock_hash(tdb, key, hash, F_WRLCK, &rec)))
1486                 return -1;
1487         ret = do_delete(tdb, rec_ptr, &rec);
1488         if (tdb_unlock(tdb, BUCKET(rec.full_hash), F_WRLCK) != 0)
1489                 TDB_LOG((tdb, 0, "tdb_delete: WARNING tdb_unlock failed!\n"));
1490         return ret;
1491 }
1492
1493 int tdb_delete(TDB_CONTEXT *tdb, TDB_DATA key)
1494 {
1495         u32 hash = tdb->hash_fn(&key);
1496         return tdb_delete_hash(tdb, key, hash);
1497 }
1498
1499 /* store an element in the database, replacing any existing element
1500    with the same key 
1501
1502    return 0 on success, -1 on failure
1503 */
1504 int tdb_store(TDB_CONTEXT *tdb, TDB_DATA key, TDB_DATA dbuf, int flag)
1505 {
1506         struct list_struct rec;
1507         u32 hash;
1508         tdb_off rec_ptr;
1509         char *p = NULL;
1510         int ret = 0;
1511
1512         /* find which hash bucket it is in */
1513         hash = tdb->hash_fn(&key);
1514         if (tdb_lock(tdb, BUCKET(hash), F_WRLCK) == -1)
1515                 return -1;
1516
1517         /* check for it existing, on insert. */
1518         if (flag == TDB_INSERT) {
1519                 if (tdb_exists_hash(tdb, key, hash)) {
1520                         tdb->ecode = TDB_ERR_EXISTS;
1521                         goto fail;
1522                 }
1523         } else {
1524                 /* first try in-place update, on modify or replace. */
1525                 if (tdb_update_hash(tdb, key, hash, dbuf) == 0)
1526                         goto out;
1527                 if (tdb->ecode == TDB_ERR_NOEXIST &&
1528                     flag == TDB_MODIFY) {
1529                         /* if the record doesn't exist and we are in TDB_MODIFY mode then
1530                          we should fail the store */
1531                         goto fail;
1532                 }
1533         }
1534         /* reset the error code potentially set by the tdb_update() */
1535         tdb->ecode = TDB_SUCCESS;
1536
1537         /* delete any existing record - if it doesn't exist we don't
1538            care.  Doing this first reduces fragmentation, and avoids
1539            coalescing with `allocated' block before it's updated. */
1540         if (flag != TDB_INSERT)
1541                 tdb_delete_hash(tdb, key, hash);
1542
1543         /* Copy key+value *before* allocating free space in case malloc
1544            fails and we are left with a dead spot in the tdb. */
1545
1546         if (!(p = (char *)malloc(key.dsize + dbuf.dsize))) {
1547                 tdb->ecode = TDB_ERR_OOM;
1548                 goto fail;
1549         }
1550
1551         memcpy(p, key.dptr, key.dsize);
1552         if (dbuf.dsize)
1553                 memcpy(p+key.dsize, dbuf.dptr, dbuf.dsize);
1554
1555         /* we have to allocate some space */
1556         if (!(rec_ptr = tdb_allocate(tdb, key.dsize + dbuf.dsize, &rec)))
1557                 goto fail;
1558
1559         /* Read hash top into next ptr */
1560         if (ofs_read(tdb, TDB_HASH_TOP(hash), &rec.next) == -1)
1561                 goto fail;
1562
1563         rec.key_len = key.dsize;
1564         rec.data_len = dbuf.dsize;
1565         rec.full_hash = hash;
1566         rec.magic = TDB_MAGIC;
1567
1568         /* write out and point the top of the hash chain at it */
1569         if (rec_write(tdb, rec_ptr, &rec) == -1
1570             || tdb_write(tdb, rec_ptr+sizeof(rec), p, key.dsize+dbuf.dsize)==-1
1571             || ofs_write(tdb, TDB_HASH_TOP(hash), &rec_ptr) == -1) {
1572                 /* Need to tdb_unallocate() here */
1573                 goto fail;
1574         }
1575  out:
1576         SAFE_FREE(p); 
1577         tdb_unlock(tdb, BUCKET(hash), F_WRLCK);
1578         return ret;
1579 fail:
1580         ret = -1;
1581         goto out;
1582 }
1583
1584 /* Attempt to append data to an entry in place - this only works if the new data size
1585    is <= the old data size and the key exists.
1586    on failure return -1. Record must be locked before calling.
1587 */
1588 static int tdb_append_inplace(TDB_CONTEXT *tdb, TDB_DATA key, u32 hash, TDB_DATA new_dbuf)
1589 {
1590         struct list_struct rec;
1591         tdb_off rec_ptr;
1592
1593         /* find entry */
1594         if (!(rec_ptr = tdb_find(tdb, key, hash, &rec)))
1595                 return -1;
1596
1597         /* Append of 0 is always ok. */
1598         if (new_dbuf.dsize == 0)
1599                 return 0;
1600
1601         /* must be long enough for key, old data + new data and tailer */
1602         if (rec.rec_len < key.dsize + rec.data_len + new_dbuf.dsize + sizeof(tdb_off)) {
1603                 /* No room. */
1604                 tdb->ecode = TDB_SUCCESS; /* Not really an error */
1605                 return -1;
1606         }
1607
1608         if (tdb_write(tdb, rec_ptr + sizeof(rec) + rec.key_len + rec.data_len,
1609                       new_dbuf.dptr, new_dbuf.dsize) == -1)
1610                 return -1;
1611
1612         /* update size */
1613         rec.data_len += new_dbuf.dsize;
1614         return rec_write(tdb, rec_ptr, &rec);
1615 }
1616
1617 /* Append to an entry. Create if not exist. */
1618
1619 int tdb_append(TDB_CONTEXT *tdb, TDB_DATA key, TDB_DATA new_dbuf)
1620 {
1621         struct list_struct rec;
1622         u32 hash;
1623         tdb_off rec_ptr;
1624         char *p = NULL;
1625         int ret = 0;
1626         size_t new_data_size = 0;
1627
1628         /* find which hash bucket it is in */
1629         hash = tdb->hash_fn(&key);
1630         if (tdb_lock(tdb, BUCKET(hash), F_WRLCK) == -1)
1631                 return -1;
1632
1633         /* first try in-place. */
1634         if (tdb_append_inplace(tdb, key, hash, new_dbuf) == 0)
1635                 goto out;
1636
1637         /* reset the error code potentially set by the tdb_append_inplace() */
1638         tdb->ecode = TDB_SUCCESS;
1639
1640         /* find entry */
1641         if (!(rec_ptr = tdb_find(tdb, key, hash, &rec))) {
1642                 if (tdb->ecode != TDB_ERR_NOEXIST)
1643                         goto fail;
1644
1645                 /* Not found - create. */
1646
1647                 ret = tdb_store(tdb, key, new_dbuf, TDB_INSERT);
1648                 goto out;
1649         }
1650
1651         new_data_size = rec.data_len + new_dbuf.dsize;
1652
1653         /* Copy key+old_value+value *before* allocating free space in case malloc
1654            fails and we are left with a dead spot in the tdb. */
1655
1656         if (!(p = (char *)malloc(key.dsize + new_data_size))) {
1657                 tdb->ecode = TDB_ERR_OOM;
1658                 goto fail;
1659         }
1660
1661         /* Copy the key in place. */
1662         memcpy(p, key.dptr, key.dsize);
1663
1664         /* Now read the old data into place. */
1665         if (rec.data_len &&
1666                 tdb_read(tdb, rec_ptr + sizeof(rec) + rec.key_len, p + key.dsize, rec.data_len, 0) == -1)
1667                         goto fail;
1668
1669         /* Finally append the new data. */
1670         if (new_dbuf.dsize)
1671                 memcpy(p+key.dsize+rec.data_len, new_dbuf.dptr, new_dbuf.dsize);
1672
1673         /* delete any existing record - if it doesn't exist we don't
1674            care.  Doing this first reduces fragmentation, and avoids
1675            coalescing with `allocated' block before it's updated. */
1676
1677         tdb_delete_hash(tdb, key, hash);
1678
1679         if (!(rec_ptr = tdb_allocate(tdb, key.dsize + new_data_size, &rec)))
1680                 goto fail;
1681
1682         /* Read hash top into next ptr */
1683         if (ofs_read(tdb, TDB_HASH_TOP(hash), &rec.next) == -1)
1684                 goto fail;
1685
1686         rec.key_len = key.dsize;
1687         rec.data_len = new_data_size;
1688         rec.full_hash = hash;
1689         rec.magic = TDB_MAGIC;
1690
1691         /* write out and point the top of the hash chain at it */
1692         if (rec_write(tdb, rec_ptr, &rec) == -1
1693             || tdb_write(tdb, rec_ptr+sizeof(rec), p, key.dsize+new_data_size)==-1
1694             || ofs_write(tdb, TDB_HASH_TOP(hash), &rec_ptr) == -1) {
1695                 /* Need to tdb_unallocate() here */
1696                 goto fail;
1697         }
1698
1699  out:
1700         SAFE_FREE(p); 
1701         tdb_unlock(tdb, BUCKET(hash), F_WRLCK);
1702         return ret;
1703
1704 fail:
1705         ret = -1;
1706         goto out;
1707 }
1708
1709 static int tdb_already_open(dev_t device,
1710                             ino_t ino)
1711 {
1712         TDB_CONTEXT *i;
1713         
1714         for (i = tdbs; i; i = i->next) {
1715                 if (i->device == device && i->inode == ino) {
1716                         return 1;
1717                 }
1718         }
1719
1720         return 0;
1721 }
1722
1723 /* open the database, creating it if necessary 
1724
1725    The open_flags and mode are passed straight to the open call on the
1726    database file. A flags value of O_WRONLY is invalid. The hash size
1727    is advisory, use zero for a default value.
1728
1729    Return is NULL on error, in which case errno is also set.  Don't 
1730    try to call tdb_error or tdb_errname, just do strerror(errno).
1731
1732    @param name may be NULL for internal databases. */
1733 TDB_CONTEXT *tdb_open(const char *name, int hash_size, int tdb_flags,
1734                       int open_flags, mode_t mode)
1735 {
1736         return tdb_open_ex(name, hash_size, tdb_flags, open_flags, mode, NULL, NULL);
1737 }
1738
1739 /* a default logging function */
1740 static void null_log_fn(TDB_CONTEXT *tdb, int level, const char *fmt, ...)
1741 {
1742 }
1743
1744
1745 TDB_CONTEXT *tdb_open_ex(const char *name, int hash_size, int tdb_flags,
1746                          int open_flags, mode_t mode,
1747                          tdb_log_func log_fn,
1748                          tdb_hash_func hash_fn)
1749 {
1750         TDB_CONTEXT *tdb;
1751         struct stat st;
1752         int rev = 0, locked = 0;
1753         uint8_t *vp;
1754         u32 vertest;
1755
1756         if (!(tdb = calloc(1, sizeof *tdb))) {
1757                 /* Can't log this */
1758                 errno = ENOMEM;
1759                 goto fail;
1760         }
1761         tdb->fd = -1;
1762         tdb->name = NULL;
1763         tdb->map_ptr = NULL;
1764         tdb->flags = tdb_flags;
1765         tdb->open_flags = open_flags;
1766         tdb->log_fn = log_fn?log_fn:null_log_fn;
1767         tdb->hash_fn = hash_fn ? hash_fn : default_tdb_hash;
1768
1769         if ((open_flags & O_ACCMODE) == O_WRONLY) {
1770                 TDB_LOG((tdb, 0, "tdb_open_ex: can't open tdb %s write-only\n",
1771                          name));
1772                 errno = EINVAL;
1773                 goto fail;
1774         }
1775         
1776         if (hash_size == 0)
1777                 hash_size = DEFAULT_HASH_SIZE;
1778         if ((open_flags & O_ACCMODE) == O_RDONLY) {
1779                 tdb->read_only = 1;
1780                 /* read only databases don't do locking or clear if first */
1781                 tdb->flags |= TDB_NOLOCK;
1782                 tdb->flags &= ~TDB_CLEAR_IF_FIRST;
1783         }
1784
1785         /* internal databases don't mmap or lock, and start off cleared */
1786         if (tdb->flags & TDB_INTERNAL) {
1787                 tdb->flags |= (TDB_NOLOCK | TDB_NOMMAP);
1788                 tdb->flags &= ~TDB_CLEAR_IF_FIRST;
1789                 if (tdb_new_database(tdb, hash_size) != 0) {
1790                         TDB_LOG((tdb, 0, "tdb_open_ex: tdb_new_database failed!"));
1791                         goto fail;
1792                 }
1793                 goto internal;
1794         }
1795
1796         if ((tdb->fd = open(name, open_flags, mode)) == -1) {
1797                 TDB_LOG((tdb, 5, "tdb_open_ex: could not open file %s: %s\n",
1798                          name, strerror(errno)));
1799                 goto fail;      /* errno set by open(2) */
1800         }
1801
1802         /* ensure there is only one process initialising at once */
1803         if (tdb_brlock(tdb, GLOBAL_LOCK, F_WRLCK, F_SETLKW, 0) == -1) {
1804                 TDB_LOG((tdb, 0, "tdb_open_ex: failed to get global lock on %s: %s\n",
1805                          name, strerror(errno)));
1806                 goto fail;      /* errno set by tdb_brlock */
1807         }
1808
1809         /* we need to zero database if we are the only one with it open */
1810         if ((tdb_flags & TDB_CLEAR_IF_FIRST) &&
1811                 (locked = (tdb_brlock(tdb, ACTIVE_LOCK, F_WRLCK, F_SETLK, 0) == 0))) {
1812                 open_flags |= O_CREAT;
1813                 if (ftruncate(tdb->fd, 0) == -1) {
1814                         TDB_LOG((tdb, 0, "tdb_open_ex: "
1815                                  "failed to truncate %s: %s\n",
1816                                  name, strerror(errno)));
1817                         goto fail; /* errno set by ftruncate */
1818                 }
1819         }
1820
1821         if (read(tdb->fd, &tdb->header, sizeof(tdb->header)) != sizeof(tdb->header)
1822             || strcmp(tdb->header.magic_food, TDB_MAGIC_FOOD) != 0
1823             || (tdb->header.version != TDB_VERSION
1824                 && !(rev = (tdb->header.version==TDB_BYTEREV(TDB_VERSION))))) {
1825                 /* its not a valid database - possibly initialise it */
1826                 if (!(open_flags & O_CREAT) || tdb_new_database(tdb, hash_size) == -1) {
1827                         errno = EIO; /* ie bad format or something */
1828                         goto fail;
1829                 }
1830                 rev = (tdb->flags & TDB_CONVERT);
1831         }
1832         vp = (uint8_t *)&tdb->header.version;
1833         vertest = (((u32)vp[0]) << 24) | (((u32)vp[1]) << 16) |
1834                   (((u32)vp[2]) << 8) | (u32)vp[3];
1835         tdb->flags |= (vertest==TDB_VERSION) ? TDB_BIGENDIAN : 0;
1836         if (!rev)
1837                 tdb->flags &= ~TDB_CONVERT;
1838         else {
1839                 tdb->flags |= TDB_CONVERT;
1840                 convert(&tdb->header, sizeof(tdb->header));
1841         }
1842         if (fstat(tdb->fd, &st) == -1)
1843                 goto fail;
1844
1845         /* Is it already in the open list?  If so, fail. */
1846         if (tdb_already_open(st.st_dev, st.st_ino)) {
1847                 TDB_LOG((tdb, 2, "tdb_open_ex: "
1848                          "%s (%d,%d) is already open in this process\n",
1849                          name, (int)st.st_dev, (int)st.st_ino));
1850                 errno = EBUSY;
1851                 goto fail;
1852         }
1853
1854         if (!(tdb->name = (char *)strdup(name))) {
1855                 errno = ENOMEM;
1856                 goto fail;
1857         }
1858
1859         tdb->map_size = st.st_size;
1860         tdb->device = st.st_dev;
1861         tdb->inode = st.st_ino;
1862         tdb->locked = calloc(tdb->header.hash_size+1, sizeof(tdb->locked[0]));
1863         if (!tdb->locked) {
1864                 TDB_LOG((tdb, 2, "tdb_open_ex: "
1865                          "failed to allocate lock structure for %s\n",
1866                          name));
1867                 errno = ENOMEM;
1868                 goto fail;
1869         }
1870         tdb_mmap(tdb);
1871         if (locked) {
1872                 if (!tdb->read_only)
1873                         if (tdb_clear_spinlocks(tdb) != 0) {
1874                                 TDB_LOG((tdb, 0, "tdb_open_ex: "
1875                                 "failed to clear spinlock\n"));
1876                                 goto fail;
1877                         }
1878                 if (tdb_brlock(tdb, ACTIVE_LOCK, F_UNLCK, F_SETLK, 0) == -1) {
1879                         TDB_LOG((tdb, 0, "tdb_open_ex: "
1880                                  "failed to take ACTIVE_LOCK on %s: %s\n",
1881                                  name, strerror(errno)));
1882                         goto fail;
1883                 }
1884
1885         }
1886
1887         /* We always need to do this if the CLEAR_IF_FIRST flag is set, even if
1888            we didn't get the initial exclusive lock as we need to let all other
1889            users know we're using it. */
1890
1891         if (tdb_flags & TDB_CLEAR_IF_FIRST) {
1892         /* leave this lock in place to indicate it's in use */
1893         if (tdb_brlock(tdb, ACTIVE_LOCK, F_RDLCK, F_SETLKW, 0) == -1)
1894                 goto fail;
1895         }
1896
1897
1898  internal:
1899         /* Internal (memory-only) databases skip all the code above to
1900          * do with disk files, and resume here by releasing their
1901          * global lock and hooking into the active list. */
1902         if (tdb_brlock(tdb, GLOBAL_LOCK, F_UNLCK, F_SETLKW, 0) == -1)
1903                 goto fail;
1904         tdb->next = tdbs;
1905         tdbs = tdb;
1906         return tdb;
1907
1908  fail:
1909         { int save_errno = errno;
1910
1911         if (!tdb)
1912                 return NULL;
1913         
1914         if (tdb->map_ptr) {
1915                 if (tdb->flags & TDB_INTERNAL)
1916                         SAFE_FREE(tdb->map_ptr);
1917                 else
1918                         tdb_munmap(tdb);
1919         }
1920         SAFE_FREE(tdb->name);
1921         if (tdb->fd != -1)
1922                 if (close(tdb->fd) != 0)
1923                         TDB_LOG((tdb, 5, "tdb_open_ex: failed to close tdb->fd on error!\n"));
1924         SAFE_FREE(tdb->locked);
1925         SAFE_FREE(tdb);
1926         errno = save_errno;
1927         return NULL;
1928         }
1929 }
1930
1931 /**
1932  * Close a database.
1933  *
1934  * @returns -1 for error; 0 for success.
1935  **/
1936 int tdb_close(TDB_CONTEXT *tdb)
1937 {
1938         TDB_CONTEXT **i;
1939         int ret = 0;
1940
1941         if (tdb->map_ptr) {
1942                 if (tdb->flags & TDB_INTERNAL)
1943                         SAFE_FREE(tdb->map_ptr);
1944                 else
1945                         tdb_munmap(tdb);
1946         }
1947         SAFE_FREE(tdb->name);
1948         if (tdb->fd != -1)
1949                 ret = close(tdb->fd);
1950         SAFE_FREE(tdb->locked);
1951
1952         /* Remove from contexts list */
1953         for (i = &tdbs; *i; i = &(*i)->next) {
1954                 if (*i == tdb) {
1955                         *i = tdb->next;
1956                         break;
1957                 }
1958         }
1959
1960         memset(tdb, 0, sizeof(*tdb));
1961         SAFE_FREE(tdb);
1962
1963         return ret;
1964 }
1965
1966 /* lock/unlock entire database */
1967 int tdb_lockall(TDB_CONTEXT *tdb)
1968 {
1969         u32 i;
1970
1971         /* There are no locks on read-only dbs */
1972         if (tdb->read_only)
1973                 return TDB_ERRCODE(TDB_ERR_LOCK, -1);
1974         for (i = 0; i < tdb->header.hash_size; i++) 
1975                 if (tdb_lock(tdb, i, F_WRLCK))
1976                         break;
1977
1978         /* If error, release locks we have... */
1979         if (i < tdb->header.hash_size) {
1980                 u32 j;
1981
1982                 for ( j = 0; j < i; j++)
1983                         tdb_unlock(tdb, j, F_WRLCK);
1984                 return TDB_ERRCODE(TDB_ERR_NOLOCK, -1);
1985         }
1986
1987         return 0;
1988 }
1989 void tdb_unlockall(TDB_CONTEXT *tdb)
1990 {
1991         u32 i;
1992         for (i=0; i < tdb->header.hash_size; i++)
1993                 tdb_unlock(tdb, i, F_WRLCK);
1994 }
1995
1996 /* lock/unlock one hash chain. This is meant to be used to reduce
1997    contention - it cannot guarantee how many records will be locked */
1998 int tdb_chainlock(TDB_CONTEXT *tdb, TDB_DATA key)
1999 {
2000         return tdb_lock(tdb, BUCKET(tdb->hash_fn(&key)), F_WRLCK);
2001 }
2002
2003 int tdb_chainunlock(TDB_CONTEXT *tdb, TDB_DATA key)
2004 {
2005         return tdb_unlock(tdb, BUCKET(tdb->hash_fn(&key)), F_WRLCK);
2006 }
2007
2008 int tdb_chainlock_read(TDB_CONTEXT *tdb, TDB_DATA key)
2009 {
2010         return tdb_lock(tdb, BUCKET(tdb->hash_fn(&key)), F_RDLCK);
2011 }
2012
2013 int tdb_chainunlock_read(TDB_CONTEXT *tdb, TDB_DATA key)
2014 {
2015         return tdb_unlock(tdb, BUCKET(tdb->hash_fn(&key)), F_RDLCK);
2016 }
2017
2018
2019 /* register a loging function */
2020 void tdb_logging_function(TDB_CONTEXT *tdb, void (*fn)(TDB_CONTEXT *, int , const char *, ...))
2021 {
2022         tdb->log_fn = fn?fn:null_log_fn;
2023 }
2024
2025
2026 /* reopen a tdb - this can be used after a fork to ensure that we have an independent
2027    seek pointer from our parent and to re-establish locks */
2028 int tdb_reopen(TDB_CONTEXT *tdb)
2029 {
2030         struct stat st;
2031
2032         if (tdb->flags & TDB_INTERNAL)
2033                 return 0; /* Nothing to do. */
2034         if (tdb_munmap(tdb) != 0) {
2035                 TDB_LOG((tdb, 0, "tdb_reopen: munmap failed (%s)\n", strerror(errno)));
2036                 goto fail;
2037         }
2038         if (close(tdb->fd) != 0)
2039                 TDB_LOG((tdb, 0, "tdb_reopen: WARNING closing tdb->fd failed!\n"));
2040         tdb->fd = open(tdb->name, tdb->open_flags & ~(O_CREAT|O_TRUNC), 0);
2041         if (tdb->fd == -1) {
2042                 TDB_LOG((tdb, 0, "tdb_reopen: open failed (%s)\n", strerror(errno)));
2043                 goto fail;
2044         }
2045         if (fstat(tdb->fd, &st) != 0) {
2046                 TDB_LOG((tdb, 0, "tdb_reopen: fstat failed (%s)\n", strerror(errno)));
2047                 goto fail;
2048         }
2049         if (st.st_ino != tdb->inode || st.st_dev != tdb->device) {
2050                 TDB_LOG((tdb, 0, "tdb_reopen: file dev/inode has changed!\n"));
2051                 goto fail;
2052         }
2053         tdb_mmap(tdb);
2054         if ((tdb->flags & TDB_CLEAR_IF_FIRST) && (tdb_brlock(tdb, ACTIVE_LOCK, F_RDLCK, F_SETLKW, 0) == -1)) {
2055                 TDB_LOG((tdb, 0, "tdb_reopen: failed to obtain active lock\n"));
2056                 goto fail;
2057         }
2058
2059         return 0;
2060
2061 fail:
2062         tdb_close(tdb);
2063         return -1;
2064 }
2065
2066 /* reopen all tdb's */
2067 int tdb_reopen_all(void)
2068 {
2069         TDB_CONTEXT *tdb;
2070
2071         for (tdb=tdbs; tdb; tdb = tdb->next) {
2072                 /* Ensure no clear-if-first. */
2073                 tdb->flags &= ~TDB_CLEAR_IF_FIRST;
2074                 if (tdb_reopen(tdb) != 0)
2075                         return -1;
2076         }
2077
2078         return 0;
2079 }