Speed up TaskSpecification copy (#5709)

build.sh

@@ -100,9 +100,11 @@ pushd "$BUILD_DIR"
 # The following line installs pyarrow from S3, these wheels have been
 # generated from https://github.com/ray-project/arrow-build from
 # the commit listed in the command.
-$PYTHON_EXECUTABLE -m pip install -q \
-    --target="$ROOT_DIR/python/ray/pyarrow_files" pyarrow==0.14.0.RAY \
-    --find-links https://s3-us-west-2.amazonaws.com/arrow-wheels/516e15028091b5e287200b5df77d77f72d9a6c9a/index.html
+if [ -z "$SKIP_PYARROW_INSTALL" ]; then
+    $PYTHON_EXECUTABLE -m pip install -q \
+        --target="$ROOT_DIR/python/ray/pyarrow_files" pyarrow==0.14.0.RAY \
+        --find-links https://s3-us-west-2.amazonaws.com/arrow-wheels/516e15028091b5e287200b5df77d77f72d9a6c9a/index.html
+fi
 export PYTHON_BIN_PATH="$PYTHON_EXECUTABLE"
 
 if [ "$RAY_BUILD_JAVA" == "YES" ]; then

src/ray/common/task/task_spec.cc

@@ -12,8 +12,8 @@ void TaskSpecification::ComputeResources() {
   if (required_placement_resources.empty()) {
     required_placement_resources = required_resources;
   }
-  required_resources_ = ResourceSet(required_resources);
-  required_placement_resources_ = ResourceSet(required_placement_resources);
+  required_resources_.reset(new ResourceSet(required_resources));
+  required_placement_resources_.reset(new ResourceSet(required_placement_resources));
 }
 
 // Task specification getter methods.
@@ -69,12 +69,12 @@ size_t TaskSpecification::ArgMetadataSize(size_t arg_index) const {
   return message_->args(arg_index).metadata().size();
 }
 
-const ResourceSet TaskSpecification::GetRequiredResources() const {
-  return required_resources_;
+const ResourceSet &TaskSpecification::GetRequiredResources() const {
+  return *required_resources_;
 }
 
-const ResourceSet TaskSpecification::GetRequiredPlacementResources() const {
-  return required_placement_resources_;
+const ResourceSet &TaskSpecification::GetRequiredPlacementResources() const {
+  return *required_placement_resources_;
 }
 
 bool TaskSpecification::IsDriverTask() const {

src/ray/common/task/task_spec.h

@@ -83,7 +83,7 @@ class TaskSpecification : public MessageWrapper<rpc::TaskSpec> {
   ///
   /// \return The resources that will be acquired during the execution of this
   /// task.
-  const ResourceSet GetRequiredResources() const;
+  const ResourceSet &GetRequiredResources() const;
 
   /// Return the resources that are required for a task to be placed on a node.
   /// This will typically be the same as the resources acquired during execution
@@ -94,7 +94,7 @@ class TaskSpecification : public MessageWrapper<rpc::TaskSpec> {
   /// so the placement of the actor should take this into account.
   ///
   /// \return The resources that are required to place a task on a node.
-  const ResourceSet GetRequiredPlacementResources() const;
+  const ResourceSet &GetRequiredPlacementResources() const;
 
   bool IsDriverTask() const;
 
@@ -140,9 +140,11 @@ class TaskSpecification : public MessageWrapper<rpc::TaskSpec> {
  private:
   void ComputeResources();
   /// Field storing required resources. Initalized in constructor.
-  ResourceSet required_resources_;
+  /// TODO(ekl) consider optimizing the representation of ResourceSet for fast copies
+  /// instead of keeping shared ptrs here.
+  std::shared_ptr<ResourceSet> required_resources_;
   /// Field storing required placement resources. Initalized in constructor.
-  ResourceSet required_placement_resources_;
+  std::shared_ptr<ResourceSet> required_placement_resources_;
 };
 
 }  // namespace ray