implement event-based retry strategy

Author: a114j0y

orchagent/cbf/cbfnhgorch.cpp

@@ -177,6 +177,7 @@ void CbfNhgOrch::doTask(Consumer& consumer)
                 if (success)
                 {
                     m_syncdNextHopGroups.erase(cbf_nhg_it);
+                    notifyRetry(gRouteOrch, APP_ROUTE_TABLE_NAME, make_constraint(RETRY_CST_ECMP));
                 }
             }
         }

orchagent/nhgbase.h

@@ -440,6 +440,13 @@ class NhgOrchCommon : public Orch
         --nhg_entry.ref_count;
     }
 
+    unsigned getNhgRefCount(const string& index)
+    {
+        SWSS_LOG_ENTER();
+        auto& nhg_entry = m_syncdNextHopGroups.at(index);
+        return nhg_entry.ref_count;
+    }
+
     /* Getters / Setters. */
     static inline unsigned getSyncedNhgCount() { return NhgBase::getSyncedCount(); }
 

orchagent/nhgorch.cpp

@@ -23,6 +23,7 @@ extern sai_next_hop_api_t*         sai_next_hop_api;
 NhgOrch::NhgOrch(DBConnector *db, string tableName) : NhgOrchCommon(db, tableName)
 {
     SWSS_LOG_ENTER();
+    createRetryCache(tableName);
 }
 
 /*
@@ -266,6 +267,7 @@ void NhgOrch::doTask(Consumer& consumer)
                         if (nhg->sync())
                         {
                             m_syncdNextHopGroups.emplace(index, NhgEntry<NextHopGroup>(std::move(nhg)));
+                            notifyRetry(gRouteOrch, APP_ROUTE_TABLE_NAME, make_constraint(RETRY_CST_NHG, index));
                         }
                         else
                         {
@@ -300,6 +302,7 @@ void NhgOrch::doTask(Consumer& consumer)
                             success = false;
                         }
                         m_syncdNextHopGroups.emplace(index, NhgEntry<NextHopGroup>(std::move(nhg)));
+                        notifyRetry(gRouteOrch, APP_ROUTE_TABLE_NAME, make_constraint(RETRY_CST_NHG, index));
                     }
                 }
             }
@@ -413,7 +416,9 @@ void NhgOrch::doTask(Consumer& consumer)
             /* If the group does exist, but it's still referenced, skip. */
             else if (nhg_it->second.ref_count > 0)
             {
-                SWSS_LOG_INFO("Unable to remove group %s which is referenced", index.c_str());
+                SWSS_LOG_INFO("Unable to remove group %s which is referenced, move task entry to RetryCache", index.c_str());
+                consumer.addToRetry(std::move(it->second), make_constraint(RETRY_CST_NHG_REF, index));
+                success = true;
             }
             /* Else, if the group is no more referenced, remove it. */
             else
@@ -425,6 +430,7 @@ void NhgOrch::doTask(Consumer& consumer)
                 if (success)
                 {
                     m_syncdNextHopGroups.erase(nhg_it);
+                    notifyRetry(gRouteOrch, APP_ROUTE_TABLE_NAME, make_constraint(RETRY_CST_ECMP));
                 }
             }
         }

orchagent/orch.cpp

