[orchagent] Event-based Retry Strategy (#3699)

What I did
I have implemented RetryCache
allows agents to communicate in the push mode to schedule retry.
Document - RetryCache High Level Design (HLD PR#1822)

The document explains

how to parse the failure reason for a task (constraint)
how to detect the resolution and send notifications
how to quickly archive and restore pending tasks
how to sync the retry cache with the current syncing cac

Author: a114j0y

lib/recorder.cpp

@@ -10,7 +10,7 @@ const std::string Recorder::REC_START = "|recording started";
 const std::string Recorder::SWSS_FNAME = "swss.rec";
 const std::string Recorder::SAIREDIS_FNAME = "sairedis.rec";
 const std::string Recorder::RESPPUB_FNAME = "responsepublisher.rec";
-
+const std::string Recorder::RETRY_FNAME = "retry.rec";
 
 Recorder& Recorder::Instance()
 {
@@ -19,6 +19,17 @@ Recorder& Recorder::Instance()
 }
 
 
+RetryRec::RetryRec() 
+{
+    /* Set Default values */
+    setRecord(true);
+    setRotate(false);
+    setLocation(Recorder::DEFAULT_DIR);
+    setFileName(Recorder::RETRY_FNAME);
+    setName("Retry");
+}
+
+
 SwSSRec::SwSSRec() 
 {
     /* Set Default values */

lib/recorder.h

@@ -48,6 +48,11 @@ class RecWriter : public RecBase {
     std::string fname;
 };
 
+class RetryRec : public RecWriter {
+public:
+    RetryRec();
+};
+
 class SwSSRec : public RecWriter {
 public:
     SwSSRec();
@@ -73,12 +78,14 @@ class Recorder {
     static const std::string SWSS_FNAME;
     static const std::string SAIREDIS_FNAME;
     static const std::string RESPPUB_FNAME;
+    static const std::string RETRY_FNAME;
 
     Recorder() = default;
     /* Individual Handlers */
     SwSSRec swss;
     SaiRedisRec sairedis;
     ResPubRec respub;
+    RetryRec retry;
 };
 
 }

orchagent/main.cpp

@@ -64,6 +64,7 @@ extern bool gIsNatSupported;
 #define SAIREDIS_RECORD_ENABLE 0x1
 #define SWSS_RECORD_ENABLE (0x1 << 1)
 #define RESPONSE_PUBLISHER_RECORD_ENABLE (0x1 << 2)
+#define RETRY_RECORD_ENABLE (0x1 << 3)
 
 /* orchagent heart beat message interval */
 #define HEART_BEAT_INTERVAL_MSECS_DEFAULT 10 * 1000
@@ -113,6 +114,7 @@ void sighup_handler(int signo)
     /*
      * Don't do any logging since they are using mutexes.
      */
+    Recorder::Instance().retry.setRotate(true);
     Recorder::Instance().swss.setRotate(true);
     Recorder::Instance().sairedis.setRotate(true);
     Recorder::Instance().respub.setRotate(true);
@@ -366,10 +368,11 @@ int main(int argc, char **argv)
     string record_location = Recorder::DEFAULT_DIR;
     string swss_rec_filename = Recorder::SWSS_FNAME;
     string sairedis_rec_filename = Recorder::SAIREDIS_FNAME;
+    string retry_rec_filename = Recorder::RETRY_FNAME;
     string zmq_server_address = "";
     string vrf;
     string responsepublisher_rec_filename = Recorder::RESPPUB_FNAME;
-    int record_type = 3; // Only swss and sairedis recordings enabled by default.
+    int record_type = SAIREDIS_RECORD_ENABLE | SWSS_RECORD_ENABLE | RETRY_RECORD_ENABLE; // Only swss, retrycache and sairedis recordings enabled by default.
     long heartBeatInterval = HEART_BEAT_INTERVAL_MSECS_DEFAULT;
 
     // Disable SAI MACSec POST by default. Use option -M to enable it.
@@ -402,7 +405,7 @@ int main(int argc, char **argv)
             // Disable all recordings if atoi() fails i.e. returns 0 due to
             // invalid command line argument.
             record_type = atoi(optarg);
-            if (record_type < 0 || record_type > 7) 
+            if (record_type < 0 || record_type > 15) 
             {
                 usage();
                 exit(EXIT_FAILURE);
@@ -532,6 +535,13 @@ int main(int argc, char **argv)
     Recorder::Instance().respub.setFileName(responsepublisher_rec_filename);
     Recorder::Instance().respub.startRec(false);
 
+    Recorder::Instance().retry.setRecord(
+        (record_type & RETRY_RECORD_ENABLE) == RETRY_RECORD_ENABLE
+    );
+    Recorder::Instance().retry.setLocation(record_location);
+    Recorder::Instance().retry.setFileName(retry_rec_filename);
+    Recorder::Instance().retry.startRec(true);
+
     // Instantiate database connectors
     DBConnector appl_db("APPL_DB", 0);
     DBConnector config_db("CONFIG_DB", 0);

orchagent/orch.cpp

@@ -147,15 +147,177 @@ vector<Selectable *> Orch::getSelectables()
     return selectables;
 }
 
-void ConsumerBase::addToSync(const KeyOpFieldsValuesTuple &entry)
+void Orch::createRetryCache(const std::string &executorName) {
+    if (m_retryCaches.find(executorName) == m_retryCaches.end())
+        m_retryCaches[executorName] = std::make_shared<RetryCache>(executorName);
+}
+
+RetryCache *Orch::getRetryCache(const std::string &executorName)
+{
+    if (m_retryCaches.find(executorName) == m_retryCaches.end())
+        return nullptr;
+    else
+        return m_retryCaches[executorName].get();
+}
+
+ConsumerBase* Orch::getConsumerBase(const std::string &executorName)
+{
+    if (m_consumerMap.find(executorName) == m_consumerMap.end())
+        return nullptr;
+    return dynamic_cast<ConsumerBase*>(m_consumerMap[executorName].get());
+}
+
+bool ConsumerBase::addToRetry(const Task &task, const Constraint &cst) {
+    auto retryCache = getOrch() ? getOrch()->getRetryCache(getName()) : nullptr;
+    if (retryCache)
+    {
+        Recorder::Instance().retry.record(dumpTuple(task).append(CACHE));
+        retryCache->insert(task, cst);
+        return true;
+    }
+    return false;
+}
+
+bool Orch::addToRetry(const std::string &executorName, const Task &task, const Constraint &cst) {
+    auto retryCache = getRetryCache(executorName);
+    if (retryCache)
+    {
+        Recorder::Instance().retry.record(getConsumerBase(executorName)->dumpTuple(task).append(CACHE));
+        retryCache->insert(task, cst);
+        return true;
+    }
+    return false;
+}
+
+/**
+ * @brief Check the consumer's RetryCache, if the set of resolved constraints is not empty,
+ * query RetryMap for failed tasks indexed by these resolved constraints,
+ * and move them back to the consumer's SyncMap, such that they can be retried in the next iteration.
+ * @param executorName - name of the consumer
+ * @param quota - maximum number of tasks to be moved back to SyncMap in a single call
+ * @return number of tasks moved back to SyncMap
+ */
+size_t Orch::retryToSync(const std::string &executorName, size_t quota)
+{
+    auto retryCache = getRetryCache(executorName);
+
+    // directly return 0 if no retry cache for this executor or quota is non-positive
+    if (!retryCache || quota <= 0)
+        return 0;
+
+    std::unordered_set<Constraint>& constraints = retryCache->getResolvedConstraints();
+
+    size_t count = 0;
+
+    while (!constraints.empty() && count < quota)
+    {
+        auto cst = *constraints.begin();
+
+        auto tasks = retryCache->resolve(cst, quota - count);
+
+        count += tasks->size();
+
+        getConsumerBase(executorName)->addToSync(tasks, true);
+
+    }
+    return count;
+}
+
+void Orch::notifyRetry(Orch *retryOrch, const std::string &executorName, const Constraint &cst)
+{
+    auto retryCache = retryOrch->getRetryCache(executorName);
+    if (!retryCache)
+    {
+        SWSS_LOG_ERROR("RetryCache not initialized for %s", executorName.c_str());
+    }
+    else
+    {
+        retryCache->mark_resolved(cst);
+    }
+}
+
+size_t ConsumerBase::addToSync(std::shared_ptr<std::deque<swss::KeyOpFieldsValuesTuple>> entries, bool onRetry) {
+    return addToSync(*entries, onRetry);
+}
+
+void ConsumerBase::addToSync(const KeyOpFieldsValuesTuple &entry, bool onRetry)
 {
     SWSS_LOG_ENTER();
 
     string key = kfvKey(entry);
     string op  = kfvOp(entry);
 
-    /* Record incoming tasks */
-    Recorder::Instance().swss.record(dumpTuple(entry));
+    if (!onRetry)
+        /* Record incoming tasks */
+        Recorder::Instance().swss.record(dumpTuple(entry));
+    else
+        Recorder::Instance().retry.record(dumpTuple(entry).append(DECACHE));
+
+    auto retryCache = getOrch() ? getOrch()->getRetryCache(getName()) : nullptr;
+
+    if (retryCache && !onRetry)
+    {
+        size_t count = retryCache->getRetryMap().count(key);
+
+        switch (count)
+        {
+        case 0:
+            // No task with the same key found in the retrycache
+            break;
+
+        case 1:
+        {
+            // Single task found
+            auto it = retryCache->getRetryMap().find(key);
+            if (it->second.second == entry) // skip duplicate task
+            {
+                SWSS_LOG_DEBUG("Skip, already in retry cache: %s", dumpTuple(entry).c_str());
+                return;
+            }
+
+            if (op == DEL_COMMAND)
+            {
+                if (kfvOp(it->second.second) == SET_COMMAND)
+                {
+                    auto old_task = retryCache->evict(key);
+                    Recorder::Instance().retry.record(dumpTuple(*old_task).append(DECACHE));
+                }
+            }
+            else if (op == SET_COMMAND)
+            {
+                if (kfvOp(it->second.second) == SET_COMMAND)
+                {
+                    // move the old SET back to m_toSync for later merge
+                    auto old_task = retryCache->evict(key);
+                    m_toSync.emplace(key, *old_task);
+                    Recorder::Instance().retry.record(dumpTuple(*old_task).append(DECACHE));
+                }
+            }
+            break;
+        }
+        case 2:
+        {
+            // 2 tasks found, must be a DEL + a SET
+            if (op == DEL_COMMAND)
+            {
+                // remove the SET task from the cache, reuse the DEL task
+                auto old_task = retryCache->evict(key);
+                Recorder::Instance().retry.record(dumpTuple(*old_task).append(DECACHE));
+                return;
+            }
+            else if (op == SET_COMMAND)
+            {
+                // Keep the DEL task, move the old SET back to m_toSync for later merge
+                auto old_task = retryCache->evict(key);
+                Recorder::Instance().retry.record(dumpTuple(*old_task).append(DECACHE));
+                m_toSync.emplace(key, *old_task);
+            }
+            break;
+        }
+        default:
+            SWSS_LOG_ERROR("Maximum two values per key, found: %zu", count);
+        }
+    }
 
     /*
     * m_toSync is a multimap which will allow one key with multiple values,
@@ -230,22 +392,18 @@ void ConsumerBase::addToSync(const KeyOpFieldsValuesTuple &entry)
 
 }
 
-size_t ConsumerBase::addToSync(const std::deque<KeyOpFieldsValuesTuple> &entries)
+size_t ConsumerBase::addToSync(const std::deque<KeyOpFieldsValuesTuple> &entries, bool onRetry)
 {
     SWSS_LOG_ENTER();
 
     for (auto& entry: entries)
     {
-        addToSync(entry);
+        addToSync(entry, onRetry);
     }
 
     return entries.size();
 }
 
-size_t ConsumerBase::addToSync(std::shared_ptr<std::deque<swss::KeyOpFieldsValuesTuple>> entries) {
-    return addToSync(*entries);
-}
-
 // TODO: Table should be const
 size_t ConsumerBase::refillToSync(Table* table)
 {
@@ -326,6 +484,20 @@ void ConsumerBase::dumpPendingTasks(vector<string> &ts)
 
         ts.push_back(s);
     }
+
+    // check pending tasks in m_toRetry if orch has allocated a retry cache for this consumer
+    auto rc = getOrch() ? getOrch()->getRetryCache(getTableName()) : nullptr;
+    if (rc)
+    {
+        for (auto &tm : rc->getRetryMap())
+        {
+            KeyOpFieldsValuesTuple& tuple = tm.second.second;
+
+            string s = dumpTuple(tuple);
+
+            ts.push_back(s);
+        }
+    }
 }
 
 void Consumer::execute()
@@ -663,8 +835,15 @@ string Orch::objectReferenceInfo(
 
 void Orch::doTask()
 {
+    // limit the number of tasks moved from RetryMap to SyncMap in one iteration 
+    // to avoid starvation of new tasks in SyncMap
+    auto threshold = gBatchSize == 0 ? 30000 : gBatchSize;
+
+    size_t count = 0;
+
     for (auto &it : m_consumerMap)
     {
+        count += retryToSync(it.first, threshold - count);
         it.second->drain();
     }
 }