[FEAT] Add stateful actor context and set CUDA_VISIBLE_DEVICES

[kevinzwang]

Resolves #2896

Some details about this PR:

• I moved the actor-local singleton out of PyActorPool into a specialized class PyStatefulActorSingleton
• I changed GPU resources to be accounted on a per-device level. That resulted in creating the data class AcquiredResources to store the resources used by a task or actor. The runner resources includes not only amount of CPU and memory resources, but the exact GPUs that each task/actor is using, which enables setting CUDA_VISIBLE_DEVICES in actors.
• I also moved resource acquisition and releasing logic into a PyRunnerResources class
• I added validation that GPU resources must be integers if greater than 1, which means it is no longer accurate to request for actor_resource_requests * num_workers anymore, so the actor pool context now asks for them individually.

[codspeed-hq]

CodSpeed Performance Report

Merging #3002 will not alter performance

_{Comparing kevin/stateful-udf-rank (0a30f41) with main (e4c6f3f)}

Summary

✅ 17 untouched benchmarks

[jaychia]

Took a quick first pass, but I'll probably need to give it a much more thorough review again given how much logic is changing in the PyRunner

[codecov]

Codecov Report

Attention: Patch coverage is 80.55556% with 35 lines in your changes missing coverage. Please review.

Project coverage is 78.34%. Comparing base (272163f) to head (0a30f41).
Report is 30 commits behind head on main.

Files with missing lines	Patch %	Lines
src/common/resource-request/src/lib.rs	70.58%	15 Missing ⚠️
daft/runners/pyrunner.py	89.71%	11 Missing ⚠️
daft/internal/gpu.py	57.14%	9 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #3002      +/-   ##
==========================================
- Coverage   78.39%   78.34%   -0.05%     
==========================================
  Files         603      611       +8     
  Lines       71443    72515    +1072     
==========================================
+ Hits        56005    56813     +808     
- Misses      15438    15702     +264     

Files with missing lines	Coverage Δ
daft/dependencies.py	58.33% <ø> (+0.64%)	⬆️
...al_optimization/rules/split_actor_pool_projects.rs	95.00% <100.00%> (-0.07%)	⬇️
daft/internal/gpu.py	56.00% <57.14%> (-5.54%)	⬇️
daft/runners/pyrunner.py	85.87% <89.71%> (+0.61%)	⬆️
src/common/resource-request/src/lib.rs	65.06% <70.58%> (-2.03%)	⬇️

... and 106 files with indirect coverage changes

[jaychia]

Still not 100% sure yet on the PR, saw quite a few gaps still on this pass-through. Let's find some time to do a live review in person?

[jaychia]

I don't quite understand this as to why we need to have this as an inline function.

Could we not do:

def _release_resources_callback(self, resources):
    return lambda _: self._resources.release(resources)

...

    future.add_done_callback(self._release_resources_callback(resources))