Get the value or values in the object store associated with the IDs.

        Return the values from the local object store for object_ids. This will
        block until all the values for object_ids have been written to the
        local object store.

        Args:
            object_ids (List[object_id.ObjectID]): A list of the object IDs
                whose values should be retrieved.

def get_object(self, object_ids):
        """Get the value or values in the object store associated with the IDs.

        Return the values from the local object store for object_ids. This will
        block until all the values for object_ids have been written to the
        local object store.

        Args:
            object_ids (List[object_id.ObjectID]): A list of the object IDs
                whose values should be retrieved.
        """
        # Make sure that the values are object IDs.
        for object_id in object_ids:
            if not isinstance(object_id, ObjectID):
                raise TypeError(
                    "Attempting to call `get` on the value {}, "
                    "which is not an ray.ObjectID.".format(object_id))
        # Do an initial fetch for remote objects. We divide the fetch into
        # smaller fetches so as to not block the manager for a prolonged period
        # of time in a single call.
        plain_object_ids = [
            plasma.ObjectID(object_id.binary()) for object_id in object_ids
        ]
        for i in range(0, len(object_ids),
                       ray._config.worker_fetch_request_size()):
            self.raylet_client.fetch_or_reconstruct(
                object_ids[i:(i + ray._config.worker_fetch_request_size())],
                True)

        # Get the objects. We initially try to get the objects immediately.
        final_results = self.retrieve_and_deserialize(plain_object_ids, 0)
        # Construct a dictionary mapping object IDs that we haven't gotten yet
        # to their original index in the object_ids argument.
        unready_ids = {
            plain_object_ids[i].binary(): i
            for (i, val) in enumerate(final_results)
            if val is plasma.ObjectNotAvailable
        }

        if len(unready_ids) > 0:
            # Try reconstructing any objects we haven't gotten yet. Try to
            # get them until at least get_timeout_milliseconds
            # milliseconds passes, then repeat.
            while len(unready_ids) > 0:
                object_ids_to_fetch = [
                    plasma.ObjectID(unready_id)
                    for unready_id in unready_ids.keys()
                ]
                ray_object_ids_to_fetch = [
                    ObjectID(unready_id) for unready_id in unready_ids.keys()
                ]
                fetch_request_size = ray._config.worker_fetch_request_size()
                for i in range(0, len(object_ids_to_fetch),
                               fetch_request_size):
                    self.raylet_client.fetch_or_reconstruct(
                        ray_object_ids_to_fetch[i:(i + fetch_request_size)],
                        False,
                        self.current_task_id,
                    )
                results = self.retrieve_and_deserialize(
                    object_ids_to_fetch,
                    max([
                        ray._config.get_timeout_milliseconds(),
                        int(0.01 * len(unready_ids)),
                    ]),
                )
                # Remove any entries for objects we received during this
                # iteration so we don't retrieve the same object twice.
                for i, val in enumerate(results):
                    if val is not plasma.ObjectNotAvailable:
                        object_id = object_ids_to_fetch[i].binary()
                        index = unready_ids[object_id]
                        final_results[index] = val
                        unready_ids.pop(object_id)

            # If there were objects that we weren't able to get locally,
            # let the raylet know that we're now unblocked.
            self.raylet_client.notify_unblocked(self.current_task_id)

        assert len(final_results) == len(object_ids)
        return final_results