[Core] dlmalloc allocate bottom-most memory chunk failed

python/ray/tests/test_plasma_unlimited.py

@@ -247,6 +247,31 @@ def test_fallback_allocation_failure(shutdown_only):
 # finally:
 # ray.shutdown()
 
+
+@pytest.mark.skipif(
+ platform.system() == "Windows", reason="Need to fix up for Windows.")
+def test_plasma_allocate(shutdown_only):
+ address = ray.init(
+ object_store_memory=300 * 1024**2,
+ _system_config={
+ "max_io_workers": 4,
+ "automatic_object_spilling_enabled": True,
+ },
+ _temp_dir="/tmp/for_test_plasma_allocate")
+ res = []
+ data = np.random.randint(
+ low=0, high=256, size=(90 * 1024**2, ), dtype=np.uint8)
+ for _ in range(3):
+ res.append(ray.put(data))
+ # keep reference for second and third object, force evict first object
+ _ = ray.get(res[1:]) # noqa
+ # keep reference for fourth object, avoid released by plasma GC.
+ __ = ray.put(data) # noqa
+
+ # Check fourth object allocate in memory.
+ _check_spilled_mb(address, spilled=180)
+
+
 if __name__ == "__main__":
  import sys
  sys.exit(pytest.main(["-v", __file__]))

src/ray/object_manager/plasma/dlmalloc.cc

@@ -56,6 +56,10 @@ int fake_munmap(void *, int64_t);
 #define HAVE_MORECORE 0
 #define DEFAULT_MMAP_THRESHOLD MAX_SIZE_T
 #define DEFAULT_GRANULARITY ((size_t)128U * 1024U)
+// Copied from plasma_allocator.cc variable kAllocationAlignment,
+// make sure to keep in sync. This for reduce memory fragmentation.
+// See https://github.com/ray-project/ray/issues/21310 for details.
+#define MALLOC_ALIGNMENT 64
 
 #include "ray/thirdparty/dlmalloc.c" // NOLINT
 

src/ray/object_manager/plasma/test/fallback_allocator_test.cc

@@ -32,11 +32,32 @@ TEST(FallbackPlasmaAllocatorTest, FallbackPassThroughTest) {
  auto plasma_directory = CreateTestDir();
  auto fallback_directory = CreateTestDir();
  int64_t kLimit = 256 * sizeof(size_t) + 2 * kMB;
+ int64_t object_size = 900 * 1024;
  PlasmaAllocator allocator(plasma_directory, fallback_directory,
  /* hugepage_enabled */ false, kLimit);
 
  EXPECT_EQ(kLimit, allocator.GetFootprintLimit());
 
+ {
+ auto allocation_1 = allocator.Allocate(object_size);
+ EXPECT_TRUE(allocation_1.has_value());
+
+ auto allocation_2 = allocator.Allocate(object_size);
+ EXPECT_TRUE(allocation_2.has_value());
+
+ EXPECT_EQ(2 * object_size, allocator.Allocated());
+
+ allocator.Free(std::move(allocation_1.value()));
+ auto allocation_3 = allocator.Allocate(object_size);
+ EXPECT_TRUE(allocation_3.has_value());
+ EXPECT_EQ(0, allocator.FallbackAllocated());
+ EXPECT_EQ(2 * object_size, allocator.Allocated());
+
+ allocator.Free(std::move(allocation_2.value()));
+ allocator.Free(std::move(allocation_3.value()));
+ EXPECT_EQ(0, allocator.Allocated());
+ }
+
  int64_t expect_allocated = 0;
  int64_t expect_fallback_allocated = 0;
  std::vector<Allocation> allocations;