Add REDIS async support

common/dbconnector.cpp

@@ -688,4 +688,67 @@ shared_ptr<string> DBConnector::blpop(const string &list, int timeout)
     throw runtime_error("GET failed, memory exception");
 }
 
+
+/*******************************************************************************
+    / \   ___ _   _ _ __   ___
+   / _ \ / __| | | | '_ \ / __|
+  / ___ \\__ \ |_| | | | | (__
+ /_/   \_\___/\__, |_| |_|\___|
+              |___/
+*******************************************************************************/
+DBConnector_async::DBConnector_async(const std::string  & db_name,
+                                     void               * user_ctx_p) :
+    db_name_m(db_name),
+    db_id_m(SonicDBConfig::getDbId(db_name_m)),
+    sock_addr_m(SonicDBConfig::getDbSock(db_name_m)),
+    user_ctx_pm(user_ctx_p)
+{
+    ac_pm = redisAsyncConnectUnix(sock_addr_m.c_str());
+
+    if (ac_pm->err != 0)
+    {
+        std::string errmsg("Unable to connect to redis: (" + std::to_string(ac_pm->err) + ')');
+
+        if ((ac_pm->errstr != nullptr) && (ac_pm->errstr[0] != '\0'))
+             errmsg += " " + std::string(ac_pm->errstr);
+
+        redisAsyncFree(ac_pm);
+        ac_pm = nullptr;
+
+        throw std::system_error(std::make_error_code(errc::address_not_available), errmsg);
+    }
+
+    ac_pm->data = this;
+
+    command(nullptr, nullptr, "SELECT %d", db_id_m);
+}
+
+DBConnector_async::~DBConnector_async()
+{
+    if (ac_pm != nullptr)
+    {
+        // We can't use redisAsyncFree() here because there may
+        // be pending messages to be sent or received. redisAsyncDisconnect()
+        // will ensure that all pending messages are processed before the
+        // context gets deleted.
+        redisAsyncDisconnect(ac_pm);
+        ac_pm = nullptr;
+    }
+}
+
+int DBConnector_async::command(redisCallbackFn * cb_func_p, void * cb_data_p, const char * format_p, ...)
+{
+    va_list ap;
+    int status;
+    va_start(ap,format_p);
+    status = redisvAsyncCommand(ac_pm, cb_func_p, cb_data_p, format_p, ap);
+    va_end(ap);
+    return status;
+}
+
+int DBConnector_async::formatted_command(redisCallbackFn * cb_func_p, void * cb_data_p, const char * cmd_p, size_t len)
+{
+    return redisAsyncFormattedCommand(ac_pm, cb_func_p, cb_data_p, cmd_p, len);
+}
+
 }

common/dbconnector.h

@@ -8,6 +8,7 @@
 #include <memory>
 
 #include <hiredis/hiredis.h>
+#include <hiredis/async.h>      // redisAsyncContext
 #include "rediscommand.h"
 #include "redisreply.h"
 #define EMPTY_NAMESPACE std::string()
@@ -171,5 +172,148 @@ void DBConnector::hmset(const std::string &key, InputIterator start, InputIterat
     RedisReply r(this, shmset, REDIS_REPLY_STATUS);
 }
 