@@ -147,7 +147,71 @@ 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());
+}
+
+void ConsumerBase::addToRetry(const Task &task, const Constraint &cst) {
+    Recorder::Instance().retry.record(dumpTuple(task).append(CACHE));
+    if (getOrch())
+        getOrch()->getRetryCache(getName())->cache_failed_task(task, cst);
+}
+
+void Orch::addToRetry(const std::string &executorName, const Task &task, const Constraint &cst) {
+    getRetryCache(executorName)->cache_failed_task(task, cst); 
+}
+
+size_t Orch::retryToSync(const std::string &executorName, size_t threshold)
+{
+    auto retryCache = getRetryCache(executorName);
+
+    if (!retryCache || threshold <= 0)
+        return 0;
+
+    std::unordered_set<Constraint>& constraints = retryCache->getResolvedConstraints();
+
+    size_t count = 0;
+
+    for (auto it = constraints.begin(); it != constraints.end() && count < threshold;)
+    {
+        auto cst = *it++;
+
+        auto tasks = retryCache->resolve(cst, threshold - count);
+
+        count += tasks->size();
+
+        getConsumerBase(executorName)->addToSync(tasks, true);
+
+    }
+    return count;
+}
+
+void Orch::notifyRetry(Orch *retryOrch, const std::string &executorName, const Constraint &cst)
+{
+    retryOrch->getRetryCache(executorName)->add_resolution(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();
 
@@ -157,6 +221,24 @@ void ConsumerBase::addToSync(const KeyOpFieldsValuesTuple &entry)
     /* Record incoming tasks */
     Recorder::Instance().swss.record(dumpTuple(entry));
 
+    auto retryCache = getOrch() ? getOrch()->getRetryCache(getName()) : nullptr;
+
+    if (retryCache)
+    {
+        auto it = retryCache->getRetryMap().find(key);
+        if (it != retryCache->getRetryMap().end()) // key exists
+        {
+            if (it->second.second == entry) // skip duplicate task
+                return;
+            
+            auto cache = retryCache->erase_stale_cache(key);
+            Recorder::Instance().retry.record(dumpTuple(*cache).append(DECACHE));
+
+            if (op == SET_COMMAND)
+                m_toSync.emplace(key, std::move(*cache));
+        }
+    }
+
     /*
     * m_toSync is a multimap which will allow one key with multiple values,
     * Also, the order of the key-value pairs whose keys compare equivalent
@@ -230,22 +312,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)
 {

orchagent/orch.h

@@ -26,6 +26,7 @@ extern "C" {
 #include "response_publisher.h"
 #include "recorder.h"
 #include "schema.h"
+#include "retrycache.h"
 
 const char delimiter           = ':';
 const char list_item_delimiter = ',';
@@ -174,11 +175,18 @@ class ConsumerBase : public Executor {
     /* record the tuple */
     void recordTuple(const swss::KeyOpFieldsValuesTuple &tuple);
 
-    void addToSync(const swss::KeyOpFieldsValuesTuple &entry);
+    void addToSync(const swss::KeyOpFieldsValuesTuple &entry, bool onRetry=false);
 
     // Returns: the number of entries added to m_toSync
-    size_t addToSync(const std::deque<swss::KeyOpFieldsValuesTuple> &entries);
-    size_t addToSync(std::shared_ptr<std::deque<swss::KeyOpFieldsValuesTuple>> entries);
+    size_t addToSync(const std::deque<swss::KeyOpFieldsValuesTuple> &entries, bool onRetry=false);
+    size_t addToSync(std::shared_ptr<std::deque<swss::KeyOpFieldsValuesTuple>> entries, bool onRetry=false); 
+
+    /**
+     * Move a task to retry cache for future processing
+     * @param task a task tuple
+     * @param cst the constraint for the task
+     */
+    void addToRetry(const Task &task, const Constraint &cst);
 
     size_t refillToSync();
     size_t refillToSync(swss::Table* table);
@@ -264,6 +272,7 @@ typedef enum
 typedef std::pair<swss::DBConnector *, std::string> TableConnector;
 typedef std::pair<swss::DBConnector *, std::vector<std::string>> TablesConnector;
 
+
 class Orch
 {
 public:
@@ -296,13 +305,33 @@ class Orch
     virtual void doTask(swss::SelectableTimer &timer) { }
 
     void dumpPendingTasks(std::vector<std::string> &ts);
+    
+    void createRetryCache(const std::string &executorName);
+    RetryCache* getRetryCache(const std::string &executorName);
+    ConsumerBase* getConsumerBase(const std::string &executorName);
+
+    // Add a task and its constraint to the retry cache 
+    void addToRetry(const std::string &executorName, const Task &task, const Constraint &cst);
+
+    /** Delete tasks whose constraints are resolved in this executor's retry cache , then add them back to its m_toSync.
+     * @param executorName name of the executor (actually a ConsumerBase instance)
+     * @param cst task constraint **/
+    virtual size_t retryToSync(const std::string &executorName, size_t threshold=30000);
+
+    /** Notify the executor that the constraint is already resolved
+     * @param retryOrch the orch to be notified
+     * @param executorName name of the executor to be notified
+     * @param cst the constraint that can be resolved
+     * **/
+    virtual void notifyRetry(Orch *retryOrch, const std::string &executorName, const Constraint &cst);
 
     /**
      * @brief Flush pending responses
      */
     void flushResponses();
 protected:
     ConsumerMap m_consumerMap;
+    RetryCacheMap m_retryCaches;
 
     Orch();
     ref_resolve_status resolveFieldRefValue(type_map&, const std::string&, const std::string&, swss::KeyOpFieldsValuesTuple&, sai_object_id_t&, std::string&);