blob_pager_and_verifier/
profile.rs

1// Copyright 2026 The Fuchsia Authors. All rights reserved.
2// Use of this source code is governed by a BSD-style license that can be
3// found in the LICENSE file.
4
5use crate::Cache;
6use fidl_fuchsia_storage_block as fblock;
7use fuchsia_async as fasync;
8use fuchsia_hash::Hash;
9use futures::future::{BoxFuture, join_all};
10use std::collections::{HashMap, HashSet, hash_map};
11use std::sync::Arc;
12use storage_profile::{BlobMessage, FileMessage, Message};
13
14// Matches the 4 KiB block alignment used by Fxfs on-disk profile files and `mapping::BLOCK_SIZE`
15// in the block driver (`BlobMessage` and `FileMessage` entries are packed into 4 KiB blocks
16// without crossing block boundaries).
17const PROFILE_BLOCK_SIZE: usize = 4096;
18
19// Matches `REPLAY_THREADS` in Fxfs's `profile.rs`: two blocking threads allow overlapping
20// `ZX_VMO_OP_PREFETCH` page-in latency without saturating the block driver or executor pool.
21const REPLAY_THREADS: usize = 2;
22
23struct ReplayRequest {
24    vmo: Arc<zx::Vmo>,
25    offset: u64,
26}
27
28struct ReplayHandle {
29    replay_threads: BoxFuture<'static, ()>,
30    // Holds the task that opens blobs and keeps their child VMOs cached in `local_cache` during
31    // replay. Dropping `ReplayHandle` drops this `fasync::Task`, which aborts the task and drops
32    // `local_cache` (allowing `Cache` to evict unused blobs) and `sender`.
33    _cache_task: fasync::Task<()>,
34}
35
36impl ReplayHandle {
37    fn start(cache: &Arc<Cache>, replay_vmo: zx::Vmo) -> Result<Self, zx::Status> {
38        let messages =
39            BlobMessage::decode_from_vmo(replay_vmo, PROFILE_BLOCK_SIZE).map_err(|error| {
40                log::warn!(error:?; "Failed to decode replay profile VMO");
41                if error == zx::Status::OUT_OF_RANGE {
42                    zx::Status::OUT_OF_RANGE
43                } else {
44                    zx::Status::IO_DATA_INTEGRITY
45                }
46            })?;
47
48        let (sender, receiver) = async_channel::unbounded::<ReplayRequest>();
49
50        // Create async_channel. An async task reads the profile, opens the blobs, and populates
51        // the channel, then `REPLAY_THREADS` blocking threads consume it and prefetch pages.
52        let mut replay_threads = Vec::with_capacity(REPLAY_THREADS);
53        let page_size = zx::system_get_page_size() as u64;
54        for _ in 0..REPLAY_THREADS {
55            let receiver = receiver.clone();
56            replay_threads.push(fasync::unblock(move || {
57                while let Ok(request) = receiver.recv_blocking() {
58                    if let Err(error) =
59                        request.vmo.op_range(zx::VmoOp::PREFETCH, request.offset, page_size)
60                    {
61                        log::warn!(
62                            error:?,
63                            offset = request.offset;
64                            "Failed to prefetch blob page during profile replay"
65                        );
66                    }
67                    // `sender.close()` below closes the channel while keeping `sender` alive
68                    // (`sender_count() == 1`) so replay threads finish draining queued requests.
69                    // If `ReplayHandle` is dropped early on abort, `_cache_task` is dropped along
70                    // with `sender` (`sender_count() == 0`), causing replay threads to exit here.
71                    if receiver.sender_count() == 0 {
72                        return;
73                    }
74                }
75            }));
76        }
77        let replay_threads = Box::pin(async move {
78            join_all(replay_threads).await;
79        });
80
81        let cache_clone = cache.clone();
82        let cache_task = cache
83            .scope
84            .spawn(async move {
85                // Hold the blobs in cache until replay is stopped. Optional as None indicates
86                // that the blob could not be opened, and we want to cache that failure.
87                let mut local_cache: HashMap<Hash, Option<Arc<zx::Vmo>>> = HashMap::new();
88                let mut prefetch_requests: usize = 0;
89                for msg in messages {
90                    if msg.is_open_marker() {
91                        continue;
92                    }
93                    let vmo = match local_cache.entry(msg.id) {
94                        hash_map::Entry::Occupied(entry) => match entry.get() {
95                            Some(vmo) => vmo.clone(),
96                            // Found a cached error.
97                            None => continue,
98                        },
99                        hash_map::Entry::Vacant(entry) => {
100                            match cache_clone.open_vmo(&msg.id.into()).await {
101                                Ok(vmo) => {
102                                    let vmo = Arc::new(vmo);
103                                    entry.insert(Some(vmo.clone()));
104                                    vmo
105                                }
106                                Err(error) => {
107                                    log::debug!(
108                                        error:?,
109                                        hash:? = msg.id;
110                                        "Failed to open blob from profile replay"
111                                    );
112                                    // Cache the error.
113                                    entry.insert(None);
114                                    continue;
115                                }
116                            }
117                        }
118                    };
119                    if let Err(error) = sender.send(ReplayRequest { vmo, offset: msg.offset }).await
120                    {
121                        log::warn!(error:?; "Failed to send request to replay thread");
122                        break;
123                    }
124                    prefetch_requests += 1;
125                }
126                drop(cache_clone);
127                sender.close();
128                let opened_blobs = local_cache.values().filter(|entry| entry.is_some()).count();
129                log::info!(
130                    opened_blobs,
131                    total_blobs = local_cache.len(),
132                    prefetch_requests;
133                    "Queued blob profile replay requests"
134                );
135                // Keep `local_cache` alive so replayed blobs stay cached until replay is stopped.
136                let () = std::future::pending().await;
137            })
138            .into();
139
140        Ok(Self { replay_threads, _cache_task: cache_task })
141    }
142}
143
144/// An active profile recording and replay session driven by [`crate::BlobPagerAndVerifier`].
145///
146/// If dropped without calling [`ProfileRecording::stop`], any in-flight replay and driver
147/// recording are automatically aborted without writing a profile to the filesystem.
148pub struct ProfileRecording {
149    name: String,
150    cache: Arc<Cache>,
151    recording_proxy: Option<fblock::RecordingProxy>,
152    replay_handle: Option<ReplayHandle>,
153}
154
155impl std::fmt::Debug for ProfileRecording {
156    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
157        f.debug_struct("ProfileRecording")
158            .field("name", &self.name)
159            .field("has_replay", &self.replay_handle.is_some())
160            .finish_non_exhaustive()
161    }
162}
163
164impl ProfileRecording {
165    pub(crate) async fn start(
166        cache: Arc<Cache>,
167        mapper_session: &fblock::MapperSessionProxy,
168        name: String,
169    ) -> Result<Self, zx::Status> {
170        cache.start_recording()?;
171        // Construct `Self` immediately so `Drop` calls `self.cache.abort_recording()` if any
172        // subsequent step fails or if the `start` future is cancelled at an `.await` point.
173        let mut this = Self { name, cache, recording_proxy: None, replay_handle: None };
174
175        let (recording_proxy, recording_server) =
176            fidl::endpoints::create_proxy::<fblock::RecordingMarker>();
177        match mapper_session.record_profile(recording_server).await {
178            Ok(Ok(())) => {}
179            Ok(Err(raw_status)) => return Err(zx::Status::err_from_raw(raw_status)),
180            Err(error) => {
181                log::warn!(error:?; "FIDL error calling MapperSession.RecordProfile");
182                return Err(zx::Status::INTERNAL);
183            }
184        }
185        this.recording_proxy = Some(recording_proxy);
186
187        let replay_vmo = match this.cache.mapping_session.get_profile(&this.name).await {
188            Ok(Ok(vmo)) => Some(vmo),
189            Ok(Err(raw_status)) => {
190                let status = zx::Status::err_from_raw(raw_status);
191                if status == zx::Status::NOT_FOUND {
192                    None
193                } else {
194                    log::warn!(
195                        status:?,
196                        name = this.name.as_str();
197                        "MappingSession.GetProfile failed"
198                    );
199                    return Err(status);
200                }
201            }
202            Err(error) => {
203                log::warn!(
204                    error:?,
205                    name = this.name.as_str();
206                    "FIDL error calling MappingSession.GetProfile"
207                );
208                return Err(zx::Status::INTERNAL);
209            }
210        };
211
212        if let Some(vmo) = replay_vmo {
213            this.replay_handle = Some(ReplayHandle::start(&this.cache, vmo)?);
214        }
215
216        Ok(this)
217    }
218
219    /// Stops any active replay, finalizes the driver recording, filters by blobs opened during the
220    /// recording window, and writes the encoded profile to the filesystem via `MappingSession`.
221    pub async fn stop(mut self) -> Result<(), zx::Status> {
222        if let Some(replay_handle) = self.replay_handle.take() {
223            replay_handle.replay_threads.await;
224        }
225
226        let recording_proxy = self.recording_proxy.take().ok_or(zx::Status::BAD_STATE)?;
227        let stop_res = recording_proxy.stop().await;
228        let active_recording = self.cache.take_recording().ok_or(zx::Status::BAD_STATE)?;
229
230        let driver_vmo = match stop_res {
231            Ok(Ok(vmo)) => vmo,
232            Ok(Err(raw_status)) => return Err(zx::Status::err_from_raw(raw_status)),
233            Err(error) => {
234                log::warn!(error:?; "FIDL error calling Recording.Stop");
235                return Err(zx::Status::INTERNAL);
236            }
237        };
238
239        let file_messages =
240            FileMessage::decode_from_vmo(driver_vmo, PROFILE_BLOCK_SIZE).map_err(|error| {
241                if error == zx::Status::OUT_OF_RANGE {
242                    log::warn!("Driver profile VMO exceeds maximum allowed size");
243                    zx::Status::OUT_OF_RANGE
244                } else {
245                    log::warn!(error:?; "Failed to decode driver profile VMO");
246                    zx::Status::IO_DATA_INTEGRITY
247                }
248            })?;
249
250        let opened_blobs = active_recording.opened_hashes.len();
251        let mut seen = HashSet::new();
252        let blob_messages = file_messages
253            .filter_map(|msg| {
254                let &hash = active_recording.hash_by_key.get(&msg.id)?;
255                active_recording
256                    .opened_hashes
257                    .contains(&hash)
258                    .then_some(BlobMessage { id: hash, offset: msg.offset })
259            })
260            .filter(|&blob_msg| seen.insert(blob_msg));
261
262        let recorded_vmo = BlobMessage::encode_to_vmo(blob_messages, PROFILE_BLOCK_SIZE)?;
263        let recorded_entries = seen.len();
264
265        match self.cache.mapping_session.write_profile(&self.name, recorded_vmo).await {
266            Ok(Ok(())) => {
267                log::info!(
268                    opened_blobs,
269                    recorded_entries;
270                    "Committed blob profile recording"
271                );
272                Ok(())
273            }
274            Ok(Err(raw_status)) => {
275                let status = zx::Status::err_from_raw(raw_status);
276                log::warn!(
277                    status:?,
278                    name = self.name.as_str();
279                    "MappingSession.WriteProfile failed"
280                );
281                Err(status)
282            }
283            Err(error) => {
284                log::warn!(
285                    error:?,
286                    name = self.name.as_str();
287                    "FIDL error calling MappingSession.WriteProfile"
288                );
289                Err(zx::Status::INTERNAL)
290            }
291        }
292    }
293}
294
295impl Drop for ProfileRecording {
296    fn drop(&mut self) {
297        self.cache.abort_recording();
298    }
299}
300
301#[cfg(test)]
302mod tests {
303    use super::*;
304    use crate::BlobPagerAndVerifier;
305    use fidl_fuchsia_storage_mapping as fmapping;
306    use fuchsia_sync::Mutex;
307    use futures::channel::oneshot;
308    use futures::{StreamExt, TryStreamExt};
309    use mapping::{DeliveryCommand, RawDeliveryCommand};
310    use vmo_fifo::SyncSender;
311
312    #[fuchsia::test]
313    async fn test_profile_recorder_replay_filtering_and_eviction() {
314        let (mapping_proxy, mut mapping_stream) =
315            fidl::endpoints::create_proxy_and_stream::<fmapping::MappingProviderMarker>();
316        let (mapper_proxy, mut mapper_stream) =
317            fidl::endpoints::create_proxy_and_stream::<fblock::MapperMarker>();
318
319        let blob_data = vec![0x42u8; 8192];
320        let (root1, _) = fuchsia_merkle::MerkleRootBuilder::new(Vec::new()).complete(&blob_data);
321        let hash1: [u8; 32] = root1.into();
322        let blob_data2 = vec![0x24u8; 8192];
323        let (root2, _) = fuchsia_merkle::MerkleRootBuilder::new(Vec::new()).complete(&blob_data2);
324        let hash2: [u8; 32] = root2.into();
325        let blob_data3 = vec![0x99u8; 8192];
326        let (root3, _) = fuchsia_merkle::MerkleRootBuilder::new(Vec::new()).complete(&blob_data3);
327        let hash3: [u8; 32] = root3.into();
328        let missing_hash: [u8; 32] = [0xde; 32];
329
330        // Replay profile contains `missing_hash` (twice, to verify negative lookup caching),
331        // `hash1` (which a client will also open), and `hash2` (replay-only, never opened by a
332        // client).
333        let replay_entries = vec![
334            BlobMessage { id: Hash::from(missing_hash), offset: 0 },
335            BlobMessage { id: Hash::from(missing_hash), offset: 4096 },
336            BlobMessage { id: Hash::from(hash1), offset: 0 },
337            BlobMessage { id: Hash::from(hash2), offset: 0 },
338        ];
339        let next_replay_vmo = Arc::new(Mutex::new(Some(
340            BlobMessage::encode_to_vmo(&replay_entries, PROFILE_BLOCK_SIZE)
341                .expect("encode_to_vmo failed"),
342        )));
343        let next_replay_vmo_for_server = next_replay_vmo.clone();
344
345        let (written_tx, written_rx) = oneshot::channel();
346        let (close_tx, mut close_rx) = futures::channel::mpsc::unbounded();
347        let opened_keys = Arc::new(Mutex::new(HashMap::<[u8; 32], Vec<u64>>::new()));
348        let opened_keys_clone = opened_keys.clone();
349
350        let mapping_task = fasync::Task::spawn(async move {
351            let mut written_tx = Some(written_tx);
352            if let Some(fmapping::MappingProviderRequest::OpenSession { session, responder }) =
353                mapping_stream.try_next().await.expect("try_next failed")
354            {
355                let mapping_vmo = zx::Vmo::create(zx::system_get_page_size().into())
356                    .expect("zx::Vmo::create failed");
357                responder.send(Ok(mapping_vmo)).expect("send failed");
358
359                let mut session_stream = session.into_stream();
360                while let Some(request) = session_stream.try_next().await.expect("try_next failed")
361                {
362                    match request {
363                        fmapping::MappingSessionRequest::GetProfile { name, responder } => {
364                            assert_eq!(name, ".boot");
365                            match next_replay_vmo_for_server.lock().take() {
366                                Some(vmo) => responder.send(Ok(vmo)).expect("send failed"),
367                                None => responder
368                                    .send(Err(zx::Status::NOT_FOUND.into_raw()))
369                                    .expect("send failed"),
370                            }
371                        }
372                        fmapping::MappingSessionRequest::WriteProfile {
373                            name,
374                            profile_vmo,
375                            responder,
376                        } => {
377                            assert_eq!(name, ".boot");
378                            if let Some(tx) = written_tx.take() {
379                                let _ = tx.send(profile_vmo);
380                            }
381                            responder.send(Ok(())).expect("send failed");
382                        }
383                        fmapping::MappingSessionRequest::Open { key, identifier, responder } => {
384                            let arr: [u8; 32] =
385                                identifier.as_slice().try_into().expect("invalid hash slice");
386                            opened_keys_clone.lock().entry(arr).or_default().push(key);
387                            if arr == missing_hash {
388                                responder
389                                    .send(Err(zx::Status::NOT_FOUND.into_raw()))
390                                    .expect("send failed");
391                            } else {
392                                responder.send(Ok(8192)).expect("send failed");
393                            }
394                        }
395                        fmapping::MappingSessionRequest::Close { key, responder } => {
396                            responder.send(Ok(())).expect("send failed");
397                            let _ = close_tx.unbounded_send(key);
398                        }
399                        _ => {}
400                    }
401                }
402            }
403        });
404
405        let (tx_delivery, rx_delivery) = oneshot::channel();
406        let (recording_dropped_tx, recording_dropped_rx) = oneshot::channel();
407        let opened_keys_for_mapper = opened_keys.clone();
408        let mapper_task = fasync::Task::spawn(async move {
409            let mut recording_dropped_tx = Some(recording_dropped_tx);
410            if let Some(fblock::MapperRequest::OpenSession {
411                session,
412                port,
413                delivery_queue,
414                responder,
415                ..
416            }) = mapper_stream.try_next().await.expect("try_next failed")
417            {
418                let port = port.expect("expected port");
419                let delivery_vmo = delivery_queue.expect("expected delivery_queue");
420                tx_delivery
421                    .send(
422                        delivery_vmo
423                            .duplicate_handle(zx::Rights::SAME_RIGHTS)
424                            .expect("duplicate_handle failed"),
425                    )
426                    .expect("send failed");
427                responder.send(Ok(())).expect("send failed");
428
429                // Spawn a mock pager thread on `port` that supplies pages when `ZX_VMO_OP_PREFETCH`
430                // triggers a page request during replay.
431                let opened_keys_for_pager = opened_keys_for_mapper.clone();
432                std::thread::spawn(move || {
433                    let mut sender = SyncSender::<RawDeliveryCommand>::new(
434                        delivery_vmo,
435                        zx::system_get_page_size() as usize,
436                        mapping::PENDING_DELIVERY_COMMANDS_CAPACITY,
437                    )
438                    .expect("SyncSender::new failed");
439                    while let Ok(packet) = port.wait(zx::MonotonicInstant::INFINITE) {
440                        if let zx::PacketContents::Pager(pager_packet) = packet.contents() {
441                            if pager_packet.command()
442                                == zx::sys::zx_page_request_command_t::ZX_PAGER_VMO_READ
443                            {
444                                let key = packet.key();
445                                let range = pager_packet.range();
446                                let reg_payload =
447                                    sender.reserve_payload(0).expect("reserve_payload failed");
448                                let reg_cmd: RawDeliveryCommand = DeliveryCommand::RegisterBlob {
449                                    key,
450                                    offset: reg_payload.offset(),
451                                    length: 0,
452                                    read_alignment: 8192,
453                                }
454                                .into();
455                                reg_payload.commit(reg_cmd).expect("commit failed");
456
457                                let data = if opened_keys_for_pager
458                                    .lock()
459                                    .get(&hash1)
460                                    .is_some_and(|ks| ks.contains(&key))
461                                {
462                                    &blob_data
463                                } else {
464                                    &blob_data2
465                                };
466                                let mut payload = sender
467                                    .reserve_payload(data.len())
468                                    .expect("reserve_payload failed");
469                                payload.data().copy_from_slice(data);
470                                let cmd: RawDeliveryCommand = DeliveryCommand::Data {
471                                    key,
472                                    offset: payload.offset(),
473                                    length: data.len() as u32,
474                                    target_offset: range.start,
475                                }
476                                .into();
477                                payload.commit(cmd).expect("commit failed");
478                            }
479                        }
480                    }
481                });
482
483                let mut session_stream = session.into_stream();
484                while let Some(request) = session_stream.try_next().await.expect("try_next failed")
485                {
486                    if let fblock::MapperSessionRequest::RecordProfile { recording, responder } =
487                        request
488                    {
489                        responder.send(Ok(())).expect("send failed");
490                        let mut rec_stream = recording.into_stream();
491                        match rec_stream.try_next().await.expect("try_next failed") {
492                            Some(fblock::RecordingRequest::Stop { responder }) => {
493                                let keys = opened_keys_for_mapper.lock().clone();
494                                let k1 = keys.get(&hash1).expect("hash1 keys missing")[0];
495                                let k2 = keys.get(&hash2).expect("hash2 keys missing")[0];
496                                let k3_first = keys.get(&hash3).expect("hash3 keys missing")[0];
497                                let k3_second = keys.get(&hash3).expect("hash3 keys missing")[1];
498                                let driver_entries = vec![
499                                    FileMessage { id: k1, offset: 0 },
500                                    FileMessage { id: k2, offset: 0 },
501                                    FileMessage { id: k3_first, offset: 0 },
502                                    // Duplicate `(hash3, 0)` under the reopened key `k3_second`
503                                    FileMessage { id: k3_second, offset: 0 },
504                                    FileMessage { id: k3_second, offset: 4096 },
505                                ];
506                                let vmo =
507                                    FileMessage::encode_to_vmo(&driver_entries, PROFILE_BLOCK_SIZE)
508                                        .expect("encode_to_vmo failed");
509                                responder.send(Ok(vmo)).expect("send failed");
510                            }
511                            None => {
512                                if let Some(tx) = recording_dropped_tx.take() {
513                                    let _ = tx.send(());
514                                }
515                            }
516                            _ => {}
517                        }
518                    }
519                }
520            }
521        });
522
523        let pager_and_verifier = Arc::new(
524            BlobPagerAndVerifier::new_with_cache_capacity(&mapping_proxy, &mapper_proxy, 0)
525                .await
526                .expect("BlobPagerAndVerifier::new_with_cache_capacity failed"),
527        );
528        let _delivery_vmo = rx_delivery.await.expect("rx_delivery failed");
529
530        let recording =
531            pager_and_verifier.start_profile(".boot").await.expect("start_profile failed");
532
533        // Concurrent `start_profile` while recording is active must fail with `BAD_STATE`.
534        let err = pager_and_verifier
535            .start_profile(".boot")
536            .await
537            .expect_err("Concurrent start_profile succeeded");
538        assert_eq!(err, zx::Status::BAD_STATE);
539
540        // Wait for the failed open of `missing_hash` to clean up its key reservation via `Close`.
541        let missing_closed_key = close_rx.next().await.expect("expected close for missing_hash");
542        assert_eq!(
543            missing_closed_key,
544            opened_keys.lock().get(&missing_hash).expect("missing_hash key missing")[0]
545        );
546
547        // Client opens `hash1` and `hash3` (leaving `hash2` unopened by any client).
548        let vmo1 = pager_and_verifier.create_vmo(&hash1).await.expect("create_vmo hash1 failed");
549        let vmo3 = pager_and_verifier.create_vmo(&hash3).await.expect("create_vmo hash3 failed");
550
551        // Evict `hash3` and re-open it before `stop()` to verify `hash_by_key` retains evicted
552        // keys and deduplicates `(hash3, 0)` across keys.
553        drop(vmo3);
554        let _ = close_rx.next().await.expect("expected close for first vmo3");
555        let vmo3_reopened =
556            pager_and_verifier.create_vmo(&hash3).await.expect("re-open hash3 failed");
557        drop(vmo3_reopened);
558        let _ = close_rx.next().await.expect("expected close for second vmo3");
559
560        // Stop recording: this waits for replay to finish, unpins `hash2` (causing its eviction),
561        // filters out `hash2`, retains `hash1` and `hash3`, deduplicates `(hash3, 0)`, and writes
562        // the profile VMO via `MappingSession.WriteProfile`.
563        recording.stop().await.expect("recording.stop failed");
564        let recorded_vmo = written_rx.await.expect("written_rx failed");
565
566        // Verify negative lookup caching: `missing_hash` appeared twice in `replay_entries`, so
567        // `MappingSession.Open` must have been called only once for it.
568        assert_eq!(
569            opened_keys.lock().get(&missing_hash).expect("missing_hash key missing").len(),
570            1
571        );
572
573        // `hash2` was only opened by replay and must be evicted upon `stop()`.
574        let closed_key = close_rx.next().await.expect("expected close for unpinned replay blob");
575        assert_eq!(closed_key, opened_keys.lock().get(&hash2).expect("hash2 key missing")[0]);
576
577        let decoded: Vec<_> = BlobMessage::decode_from_vmo(recorded_vmo, PROFILE_BLOCK_SIZE)
578            .expect("decode_from_vmo failed")
579            .collect();
580        assert_eq!(
581            decoded,
582            vec![
583                BlobMessage { id: Hash::from(hash1), offset: 0 },
584                BlobMessage { id: Hash::from(hash3), offset: 0 },
585                BlobMessage { id: Hash::from(hash3), offset: 4096 },
586            ]
587        );
588
589        // Start a second recording and drop `ProfileRecording` without calling `stop()` to verify
590        // that the driver recording is aborted and replay-pinned blobs are unpinned.
591        *next_replay_vmo.lock() = Some(
592            BlobMessage::encode_to_vmo(
593                &[BlobMessage { id: Hash::from(hash2), offset: 0 }],
594                PROFILE_BLOCK_SIZE,
595            )
596            .expect("encode_to_vmo failed"),
597        );
598        let recording2 =
599            pager_and_verifier.start_profile(".boot").await.expect("second start_profile failed");
600        drop(recording2);
601        recording_dropped_rx.await.expect("expected driver Recording channel to close on abort");
602
603        drop(vmo1);
604        let _ = close_rx.next().await.expect("expected close for vmo1 or aborted replay");
605        drop(pager_and_verifier);
606        mapping_task.await;
607        mapper_task.await;
608    }
609}
blob_pager_and_verifier/profile.rs

blob_pager_and_verifier/
profile.rs
