Features/batch operation fix whitespace

.gitmodules

@@ -1,7 +1,7 @@
 [submodule "kinetic-protocol"]
 	path = vendor/kinetic-protocol
 	url = https://github.com/Seagate/kinetic-protocol.git
-	branch = 3.0.5
+	branch = features/batch-operation
 [submodule "socket99"]
 	path = vendor/socket99
 	url = https://github.com/silentbicycle/socket99.git

README.md

@@ -15,7 +15,7 @@ Kinetic Protocol Support
 ------------------------
 Built using:
 
-* [Kinetic Protocol v3.0.5](https://github.com/Kinetic/kinetic-protocol/tree/3.0.5)
+* [Kinetic Protocol v3.0.5](https://github.com/Kinetic/kinetic-protocol/tree/3.0.6)
 * [ProtoBuf-C v1.1.0](https://github.com/protobuf-c/protobuf-c)
 * [Google ProtoBuf v2.6.0](https://developers.google.com/protocol-buffers/docs/downloads)
 
@@ -81,10 +81,10 @@ Getting Started
 API Documentation
 =================
 
-[Kinetic-C API Documentation](http://seagate.github.io/kinetic-c/) (generated with Doxygen)
-* [Kinetic-C API](http://seagate.github.io/kinetic-c/kinetic__client_8h.html)
-* [Kinetic-C types](http://seagate.github.io/kinetic-c/kinetic__types_8h.html)
-* [ByteArray API](http://seagate.github.io/kinetic-c/byte__array_8h.html)
+[Kinetic-C API Documentation](http://kinetic.github.io/kinetic-c/) (generated with Doxygen)
+* [Kinetic-C API](http://kinetic.github.io/kinetic-c/kinetic__client_8h.html)
+* [Kinetic-C types](http://kinetic.github.io/kinetic-c/kinetic__types_8h.html)
+* [ByteArray API](http://kinetic.github.io/kinetic-c/byte__array_8h.html)
     * The ByteArray and ByteBuffer types are used for exchanging variable length byte-arrays with kinetic-c
         * e.g. object keys, object value data, etc.
 

include/kinetic_client.h

@@ -260,4 +260,86 @@ KineticStatus KineticClient_P2POperation(KineticSession* const session,
                                          KineticP2P_Operation* const p2pOp,
                                          KineticCompletionClosure* closure);
 
+/**
+ * @brief Generates a session-unique batch id for this transaction and initializes the structure
+ *
+ * @param session   The connected KineticSession to use for the operation
+ *
+ * @return          Returns a pointer to a KineticBatch_Operation. You need to pass
+ *                  this pointer to other batch interfaces to execute batch commands.
+ *                  Once you are finished with the KineticBatch_Operation, the pointer
+ *                  should be released.
+ */
+KineticBatch_Operation * KineticClient_InitBatchOperation(KineticSession* const session);
+
+/**
+ * @brief Executes a `PUT` operation to store/update an entry on the Kinetic Device.
+ * This command is not committed until the KineticClient_BatchEnd method is invoked
+ * and returned successfully.
+ *
+ * @param batchOp       The session-unique KineticBatch_Operation to use for the operation.
+ * @param entry         Key/value entry for object to store. 'value' must
+ *                      specify the data to be stored. If a closure is provided
+ *                      this pointer must remain valid until the closure callback
+ *                      is called.
+ * @param closure       Must be specified, operation will be executed in asynchronous
+ *                      mode, and closure callback will be called upon completion in
+ *                      another thread.
+ *
+ * @return              Returns the resulting KineticStatus.
+ */
+KineticStatus KineticClient_BatchPut(KineticBatch_Operation* const batchOp,
+                                     KineticEntry* const entry);
+
+/**
+ * @brief Executes a `DELETE` operation to delete an entry from the Kinetic Device.
+ * This command is not committed until the KineticClient_BatchEnd method is invoked
+ * and returned successfully.
+ *
+ * @param batchOp       The session-unique KineticBatch_Operation to use for the operation.
+ * @param entry         Key/value entry for object to delete. 'value' is
+ *                      not used for this operation.
+ * @param closure       Must specified, operation will be executed in asynchronous mode,
+ *                      and closure callback will be called upon completion in another thread.
+ *
+ * @return              Returns the resulting KineticStatus.
+ */
+KineticStatus KineticClient_BatchDelete(KineticBatch_Operation* const batchOp,
+                                        KineticEntry* const entry);
+
+/**
+ * @brief Start the batch operation.
+ *
+ * @param batchOp       The session-unique KineticBatch_Operation to use for
+ *                      the operation.
+ *
+ * @return              Returns the resulting KineticStatus.
+ */
+KineticStatus KineticClient_BatchStart(KineticBatch_Operation* const batchOp);
+
+/**
+ * @brief Commit the batch operation. When this call returned successfully,
+ * all the commands performed in this batch are executed and committed to
+ * store successfully. Otherwise, no commands in this batch were committed
+ * to the persistent store.
+ *
+ * @param batchOp       The session-unique KineticBatch_Operation to use for
+ *                      this operation.
+ *
+ * @return              Returns the resulting KineticStatus.
+ */
+KineticStatus KineticClient_BatchEnd(KineticBatch_Operation* const batchOp);
+
+/**
+ * @brief Abort the current batch operation. When this call returned successfully,
+ * all the commands queued in this batch are aborted. Resources related to this
+ * batch are cleaned up and released.
+ *
+ * @param batchOp       The session-unique KineticBatch_Operation to use for
+ *                      this operation.
+ *
+ * @return              Returns the resulting KineticStatus.
+ */
+KineticStatus KineticClient_BatchAbort(KineticBatch_Operation* const batchOp);
+
 #endif // _KINETIC_CLIENT_H

include/kinetic_types.h

@@ -146,6 +146,12 @@ typedef struct _KineticSessionConfig {
  */
 typedef struct _KineticSession KineticSession;
 
+
+typedef struct {
+	uint32_t batchId;
+	KineticSession* session;
+} KineticBatch_Operation;
+
 /**
 * @brief Kinetic status codes.
 */

src/examples/batch_put_delete.c

@@ -0,0 +1,138 @@
+/**
+ * Copyright 2013-2015 Seagate Technology LLC.
+ *
+ * This Source Code Form is subject to the terms of the Mozilla
+ * Public License, v. 2.0. If a copy of the MPL was not
+ * distributed with this file, You can obtain one at
+ * https://mozilla.org/MP:/2.0/.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but is provided AS-IS, WITHOUT ANY WARRANTY; including without
+ * the implied warranty of MERCHANTABILITY, NON-INFRINGEMENT or
+ * FITNESS FOR A PARTICULAR PURPOSE. See the Mozilla Public
+ * License for more details.
+ *
+ * See www.openkinetic.org for more project information
+ */
+
+#include "kinetic_client.h"
+#include "kinetic_types.h"
+#include <stdlib.h>
+#include <openssl/sha.h>
+
+int main(int argc, char** argv)
+{
+    (void)argc;
+    (void)argv;
+
+    // Initialize kinetic-c and establish session
+    KineticSession* session;
+    KineticClientConfig client_config = {
+        .logFile = "stdout",
+        .logLevel = 0,
+    };
+    KineticClient * client = KineticClient_Init(&client_config);
+    if (client == NULL) { return 1; }
+    const char HmacKeyString[] = "asdfasdf";
+    KineticSessionConfig config = {
+        .host = "127.0.0.1",
+        .port = KINETIC_PORT,
+        .clusterVersion = 0,
+        .identity = 1,
+        .hmacKey = ByteArray_CreateWithCString(HmacKeyString),
+    };
+    KineticStatus connect_status = KineticClient_CreateSession(&config, client, &session);
+    if (connect_status != KINETIC_STATUS_SUCCESS) {
+        fprintf(stderr, "Failed connecting to the Kinetic device w/status: %s\n",
+            Kinetic_GetStatusDescription(connect_status));
+        exit(1);
+    }
+
+    //prepare data for foo entry
+    uint8_t foo_value_data[] = { 0x00, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x09, 0x0A, 0x0B, 0x0C, 0x0D, 0x0E, 0x0F };
+    ByteBuffer foo_value = ByteBuffer_MallocAndAppend(foo_value_data, sizeof(foo_value_data));
+    uint8_t foo_key_data[] = {0x00, 0x01, 0x02, 0x03, 0x04};
+    ByteBuffer foo_key = ByteBuffer_MallocAndAppend(foo_key_data, sizeof(foo_key_data));
+    ByteBuffer foo_tag = ByteBuffer_Malloc(20);
+    uint8_t foo_sha1[20];
+    SHA1(foo_value.array.data, foo_value.bytesUsed, &foo_sha1[0]);
+    ByteBuffer_Append(&foo_tag, foo_sha1, sizeof(foo_sha1));
+
+    //prapare data for bar entry
+    uint8_t bar_value_data[] = { 0x01, 0x00, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x09, 0x0A, 0x0B, 0x0C, 0x0D, 0x0E, 0x0F };
+    ByteBuffer bar_value = ByteBuffer_MallocAndAppend(bar_value_data, sizeof(bar_value_data));
+    uint8_t bar_key_data[] = {0x01, 0x00, 0x02, 0x03, 0x04};
+    ByteBuffer bar_key = ByteBuffer_MallocAndAppend(bar_key_data, sizeof(bar_key_data));
+    ByteBuffer bar_tag = ByteBuffer_Malloc(20);
+    uint8_t bar_sha1[20];
+    SHA1(bar_value.array.data, bar_value.bytesUsed, &bar_sha1[0]);
+    ByteBuffer_Append(&bar_tag, bar_sha1, sizeof(bar_sha1));
+
+    KineticEntry put_foo_entry = {
+        .key = foo_key,
+        .tag = foo_tag,
+        .algorithm = KINETIC_ALGORITHM_SHA1,
+        .value = foo_value,
+        .synchronization = KINETIC_SYNCHRONIZATION_WRITETHROUGH,
+    };
+
+    KineticStatus status = KineticClient_Put(session,&put_foo_entry,NULL);
+    if (status != KINETIC_STATUS_SUCCESS) {
+        fprintf(stderr, "PUT failed w/status: %s\n", Kinetic_GetStatusDescription(status));
+        exit(1);
+    }
+
+    // Delete foo entry and put bar entry in a batch
+    KineticBatch_Operation * batchOp = KineticClient_InitBatchOperation(session);
+    KineticEntry delete_foo_entry = {
+    	.key = foo_key,
+    };
+
+    status = KineticClient_BatchStart(batchOp);
+    if (status != KINETIC_STATUS_SUCCESS) {
+        fprintf(stderr, "Start batch failed w/status: %s\n", Kinetic_GetStatusDescription(status));
+        exit(1);
+    }
+
+    status = KineticClient_BatchDelete(batchOp, &delete_foo_entry);
+    if (status != KINETIC_STATUS_SUCCESS) {
+        fprintf(stderr, "Batch delete failed w/status: %s\n", Kinetic_GetStatusDescription(status));
+        exit(1);
+    };
+
+    KineticEntry put_bar_entry = {
+        .key = bar_key,
+        .tag = bar_tag,
+        .algorithm = KINETIC_ALGORITHM_SHA1,
+        .value = bar_value,
+        .synchronization = KINETIC_SYNCHRONIZATION_WRITETHROUGH,
+    };
+    status = KineticClient_BatchPut(batchOp, &put_bar_entry);
+    if (status != KINETIC_STATUS_SUCCESS) {
+        fprintf(stderr, "Batch put failed w/status: %s\n", Kinetic_GetStatusDescription(status));
+        exit(1);
+    };
+
+    status = KineticClient_BatchEnd(batchOp);
+    if (status != KINETIC_STATUS_SUCCESS) {
+        fprintf(stderr, "End batch failed w/status: %s\n", Kinetic_GetStatusDescription(status));
+        exit(1);
+    };
+
+    // Free buffers
+    ByteBuffer_Free(foo_value);
+    ByteBuffer_Free(foo_key);
+    ByteBuffer_Free(foo_tag);
+    ByteBuffer_Free(bar_value);
+    ByteBuffer_Free(bar_key);
+    ByteBuffer_Free(bar_tag);
+
+    // Free batch operation
+    free(batchOp);
+
+    // Shutdown client connection and cleanup
+    KineticClient_DestroySession(session);
+    KineticClient_Shutdown(client);
+
+    return 0;
+}

src/examples/get_key_range.c

@@ -117,18 +117,19 @@ static bool create_entries(KineticSession * const session, const int count)
     static const ssize_t sz = 20;
     char key_buf[sz];
     char value_buf[sz];
+    char tag_buf[sz];
 
     for (int i = 0; i < count; i++) {
 
         ByteBuffer KeyBuffer = ByteBuffer_CreateAndAppendFormattedCString(key_buf, sz, "key_prefix_%02d", i);
         ByteBuffer ValueBuffer = ByteBuffer_CreateAndAppendFormattedCString(value_buf, sz, "val_%02d", i);
-        
+
         /* Populate tag with SHA1 of value */
         ByteBuffer put_tag_buf = ByteBuffer_Malloc(20);
         uint8_t sha1[20];
         SHA1(ValueBuffer.array.data, ValueBuffer.bytesUsed, &sha1[0]);
         ByteBuffer_Append(&put_tag_buf, sha1, sizeof(sha1));
-        
+
         KineticEntry entry = {
             .key = KeyBuffer,
             .value = ValueBuffer,

src/lib/bus/bus.c

@@ -280,6 +280,8 @@ static boxed_msg *box_msg(struct bus *b, bus_user_msg *msg) {
         box->timeout_sec = BUS_DEFAULT_TIMEOUT_SEC;
     }
 
+    box->no_response = msg->no_response;
+
     box->out_seq_id = msg->seq_id;
     box->out_msg_size = msg->msg_size;
 

src/lib/bus/bus_internal_types.h

@@ -40,6 +40,8 @@ typedef struct boxed_msg {
     /** Message send timeout. */
     time_t timeout_sec;
 
+    bool no_response;
+
     /** Callback and userdata to which the bus_msg_result_t above will be sunk. */
     bus_msg_cb *cb;
     void *udata;

src/lib/bus/bus_types.h

@@ -239,7 +239,7 @@ typedef struct {
     uint8_t *msg;
     size_t msg_size;
     uint16_t timeout_sec;
-
+    bool no_response;
     bus_msg_cb *cb;
     void *udata;
 } bus_user_msg;

src/lib/bus/listener_task.c

@@ -129,7 +129,9 @@ static void tick_handler(listener *l) {
         }
     }
 
-    if (b->log_level > 5 || 0) { ListenerTask_DumpRXInfoTable(l); }
+    if (b->log_level > 5 || 0) {
+        ListenerTask_DumpRXInfoTable(l);
+    }
 
     for (int i = 0; i <= l->rx_info_max_used; i++) {
         rx_info_t *info = &l->rx_info[i];
@@ -166,6 +168,13 @@ static void tick_handler(listener *l) {
             }
             break;
         case RIS_EXPECT:
+        	if (info->u.expect.box->no_response)
+        	{
+                info->u.expect.error = RX_ERROR_DONE;
+                info->u.expect.box->result.status = BUS_SEND_SUCCESS;
+                clean_up_completed_info(l, info);
+        	}
+
             any_work = true;
             if (info->u.expect.error == RX_ERROR_READY_FOR_DELIVERY) {
                 BUS_LOG(b, 4, LOG_LISTENER,