[flowstate_ros_bridge] Make page_size and other params configurable in ExecutiveBridge (#72)

* Make page_size and other params configurable

Signed-off-by: Yadunund <[email protected]>

* Change to integer types

Signed-off-by: Luca Della Vedova <[email protected]>

* Add support for multiple pages, remove page param

Signed-off-by: Luca Della Vedova <[email protected]>

---------

Signed-off-by: Yadunund <[email protected]>
Signed-off-by: Luca Della Vedova <[email protected]>
Co-authored-by: Luca Della Vedova <[email protected]>

Author: Yadunund

flowstate_ros_bridge/flowstate_ros_bridge.proto

@@ -12,4 +12,6 @@ message FlowstateRosBridgeConfig {
   string flowstate_zenoh_router_address = 7;
   optional string external_zenoh_router_address = 8;
   repeated string bridge_plugins = 9;
+  int64 executive_deadline_seconds = 10;
+  int64 executive_update_rate_millis = 11;
 }

flowstate_ros_bridge/flowstate_ros_bridge_default_config.pbtxt

@@ -14,4 +14,6 @@
     "flowstate_ros_bridge::WorldBridge",
     "flowstate_ros_bridge::ExecutiveBridge"
   ],
+  executive_deadline_seconds: 5
+  executive_update_rate_millis: 1000
 }

flowstate_ros_bridge/src/bridges/world_bridge.cpp

@@ -267,7 +267,7 @@ absl::Status WorldBridge::Data::SendObjectVisualizationMessages(
   }
   LOG(INFO) << "Total gltf size: " << total_gltf_size << " bytes";
 
