diff --git a/lib/BUILD.bazel b/lib/BUILD.bazel
index 660fd1dc0..461e8ea1b 100644
--- a/lib/BUILD.bazel
+++ b/lib/BUILD.bazel
@@ -26,17 +26,31 @@ cc_library(
     deps = [
         "//gutil:status",
         "//lib/gnmi:gnmi_helper",
+        "//lib/p4rt:packet_listener",
         "//lib/validator:validator_lib",
+        "//p4_pdpi:p4_runtime_session",
+        "//p4_pdpi:ir",
+        "//p4_pdpi:ir_cc_proto",
+        "//p4_pdpi:pd",
+        "//p4_pdpi/packetlib",
+        "//p4_pdpi/packetlib:packetlib_cc_proto",
+        "//sai_p4/instantiations/google:instantiations",
+        "//sai_p4/instantiations/google:sai_p4info_cc",
+        "//tests/forwarding:util",
         "//thinkit:control_interface",
+        "//thinkit:packet_generation_finalizer",
         "//thinkit:switch",
         "@com_github_gnmi//proto/gnmi:gnmi_cc_proto",
         "@com_github_gnoi//diag:diag_cc_grpc_proto",
         "@com_github_gnoi//system:system_cc_grpc_proto",
         "@com_github_google_glog//:glog",
+        "@com_github_nlohmann_json//:nlohmann_json",
         "@com_google_absl//absl/container:flat_hash_map",
         "@com_google_absl//absl/status",
         "@com_google_absl//absl/status:statusor",
         "@com_google_absl//absl/strings",
+        "@com_google_absl//absl/time",
+        "@com_google_absl//absl/types:span",
     ],
 )
 
diff --git a/lib/gnmi/gnmi_helper.cc b/lib/gnmi/gnmi_helper.cc
index a289f8068..78a56e774 100644
--- a/lib/gnmi/gnmi_helper.cc
+++ b/lib/gnmi/gnmi_helper.cc
@@ -205,13 +205,13 @@ absl::Status PushGnmiConfig(thinkit::Switch& chassis,
   return pins_test::PushGnmiConfig(*stub, chassis.ChassisName(), gnmi_config);
 }
 
