gitadmin/foxhunt
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix(ml): add tokio runtime to training binaries for OTLP export

Browse Source 
All 6 training binaries (hyperopt_baseline_rl, hyperopt_baseline_supervised,
train_baseline_rl, train_baseline_supervised, evaluate_baseline,
evaluate_supervised) used sync fn main() but the OTLP batch exporter
requires a tokio runtime (tonic/hyper-util gRPC transport). This caused
an immediate panic on CI when OTEL_EXPORTER_OTLP_ENDPOINT was set.

Fix: #[tokio::main(flavor = "current_thread")] on all 6 binaries.
Also fix pre-existing clippy warnings (shadow, let_underscore_must_use,
doc_markdown, cognitive_complexity, integer_division, unsafe_code).

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
jgrusewski
2026-03-02 18:16:37 +01:00
parent a57e4911ff
commit fcf87a5f72
6 changed files with 34 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
crates/ml/examples/evaluate_baseline.rs
View File

@@ -128,6 +128,7 @@ struct Args {
///
/// Expected format: `{ "model_key": { "best_params": { ... }, ... } }`
/// Returns `None` if the file doesn't exist or can't be parsed.
#[allow(clippy::cognitive_complexity)]
fn load_hyperopt_params(hp_path: &Option<PathBuf>, model_key: &str) -> Option<Value> {
let file_path = hp_path.as_ref()?;
if !file_path.exists() {
@@ -347,6 +348,7 @@ fn evaluate_dqn_fold(
}
// Create DQN with same config as training (gamma must match train_baseline)
#[allow(clippy::integer_division)]
let config = DQNConfig {
state_dim: args.feature_dim,
num_actions: args.num_actions,
@@ -478,6 +480,7 @@ fn evaluate_ppo_fold(
}
// Create PPO config matching training
#[allow(clippy::integer_division)]
let config = PPOConfig {
state_dim: args.feature_dim,
num_actions: args.num_actions,
@@ -636,7 +639,8 @@ fn run_sanity_checks(
// ---------------------------------------------------------------------------
#[allow(clippy::cognitive_complexity, clippy::too_many_lines)]
fn main() -> Result<()> {
#[tokio::main(flavor = "current_thread")]
async fn main() -> Result<()> {
// Initialize tracing with optional OTLP export to Tempo
let otlp_endpoint = std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT").ok();
if let Err(e) = common::observability::init_observability(

3
crates/ml/examples/evaluate_supervised.rs
View File

@@ -532,7 +532,8 @@ fn evaluate_fold(
// ---------------------------------------------------------------------------
#[allow(clippy::cognitive_complexity, clippy::too_many_lines)]
fn main() -> Result<()> {
#[tokio::main(flavor = "current_thread")]
async fn main() -> Result<()> {
// Initialize tracing with optional OTLP export to Tempo
let otlp_endpoint = std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT").ok();
if let Err(e) = common::observability::init_observability(

23
crates/ml/examples/hyperopt_baseline_rl.rs
View File

@@ -33,7 +33,7 @@
//! }
//! ```
#![allow(unused_crate_dependencies)]
#![allow(unused_crate_dependencies, unsafe_code)]
use anyhow::{Context, Result};
use clap::Parser;
@@ -261,7 +261,8 @@ fn run_ppo_hyperopt(args: &Args, parallel: usize, device: &candle_core::Device)
}
#[allow(clippy::cognitive_complexity)]
fn main() -> Result<()> {
#[tokio::main(flavor = "current_thread")]
async fn main() -> Result<()> {
// Initialize tracing with optional OTLP export to Tempo
let otlp_endpoint = std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT").ok();
if let Err(e) = common::observability::init_observability(
@@ -275,7 +276,8 @@ fn main() -> Result<()> {
metrics_server::start_metrics_server(9094);
training_metrics::set_active_workers(1.0);
// Pre-allocate CUBLAS workspace for deterministic + faster tensor core ops
// Pre-allocate CUBLAS workspace for deterministic + faster tensor core ops.
// SAFETY: called once at startup before any multi-threading or CUDA work begins.
unsafe { std::env::set_var("CUBLAS_WORKSPACE_CONFIG", ":4096:8"); }
let args = Args::parse();
@@ -309,6 +311,7 @@ fn main() -> Result<()> {
// Smart cap: on GPU instances with few vCPUs (e.g. L40S-1-48G has 8 vCPU),
// more threads than vCPU/2 causes CPU contention that starves GPU.
let cpu_cap = cpus.saturating_sub(2).max(1);
#[allow(clippy::integer_division)]
let cpu_smart_cap = (cpus / 2).max(1);
let parallel = if args.parallel == 0 {
cpu_cap.min(cpu_smart_cap)
@@ -319,18 +322,18 @@ fn main() -> Result<()> {
// Cap by VRAM budget with corrected constants (DQN ~50MB overhead, ~0.0005MB/sample)
let budget = ml::hyperopt::HardwareBudget::detect();
let vram_cap = budget.plan_hyperopt(50.0, 0.0005, 64.0, 4096.0).max_concurrent_trials;
let parallel = parallel.min(vram_cap);
info!("Parallel: {} threads (CPU: {}, CPU smart cap: {}, VRAM cap: {})", parallel, cpus, cpu_smart_cap, vram_cap);
let num_threads = parallel.min(vram_cap);
info!("Parallel: {} threads (CPU: {}, CPU smart cap: {}, VRAM cap: {})", num_threads, cpus, cpu_smart_cap, vram_cap);
// Configure rayon thread pool for parallel trial evaluation
if parallel > 1 {
if num_threads > 1 {
rayon::ThreadPoolBuilder::new()
.num_threads(parallel)
.num_threads(num_threads)
.build_global()
.unwrap_or_else(|e| {
warn!("Failed to set rayon thread pool size: {}", e);
});
info!("Rayon thread pool configured: {} threads", parallel);
info!("Rayon thread pool configured: {} threads", num_threads);
}
// Scope data directory to symbol subdirectory to avoid mixing instruments
@@ -383,7 +386,7 @@ fn main() -> Result<()> {
// Run DQN hyperopt
if run_dqn {
match run_dqn_hyperopt(&args, parallel, &device) {
match run_dqn_hyperopt(&args, num_threads, &device) {
Ok(dqn_result) => {
results.insert("dqn".to_owned(), dqn_result);
},
@@ -400,7 +403,7 @@ fn main() -> Result<()> {
// Run PPO hyperopt
if run_ppo {
match run_ppo_hyperopt(&args, parallel, &device) {
match run_ppo_hyperopt(&args, num_threads, &device) {
Ok(ppo_result) => {
results.insert("ppo".to_owned(), ppo_result);
},

3
crates/ml/examples/hyperopt_baseline_supervised.rs
View File

@@ -543,7 +543,8 @@ const VALID_MODELS: &[&str] = &[
];
#[allow(clippy::cognitive_complexity)]
fn main() -> Result<()> {
#[tokio::main(flavor = "current_thread")]
async fn main() -> Result<()> {
// Initialize tracing with optional OTLP export to Tempo
let otlp_endpoint = std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT").ok();
if let Err(e) = common::observability::init_observability(

16
crates/ml/examples/train_baseline_rl.rs
View File

@@ -188,9 +188,10 @@ fn hp_usize(params: &Option<Value>, key: &str) -> Option<usize> {
// Fold data preparation (extracted for prefetching)
// ---------------------------------------------------------------------------
/// Prepared fold data: (train_features, val_features, train_bars, val_bars)
/// Prepared fold data: (`train_features`, `val_features`, `train_bars`, `val_bars`)
type FoldData = (Vec<[f64; 51]>, Vec<[f64; 51]>, Vec<OHLCVBar>, Vec<OHLCVBar>);
#[allow(clippy::cognitive_complexity)]
/// Prepare a fold's data for training: extract features, normalize, align bars.
///
/// Returns `None` if feature extraction fails or produces empty features.
@@ -268,6 +269,7 @@ fn features_to_trainer_format(
/// fold structure stays in this binary; only per-fold training delegates to `DQNTrainer`.
///
/// Returns the best validation loss achieved.
#[allow(clippy::cognitive_complexity, clippy::too_many_arguments)]
fn train_dqn_fold(
fold: usize,
train_features: &[[f64; 51]],
@@ -400,18 +402,19 @@ fn train_dqn_fold(
/// production trainer. The walk-forward fold structure stays in this binary;
/// only per-fold training delegates to `PpoTrainer`.
///
/// Returns the best validation loss achieved (approximated by final value_loss).
/// Returns the best validation loss achieved (approximated by final `value_loss`).
#[allow(clippy::cognitive_complexity)]
fn train_ppo_fold(
fold: usize,
train_features: &[[f64; 51]],
_val_features: &[[f64; 51]],
val_features: &[[f64; 51]],
train_bars: &[OHLCVBar],
_val_bars: &[OHLCVBar],
args: &Args,
output_dir: &Path,
hp: &Option<Value>,
) -> Result<f64> {
info!(" [PPO] Fold {} -- {} train, {} val features", fold, train_features.len(), _val_features.len());
info!(" [PPO] Fold {} -- {} train, {} val features", fold, train_features.len(), val_features.len());
let n_train = train_features.len();
if n_train < 2 {
@@ -652,7 +655,7 @@ fn run_training(args: &Args) -> Result<Vec<RlTrainingResult>> {
.spawn(move || {
info!(" [Prefetch] Loading fold {} data on background thread", nw.fold);
let result = prepare_fold_data(&nw, &out_dir);
let _ = tx.send(result);
drop(tx.send(result));
});
Some(rx)
} else {
@@ -824,7 +827,8 @@ fn run_training(args: &Args) -> Result<Vec<RlTrainingResult>> {
// Main
// ---------------------------------------------------------------------------
fn main() -> Result<()> {
#[tokio::main(flavor = "current_thread")]
async fn main() -> Result<()> {
// Initialize tracing with optional OTLP export to Tempo
let otlp_endpoint = std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT").ok();
if let Err(e) = common::observability::init_observability(

3
crates/ml/examples/train_baseline_supervised.rs
View File

@@ -796,7 +796,8 @@ fn run_training(args: &Args) -> Result<Vec<TrainingResult>> {
// Main
// ---------------------------------------------------------------------------
fn main() -> Result<()> {
#[tokio::main(flavor = "current_thread")]
async fn main() -> Result<()> {
// Initialize tracing with optional OTLP export to Tempo
let otlp_endpoint = std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT").ok();
if let Err(e) = common::observability::init_observability(