Skip to content

refactor: migrate all tests/benches from TaskExecutor to TypedExecutor #66

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
deepjoy merged 1 commit into from
Mar 21, 2026
Merged

refactor: migrate all tests/benches from TaskExecutor to TypedExecutor #66

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 13 additions & 5 deletions benches/dependencies.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,9 +7,10 @@ use std::time::{Duration, Instant};
use criterion::{criterion_group, criterion_main, BenchmarkId, Criterion};
#[cfg(feature = "profile")]
use pprof::criterion::{Output, PProfProfiler};
use serde::{Deserialize, Serialize};
use taskmill::{
Domain, DomainKey, Scheduler, SchedulerEvent, TaskContext, TaskError, TaskExecutor, TaskStore,
TaskSubmission,
Domain, DomainKey, Scheduler, SchedulerEvent, TaskContext, TaskError, TaskStore,
TaskSubmission, TypedExecutor, TypedTask,
};
use tokio::runtime::Runtime;
use tokio_util::sync::CancellationToken;
Expand All @@ -19,18 +20,25 @@ impl DomainKey for BenchDomain {
const NAME: &'static str = "bench";
}

#[derive(Serialize, Deserialize)]
struct BenchTask;
impl TypedTask for BenchTask {
type Domain = BenchDomain;
const TASK_TYPE: &'static str = "test";
}

struct NoopExecutor;