-absl::Status CanGetAllInterfaceOverGnmi(gnmi::gNMI::Stub& stub,
+absl::Status CanGetAllInterfaceOverGnmi(gnmi::gNMI::StubInterface& stub,
                                         absl::Duration timeout) {
   return GetAllInterfaceOverGnmi(stub).status();
 }
 
 absl::StatusOr<gnmi::GetResponse> GetAllInterfaceOverGnmi(
-    gnmi::gNMI::Stub& stub, absl::Duration timeout) {
+    gnmi::gNMI::StubInterface& stub, absl::Duration timeout) {
   ASSIGN_OR_RETURN(auto req, BuildGnmiGetRequest("", gnmi::GetRequest::ALL));
   gnmi::GetResponse resp;
   grpc::ClientContext context;
@@ -285,12 +285,17 @@ GetInterfaceToOperStatusMapOverGnmi(gnmi::gNMI::StubInterface& stub,
 
 absl::Status CheckAllInterfaceOperStateOverGnmi(
     gnmi::gNMI::StubInterface& stub, absl::string_view interface_oper_state,
-    absl::Duration timeout) {
+    bool skip_non_ethernet_interfaces, absl::Duration timeout) {
   ASSIGN_OR_RETURN(const auto interface_to_oper_status_map,
                    GetInterfaceToOperStatusMapOverGnmi(stub, timeout));
 
   std::vector<std::string> unavailable_interfaces;
   for (const auto& [interface, oper_status] : interface_to_oper_status_map) {
+    if (skip_non_ethernet_interfaces &&
+        !absl::StrContains(interface, "Ethernet")) {
+      LOG(INFO) << "Skipping check on interface: " << interface;
+      continue;
+    }
     if (oper_status != interface_oper_state) {
       unavailable_interfaces.push_back(interface);
     }
@@ -413,6 +418,61 @@ absl::StatusOr<OperStatus> GetInterfaceOperStatusOverGnmi(
   return OperStatus::kUnknown;
 }
 
+absl::StatusOr<absl::flat_hash_map<std::string, std::string>>
+GetAllInterfaceNameToPortId(gnmi::gNMI::StubInterface& stub) {
+  ASSIGN_OR_RETURN(
+      gnmi::GetResponse response,
+      pins_test::GetAllInterfaceOverGnmi(stub, absl::ZeroDuration()));
+  if (response.notification_size() < 1) {
+    return absl::InternalError(
+        absl::StrCat("Invalid response: ", response.DebugString()));
+  }
+
+  const auto response_json = nlohmann::json::parse(
+      response.notification(0).update(0).val().json_ietf_val());
+  const auto oc_intf_json =
+      response_json.find("openconfig-interfaces:interfaces");
+  if (oc_intf_json == response_json.end()) {
+    return absl::NotFoundError(
+        absl::StrCat("'openconfig-interfaces:interfaces' not found: ",
+                     response_json.dump()));
+  }
+  const auto oc_intf_list_json = oc_intf_json->find("interface");
+  if (oc_intf_list_json == oc_intf_json->end()) {
+    return absl::NotFoundError(
+        absl::StrCat("'interface' not found: ", oc_intf_json->dump()));
+  }
+
+  absl::flat_hash_map<std::string, std::string> interface_name_to_port_id;
+  for (auto const& element : oc_intf_list_json->items()) {
+    const auto element_name_json = element.value().find("name");
+    if (element_name_json == element.value().end()) {
+      return absl::NotFoundError(
+          absl::StrCat("'name' not found: ", element.value().dump()));
+    }
+    std::string name = element_name_json->get<std::string>();
+
+    // TODO: Remove once CpuX contains the oper-state subtree.
+    if (absl::StartsWith(name, "Cpu")) {
+      LOG(INFO) << "Skipping " << name << ".";
+      continue;
+    }
+
+    const auto element_interface_state_json = element.value().find("state");
+    if (element_interface_state_json == element.value().end()) {
+      return absl::NotFoundError(
+          absl::StrCat("'state' not found: ", element.value().dump()));
+    }
+    const auto element_id_json = element_interface_state_json->find("id");
+    if (element_id_json == element_interface_state_json->end()) {
+      return absl::NotFoundError(
+          absl::StrCat("'id' not found: ", element.value().dump()));
+    }
+    interface_name_to_port_id[name] = element_id_json->get<std::string>();
+  }
+  return interface_name_to_port_id;
+}
+
 absl::StatusOr<std::vector<std::string>> ParseAlarms(
     const std::string& alarms_json) {
   auto alarms_array = json::parse(alarms_json);
diff --git a/lib/gnmi/gnmi_helper.h b/lib/gnmi/gnmi_helper.h
index 3ed1d0a8b..b62b085a7 100644
--- a/lib/gnmi/gnmi_helper.h
+++ b/lib/gnmi/gnmi_helper.h
@@ -105,10 +105,12 @@ absl::Status PushGnmiConfig(thinkit::Switch& chassis,
                             const std::string& gnmi_config);
 
 absl::Status CanGetAllInterfaceOverGnmi(
-    gnmi::gNMI::Stub& stub, absl::Duration timeout = absl::Seconds(60));
+    gnmi::gNMI::StubInterface& stub,
+    absl::Duration timeout = absl::Seconds(60));
 
 absl::StatusOr<gnmi::GetResponse> GetAllInterfaceOverGnmi(
-    gnmi::gNMI::Stub& stub, absl::Duration timeout = absl::Seconds(60));
+    gnmi::gNMI::StubInterface& stub,
+    absl::Duration timeout = absl::Seconds(60));
 
 // Gets the interface to oper status map.
 absl::StatusOr<absl::flat_hash_map<std::string, std::string>>
@@ -118,6 +120,7 @@ GetInterfaceToOperStatusMapOverGnmi(gnmi::gNMI::StubInterface& stub,
 // Checks if all interfaces oper-status is up/down.
 absl::Status CheckAllInterfaceOperStateOverGnmi(
     gnmi::gNMI::StubInterface& stub, absl::string_view interface_oper_state,
+    bool skip_non_ethernet_interfaces = false,
     absl::Duration timeout = absl::Seconds(60));
 
 // Returns gNMI Path for OC strings.
@@ -132,6 +135,10 @@ absl::StatusOr<std::vector<std::string>> GetUpInterfacesOverGnmi(
 absl::StatusOr<OperStatus> GetInterfaceOperStatusOverGnmi(
     gnmi::gNMI::Stub& stub, absl::string_view if_name);
 
+// Gets the interface name to port id map.
+absl::StatusOr<absl::flat_hash_map<std::string, std::string>>
+GetAllInterfaceNameToPortId(gnmi::gNMI::StubInterface& stub);
+
 // Parses the alarms JSON array returned from a gNMI Get request to
 // "openconfig-system:system/alarms/alarm". Returns the list of alarms.
 absl::StatusOr<std::vector<std::string>> ParseAlarms(
diff --git a/lib/gnmi/gnmi_helper_test.cc b/lib/gnmi/gnmi_helper_test.cc
index 41e9bd15b..61e20f910 100644
--- a/lib/gnmi/gnmi_helper_test.cc
+++ b/lib/gnmi/gnmi_helper_test.cc
@@ -39,10 +39,13 @@ using ::gutil::IsOkAndHolds;
 using ::gutil::StatusIs;
 using ::testing::_;
 using ::testing::DoAll;
+using ::testing::Eq;
+using ::testing::HasSubstr;
 using ::testing::IsEmpty;
 using ::testing::Return;
 using ::testing::SetArgPointee;
 using ::testing::UnorderedElementsAre;
+using ::testing::UnorderedPointwise;
 
 static constexpr char kAlarmsJson[] = R"([
     {
@@ -370,9 +373,9 @@ TEST(GetInterfaceOperStatusMap, GnmiGetRpcFails) {
 TEST(GetInterfaceOperStatusMap, InvalidGnmiGetResponse) {
   gnmi::MockgNMIStub stub;
   EXPECT_CALL(stub, Get).WillOnce(Return(grpc::Status::OK));
-  EXPECT_THAT(GetInterfaceToOperStatusMapOverGnmi(stub, absl::Seconds(60)),
-              StatusIs(absl::StatusCode::kInternal,
-                       testing::HasSubstr("Invalid response")));
+  EXPECT_THAT(
+      GetInterfaceToOperStatusMapOverGnmi(stub, absl::Seconds(60)),
+      StatusIs(absl::StatusCode::kInternal, HasSubstr("Invalid response")));
 }
 
 TEST(GetInterfaceOperStatusMap, GnmiGetResponseWithoutOpenconfigInterface) {
@@ -389,10 +392,10 @@ TEST(GetInterfaceOperStatusMap, GnmiGetResponseWithoutOpenconfigInterface) {
                })pb")),
       Return(grpc::Status::OK)));
 
-  EXPECT_THAT(GetInterfaceToOperStatusMapOverGnmi(stub, absl::Seconds(60)),
-              StatusIs(absl::StatusCode::kNotFound,
-                       testing::HasSubstr(
-                           "'openconfig-interfaces:interfaces' not found")));
+  EXPECT_THAT(
+      GetInterfaceToOperStatusMapOverGnmi(stub, absl::Seconds(60)),
+      StatusIs(absl::StatusCode::kNotFound,
+               HasSubstr("'openconfig-interfaces:interfaces' not found")));
 }
 
 TEST(GetInterfaceOperStatusMap, InterfaceNotFoundInGnmiGetResponse) {
@@ -413,7 +416,7 @@ TEST(GetInterfaceOperStatusMap, InterfaceNotFoundInGnmiGetResponse) {
 
   EXPECT_THAT(GetInterfaceToOperStatusMapOverGnmi(stub, absl::Seconds(60)),
               StatusIs(absl::StatusCode::kNotFound,
-                       testing::HasSubstr("'interface' not found")));
+                       HasSubstr("'interface' not found")));
 }
 
 TEST(GetInterfaceOperStatusMap, InterfaceNameNotFound) {
@@ -432,9 +435,9 @@ TEST(GetInterfaceOperStatusMap, InterfaceNameNotFound) {
                })pb")),
       Return(grpc::Status::OK)));
 
-  EXPECT_THAT(GetInterfaceToOperStatusMapOverGnmi(stub, absl::Seconds(60)),
-              StatusIs(absl::StatusCode::kNotFound,
-                       testing::HasSubstr("'name' not found")));
+  EXPECT_THAT(
+      GetInterfaceToOperStatusMapOverGnmi(stub, absl::Seconds(60)),
+      StatusIs(absl::StatusCode::kNotFound, HasSubstr("'name' not found")));
 }
 
 TEST(GetInterfaceOperStatusMap, InterfaceStateNotFound) {
@@ -453,9 +456,9 @@ TEST(GetInterfaceOperStatusMap, InterfaceStateNotFound) {
                })pb")),
       Return(grpc::Status::OK)));
 
