1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
use arc_gc::gc::GC;
use std::{collections::VecDeque, sync::Arc};
use crate::{
lambda::runnable::{Runnable, RuntimeError, StepResult},
types::{
async_handle::OnionAsyncHandle,
object::{GCArcStorage, OnionObjectCell, OnionObjectExt},
},
unwrap_step_result,
};
const NUM_PRIORITY_LEVELS: usize = 3; // 优先级级别数量 - 1
/// 2^n-1 序列
#[inline(always)]
fn generate_sched_step(n: usize) -> u64 {
(1u64 << (n + 1)) - 1
}
pub struct Task {
runnable: Box<dyn Runnable>,
task_handler: (Arc<OnionAsyncHandle>, GCArcStorage),
priority: usize, // 优先级,决定调度间隔
}
impl Task {
pub fn new(
runnable: Box<dyn Runnable>,
task_handler: (Arc<OnionAsyncHandle>, GCArcStorage),
priority: usize,
) -> Self {
Self {
runnable,
task_handler,
priority,
}
}
}
pub struct AsyncScheduler {
queue: VecDeque<Task>,
main_task_handler: (Arc<OnionAsyncHandle>, GCArcStorage), // 主任务处理器
step: u64, // 当前调度步数
}
impl AsyncScheduler {
pub fn new(main_task: Task) -> Self {
let mut queue = VecDeque::new();
let main_task_handler = main_task.task_handler.clone();
queue.push_back(main_task);
AsyncScheduler {
queue,
main_task_handler,
step: 0,
}
}
}
impl Runnable for AsyncScheduler {
fn step(&mut self, gc: &mut GC<OnionObjectCell>) -> StepResult {
// 单队列调度:遍历队列,按步数调度
let len = self.queue.len();
if len == 0 {
// 所有任务都已完成,此时对 main_task_handler 执行valueof
return StepResult::Return(
unwrap_step_result!(self.main_task_handler.0.value_of()).into(),
);
}
let mut i = 0;
self.step += 1;
while i < len {
if let Some(mut task) = self.queue.pop_front() {
// 只有 step % generate_sched_step(priority) == 0 时才调度
if self.step % generate_sched_step(task.priority) == 0 {
let step_result = task.runnable.step(gc);
match step_result {
StepResult::Continue => {
task.priority = 0; // 重置优先级
self.queue.push_back(task);
}
StepResult::Return(ref result) => {
unwrap_step_result!(task.task_handler.0.set_result(result.weak()));
}
StepResult::Error(RuntimeError::Pending) => {
// 一旦pending立即降级
task.priority = std::cmp::min(task.priority + 1, NUM_PRIORITY_LEVELS);
self.queue.push_back(task);
}
e @ StepResult::Error(_) => return e,
StepResult::NewRunnable(_) => {
// AsyncScheduler 不支持 NewRunnable 因为它没有意义,出现 NewRunnable 就意味着逻辑有问题
return StepResult::Error(RuntimeError::DetailedError(
"AsyncScheduler does not support NewRunnable"
.to_string()
.into(),
));
}
StepResult::ReplaceRunnable(_) => {
// 同上
return StepResult::Error(RuntimeError::DetailedError(
"AsyncScheduler does not support ReplaceRunnable"
.to_string()
.into(),
));
}
StepResult::SpawnRunnable(new_task) => {
self.queue.push_back(*new_task);
self.queue.push_back(task);
}
StepResult::SetSelfObject(_) => {
self.queue.push_back(task);
}
}
} else {
// 未到调度步,放回队尾
self.queue.push_back(task);
}
}
i += 1;
}
StepResult::Continue
}
fn receive(
&mut self,
_step_result: &StepResult,
_gc: &mut GC<OnionObjectCell>,
) -> Result<(), RuntimeError> {
Err(RuntimeError::DetailedError(
"AsyncScheduler does not support receive".to_string().into(),
))
}
fn format_context(&self) -> String {
let mut output = Vec::new();
// 1. 调度器自身的状态
output.push(format!(
"-> AsyncScheduler Status:\n - Current Step: {}\n - Total Tasks in Queue: {}",
self.step,
self.queue.len()
));
// 2. 遍历队列中的所有任务
if self.queue.is_empty() {
output.push(" - Queue is empty.".to_string());
} else {
output.push("--- Task Queue Details ---".to_string());
for (index, task) in self.queue.iter().enumerate() {
// 下一次轮到该任务执行的步数
let next_run_step = {
let sched_interval = generate_sched_step(task.priority);
// 计算下一个能被 sched_interval 整除的 step
if self.step % sched_interval == 0 {
self.step // 就是当前步
} else {
self.step - (self.step % sched_interval) + sched_interval
}
};
// 获取 Runnable 的类型名
let runnable_type = std::any::type_name_of_val(&*task.runnable);
// 3. 为每个任务创建一个摘要条目
let task_summary = format!(
" [Task #{}] Priority: {} (Next run at step {}), Type: {}",
index,
task.priority,
next_run_step,
runnable_type.split("::").last().unwrap_or(runnable_type) // 简化类型名显示
);
output.push(task_summary);
// 4. 获取并缩进该任务内部的上下文
let inner_context = task.runnable.format_context();
for line in inner_context.lines() {
// 为内部上下文的每一行添加缩进,以保持层次结构清晰
output.push(format!(" {}", line));
}
}
}
output.join("\n")
}
}