[P4RT] Use zmq to send msgs to OrchAgent instead of redis notifications.

Author: kishanps

p4rt_app/BUILD.bazel

@@ -32,6 +32,7 @@ cc_binary(
         "@com_github_gflags_gflags//:gflags",
         "@com_github_grpc_grpc//:grpc++_authorization_provider",
         "@com_google_absl//absl/functional:bind_front",
+        "@com_google_absl//absl/memory",
         "@com_google_absl//absl/status",
         "@com_google_absl//absl/status:statusor",
         "@com_google_absl//absl/strings",
@@ -51,11 +52,13 @@ cc_binary(
         "//p4rt_app/event_monitoring:state_event_monitor",
         "//p4rt_app/event_monitoring:state_verification_events",
         "//p4rt_app/p4runtime:p4runtime_impl",
+        "//p4rt_app/sonic/adapters:subscriber_state_table_adapter",
+        "//p4rt_app/sonic/adapters:producer_state_table_adapter",
+        "//p4rt_app/sonic/adapters:zmq_producer_state_table_adapter",
         "//p4rt_app/sonic:packetio_impl",
         "//p4rt_app/sonic:redis_connections",
         "//p4rt_app/sonic/adapters:consumer_notifier_adapter",
         "//p4rt_app/sonic/adapters:notification_producer_adapter",
-        "//p4rt_app/sonic/adapters:producer_state_table_adapter",
         "//p4rt_app/sonic/adapters:system_call_adapter",
         "//p4rt_app/sonic/adapters:table_adapter",
         "//p4rt_app/sonic/adapters:warm_boot_state_adapter",

p4rt_app/p4rt.cc

@@ -21,6 +21,7 @@
 
 #include "absl/flags/parse.h"
 #include "absl/functional/bind_front.h"
+#include "absl/memory/memory.h"
 #include "absl/status/status.h"
 #include "absl/status/statusor.h"
 #include "absl/strings/str_cat.h"
@@ -54,13 +55,16 @@
 #include "p4rt_app/sonic/adapters/system_call_adapter.h"
 #include "p4rt_app/sonic/adapters/table_adapter.h"
 #include "p4rt_app/sonic/adapters/warm_boot_state_adapter.h"
+#include "p4rt_app/sonic/adapters/zmq_producer_state_table_adapter.h"
 #include "p4rt_app/sonic/packetio_impl.h"
 //TODO(PINS):
 //#include "swss/component_state_helper.h"
 //#include "swss/component_state_helper_interface.h"
 #include "p4rt_app/sonic/redis_connections.h"
 #include "swss/dbconnector.h"
 #include "swss/schema.h"
+#include "swss/warm_restart.h"
+#include "swss/zmqclient.h"
 
 using ::grpc::Server;
 using ::grpc::ServerBuilder;
@@ -183,17 +187,11 @@ namespace p4rt_app {
 namespace {
 
 sonic::P4rtTable CreateP4rtTable(swss::DBConnector* app_db,
-                                 swss::DBConnector* counters_db) {
-  const std::string kP4rtResponseChannel =
-      std::string("APPL_DB_") + APP_P4RT_TABLE_NAME + "_RESPONSE_CHANNEL";
-
+                                 swss::DBConnector* counters_db,
+                                 swss::ZmqClient* zmq_client) {
   return sonic::P4rtTable{
-      .notification_producer =
-          absl::make_unique<sonic::NotificationProducerAdapter>(
-              app_db, APP_P4RT_CHANNEL_NAME),
-      .notification_consumer =
-          absl::make_unique<sonic::ConsumerNotifierAdapter>(
-              kP4rtResponseChannel, app_db),
+      .producer = absl::make_unique<sonic::ZmqProducerStateTableAdapter>(
+          app_db, APP_P4RT_TABLE_NAME, *zmq_client),
       .app_db = absl::make_unique<p4rt_app::sonic::TableAdapter>(
           app_db, APP_P4RT_TABLE_NAME),
       .counter_db = absl::make_unique<p4rt_app::sonic::TableAdapter>(
@@ -428,9 +426,13 @@ int main(int argc, char** argv) {
   swss::DBConnector counters_db("COUNTERS_DB", /*timeout=*/0);
   swss::DBConnector state_db("STATE_DB", /*timeout=*/0);
 
+  // Zmq request-reply one-to-one connection with the swss server.
+  swss::ZmqClient zmq_client("ipc:///zmq/zmq_swss_ep",
+                             /*waitTimeMs=*/10 * 60 * 1000);
+
   // Create interfaces to interact with the P4RT_TABLE entries.
   p4rt_app::sonic::P4rtTable p4rt_table =
-      p4rt_app::CreateP4rtTable(&app_db, &counters_db);
+      p4rt_app::CreateP4rtTable(&app_db, &counters_db, &zmq_client);
   p4rt_app::sonic::VrfTable vrf_table =
       p4rt_app::CreateVrfTable(&app_db, &app_state_db);
   p4rt_app::sonic::VlanTable vlan_table =

p4rt_app/p4runtime/p4runtime_impl.cc

@@ -1597,7 +1597,7 @@ absl::Status P4RuntimeImpl::ConfigureAppDbTables(
       ASSIGN_OR_RETURN(
           pdpi::IrUpdateStatus status,
           sonic::GetAndProcessResponseNotificationWithoutRevertingState(
-              *p4rt_table_.notification_consumer, acl_key));
+              *p4rt_table_.producer, acl_key));
 
       // Any issue with the forwarding config should be sent back to the
       // controller as an INVALID_ARGUMENT.
@@ -1614,9 +1614,9 @@ absl::Status P4RuntimeImpl::ConfigureAppDbTables(
             _ << "Could not publish Table Definition Set to APPDB");
 
       ASSIGN_OR_RETURN(
-            pdpi::IrUpdateStatus status,
-            sonic::GetAndProcessResponseNotificationWithoutRevertingState(
-                 *p4rt_table_.notification_consumer, acl_key));
+          pdpi::IrUpdateStatus status,
+          sonic::GetAndProcessResponseNotificationWithoutRevertingState(
+              *p4rt_table_.producer, acl_key));
 
       // Any issue with the forwarding config should be sent back to the
       // controller as an INVALID_ARGUMENT.

p4rt_app/sonic/BUILD.bazel

@@ -43,9 +43,9 @@ cc_test(
     deps = [
         ":app_db_acl_def_table_manager",
         ":redis_connections",
-        "//p4rt_app/sonic/adapters:fake_notification_producer_adapter",
         "//gutil/gutil:status_matchers",
         "//p4rt_app/sonic/adapters:fake_sonic_db_table",
+        "//p4rt_app/sonic/adapters:fake_zmq_producer_state_table_adapter",
         "//p4rt_app/utils:ir_builder",
         "@com_github_google_glog//:glog",
         "@com_github_nlohmann_json//:nlohmann_json",
@@ -98,9 +98,9 @@ cc_test(
         "//gutil/gutil:status_matchers",
         "//p4_pdpi:ir_cc_proto",
         "//p4rt_app/sonic/adapters:mock_consumer_notifier_adapter",
-        "//p4rt_app/sonic/adapters:mock_notification_producer_adapter",
         "//p4rt_app/sonic/adapters:mock_producer_state_table_adapter",
         "//p4rt_app/sonic/adapters:mock_table_adapter",
+        "//p4rt_app/sonic/adapters:mock_zmq_producer_state_table_adapter",
         "//p4rt_app/tests/lib:app_db_entry_builder",
         "//sai_p4/instantiations/google:instantiations",
         "//sai_p4/instantiations/google:sai_p4info_cc",
@@ -284,9 +284,9 @@ cc_library(
     hdrs = ["redis_connections.h"],
     deps = [
         "//p4rt_app/sonic/adapters:consumer_notifier_adapter",
-        "//p4rt_app/sonic/adapters:notification_producer_adapter",
         "//p4rt_app/sonic/adapters:producer_state_table_adapter",
         "//p4rt_app/sonic/adapters:table_adapter",
+        "//p4rt_app/sonic/adapters:zmq_producer_state_table_adapter",
     ],
 )
 
@@ -300,6 +300,7 @@ cc_library(
         "//p4_pdpi/utils:ir",
         "//p4rt_app/sonic/adapters:consumer_notifier_adapter",
         "//p4rt_app/sonic/adapters:table_adapter",
+        "//p4rt_app/sonic/adapters:zmq_producer_state_table_adapter",
         "@com_github_google_glog//:glog",
         "@com_google_absl//absl/container:btree",
         "@com_google_absl//absl/status",
@@ -477,8 +478,6 @@ cc_test(
         "//gutil/gutil:proto_matchers",
         "//gutil/gutil:status_matchers",
         "//p4_pdpi:ir_cc_proto",
-        "//p4rt_app/sonic/adapters:mock_consumer_notifier_adapter",
-        "//p4rt_app/sonic/adapters:mock_notification_producer_adapter",
         "//p4rt_app/sonic/adapters:mock_table_adapter",
         "@com_github_p4lang_p4runtime//:p4runtime_cc_proto",
         "@com_google_googleapis//google/rpc:code_cc_proto",

p4rt_app/sonic/adapters/BUILD.bazel

@@ -252,3 +252,34 @@ cc_library(
         "@sonic_swss_common//:common",
     ],
 )
+
+cc_library(
+    name = "zmq_producer_state_table_adapter",
+    srcs = ["zmq_producer_state_table_adapter.cc"],
+    hdrs = ["zmq_producer_state_table_adapter.h"],
+    deps = ["@sonic_swss_common//:libswsscommon"],
+)
+
+cc_library(
+    name = "fake_zmq_producer_state_table_adapter",
+    testonly = True,
+    srcs = ["fake_zmq_producer_state_table_adapter.cc"],
+    hdrs = ["fake_zmq_producer_state_table_adapter.h"],
+    deps = [
+        ":fake_sonic_db_table",
+        ":zmq_producer_state_table_adapter",
+        "@com_google_absl//absl/status",
+        "@sonic_swss_common//:common",
+    ],
+)
+
+cc_library(
+    name = "mock_zmq_producer_state_table_adapter",
+    testonly = True,
+    hdrs = ["mock_zmq_producer_state_table_adapter.h"],
+    deps = [
+        ":zmq_producer_state_table_adapter",
+        "@com_google_googletest//:gtest",
+        "@sonic_swss_common//:common",
+    ],
+)

p4rt_app/sonic/adapters/fake_consumer_notifier_adapter.cc

@@ -33,8 +33,7 @@ FakeConsumerNotifierAdapter::FakeConsumerNotifierAdapter(
 bool FakeConsumerNotifierAdapter::WaitForNotificationAndPop(
     std::string &op, std::string &data, SonicDbEntryList &values,
     int64_t timeout_ms) {
-  sonic_db_table_->GetNextNotification(op, data, values);
-  return true;
+  return sonic_db_table_->GetNextNotification(op, data, values).ok();
 }
 
 }  // namespace sonic

p4rt_app/sonic/adapters/fake_sonic_db_table.cc

@@ -101,12 +101,11 @@ bool FakeSonicDbTable::PushNotification(const std::string &key,
   return true;
 }
 
-void FakeSonicDbTable::GetNextNotification(std::string &op, std::string &data,
-                                           SonicDbEntryList &values) {
+absl::Status FakeSonicDbTable::GetNextNotification(std::string& op,
+                                                   std::string& data,
+                                                   SonicDbEntryList& values) {
   if (notifications_.empty()) {
-    // TODO: we probably want to return a timeout error if we never
-    // get a notification?
-    LOG(FATAL) << "Could not find a notification.";
+    return absl::DeadlineExceededError("No active notification to send");
   }
 
   VLOG(1) << absl::StreamFormat("'%s' get notification: %s", debug_table_name_,
@@ -124,6 +123,7 @@ void FakeSonicDbTable::GetNextNotification(std::string &op, std::string &data,
     op = "SWSS_RC_SUCCESS";
     values.push_back({"err_str", "SWSS_RC_SUCCESS"});
   }
+  return absl::OkStatus();
 }
 
 absl::StatusOr<SonicDbEntryMap> FakeSonicDbTable::ReadTableEntry(

p4rt_app/sonic/adapters/fake_sonic_db_table.h

@@ -22,6 +22,7 @@
 
 #include "absl/base/thread_annotations.h"
 #include "absl/container/flat_hash_map.h"
+#include "absl/status/status.h"
 #include "absl/status/statusor.h"
 #include "absl/synchronization/mutex.h"
 
@@ -67,8 +68,8 @@ class FakeSonicDbTable {
   bool PushNotification(const std::string &key);
   bool PushNotification(const std::string &key, const std::string &op,
                         const SonicDbEntryMap &values);
-  void GetNextNotification(std::string &op, std::string &data,
-                           SonicDbEntryList &values);
+  absl::Status GetNextNotification(std::string& op, std::string& data,
+                                   SonicDbEntryList& values);
 
   absl::StatusOr<SonicDbEntryMap> ReadTableEntry(const std::string &key) const
       ABSL_LOCKS_EXCLUDED(entries_mutex_);

p4rt_app/sonic/adapters/fake_sonic_db_table_test.cc

@@ -101,14 +101,13 @@ TEST(FakeSonicDbTest, DeleteEntry) {
   EXPECT_TRUE(table.GetAllKeys().empty());
 }
 
-TEST(FakeSonicDbDeathTest, GetNotificationDiesIfNoNotificationExists) {
+TEST(FakeSonicDbDeathTest, GetNotificationFailsIfNoNotificationExists) {
   FakeSonicDbTable table;
   std::string op;
   std::string data;
   SonicDbEntryList value;
 
-  EXPECT_DEATH(table.GetNextNotification(op, data, value),
-               "Could not find a notification");
+  EXPECT_FALSE(table.GetNextNotification(op, data, value).ok());
 }
 
 TEST(FakeSonicDbTest, DefaultNotificationResponseIsSuccess) {
@@ -119,7 +118,7 @@ TEST(FakeSonicDbTest, DefaultNotificationResponseIsSuccess) {
 
   // First insert.
   EXPECT_TRUE(table.PushNotification("entry"));
-  table.GetNextNotification(op, data, values);
+  EXPECT_OK(table.GetNextNotification(op, data, values));
   EXPECT_EQ(op, "SWSS_RC_SUCCESS");
   EXPECT_EQ(data, "entry");
   EXPECT_THAT(values,
@@ -135,7 +134,7 @@ TEST(FakeSonicDbTest, SetNotificationResponseForKey) {
   table.SetResponseForKey(/*key=*/"entry", /*code=*/"SWSS_RC_UNKNOWN",
                           /*message=*/"my error message");
   EXPECT_FALSE(table.PushNotification("entry"));
-  table.GetNextNotification(op, data, values);
+  EXPECT_OK(table.GetNextNotification(op, data, values));
   EXPECT_EQ(op, "SWSS_RC_UNKNOWN");
   EXPECT_EQ(data, "entry");
   EXPECT_THAT(values,

p4rt_app/sonic/adapters/fake_zmq_producer_state_table_adapter.cc

@@ -0,0 +1,83 @@
+// Copyright 2025 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+#include "p4rt_app/sonic/adapters/fake_zmq_producer_state_table_adapter.h"
+
+#include <memory>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "absl/status/status.h"
+#include "p4rt_app/sonic/adapters/fake_sonic_db_table.h"
+#include "swss/rediscommand.h"
+#include "swss/schema.h"
+#include "swss/table.h"
+
+namespace p4rt_app {
+namespace sonic {
+
+FakeZmqProducerStateTableAdapter::FakeZmqProducerStateTableAdapter(
+    FakeSonicDbTable* sonic_db_table)
+    : sonic_db_table_(*sonic_db_table) {}
+
+void FakeZmqProducerStateTableAdapter::send(
+    const std::vector<swss::KeyOpFieldsValuesTuple>& kcos) {
+  for (const auto& kfv : kcos) {
+    SonicDbEntryMap entry_map;
+    SonicDbEntryList entry_list;
+    for (const auto& [field, value] : kfvFieldsValues(kfv)) {
+      entry_map[field] = value;
+      entry_list.push_back(std::make_pair(field, value));
+    }
+
+    // Only if the request succeeds should we update AppDb state.
+    if (sonic_db_table_.PushNotification(kfvKey(kfv), kfvOp(kfv), entry_map)) {
+      // Notifications to the OA can only "SET" or "DEL" an entry. So "SET" is
+      // used for both inserting and modifying an entry. Therefore, we always
+      // delete the current entry and only then re-insert it on "SET".
+      sonic_db_table_.DeleteTableEntry(kfvKey(kfv));
+      if (kfvOp(kfv) == "SET") {
+        sonic_db_table_.InsertTableEntry(kfvKey(kfv), entry_list);
+      }
+    }
+  }
+}
+
+bool FakeZmqProducerStateTableAdapter::wait(
+    std::string& db, std::string& table_name,
+    std::vector<std::shared_ptr<swss::KeyOpFieldsValuesTuple>>& kcos) {
+  // db & table_name is fixed now since only p4rt tables use zmq.
+  db = "APPL_DB";
+  table_name = APP_P4RT_TABLE_NAME;
+  std::string op, key;
+  SonicDbEntryList values;
+  while (sonic_db_table_.GetNextNotification(op, key, values).code() !=
+         absl::StatusCode::kDeadlineExceeded) {
+    // Zmq response consists of key, op(empty) and field value tuples of
+    // <status, error_message>. Replace the field in values tuple(originally has
+    // `err_str`) with the actual response code string.
+    values[0].first = op;
+    kcos.push_back(std::make_shared<swss::KeyOpFieldsValuesTuple>(
+        key, "", std::vector<std::pair<std::string, std::string>>(values)));
+    values.clear();
+  }
+  if (kcos.empty()) {
+    return false;
+  }
+
+  return true;
+}
+
+}  // namespace sonic
+}  // namespace p4rt_app