the method collect exists for struct Instrumented<..., &TaskMonitorCore>, but its trait bounds were not satisfied
主线位置：任务主线的入口。
instrument不挑被包的是Future还是Stream，但包出来的Instrumented只在一种 monitor 类型上实现了Stream。
flowchart TB s["一个 Stream"] --> a["TaskMonitor::instrument"] s --> b["static TaskMonitorCore 的 instrument"] a --> ia["M = TaskMonitor"] b --> ib["M = 引用 TaskMonitorCore"] ia --> st["实现 Future 和 Stream"] ib --> fu["只实现 Future"] fu --> err["调 StreamExt 方法：E0599"]
复现
use futures::StreamExt;
use tokio_metrics::TaskMonitorCore;
static CORE: TaskMonitorCore = TaskMonitorCore::new();
#[tokio::main]
async fn main() {
let s = CORE.instrument(futures::stream::iter([1, 2, 3]));
let v: Vec<i32> = s.collect().await;
println!("{v:?}");
}
error[E0599]: the method `collect` exists for struct `Instrumented<futures::stream::Iter<std::array::IntoIter<{integer}, 3>>, &TaskMonitorCore>`, but its trait bounds were not satisfied
--> src/bin/stream.rs:9:23
|
9 | let v: Vec<i32> = s.collect().await;
| ^^^^^^^ method cannot be called due to unsatisfied trait bounds
|
::: <registry>/pin-project-lite-0.2.17/src/lib.rs:745:9
|
745 | / $vis struct $ident $($def_generics)*
746 | | $(where
747 | | $($where_clause)*)?
... |
751 | | ),+
752 | | }
| |_________- doesn't satisfy `_: Iterator`, `_: StreamExt` or `_: Stream`
|
= note: the following trait bounds were not satisfied:
`Instrumented<futures::stream::Iter<std::array::IntoIter<{integer}, 3>>, &TaskMonitorCore>: Stream`
which is required by `Instrumented<futures::stream::Iter<std::array::IntoIter<{integer}, 3>>, &TaskMonitorCore>: StreamExt`
...
报错里的第二个位置指向 pin-project-lite 的宏内部，而不是 tokio-metrics：Instrumented 是用 pin_project! 定义的，编译器只能指到展开出结构体的那一行。真正有用的是 note 的第一条：这个类型不是 Stream。
为什么：Stream 的 impl 少一个泛型参数
Future 的实现对 monitor 类型是泛型的，Stream 的实现不是：
impl<T: Future, M: AsRef<TaskMonitorCore> + Send + Sync + 'static> Future for Instrumented<T, M> {
// ...
}
impl<T: Stream> Stream for Instrumented<T> {
// ...
}
Instrumented<T> 用的是默认参数 M = TaskMonitor。TaskMonitorCore::instrument 收 &'static self，返回 Instrumented<F, &'static Self>，第二个参数不是 TaskMonitor，于是只剩 Future。为什么没加泛型，源码和 CHANGELOG 都没说，细节见 Instrumented 与 instrument_poll 的「Stream 也能包，但只限默认 monitor」一节；这个不对称是 TaskMonitorCore 与 FutureMonitor 这个变体的代价之一。
修法
use std::sync::LazyLock;
use futures::StreamExt;
use tokio_metrics::{TaskMonitor, TaskMonitorCore};
// 要包 Stream，静态变量里放 TaskMonitor（Arc 版）而不是 TaskMonitorCore
static MONITOR: LazyLock<TaskMonitor> = LazyLock::new(TaskMonitor::new);
static CORE: TaskMonitorCore = TaskMonitorCore::new();
#[tokio::main]
async fn main() {
// 修法一：LazyLock<TaskMonitor>，包出来的是 Instrumented<_, TaskMonitor>，实现了 Stream
let s = MONITOR.instrument(futures::stream::iter([1, 2, 3]));
let v: Vec<i32> = s.collect().await;
let m = MONITOR.cumulative();
println!("LazyLock<TaskMonitor>: {v:?} instrumented={} polls={}", m.instrumented_count, m.total_poll_count);
// 修法二：不包流，包「消费这个流」的 future
let v: Vec<i32> = CORE.instrument(futures::stream::iter([1, 2, 3]).collect()).await;
let m = CORE.cumulative();
println!("TaskMonitorCore + collect future: {v:?} instrumented={} polls={}", m.instrumented_count, m.total_poll_count);
}
LazyLock<TaskMonitor>: [1, 2, 3] instrumented=1 polls=4
TaskMonitorCore + collect future: [1, 2, 3] instrumented=1 polls=1
两种修法量到的东西不一样：
- 修法一每次
poll_next都算一次 poll：3 个元素各一次Ready(Some)，最后一次Ready(None)，共 4 次。代价是多一次堆分配和一层Arc（TaskMonitor就是Arc<TaskMonitorCore>），LazyLock要 Rust 1.80；更早的版本用OnceLock。 - 修法二把整个消费过程当一个 future：
stream::iter从不返回Pending，collect一次 poll 就完成，记 1 次。流里如果有等待（比如 channel 的接收端），修法二记的是外层 future 被 poll 的次数，每次 poll 里可能取了好几个元素。想知道「每个元素之间等了多久」，只能用修法一。
TaskMonitorCore 存在的理由是能放进 static 而不用 LazyLock（const fn 构造）。如果为了包流又套回 LazyLock<TaskMonitor>，这个好处就没了，但对一个按路由或按任务种类建的 monitor，多一次初始化无关紧要。
小结
Stream for Instrumented<T>只对默认的M = TaskMonitor实现；static TaskMonitorCore包出来的是Instrumented<_, &TaskMonitorCore>，只能当Future用。- 报错指向 pin-project-lite 的宏内部，看
note的第一条就知道是缺Stream。 - 修法一：静态变量放
LazyLock<TaskMonitor>，每次poll_next记一次 poll。 - 修法二：包住消费流的 future，记的是外层 future 的 poll，粒度粗一层。
💬 关联讨论
发起讨论加载中...