tdb: Vectorize _tdb_store
[samba.git] / lib / tdb / common / tdb.c
index ae98c9619d1c1b6f70c799b083acf72278abb57b..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;
 
-       if (dbuf.dsize != rec.data_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 (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);
        }
 
@@ -486,13 +526,34 @@ tdb_off_t tdb_find_dead(struct tdb_context *tdb, uint32_t hash,
        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)) {
@@ -501,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 &&
@@ -511,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
@@ -521,7 +583,7 @@ static int _tdb_store(struct tdb_context *tdb, TDB_DATA key,
                tdb_delete_hash(tdb, key, hash);
 
        /* we have to allocate some space */
-       rec_ptr = tdb_allocate(tdb, hash, key.dsize + dbuf.dsize, &rec);
+       rec_ptr = tdb_allocate(tdb, hash, rec_len, &rec);
 
        if (rec_ptr == 0) {
                goto fail;
@@ -532,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;
        }
@@ -556,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
 
@@ -785,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;