impl TaskExecutor for NoopExecutor {
async fn execute<'a>(&'a self, _ctx: &'a TaskContext) -> Result<(), TaskError> {
impl TypedExecutor<BenchTask> for NoopExecutor {
async fn execute(&self, _payload: BenchTask, _ctx: &TaskContext) -> Result<(), TaskError> {
Ok(())
}
}

async fn build_scheduler(max_concurrency: usize) -> Scheduler {
Scheduler::builder()
.store(TaskStore::open_memory().await.unwrap())
.domain(Domain::<BenchDomain>::new().raw_executor("test", NoopExecutor))
.domain(Domain::<BenchDomain>::new().task::<BenchTask>(NoopExecutor))
.max_concurrency(max_concurrency)
.poll_interval(Duration::from_millis(10))
.build()
Expand Down
22 changes: 15 additions & 7 deletions benches/groups.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,9 +5,10 @@
use std::time::{Duration, Instant};

use criterion::{criterion_group, criterion_main, BenchmarkId, Criterion};
use serde::{Deserialize, Serialize};
use taskmill::{
Domain, DomainKey, Scheduler, SchedulerEvent, TaskContext, TaskError, TaskExecutor, TaskStore,
TaskSubmission,
Domain, DomainKey, Scheduler, SchedulerEvent, TaskContext, TaskError, TaskStore,
TaskSubmission, TypedExecutor, TypedTask,
};
use tokio::runtime::Runtime;
use tokio_util::sync::CancellationToken;
Expand All @@ -17,10 +18,17 @@ impl DomainKey for BenchDomain {
const NAME: &'static str = "bench";
}

#[derive(Serialize, Deserialize)]
struct BenchTask;
impl TypedTask for BenchTask {
type Domain = BenchDomain;
const TASK_TYPE: &'static str = "test";
}

struct NoopExecutor;

impl TaskExecutor for NoopExecutor {
async fn execute<'a>(&'a self, _ctx: &'a TaskContext) -> Result<(), TaskError> {
impl TypedExecutor<BenchTask> for NoopExecutor {
async fn execute(&self, _payload: BenchTask, _ctx: &TaskContext) -> Result<(), TaskError> {
Ok(())
}
}
Expand Down Expand Up @@ -56,7 +64,7 @@ fn bench_dispatch_no_groups(c: &mut Criterion) {
for _ in 0..iters {
let sched = Scheduler::builder()
.store(TaskStore::open_memory().await.unwrap())
.domain(Domain::<BenchDomain>::new().raw_executor("test", NoopExecutor))
.domain(Domain::<BenchDomain>::new().task::<BenchTask>(NoopExecutor))
.max_concurrency(8)
.poll_interval(Duration::from_millis(10))
.build()
Expand Down Expand Up @@ -90,7 +98,7 @@ fn bench_dispatch_one_group(c: &mut Criterion) {
for _ in 0..iters {
let sched = Scheduler::builder()
.store(TaskStore::open_memory().await.unwrap())
.domain(Domain::<BenchDomain>::new().raw_executor("test", NoopExecutor))
.domain(Domain::<BenchDomain>::new().task::<BenchTask>(NoopExecutor))
.max_concurrency(8)
.group_concurrency("g0", 500) // high limit — no artificial throttling
.poll_interval(Duration::from_millis(10))
Expand Down Expand Up @@ -138,7 +146,7 @@ fn bench_dispatch_group_scaling(c: &mut Criterion) {

let mut builder = Scheduler::builder()
.store(TaskStore::open_memory().await.unwrap())
.domain(Domain::<BenchDomain>::new().raw_executor("test", NoopExecutor))
.domain(Domain::<BenchDomain>::new().task::<BenchTask>(NoopExecutor))
.max_concurrency(8)
.poll_interval(Duration::from_millis(10));

Expand Down
18 changes: 13 additions & 5 deletions benches/history.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,9 +5,10 @@
use std::time::Duration;

use criterion::{criterion_group, criterion_main, BenchmarkId, Criterion};
use serde::{Deserialize, Serialize};
use taskmill::{
Domain, DomainKey, Scheduler, SchedulerEvent, TaskContext, TaskError, TaskExecutor, TaskStore,
TaskSubmission,
Domain, DomainKey, Scheduler, SchedulerEvent, TaskContext, TaskError, TaskStore,
TaskSubmission, TypedExecutor, TypedTask,
};
use tokio::runtime::Runtime;
use tokio_util::sync::CancellationToken;
Expand All @@ -17,10 +18,17 @@ impl DomainKey for BenchDomain {
const NAME: &'static str = "bench";
}

#[derive(Serialize, Deserialize)]
struct BenchTask;
impl TypedTask for BenchTask {
type Domain = BenchDomain;
const TASK_TYPE: &'static str = "test";
}

struct NoopExecutor;

impl TaskExecutor for NoopExecutor {
async fn execute<'a>(&'a self, _ctx: &'a TaskContext) -> Result<(), TaskError> {
impl TypedExecutor<BenchTask> for NoopExecutor {
async fn execute(&self, _payload: BenchTask, _ctx: &TaskContext) -> Result<(), TaskError> {
Ok(())
}
}
Expand All @@ -29,7 +37,7 @@ impl TaskExecutor for NoopExecutor {
async fn build_scheduler_with_history(n: usize) -> Scheduler {
let sched = Scheduler::builder()
.store(TaskStore::open_memory().await.unwrap())
.domain(Domain::<BenchDomain>::new().raw_executor("test", NoopExecutor))
.domain(Domain::<BenchDomain>::new().task::<BenchTask>(NoopExecutor))
.max_concurrency(32)
.poll_interval(Duration::from_millis(10))
.build()
Expand Down
26 changes: 17 additions & 9 deletions benches/retry.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,9 +5,10 @@
use std::time::{Duration, Instant};

use criterion::{black_box, criterion_group, criterion_main, BenchmarkId, Criterion};
use serde::{Deserialize, Serialize};
use taskmill::{
BackoffStrategy, Domain, DomainKey, RetryPolicy, Scheduler, SchedulerEvent, TaskContext,
TaskError, TaskExecutor, TaskStore, TaskSubmission,
TaskError, TaskStore, TaskSubmission, TypedExecutor, TypedTask,
};
use tokio::runtime::Runtime;
use tokio_util::sync::CancellationToken;
Expand All @@ -19,20 +20,29 @@ impl DomainKey for BenchDomain {
const NAME: &'static str = "bench";
}

// ── Typed Tasks ────────────────────────────────────────────────────

#[derive(Serialize, Deserialize)]
struct FailTask;
impl TypedTask for FailTask {
type Domain = BenchDomain;
const TASK_TYPE: &'static str = "fail";
}

// ── Executors ───────────────────────────────────────────────────────

struct FailPermanentExecutor;

impl TaskExecutor for FailPermanentExecutor {
async fn execute<'a>(&'a self, _ctx: &'a TaskContext) -> Result<(), TaskError> {
impl TypedExecutor<FailTask> for FailPermanentExecutor {
async fn execute(&self, _payload: FailTask, _ctx: &TaskContext) -> Result<(), TaskError> {
Err(TaskError::permanent("bench: permanent failure"))
}
}

struct FailRetryableExecutor;

impl TaskExecutor for FailRetryableExecutor {
async fn execute<'a>(&'a self, _ctx: &'a TaskContext) -> Result<(), TaskError> {
impl TypedExecutor<FailTask> for FailRetryableExecutor {
async fn execute(&self, _payload: FailTask, _ctx: &TaskContext) -> Result<(), TaskError> {
Err(TaskError::retryable("bench: transient failure"))
}
}
Expand Down Expand Up @@ -103,9 +113,7 @@ fn bench_dispatch_permanent_failure(c: &mut Criterion) {
for _ in 0..iters {
let sched = Scheduler::builder()
.store(TaskStore::open_memory().await.unwrap())
.domain(
Domain::<BenchDomain>::new().raw_executor("fail", FailPermanentExecutor),
)
.domain(Domain::<BenchDomain>::new().task::<FailTask>(FailPermanentExecutor))
.max_concurrency(8)
.max_retries(0)
.poll_interval(Duration::from_millis(10))
Expand Down Expand Up @@ -203,7 +211,7 @@ fn bench_dispatch_retryable_dead_letter(c: &mut Criterion) {
.store(TaskStore::open_memory().await.unwrap())
.domain(
Domain::<BenchDomain>::new()
.raw_executor("fail", FailRetryableExecutor)
.task::<FailTask>(FailRetryableExecutor)
.default_retry(policy),
)
.max_concurrency(8)
Expand Down
37 changes: 26 additions & 11 deletions benches/scheduler.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,9 +5,10 @@
use std::time::{Duration, Instant};

use criterion::{criterion_group, criterion_main, BenchmarkId, Criterion};
use serde::{Deserialize, Serialize};
use taskmill::{
Domain, DomainKey, Priority, Scheduler, SchedulerEvent, TaskContext, TaskError, TaskExecutor,
TaskStore, TaskSubmission,
Domain, DomainKey, Priority, Scheduler, SchedulerEvent, TaskContext, TaskError, TaskStore,
TaskSubmission, TypedExecutor, TypedTask,
};
use tokio::runtime::Runtime;
use tokio_util::sync::CancellationToken;
Expand All @@ -19,12 +20,28 @@ impl DomainKey for BenchDomain {
const NAME: &'static str = "bench";
}

// ── Typed Tasks ────────────────────────────────────────────────────

#[derive(Serialize, Deserialize)]
struct BenchTask;
impl TypedTask for BenchTask {
type Domain = BenchDomain;
const TASK_TYPE: &'static str = "test";
}

#[derive(Serialize, Deserialize)]
struct ByteTestTask;
impl TypedTask for ByteTestTask {
type Domain = BenchDomain;
const TASK_TYPE: &'static str = "byte-test";
}

// ── Test Executors ──────────────────────────────────────────────────

struct NoopExecutor;

impl TaskExecutor for NoopExecutor {
async fn execute<'a>(&'a self, _ctx: &'a TaskContext) -> Result<(), TaskError> {
impl TypedExecutor<BenchTask> for NoopExecutor {
async fn execute(&self, _payload: BenchTask, _ctx: &TaskContext) -> Result<(), TaskError> {
Ok(())
}
}
Expand All @@ -35,8 +52,8 @@ struct ByteProgressExecutor {
chunk_size: u64,
}

impl TaskExecutor for ByteProgressExecutor {
async fn execute<'a>(&'a self, ctx: &'a TaskContext) -> Result<(), TaskError> {
impl TypedExecutor<ByteTestTask> for ByteProgressExecutor {
async fn execute(&self, _payload: ByteTestTask, ctx: &TaskContext) -> Result<(), TaskError> {
ctx.set_bytes_total(self.total);
let mut remaining = self.total;
while remaining > 0 {
Expand All @@ -53,7 +70,7 @@ impl TaskExecutor for ByteProgressExecutor {
async fn build_scheduler(max_concurrency: usize) -> Scheduler {
Scheduler::builder()
.store(TaskStore::open_memory().await.unwrap())
.domain(Domain::<BenchDomain>::new().raw_executor("test", NoopExecutor))
.domain(Domain::<BenchDomain>::new().task::<BenchTask>(NoopExecutor))
.max_concurrency(max_concurrency)
.poll_interval(std::time::Duration::from_millis(10))
.build()
Expand Down Expand Up @@ -360,8 +377,7 @@ fn bench_byte_progress_overhead(c: &mut Criterion) {
for _ in 0..iters {
let sched = Scheduler::builder()
.store(TaskStore::open_memory().await.unwrap())
.domain(Domain::<BenchDomain>::new().raw_executor(
"byte-test",
.domain(Domain::<BenchDomain>::new().task::<ByteTestTask>(
ByteProgressExecutor {
total: 1_048_576,
chunk_size: 1024,
Expand Down Expand Up @@ -417,8 +433,7 @@ fn bench_byte_progress_snapshot(c: &mut Criterion) {
for _ in 0..iters {
let sched = Scheduler::builder()
.store(TaskStore::open_memory().await.unwrap())
.domain(Domain::<BenchDomain>::new().raw_executor(
"byte-test",
.domain(Domain::<BenchDomain>::new().task::<ByteTestTask>(
ByteProgressExecutor {
total: 10_485_760,
chunk_size: 65_536,
Expand Down
17 changes: 13 additions & 4 deletions benches/tags.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,8 +5,10 @@
use std::time::{Duration, Instant};

use criterion::{criterion_group, criterion_main, BenchmarkId, Criterion};
use serde::{Deserialize, Serialize};
use taskmill::{
Domain, DomainKey, Scheduler, TaskContext, TaskError, TaskExecutor, TaskStore, TaskSubmission,
Domain, DomainKey, Scheduler, TaskContext, TaskError, TaskStore, TaskSubmission, TypedExecutor,
TypedTask,
};
use tokio::runtime::Runtime;

Expand All @@ -15,10 +17,17 @@ impl DomainKey for BenchDomain {
const NAME: &'static str = "bench";
}

#[derive(Serialize, Deserialize)]
struct BenchTask;
impl TypedTask for BenchTask {
type Domain = BenchDomain;
const TASK_TYPE: &'static str = "test";
}

struct NoopExecutor;

impl TaskExecutor for NoopExecutor {
async fn execute<'a>(&'a self, _ctx: &'a TaskContext) -> Result<(), TaskError> {
impl TypedExecutor<BenchTask> for NoopExecutor {
async fn execute(&self, _payload: BenchTask, _ctx: &TaskContext) -> Result<(), TaskError> {
Ok(())
}
}
Expand Down Expand Up @@ -57,7 +66,7 @@ fn bench_submit_with_tags(c: &mut Criterion) {
for _ in 0..iters {
let sched = Scheduler::builder()
.store(TaskStore::open_memory().await.unwrap())
.domain(Domain::<BenchDomain>::new().raw_executor("test", NoopExecutor))
.domain(Domain::<BenchDomain>::new().task::<BenchTask>(NoopExecutor))
.max_concurrency(4)
.poll_interval(Duration::from_millis(10))
.build()
Expand Down
21 changes: 17 additions & 4 deletions examples/profile_dep_chain.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,26 +2,39 @@
/// Run with: cargo run --release --example profile_dep_chain
use std::time::{Duration, Instant};

use serde::{Deserialize, Serialize};
use taskmill::{
Domain, DomainKey, Scheduler, TaskContext, TaskError, TaskExecutor, TaskStore, TaskSubmission,
Domain, DomainKey, Scheduler, TaskContext, TaskError, TaskStore, TaskSubmission, TypedExecutor,
TypedTask,
};

struct BenchDomain;
impl DomainKey for BenchDomain {
const NAME: &'static str = "bench";
}

#[derive(Serialize, Deserialize)]
struct BenchTask;
impl TypedTask for BenchTask {
type Domain = BenchDomain;
const TASK_TYPE: &'static str = "test";
}

struct NoopExecutor;
impl TaskExecutor for NoopExecutor {
async fn execute<'a>(&'a self, _ctx: &'a TaskContext) -> Result<(), TaskError> {
impl TypedExecutor<BenchTask> for NoopExecutor {
async fn execute<'a>(
&'a self,
_payload: BenchTask,
_ctx: &'a TaskContext,
) -> Result<(), TaskError> {
Ok(())
}
}

async fn build_scheduler() -> Scheduler {
Scheduler::builder()
.store(TaskStore::open_memory().await.unwrap())
.domain(Domain::<BenchDomain>::new().raw_executor("test", NoopExecutor))
.domain(Domain::<BenchDomain>::new().task::<BenchTask>(NoopExecutor))
.max_concurrency(8)
.poll_interval(Duration::from_millis(10))
.build()
Expand Down
Loading
Loading