Add REDIS async support

common/dbconnector.cpp

@@ -840,3 +840,65 @@ void DBConnector::del(const std::vector<std::string>& keys)
 
     RedisReply r(this, command, REDIS_REPLY_NIL);
 }
+
+/*******************************************************************************
+    / \   ___ _   _ _ __   ___
+   / _ \ / __| | | | '_ \ / __|
+  / ___ \\__ \ |_| | | | | (__
+ /_/   \_\___/\__, |_| |_|\___|
+              |___/
+*******************************************************************************/
+DBConnector_async::DBConnector_async(const std::string  &dbName,
+                                     void               *userCtxPtr) :
+    m_dbName(dbName),
+    m_dbId(SonicDBConfig::getDbId(m_dbName)),
+    m_sockAddr(SonicDBConfig::getDbSock(m_dbName)),
+    m_userCtxPtr(userCtxPtr)
+{
+    m_acPtr = redisAsyncConnectUnix(m_sockAddr.c_str());
+
+    if (m_acPtr->err != 0)
+    {
+        std::string errmsg("Unable to connect to redis: (" + std::to_string(m_acPtr->err) + ')');
+
+        if ((m_acPtr->errstr != nullptr) && (m_acPtr->errstr[0] != '\0'))
+             errmsg += " " + std::string(m_acPtr->errstr);
+
+        redisAsyncFree(m_acPtr);
+        m_acPtr = nullptr;
+
+        throw std::system_error(std::make_error_code(errc::address_not_available), errmsg);
+    }
+
+    m_acPtr->data = this;
+
+    command(nullptr, nullptr, "SELECT %d", m_dbId);
+}
+
+DBConnector_async::~DBConnector_async()
+{
+    if (m_acPtr != 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(m_acPtr);
+        m_acPtr = 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(m_acPtr, 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(m_acPtr, 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()
@@ -235,5 +236,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 p_dbName Name of the DB we want to connect to (e.g. CONFIG_DB,
+     *                  APPL_DB, ...)
+     *
+     * @param userCtxPtr Optional user context (future use).
+     */
+    DBConnector_async(const std::string &dbName,
+                      void              *userCtxPtr=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 m_acPtr; }
+
+    /**
+     * @brief Return the user context that was provided in the constructor.
+     *
+     * @return User context pointer.
+     */
+    void *getUserCtx() const { return m_userCtxPtr; }
+
+    /**
+     * @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 char *getDbName() const { return m_dbName.c_str(); }
+
+    /**
+     * @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 getDbId() const { return m_dbId; }
+
+    /**
+     * @brief Get the socket address associated with this connector.
+     *
+     * @return The Unix Domain Socket name.
+     */
+    const char *getSockAddr() const { return m_sockAddr.c_str(); }
+
+    /**
+     * @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   m_dbName;
+    const int           m_dbId = -1;
+    std::string         m_sockAddr;
+    redisAsyncContext  *m_acPtr = nullptr;
+    void               *m_userCtxPtr = nullptr;
+};
+
+} // namespace swss
 #endif

common/table.cpp

@@ -242,3 +242,45 @@ 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.getDbName())),
+    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

@@ -232,5 +232,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