address comments

BUILD.bazel

@@ -295,7 +295,6 @@ cc_library(
         ],
     }),
     hdrs = [
-        "src/ray/object_manager/plasma/allocator.h",
         "src/ray/object_manager/plasma/client.h",
         "src/ray/object_manager/plasma/common.h",
         "src/ray/object_manager/plasma/compat.h",

src/ray/object_manager/plasma/eviction_policy.cc

@@ -90,7 +90,7 @@ int64_t LRUCache::ChooseObjectsToEvict(int64_t num_bytes_required,
   return bytes_evicted;
 }
 
-EvictionPolicy::EvictionPolicy(PlasmaStoreInfo *store_info, IAllocator &allocator)
+EvictionPolicy::EvictionPolicy(PlasmaStoreInfo *store_info, const IAllocator &allocator)
     : pinned_memory_bytes_(0),
       store_info_(store_info),
       allocator_(allocator),

src/ray/object_manager/plasma/eviction_policy.h

@@ -98,7 +98,7 @@ class EvictionPolicy {
   /// \param store_info Information about the Plasma store that is exposed
   ///        to the eviction policy.
   /// \param allocator Memory allocator.
-  explicit EvictionPolicy(PlasmaStoreInfo *store_info, IAllocator &allocator);
+  explicit EvictionPolicy(PlasmaStoreInfo *store_info, const IAllocator &allocator);
 
   /// Destroy an eviction policy.
   virtual ~EvictionPolicy() {}
@@ -175,7 +175,7 @@ class EvictionPolicy {
   /// Pointer to the plasma store info.
   PlasmaStoreInfo *store_info_;
 
-  IAllocator &allocator_;
+  const IAllocator &allocator_;
 
   /// Datastructure for the LRU cache.
   LRUCache cache_;

src/ray/object_manager/plasma/malloc.cc

@@ -32,7 +32,8 @@ static ptrdiff_t pointer_distance(void const *pfrom, void const *pto) {
   return (unsigned char const *)pto - (unsigned char const *)pfrom;
 }
 
-bool GetMallocMapinfo(void *addr, MEMFD_TYPE *fd, int64_t *map_size, ptrdiff_t *offset) {
+bool GetMallocMapinfo(const void *const addr, MEMFD_TYPE *fd, int64_t *map_size,
+                      ptrdiff_t *offset) {
   // TODO(rshin): Implement a more efficient search through mmap_records.
   for (const auto &entry : mmap_records) {
     if (addr >= entry.first && addr < pointer_advance(entry.first, entry.second.size)) {

src/ray/object_manager/plasma/malloc.h

@@ -41,6 +41,11 @@ struct MmapRecord {
 /// and size.
 extern std::unordered_map<void *, MmapRecord> mmap_records;
 
-/// private function, only used by PlasmaAllocator
-bool GetMallocMapinfo(void *addr, MEMFD_TYPE *fd, int64_t *map_length, ptrdiff_t *offset);
+/// private function, only used by PlasmaAllocator to look up Mmap information
+/// given an address allocated by dlmalloc.
+///
+/// \return true if look up succeed. false means the address is not allocated
+/// by dlmalloc.
+bool GetMallocMapinfo(const void *const addr, MEMFD_TYPE *fd, int64_t *map_length,
+                      ptrdiff_t *offset);
 }  // namespace plasma

src/ray/object_manager/plasma/plasma_allocator.cc

@@ -111,7 +111,7 @@ absl::optional<Allocation> PlasmaAllocator::FallbackAllocate(size_t bytes) {
 }
 
 void PlasmaAllocator::Free(const Allocation &allocation) {
-  RAY_CHECK(allocation.address != nullptr);
+  RAY_CHECK(allocation.address != nullptr) << "Cannot free the nullptr";
   RAY_LOG(DEBUG) << "deallocating " << allocation.size << " at " << allocation.address;
   dlfree(allocation.address);
   allocated_ -= allocation.size;

src/ray/object_manager/plasma/plasma_allocator.h

@@ -24,7 +24,8 @@
 namespace plasma {
 
 // PlasmaAllocator that allocates memory from mmaped file to
-// enable memory sharing between processes.
+// enable memory sharing between processes. It's not thread
+// safe and can only be created once per process.
 //
 // PlasmaAllocator is optimized for linux. On linux,
 // the Allocate call allocates memory from a pre-mmap file
@@ -42,13 +43,18 @@ class PlasmaAllocator : public IAllocator {
 
   /// On linux, it allocates memory from a pre-mmaped file from /dev/shm.
   /// On other system, it allocates memory from a pre-mmaped file on disk.
-  /// return null if running out of space.
+  /// NOTE: due to fragmentation, there is a possibility that the
+  /// allocator has the capacity but fails to fulfill the allocation
+  /// request.
   ///
   /// \param bytes Number of bytes.
   /// \return allocated memory. returns empty if not enough space.
   absl::optional<Allocation> Allocate(size_t bytes) override;
 
-  /// Fallback allocate memory from disk mmaped file.
+  /// Fallback allocate memory from disk mmaped file. This is useful
+  /// when we running out of memory but still want to allocate memory
+  /// with sub-optimal peformance.
+  ///
   /// On linux with fallocate support, it returns null if running out of
   /// space; On linux without fallocate it raises SIGBUS interrupt.
   /// TODO(scv119): On other system the behavior is undefined.
@@ -58,7 +64,7 @@ class PlasmaAllocator : public IAllocator {
   absl::optional<Allocation> FallbackAllocate(size_t bytes) override;
 
   /// Frees the memory space pointed to by mem, which must have been returned by
-  /// a previous call to Allocate/FallbackAllocate or it yield undefined behavior.
+  /// a previous call to Allocate/FallbackAllocate or it yields undefined behavior.
   ///
   /// \param allocation allocation to free.
   void Free(const Allocation &allocation) override;

src/ray/object_manager/plasma/store.cc

@@ -358,8 +358,10 @@ PlasmaError PlasmaStore::CreateObject(const ObjectID &object_id,
   }
 
   RAY_LOG(DEBUG) << "create object " << object_id << " succeeded";
-  auto ptr = std::make_unique<ObjectTableEntry>(std::move(allocation.value()));
-  entry = store_info_.objects.emplace(object_id, std::move(ptr)).first->second.get();
+  auto object_table_entry =
+      std::make_unique<ObjectTableEntry>(std::move(allocation.value()));
+  entry = store_info_.objects.emplace(object_id, std::move(object_table_entry))
+              .first->second.get();
   entry->data_size = data_size;
   entry->metadata_size = metadata_size;
   entry->state = ObjectState::PLASMA_CREATED;

src/ray/object_manager/plasma/store_runner.cc

@@ -101,7 +101,9 @@ void PlasmaStoreRunner::Start(ray::SpillObjectsCallback spill_objects_callback,
     // should be part of PlasmaAllocator contruction.
     auto allocation = PlasmaAllocator::GetInstance().Allocate(
         PlasmaAllocator::GetInstance().GetFootprintLimit() - 256 * sizeof(size_t));
-    RAY_CHECK(allocation.has_value());
+    RAY_CHECK(allocation.has_value())
+        << "Plasma store initial allocation failed. Probably not enough space in "
+        << plasma_directory_;
     // This will unmap the file, but the next one created will be as large
     // as this one (this is an implementation detail of dlmalloc).
     PlasmaAllocator::GetInstance().Free(allocation.value());