+
+
+/*******************************************************************************
+    / \   ___ _   _ _ __   ___
+   / _ \ / __| | | | '_ \ / __|
+  / ___ \\__ \ |_| | | | | (__
+ /_/   \_\___/\__, |_| |_|\___|
+              |___/
+*******************************************************************************/
+
+/**
+ * @brief   This uses hiredis asynchronous APIs (hiredis/async.h) to connect
+ *          to the REDIS server.
+ *
+ * @details Asyncronous architecture means that the application is designed
+ *          around an event loop that calls callback functions for each event.
+ *          The hiredis library provides a number of adapters for 3rd party
+ *          event loop libraries such as GLib, libevent, qt, etc.
+ *          Ref: <a href="https://github.com/redis/hiredis/tree/master/adapters">hiredis/adapters/[*.h]</a>
+ *
+ *          No additional thread is created when using the async hiredis APIs.
+ *          The hiredis context is simply "hooked up" to the event loop and
+ *          that allows the event loop to dispatch the "events" (i.e. replies
+ *          from the REDIS server) to their corresponding callback functions.
+ */
+class DBConnector_async
+{
+public:
+    /**
+     * @brief Asynchronous Connector object to the REDIS server.
+     *
+     * @param db_name_p Name of the DB we want to connect to (e.g. CONFIG_DB,
+     *                  APPL_DB, ...)
+     *
+     * @param user_ctx_p Optional user context (future use).
+     */
+    DBConnector_async(const std::string & db_name_p,
+                      void              * user_ctx_p=nullptr);
+    ~DBConnector_async();
+
+    /**
+     * @brief Get the hiredis context
+     *
+     * @return Return the hiredis async connection context. This API should be
+     *         used when your application needs to hook up the hiredis context
+     *         to the event loop. For example, if your application uses the
+     *         GLib main loop, you would hook up the hiredis context as in the
+     *         example below:
+     *
+     * @code
+     *  // --------------------------------------------------------------------
+     *  // Example showing how to hook up the hiredis context retrived
+     *  // from a DBConnector_async object to a GLib main event loop.
+     *  --------------------------------------------------------------------
+     *  #include <glib.h>                   // g_main_context_default()
+     *  #include <hiredis/adapters/glib.h>  // redis_source_new()
+     *  #include "dbconnector.h"            // class DBConnector_async
+     *
+     *  GMainContext * main_ctx_p = g_main_context_default();
+     *  GMainLoop    * loop_p     = g_main_loop_new(main_ctx_p, FALSE);
+     *  .
+     *  .
+     *
+     *  DBConnector_async  db("APPL_DB");
+     *  .
+     *  .
+     *
+     *  // Hook up hiredis context to GLib main loop
+     *  g_source_attach(redis_source_new(db.context()), main_ctx_p);
+     *  .
+     *  .
+     *
+     *  g_main_loop_run(loop_p);
+     *
+     * @endcode
+     *
+     */
+    redisAsyncContext * context() const { return ac_pm; }
+
+    /**
+     * @brief Return the user context that was provided in the constructor.
+     *
+     * @return User context pointer.
+     */
+    void * get_user_ctx() const { return user_ctx_pm; }
+
+    /**
+     * @brief Get the DB name (e.g. "CONFIG_DB", "APPL_DB", etc.). This is the
+     *        same name that was provided to the constructor.
+     *
+     * @return The DB name associated with this connector.
+     */
+    const std::string & db_name() const { return db_name_m; }
+
+    /**
+     * @brief Get the DB ID (e.g. 0 for "APPL_DB", 4 for "CONFIG_DB", etc.)
+     *
+     * @return The DB ID associated with this connector.
+     */
+    int db_id() const { return db_id_m; }
+
+    /**
+     * @brief Get the socket address associated with this connector.
+     *
+     * @return The Unix Domain Socket name.
+     */
+    const std::string & sock_addr() const { return sock_addr_m; }
+
+    /**
+     * @brief Invoke a REDIS a command
+     *
+     * @param cb_func_p Callback function to be invoked when reply is received
+     * @param cb_data_p User data passed to the callback function
+     * @param format A format string similar to printf(), but specific to
+     *               hiredis. For more info refer to the documentation for <a
+     *               href="https://github.com/redis/hiredis/blob/master/hiredis.c#L531">redisFormatCommand()</a>.
+     *
+     * @return 0 on success, otherwise the status returned by
+     *         redisvAsyncCommand()
+     */
+    int command(redisCallbackFn * cb_func_p, void * cb_data_p, const char * format, ...);
+
+    /**
+     * @brief Invoke a REDIS a command
+     *
+     * @param cb_func_p Callback function to be invoked when reply is received
+     * @param cb_data_p User data passed to the callback function
+     * @param cmd A formatted command to be sent to the REDIS server
+     * @param len The length of %cmd
+     *
+     * @return 0 on success, otherwise the status returned by
+     *         redisAsyncFormattedCommand()
+     */
+    int formatted_command(redisCallbackFn * cb_func_p, void * cb_data_p, const char * cmd_p, size_t len);
+
+private:
+    const std::string    db_name_m;
+    const int            db_id_m = -1;
+    std::string          sock_addr_m;
+    redisAsyncContext  * ac_pm = nullptr;
+    void               * user_ctx_pm = nullptr;
+};
+
 }
 #endif

common/rediscommand.cpp

