First functional INCR command

pink/Makefile

@@ -63,11 +63,7 @@ ifeq ($(NO_PB),1)
   				$(filter-out $(VERSION_CC) $(wildcard $(SRC_DIR)/pb_*), $(wildcard $(SRC_DIR)/*.cc))
 endif
 
-
-ifndef SLASH_PATH
-  $(warning Warning: missing slash path, using default)
-  SLASH_PATH=$(CURDIR)/third/slash
-endif
+SLASH_PATH=$(CURDIR)/third/slash
 SLASH_INCLUDE_DIR=$(SLASH_PATH)
 SLASH_LIBRARY=$(SLASH_PATH)/slash/lib/libslash.a
 

pink/include/redis_conn.h

@@ -34,9 +34,9 @@ class RedisConn: public PinkConn {
             const int rbuf_max_len = REDIS_MAX_MESSAGE);
   virtual ~RedisConn();
 
-  virtual ReadStatus GetRequest();
-  virtual WriteStatus SendReply();
-  virtual int WriteResp(const std::string& resp);
+  virtual ReadStatus GetRequest() override;
+  virtual WriteStatus SendReply() override;
+  virtual int WriteResp(const std::string& resp) override;
 
   void TryResizeBuffer() override;
   void SetHandleType(const HandleType& handle_type);

pink/rondis/Makefile

@@ -14,10 +14,7 @@ endif
 
 all: rondis
 
-ifndef PINK_PATH
-  $(warning Warning: missing pink path, using default)
-  PINK_PATH=$(CURDIR)/../..
-endif
+PINK_PATH=$(CURDIR)/../..
 PINK_INCLUDE_DIR=$(PINK_PATH)
 PINK_LIBRARY=$(PINK_PATH)/pink/lib/libpink.a
 

pink/rondis/common.h

@@ -1,6 +1,9 @@
 #include <ndbapi/NdbApi.hpp>
 #include <ndbapi/Ndb.hpp>
 
+#ifndef RONDIS_COMMON_H
+#define RONDIS_COMMON_H
+
 #define MAX_CONNECTIONS 2
 
 #define REDIS_DB_NAME "redis"
@@ -20,6 +23,7 @@ void assign_generic_err_to_response(std::string *response, const char *app_str);
 #define FAILED_CREATE_TXN_OBJECT "Failed to create transaction object"
 #define FAILED_EXEC_TXN "Failed to execute transaction"
 #define FAILED_READ_KEY "Failed to read key"
+#define FAILED_INCR_KEY "Failed to increment key"
 #define FAILED_GET_OP "Failed to get NdbOperation object"
 #define FAILED_DEFINE_OP "Failed to define RonDB operation"
 
@@ -28,3 +32,4 @@ void assign_generic_err_to_response(std::string *response, const char *app_str);
 #define REDIS_WRONG_NUMBER_OF_ARGS "wrong number of arguments for '%s' command"
 #define REDIS_NO_SUCH_KEY "$-1\r\n"
 #define REDIS_KEY_TOO_LARGE "key is too large (3000 bytes max)"
+#endif

pink/rondis/rondb.cc

@@ -140,6 +140,19 @@ int rondb_redis_handler(const pink::RedisCmdArgsType &argv,
                 assign_generic_err_to_response(response, error_message);
             }
         }
+        else if (argv[0] == "INCR")
+        {
+            if (argv.size() == 2)
+            {
+                rondb_incr_command(ndb, argv, response);
+            }
+            else
+            {
+                char error_message[256];
+                snprintf(error_message, sizeof(error_message), REDIS_WRONG_NUMBER_OF_ARGS, argv[0].c_str());
+                assign_generic_err_to_response(response, error_message);
+            }
+        }
         else
         {
             printf("Unsupported command: ");

pink/rondis/rondb.h

@@ -3,6 +3,9 @@
 #include <ndbapi/NdbApi.hpp>
 #include <ndbapi/Ndb.hpp>
 
+#ifndef RONDIS_RONDB_H
+#define RONDIS_RONDB_H
+
 extern std::vector<Ndb *> ndb_objects;
 
 int initialize_ndb_objects(const char *connect_string, int num_ndb_objects);
@@ -14,3 +17,4 @@ void rondb_end();
 int rondb_redis_handler(const pink::RedisCmdArgsType &argv,
                         std::string *response,
                         int fd);
+#endif

pink/rondis/rondis.cc

@@ -115,7 +115,7 @@ static void SignalSetup()
 int main(int argc, char *argv[])
 {
     int port = 6379;
-    char *connect_string = "localhost:13000";
+    const char *connect_string = "localhost:13000";
     int worker_threads = 2;
     if (argc != 4)
     {

pink/rondis/string/commands.cc

@@ -6,9 +6,56 @@
 #include <ndbapi/Ndb.hpp>
 
 #include "db_operations.h"
+#include "commands.h"
 #include "../common.h"
 #include "table_definitions.h"
 
+bool setup_transaction(
+    Ndb *ndb,
+    const pink::RedisCmdArgsType &argv,
+    std::string *response,
+    struct key_table *key_row,
+    const char *key_str,
+    Uint32 key_len,
+    const NdbDictionary::Dictionary **ret_dict,
+    const NdbDictionary::Table **ret_tab,
+    NdbTransaction **ret_trans)
+{
+    if (key_len > MAX_KEY_VALUE_LEN)
+    {
+        assign_generic_err_to_response(response, REDIS_KEY_TOO_LARGE);
+        return false;
+    }
+    const NdbDictionary::Dictionary *dict = ndb->getDictionary();
+    if (dict == nullptr)
+    {
+        assign_ndb_err_to_response(response, FAILED_GET_DICT, ndb->getNdbError());
+        return false;
+    }
+    const NdbDictionary::Table *tab = dict->getTable(KEY_TABLE_NAME);
+    if (tab == nullptr)
+    {
+        assign_ndb_err_to_response(response, FAILED_CREATE_TABLE_OBJECT, dict->getNdbError());
+        return false;
+    }
+    memcpy(&key_row->redis_key[2], key_str, key_len);
+    set_length((char*)&key_row->redis_key[0], key_len);
+    NdbTransaction *trans = ndb->startTransaction(tab,
+                                                  &key_row->redis_key[0],
+                                                  key_len + 2);
+    if (trans == nullptr)
+    {
+        assign_ndb_err_to_response(response,
+                                   FAILED_CREATE_TXN_OBJECT,
+                                   ndb->getNdbError());
+        return false;
+    }
+    *ret_tab = tab;
+    *ret_trans = trans;
+    *ret_dict = dict;
+    return true;
+}
+
 /*
     A successful GET will return in this format:
         $5
@@ -27,45 +74,22 @@ void rondb_get_command(Ndb *ndb,
                        const pink::RedisCmdArgsType &argv,
                        std::string *response)
 {
-    Uint32 key_len = argv[1].size();
-    if (key_len > MAX_KEY_VALUE_LEN)
-    {
-        assign_generic_err_to_response(response, REDIS_KEY_TOO_LARGE);
-        return;
-    }
-    const char *key_str = argv[1].c_str();
-
-    const NdbDictionary::Dictionary *dict = ndb->getDictionary();
-    const NdbDictionary::Table *tab = dict->getTable(KEY_TABLE_NAME);
-    if (tab == nullptr)
-    {
-        assign_ndb_err_to_response(response,
-                                   FAILED_CREATE_TABLE_OBJECT,
-                                   dict->getNdbError());
-        return;
-    }
-
+    const NdbDictionary::Dictionary *dict;
+    const NdbDictionary::Table *tab = nullptr;
+    NdbTransaction *trans = nullptr;
     struct key_table key_row;
-    // varbinary -> first 2 bytes are length if bigger than 255
-    // start copying from 3rd byte
-    memcpy(&key_row.redis_key[2], key_str, key_len);
-    // Length as little endian
-    Uint8 *ptr = (Uint8 *)&key_row.redis_key[0];
-    ptr[0] = Uint8(key_len & 255);
-    ptr[1] = Uint8(key_len >> 8);
-
-    // This is (usually) a local operation to calculate the correct data node, using the
-    // hash of the pk value.
-    NdbTransaction *trans = ndb->startTransaction(tab,
-                                                  &(key_row.redis_key[0]),
-                                                  key_len + 2);
-    if (trans == nullptr)
-    {
-        assign_ndb_err_to_response(response,
-                                   FAILED_CREATE_TXN_OBJECT,
-                                   ndb->getNdbError());
-        return;
-    }
+    const char *key_str = argv[1].c_str();
+    Uint32 key_len = argv[1].size();
+    if (!setup_transaction(ndb,
+                           argv,
+                           response,
+                           &key_row,
+                           key_str,
+                           key_len,
+                           &dict,
+                           &tab,
+                           &trans))
+      return;
 
     int ret_code = get_simple_key_row(
         response,
@@ -81,22 +105,11 @@ void rondb_get_command(Ndb *ndb,
     }
     printf("Getting %d value rows\n", key_row.num_rows);
     {
-        // For some reason this needs to be used from scratch
-        struct key_table key_row;
-        // varbinary -> first 2 bytes are length if bigger than 255
-        // start copying from 3rd byte
-        memcpy(&key_row.redis_key[2], key_str, key_len);
-        // Length as little endian
-        Uint8 *ptr = (Uint8 *)&key_row.redis_key[0];
-        ptr[0] = Uint8(key_len & 255);
-        ptr[1] = Uint8(key_len >> 8);
-
         /*
             Our value uses value rows, so a more complex read is required.
             We're starting from scratch here since we'll use a shared lock
             on the key table this time we read from it.
         */
-
         NdbTransaction *trans = ndb->startTransaction(tab,
                                                       &(key_row.redis_key[0]),
                                                       key_len + 2);
@@ -124,37 +137,25 @@ void rondb_set_command(
     const pink::RedisCmdArgsType &argv,
     std::string *response)
 {
+    const NdbDictionary::Dictionary *dict;
+    const NdbDictionary::Table *tab = nullptr;
+    NdbTransaction *trans = nullptr;
+    struct key_table key_row;
+    const char *key_str = argv[1].c_str();
     Uint32 key_len = argv[1].size();
-    if (key_len > MAX_KEY_VALUE_LEN)
-    {
-        assign_generic_err_to_response(response, REDIS_KEY_TOO_LARGE);
-        return;
-    }
+    if (!setup_transaction(ndb,
+                           argv,
+                           response,
+                           &key_row,
+                           key_str,
+                           key_len,
+                           &dict,
+                           &tab,
+                           &trans))
+      return;
 
-    const char *key_str = argv[1].c_str();
     const char *value_str = argv[2].c_str();
     Uint32 value_len = argv[2].size();
-
-    const NdbDictionary::Dictionary *dict = ndb->getDictionary();
-    if (dict == nullptr)
-    {
-        assign_ndb_err_to_response(response, FAILED_GET_DICT, ndb->getNdbError());
-        return;
-    }
-    const NdbDictionary::Table *tab = dict->getTable(KEY_TABLE_NAME);
-    if (tab == nullptr)
-    {
-        assign_ndb_err_to_response(response, FAILED_CREATE_TABLE_OBJECT, dict->getNdbError());
-        return;
-    }
-
-    NdbTransaction *trans = ndb->startTransaction(tab, key_str, key_len);
-    if (trans == nullptr)
-    {
-        assign_ndb_err_to_response(response, FAILED_CREATE_TXN_OBJECT, ndb->getNdbError());
-        return;
-    }
-
     char varsize_param[EXTENSION_VALUE_LEN + 500];
     Uint32 num_value_rows = 0;
     Uint64 rondb_key = 0;
@@ -213,7 +214,9 @@ void rondb_set_command(
                 is best done via a cascade delete. We do a delete & insert in
                 a single transaction (plus writing the value rows).
             */
-            trans = ndb->startTransaction(tab, key_str, key_len);
+            NdbTransaction *trans = ndb->startTransaction(tab,
+                                                          &(key_row.redis_key[0]),
+                                                          key_len + 2);
             if (trans == nullptr)
             {
                 assign_ndb_err_to_response(response, FAILED_CREATE_TXN_OBJECT, ndb->getNdbError());
@@ -246,7 +249,6 @@ void rondb_set_command(
         return;
     }
     printf("Inserting %d value rows\n", num_value_rows);
-
     create_all_value_rows(response,
                           ndb,
                           dict,
@@ -259,3 +261,34 @@ void rondb_set_command(
     ndb->closeTransaction(trans);
     return;
 }
+
+void rondb_incr_command(
+    Ndb *ndb,
+    const pink::RedisCmdArgsType &argv,
+    std::string *response)
+{
+    const NdbDictionary::Dictionary *dict;
+    const NdbDictionary::Table *tab = nullptr;
+    NdbTransaction *trans = nullptr;
+    struct key_table key_row;
+    const char *key_str = argv[1].c_str();
+    Uint32 key_len = argv[1].size();
+    if (!setup_transaction(ndb,
+                           argv,
+                           response,
+                           &key_row,
+                           key_str,
+                           key_len,
+                           &dict,
+                           &tab,
+                           &trans))
+      return;
+
+    incr_key_row(response,
+                 ndb,
+                 tab,
+                 trans,
+                 &key_row);
+    ndb->closeTransaction(trans);
+    return;
+}

pink/rondis/string/commands.h

@@ -4,9 +4,10 @@
 #include "pink/include/redis_conn.h"
 #include <ndbapi/NdbApi.hpp>
 #include <ndbapi/Ndb.hpp>
-
 #include "db_operations.h"
 
+#ifndef STRING_COMMANDS_H
+#define STRING_COMMANDS_H
 /*
     All STRING commands:
     https://redis.io/docs/latest/commands/?group=string
@@ -21,6 +22,8 @@
     Most importantly, it writes Ndb error messages to the response string. This may
     however change in the future, since this causes redundancy.
 */
+void set_length(char* buf, Uint32 key_len);
+Uint32 get_length(char* buf);
 
 void rondb_get_command(Ndb *ndb,
                        const pink::RedisCmdArgsType &argv,
@@ -29,3 +32,8 @@ void rondb_get_command(Ndb *ndb,
 void rondb_set_command(Ndb *ndb,
                        const pink::RedisCmdArgsType &argv,
                        std::string *response);
+
+void rondb_incr_command(Ndb *ndb,
+                        const pink::RedisCmdArgsType &argv,
+                        std::string *response);
+#endif