-  EXPECT_THAT(GetInterfaceToOperStatusMapOverGnmi(stub, absl::Seconds(60)),
-              StatusIs(absl::StatusCode::kNotFound,
-                       testing::HasSubstr("'state' not found")));
+  EXPECT_THAT(
+      GetInterfaceToOperStatusMapOverGnmi(stub, absl::Seconds(60)),
+      StatusIs(absl::StatusCode::kNotFound, HasSubstr("'state' not found")));
 }
 
 TEST(GetInterfaceOperStatusMap, OperStatusNotFoundInState) {
@@ -476,7 +479,7 @@ TEST(GetInterfaceOperStatusMap, OperStatusNotFoundInState) {
 
   EXPECT_THAT(GetInterfaceToOperStatusMapOverGnmi(stub, absl::Seconds(60)),
               StatusIs(absl::StatusCode::kNotFound,
-                       testing::HasSubstr("'oper-status' not found")));
+                       HasSubstr("'oper-status' not found")));
 }
 
 TEST(GetInterfaceOperStatusMap, SuccessfullyReturnsInterfaceOperStatusMap) {
@@ -499,8 +502,143 @@ TEST(GetInterfaceOperStatusMap, SuccessfullyReturnsInterfaceOperStatusMap) {
   ASSERT_OK(statusor);
   const absl::flat_hash_map<std::string, std::string> expected_map = {
       {"Ethernet0", "DOWN"}};
-  EXPECT_THAT(*statusor,
-              ::testing::UnorderedPointwise(::testing::Eq(), expected_map));
+  EXPECT_THAT(*statusor, UnorderedPointwise(Eq(), expected_map));
+}
+
+TEST(GetInterfacePortIdMap, SuccessfullyReturnsInterfacePortIdMap) {
+  gnmi::MockgNMIStub stub;
+  EXPECT_CALL(stub, Get).WillOnce(DoAll(
+      SetArgPointee<2>(gutil::ParseProtoOrDie<gnmi::GetResponse>(
+          R"pb(notification {
+                 timestamp: 1620348032128305716
+                 prefix { origin: "openconfig" }
+                 update {
+                   path { elem { name: "interfaces" } }
+                   val {
+                     json_ietf_val: "{\"openconfig-interfaces:interfaces\":{\"interface\":[{\"name\":\"Ethernet0\",\"state\":{\"id\":\"1\"}}]}}"
+                   }
+                 }
+               })pb")),
+      Return(grpc::Status::OK)));
+
+  auto interface_name_to_port_id = GetAllInterfaceNameToPortId(stub);
+  ASSERT_OK(interface_name_to_port_id);
+  const absl::flat_hash_map<std::string, std::string> expected_map = {
+      {"Ethernet0", "1"}};
+  EXPECT_THAT(*interface_name_to_port_id,
+              UnorderedPointwise(Eq(), expected_map));
+}
+
+TEST(GetInterfacePortIdMap, PortIdNotFoundInState) {
+  gnmi::MockgNMIStub stub;
+  EXPECT_CALL(stub, Get).WillOnce(DoAll(
+      SetArgPointee<2>(gutil::ParseProtoOrDie<gnmi::GetResponse>(
+          R"pb(notification {
+                 timestamp: 1620348032128305716
+                 prefix { origin: "openconfig" }
+                 update {
+                   path { elem { name: "interfaces" } }
+                   val {
+                     json_ietf_val: "{\"openconfig-interfaces:interfaces\":{\"interface\":[{\"name\":\"Cpu0\"},{\"name\":\"Ethernet0\",\"state\":{\"name\":\"Ethernet0\"}}]}}"
+                   }
+                 }
+               })pb")),
+      Return(grpc::Status::OK)));
+
+  EXPECT_THAT(
+      GetAllInterfaceNameToPortId(stub),
+      StatusIs(absl::StatusCode::kNotFound, HasSubstr("'id' not found")));
+}
+
+TEST(GetInterfacePortIdMap, InterfaceStateNotFound) {
+  gnmi::MockgNMIStub stub;
+  EXPECT_CALL(stub, Get).WillOnce(DoAll(
+      SetArgPointee<2>(gutil::ParseProtoOrDie<gnmi::GetResponse>(
+          R"pb(notification {
+                 timestamp: 1620348032128305716
+                 prefix { origin: "openconfig" }
+                 update {
+                   path { elem { name: "interfaces" } }
+                   val {
+                     json_ietf_val: "{\"openconfig-interfaces:interfaces\":{\"interface\":[{\"name\":\"Ethernet0\"}]}}"
+                   }
+                 }
+               })pb")),
+      Return(grpc::Status::OK)));
+
+  EXPECT_THAT(
+      GetAllInterfaceNameToPortId(stub),
+      StatusIs(absl::StatusCode::kNotFound, HasSubstr("'state' not found")));
+}
+
+TEST(GetInterfacePortIdMap, InterfaceNameNotFound) {
+  gnmi::MockgNMIStub stub;
+  EXPECT_CALL(stub, Get).WillOnce(DoAll(
+      SetArgPointee<2>(gutil::ParseProtoOrDie<gnmi::GetResponse>(
+          R"pb(notification {
+                 timestamp: 1620348032128305716
+                 prefix { origin: "openconfig" }
+                 update {
+                   path { elem { name: "interfaces" } }
+                   val {
+                     json_ietf_val: "{\"openconfig-interfaces:interfaces\":{\"interface\":[{}]}}"
+                   }
+                 }
+               })pb")),
+      Return(grpc::Status::OK)));
+
+  EXPECT_THAT(
+      GetAllInterfaceNameToPortId(stub),
+      StatusIs(absl::StatusCode::kNotFound, HasSubstr("'name' not found")));
+}
+
+TEST(GetInterfacePortIdMap, InterfaceNotFoundInGnmiGetResponse) {
+  gnmi::MockgNMIStub stub;
+  EXPECT_CALL(stub, Get).WillOnce(DoAll(
+      SetArgPointee<2>(gutil::ParseProtoOrDie<gnmi::GetResponse>(
+          R"pb(notification {
+                 timestamp: 1620348032128305716
+                 prefix { origin: "openconfig" }
+                 update {
+                   path { elem { name: "interfaces" } }
+                   val {
+                     json_ietf_val: "{\"openconfig-interfaces:interfaces\":{}}"
+                   }
+                 }
+               })pb")),
+      Return(grpc::Status::OK)));
+
+  EXPECT_THAT(GetAllInterfaceNameToPortId(stub),
+              StatusIs(absl::StatusCode::kNotFound,
+                       HasSubstr("'interface' not found")));
+}
+
+TEST(GetInterfacePortIdMap, GnmiGetResponseWithoutOpenconfigInterface) {
+  gnmi::MockgNMIStub stub;
+  EXPECT_CALL(stub, Get).WillOnce(DoAll(
+      SetArgPointee<2>(gutil::ParseProtoOrDie<gnmi::GetResponse>(
+          R"pb(notification {
+                 timestamp: 1620348032128305716
+                 prefix { origin: "openconfig" }
+                 update {
+                   path { elem { name: "interfaces" } }
+                   val { json_ietf_val: "{\"openconfig-system:alarms\":{}}" }
+                 }
+               })pb")),
+      Return(grpc::Status::OK)));
+
+  EXPECT_THAT(
+      GetAllInterfaceNameToPortId(stub),
+      StatusIs(absl::StatusCode::kNotFound,
+               HasSubstr("'openconfig-interfaces:interfaces' not found")));
+}
+
+TEST(GetInterfacePortIdMap, InvalidGnmiGetResponse) {
+  gnmi::MockgNMIStub stub;
+  EXPECT_CALL(stub, Get).WillOnce(Return(grpc::Status::OK));
+  EXPECT_THAT(
+      GetAllInterfaceNameToPortId(stub),
+      StatusIs(absl::StatusCode::kInternal, HasSubstr("Invalid response")));
 }
 
 TEST(CheckAllInterfaceOperState, FailsToGetInterfaceOperStatusMap) {
@@ -531,7 +669,7 @@ TEST(CheckAllInterfaceOperState, InterfaceNotUp) {
   EXPECT_THAT(
       CheckAllInterfaceOperStateOverGnmi(stub, /*interface_oper_state=*/"UP"),
       StatusIs(absl::StatusCode::kUnavailable,
-               testing::HasSubstr("Interfaces are not ready")));
+               HasSubstr("Interfaces are not ready")));
 }
 
 TEST(CheckAllInterfaceOperState, InterfaceNotDown) {
@@ -553,7 +691,7 @@ TEST(CheckAllInterfaceOperState, InterfaceNotDown) {
   EXPECT_THAT(
       CheckAllInterfaceOperStateOverGnmi(stub, /*interface_oper_state=*/"DOWN"),
       StatusIs(absl::StatusCode::kUnavailable,
-               testing::HasSubstr("Interfaces are not ready")));
+               HasSubstr("Interfaces are not ready")));
 }
 
 TEST(CheckAllInterfaceOperState, AllInterfacesUp) {
diff --git a/lib/gpins_control_interface.cc b/lib/gpins_control_interface.cc
index 535d84562..a775eef03 100644
--- a/lib/gpins_control_interface.cc
+++ b/lib/gpins_control_interface.cc
@@ -14,18 +14,30 @@
 
 #include "lib/gpins_control_interface.h"
 
+#include "absl/container/flat_hash_map.h"
 #include "absl/status/status.h"
 #include "absl/status/statusor.h"
+#include "absl/strings/escaping.h"
 #include "absl/strings/match.h"
 #include "absl/strings/string_view.h"
 #include "absl/strings/substitute.h"
+#include "absl/time/time.h"
 #include "diag/diag.grpc.pb.h"
 #include "glog/logging.h"
 #include "gutil/status.h"
 #include "lib/gnmi/gnmi_helper.h"
+#include "lib/p4rt/packet_listener.h"
 #include "lib/validator/validator_lib.h"
+#include "p4_pdpi/packetlib/packetlib.h"
+#include "p4_pdpi/packetlib/packetlib.pb.h"
+#include "p4_pdpi/pd.h"
 #include "proto/gnmi/gnmi.pb.h"
+#include "sai_p4/instantiations/google/instantiations.h"
+#include "sai_p4/instantiations/google/sai_p4info.h"
+#include "include/nlohmann/json.hpp"
+#include "tests/forwarding/util.h"
 #include "thinkit/control_interface.h"
+#include "thinkit/packet_generation_finalizer.h"
 
 namespace pins_test {
 
@@ -49,6 +61,42 @@ absl::StatusOr<absl::string_view> GetLinkState(thinkit::LinkState state) {
 
 }  // namespace
 
+GpinsControlInterface::GpinsControlInterface(
+    std::unique_ptr<thinkit::Switch> sut,
+    std::unique_ptr<pdpi::P4RuntimeSession> control_p4_session,
+    pdpi::IrP4Info ir_p4info,
+    absl::flat_hash_map<std::string, std::string> interface_name_to_port_id)
+    : sut_(std::move(sut)),
+      control_p4_session_(std::move(control_p4_session)),
+      ir_p4info_(std::move(ir_p4info)),
+      interface_name_to_port_id_(std::move(interface_name_to_port_id)) {
+  for (const auto& [name, port_id] : interface_name_to_port_id_) {
+    interface_port_id_to_name_[port_id] = name;
+  }
+}
+
+absl::StatusOr<std::unique_ptr<thinkit::PacketGenerationFinalizer>>
+GpinsControlInterface::CollectPackets(thinkit::PacketCallback callback) {
+  return absl::make_unique<PacketListener>(
+      control_p4_session_.get(), &ir_p4info_, &interface_name_to_port_id_,
+      callback);
+}
+
+absl::Status GpinsControlInterface::SendPacket(absl::string_view interface,
+                                               absl::string_view packet) {
+  return gpins::InjectEgressPacket(interface_name_to_port_id_[interface],
+                                   std::string(packet), ir_p4info_,
+                                   control_p4_session_.get());
+}
+
+absl::Status GpinsControlInterface::SendPackets(
+    absl::string_view interface, absl::Span<const std::string> packets) {
+  for (absl::string_view packet : packets) {
+    RETURN_IF_ERROR(SendPacket(interface, packet));
+  }
+  return absl::OkStatus();
+}
+
 absl::Status GpinsControlInterface::SetAdminLinkState(
     absl::Span<const std::string> interfaces, thinkit::LinkState state) {
   ASSIGN_OR_RETURN(auto gnmi_stub, sut_->CreateGnmiStub());
@@ -162,8 +210,6 @@ GpinsControlInterface::GetUpLinks(absl::Span<const std::string> interfaces) {
   return up_links;
 }
 
-absl::Status GpinsControlInterface::CheckUp() {
-  return pins_test::SwitchReady(*sut_);
-}
+absl::Status GpinsControlInterface::CheckUp() { return SwitchReady(*sut_); }
 
 }  // namespace pins_test
diff --git a/lib/gpins_control_interface.h b/lib/gpins_control_interface.h
index e29ee83a1..add3e6fd9 100644
--- a/lib/gpins_control_interface.h
+++ b/lib/gpins_control_interface.h
@@ -16,21 +16,44 @@
 #define GOOGLE_LIB_GPINS_CONTROL_INTERFACE_H_
 
 #include <memory>
+#include <vector>
 
 #include "absl/container/flat_hash_map.h"
-#include "absl/status/status.h"
 #include "absl/status/statusor.h"
+#include "absl/strings/string_view.h"
+#include "absl/types/span.h"
 #include "diag/diag.grpc.pb.h"
+#include "p4_pdpi/p4_runtime_session.h"
+#include "p4_pdpi/ir.h"
+#include "p4_pdpi/ir.pb.h"
+#include "sai_p4/instantiations/google/instantiations.h"
+#include "sai_p4/instantiations/google/sai_p4info.h"
 #include "system/system.grpc.pb.h"
 #include "thinkit/control_interface.h"
+#include "thinkit/packet_generation_finalizer.h"
 #include "thinkit/switch.h"
 
 namespace pins_test {
 
 class GpinsControlInterface : public thinkit::ControlInterface {
  public:
-  GpinsControlInterface(std::unique_ptr<thinkit::Switch> sut)
-      : sut_(std::move(sut)) {}
+  static absl::StatusOr<GpinsControlInterface> CreateGpinsControlInterface(
+      std::unique_ptr<thinkit::Switch> sut);
+
+  GpinsControlInterface(
+      std::unique_ptr<thinkit::Switch> sut,
+      std::unique_ptr<pdpi::P4RuntimeSession> control_p4_session,
+      pdpi::IrP4Info ir_p4info,
+      absl::flat_hash_map<std::string, std::string> interface_name_to_port_id);
+
+  absl::StatusOr<std::unique_ptr<thinkit::PacketGenerationFinalizer>>
+  CollectPackets(thinkit::PacketCallback callback) override;
+
+  absl::Status SendPacket(absl::string_view interface,
+                          absl::string_view packet) override;
+
+  absl::Status SendPackets(absl::string_view interface,
+                           absl::Span<const std::string> packets) override;
 
   absl::Status SetAdminLinkState(absl::Span<const std::string> interfaces,
                                  thinkit::LinkState state) override;
@@ -53,6 +76,10 @@ class GpinsControlInterface : public thinkit::ControlInterface {
 
  private:
   std::unique_ptr<thinkit::Switch> sut_;
+  std::unique_ptr<pdpi::P4RuntimeSession> control_p4_session_;
+  pdpi::IrP4Info ir_p4info_;
+  absl::flat_hash_map<std::string, std::string> interface_name_to_port_id_;
+  absl::flat_hash_map<std::string, std::string> interface_port_id_to_name_;
 };
 
 }  // namespace pins_test
diff --git a/lib/p4rt/BUILD.bazel b/lib/p4rt/BUILD.bazel
index 338d9a8a3..d834d7f0d 100644
--- a/lib/p4rt/BUILD.bazel
+++ b/lib/p4rt/BUILD.bazel
@@ -31,3 +31,25 @@ cc_library(
     ],
 )
 
+cc_library(
+    name = "packet_listener",
+    testonly = 1,
+    srcs = ["packet_listener.cc"],
+    hdrs = ["packet_listener.h"],
+    deps = [
+        "//gutil:status_matchers",
+        "//p4_pdpi:p4_runtime_session",
+        "//p4_pdpi:ir_cc_proto",
+        "//p4_pdpi:pd",
+        "//sai_p4/instantiations/google:instantiations",
+        "//sai_p4/instantiations/google:sai_p4info_cc",
+        "//sai_p4/instantiations/google:sai_pd_cc_proto",
+        "//thinkit:control_interface",
+        "//thinkit:packet_generation_finalizer",
+        "//thinkit:switch",
+        "@com_github_google_glog//:glog",
+        "@com_github_p4lang_p4runtime//:p4runtime_cc_proto",
+        "@com_google_absl//absl/container:flat_hash_map",
+        "@com_google_absl//absl/strings",
+    ],
+)
diff --git a/lib/p4rt/packet_listener.cc b/lib/p4rt/packet_listener.cc
new file mode 100644
index 000000000..e6055e731
--- /dev/null
+++ b/lib/p4rt/packet_listener.cc
@@ -0,0 +1,36 @@
+#include "lib/p4rt/packet_listener.h"
+
+namespace pins_test {
+PacketListener::PacketListener(
+    pdpi::P4RuntimeSession* session, const pdpi::IrP4Info* ir_p4info,
+    const absl::flat_hash_map<std::string, std::string>*
+        interface_port_id_to_name,
+    thinkit::PacketCallback callback)
+    : session_(std::move(session)),
+      receive_packet_thread_([this, ir_p4info, interface_port_id_to_name,
+                              callback = std::move(callback)]() {
+        p4::v1::StreamMessageResponse pi_response;
+        while (session_->StreamChannelRead(pi_response)) {
+          sai::StreamMessageResponse pd_response;
+          if (!pdpi::PiStreamMessageResponseToPd(*ir_p4info, pi_response,
+                                                 &pd_response)
+                   .ok()) {
+            LOG(ERROR) << "Failed to convert PI stream message response to PD.";
+            return;
+          }
+          if (!pd_response.has_packet()) {
+            LOG(ERROR) << "PD response has no packet.";
+            return;
+          }
+          std::string port_id = pd_response.packet().metadata().ingress_port();
+          auto port_name = interface_port_id_to_name->find(port_id);
+          if (port_name == interface_port_id_to_name->end()) {
+            LOG(WARNING) << port_id << " not found.";
+            return;
+          }
+          callback(port_name->second,
+                   absl::BytesToHexString(pd_response.packet().payload()));
+        }
+      }) {}
+
+}  // namespace pins_test
diff --git a/lib/p4rt/packet_listener.h b/lib/p4rt/packet_listener.h
index fe17ca3ba..7aaf460e8 100644
--- a/lib/p4rt/packet_listener.h
+++ b/lib/p4rt/packet_listener.h
@@ -1,74 +1,53 @@
-// Copyright 2021 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.
-#ifndef PINS_LIB_P4RT_PACKET_LISTENER_H_
-#define PINS_LIB_P4RT_PACKET_LISTENER_H_
+#ifndef GOOGLE_LIB_P4RT_PACKET_LISTENER_H_
+#define GOOGLE_LIB_P4RT_PACKET_LISTENER_H_
 
-#include <functional>
+#include <memory>
 #include <string>
 #include <thread>  // NOLINT
 
 #include "absl/container/flat_hash_map.h"
-#include "absl/status/status.h"
+#include "absl/strings/escaping.h"
 #include "glog/logging.h"
-#include "lib/p4rt/p4rt_programming_context.h"
+#include "gutil/status_matchers.h"
 #include "p4/v1/p4runtime.pb.h"
-#include "p4_pdpi/ir.pb.h"
 #include "p4_pdpi/p4_runtime_session.h"
+#include "p4_pdpi/ir.pb.h"
 #include "p4_pdpi/pd.h"
-#include "sai_p4/instantiations/google/instantiations.h"
 #include "sai_p4/instantiations/google/sai_pd.pb.h"
-#include "thinkit/control_device.h"
+#include "thinkit/control_interface.h"
 #include "thinkit/packet_generation_finalizer.h"
 #include "thinkit/switch.h"
 
 namespace pins_test {
 
-// `PacketListener` will callback once a packet is received and stop listening
-// for packets when it goes out of scope.
+// PacketListener will callback once a packet is received and stop listening for
+// packets when it goes out of scope.
 class PacketListener : public thinkit::PacketGenerationFinalizer {
  public:
-  // Calls PacketCallback once a packet is received. `interface_port_id_to_name`
-  // needs to outlive this class. `on_finish` will get called when the listener
-  // is finished.
+  // Calls PacketCallback once a packet is received. Parameters passed in
+  // (besides the callback) cannot be null and need to outlive this class.
   PacketListener(pdpi::P4RuntimeSession* session,
-                 P4rtProgrammingContext context,
-                 sai::Instantiation instantiation,
+                 const pdpi::IrP4Info* ir_p4info,
                  const absl::flat_hash_map<std::string, std::string>*
                      interface_port_id_to_name,
-                 thinkit::PacketCallback callback,
-                 std::function<void()> on_finish);
+                 thinkit::PacketCallback callback);
 
-  ~PacketListener() {
-    absl::Status status = context_.Revert();
-    if (!status.ok()) {
-      LOG(WARNING) << "Failed to revert packet listening flows: " << status;
-    }
-    status = session_->Finish();
-    if (!status.ok()) {
-      LOG(WARNING) << "P4RuntimeSession finished abnormally: " << status;
+ ~PacketListener() ABSL_LOCKS_EXCLUDED(mutex_) {
+    {
+      absl::MutexLock lock(&mutex_);
+      time_to_exit_ = true;
     }
+    LOG(INFO) << "receive packet thread join.";
     receive_packet_thread_.join();
-    on_finish_();
-  }
+  } 
 
  private:
   pdpi::P4RuntimeSession* session_;
-  P4rtProgrammingContext context_;
+  bool time_to_exit_ ABSL_GUARDED_BY(mutex_);
+  absl::Mutex mutex_;
   std::thread receive_packet_thread_;
-  std::function<void()> on_finish_;
 };
 
 }  // namespace pins_test
 
-#endif  // PINS_LIB_P4RT_PACKET_LISTENERR_H_
+#endif  // GOOGLE_LIB_P4RT_PACKET_LISTENERR_H_
diff --git a/lib/validator/validator_lib.cc b/lib/validator/validator_lib.cc
index 76b69e682..266b5f1f2 100644
--- a/lib/validator/validator_lib.cc
+++ b/lib/validator/validator_lib.cc
@@ -116,7 +116,8 @@ absl::Status PortsUp(thinkit::Switch& thinkit_switch, absl::Duration timeout) {
   ASSIGN_OR_RETURN(std::unique_ptr<gnmi::gNMI::Stub> gnmi_stub,
                    thinkit_switch.CreateGnmiStub());
   return pins_test::CheckAllInterfaceOperStateOverGnmi(
-      *gnmi_stub, /*interface_oper_state=*/"UP", timeout);
+      *gnmi_stub, /*interface_oper_state=*/"UP",
+      /*skip_non_ethernet_interfaces=*/false, timeout);
 }
 
 absl::Status NoAlarms(thinkit::Switch& thinkit_switch, absl::Duration timeout) {
diff --git a/p4_pdpi/p4_runtime_session.cc b/p4_pdpi/p4_runtime_session.cc
index 56fbead59..f70d8cbf2 100644
--- a/p4_pdpi/p4_runtime_session.cc
+++ b/p4_pdpi/p4_runtime_session.cc
@@ -218,6 +218,26 @@ P4RuntimeSession::GetForwardingPipelineConfig(
   return response;
 }
 
+bool P4RuntimeSession::StreamChannelRead(
+    p4::v1::StreamMessageResponse& response,
+    std::optional<absl::Duration> timeout) {
+  absl::MutexLock lock(&stream_read_lock_);
+  auto cond = [&]() ABSL_SHARED_LOCKS_REQUIRED(stream_read_lock_) {
+    return !stream_messages_.empty() || !is_stream_up_;
+  };
+  if (timeout.has_value()) {
+    stream_read_lock_.AwaitWithTimeout(absl::Condition(&cond), *timeout);
+  } else {
+    stream_read_lock_.Await(absl::Condition(&cond));
+  }
+  if (!stream_messages_.empty()) {
+    response = stream_messages_.front();
+    stream_messages_.pop();
+    return true;
+  }
+  return false;
+}
+
 bool P4RuntimeSession::StreamChannelWrite(
     const p4::v1::StreamMessageRequest& request) {
   absl::MutexLock lock(&stream_write_lock_);
diff --git a/p4_pdpi/p4_runtime_session.h b/p4_pdpi/p4_runtime_session.h
index 008c98edf..ee0eebf68 100644
--- a/p4_pdpi/p4_runtime_session.h
+++ b/p4_pdpi/p4_runtime_session.h
@@ -181,7 +181,13 @@ class P4RuntimeSession {
   p4::v1::Uint128 ElectionId() const { return election_id_; }
   // Returns the role of this session.
   std::string Role() const { return role_; }
-
+  // Thread-safe wrapper around the stream channel's `Read` method.
+  // It blocks until the stream message queue is non-empty, the
+  // stream channel is closed, or (if specified) the `timeout` is expired .
+  ABSL_MUST_USE_RESULT bool StreamChannelRead(
+      p4::v1::StreamMessageResponse& response,
+      std::optional<absl::Duration> timeout = std::nullopt)
+      ABSL_LOCKS_EXCLUDED(stream_read_lock_);
   // Thread-safe wrapper around the stream channel's `Write` method.
   ABSL_MUST_USE_RESULT bool StreamChannelWrite(
       const p4::v1::StreamMessageRequest& request)