Support bulk API for buffer, queue, schedulergroup and bulk get

lib/ClientSai.cpp

@@ -1228,6 +1228,38 @@ sai_status_t ClientSai::bulkCreate(
             object_statuses);
 }
 
+sai_status_t ClientSai::bulkCreate(
+        _In_ uint32_t object_count,
+        _In_ const sai_neighbor_entry_t* neighbor_entry,
+        _In_ const uint32_t *attr_count,
+        _In_ const sai_attribute_t **attr_list,
+        _In_ sai_bulk_op_error_mode_t mode,
+        _Out_ sai_status_t *object_statuses)
+{
+    MUTEX();
+    SWSS_LOG_ENTER();
+    REDIS_CHECK_API_INITIALIZED();
+
+    // TODO support mode
+
+    std::vector<std::string> serialized_object_ids;
+
+    // on create vid is put in db by syncd
+    for (uint32_t idx = 0; idx < object_count; idx++)
+    {
+        std::string str_object_id = sai_serialize_neighbor_entry(neighbor_entry[idx]);
+        serialized_object_ids.push_back(str_object_id);
+    }
+
+    return bulkCreate(
+            SAI_OBJECT_TYPE_NEIGHBOR_ENTRY,
+            serialized_object_ids,
+            attr_count,
+            attr_list,
+            mode,
+            object_statuses);
+}
+
 sai_status_t ClientSai::bulkCreate(
         _In_ uint32_t object_count,
         _In_ const sai_my_sid_entry_t* my_sid_entry,
@@ -1416,6 +1448,26 @@ sai_status_t ClientSai::bulkRemove(
     return bulkRemove(SAI_OBJECT_TYPE_FDB_ENTRY, serializedObjectIds, mode, object_statuses);
 }
 
+sai_status_t ClientSai::bulkRemove(
+        _In_ uint32_t object_count,
+        _In_ const sai_neighbor_entry_t *neighbor_entry,
+        _In_ sai_bulk_op_error_mode_t mode,
+        _Out_ sai_status_t *object_statuses)
+{
+    MUTEX();
+    SWSS_LOG_ENTER();
+    REDIS_CHECK_API_INITIALIZED();
+
+    std::vector<std::string> serializedObjectIds;
+
+    for (uint32_t idx = 0; idx < object_count; idx++)
+    {
+        serializedObjectIds.emplace_back(sai_serialize_neighbor_entry(neighbor_entry[idx]));
+    }
+
+    return bulkRemove(SAI_OBJECT_TYPE_NEIGHBOR_ENTRY, serializedObjectIds, mode, object_statuses);
+}
+
 sai_status_t ClientSai::bulkRemove(
         _In_ uint32_t object_count,
         _In_ const sai_my_sid_entry_t *my_sid_entry,
@@ -1586,6 +1638,27 @@ sai_status_t ClientSai::bulkSet(
     return bulkSet(SAI_OBJECT_TYPE_FDB_ENTRY, serializedObjectIds, attr_list, mode, object_statuses);
 }
 
+sai_status_t ClientSai::bulkSet(
+        _In_ uint32_t object_count,
+        _In_ const sai_neighbor_entry_t *neighbor_entry,
+        _In_ const sai_attribute_t *attr_list,
+        _In_ sai_bulk_op_error_mode_t mode,
+        _Out_ sai_status_t *object_statuses)
+{
+    MUTEX();
+    SWSS_LOG_ENTER();
+    REDIS_CHECK_API_INITIALIZED();
+
+    std::vector<std::string> serializedObjectIds;
+
+    for (uint32_t idx = 0; idx < object_count; idx++)
+    {
+        serializedObjectIds.emplace_back(sai_serialize_neighbor_entry(neighbor_entry[idx]));
+    }
+
+    return bulkSet(SAI_OBJECT_TYPE_NEIGHBOR_ENTRY, serializedObjectIds, attr_list, mode, object_statuses);
+}
+
 sai_status_t ClientSai::bulkSet(
         _In_ uint32_t object_count,
         _In_ const sai_my_sid_entry_t *my_sid_entry,
@@ -1649,6 +1722,129 @@ sai_status_t ClientSai::bulkSet(
     return waitForBulkResponse(SAI_COMMON_API_BULK_SET, (uint32_t)serialized_object_ids.size(), object_statuses);
 }
 
+sai_status_t ClientSai::bulkGet(
+        _In_ sai_object_type_t object_type,
+        _In_ uint32_t object_count,
+        _In_ const sai_object_id_t *object_id,
+        _In_ const uint32_t *attr_count,
+        _Inout_ sai_attribute_t **attr_list,
+        _In_ sai_bulk_op_error_mode_t mode,
+        _Out_ sai_status_t *object_statuses)
+{
+    MUTEX();
+    SWSS_LOG_ENTER();
+    REDIS_CHECK_API_INITIALIZED();
+
+    std::vector<std::string> serializedObjectIds;
+
+    SWSS_LOG_NOTICE("SYNCD BULK ENTER SHELL");
+
+    for (uint32_t idx = 0; idx < object_count; idx++)
+    {
+//        SWSS_LOG_NOTICE("SYNCD BULK EMPLACE OID %" PRIx64, object_id[idx]);
+        serializedObjectIds.emplace_back(sai_serialize_object_id(object_id[idx]));
+    }
+
+    return bulkGet(object_type, serializedObjectIds, attr_count, attr_list, mode, object_statuses);
+}
+
+sai_status_t ClientSai::bulkGet(
+        _In_ sai_object_type_t object_type,
+        _In_ const std::vector<std::string> &serialized_object_ids,
+        _In_ const uint32_t *attr_count,
+        _Inout_ sai_attribute_t **attr_list,
+        _In_ sai_bulk_op_error_mode_t mode,
+        _Out_ sai_status_t *object_statuses)
+{
+    SWSS_LOG_ENTER();
+
+    // TODO support mode
+
+    SWSS_LOG_NOTICE("SYNCD BULK ENTER DO");
+
+    std::string str_object_type = sai_serialize_object_type(object_type);
+
+    std::vector<swss::FieldValueTuple> entries;
+
+    for (size_t idx = 0; idx < serialized_object_ids.size(); ++idx)
+    {
+        auto entry = SaiAttributeList::serialize_attr_list(object_type, attr_count[idx], attr_list[idx], false);
+
+        std::string str_attr = Globals::joinFieldValues(entry);
+
+        swss::FieldValueTuple fvtNoStatus(serialized_object_ids[idx] , str_attr);
+
+        entries.push_back(fvtNoStatus);
+    }
+
+    /*
+     * We are adding number of entries to actually add ':' to be compatible
+     * with previous
+     */
+
+    // key:         object_type:count
+    // field:       object_id
+    // value:       object_attrs
+    std::string key = str_object_type + ":" + std::to_string(entries.size());
+
+//    m_recorder->recordBulkGenericGet(str_object_type, entries);
+
+    m_communicationChannel->set(key, entries, REDIS_ASIC_STATE_COMMAND_BULK_GET);
+
+    swss::KeyOpFieldsValuesTuple kco;
+    auto status = waitForBulkGetResponse(object_type, serialized_object_ids, attr_count, attr_list, object_statuses, kco);
+
+//    m_recorder->recordBulkGenericGetResponse(str_object_type, kfvFieldsValues(kco));
+
+    return status;
+}
+
+sai_status_t ClientSai::waitForBulkGetResponse(
+        _In_ sai_object_type_t object_type,
+        _In_ const std::vector<std::string> &serialized_object_ids,
+        _In_ const uint32_t *attr_count,
+        _Inout_ sai_attribute_t **attr_list,
+        _Out_ sai_status_t *object_statuses,
+        _Out_ swss::KeyOpFieldsValuesTuple &kco)
+{
+    SWSS_LOG_ENTER();
+
+    auto status = m_communicationChannel->wait(REDIS_ASIC_STATE_COMMAND_GETRESPONSE, kco);
+
+    auto &values = kfvFieldsValues(kco);
+
+    if (status != SAI_STATUS_SUCCESS)
+    {
+        SWSS_LOG_THROW("logic error, get bulk response returned non zero %d", status);
+        return status;
+    }
+
+    uint32_t attr_index = 0;
+
+    // Fetch all the status
+    for(uint32_t obj_index = 0; obj_index < serialized_object_ids.size(); obj_index++)
+    {
+//        sai_deserialize_status(fvField(values[obj_index]), object_statuses[obj_index]);
+        object_statuses[obj_index] = SAI_STATUS_SUCCESS;
+    }
+//    values.erase(values.begin(), values.begin() + serialized_object_ids.size());
+
+    // Fetch attribute list for each object
+    SaiAttributeList list(object_type, values, false);
+
+//    SWSS_LOG_NOTICE("BULK GET OBJECTS %d", serialized_object_ids.size());
+
+    for(uint32_t obj_index = 0; obj_index < serialized_object_ids.size(); obj_index++)
+    {
+//        SWSS_LOG_NOTICE("BULK GET BEFORE TRANSFER %s %d attr index %d", serialized_object_ids[obj_index].c_str(), list.get_attr_list()[attr_index].value.u32, attr_index);
+        transfer_attributes(object_type, attr_count[obj_index], &(list.get_attr_list()[attr_index]), attr_list[obj_index], false);
+//        SWSS_LOG_NOTICE("BULK GET AFTER TRANSFER %s %d attr index %d", serialized_object_ids[obj_index].c_str(), attr_list[obj_index][0].value.u32);
+        attr_index += attr_count[obj_index];
+    }
+
+    return status;
+}
+
 // BULK RESPONSE HELPERS
 
 sai_status_t ClientSai::waitForBulkResponse(

lib/ClientSai.h

@@ -88,6 +88,15 @@ namespace sairedis
                     _In_ sai_bulk_op_error_mode_t mode,
                     _Out_ sai_status_t *object_statuses) override;
 
+            virtual sai_status_t bulkGet(
+                    _In_ sai_object_type_t object_type,
+                    _In_ uint32_t object_count,
+                    _In_ const sai_object_id_t *object_id,
+                    _In_ const uint32_t *attr_count,
+                    _Inout_ sai_attribute_t **attr_list,
+                    _In_ sai_bulk_op_error_mode_t mode,
+                    _Out_ sai_status_t *object_statuses) override;
+
         public: // stats API
 
             virtual sai_status_t getStats(
@@ -221,6 +230,14 @@ namespace sairedis
                     _In_ sai_bulk_op_error_mode_t mode,
                     _Out_ sai_status_t *object_statuses);
 
+            sai_status_t bulkGet(
+                    _In_ sai_object_type_t object_type,
+                    _In_ const std::vector<std::string> &serialized_object_ids,
+                    _In_ const uint32_t *attr_count,
+                    _Inout_ sai_attribute_t **attr_list,
+                    _In_ sai_bulk_op_error_mode_t mode,
+                    _Out_ sai_status_t *object_statuses);
+
         private: // QUAD API response
 
             /**
@@ -249,6 +266,13 @@ namespace sairedis
                     _Inout_ sai_attribute_t *attr_list);
 
         private: // bulk QUAD API response
+            sai_status_t waitForBulkGetResponse(
+                    _In_ sai_object_type_t object_type,
+                    _In_ const std::vector<std::string> &serialized_object_ids,
+                    _In_ const uint32_t *attr_count,
+                    _Inout_ sai_attribute_t **attr_list,
+                    _Out_ sai_status_t *object_statuses,
+                    _Out_ swss::KeyOpFieldsValuesTuple &kco);
 
             /**
              * @brief Wait for bulk response.

lib/ClientServerSai.cpp

@@ -414,6 +414,31 @@ sai_status_t ClientServerSai::bulkSet(
             object_statuses);
 }
 
+sai_status_t ClientServerSai::bulkGet(
+        _In_ sai_object_type_t object_type,
+        _In_ uint32_t object_count,
+        _In_ const sai_object_id_t *object_id,
+        _In_ const uint32_t *attr_count,
+        _Inout_ sai_attribute_t **attr_list,
+        _In_ sai_bulk_op_error_mode_t mode,
+        _Out_ sai_status_t *object_statuses)
+{
+    MUTEX();
+    SWSS_LOG_ENTER();
+    REDIS_CHECK_API_INITIALIZED();
+
+//    SWSS_LOG_NOTICE("SYNCD BULK GET Sai stub");
+
+    return m_sai->bulkGet(
+            object_type,
+            object_count,
+            object_id,
+            attr_count,
+            attr_list,
+            mode,
+            object_statuses);
+}
+
 // BULK QUAD ENTRY
 
 #define DECLARE_BULK_CREATE_ENTRY(OT,ot)                    \

lib/ClientServerSai.h

@@ -80,6 +80,15 @@ namespace sairedis
                     _In_ sai_bulk_op_error_mode_t mode,
                     _Out_ sai_status_t *object_statuses) override;
 
+            virtual sai_status_t bulkGet(
+                    _In_ sai_object_type_t object_type,
+                    _In_ uint32_t object_count,
+                    _In_ const sai_object_id_t *object_id,
+                    _In_ const uint32_t *attr_count,
+                    _Inout_ sai_attribute_t **attr_list,
+                    _In_ sai_bulk_op_error_mode_t mode,
+                    _Out_ sai_status_t *object_statuses) override;
+
         public: // stats API
 
             virtual sai_status_t getStats(

lib/Recorder.cpp

@@ -608,6 +608,46 @@ void Recorder::recordGenericGetResponse(
     recordLine("G|" + sai_serialize_status(status) + "|" + Globals::joinFieldValues(arguments));
 }
 
+void Recorder::recordBulkGenericGet(
+        _In_ const std::string& objectType,
+        _In_ const std::vector<swss::FieldValueTuple>& entriesWithStatus)
+{
+    SWSS_LOG_ENTER();
+
+    std::string joined;
+
+    for (const auto &e: entriesWithStatus)
+    {
+        // ||obj_id|attr=val|attr=val|status||obj_id|attr=val|attr=val|status
+
+        joined += "||" + fvField(e) + "|" + fvValue(e);
+    }
+
+    // capital 'b' stands for bulk GET operation.
+
+    recordLine("b|" + objectType + joined);
+}
+
+void Recorder::recordBulkGenericGetResponse(
+        _In_ const std::string& objectType,
+        _In_ const std::vector<swss::FieldValueTuple>& entriesWithStatus)
+{
+    SWSS_LOG_ENTER();
+
+    std::string joined;
+
+    for (const auto &e: entriesWithStatus)
+    {
+        // ||obj_id|attr=val|attr=val|status||obj_id|attr=val|attr=val|status
+
+        joined += "||" + fvField(e) + "|" + fvValue(e);
+    }
+
+    // capital 'B' stands for bulk GET operation.
+
+    recordLine("B|" + objectType + joined);
+}
+
 void Recorder::recordGenericGetStats(
         _In_ sai_object_type_t object_type,
         _In_ sai_object_id_t object_id,

lib/Recorder.h

@@ -160,6 +160,14 @@ namespace sairedis
             void recordGenericResponse(
                     _In_ sai_status_t status);
 
+            void recordBulkGenericGet(
+                    _In_ const std::string& objectType,
+                    _In_ const std::vector<swss::FieldValueTuple>& entriesWithStatus);
+
+            void recordBulkGenericGetResponse(
+                    _In_ const std::string& objectType,
+                    _In_ const std::vector<swss::FieldValueTuple>& entriesWithStatus);
+
         public: // create ENTRY
 
             SAI_REDIS_RECORDER_DECLARE_RECORD_CREATE(fdb_entry);