hunk.code_window:                             f"allow mount {remote}:{local} to take hold")
               except json.JSONDecodeError:
                   cli_logger.verbose(
                       "Unable to check if file_mounts specified in the YAML "
                       "differ from those on the running container.")
           self.initialized = True
hunk.labels: keep keep keep keep add keep
hunk.after_edit:         # Explicitly copy in ray bootstrap files.
           for mount in BOOTSTRAP_MOUNTS:
               if mount in file_mounts:
                   self.ssh_command_runner.run(
                       "docker cp {src} {container}:{dst}".format(
                           src=os.path.join(DOCKER_MOUNT_PREFIX, mount),
                           container=self.container_name,
                           dst=self._docker_expand_user(mount)))
  
hunk.file_path: python/ray/autoscaler/command_runner.py
hunk.type: add
file: // Licensed to the Apache Software Foundation (ASF) under one
  // or more contributor license agreements.  See the NOTICE file
  // distributed with this work for additional information
  // regarding copyright ownership.  The ASF licenses this file
  // to you 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.
  
  // PLASMA CLIENT: Client library for using the plasma store and manager
  
  #include "ray/object_manager/plasma/client.h"
  
  #include <cstring>
  
  #include <algorithm>
  #include <deque>
  #include <mutex>
  #include <tuple>
  #include <unordered_map>
  #include <unordered_set>
  #include <vector>
  
  #include <boost/asio.hpp>
  
  #include "arrow/buffer.h"
  
  #include "ray/object_manager/plasma/connection.h"
  #include "ray/object_manager/plasma/plasma.h"
  #include "ray/object_manager/plasma/protocol.h"
  #include "ray/object_manager/plasma/shared_memory.h"
  
  #ifdef PLASMA_CUDA
  #include "arrow/gpu/cuda_api.h"
  
  using arrow::cuda::CudaBuffer;
  using arrow::cuda::CudaBufferWriter;
  using arrow::cuda::CudaContext;
  using arrow::cuda::CudaDeviceManager;
  #endif
  
  namespace fb = plasma::flatbuf;
  
  namespace plasma {
  
  using fb::MessageType;
  using fb::PlasmaError;
  
  using arrow::MutableBuffer;
  
  // ----------------------------------------------------------------------
  // GPU support
  
  #ifdef PLASMA_CUDA
  
  namespace {
  
  struct GpuProcessHandle {
    /// Pointer to CUDA buffer that is backing this GPU object.
    std::shared_ptr<CudaBuffer> ptr;
    /// Number of client using this GPU object.
    int client_count;
  };
  
  // This is necessary as IPC handles can only be mapped once per process.
  // Thus if multiple clients in the same process get the same gpu object,
  // they need to access the same mapped CudaBuffer.
  std::unordered_map<ObjectID, GpuProcessHandle*> gpu_object_map;
  std::mutex gpu_mutex;
  
  // Return a new CudaBuffer pointing to the same data as the GpuProcessHandle,
  // but able to persist after the original IPC-backed buffer is closed
  // (ARROW-5924).
  std::shared_ptr<Buffer> MakeBufferFromGpuProcessHandle(GpuProcessHandle* handle) {
    return std::make_shared<CudaBuffer>(handle->ptr->address(), handle->ptr->size(),
                                        handle->ptr->context());
  }
  
  }  // namespace
  
  #endif
  
  // ----------------------------------------------------------------------
  // PlasmaBuffer
  
  /// A Buffer class that automatically releases the backing plasma object
  /// when it goes out of scope. This is returned by Get.
  class RAY_NO_EXPORT PlasmaBuffer : public Buffer {
   public:
    ~PlasmaBuffer();
  
    PlasmaBuffer(std::shared_ptr<PlasmaClient::Impl> client, const ObjectID& object_id,
                 const std::shared_ptr<Buffer>& buffer)
        : Buffer(buffer, 0, buffer->size()), client_(client), object_id_(object_id) {
      if (buffer->is_mutable()) {
        is_mutable_ = true;
      }
    }
  
   private:
    std::shared_ptr<PlasmaClient::Impl> client_;
    ObjectID object_id_;
  };
  
  /// A mutable Buffer class that keeps the backing data alive by keeping a
  /// PlasmaClient shared pointer. This is returned by Create. Release will
  /// be called in the associated Seal call.
  class RAY_NO_EXPORT PlasmaMutableBuffer : public MutableBuffer {
   public:
    PlasmaMutableBuffer(std::shared_ptr<PlasmaClient::Impl> client, uint8_t* mutable_data,
                        int64_t data_size)
        : MutableBuffer(mutable_data, data_size), client_(client) {}
  
   private:
    std::shared_ptr<PlasmaClient::Impl> client_;
  };
  
  // ----------------------------------------------------------------------
  // PlasmaClient::Impl
  
  struct ObjectInUseEntry {
    /// A count of the number of times this client has called PlasmaClient::Create
    /// or
    /// PlasmaClient::Get on this object ID minus the number of calls to
    /// PlasmaClient::Release.
    /// When this count reaches zero, we remove the entry from the ObjectsInUse
    /// and decrement a count in the relevant ClientMmapTableEntry.
    int count;
    /// Cached information to read the object.
    PlasmaObject object;
    /// A flag representing whether the object has been sealed.
    bool is_sealed;
  };
  
  class PlasmaClient::Impl : public std::enable_shared_from_this<PlasmaClient::Impl> {
   public:
    Impl();
    ~Impl();
  
    // PlasmaClient method implementations
  
    Status Connect(const std::string& store_socket_name,
                   const std::string& manager_socket_name, int release_delay = 0,
                   int num_retries = -1);
  
    Status SetClientOptions(const std::string& client_name, int64_t output_memory_quota);
  
    Status Create(const ObjectID& object_id, const ray::rpc::Address& owner_address,
                  int64_t data_size, const uint8_t* metadata, int64_t metadata_size,
                  std::shared_ptr<Buffer>* data, int device_num = 0,
                  bool evict_if_full = true);
  
    Status Get(const std::vector<ObjectID>& object_ids, int64_t timeout_ms,
               std::vector<ObjectBuffer>* object_buffers);
  
    Status Get(const ObjectID* object_ids, int64_t num_objects, int64_t timeout_ms,
               ObjectBuffer* object_buffers);
  
    Status Release(const ObjectID& object_id);
  
    Status Contains(const ObjectID& object_id, bool* has_object);
  
    Status Abort(const ObjectID& object_id);
  
    Status Seal(const ObjectID& object_id);
  
    Status Delete(const std::vector<ObjectID>& object_ids);
  
    Status Evict(int64_t num_bytes, int64_t& num_bytes_evicted);
  
    Status Refresh(const std::vector<ObjectID>& object_ids);
  
    Status Disconnect();
  
    std::string DebugString();
  
    bool IsInUse(const ObjectID& object_id);
  
    int64_t store_capacity() { return store_capacity_; }
  
   private:
    /// Check if store_fd has already been received from the store. If yes,
    /// return it. Otherwise, receive it from the store (see analogous logic
    /// in store.cc).
    ///
    /// \param store_fd File descriptor to fetch from the store.
    /// \return The pointer corresponding to store_fd.
    uint8_t* GetStoreFdAndMmap(MEMFD_TYPE store_fd, int64_t map_size);
  
    /// This is a helper method for marking an object as unused by this client.
    ///
    /// \param object_id The object ID we mark unused.
    /// \return The return status.
    Status MarkObjectUnused(const ObjectID& object_id);
  
    /// Common helper for Get() variants
    Status GetBuffers(const ObjectID* object_ids, int64_t num_objects, int64_t timeout_ms,
                      const std::function<std::shared_ptr<Buffer>(
                          const ObjectID&, const std::shared_ptr<Buffer>&)>& wrap_buffer,
                      ObjectBuffer* object_buffers);
  
    uint8_t* LookupMmappedFile(MEMFD_TYPE store_fd_val);
  
    void IncrementObjectCount(const ObjectID& object_id, PlasmaObject* object,
                              bool is_sealed);
  
    /// The boost::asio IO context for the client.
    boost::asio::io_service main_service_;
    /// The connection to the store service.
    std::shared_ptr<StoreConn> store_conn_;
    /// Table of dlmalloc buffer files that have been memory mapped so far. This
    /// is a hash table mapping a file descriptor to a struct containing the
    /// address of the corresponding memory-mapped file.
    std::unordered_map<MEMFD_TYPE, std::unique_ptr<ClientMmapTableEntry>> mmap_table_;
    /// A hash table of the object IDs that are currently being used by this
    /// client.
    std::unordered_map<ObjectID, std::unique_ptr<ObjectInUseEntry>> objects_in_use_;
    /// The amount of memory available to the Plasma store. The client needs this
    /// information to make sure that it does not delay in releasing so much
    /// memory that the store is unable to evict enough objects to free up space.
    int64_t store_capacity_;
    /// A hash set to record the ids that users want to delete but still in use.
    std::unordered_set<ObjectID> deletion_cache_;
    /// A mutex which protects this class.
    std::recursive_mutex client_mutex_;
  
  #ifdef PLASMA_CUDA
    /// Cuda Device Manager.
    arrow::cuda::CudaDeviceManager* manager_;
  #endif
  };
  
  PlasmaBuffer::~PlasmaBuffer() { RAY_UNUSED(client_->Release(object_id_)); }
  
  PlasmaClient::Impl::Impl() : store_capacity_(0) {
  #ifdef PLASMA_CUDA
    auto maybe_manager = CudaDeviceManager::Instance();
    DCHECK_OK(maybe_manager.status());
    manager_ = *maybe_manager;
  #endif
  }
  
  PlasmaClient::Impl::~Impl() {}
  
  // If the file descriptor fd has been mmapped in this client process before,
  // return the pointer that was returned by mmap, otherwise mmap it and store the
  // pointer in a hash table.
  uint8_t* PlasmaClient::Impl::GetStoreFdAndMmap(MEMFD_TYPE store_fd_val, int64_t map_size) {
    auto entry = mmap_table_.find(store_fd_val);
    if (entry != mmap_table_.end()) {
      return entry->second->pointer();
    } else {
      MEMFD_TYPE fd;
      RAY_CHECK_OK(store_conn_->RecvFd(&fd));
      mmap_table_[store_fd_val] =
          std::unique_ptr<ClientMmapTableEntry>(new ClientMmapTableEntry(fd, map_size));
      return mmap_table_[store_fd_val]->pointer();
    }
  }
  
  // Get a pointer to a file that we know has been memory mapped in this client
  // process before.
  uint8_t* PlasmaClient::Impl::LookupMmappedFile(MEMFD_TYPE store_fd_val) {
    auto entry = mmap_table_.find(store_fd_val);
    RAY_CHECK(entry != mmap_table_.end());
    return entry->second->pointer();
  }
  
  bool PlasmaClient::Impl::IsInUse(const ObjectID& object_id) {
    std::lock_guard<std::recursive_mutex> guard(client_mutex_);
  
    const auto elem = objects_in_use_.find(object_id);
    return (elem != objects_in_use_.end());
  }
  
  void PlasmaClient::Impl::IncrementObjectCount(const ObjectID& object_id,
                                                PlasmaObject* object, bool is_sealed) {
    // Increment the count of the object to track the fact that it is being used.
    // The corresponding decrement should happen in PlasmaClient::Release.
    auto elem = objects_in_use_.find(object_id);
    ObjectInUseEntry* object_entry;
    if (elem == objects_in_use_.end()) {
      // Add this object ID to the hash table of object IDs in use. The
      // corresponding call to free happens in PlasmaClient::Release.
      objects_in_use_[object_id] =
          std::unique_ptr<ObjectInUseEntry>(new ObjectInUseEntry());
      objects_in_use_[object_id]->object = *object;
      objects_in_use_[object_id]->count = 0;
      objects_in_use_[object_id]->is_sealed = is_sealed;
      object_entry = objects_in_use_[object_id].get();
    } else {
      object_entry = elem->second.get();
      RAY_CHECK(object_entry->count > 0);
    }
    // Increment the count of the number of instances of this object that are
    // being used by this client. The corresponding decrement should happen in
    // PlasmaClient::Release.
    object_entry->count += 1;
  }
  
  Status PlasmaClient::Impl::Create(const ObjectID& object_id, const ray::rpc::Address& owner_address,
                                    int64_t data_size, const uint8_t* metadata, int64_t metadata_size,
                                    std::shared_ptr<Buffer>* data, int device_num,
                                    bool evict_if_full) {
    std::lock_guard<std::recursive_mutex> guard(client_mutex_);
  
    RAY_LOG(DEBUG) << "called plasma_create on conn " << store_conn_ << " with size "
                     << data_size << " and metadata size " << metadata_size;
    RAY_RETURN_NOT_OK(SendCreateRequest(store_conn_, object_id, owner_address, evict_if_full, data_size,
                                    metadata_size, device_num));
    std::vector<uint8_t> buffer;
    RAY_RETURN_NOT_OK(PlasmaReceive(store_conn_, MessageType::PlasmaCreateReply, &buffer));
    ObjectID id;
    PlasmaObject object;
    MEMFD_TYPE store_fd;
    int64_t mmap_size;
    RAY_RETURN_NOT_OK(
        ReadCreateReply(buffer.data(), buffer.size(), &id, &object, &store_fd, &mmap_size));
    // If the CreateReply included an error, then the store will not send a file
    // descriptor.
    if (device_num == 0) {
      RAY_CHECK(object.data_size == data_size);
      RAY_CHECK(object.metadata_size == metadata_size);
      // The metadata should come right after the data.
      RAY_CHECK(object.metadata_offset == object.data_offset + data_size);
      *data = std::make_shared<PlasmaMutableBuffer>(
          shared_from_this(), GetStoreFdAndMmap(store_fd, mmap_size) + object.data_offset,
          data_size);
      // If plasma_create is being called from a transfer, then we will not copy the
      // metadata here. The metadata will be written along with the data streamed
      // from the transfer.
      if (metadata != NULL) {
        // Copy the metadata to the buffer.
        memcpy((*data)->mutable_data() + object.data_size, metadata, metadata_size);
      }
    } else {
  #ifdef PLASMA_CUDA
      std::shared_ptr<CudaContext> context;
      ARROW_ASSIGN_OR_RAISE(context, manager_->GetContext(device_num - 1));
      GpuProcessHandle* handle = new GpuProcessHandle();
      handle->client_count = 2;
      ARROW_ASSIGN_OR_RAISE(handle->ptr, context->OpenIpcBuffer(*object.ipc_handle));
      {
        std::lock_guard<std::mutex> lock(gpu_mutex);
        gpu_object_map[object_id] = handle;
      }
      if (metadata != NULL) {
        // Copy the metadata to the buffer.
        CudaBufferWriter writer(handle->ptr);
        RAY_RETURN_NOT_OK(writer.WriteAt(object.data_size, metadata, metadata_size));
      }
      *data = MakeBufferFromGpuProcessHandle(handle);
  #else
      RAY_LOG(FATAL) << "Arrow GPU library is not enabled.";
  #endif
    }
  
    // Increment the count of the number of instances of this object that this
    // client is using. A call to PlasmaClient::Release is required to decrement
    // this count. Cache the reference to the object.
    IncrementObjectCount(object_id, &object, false);
    // We increment the count a second time (and the corresponding decrement will
    // happen in a PlasmaClient::Release call in plasma_seal) so even if the
    // buffer returned by PlasmaClient::Create goes out of scope, the object does
    // not get released before the call to PlasmaClient::Seal happens.
    IncrementObjectCount(object_id, &object, false);
    return Status::OK();
  }
  
  Status PlasmaClient::Impl::GetBuffers(
      const ObjectID* object_ids, int64_t num_objects, int64_t timeout_ms,
      const std::function<std::shared_ptr<Buffer>(
          const ObjectID&, const std::shared_ptr<Buffer>&)>& wrap_buffer,
      ObjectBuffer* object_buffers) {
    // Fill out the info for the objects that are already in use locally.
    bool all_present = true;
    for (int64_t i = 0; i < num_objects; ++i) {
      auto object_entry = objects_in_use_.find(object_ids[i]);
      if (object_entry == objects_in_use_.end()) {
        // This object is not currently in use by this client, so we need to send
        // a request to the store.
        all_present = false;
      } else if (!object_entry->second->is_sealed) {
        // This client created the object but hasn't sealed it. If we call Get
        // with no timeout, we will deadlock, because this client won't be able to
        // call Seal.
        RAY_CHECK(timeout_ms != -1)
            << "Plasma client called get on an unsealed object that it created";
        RAY_LOG(WARNING)
            << "Attempting to get an object that this client created but hasn't sealed.";
        all_present = false;
      } else {
        PlasmaObject* object = &object_entry->second->object;
        std::shared_ptr<Buffer> physical_buf;
  
        if (object->device_num == 0) {
          uint8_t* data = LookupMmappedFile(object->store_fd);
          physical_buf = std::make_shared<Buffer>(
              data + object->data_offset, object->data_size + object->metadata_size);
        } else {
  #ifdef PLASMA_CUDA
          std::lock_guard<std::mutex> lock(gpu_mutex);
          auto iter = gpu_object_map.find(object_ids[i]);
          RAY_CHECK(iter != gpu_object_map.end());
          iter->second->client_count++;
          physical_buf = MakeBufferFromGpuProcessHandle(iter->second);
  #else
          RAY_LOG(FATAL) << "Arrow GPU library is not enabled.";
  #endif
        }
        physical_buf = wrap_buffer(object_ids[i], physical_buf);
        object_buffers[i].data = SliceBuffer(physical_buf, 0, object->data_size);
        object_buffers[i].metadata =
            SliceBuffer(physical_buf, object->data_size, object->metadata_size);
        object_buffers[i].device_num = object->device_num;
        // Increment the count of the number of instances of this object that this
        // client is using. Cache the reference to the object.
        IncrementObjectCount(object_ids[i], object, true);
      }
    }
  
    if (all_present) {
      return Status::OK();
    }
  
    // If we get here, then the objects aren't all currently in use by this
    // client, so we need to send a request to the plasma store.
    RAY_RETURN_NOT_OK(SendGetRequest(store_conn_, &object_ids[0], num_objects, timeout_ms));
    std::vector<uint8_t> buffer;
    RAY_RETURN_NOT_OK(PlasmaReceive(store_conn_, MessageType::PlasmaGetReply, &buffer));
    std::vector<ObjectID> received_object_ids(num_objects);
    std::vector<PlasmaObject> object_data(num_objects);
    PlasmaObject* object;
    std::vector<MEMFD_TYPE> store_fds;
    std::vector<int64_t> mmap_sizes;
    RAY_RETURN_NOT_OK(ReadGetReply(buffer.data(), buffer.size(), received_object_ids.data(),
                               object_data.data(), num_objects, store_fds, mmap_sizes));
  
    // We mmap all of the file descriptors here so that we can avoid look them up
    // in the subsequent loop based on just the store file descriptor and without
    // having to know the relevant file descriptor received from recv_fd.
    for (size_t i = 0; i < store_fds.size(); i++) {
      GetStoreFdAndMmap(store_fds[i], mmap_sizes[i]);
    }
  
    for (int64_t i = 0; i < num_objects; ++i) {
      RAY_DCHECK(received_object_ids[i] == object_ids[i]);
      object = &object_data[i];
      if (object_buffers[i].data) {
        // If the object was already in use by the client, then the store should
        // have returned it.
        RAY_DCHECK(object->data_size != -1);
        // We've already filled out the information for this object, so we can
        // just continue.
        continue;
      }
      // If we are here, the object was not currently in use, so we need to
      // process the reply from the object store.
      if (object->data_size != -1) {
        std::shared_ptr<Buffer> physical_buf;
        if (object->device_num == 0) {
          uint8_t* data = LookupMmappedFile(object->store_fd);
          physical_buf = std::make_shared<Buffer>(
              data + object->data_offset, object->data_size + object->metadata_size);
        } else {
  #ifdef PLASMA_CUDA
          std::lock_guard<std::mutex> lock(gpu_mutex);
          auto iter = gpu_object_map.find(object_ids[i]);
          if (iter == gpu_object_map.end()) {
            std::shared_ptr<CudaContext> context;
            ARROW_ASSIGN_OR_RAISE(context, manager_->GetContext(object->device_num - 1));
            GpuProcessHandle* obj_handle = new GpuProcessHandle();
            obj_handle->client_count = 1;
            ARROW_ASSIGN_OR_RAISE(obj_handle->ptr,
                                  context->OpenIpcBuffer(*object->ipc_handle));
            gpu_object_map[object_ids[i]] = obj_handle;
            physical_buf = MakeBufferFromGpuProcessHandle(obj_handle);
          } else {
            iter->second->client_count++;
            physical_buf = MakeBufferFromGpuProcessHandle(iter->second);
          }
  #else
          RAY_LOG(FATAL) << "Arrow GPU library is not enabled.";
  #endif
        }
        // Finish filling out the return values.
        physical_buf = wrap_buffer(object_ids[i], physical_buf);
        object_buffers[i].data = SliceBuffer(physical_buf, 0, object->data_size);
        object_buffers[i].metadata =
            SliceBuffer(physical_buf, object->data_size, object->metadata_size);
        object_buffers[i].device_num = object->device_num;
        // Increment the count of the number of instances of this object that this
        // client is using. Cache the reference to the object.
        IncrementObjectCount(received_object_ids[i], object, true);
      } else {
        // The object was not retrieved.  The caller can detect this condition
        // by checking the boolean value of the metadata/data buffers.
        RAY_DCHECK(!object_buffers[i].metadata);
        RAY_DCHECK(!object_buffers[i].data);
      }
    }
    return Status::OK();
  }
  
  Status PlasmaClient::Impl::Get(const std::vector<ObjectID>& object_ids,
                                 int64_t timeout_ms, std::vector<ObjectBuffer>* out) {
    std::lock_guard<std::recursive_mutex> guard(client_mutex_);
  
    const auto wrap_buffer = [=](const ObjectID& object_id,
                                 const std::shared_ptr<Buffer>& buffer) {
      return std::make_shared<PlasmaBuffer>(shared_from_this(), object_id, buffer);
    };
    const size_t num_objects = object_ids.size();
    *out = std::vector<ObjectBuffer>(num_objects);
    return GetBuffers(&object_ids[0], num_objects, timeout_ms, wrap_buffer, &(*out)[0]);
  }
  
  Status PlasmaClient::Impl::Get(const ObjectID* object_ids, int64_t num_objects,
                                 int64_t timeout_ms, ObjectBuffer* out) {
    std::lock_guard<std::recursive_mutex> guard(client_mutex_);
  
    const auto wrap_buffer = [](const ObjectID& object_id,
                                const std::shared_ptr<Buffer>& buffer) { return buffer; };
    return GetBuffers(object_ids, num_objects, timeout_ms, wrap_buffer, out);
  }
  
  Status PlasmaClient::Impl::MarkObjectUnused(const ObjectID& object_id) {
    auto object_entry = objects_in_use_.find(object_id);
    RAY_CHECK(object_entry != objects_in_use_.end());
    RAY_CHECK(object_entry->second->count == 0);
  
    // Remove the entry from the hash table of objects currently in use.
    objects_in_use_.erase(object_id);
    return Status::OK();
  }
  
  Status PlasmaClient::Impl::Release(const ObjectID& object_id) {
    std::lock_guard<std::recursive_mutex> guard(client_mutex_);
  
    // If the client is already disconnected, ignore release requests.
    if (!store_conn_) {
      return Status::OK();
    }
    auto object_entry = objects_in_use_.find(object_id);
    RAY_CHECK(object_entry != objects_in_use_.end());
  
  #ifdef PLASMA_CUDA
    if (object_entry->second->object.device_num != 0) {
      std::lock_guard<std::mutex> lock(gpu_mutex);
      auto iter = gpu_object_map.find(object_id);
      RAY_CHECK(iter != gpu_object_map.end());
      if (--iter->second->client_count == 0) {
        delete iter->second;
        gpu_object_map.erase(iter);
      }
    }
  #endif
  
    object_entry->second->count -= 1;
    RAY_CHECK(object_entry->second->count >= 0);
    // Check if the client is no longer using this object.
    if (object_entry->second->count == 0) {
      // Tell the store that the client no longer needs the object.
      RAY_RETURN_NOT_OK(MarkObjectUnused(object_id));
      RAY_RETURN_NOT_OK(SendReleaseRequest(store_conn_, object_id));
      auto iter = deletion_cache_.find(object_id);
      if (iter != deletion_cache_.end()) {
        deletion_cache_.erase(object_id);
        RAY_RETURN_NOT_OK(Delete({object_id}));
      }
    }
    return Status::OK();
  }
  
  // This method is used to query whether the plasma store contains an object.
  Status PlasmaClient::Impl::Contains(const ObjectID& object_id, bool* has_object) {
    std::lock_guard<std::recursive_mutex> guard(client_mutex_);
  
    // Check if we already have a reference to the object.
    if (objects_in_use_.count(object_id) > 0) {
      *has_object = 1;
    } else {
      // If we don't already have a reference to the object, check with the store
      // to see if we have the object.
      RAY_RETURN_NOT_OK(SendContainsRequest(store_conn_, object_id));
      std::vector<uint8_t> buffer;
      RAY_RETURN_NOT_OK(PlasmaReceive(store_conn_, MessageType::PlasmaContainsReply, &buffer));
      ObjectID object_id2;
      RAY_DCHECK(buffer.size() > 0);
      RAY_RETURN_NOT_OK(
          ReadContainsReply(buffer.data(), buffer.size(), &object_id2, has_object));
    }
    return Status::OK();
  }
  
  Status PlasmaClient::Impl::Seal(const ObjectID& object_id) {
    std::lock_guard<std::recursive_mutex> guard(client_mutex_);
  
    // Make sure this client has a reference to the object before sending the
    // request to Plasma.
    auto object_entry = objects_in_use_.find(object_id);
  
    if (object_entry == objects_in_use_.end()) {
      return Status::ObjectNotFound(
          "Seal() called on an object without a reference to it");
    }
    if (object_entry->second->is_sealed) {
      return Status::ObjectAlreadySealed(
          "Seal() called on an already sealed object");
    }
  
    object_entry->second->is_sealed = true;
    /// Send the seal request to Plasma.
    RAY_RETURN_NOT_OK(SendSealRequest(store_conn_, object_id));
    std::vector<uint8_t> buffer;
    RAY_RETURN_NOT_OK(PlasmaReceive(store_conn_, MessageType::PlasmaSealReply, &buffer));
    ObjectID sealed_id;
    RAY_RETURN_NOT_OK(ReadSealReply(buffer.data(), buffer.size(), &sealed_id));
    RAY_CHECK(sealed_id == object_id);
    // We call PlasmaClient::Release to decrement the number of instances of this
    // object
    // that are currently being used by this client. The corresponding increment
    // happened in plasma_create and was used to ensure that the object was not
    // released before the call to PlasmaClient::Seal.
    return Release(object_id);
  }
  
  Status PlasmaClient::Impl::Abort(const ObjectID& object_id) {
    std::lock_guard<std::recursive_mutex> guard(client_mutex_);
    auto object_entry = objects_in_use_.find(object_id);
    RAY_CHECK(object_entry != objects_in_use_.end())
        << "Plasma client called abort on an object without a reference to it";
    RAY_CHECK(!object_entry->second->is_sealed)
        << "Plasma client called abort on a sealed object";
  
    // Make sure that the Plasma client only has one reference to the object. If
    // it has more, then the client needs to release the buffer before calling
    // abort.
    if (object_entry->second->count > 1) {
      return Status::Invalid("Plasma client cannot have a reference to the buffer.");
    }
  
  #ifdef PLASMA_CUDA
    if (object_entry->second->object.device_num != 0) {
      std::lock_guard<std::mutex> lock(gpu_mutex);
      auto iter = gpu_object_map.find(object_id);
      RAY_CHECK(iter != gpu_object_map.end());
      RAY_CHECK(iter->second->client_count == 1);
      delete iter->second;
      gpu_object_map.erase(iter);
    }
  #endif
  
    // Send the abort request.
    RAY_RETURN_NOT_OK(SendAbortRequest(store_conn_, object_id));
    // Decrease the reference count to zero, then remove the object.
    object_entry->second->count--;
    RAY_RETURN_NOT_OK(MarkObjectUnused(object_id));
  
    std::vector<uint8_t> buffer;
    ObjectID id;
    RAY_RETURN_NOT_OK(PlasmaReceive(store_conn_, MessageType::PlasmaAbortReply, &buffer));
    return ReadAbortReply(buffer.data(), buffer.size(), &id);
  }
  
  Status PlasmaClient::Impl::Delete(const std::vector<ObjectID>& object_ids) {
    std::lock_guard<std::recursive_mutex> guard(client_mutex_);
  
    std::vector<ObjectID> not_in_use_ids;
    for (auto& object_id : object_ids) {
      // If the object is in used, skip it.
      if (objects_in_use_.count(object_id) == 0) {
        not_in_use_ids.push_back(object_id);
      } else {
        deletion_cache_.emplace(object_id);
      }
    }
    if (not_in_use_ids.size() > 0) {
      RAY_RETURN_NOT_OK(SendDeleteRequest(store_conn_, not_in_use_ids));
      std::vector<uint8_t> buffer;
      RAY_RETURN_NOT_OK(PlasmaReceive(store_conn_, MessageType::PlasmaDeleteReply, &buffer));
      RAY_DCHECK(buffer.size() > 0);
      std::vector<PlasmaError> error_codes;
      not_in_use_ids.clear();
      RAY_RETURN_NOT_OK(
          ReadDeleteReply(buffer.data(), buffer.size(), &not_in_use_ids, &error_codes));
    }
    return Status::OK();
  }
  
  Status PlasmaClient::Impl::Evict(int64_t num_bytes, int64_t& num_bytes_evicted) {
    std::lock_guard<std::recursive_mutex> guard(client_mutex_);
  
    // Send a request to the store to evict objects.
    RAY_RETURN_NOT_OK(SendEvictRequest(store_conn_, num_bytes));
    // Wait for a response with the number of bytes actually evicted.
    std::vector<uint8_t> buffer;
    RAY_RETURN_NOT_OK(PlasmaReceive(store_conn_, MessageType::PlasmaEvictReply, &buffer));
    return ReadEvictReply(buffer.data(), buffer.size(), num_bytes_evicted);
  }
  
  Status PlasmaClient::Impl::Refresh(const std::vector<ObjectID>& object_ids) {
    std::lock_guard<std::recursive_mutex> guard(client_mutex_);
  
    RAY_RETURN_NOT_OK(SendRefreshLRURequest(store_conn_, object_ids));
    std::vector<uint8_t> buffer;
    RAY_RETURN_NOT_OK(PlasmaReceive(store_conn_, MessageType::PlasmaRefreshLRUReply, &buffer));
    return ReadRefreshLRUReply(buffer.data(), buffer.size());
  }
  
  Status PlasmaClient::Impl::Connect(const std::string& store_socket_name,
                                     const std::string& manager_socket_name,
                                     int release_delay, int num_retries) {
    std::lock_guard<std::recursive_mutex> guard(client_mutex_);
  
    /// The local stream socket that connects to store.
    ray::local_stream_socket socket(main_service_);
    RAY_RETURN_NOT_OK(ray::ConnectSocketRetry(socket, store_socket_name));
    store_conn_.reset(new StoreConn(std::move(socket)));
    // Send a ConnectRequest to the store to get its memory capacity.
    RAY_RETURN_NOT_OK(SendConnectRequest(store_conn_));
    std::vector<uint8_t> buffer;
    RAY_RETURN_NOT_OK(PlasmaReceive(store_conn_, MessageType::PlasmaConnectReply, &buffer));
    RAY_RETURN_NOT_OK(ReadConnectReply(buffer.data(), buffer.size(), &store_capacity_));
    return Status::OK();
  }
  
  Status PlasmaClient::Impl::SetClientOptions(const std::string& client_name,
                                              int64_t output_memory_quota) {
    std::lock_guard<std::recursive_mutex> guard(client_mutex_);
    RAY_RETURN_NOT_OK(SendSetOptionsRequest(store_conn_, client_name, output_memory_quota));
    std::vector<uint8_t> buffer;
    RAY_RETURN_NOT_OK(PlasmaReceive(store_conn_, MessageType::PlasmaSetOptionsReply, &buffer));
    return ReadSetOptionsReply(buffer.data(), buffer.size());
  }
  
  Status PlasmaClient::Impl::Disconnect() {
    std::lock_guard<std::recursive_mutex> guard(client_mutex_);
  
    // NOTE: We purposefully do not finish sending release calls for objects in
    // use, so that we don't duplicate PlasmaClient::Release calls (when handling
    // a SIGTERM, for example).
  
    // Close the connections to Plasma. The Plasma store will release the objects
    // that were in use by us when handling the SIGPIPE.
    store_conn_.reset();
    return Status::OK();
  }
  
  std::string PlasmaClient::Impl::DebugString() {
    std::lock_guard<std::recursive_mutex> guard(client_mutex_);
    if (!SendGetDebugStringRequest(store_conn_).ok()) {
      return "error sending request";
    }
    std::vector<uint8_t> buffer;
    if (!PlasmaReceive(store_conn_, MessageType::PlasmaGetDebugStringReply, &buffer).ok()) {
      return "error receiving reply";
    }
    std::string debug_string;
    if (!ReadGetDebugStringReply(buffer.data(), buffer.size(), &debug_string).ok()) {
      return "error parsing reply";
    }
    return debug_string;
  }
  
  // ----------------------------------------------------------------------
  // PlasmaClient
  
  PlasmaClient::PlasmaClient() : impl_(std::make_shared<PlasmaClient::Impl>()) {}
  
  PlasmaClient::~PlasmaClient() {}
  
  Status PlasmaClient::Connect(const std::string& store_socket_name,
                               const std::string& manager_socket_name, int release_delay,
                               int num_retries) {
    return impl_->Connect(store_socket_name, manager_socket_name, release_delay,
                          num_retries);
  }
  
  Status PlasmaClient::SetClientOptions(const std::string& client_name,
                                        int64_t output_memory_quota) {
    return impl_->SetClientOptions(client_name, output_memory_quota);
  }
  
  Status PlasmaClient::Create(const ObjectID& object_id, const ray::rpc::Address& owner_address, int64_t data_size,
                              const uint8_t* metadata, int64_t metadata_size,
                              std::shared_ptr<Buffer>* data, int device_num,
                              bool evict_if_full) {
    return impl_->Create(object_id, owner_address, data_size, metadata, metadata_size, data, device_num,
                         evict_if_full);
  }
  
  Status PlasmaClient::Get(const std::vector<ObjectID>& object_ids, int64_t timeout_ms,
                           std::vector<ObjectBuffer>* object_buffers) {
    return impl_->Get(object_ids, timeout_ms, object_buffers);
  }
  
  Status PlasmaClient::Get(const ObjectID* object_ids, int64_t num_objects,
                           int64_t timeout_ms, ObjectBuffer* object_buffers) {
    return impl_->Get(object_ids, num_objects, timeout_ms, object_buffers);
  }
  
  Status PlasmaClient::Release(const ObjectID& object_id) {
    return impl_->Release(object_id);
  }
  
  Status PlasmaClient::Contains(const ObjectID& object_id, bool* has_object) {
    return impl_->Contains(object_id, has_object);
  }
  
  Status PlasmaClient::Abort(const ObjectID& object_id) { return impl_->Abort(object_id); }
  
  Status PlasmaClient::Seal(const ObjectID& object_id) { return impl_->Seal(object_id); }
  
  Status PlasmaClient::Delete(const ObjectID& object_id) {
    return impl_->Delete(std::vector<ObjectID>{object_id});
  }
  
  Status PlasmaClient::Delete(const std::vector<ObjectID>& object_ids) {
    return impl_->Delete(object_ids);
  }
  
  Status PlasmaClient::Evict(int64_t num_bytes, int64_t& num_bytes_evicted) {
    return impl_->Evict(num_bytes, num_bytes_evicted);
  }
  
  Status PlasmaClient::Refresh(const std::vector<ObjectID>& object_ids) {
    return impl_->Refresh(object_ids);
  }
  
  Status PlasmaClient::Disconnect() { return impl_->Disconnect(); }
  
  std::string PlasmaClient::DebugString() { return impl_->DebugString(); }
  
  bool PlasmaClient::IsInUse(const ObjectID& object_id) {
    return impl_->IsInUse(object_id);
  }
  
  int64_t PlasmaClient::store_capacity() { return impl_->store_capacity(); }
  
  }  // namespace plasma
  