[FEAT] Compute pool for native executor

[colin-ho]

Create a multithreaded compute runtime for swordfish compute tasks. Switch query runtime to be single threaded, and use IO pool for scan task streams.

Additionally, adds in a tokio_select together with the tokio::signal::ctrlc and main async execution loop so that queries can be cancelled.

import os
import daft
import numpy
import time
import psutil

current_process = psutil.Process(os.getpid())

daft.set_execution_config(enable_native_executor=True, default_morsel_size=1)
dfs = [
    iter(
        daft.from_pydict({"a": numpy.random.rand(10)}).with_column(
            "plus_one", daft.col("a") + 1
        )
    )
    for _ in range(10)
]
while True:
    for i, df in enumerate(dfs):
        time.sleep(0.1)
        try:
            print("threads: ", current_process.num_threads())
            print(next(df))
        except StopIteration:
            dfs.pop(i)
    if not dfs:
        break

If you run this script you can see that the number of threads increases by only 1 per dataframe.

[codspeed-hq]

CodSpeed Performance Report

Merging #2986 will not alter performance

_{Comparing colin/compute-pool (fa8e8bb) with main (73ff3f3)}

Summary

✅ 17 untouched benchmarks

[codecov]

Codecov Report

Attention: Patch coverage is 77.70270% with 33 lines in your changes missing coverage. Please review.

Project coverage is 78.48%. Comparing base (73ff3f3) to head (fa8e8bb).

Files with missing lines	Patch %	Lines
src/daft-local-execution/src/run.rs	58.62%	12 Missing ⚠️
src/daft-io/src/lib.rs	70.96%	9 Missing ⚠️
src/daft-local-execution/src/lib.rs	84.21%	9 Missing ⚠️
...-execution/src/intermediate_ops/intermediate_op.rs	66.66%	3 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #2986      +/-   ##
==========================================
+ Coverage   78.14%   78.48%   +0.33%     
==========================================
  Files         610      610              
  Lines       72146    71837     -309     
==========================================
- Hits        56381    56379       -2     
+ Misses      15765    15458     -307     

Files with missing lines	Coverage Δ
...-local-execution/src/intermediate_ops/aggregate.rs	100.00% <ø> (ø)
.../src/intermediate_ops/anti_semi_hash_join_probe.rs	96.36% <100.00%> (ø)
...aft-local-execution/src/intermediate_ops/filter.rs	100.00% <ø> (ø)
...-execution/src/intermediate_ops/hash_join_probe.rs	97.41% <100.00%> (ø)
...ft-local-execution/src/intermediate_ops/project.rs	100.00% <ø> (ø)
...rc/daft-local-execution/src/sinks/blocking_sink.rs	74.39% <100.00%> (+3.96%)	⬆️
src/daft-local-execution/src/sources/scan_task.rs	81.95% <ø> (ø)
...-execution/src/intermediate_ops/intermediate_op.rs	77.44% <66.66%> (-1.30%)	⬇️
src/daft-io/src/lib.rs	77.96% <70.96%> (+1.60%)	⬆️
src/daft-local-execution/src/lib.rs	86.45% <84.21%> (-3.29%)	⬇️
... and 1 more

... and 12 files with indirect coverage changes

[colin-ho]

Didn't like the Option<&mut Box<dyn IntermediateOperatorState>> parameter, instead changing it to just &mut dyn IntermediateOperatorState

[colin-ho]

Todo: look into thread priority: https://docs.rs/thread-priority/latest/thread_priority/, we could potentially set priorities for compute and io pool.

Would do it on https://docs.rs/tokio/latest/tokio/runtime/struct.Builder.html#method.on_thread_start