Closes #1575 Added toku_omt_create_steal_sorted_array

git-svn-id: file:///svn/toku/tokudb@10378 c7de825b-a66e-492c-adef-691d508d4ae1

newbrt/brt-serialize.c

@@ -573,12 +573,13 @@ int toku_deserialize_brtnode_from (int fd, BLOCKNUM blocknum, u_int32_t fullhash
 	u_int32_t end_of_data = rc.ndone;
 	result->u.l.n_bytes_in_buffer += end_of_data-start_of_data + n_in_buf*OMT_ITEM_OVERHEAD;
 	actual_sum *= result->rand4fingerprint;
-	r = toku_omt_create_from_sorted_array(&result->u.l.buffer, array, n_in_buf);
-	toku_free(array);
+	r = toku_omt_create_steal_sorted_array(&result->u.l.buffer, &array, n_in_buf, n_in_buf);
 	if (r!=0) {
+            toku_free(array);
 	    if (0) { died_21: toku_omt_destroy(&result->u.l.buffer); }
 	    return toku_db_badformat();
 	}
+        assert(array==NULL);
         r = toku_leaflock_borrow(&result->u.l.leaflock);
         if (r!=0) goto died_21;
 

newbrt/brt.c

@@ -807,9 +807,8 @@ brtleaf_split (BRT t, BRTNODE node, BRTNODE *nodea, BRTNODE *nodeb, DBT *splitk)
             B   ->u.l.n_bytes_in_buffer += diff_size;
         }
         if ((r = toku_omt_create_from_sorted_array(&B->u.l.buffer,    leafentries+break_at, n_leafentries-break_at))) return r;
-        if ((r = toku_omt_create_from_sorted_array(&node->u.l.buffer, leafentries,          break_at))) return r;
-
-        toku_free(leafentries);
+        if ((r = toku_omt_create_steal_sorted_array(&node->u.l.buffer, &leafentries,          break_at, n_leafentries))) return r;
+        assert(leafentries==NULL);
 
         toku_verify_all_in_mempool(node);
         toku_verify_all_in_mempool(B);

newbrt/omt.c

@@ -58,24 +58,46 @@ struct omt_cursor {
     OMTCURSOR next,prev; // circular linked list of all OMTCURSORs associated with omt.
 };
 
-static int omt_create_internal(OMT *omtp, u_int32_t num_starting_nodes) {
-    if (num_starting_nodes < 2) num_starting_nodes = 2;
+static inline int
+omt_create_no_array(OMT *omtp) {
     OMT MALLOC(result);
-    if (result==NULL) return errno;
+    if (result==NULL) return ENOMEM;
     result->is_array       = TRUE;
-    result->capacity       = 2*num_starting_nodes;
     result->i.a.num_values = 0;
     result->i.a.start_idx  = 0;
+    result->associated = NULL;
+    *omtp = result;
+    return 0;
+}
+
+static int omt_create_internal(OMT *omtp, u_int32_t num_starting_nodes) {
+    OMT result;
+    int r = omt_create_no_array(&result);
+    if (r) return r;
+    if (num_starting_nodes < 2) num_starting_nodes = 2;
+    result->capacity       = 2*num_starting_nodes;
     MALLOC_N(result->capacity, result->i.a.values);
     if (result->i.a.values==NULL) {
         toku_free(result);
         return errno;
     }
-    result->associated = NULL;
     *omtp = result;
     return 0;
 }
 
+int
+toku_omt_create_steal_sorted_array(OMT *omtp, OMTVALUE **valuesp, u_int32_t numvalues, u_int32_t capacity) {
+    if (numvalues>capacity || !*valuesp) return EINVAL;
+    int r = omt_create_no_array(omtp);
+    if (r) return r;
+    OMT result = *omtp;
+    result->capacity       = capacity;
+    result->i.a.num_values = numvalues;
+    result->i.a.values     = *valuesp;
+    *valuesp = NULL; //Remove caller's reference.
+    return 0;
+}
+
 int toku_omt_cursor_create (OMTCURSOR *omtcp) {
     OMTCURSOR MALLOC(c);
     if (c==NULL) return errno;

newbrt/omt.h

@@ -171,6 +171,24 @@ int toku_omt_create_from_sorted_array(OMT *omtp, OMTVALUE *values, u_int32_t num
 //               If the N values are known in advance, are sorted, and
 //               the structure is empty, we can batch insert them much faster.
 
+int toku_omt_create_steal_sorted_array(OMT *omtp, OMTVALUE **valuesp, u_int32_t numvalues, u_int32_t steal_capacity);
+// Effect: Create an OMT containing values.  The number of values is in numvalues.
+//         On success the OMT takes ownership of *valuesp array, and sets valuesp=NULL.
+// Requires: omtp != NULL
+// Requires: valuesp != NULL
+// Requires: *valuesp is sorted
+// Requires: *valuesp was allocated with toku_malloc
+// Requires: Capacity of the *valuesp array is <= steal_capacity
+// Requires: On success, *valuesp may not be accessed again by the caller.
+// Returns:
+//   0        success
+//   ENOMEM   out of memory (and doesn't modify *omtp)
+//   EINVAL   *valuesp == NULL or numvalues > capacity
+// Performance:  time=O(1)
+// Rational:     toku_omt_create_from_sorted_array takes O(numvalues) time.
+//               By taking ownership of the array, we save a malloc and memcpy,
+//               and possibly a free (if the caller is done with the array).
+
 void toku_omt_destroy(OMT *omtp);
 // Effect:  Destroy an OMT, freeing all its memory.
 //   Does not free the OMTVALUEs stored in the OMT.

newbrt/tests/omt-test.c

@@ -55,6 +55,7 @@ enum close_when_done {
     KEEP_WHEN_DONE
 };
 enum create_type {
+    STEAL_ARRAY,
     BATCH_INSERT,
     INSERT_AT,
     INSERT_AT_ALMOST_RANDOM,
@@ -234,6 +235,13 @@ test_create_from_sorted_array (enum create_type create_choice, enum close_when_d
         r = toku_omt_create_from_sorted_array(&omt, values, length);
         CKERR(r);
     }
+    else if (create_choice == STEAL_ARRAY) {
+        TESTVALUE* MALLOC_N(length, values_copy);
+        memcpy(values_copy, values, length*sizeof(*values));
+        r = toku_omt_create_steal_sorted_array(&omt, &values_copy, length, length);
+        CKERR(r);
+        assert(values_copy==NULL);
+    }
     else if (create_choice == INSERT_AT) {
         test_create_insert_at_sequential(KEEP_WHEN_DONE);
     }
@@ -974,6 +982,7 @@ test_main(int argc, const char *argv[]) {
     test_create(      CLOSE_WHEN_DONE);
     test_create_size( CLOSE_WHEN_DONE);
     runtests_create_choice(BATCH_INSERT);
+    runtests_create_choice(STEAL_ARRAY);
     runtests_create_choice(INSERT_AT);
     runtests_create_choice(INSERT_AT_ALMOST_RANDOM);
     cleanup_globals();