Changeset 745 for trunk/server/lib/tdb/common/lock.c

Timestamp:

Nov 27, 2012, 4:43:17 PM (13 years ago)

Author:

Silvan Scherrer

Message:

Samba Server: updated trunk to 3.6.0

Location:

trunk/server

Files:

• . (modified) (1 prop)
• lib/tdb/common/lock.c (modified) (37 diffs)

trunk/server/lib/tdb/common/lock.c

@@ -7 +7 @@
    Copyright (C) Paul `Rusty' Russell              2000
    Copyright (C) Jeremy Allison                    2000-2003
-   
+
      ** NOTE! The following LGPL license applies to the tdb
      ** library. This does NOT imply that all of Samba is released
      ** under the LGPL
-   
+
    This library is free software; you can redistribute it and/or
    modify it under the terms of the GNU Lesser General Public
@@ -27 +27 @@
 
 #include "tdb_private.h"
-
-#define TDB_MARK_LOCK 0x80000000
-
 #ifdef __OS2__
 
-static char* lock_type( int lck)
-{
-        static char buffer[16];
-        switch(lck) {
-        case F_GETLK: return "F_GETLK";
-        case F_SETLK: return "F_SETLK";
-        case F_SETLKW: return "F_SETLKW";
-        default:
-                sprintf( buffer, "unknown %d", lck);
-        }
-        return buffer;
+static char* lock_type( bool waitflag)
+{
+        if (waitflag)
+                return "F_SETLKW";
+        else
+                return "F_SETLK";
 }
 
@@ -59 +51 @@
 
 static int _mutex_brlock(struct tdb_context *tdb, tdb_off_t offset, 
-               int rw_type, int lck_type, int probe, size_t len)
+               int rw_type, bool waitflag, size_t len)
 {
         HMTX    hSem;
@@ -66 +58 @@
         
         switch( offset) {
-        case GLOBAL_LOCK:
+        case OPEN_LOCK:
                 hSem = tdb->hGlobalLock;
                 break;
@@ -86 +78 @@
         TDB_LOG((tdb, TDB_DEBUG_TRACE,"_mutex_brlock handle %d, offset %d\n", hSem, offset));
 
-        if (lck_type == F_SETLKW)
+        if (waitflag)
                 ulTimeout = SEM_INDEFINITE_WAIT;
         else
@@ -112 +104 @@
         errno = EINVAL;
         TDB_LOG(( tdb, TDB_DEBUG_ERROR, "_mutex_brlock pid %X, failed (fd=%d) at offset %d rw_type=%d lck_type=%d len=%d, rc=%d\n", 
-                 getpid(), tdb->fd, offset, rw_type, lck_type, (int)len, rc));
+                 getpid(), tdb->fd, offset, rw_type, waitflag, (int)len, rc));
         tdb->ecode = TDB_ERR_LOCK;
         return -1;
@@ -118 +110 @@
 #endif
 
-void tdb_setalarm_sigptr(struct tdb_context *tdb, volatile sig_atomic_t *ptr)
+_PUBLIC_ void tdb_setalarm_sigptr(struct tdb_context *tdb, volatile sig_atomic_t *ptr)
 {
         tdb->interrupt_sig_ptr = ptr;
 }
 
-/* a byte range locking function - return 0 on success
-   this functions locks/unlocks 1 byte at the specified offset.
-
-   On error, errno is also set so that errors are passed back properly
-   through tdb_open(). 
-
-   note that a len of zero means lock to end of file
-*/
-int tdb_brlock(struct tdb_context *tdb, tdb_off_t offset, 
-               int rw_type, int lck_type, int probe, size_t len)
-{
-
+static int fcntl_lock(struct tdb_context *tdb,
+                      int rw, tdb_off_t off, size_t len, bool waitflag)
+{
 #ifdef __OS2__
         APIRET      rc;
-        ULONG       fAccess = 0;
-        int         fLock = 0;
+        ULONG       fAccess = 0; // default exclusiv
+        int         fLock = 1;   // default lock 
         off_t       cbFile;
         off_t       offStart;
         off_t       cbRange;
-
-        TDB_LOG((tdb, TDB_DEBUG_TRACE, "tdb_brlock pid %X, fd %d, lck_type %s, rw_type %s, offset %d, len %d\n",
-                getpid(), tdb->fd, lock_type(lck_type), read_type(rw_type), offset, len));
-
-        if (tdb->flags & TDB_NOLOCK) {
-                return 0;
-        }
-
-        if ((rw_type == F_WRLCK) && (tdb->read_only || tdb->traverse_read)) {
-                tdb->ecode = TDB_ERR_RDONLY;
-                return -1;
-        }
+        FILELOCK    lockArea = {0}, unlockArea = {0};
+
+        TDB_LOG((tdb, TDB_DEBUG_TRACE, "fcntl_lock in pid %X, fd %d, lck_type %s, rw_type %s, offset %d, len %d\n",
+                getpid(), tdb->fd, lock_type(waitflag), read_type(rw), off, len));
         
-        switch( offset) {
-        case GLOBAL_LOCK:
+        switch(off) {
+        case OPEN_LOCK:
         case ACTIVE_LOCK:
         case TRANSACTION_LOCK:
-                return _mutex_brlock( tdb, offset, rw_type,  lck_type,  probe, len);
+                return _mutex_brlock(tdb, off, rw, waitflag, len);
         }
 
         /* flags and order */
-        fAccess = 0; /* exclusive */
-        switch (rw_type)
+        switch (rw)
         {
                 case F_UNLCK:
                         fLock = 0;
+                        unlockArea.lOffset = off;
+                        unlockArea.lRange  = len ? len : tdb->header.hash_size *4; //was LONG_MAX
                         break;
                 case F_RDLCK:
+                        lockArea.lOffset = off;
+                        lockArea.lRange  = len ? len : LONG_MAX;
                         fAccess = 1; /* read-only */
+                        break;
                 case F_WRLCK:
-                        fLock = 1;
+                        lockArea.lOffset = off;
+                        lockArea.lRange  = len ? len : LONG_MAX;
                         break;
                 default:
@@ -178 +158 @@
         }
                 
-        FILELOCK   aflock[2];
-        bzero(&aflock[(fLock + 1) & 1], sizeof(aflock[0]));
-        aflock[fLock].lOffset = offset;
-        aflock[fLock].lRange  = len ? len : LONG_MAX;
-        rc = DosSetFileLocks(tdb->fd, &aflock[0], &aflock[1], SEM_IMMEDIATE_RETURN, fAccess);
-
-        if (rc != NO_ERROR && lck_type == F_SETLKW) {
+        rc = DosSetFileLocks(tdb->fd, &unlockArea, &lockArea, SEM_IMMEDIATE_RETURN, fAccess);
+
+        if (rc != NO_ERROR && waitflag) {
                 int     count = 20;
                 do {
-                        rc = DosSetFileLocks(tdb->fd, &aflock[0], &aflock[1], 100, fAccess);
+                        rc = DosSetFileLocks(tdb->fd, &unlockArea, &lockArea, 100, fAccess);
                         count--;
                 } while( count>0 && rc !=NO_ERROR);
 
         }
+
+        TDB_LOG(( tdb, TDB_DEBUG_TRACE, "fcntl_lock out pid %X, fd %d, lck_type %s, rw_type %s, offset %d, len %d, rc=%d\n",
+                getpid(), tdb->fd, lock_type(waitflag), read_type(rw), off, len, rc));
+
         if (rc != NO_ERROR) {
                 errno  = EINVAL;
-                /* Generic lock error. errno set by fcntl.
-                 * EAGAIN is an expected return from non-blocking
-                 * locks. */
-                if (!probe && lck_type != F_SETLK) {
-                        /* Ensure error code is set for log fun to examine. */
-                        tdb->ecode = TDB_ERR_LOCK;
-                        TDB_LOG((tdb, TDB_DEBUG_TRACE,"tdb_brlock failed (fd=%d) at offset %d rw_type=%d lck_type=%d len=%d\n", 
-                                 tdb->fd, offset, rw_type, lck_type, (int)len));
+                return -1;
+        }
+
+#else
+        struct flock fl;
+
+        fl.l_type = rw;
+        fl.l_whence = SEEK_SET;
+        fl.l_start = off;
+        fl.l_len = len;
+        fl.l_pid = 0;
+
+        if (waitflag)
+                return fcntl(tdb->fd, F_SETLKW, &fl);
+        else
+                return fcntl(tdb->fd, F_SETLK, &fl);
+#endif
+}
+
+static int fcntl_unlock(struct tdb_context *tdb, int rw, tdb_off_t off, size_t len)
+{
+        struct flock fl;
+#if 0 /* Check they matched up locks and unlocks correctly. */
+        char line[80];
+        FILE *locks;
+        bool found = false;
+
+        locks = fopen("/proc/locks", "r");
+
+        while (fgets(line, 80, locks)) {
+                char *p;
+                int type, start, l;
+
+                /* eg. 1: FLOCK  ADVISORY  WRITE 2440 08:01:2180826 0 EOF */
+                p = strchr(line, ':') + 1;
+                if (strncmp(p, " POSIX  ADVISORY  ", strlen(" POSIX  ADVISORY  ")))
+                        continue;
+                p += strlen(" FLOCK  ADVISORY  ");
+                if (strncmp(p, "READ  ", strlen("READ  ")) == 0)
+                        type = F_RDLCK;
+                else if (strncmp(p, "WRITE ", strlen("WRITE ")) == 0)
+                        type = F_WRLCK;
+                else
+                        abort();
+                p += 6;
+                if (atoi(p) != getpid())
+                        continue;
+                p = strchr(strchr(p, ' ') + 1, ' ') + 1;
+                start = atoi(p);
+                p = strchr(p, ' ') + 1;
+                if (strncmp(p, "EOF", 3) == 0)
+                        l = 0;
+                else
+                        l = atoi(p) - start + 1;
+
+                if (off == start) {
+                        if (len != l) {
+                                fprintf(stderr, "Len %u should be %u: %s",
+                                        (int)len, l, line);
+                                abort();
+                        }
+                        if (type != rw) {
+                                fprintf(stderr, "Type %s wrong: %s",
+                                        rw == F_RDLCK ? "READ" : "WRITE", line);
+                                abort();
+                        }
+                        found = true;
+                        break;
                 }
-
-                TDB_LOG(( tdb, TDB_DEBUG_TRACE, "tdb_brlock pid %X, failed (fd=%d) at offset %d rw_type=%d lck_type=%d len=%d\n", 
-                         getpid(), tdb->fd, offset, rw_type, lck_type, (int)len));
-                tdb->ecode = TDB_ERR_LOCK;
-                return -1;
-        }
-
-        TDB_LOG(( tdb, TDB_DEBUG_TRACE, "tdb_brlock pid %X, fd %d, lck_type %s, rw_type %s, offset %d, len %d DONE\n",
-                getpid(), tdb->fd, lock_type(lck_type), read_type(rw_type), offset, len));
-
+        }
+
+        if (!found) {
+                fprintf(stderr, "Unlock on %u@%u not found!\n",
+                        (int)off, (int)len);
+                abort();
+        }
+
+        fclose(locks);
+#endif
+
+        fl.l_type = F_UNLCK;
+        fl.l_whence = SEEK_SET;
+        fl.l_start = off;
+        fl.l_len = len;
+        fl.l_pid = 0;
+#ifdef __OS2__
+        return fcntl_lock(tdb, F_UNLCK, off, len, 1);
 #else
-
-        struct flock fl;
+        return fcntl(tdb->fd, F_SETLKW, &fl);
+#endif
+}
+
+/* list -1 is the alloc list, otherwise a hash chain. */
+static tdb_off_t lock_offset(int list)
+{
+        return FREELIST_TOP + 4*list;
+}
+
+/* a byte range locking function - return 0 on success
+   this functions locks/unlocks 1 byte at the specified offset.
+
+   On error, errno is also set so that errors are passed back properly
+   through tdb_open(). 
+
+   note that a len of zero means lock to end of file
+*/
+int tdb_brlock(struct tdb_context *tdb,
+               int rw_type, tdb_off_t offset, size_t len,
+               enum tdb_lock_flags flags)
+{
         int ret;
 
         if (tdb->flags & TDB_NOLOCK) {
+                return 0;
+        }
+
+        if (flags & TDB_LOCK_MARK_ONLY) {
                 return 0;
         }
@@ -227 +299 @@
         }
 
-        fl.l_type = rw_type;
-        fl.l_whence = SEEK_SET;
-        fl.l_start = offset;
-        fl.l_len = len;
-        fl.l_pid = 0;
-
         do {
-                ret = fcntl(tdb->fd,lck_type,&fl);
-
+                ret = fcntl_lock(tdb, rw_type, offset, len,
+                                 flags & TDB_LOCK_WAIT);
                 /* Check for a sigalarm break. */
                 if (ret == -1 && errno == EINTR &&
@@ -249 +315 @@
                  * EAGAIN is an expected return from non-blocking
                  * locks. */
-                if (!probe && lck_type != F_SETLK) {
-                        TDB_LOG((tdb, TDB_DEBUG_TRACE,"tdb_brlock failed (fd=%d) at offset %d rw_type=%d lck_type=%d len=%d\n", 
-                                 tdb->fd, offset, rw_type, lck_type, (int)len));
+                if (!(flags & TDB_LOCK_PROBE) && errno != EAGAIN) {
+                        TDB_LOG((tdb, TDB_DEBUG_TRACE,"tdb_brlock failed (fd=%d) at offset %d rw_type=%d flags=%d len=%d\n",
+                                 tdb->fd, offset, rw_type, flags, (int)len));
                 }
                 return -1;
         }
         return 0;
-#endif
-}
-
+}
+
+int tdb_brunlock(struct tdb_context *tdb,
+                 int rw_type, tdb_off_t offset, size_t len)
+{
+        int ret;
+
+        if (tdb->flags & TDB_NOLOCK) {
+                return 0;
+        }
+
+        do {
+                ret = fcntl_unlock(tdb, rw_type, offset, len);
+        } while (ret == -1 && errno == EINTR);
+
+        if (ret == -1) {
+                TDB_LOG((tdb, TDB_DEBUG_TRACE,"tdb_brunlock failed (fd=%d) at offset %d rw_type=%d len=%d\n",
+                         tdb->fd, offset, rw_type, (int)len));
+        }
+        return ret;
+}
 
 /*
@@ -266 +350 @@
   made. For those OSes we may loop for a while.  
 */
-int tdb_brlock_upgrade(struct tdb_context *tdb, tdb_off_t offset, size_t len)
+int tdb_allrecord_upgrade(struct tdb_context *tdb)
 {
         int count = 1000;
+
+        if (tdb->allrecord_lock.count != 1) {
+                TDB_LOG((tdb, TDB_DEBUG_ERROR,
+                         "tdb_allrecord_upgrade failed: count %u too high\n",
+                         tdb->allrecord_lock.count));
+                return -1;
+        }
+
+        if (tdb->allrecord_lock.off != 1) {
+                TDB_LOG((tdb, TDB_DEBUG_ERROR,
+                         "tdb_allrecord_upgrade failed: already upgraded?\n"));
+                return -1;
+        }
+
         while (count--) {
                 struct timeval tv;
-
 #ifdef __OS2__
                 // YD we cannot upgrade without an unlock first...
-                tdb_brlock(tdb, offset, F_UNLCK, F_SETLKW, 1, len);
+                tdb_brlock(tdb, F_UNLCK, FREELIST_TOP, 0, TDB_LOCK_WAIT|TDB_LOCK_PROBE);
 #endif
-
-                if (tdb_brlock(tdb, offset, F_WRLCK, F_SETLKW, 1, len) == 0) {
+                if (tdb_brlock(tdb, F_WRLCK, FREELIST_TOP, 0,
+                               TDB_LOCK_WAIT|TDB_LOCK_PROBE) == 0) {
+                        tdb->allrecord_lock.ltype = F_WRLCK;
+                        tdb->allrecord_lock.off = 0;
                         return 0;
                 }
@@ -288 +387 @@
                 select(0, NULL, NULL, NULL, &tv);
         }
-        TDB_LOG((tdb, TDB_DEBUG_TRACE,"tdb_brlock_upgrade failed at offset %d\n", offset));
+        TDB_LOG((tdb, TDB_DEBUG_TRACE,"tdb_allrecord_upgrade failed\n"));
         return -1;
 }
 
-
-/* lock a list in the database. list -1 is the alloc list */
-static int _tdb_lock(struct tdb_context *tdb, int list, int ltype, int op)
+static struct tdb_lock_type *find_nestlock(struct tdb_context *tdb,
+                                           tdb_off_t offset)
+{
+        unsigned int i;
+
+        for (i=0; i<tdb->num_lockrecs; i++) {
+                if (tdb->lockrecs[i].off == offset) {
+                        return &tdb->lockrecs[i];
+                }
+        }
+        return NULL;
+}
+
+/* lock an offset in the database. */
+int tdb_nest_lock(struct tdb_context *tdb, uint32_t offset, int ltype,
+                  enum tdb_lock_flags flags)
 {
         struct tdb_lock_type *new_lck;
-        int i;
-        bool mark_lock = ((ltype & TDB_MARK_LOCK) == TDB_MARK_LOCK);
-
-        ltype &= ~TDB_MARK_LOCK;
-
-        /* a global lock allows us to avoid per chain locks */
-        if (tdb->global_lock.count && 
-            (ltype == tdb->global_lock.ltype || ltype == F_RDLCK)) {
-                return 0;
-        }
-
-        if (tdb->global_lock.count) {
+
+        if (offset >= lock_offset(tdb->header.hash_size)) {
                 tdb->ecode = TDB_ERR_LOCK;
-                return -1;
-        }
-
-        if (list < -1 || list >= (int)tdb->header.hash_size) {
-                tdb->ecode = TDB_ERR_LOCK;
-                TDB_LOG((tdb, TDB_DEBUG_ERROR,"tdb_lock: invalid list %d for ltype=%d\n", 
-                           list, ltype));
+                TDB_LOG((tdb, TDB_DEBUG_ERROR,"tdb_lock: invalid offset %u for ltype=%d\n",
+                         offset, ltype));
                 return -1;
         }
@@ -322 +419 @@
                 return 0;
 
-        for (i=0; i<tdb->num_lockrecs; i++) {
-                if (tdb->lockrecs[i].list == list) {
-                        if (tdb->lockrecs[i].count == 0) {
-                                /*
-                                 * Can't happen, see tdb_unlock(). It should
-                                 * be an assert.
-                                 */
-                                TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_lock: "
-                                         "lck->count == 0 for list %d", list));
-                        }
-                        /*
-                         * Just increment the in-memory struct, posix locks
-                         * don't stack.
-                         */
-                        tdb->lockrecs[i].count++;
-                        return 0;
-                }
+        new_lck = find_nestlock(tdb, offset);
+        if (new_lck) {
+                /*
+                 * Just increment the in-memory struct, posix locks
+                 * don't stack.
+                 */
+                new_lck->count++;
+                return 0;
         }
 
@@ -352 +440 @@
         /* Since fcntl locks don't nest, we do a lock for the first one,
            and simply bump the count for future ones */
-        if (!mark_lock &&
-            tdb->methods->tdb_brlock(tdb,FREELIST_TOP+4*list, ltype, op,
-                                     0, 1)) {
-                return -1;
-        }
-
-        tdb->num_locks++;
-
-        tdb->lockrecs[tdb->num_lockrecs].list = list;
+        if (tdb_brlock(tdb, ltype, offset, 1, flags)) {
+                return -1;
+        }
+
+        tdb->lockrecs[tdb->num_lockrecs].off = offset;
         tdb->lockrecs[tdb->num_lockrecs].count = 1;
         tdb->lockrecs[tdb->num_lockrecs].ltype = ltype;
-        tdb->num_lockrecs += 1;
+        tdb->num_lockrecs++;
 
         return 0;
+}
+
+static int tdb_lock_and_recover(struct tdb_context *tdb)
+{
+        int ret;
+
+        /* We need to match locking order in transaction commit. */
+        if (tdb_brlock(tdb, F_WRLCK, FREELIST_TOP, 0, TDB_LOCK_WAIT)) {
+                return -1;
+        }
+
+        if (tdb_brlock(tdb, F_WRLCK, OPEN_LOCK, 1, TDB_LOCK_WAIT)) {
+                tdb_brunlock(tdb, F_WRLCK, FREELIST_TOP, 0);
+                return -1;
+        }
+
+        ret = tdb_transaction_recover(tdb);
+
+        tdb_brunlock(tdb, F_WRLCK, OPEN_LOCK, 1);
+        tdb_brunlock(tdb, F_WRLCK, FREELIST_TOP, 0);
+
+        return ret;
+}
+
+static bool have_data_locks(const struct tdb_context *tdb)
+{
+        unsigned int i;
+
+        for (i = 0; i < tdb->num_lockrecs; i++) {
+                if (tdb->lockrecs[i].off >= lock_offset(-1))
+                        return true;
+        }
+        return false;
+}
+
+static int tdb_lock_list(struct tdb_context *tdb, int list, int ltype,
+                         enum tdb_lock_flags waitflag)
+{
+        int ret;
+        bool check = false;
+
+        /* a allrecord lock allows us to avoid per chain locks */
+        if (tdb->allrecord_lock.count &&
+            (ltype == tdb->allrecord_lock.ltype || ltype == F_RDLCK)) {
+                return 0;
+        }
+
+        if (tdb->allrecord_lock.count) {
+                tdb->ecode = TDB_ERR_LOCK;
+                ret = -1;
+        } else {
+                /* Only check when we grab first data lock. */
+                check = !have_data_locks(tdb);
+                ret = tdb_nest_lock(tdb, lock_offset(list), ltype, waitflag);
+
+                if (ret == 0 && check && tdb_needs_recovery(tdb)) {
+                        tdb_nest_unlock(tdb, lock_offset(list), ltype, false);
+
+                        if (tdb_lock_and_recover(tdb) == -1) {
+                                return -1;
+                        }
+                        return tdb_lock_list(tdb, list, ltype, waitflag);
+                }
+        }
+        return ret;
 }
 
@@ -372 +521 @@
 {
         int ret;
-        ret = _tdb_lock(tdb, list, ltype, F_SETLKW);
+
+        ret = tdb_lock_list(tdb, list, ltype, TDB_LOCK_WAIT);
         if (ret) {
                 TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_lock failed on list %d "
@@ -383 +533 @@
 int tdb_lock_nonblock(struct tdb_context *tdb, int list, int ltype)
 {
-        return _tdb_lock(tdb, list, ltype, F_SETLK);
-}
-
-
-/* unlock the database: returns void because it's too late for errors. */
-        /* changed to return int it may be interesting to know there
-           has been an error  --simo */
-int tdb_unlock(struct tdb_context *tdb, int list, int ltype)
+        return tdb_lock_list(tdb, list, ltype, TDB_LOCK_NOWAIT);
+}
+
+
+int tdb_nest_unlock(struct tdb_context *tdb, uint32_t offset, int ltype,
+                    bool mark_lock)
 {
         int ret = -1;
-        int i;
-        struct tdb_lock_type *lck = NULL;
-        bool mark_lock = ((ltype & TDB_MARK_LOCK) == TDB_MARK_LOCK);
-
-        ltype &= ~TDB_MARK_LOCK;
-
-        /* a global lock allows us to avoid per chain locks */
-        if (tdb->global_lock.count && 
-            (ltype == tdb->global_lock.ltype || ltype == F_RDLCK)) {
-                return 0;
-        }
-
-        if (tdb->global_lock.count) {
-                tdb->ecode = TDB_ERR_LOCK;
-                return -1;
-        }
+        struct tdb_lock_type *lck;
 
         if (tdb->flags & TDB_NOLOCK)
@@ -414 +547 @@
 
         /* Sanity checks */
-        if (list < -1 || list >= (int)tdb->header.hash_size) {
-                TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_unlock: list %d invalid (%d)\n", list, tdb->header.hash_size));
+        if (offset >= lock_offset(tdb->header.hash_size)) {
+                TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_unlock: offset %u invalid (%d)\n", offset, tdb->header.hash_size));
                 return ret;
         }
 
-        for (i=0; i<tdb->num_lockrecs; i++) {
-                if (tdb->lockrecs[i].list == list) {
-                        lck = &tdb->lockrecs[i];
-                        break;
-                }
-        }
-
+        lck = find_nestlock(tdb, offset);
         if ((lck == NULL) || (lck->count == 0)) {
                 TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_unlock: count is 0\n"));
@@ -446 +573 @@
                 ret = 0;
         } else {
-                ret = tdb->methods->tdb_brlock(tdb, FREELIST_TOP+4*list, F_UNLCK,
-                                               F_SETLKW, 0, 1);
-        }
-        tdb->num_locks--;
+                ret = tdb_brunlock(tdb, ltype, offset, 1);
+        }
 
         /*
@@ -455 +580 @@
          * last array element.
          */
-
-        if (tdb->num_lockrecs > 1) {
-                *lck = tdb->lockrecs[tdb->num_lockrecs-1];
-        }
-        tdb->num_lockrecs -= 1;
+        *lck = tdb->lockrecs[--tdb->num_lockrecs];
 
         /*
@@ -475 +596 @@
 }
 
+int tdb_unlock(struct tdb_context *tdb, int list, int ltype)
+{
+        /* a global lock allows us to avoid per chain locks */
+        if (tdb->allrecord_lock.count &&
+            (ltype == tdb->allrecord_lock.ltype || ltype == F_RDLCK)) {
+                return 0;
+        }
+
+        if (tdb->allrecord_lock.count) {
+                tdb->ecode = TDB_ERR_LOCK;
+                return -1;
+        }
+
+        return tdb_nest_unlock(tdb, lock_offset(list), ltype, false);
+}
+
 /*
   get the transaction lock
  */
-int tdb_transaction_lock(struct tdb_context *tdb, int ltype)
-{
-        if (tdb->global_lock.count) {
-                return 0;
-        }
-        if (tdb->transaction_lock_count > 0) {
-                tdb->transaction_lock_count++;
-                return 0;
-        }
-
-        if (tdb->methods->tdb_brlock(tdb, TRANSACTION_LOCK, ltype, 
-                                     F_SETLKW, 0, 1) == -1) {
-                TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_transaction_lock: failed to get transaction lock\n"));
-                tdb->ecode = TDB_ERR_LOCK;
-                return -1;
-        }
-        tdb->transaction_lock_count++;
-        return 0;
+int tdb_transaction_lock(struct tdb_context *tdb, int ltype,
+                         enum tdb_lock_flags lockflags)
+{
+        return tdb_nest_lock(tdb, TRANSACTION_LOCK, ltype, lockflags);
 }
 
@@ -501 +624 @@
   release the transaction lock
  */
-int tdb_transaction_unlock(struct tdb_context *tdb)
-{
-        int ret;
-        if (tdb->global_lock.count) {
-                return 0;
-        }
-        if (tdb->transaction_lock_count > 1) {
-                tdb->transaction_lock_count--;
-                return 0;
-        }
-        ret = tdb->methods->tdb_brlock(tdb, TRANSACTION_LOCK, F_UNLCK, F_SETLKW, 0, 1);
-        if (ret == 0) {
-                tdb->transaction_lock_count = 0;
-        }
-        return ret;
-}
-
-
-
-
-/* lock/unlock entire database */
-static int _tdb_lockall(struct tdb_context *tdb, int ltype, int op)
-{
-        bool mark_lock = ((ltype & TDB_MARK_LOCK) == TDB_MARK_LOCK);
-
-        ltype &= ~TDB_MARK_LOCK;
-
+int tdb_transaction_unlock(struct tdb_context *tdb, int ltype)
+{
+        return tdb_nest_unlock(tdb, TRANSACTION_LOCK, ltype, false);
+}
+
+/* Returns 0 if all done, -1 if error, 1 if ok. */
+static int tdb_allrecord_check(struct tdb_context *tdb, int ltype,
+                               enum tdb_lock_flags flags, bool upgradable)
+{
         /* There are no locks on read-only dbs */
         if (tdb->read_only || tdb->traverse_read) {
@@ -534 +639 @@
         }
 
-        if (tdb->global_lock.count && tdb->global_lock.ltype == ltype) {
-                tdb->global_lock.count++;
-                return 0;
-        }
-
-        if (tdb->global_lock.count) {
+        if (tdb->allrecord_lock.count && tdb->allrecord_lock.ltype == ltype) {
+                tdb->allrecord_lock.count++;
+                return 0;
+        }
+
+        if (tdb->allrecord_lock.count) {
                 /* a global lock of a different type exists */
                 tdb->ecode = TDB_ERR_LOCK;
                 return -1;
         }
-        
-        if (tdb->num_locks != 0) {
+
+        if (tdb_have_extra_locks(tdb)) {
                 /* can't combine global and chain locks */
                 tdb->ecode = TDB_ERR_LOCK;
@@ -551 +656 @@
         }
 
-        if (!mark_lock &&
-            tdb->methods->tdb_brlock(tdb, FREELIST_TOP, ltype, op,
-                                     0, 4*tdb->header.hash_size)) {
-                if (op == F_SETLKW) {
-                        TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_lockall failed (%s)\n", strerror(errno)));
+        if (upgradable && ltype != F_RDLCK) {
+                /* tdb error: you can't upgrade a write lock! */
+                tdb->ecode = TDB_ERR_LOCK;
+                return -1;
+        }
+        return 1;
+}
+
+/* We only need to lock individual bytes, but Linux merges consecutive locks
+ * so we lock in contiguous ranges. */
+static int tdb_chainlock_gradual(struct tdb_context *tdb,
+                                 int ltype, enum tdb_lock_flags flags,
+                                 size_t off, size_t len)
+{
+        int ret;
+        enum tdb_lock_flags nb_flags = (flags & ~TDB_LOCK_WAIT);
+
+        if (len <= 4) {
+                /* Single record.  Just do blocking lock. */
+                return tdb_brlock(tdb, ltype, off, len, flags);
+        }
+
+        /* First we try non-blocking. */
+        ret = tdb_brlock(tdb, ltype, off, len, nb_flags);
+        if (ret == 0) {
+                return 0;
+        }
+
+        /* Try locking first half, then second. */
+        ret = tdb_chainlock_gradual(tdb, ltype, flags, off, len / 2);
+        if (ret == -1)
+                return -1;
+
+        ret = tdb_chainlock_gradual(tdb, ltype, flags,
+                                    off + len / 2, len - len / 2);
+        if (ret == -1) {
+                tdb_brunlock(tdb, ltype, off, len / 2);
+                return -1;
+        }
+        return 0;
+}
+
+/* lock/unlock entire database.  It can only be upgradable if you have some
+ * other way of guaranteeing exclusivity (ie. transaction write lock).
+ * We do the locking gradually to avoid being starved by smaller locks. */
+int tdb_allrecord_lock(struct tdb_context *tdb, int ltype,
+                       enum tdb_lock_flags flags, bool upgradable)
+{
+        switch (tdb_allrecord_check(tdb, ltype, flags, upgradable)) {
+        case -1:
+                return -1;
+        case 0:
+                return 0;
+        }
+
+        /* We cover two kinds of locks:
+         * 1) Normal chain locks.  Taken for almost all operations.
+         * 3) Individual records locks.  Taken after normal or free
+         *    chain locks.
+         *
+         * It is (1) which cause the starvation problem, so we're only
+         * gradual for that. */
+        if (tdb_chainlock_gradual(tdb, ltype, flags, FREELIST_TOP,
+                                  tdb->header.hash_size * 4) == -1) {
+                return -1;
+        }
+
+        /* Grab individual record locks. */
+        if (tdb_brlock(tdb, ltype, lock_offset(tdb->header.hash_size), 0,
+                       flags) == -1) {
+                tdb_brunlock(tdb, ltype, FREELIST_TOP,
+                             tdb->header.hash_size * 4);
+                return -1;
+        }
+
+        tdb->allrecord_lock.count = 1;
+        /* If it's upgradable, it's actually exclusive so we can treat
+         * it as a write lock. */
+        tdb->allrecord_lock.ltype = upgradable ? F_WRLCK : ltype;
+        tdb->allrecord_lock.off = upgradable;
+
+        if (tdb_needs_recovery(tdb)) {
+                bool mark = flags & TDB_LOCK_MARK_ONLY;
+                tdb_allrecord_unlock(tdb, ltype, mark);
+                if (mark) {
+                        tdb->ecode = TDB_ERR_LOCK;
+                        TDB_LOG((tdb, TDB_DEBUG_ERROR,
+                                 "tdb_lockall_mark cannot do recovery\n"));
+                        return -1;
                 }
-                return -1;
-        }
-
-        tdb->global_lock.count = 1;
-        tdb->global_lock.ltype = ltype;
+                if (tdb_lock_and_recover(tdb) == -1) {
+                        return -1;
+                }
+                return tdb_allrecord_lock(tdb, ltype, flags, upgradable);
+        }
 
         return 0;
@@ -569 +758 @@
 
 /* unlock entire db */
-static int _tdb_unlockall(struct tdb_context *tdb, int ltype)
-{
-        bool mark_lock = ((ltype & TDB_MARK_LOCK) == TDB_MARK_LOCK);
-
-        ltype &= ~TDB_MARK_LOCK;
-
+int tdb_allrecord_unlock(struct tdb_context *tdb, int ltype, bool mark_lock)
+{
         /* There are no locks on read-only dbs */
         if (tdb->read_only || tdb->traverse_read) {
@@ -581 +766 @@
         }
 
-        if (tdb->global_lock.ltype != ltype || tdb->global_lock.count == 0) {
+        if (tdb->allrecord_lock.count == 0) {
                 tdb->ecode = TDB_ERR_LOCK;
                 return -1;
         }
 
-        if (tdb->global_lock.count > 1) {
-                tdb->global_lock.count--;
-                return 0;
-        }
-
-        if (!mark_lock &&
-            tdb->methods->tdb_brlock(tdb, FREELIST_TOP, F_UNLCK, F_SETLKW, 
-                                     0, 4*tdb->header.hash_size)) {
+        /* Upgradable locks are marked as write locks. */
+        if (tdb->allrecord_lock.ltype != ltype
+            && (!tdb->allrecord_lock.off || ltype != F_RDLCK)) {
+                tdb->ecode = TDB_ERR_LOCK;
+                return -1;
+        }
+
+        if (tdb->allrecord_lock.count > 1) {
+                tdb->allrecord_lock.count--;
+                return 0;
+        }
+
+        if (!mark_lock && tdb_brunlock(tdb, ltype, FREELIST_TOP, 0)) {
                 TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_unlockall failed (%s)\n", strerror(errno)));
                 return -1;
         }
 
-        tdb->global_lock.count = 0;
-        tdb->global_lock.ltype = 0;
+        tdb->allrecord_lock.count = 0;
+        tdb->allrecord_lock.ltype = 0;
 
         return 0;
@@ -605 +795 @@
 
 /* lock entire database with write lock */
-int tdb_lockall(struct tdb_context *tdb)
+_PUBLIC_ int tdb_lockall(struct tdb_context *tdb)
 {
         tdb_trace(tdb, "tdb_lockall");
-        return _tdb_lockall(tdb, F_WRLCK, F_SETLKW);
+        return tdb_allrecord_lock(tdb, F_WRLCK, TDB_LOCK_WAIT, false);
 }
 
 /* lock entire database with write lock - mark only */
-int tdb_lockall_mark(struct tdb_context *tdb)
+_PUBLIC_ int tdb_lockall_mark(struct tdb_context *tdb)
 {
         tdb_trace(tdb, "tdb_lockall_mark");
-        return _tdb_lockall(tdb, F_WRLCK | TDB_MARK_LOCK, F_SETLKW);
+        return tdb_allrecord_lock(tdb, F_WRLCK, TDB_LOCK_MARK_ONLY, false);
 }
 
 /* unlock entire database with write lock - unmark only */
-int tdb_lockall_unmark(struct tdb_context *tdb)
+_PUBLIC_ int tdb_lockall_unmark(struct tdb_context *tdb)
 {
         tdb_trace(tdb, "tdb_lockall_unmark");
-        return _tdb_unlockall(tdb, F_WRLCK | TDB_MARK_LOCK);
+        return tdb_allrecord_unlock(tdb, F_WRLCK, true);
 }
 
 /* lock entire database with write lock - nonblocking varient */
-int tdb_lockall_nonblock(struct tdb_context *tdb)
-{
-        int ret = _tdb_lockall(tdb, F_WRLCK, F_SETLK);
+_PUBLIC_ int tdb_lockall_nonblock(struct tdb_context *tdb)
+{
+        int ret = tdb_allrecord_lock(tdb, F_WRLCK, TDB_LOCK_NOWAIT, false);
         tdb_trace_ret(tdb, "tdb_lockall_nonblock", ret);
         return ret;
@@ -634 +824 @@
 
 /* unlock entire database with write lock */
-int tdb_unlockall(struct tdb_context *tdb)
+_PUBLIC_ int tdb_unlockall(struct tdb_context *tdb)
 {
         tdb_trace(tdb, "tdb_unlockall");
-        return _tdb_unlockall(tdb, F_WRLCK);
+        return tdb_allrecord_unlock(tdb, F_WRLCK, false);
 }
 
 /* lock entire database with read lock */
-int tdb_lockall_read(struct tdb_context *tdb)
+_PUBLIC_ int tdb_lockall_read(struct tdb_context *tdb)
 {
         tdb_trace(tdb, "tdb_lockall_read");
-        return _tdb_lockall(tdb, F_RDLCK, F_SETLKW);
+        return tdb_allrecord_lock(tdb, F_RDLCK, TDB_LOCK_WAIT, false);
 }
 
 /* lock entire database with read lock - nonblock varient */
-int tdb_lockall_read_nonblock(struct tdb_context *tdb)
-{
-        int ret = _tdb_lockall(tdb, F_RDLCK, F_SETLK);
+_PUBLIC_ int tdb_lockall_read_nonblock(struct tdb_context *tdb)
+{
+        int ret = tdb_allrecord_lock(tdb, F_RDLCK, TDB_LOCK_NOWAIT, false);
         tdb_trace_ret(tdb, "tdb_lockall_read_nonblock", ret);
         return ret;
@@ -656 +846 @@
 
 /* unlock entire database with read lock */
-int tdb_unlockall_read(struct tdb_context *tdb)
+_PUBLIC_ int tdb_unlockall_read(struct tdb_context *tdb)
 {
         tdb_trace(tdb, "tdb_unlockall_read");
-        return _tdb_unlockall(tdb, F_RDLCK);
+        return tdb_allrecord_unlock(tdb, F_RDLCK, false);
 }
 
 /* lock/unlock one hash chain. This is meant to be used to reduce
    contention - it cannot guarantee how many records will be locked */
-int tdb_chainlock(struct tdb_context *tdb, TDB_DATA key)
+_PUBLIC_ int tdb_chainlock(struct tdb_context *tdb, TDB_DATA key)
 {
         int ret = tdb_lock(tdb, BUCKET(tdb->hash_fn(&key)), F_WRLCK);
@@ -674 +864 @@
    to reduce contention - it cannot guarantee how many records will be
    locked */
-int tdb_chainlock_nonblock(struct tdb_context *tdb, TDB_DATA key)
+_PUBLIC_ int tdb_chainlock_nonblock(struct tdb_context *tdb, TDB_DATA key)
 {
         int ret = tdb_lock_nonblock(tdb, BUCKET(tdb->hash_fn(&key)), F_WRLCK);
@@ -682 +872 @@
 
 /* mark a chain as locked without actually locking it. Warning! use with great caution! */
-int tdb_chainlock_mark(struct tdb_context *tdb, TDB_DATA key)
-{
-        int ret = tdb_lock(tdb, BUCKET(tdb->hash_fn(&key)), F_WRLCK | TDB_MARK_LOCK);
+_PUBLIC_ int tdb_chainlock_mark(struct tdb_context *tdb, TDB_DATA key)
+{
+        int ret = tdb_nest_lock(tdb, lock_offset(BUCKET(tdb->hash_fn(&key))),
+                                F_WRLCK, TDB_LOCK_MARK_ONLY);
         tdb_trace_1rec(tdb, "tdb_chainlock_mark", key);
         return ret;
@@ -690 +881 @@
 
 /* unmark a chain as locked without actually locking it. Warning! use with great caution! */
-int tdb_chainlock_unmark(struct tdb_context *tdb, TDB_DATA key)
+_PUBLIC_ int tdb_chainlock_unmark(struct tdb_context *tdb, TDB_DATA key)
 {
         tdb_trace_1rec(tdb, "tdb_chainlock_unmark", key);
-        return tdb_unlock(tdb, BUCKET(tdb->hash_fn(&key)), F_WRLCK | TDB_MARK_LOCK);
-}
-
-int tdb_chainunlock(struct tdb_context *tdb, TDB_DATA key)
+        return tdb_nest_unlock(tdb, lock_offset(BUCKET(tdb->hash_fn(&key))),
+                               F_WRLCK, true);
+}
+
+_PUBLIC_ int tdb_chainunlock(struct tdb_context *tdb, TDB_DATA key)
 {
         tdb_trace_1rec(tdb, "tdb_chainunlock", key);
@@ -702 +894 @@
 }
 
-int tdb_chainlock_read(struct tdb_context *tdb, TDB_DATA key)
+_PUBLIC_ int tdb_chainlock_read(struct tdb_context *tdb, TDB_DATA key)
 {
         int ret;
@@ -710 +902 @@
 }
 
-int tdb_chainunlock_read(struct tdb_context *tdb, TDB_DATA key)
+_PUBLIC_ int tdb_chainunlock_read(struct tdb_context *tdb, TDB_DATA key)
 {
         tdb_trace_1rec(tdb, "tdb_chainunlock_read", key);
@@ -716 +908 @@
 }
 
-
-
 /* record lock stops delete underneath */
 int tdb_lock_record(struct tdb_context *tdb, tdb_off_t off)
 {
-        if (tdb->global_lock.count) {
-                return 0;
-        }
-        return off ? tdb->methods->tdb_brlock(tdb, off, F_RDLCK, F_SETLKW, 0, 1) : 0;
+        if (tdb->allrecord_lock.count) {
+                return 0;
+        }
+        return off ? tdb_brlock(tdb, F_RDLCK, off, 1, TDB_LOCK_WAIT) : 0;
 }
 
@@ -738 +928 @@
                 if (i->off == off)
                         return -1;
-        return tdb->methods->tdb_brlock(tdb, off, F_WRLCK, F_SETLK, 1, 1);
-}
-
-/*
-  Note this is meant to be F_SETLK, *not* F_SETLKW, as it's not
-  an error to fail to get the lock here.
-*/
+        if (tdb->allrecord_lock.count) {
+                if (tdb->allrecord_lock.ltype == F_WRLCK) {
+                        return 0;
+                }
+                return -1;
+        }
+        return tdb_brlock(tdb, F_WRLCK, off, 1, TDB_LOCK_NOWAIT|TDB_LOCK_PROBE);
+}
+
 int tdb_write_unlock_record(struct tdb_context *tdb, tdb_off_t off)
 {
-        return tdb->methods->tdb_brlock(tdb, off, F_UNLCK, F_SETLK, 0, 1);
+        if (tdb->allrecord_lock.count) {
+                return 0;
+        }
+        return tdb_brunlock(tdb, F_WRLCK, off, 1);
 }
 
@@ -756 +951 @@
         uint32_t count = 0;
 
-        if (tdb->global_lock.count) {
+        if (tdb->allrecord_lock.count) {
                 return 0;
         }
@@ -765 +960 @@
                 if (i->off == off)
                         count++;
-        return (count == 1 ? tdb->methods->tdb_brlock(tdb, off, F_UNLCK, F_SETLKW, 0, 1) : 0);
-}
+        return (count == 1 ? tdb_brunlock(tdb, F_RDLCK, off, 1) : 0);
+}
+
+bool tdb_have_extra_locks(struct tdb_context *tdb)
+{
+        unsigned int extra = tdb->num_lockrecs;
+
+        /* A transaction holds the lock for all records. */
+        if (!tdb->transaction && tdb->allrecord_lock.count) {
+                return true;
+        }
+
+        /* We always hold the active lock if CLEAR_IF_FIRST. */
+        if (find_nestlock(tdb, ACTIVE_LOCK)) {
+                extra--;
+        }
+
+        /* In a transaction, we expect to hold the transaction lock */
+        if (tdb->transaction && find_nestlock(tdb, TRANSACTION_LOCK)) {
+                extra--;
+        }
+
+        return extra;
+}
+
+/* The transaction code uses this to remove all locks. */
+void tdb_release_transaction_locks(struct tdb_context *tdb)
+{
+        unsigned int i, active = 0;
+
+        if (tdb->allrecord_lock.count != 0) {
+                tdb_brunlock(tdb, tdb->allrecord_lock.ltype, FREELIST_TOP, 0);
+                tdb->allrecord_lock.count = 0;
+        }
+
+        for (i=0;i<tdb->num_lockrecs;i++) {
+                struct tdb_lock_type *lck = &tdb->lockrecs[i];
+
+                /* Don't release the active lock!  Copy it to first entry. */
+                if (lck->off == ACTIVE_LOCK) {
+                        tdb->lockrecs[active++] = *lck;
+                } else {
+                        tdb_brunlock(tdb, lck->ltype, lck->off, 1);
+                }
+        }
+        tdb->num_lockrecs = active;
+        if (tdb->num_lockrecs == 0) {
+                SAFE_FREE(tdb->lockrecs);
+        }
+}