tdb: Vectorize _tdb_store
[samba.git] / lib / tdb / common / tdb.c
index 8242d2bb3fd3dda7989abe5689cf37a764a6eeee..6d4ad6d005c41720e06a10711c08a486b1313f38 100644 (file)
@@ -59,6 +59,11 @@ static void tdb_increment_seqnum(struct tdb_context *tdb)
                return;
        }
 
+       if (tdb->transaction != NULL) {
+               tdb_increment_seqnum_nonblock(tdb);
+               return;
+       }
+
        if (tdb_nest_lock(tdb, TDB_SEQNUM_OFS, F_WRLCK,
                          TDB_LOCK_WAIT|TDB_LOCK_PROBE) != 0) {
                return;
@@ -124,16 +129,32 @@ tdb_off_t tdb_find_lock_hash(struct tdb_context *tdb, TDB_DATA key, uint32_t has
 
 static TDB_DATA _tdb_fetch(struct tdb_context *tdb, TDB_DATA key);
 
+struct tdb_update_hash_state {
+       const TDB_DATA *dbufs;
+       int num_dbufs;
+       tdb_len_t dbufs_len;
+};
+
 static int tdb_update_hash_cmp(TDB_DATA key, TDB_DATA data, void *private_data)
 {
-       TDB_DATA *dbuf = (TDB_DATA *)private_data;
+       struct tdb_update_hash_state *state = private_data;
+       unsigned char *dptr = data.dptr;
+       int i;
 
-       if (dbuf->dsize != data.dsize) {
+       if (state->dbufs_len != data.dsize) {
                return -1;
        }
-       if (memcmp(dbuf->dptr, data.dptr, data.dsize) != 0) {
-               return -1;
+
+       for (i=0; i<state->num_dbufs; i++) {
+               TDB_DATA dbuf = state->dbufs[i];
+               int ret;
+               ret = memcmp(dptr, dbuf.dptr, dbuf.dsize);
+               if (ret != 0) {
+                       return -1;
+               }
+               dptr += dbuf.dsize;
        }
+
        return 0;
 }
 
@@ -141,10 +162,14 @@ static int tdb_update_hash_cmp(TDB_DATA key, TDB_DATA data, void *private_data)
    is <= the old data size and the key exists.
    on failure return -1.
 */
-static int tdb_update_hash(struct tdb_context *tdb, TDB_DATA key, uint32_t hash, TDB_DATA dbuf)
+static int tdb_update_hash(struct tdb_context *tdb, TDB_DATA key,
+                          uint32_t hash,
+                          const TDB_DATA *dbufs, int num_dbufs,
+                          tdb_len_t dbufs_len)
 {
        struct tdb_record rec;
-       tdb_off_t rec_ptr;
+       tdb_off_t rec_ptr, ofs;
+       int i;
 
        /* find entry */
        if (!(rec_ptr = tdb_find(tdb, key, hash, &rec)))
@@ -152,26 +177,41 @@ static int tdb_update_hash(struct tdb_context *tdb, TDB_DATA key, uint32_t hash,
 
        /* it could be an exact duplicate of what is there - this is
         * surprisingly common (eg. with a ldb re-index). */
-       if (rec.key_len == key.dsize &&
-           rec.data_len == dbuf.dsize &&
-           rec.full_hash == hash &&
-           tdb_parse_record(tdb, key, tdb_update_hash_cmp, &dbuf) == 0) {
-               return 0;
+       if (rec.data_len == dbufs_len) {
+               struct tdb_update_hash_state state = {
+                       .dbufs = dbufs, .num_dbufs = num_dbufs,
+                       .dbufs_len = dbufs_len
+               };
+               int ret;
+
+               ret = tdb_parse_record(tdb, key, tdb_update_hash_cmp, &state);
+               if (ret == 0) {
+                       return 0;
+               }
        }
 
        /* must be long enough key, data and tailer */
-       if (rec.rec_len < key.dsize + dbuf.dsize + sizeof(tdb_off_t)) {
+       if (rec.rec_len < key.dsize + dbufs_len + sizeof(tdb_off_t)) {
                tdb->ecode = TDB_SUCCESS; /* Not really an error */
                return -1;
        }
 
-       if (tdb->methods->tdb_write(tdb, rec_ptr + sizeof(rec) + rec.key_len,
-                     dbuf.dptr, dbuf.dsize) == -1)
-               return -1;
+       ofs = rec_ptr + sizeof(rec) + rec.key_len;
+
+       for (i=0; i<num_dbufs; i++) {
+               TDB_DATA dbuf = dbufs[i];
+               int ret;
+
+               ret = tdb->methods->tdb_write(tdb, ofs, dbuf.dptr, dbuf.dsize);
+               if (ret == -1) {
+                       return -1;
+               }
+               ofs += dbuf.dsize;
+       }
 
-       if (dbuf.dsize != rec.data_len) {
+       if (dbufs_len != rec.data_len) {
                /* update size */
-               rec.data_len = dbuf.dsize;
+               rec.data_len = dbufs_len;
                return tdb_rec_write(tdb, rec_ptr, &rec);
        }
 
@@ -345,7 +385,7 @@ static int tdb_count_dead(struct tdb_context *tdb, uint32_t hash)
 /*
  * Purge all DEAD records from a hash chain
  */
-static int tdb_purge_dead(struct tdb_context *tdb, uint32_t hash)
+int tdb_purge_dead(struct tdb_context *tdb, uint32_t hash)
 {
        int res = -1;
        struct tdb_record rec;
@@ -445,15 +485,21 @@ _PUBLIC_ int tdb_delete(struct tdb_context *tdb, TDB_DATA key)
 /*
  * See if we have a dead record around with enough space
  */
-static tdb_off_t tdb_find_dead(struct tdb_context *tdb, uint32_t hash,
-                              struct tdb_record *r, tdb_len_t length)
+tdb_off_t tdb_find_dead(struct tdb_context *tdb, uint32_t hash,
+                       struct tdb_record *r, tdb_len_t length,
+                       tdb_off_t *p_last_ptr)
 {
-       tdb_off_t rec_ptr;
+       tdb_off_t rec_ptr, last_ptr;
        tdb_off_t best_rec_ptr = 0;
+       tdb_off_t best_last_ptr = 0;
        struct tdb_record best = { .rec_len = UINT32_MAX };
 
+       length += sizeof(tdb_off_t); /* tailer */
+
+       last_ptr = TDB_HASH_TOP(hash);
+
        /* read in the hash top */
-       if (tdb_ofs_read(tdb, TDB_HASH_TOP(hash), &rec_ptr) == -1)
+       if (tdb_ofs_read(tdb, last_ptr, &rec_ptr) == -1)
                return 0;
 
        /* keep looking until we find the right record */
@@ -464,8 +510,10 @@ static tdb_off_t tdb_find_dead(struct tdb_context *tdb, uint32_t hash,
                if (TDB_DEAD(r) && (r->rec_len >= length) &&
                    (r->rec_len < best.rec_len)) {
                        best_rec_ptr = rec_ptr;
+                       best_last_ptr = last_ptr;
                        best = *r;
                }
+               last_ptr = rec_ptr;
                rec_ptr = r->next;
        }
 
@@ -474,16 +522,38 @@ static tdb_off_t tdb_find_dead(struct tdb_context *tdb, uint32_t hash,
        }
 
        *r = best;
+       *p_last_ptr = best_last_ptr;
        return best_rec_ptr;
 }
 
-static int _tdb_store(struct tdb_context *tdb, TDB_DATA key,
-                      TDB_DATA dbuf, int flag, uint32_t hash)
+static int _tdb_storev(struct tdb_context *tdb, TDB_DATA key,
+                      const TDB_DATA *dbufs, int num_dbufs,
+                      int flag, uint32_t hash)
 {
        struct tdb_record rec;
-       tdb_off_t rec_ptr;
+       tdb_off_t rec_ptr, ofs;
+       tdb_len_t rec_len, dbufs_len;
+       int i;
        int ret = -1;
 
+       dbufs_len = 0;
+
+       for (i=0; i<num_dbufs; i++) {
+               size_t dsize = dbufs[i].dsize;
+
+               dbufs_len += dsize;
+               if (dbufs_len < dsize) {
+                       tdb->ecode = TDB_ERR_OOM;
+                       goto fail;
+               }
+       }
+
+       rec_len = key.dsize + dbufs_len;
+       if ((rec_len < key.dsize) || (rec_len < dbufs_len)) {
+               tdb->ecode = TDB_ERR_OOM;
+               goto fail;
+       }
+
        /* check for it existing, on insert. */
        if (flag == TDB_INSERT) {
                if (tdb_exists_hash(tdb, key, hash)) {
@@ -492,7 +562,8 @@ static int _tdb_store(struct tdb_context *tdb, TDB_DATA key,
                }
        } else {
                /* first try in-place update, on modify or replace. */
-               if (tdb_update_hash(tdb, key, hash, dbuf) == 0) {
+               if (tdb_update_hash(tdb, key, hash, dbufs, num_dbufs,
+                                   dbufs_len) == 0) {
                        goto done;
                }
                if (tdb->ecode == TDB_ERR_NOEXIST &&
@@ -502,7 +573,7 @@ static int _tdb_store(struct tdb_context *tdb, TDB_DATA key,
                        goto fail;
                }
        }
-       /* reset the error code potentially set by the tdb_update() */
+       /* reset the error code potentially set by the tdb_update_hash() */
        tdb->ecode = TDB_SUCCESS;
 
        /* delete any existing record - if it doesn't exist we don't
@@ -511,55 +582,8 @@ static int _tdb_store(struct tdb_context *tdb, TDB_DATA key,
        if (flag != TDB_INSERT)
                tdb_delete_hash(tdb, key, hash);
 
-       if (tdb->max_dead_records != 0) {
-               /*
-                * Allow for some dead records per hash chain, look if we can
-                * find one that can hold the new record. We need enough space
-                * for key, data and tailer. If we find one, we don't have to
-                * consult the central freelist.
-                */
-               rec_ptr = tdb_find_dead(
-                       tdb, hash, &rec,
-                       key.dsize + dbuf.dsize + sizeof(tdb_off_t));
-
-               if (rec_ptr != 0) {
-                       rec.key_len = key.dsize;
-                       rec.data_len = dbuf.dsize;
-                       rec.full_hash = hash;
-                       rec.magic = TDB_MAGIC;
-                       if (tdb_rec_write(tdb, rec_ptr, &rec) == -1
-                           || tdb->methods->tdb_write(
-                                   tdb, rec_ptr + sizeof(rec),
-                                   key.dptr, key.dsize) == -1
-                           || tdb->methods->tdb_write(
-                                   tdb, rec_ptr + sizeof(rec) + key.dsize,
-                                   dbuf.dptr, dbuf.dsize) == -1) {
-                               goto fail;
-                       }
-                       goto done;
-               }
-       }
-
-       /*
-        * We have to allocate some space from the freelist, so this means we
-        * have to lock it. Use the chance to purge all the DEAD records from
-        * the hash chain under the freelist lock.
-        */
-
-       if (tdb_lock(tdb, -1, F_WRLCK) == -1) {
-               goto fail;
-       }
-
-       if ((tdb->max_dead_records != 0)
-           && (tdb_purge_dead(tdb, hash) == -1)) {
-               tdb_unlock(tdb, -1, F_WRLCK);
-               goto fail;
-       }
-
        /* we have to allocate some space */
-       rec_ptr = tdb_allocate(tdb, key.dsize + dbuf.dsize, &rec);
-
-       tdb_unlock(tdb, -1, F_WRLCK);
+       rec_ptr = tdb_allocate(tdb, hash, rec_len, &rec);
 
        if (rec_ptr == 0) {
                goto fail;
@@ -570,17 +594,36 @@ static int _tdb_store(struct tdb_context *tdb, TDB_DATA key,
                goto fail;
 
        rec.key_len = key.dsize;
-       rec.data_len = dbuf.dsize;
+       rec.data_len = dbufs_len;
        rec.full_hash = hash;
        rec.magic = TDB_MAGIC;
 
+       ofs = rec_ptr;
+
        /* write out and point the top of the hash chain at it */
-       if (tdb_rec_write(tdb, rec_ptr, &rec) == -1
-           || tdb->methods->tdb_write(tdb, rec_ptr+sizeof(rec),
-                                      key.dptr, key.dsize) == -1
-           || tdb->methods->tdb_write(tdb, rec_ptr+sizeof(rec)+key.dsize,
-                                      dbuf.dptr, dbuf.dsize) == -1
-           || tdb_ofs_write(tdb, TDB_HASH_TOP(hash), &rec_ptr) == -1) {
+       ret = tdb_rec_write(tdb, ofs, &rec);
+       if (ret == -1) {
+               goto fail;
+       }
+       ofs += sizeof(rec);
+
+       ret = tdb->methods->tdb_write(tdb, ofs, key.dptr, key.dsize);
+       if (ret == -1) {
+               goto fail;
+       }
+       ofs += key.dsize;
+
+       for (i=0; i<num_dbufs; i++) {
+               ret = tdb->methods->tdb_write(tdb, ofs, dbufs[i].dptr,
+                                             dbufs[i].dsize);
+               if (ret == -1) {
+                       goto fail;
+               }
+               ofs += dbufs[i].dsize;
+       }
+
+       ret = tdb_ofs_write(tdb, TDB_HASH_TOP(hash), &rec_ptr);
+       if (ret == -1) {
                /* Need to tdb_unallocate() here */
                goto fail;
        }
@@ -594,6 +637,12 @@ static int _tdb_store(struct tdb_context *tdb, TDB_DATA key,
        return ret;
 }
 
+static int _tdb_store(struct tdb_context *tdb, TDB_DATA key,
+                     TDB_DATA dbuf, int flag, uint32_t hash)
+{
+       return _tdb_storev(tdb, key, &dbuf, 1, flag, hash);
+}
+
 /* store an element in the database, replacing any existing element
    with the same key
 
@@ -761,6 +810,15 @@ _PUBLIC_ void tdb_remove_flags(struct tdb_context *tdb, unsigned flags)
                return;
        }
 
+       if ((flags & TDB_NOLOCK) &&
+           (tdb->feature_flags & TDB_FEATURE_FLAG_MUTEX) &&
+           (tdb->mutexes == NULL)) {
+               tdb->ecode = TDB_ERR_LOCK;
+               TDB_LOG((tdb, TDB_DEBUG_FATAL, "tdb_remove_flags: "
+                        "Can not remove NOLOCK flag on mutexed databases"));
+               return;
+       }
+
        if (flags & TDB_ALLOW_NESTING) {
                tdb->flags |= TDB_DISALLOW_NESTING;
        }
@@ -814,7 +872,7 @@ static int tdb_free_region(struct tdb_context *tdb, tdb_off_t offset, ssize_t le
  */
 _PUBLIC_ int tdb_wipe_all(struct tdb_context *tdb)
 {
-       int i;
+       uint32_t i;
        tdb_off_t offset = 0;
        ssize_t data_len;
        tdb_off_t recovery_head;