CLIENT-681: Create dynamically allocate byte pool.

src/include/bit_operations.h

@@ -25,4 +25,4 @@
 as_status add_new_bit_op(AerospikeClient *self, as_error *err,
                          PyObject *op_dict, as_vector *unicodeStrVector,
                          as_operations *ops, long operation_code,
-                         long *ret_type, int serializer_type);
+                         long *ret_type);

src/include/cdt_list_operations.h

@@ -24,5 +24,4 @@
 as_status add_new_list_op(AerospikeClient *self, as_error *err,
                           PyObject *op_dict, as_vector *unicodeStrVector,
                           as_dynamic_pool *dynamic_pool, as_operations *ops,
-                          long operation_code, long *ret_type,
-                          int serializer_type);
+                          long operation_code, long *ret_type);

src/include/cdt_map_operations.h

@@ -24,5 +24,4 @@
 as_status add_new_map_op(AerospikeClient *self, as_error *err,
                          PyObject *op_dict, as_vector *unicodeStrVector,
                          as_dynamic_pool *dynamic_pool, as_operations *ops,
-                         long operation_code, long *ret_type,
-                         int serializer_type);
+                         long operation_code, long *ret_type);

src/include/cdt_operation_utils.h

@@ -51,12 +51,11 @@ as_status get_bin(as_error *err, PyObject *op_dict, as_vector *unicodeStrVector,
 
 as_status get_asval(AerospikeClient *self, as_error *err, char *key,
                     PyObject *op_dict, as_val **val,
-                    as_dynamic_pool *dynamic_pool, int serializer_type,
-                    bool required);
+                    as_dynamic_pool *dynamic_pool, bool required);
 
 as_status get_val_list(AerospikeClient *self, as_error *err,
                        const char *list_key, PyObject *op_dict, as_list **list,
-                       as_dynamic_pool *dynamic_pool, int serializer_type);
+                       as_dynamic_pool *dynamic_pool);
 
 as_status get_int64_t(as_error *err, const char *key, PyObject *op_dict,
                       int64_t *i64_valptr);

src/include/conversions.h

@@ -86,15 +86,15 @@ as_status pyobject_to_strArray(as_error *err, PyObject *py_list, char **arr,
 as_status as_val_new_from_pyobject(AerospikeClient *self, as_error *err,
                                    PyObject *py_obj, as_val **val,
                                    as_dynamic_pool *dynamic_pool, int serializer_type,
-                                   bool allocate_buffer);
+                                   bool destroy_buffers);
 
 as_status pyobject_to_map(AerospikeClient *self, as_error *err,
                           PyObject *py_dict, as_map **map,
-                          as_dynamic_pool *dynamic_pool, int serializer_type, bool allocate_buffer);
+                          as_dynamic_pool *dynamic_pool, int serializer_type, bool destroy_buffers);
 
 as_status pyobject_to_list(AerospikeClient *self, as_error *err,
                            PyObject *py_list, as_list **list,
-                           as_dynamic_pool *dynamic_pool, int serializer_type, bool allocate_buffer);
+                           as_dynamic_pool *dynamic_pool, int serializer_type, bool destroy_buffers);
 
 as_status pyobject_to_key(as_error *err, PyObject *py_key, as_key *key);
 
@@ -197,8 +197,7 @@ as_status string_and_pyuni_from_pystring(PyObject *py_string,
 
 as_status get_cdt_ctx(AerospikeClient *self, as_error *err, as_cdt_ctx *cdt_ctx,
                       PyObject *op_dict, bool *ctx_in_use,
-                      as_dynamic_pool *dynamic_pool, int serializer_type,
-                      bool allocate_buffer);
+                      as_dynamic_pool *dynamic_pool, bool destroy_buffers);
 
 as_status convert_exp_list(AerospikeClient *self, PyObject *py_exp_list,
                            as_exp **exp_list, as_error *err);

src/include/dynamic_pool.hpp

@@ -1,8 +1,13 @@
 /*
  *******************************************************************************************************
- * Dynamic pool maintained to avoid excessive runtime mallocs.
- * It is comprised of a Pool of pointers to pools of
- * as_bytes containing AS_DYNAMIC_POOL_BLOCK_SIZE as_bytes.
+ * Dynamic pool maintained to avoid excessive runtime mallocs and efficiently use memory.
+ * 
+ * The dynamic pool maintains a table composed of several groups of as_bytes buffers.
+ * New groups are allocated dynamically after the current group is exhausted.
+ * As more as_bytes are used, group sizes will grow to reduce malloc calls.
+
+ * The dynamic pool does not allocate any memory unless bytes are used in a command.
+ * 
  *******************************************************************************************************
  */
 #define AS_DYNAMIC_POOL_BYTES_PER_GROUP_MIN 128
