Enable taking Python Streams from Context

[nirandaperera]

This PR enables taking python Streams from the stream pool in the buffer resource obj.
This achieved by calling Stream._from_cudaStream_t(stream_view.value(), self). Context python obj is passed as the owner of the Stream so that it outlives the returned Stream obj.

Fixes #592

[TomAugspurger]

Thanks @nirandaperera. Seems reasonable enough, and it should be sufficient for cudf-polars.

I don't think cudf-polars has a preference for whether this lives on Context or BufferResource, since we'll only access it via a Context. Though I don't know enough Cython / C++ to say whether it being on BufferResource would make getting the stream_pool in the cython method any easier.

[TomAugspurger]

I wonder if the lifetime should be self.br rather than self. My understanding is that the stream pool is attached to the (C++) BufferResource, not the context.

But the Context being alive implies that the context.br is alive, so it ends up not mattering, maybe?

[nirandaperera]

I think you are right. We accept a BufferResource to the Context. So, its best to pass br here.

[TomAugspurger]

Thanks @nirandaperera. I think things are good from the cudf-polars side. When the user configures the query to run with cudf-polars, we'll have some sort of bootstrapping to do (either explicitly or implicitly) to build the context:

import rapidsmpf.buffer.resource
import rapidsmpf.streaming.core.context
import rapidsmpf.communicator.single
import rapidsmpf.config
import rmm.mr

opts = rapidsmpf.config.Options()
comm = rapidsmpf.communicator.single.new_communicator(opts)
br = rapidsmpf.buffer.resource.BufferResource(rmm.mr.get_current_device_resource())
ctx = rapidsmpf.streaming.core.context.Context(comm, br, opts)

And then we'll get streams from the pool as needed (source IR nodes):

In [22]: stream = ctx.get_stream_from_pool()

In [23]: stream
Out[23]: <rmm.pylibrmm.stream.Stream at 0x7f3b024fdac0>

[nirandaperera]

I have addressed @wence- 's comments. So merging this

[nirandaperera]

/merge