@@ -18,6 +18,12 @@ RedisCommand::~RedisCommand()
 
 void RedisCommand::format(const char *fmt, ...)
 {
+    if (temp != nullptr)
+    {
+        redisFreeCommand(temp);
+        temp = nullptr;
+    }
+
     va_list ap;
     va_start(ap, fmt);
     int len = redisvFormatCommand(&temp, fmt, ap);
@@ -31,6 +37,12 @@ void RedisCommand::format(const char *fmt, ...)
 
 void RedisCommand::formatArgv(int argc, const char **argv, const size_t *argvlen)
 {
+    if (temp != nullptr)
+    {
+        redisFreeCommand(temp);
+        temp = nullptr;
+    }
+
     int len = redisFormatCommandArgv(&temp, argc, argv, argvlen);
     if (len == -1) {
         throw std::bad_alloc();

common/table.cpp

@@ -241,3 +241,46 @@ string Table::stripSpecialSym(const string &key)
 
     return key;
 }
+
+
+/*******************************************************************************
+    / \   ___ _   _ _ __   ___
+   / _ \ / __| | | | '_ \ / __|
+  / ___ \\__ \ |_| | | | | (__
+ /_/   \_\___/\__, |_| |_|\___|
+              |___/
+*******************************************************************************/
+Table_async::Table_async(DBConnector_async & dbconn_r, const std::string & table_name_r) :
+    TableBase(table_name_r, SonicDBConfig::getSeparator(dbconn_r.db_name())),
+    dbconn_rm(dbconn_r)
+{
+}
+
+Table_async::~Table_async()
+{
+}
+
+int Table_async::hdel(redisCallbackFn * cb_func_p, void * cb_data_p, const std::string & key_r, const std::string & field_r)
+{
+    std::string  key = getKeyName(key_r);
+    swss::RedisCommand cmd;
+    cmd.formatHDEL(key, field_r);
+    return dbconn_rm.formatted_command(cb_func_p, cb_data_p, cmd.c_str(), cmd.length());
+}
+
+int Table_async::hget(redisCallbackFn * cb_func_p, void * cb_data_p, const std::string & key_r, const std::string & field_r)
+{
+    std::string  key = getKeyName(key_r);
+    swss::RedisCommand cmd;
+    cmd.formatHGET(key, field_r);
+    return dbconn_rm.formatted_command(cb_func_p, cb_data_p, cmd.c_str(), cmd.length());
+}
+
+int Table_async::hset(redisCallbackFn * cb_func_p, void * cb_data_p, const std::string & key_r, const std::string & field_r, const std::string & value_r)
+{
+    std::string  key = getKeyName(key_r);
+    swss::RedisCommand cmd;
+    cmd.formatHSET(key, field_r, value_r);
+    return dbconn_rm.formatted_command(cb_func_p, cb_data_p, cmd.c_str(), cmd.length());
+}
+

common/table.h

@@ -220,5 +220,87 @@ class TableName_KeySet {
     std::string getStateHashPrefix() const { return "_"; }
 };
 
-}
+
+/*******************************************************************************
+    / \   ___ _   _ _ __   ___
+   / _ \ / __| | | | '_ \ / __|
+  / ___ \\__ \ |_| | | | | (__
+ /_/   \_\___/\__, |_| |_|\___|
+              |___/
+*******************************************************************************/
+/**
+ * @brief Use hiredis async APIs provided by a DBConnector_async object to
+ *        perform asynchronous Table operations with the REDIS server.
+ *
+ */
+class Table_async : public TableBase {
+public:
+    /**
+     * @brief Provides asynchronous access to a Table in the REDIS DB.
+     *
+     * @param dbconn_r      Asynchronous DB connector object
+     * @param table_name_r  Name of the Table. E.g. "CONFIG_DB", "APPL_DB",
+     *                      etc.
+     */
+    Table_async(DBConnector_async & dbconn_r, const std::string & table_name_r);
+    ~Table_async();
+
+    /**
+     * @brief Perform a HDEL REDIS DB operation
+     *
+     * @param cb_func_p Callback function invoked when the HDEL REPLY is
+     *                  received from the REDIS server
+     * @param cb_data_p User data passed to the callback function
+     * @param key_r     Key to be accessed in the Table
+     * @param field_r   Field to HDEL
+     *
+     * @return The value returned by redisAsyncFormattedCommand(). Typically,
+     *         that would be REDIS_OK=0 on success, and REDIS_ERR=-1 otherwise.
+     */
+    int hdel(redisCallbackFn   * cb_func_p,
+             void              * cb_data_p,
+             const std::string & key_r,
+             const std::string & field_r);
+
+    /**
+     * @brief Perform a HGET REDIS DB operation
+     *
+     * @param cb_func_p Callback function invoked when the HGET REPLY is
+     *                  received from the REDIS server
+     * @param cb_data_p User data passed to the callback function
+     * @param key_r     Key to be accessed in the Table
+     * @param field_r   Field to HGET
+     *
+     * @return The value returned by redisAsyncFormattedCommand(). Typically,
+     *         that would be REDIS_OK=0 on success, and REDIS_ERR=-1 otherwise.
+     */
+    int hget(redisCallbackFn   * cb_func_p,
+             void              * cb_data_p,
+             const std::string & key_r,
+             const std::string & field_r);
+
+    /**
+     * @brief Perform a HSET REDIS DB operation
+     *
+     * @param cb_func_p Callback function invoked when the HSET REPLY is
+     *                  received from the REDIS server
+     * @param cb_data_p User data passed to the callback function
+     * @param key_r     Key to be accessed in the Table
+     * @param field_r   Field to HSET
+     * @param value_r   Value that the field will be HSET to.
+     *
+     * @return The value returned by redisAsyncFormattedCommand(). Typically,
+     *         that would be REDIS_OK=0 on success, and REDIS_ERR=-1 otherwise.
+     */
+    int hset(redisCallbackFn   * cb_func_p,
+             void              * cb_data_p,
+             const std::string & key_r,
+             const std::string & field_r,
+             const std::string & value_r);
+
+private:
+    DBConnector_async          & dbconn_rm;
+};
+
+} // namespace swss
 #endif