Add BufferResource.memory_available Python API (rapidsai#495)

Changes in this PR:
- Adds a Python API for `BufferResource.memory_available`.
- ~Adds an additional `BufferResource.device_memory_available` API for convenience.~
- Adds test coverage to `test_buffer_resource.py`

**Background**: I've been experimenting with spilling improvements in cudf-polars (e.g. rapidsai#489). While prototyping next steps, I found myself needing to query the amount of available device memory within a cudf-polars task. As far as I can tell, this is not currently possible, because we do not have access to the `LimitAvailableMemory` object used to construct the workers `BufferResource`. Therefore, we need bindings for `BufferResource.memory_available` to get this information from the `BufferResource` itself (Please do let me know if I'm mistaken about this).

**Note**: My Cython skills are limited, so feel free to suggest dramatic changes :)

Authors:
  - Richard (Rick) Zamora (https://github.com/rjzamora)

Approvers:
  - Mads R. B. Kristensen (https://github.com/madsbk)

URL: rapidsai#495

Author: rjzamora

python/rapidsmpf/rapidsmpf/buffer/resource.pxd

@@ -32,6 +32,9 @@ cdef extern from "<rapidsmpf/buffer/resource.hpp>" nogil:
         size_t cpp_memory_reserved "memory_reserved"(
             MemoryType mem_type
         ) except +
+        cpp_MemoryAvailable cpp_memory_available "memory_available"(
+            MemoryType mem_type
+        ) except +
         cpp_SpillManager &cpp_spill_manager "spill_manager"() except +
 
 

python/rapidsmpf/rapidsmpf/buffer/resource.pyi

@@ -16,6 +16,7 @@ class BufferResource:
         memory_available: Mapping[MemoryType, Callable[[], int]] | None = None,
         periodic_spill_check: float | None = 1e-3,
     ) -> None: ...
+    def memory_available(self, mem_type: MemoryType) -> int: ...
     def memory_reserved(self, mem_type: MemoryType) -> int: ...
     @property
     def spill_manager(self) -> SpillManager: ...

python/rapidsmpf/rapidsmpf/buffer/resource.pyx

@@ -16,10 +16,21 @@ cdef extern from *:
     ) {
         return *functor;
     }
+
+    std::int64_t _call_memory_available(
+        rapidsmpf::BufferResource* resource,
+        rapidsmpf::MemoryType mem_type
+    ) {
+        return resource->memory_available(mem_type)();
+    }
     """
     cpp_MemoryAvailable to_MemoryAvailable(
         shared_ptr[cpp_LimitAvailableMemory]
     ) except +
+    int64_t _call_memory_available(
+        cpp_BufferResource* resource,
+        MemoryType mem_type
+    ) except + nogil
 
 
 cdef class BufferResource:
@@ -121,6 +132,17 @@ cdef class BufferResource:
             ret = deref(self._handle).cpp_memory_reserved(mem_type)
         return ret
 
+    def memory_available(self, MemoryType mem_type):
+        """
+        Get the current available memory of the specified memory type.
+        """
+        cdef int64_t ret
+        cdef cpp_BufferResource* resource_ptr = self.ptr()
+        # Use inline C++ to handle the function object call
+        with nogil:
+            ret = _call_memory_available(resource_ptr, mem_type)
+        return ret
+
 
 cdef class LimitAvailableMemory:
     """

python/rapidsmpf/rapidsmpf/tests/test_buffer_resource.py

@@ -61,4 +61,11 @@ def test_buffer_resource() -> None:
     assert br.memory_reserved(MemoryType.DEVICE) == 0
     assert br.memory_reserved(MemoryType.HOST) == 0
 
+    # Chack BufferResource.memory_available
+    assert br.memory_available(MemoryType.DEVICE) == mem_available() == KiB(100)
+    buf1 = rmm.DeviceBuffer(size=KiB(50), mr=mr)
+    assert br.memory_available(MemoryType.DEVICE) == mem_available() == KiB(50)
+    del buf1
+    assert br.memory_available(MemoryType.DEVICE) == mem_available() == KiB(100)
+
     # TODO: add more `BufferResource` checks here as we add python bindings.