@@ -41,8 +46,7 @@ static inline void dynamic_pool_malloc_group(as_dynamic_pool *dynamic_pool,
     // bytes number to allocate in the next group.
     uint16_t num_bytes_to_allocate = dynamic_pool->bytes_per_group;
 
-    // Allocate a group of bytes determined by bytes_per_group.
-    table[group_num_to_allocate] = (as_bytes *)cf_malloc(sizeof(as_bytes) * num_bytes_to_allocate);
+    table[group_num_to_allocate] = (as_bytes *)cf_malloc(num_bytes_to_allocate * sizeof(as_bytes));
 
     // If allocation fails, throw an error.
     if (table[group_num_to_allocate] == NULL) {
@@ -144,7 +148,7 @@ static inline void dynamic_pool_free_group(as_dynamic_pool *dynamic_pool, bool f
         dynamic_pool_destroy_bytes_in_group(dynamic_pool, group_index, num_bytes);
     }
     as_bytes* group = dynamic_pool->byte_group_table[group_index];
-    cf_free((as_dynamic_pool *)group);
+    cf_free(group);
 }
 
 /**
@@ -166,7 +170,7 @@ static inline void dynamic_pool_free_table(as_dynamic_pool *dynamic_pool, bool f
     // Free the current byte group.
     dynamic_pool_free_group(dynamic_pool, free_buffer, dynamic_pool->group_iterator, dynamic_pool->byte_iterator);
     // Free the table.
-    cf_free(((as_dynamic_pool *)dynamic_pool)->byte_group_table);
+    cf_free(dynamic_pool->byte_group_table);
 }
 
 
@@ -176,7 +180,7 @@ static inline void dynamic_pool_free_table(as_dynamic_pool *dynamic_pool, bool f
  * If the group table is full, the table is also expanded.
  *
  * @param dynamic_pool Pointer to a dynamic pool.
- * @param error Pointer to an as_error
+ * @param err Pointer to an as_error
  *
  */
 static inline void dynamic_pool_init(as_dynamic_pool *dynamic_pool,
@@ -197,7 +201,7 @@ static inline void dynamic_pool_init(as_dynamic_pool *dynamic_pool,
  * If the group table is full, the table is also expanded.
  *
  * @param dynamic_pool Pointer to a dynamic pool.
- * @param error Pointer to an as_error
+ * @param err Pointer to an as_error
  *
  */
 static inline void dynamic_pool_add_group(as_dynamic_pool *dynamic_pool,
@@ -220,26 +224,30 @@ static inline void dynamic_pool_add_group(as_dynamic_pool *dynamic_pool,
  */
 #define BYTE_POOL_INIT_NULL(dynamic_pool)                                                           \
     (dynamic_pool)->byte_group_table = NULL;
+
 /**
  * Fetches the address of the next as_byte in the pool.
  * 
  * @param map_bytes Pointer to an as_bytes.
  * @param dynamic_pool Pointer to a dynamic pool.
- * @param error Pointer to an as_error
+ * @param err Pointer to an as_error
  */
-#define GET_BYTES_POOL(map_bytes, dynamic_pool, err)                                                \
-    as_bytes **table = (dynamic_pool)->byte_group_table;                                              \
-    if (table == NULL) {                                                                            \
-        dynamic_pool_init(dynamic_pool, err);                                                       \
-    }                                                                                               \
-    else if ((dynamic_pool)->bytes_per_group <= (dynamic_pool)->byte_iterator) {                        \
-        dynamic_pool_add_group(dynamic_pool, err);                                                     \
-    }                                                                                               \
-    table = (dynamic_pool)->byte_group_table;                                                         \
-    uint16_t group_iterator = (dynamic_pool)->group_iterator;                                         \
-    as_bytes *group = table[group_iterator];                                                        \
-    uint16_t byte_iterator = (dynamic_pool)->byte_iterator++;                                         \
-    map_bytes = &group[byte_iterator];
+static inline as_bytes* GET_BYTES_POOL(as_dynamic_pool *dynamic_pool, as_error *err) {
+    as_bytes **table = dynamic_pool->byte_group_table;
+
+    if (table == NULL) {
+        dynamic_pool_init(dynamic_pool, err);
+    } else if (dynamic_pool->byte_iterator >= dynamic_pool->bytes_per_group) {
+        dynamic_pool_add_group(dynamic_pool, err);
+    }
+
+    table = dynamic_pool->byte_group_table;
+    uint16_t group_iterator = dynamic_pool->group_iterator;
+    as_bytes *group = table[group_iterator];
+    uint16_t byte_iterator = dynamic_pool->byte_iterator++;
+
+    return &group[byte_iterator];
+}
 
 /**
  * Destroy the dynamic pool. Must be called before the dynamic_pool loses scope.
@@ -248,7 +256,9 @@ static inline void dynamic_pool_add_group(as_dynamic_pool *dynamic_pool,
  * @param free_buffer boolean value determining if as_bytes_destroyed should be called.
  *        If the raw bytes array is heap allocated, free_buffer should be true.
  */
-#define DESTROY_DYNAMIC_POOL(dynamic_pool, free_buffer)                                             \
-    if ((dynamic_pool)->byte_group_table != NULL) {                                                   \
-        dynamic_pool_free_table(dynamic_pool, free_buffer);                                         \
+static inline void DESTROY_DYNAMIC_POOL(as_dynamic_pool *dynamic_pool, bool free_buffer) {
+    if (dynamic_pool->byte_group_table != NULL) {
+        dynamic_pool_free_table(dynamic_pool, free_buffer);
     }
+}
+

src/include/expression_operations.h

@@ -23,5 +23,4 @@
 
 as_status add_new_expr_op(AerospikeClient *self, as_error *err,
                           PyObject *op_dict, as_vector *unicodeStrVector,
-                          as_operations *ops, long operation_code,
-                          int serializer_type);
+                          as_operations *ops, long operation_code);

src/include/hll_operations.h

@@ -24,5 +24,4 @@
 as_status add_new_hll_op(AerospikeClient *self, as_error *err,
                          PyObject *op_dict, as_vector *unicodeStrVector,
                          as_dynamic_pool *dynamic_pool, as_operations *ops,
-                         long operation_code, long *ret_type,
-                         int serializer_type);
+                         long operation_code, long *ret_type);

src/include/operate.h

@@ -24,5 +24,5 @@
 #include "exceptions.h"
 
 as_status add_op(AerospikeClient *self, as_error *err, PyObject *py_val,
-                 as_vector *unicodeStrVector, as_dynamic_pool *dynamic_pool,
+                 as_vector *unicodeStrVector, as_dynamic_pool *dynamic_pool, bool destroy_buffers,
                  as_operations *ops, long *op, long *ret_type);

src/include/policy.h

@@ -31,7 +31,7 @@
 
 enum Aerospike_serializer_values {
     SERIALIZER_NONE, /* default handler for serializer type */
-    SERIALIZER_PYTHON,
+    SERIALIZER_PYTHON, /* Removed */
     SERIALIZER_JSON,
     SERIALIZER_USER,
 };

src/main/client/apply.c

@@ -27,6 +27,7 @@
 #include "exceptions.h"
 #include "policy.h"
 
+
 /**
  *******************************************************************************************************
  * This function applies a registered udf module on a particular record.
@@ -70,6 +71,8 @@ PyObject *AerospikeClient_Apply_Invoke(AerospikeClient *self, PyObject *py_key,
     // Initialisation flags
     bool key_initialised = false;
 
+    bool destroy_buffers = false;
+
     // Initialize error
     as_error_init(&err);
 
@@ -102,10 +105,9 @@ PyObject *AerospikeClient_Apply_Invoke(AerospikeClient *self, PyObject *py_key,
     // Key is initialiased successfully
     key_initialised = true;
 
-    bool allocate_buffer = false;
     // Convert python list to as_list
     pyobject_to_list(self, &err, py_arglist, &arglist, &dynamic_pool,
-                     SERIALIZER_PYTHON, allocate_buffer);
+                     SERIALIZER_NONE, destroy_buffers);
     if (err.code != AEROSPIKE_OK) {
         goto CLEANUP;
     }
@@ -169,7 +171,7 @@ PyObject *AerospikeClient_Apply_Invoke(AerospikeClient *self, PyObject *py_key,
     }
     as_list_destroy(arglist);
     as_val_destroy(result);
-    DESTROY_DYNAMIC_POOL(&dynamic_pool, false);
+    DESTROY_DYNAMIC_POOL(&dynamic_pool, destroy_buffers);
 
     if (err.code != AEROSPIKE_OK) {
         raise_exception_base(&err, py_key, Py_None, py_module, py_function,

src/main/client/batch_apply.c

@@ -134,7 +134,8 @@ static PyObject *AerospikeClient_Batch_Apply_Invoke(
     as_dynamic_pool dynamic_pool;
     BYTE_POOL_INIT_NULL(&dynamic_pool);
 
-
+    bool destroy_buffers = false;
+
     Py_ssize_t keys_size = PyList_Size(py_keys);
 
     as_list *arglist = NULL;
@@ -200,10 +201,9 @@ static PyObject *AerospikeClient_Batch_Apply_Invoke(
     const char *mod = PyUnicode_AsUTF8(py_mod);
     const char *func = PyUnicode_AsUTF8(py_func);
 
-    bool allocate_buffer = false;
 
     pyobject_to_list(self, err, py_args, &arglist, &dynamic_pool,
-                     SERIALIZER_PYTHON, allocate_buffer);
+                     SERIALIZER_NONE, destroy_buffers);
     if (err->code != AEROSPIKE_OK) {
         goto CLEANUP;
     }
@@ -290,7 +290,7 @@ static PyObject *AerospikeClient_Batch_Apply_Invoke(
         as_vector_destroy(tmp_keys_p);
     }
 
-    DESTROY_DYNAMIC_POOL(&dynamic_pool, false);
+    DESTROY_DYNAMIC_POOL(&dynamic_pool, destroy_buffers);
 
     if (err->code != AEROSPIKE_OK) {
         raise_exception(err);

src/main/client/batch_operate.c

@@ -141,6 +141,8 @@ static PyObject *AerospikeClient_Batch_Operate_Invoke(
     BYTE_POOL_INIT_NULL(&dynamic_pool);
 
 
+    bool destroy_buffers = false;
+
     as_vector *tmp_keys_p = NULL;
 
     as_operations ops;
@@ -170,7 +172,7 @@ static PyObject *AerospikeClient_Batch_Operate_Invoke(
             goto CLEANUP;
         }
 
-        if (add_op(self, err, py_val, unicodeStrVector, &dynamic_pool, &ops,
+        if (add_op(self, err, py_val, unicodeStrVector, &dynamic_pool, destroy_buffers, &ops,
                    &operation, &return_type) != AEROSPIKE_OK) {
             goto CLEANUP;
         }
@@ -312,7 +314,7 @@ static PyObject *AerospikeClient_Batch_Operate_Invoke(
     as_vector_destroy(unicodeStrVector);
     as_operations_destroy(&ops);
     as_batch_destroy(&batch);
-    DESTROY_DYNAMIC_POOL(&dynamic_pool, false);
+    DESTROY_DYNAMIC_POOL(&dynamic_pool, destroy_buffers);
 
     if (tmp_keys_p) {
         as_vector_destroy(tmp_keys_p);

src/main/client/batch_write.c

@@ -127,6 +127,7 @@ static PyObject *AerospikeClient_BatchWriteInvoke(AerospikeClient *self,
     as_policy_batch *batch_policy_p = NULL;
     as_exp exp_list;
     as_exp *exp_list_p = NULL;
+    bool destroy_buffers = false;
 
     PyObject *py_batch_type = NULL;
     PyObject *py_key = NULL;
@@ -309,7 +310,7 @@ static PyObject *AerospikeClient_BatchWriteInvoke(AerospikeClient *self,
                     goto CLEANUP_ON_ERROR;
                 }
 
-                if (add_op(self, err, py_op, unicodeStrVector, &dynamic_pool,
+                if (add_op(self, err, py_op, unicodeStrVector, &dynamic_pool, destroy_buffers,
                            ops, &operation, &return_type) != AEROSPIKE_OK) {
                     goto CLEANUP_ON_ERROR;
                 }
@@ -402,9 +403,8 @@ static PyObject *AerospikeClient_BatchWriteInvoke(AerospikeClient *self,
 
             as_list *arglist = NULL;
 
-            bool allocate_buffer = false;
             pyobject_to_list(self, err, py_args, &arglist, &dynamic_pool,
-                             SERIALIZER_PYTHON, allocate_buffer);
+                             SERIALIZER_NONE, destroy_buffers);
             if (err->code != AEROSPIKE_OK) {
                 Py_DECREF(py_args);
                 goto CLEANUP_ON_ERROR;
@@ -550,7 +550,7 @@ static PyObject *AerospikeClient_BatchWriteInvoke(AerospikeClient *self,
 
     as_vector_destroy(unicodeStrVector);
 
-    DESTROY_DYNAMIC_POOL(&dynamic_pool, false);
+    DESTROY_DYNAMIC_POOL(&dynamic_pool, destroy_buffers);
 
     if (exp_list_p != NULL) {
         as_exp_destroy(exp_list_p);