-  if (!array_msg.markers.empty()){
+  if (!array_msg.markers.empty()) {
     workcell_markers_pub_->publish(array_msg);
   }
 

flowstate_ros_bridge/src/executive.cpp

@@ -54,6 +54,8 @@ using GetSkillsResponse = intrinsic_proto::skills::GetSkillsResponse;
 
 static constexpr std::string PROCESS_PARAMS_KEY = "parameters";
 static constexpr std::string RESOURCE_PARAMS_KEY = "resources";
+static constexpr std::size_t kListBehaviorTreesPageSize = 50;
+static constexpr std::size_t kListOperationsPageSize = 100;
 
 ///=============================================================================
 Executive::Executive(const std::string& executive_service_address,
@@ -261,18 +263,26 @@ void Executive::clear_and_delete_operations() {
   // First list active operations.
   auto client_context = make_client_context(this->deadline_seconds_);
   google::longrunning::ListOperationsRequest list_request;
-  google::longrunning::ListOperationsResponse list_response;
-  auto status = executive_stub_->ListOperations(
-      client_context.get(), std::move(list_request), &list_response);
-  if (status.ok()) {
+  list_request.set_page_size(kListOperationsPageSize);
+  do {
+    google::longrunning::ListOperationsResponse list_response;
+    const auto status = executive_stub_->ListOperations(
+        client_context.get(), list_request, &list_response);
+
+    if (!status.ok()) {
+      LOG(INFO) << "Error while listing operations.";
+      return;
+    }
     for (const auto& operation : list_response.operations()) {
       // TODO(Yadunund): If we want this bridge to not override
       // all the operations started by users from the frontend, we might
       // want to attach active operations to this bridge and monitor its
       // state until completion before clearing.
       delete_operation(operation.name(), executive_stub_, deadline_seconds_);
     }
+    list_request.set_page_token(list_response.next_page_token());
   }
+  while (!list_request.page_token().empty());
 }
 
 ///=============================================================================
@@ -307,15 +317,20 @@ auto Executive::behavior_trees() const
   client_context->set_deadline(std::chrono::system_clock::now() +
                                std::chrono::seconds(deadline_seconds_));
   intrinsic_proto::solution::v1::ListBehaviorTreesRequest request;
-  intrinsic_proto::solution::v1::ListBehaviorTreesResponse response;
-  INTR_RETURN_IF_ERROR(
-      intrinsic::ToAbslStatus(solution_stub_->ListBehaviorTrees(
-          client_context.get(), request, &response)));
-
+  request.set_page_size(kListBehaviorTreesPageSize);
   std::vector<BehaviorTree> bts = {};
-  for (auto& bt : *response.mutable_behavior_trees()) {
-    bts.emplace_back(std::move(bt));
+  do {
+    intrinsic_proto::solution::v1::ListBehaviorTreesResponse response;
+    INTR_RETURN_IF_ERROR(
+        intrinsic::ToAbslStatus(solution_stub_->ListBehaviorTrees(
+            client_context.get(), request, &response)));
+
+    for (auto& bt : *response.mutable_behavior_trees()) {
+      bts.emplace_back(std::move(bt));
+    }
+    request.set_page_token(response.next_page_token());
   }
+  while (!request.page_token().empty());
   return bts;
 }
 
@@ -337,12 +352,10 @@ auto Executive::ProcessHandle::make(
     std::shared_ptr<ExecutiveService::Stub> executive_stub,
     std::size_t deadline_seconds, google::longrunning::Operation operation,
     ProcessFeedbackCallback feedback_cb, ProcessCompletedCallback completed_cb,
-    std::size_t update_interval_millis)
-    -> ProcessHandlePtr {
-  auto handle = std::shared_ptr<ProcessHandle>(
-      new ProcessHandle(std::move(executive_stub), std::move(deadline_seconds),
-                        std::move(operation), std::move(feedback_cb),
-                        std::move(completed_cb)));
+    std::size_t update_interval_millis) -> ProcessHandlePtr {
+  auto handle = std::shared_ptr<ProcessHandle>(new ProcessHandle(
+      std::move(executive_stub), std::move(deadline_seconds),
+      std::move(operation), std::move(feedback_cb), std::move(completed_cb)));
 
   // Spawn a thread to monitor lifecycle of the process.
   handle->update_thread_ =
@@ -370,9 +383,8 @@ auto Executive::ProcessHandle::make(
                 handle->feedback_cb_(false, absl::UnavailableError(ss.str()));
               }
 
-              handle->feedback_cb_(
-                  handle->current_operation_.done(),
-                  intrinsic::ToAbslStatus(status));
+              handle->feedback_cb_(handle->current_operation_.done(),
+                                   intrinsic::ToAbslStatus(status));
 
               if (handle->current_operation_.done()) {
                 // The operation may be done because it succeeded or errored
@@ -487,10 +499,10 @@ absl::StatusOr<Executive::ProcessHandlePtr> Executive::start(
       start_client_context.get(), std::move(start_request),
       &current_operation)));
 
-  current_process_ = ProcessHandle::make(
-      executive_stub_, deadline_seconds_, std::move(current_operation),
-      std::move(feedback_cb), std::move(completed_cb),
-      this->update_rate_millis_);
+  current_process_ =
+      ProcessHandle::make(executive_stub_, deadline_seconds_,
+                          std::move(current_operation), std::move(feedback_cb),
+                          std::move(completed_cb), this->update_rate_millis_);
 
   return current_process_;
 }

flowstate_ros_bridge/src/flowstate_ros_bridge.cpp

@@ -40,6 +40,10 @@ constexpr const char* kWorldAddressParamName = "world_service_address";
 constexpr const char* kGeometryAddressParamName = "geometry_service_address";
 constexpr const char* kServiceTunnelParamName = "service_tunnel";
 constexpr const char* kBridgePluginParamName = "bridge_plugins";
+constexpr const char* kExecutiveDeadlineParamName =
+    "executive_deadline_seconds";
+constexpr const char* kExecutiveUpdateRateMillisParamName =
+    "executive_update_rate_millis";
 
 ///=============================================================================
 FlowstateROSBridge::FlowstateROSBridge(const rclcpp::NodeOptions& options)
@@ -56,6 +60,9 @@ FlowstateROSBridge::FlowstateROSBridge(const rclcpp::NodeOptions& options)
       service_tunnel.empty()
           ? "executive.app-intrinsic-app-chart.svc.cluster.local:8080"
           : service_tunnel);
+  this->declare_parameter(kExecutiveDeadlineParamName, 5);
+  this->declare_parameter(kExecutiveUpdateRateMillisParamName, 1000);
+
   this->declare_parameter(
       kSkillAddressParamName,
       service_tunnel.empty()
@@ -94,7 +101,10 @@ FlowstateROSBridge::FlowstateROSBridge(const rclcpp::NodeOptions& options)
   this->executive_ = std::make_shared<Executive>(
       this->get_parameter(kExecutiveAddressParamName).get_value<std::string>(),
       this->get_parameter(kSkillAddressParamName).get_value<std::string>(),
-      this->get_parameter(kSolutionAddressParamName).get_value<std::string>());
+      this->get_parameter(kSolutionAddressParamName).get_value<std::string>(),
+      this->get_parameter(kExecutiveDeadlineParamName).get_value<std::size_t>(),
+      this->get_parameter(kExecutiveUpdateRateMillisParamName)
+          .get_value<std::size_t>());
 
   // Initialize the world client.
   this->world_ = std::make_shared<World>(

flowstate_ros_bridge/src/flowstate_ros_bridge_main.cpp

@@ -59,6 +59,10 @@ int main(int argc, char* argv[]) {
   // Get parameters from config
   params.emplace_back("executive_service_address",
                       ros_config.executive_service_address());
+  params.emplace_back("executive_deadline_seconds",
+                      ros_config.executive_deadline_seconds());
+  params.emplace_back("executive_update_rate_millis",
+                      ros_config.executive_update_rate_millis());
   params.emplace_back("skill_registry_address",
                       ros_config.skill_registry_address());
   params.emplace_back("solution_service_address",