zencodec_coroutine/
context.rs

1//! The codec author's API: read bytes, write rows, announce metadata.
2
3use alloc::sync::Arc;
4use alloc::vec::Vec;
5use core::fmt;
6use core::future::Future;
7use core::pin::Pin;
8use core::task::{Context, Poll};
9
10use enough::StopReason;
11use zencodec::ImageInfo;
12use zencodec::decode::{
13    BatchMeta, BoundaryAction, BoundaryId, ByteRange, ControlError, DecodeBoundary, InputRequest,
14    InterpretationFacts, MetadataUpdate, SinkError,
15};
16use zencodec::source::SourceError;
17use zencodec::storage::{OwnedStorage, StorageBudget, StorageError};
18use zenpixels::{PixelDescriptor, PixelSlice};
19
20use crate::input::InputStore;
21
22/// Failure inside a coroutine decode, convertible into the codec's error.
23#[derive(Debug)]
24#[non_exhaustive]
25pub enum CoroutineError {
26    /// Input payload, range index or row staging could not be admitted.
27    Storage(StorageError),
28    /// An absolute byte range overflowed the source address space.
29    InputRangeOverflow,
30    /// The admitted input range index cannot represent another disjoint span.
31    InputSpanLimit {
32        /// Maximum number of independently retained spans.
33        limit: usize,
34    },
35    /// A required interpretation boundary was violated.
36    Control(ControlError),
37    /// A row buffer size overflowed or its allocation failed.
38    RowAllocation,
39    /// A read could not be satisfied (past the announced end).
40    Source(SourceError),
41    /// The sink refused rows.
42    Sink(SinkError),
43    /// The driver's stop token tripped.
44    Cancelled(StopReason),
45    /// The decoder was stepped after it finished or failed.
46    Finished,
47    /// The coroutine suspended without asking for anything (author bug).
48    Stalled,
49    /// The coroutine wrote rows before setting the image info and output
50    /// descriptor, so the sink could not be told the geometry (author bug).
51    RowsBeforeHeader,
52    /// One read asked for more bytes than the decoder's input cap can hold,
53    /// so it could never complete.
54    ReadExceedsInputCap {
55        /// Bytes the read needs.
56        len: usize,
57        /// The configured cap.
58        cap: usize,
59    },
60}
61
62impl fmt::Display for CoroutineError {
63    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
64        match self {
65            Self::Storage(e) => e.fmt(f),
66            Self::InputRangeOverflow => f.write_str("coroutine input range overflow"),
67            Self::InputSpanLimit { limit } => {
68                write!(f, "coroutine input exceeds its {limit}-span index")
69            }
70            Self::Control(e) => e.fmt(f),
71            Self::RowAllocation => f.write_str("coroutine row allocation failed"),
72            Self::Source(e) => write!(f, "coroutine decode: {e}"),
73            Self::Sink(e) => write!(f, "coroutine decode: sink: {e}"),
74            Self::Cancelled(r) => write!(f, "coroutine decode cancelled: {r:?}"),
75            Self::Finished => write!(f, "coroutine decoder stepped after it finished"),
76            Self::Stalled => write!(f, "coroutine suspended without a request"),
77            Self::ReadExceedsInputCap { len, cap } => write!(
78                f,
79                "coroutine read of {len} bytes exceeds its {cap}-byte input cap"
80            ),
81            Self::RowsBeforeHeader => write!(
82                f,
83                "coroutine wrote rows before set_info and set_output_descriptor"
84            ),
85        }
86    }
87}
88
89impl core::error::Error for CoroutineError {
90    fn source(&self) -> Option<&(dyn core::error::Error + 'static)> {
91        match self {
92            Self::Storage(e) => Some(e),
93            Self::Control(e) => Some(e),
94            Self::Source(e) => Some(e),
95            Self::Sink(e) => Some(e.as_ref()),
96            _ => None,
97        }
98    }
99}
100
101impl From<StorageError> for CoroutineError {
102    fn from(e: StorageError) -> Self {
103        Self::Storage(e)
104    }
105}
106
107pub(crate) enum RowStorage {
108    Legacy(Vec<u8>),
109    Admitted(OwnedStorage<u8>),
110}
111impl Default for RowStorage {
112    fn default() -> Self {
113        Self::Legacy(Vec::new())
114    }
115}
116impl RowStorage {
117    fn prepare(
118        &mut self,
119        len: usize,
120        budget: Option<&StorageBudget>,
121    ) -> Result<(), CoroutineError> {
122        if let Some(budget) = budget {
123            if matches!(self, Self::Admitted(data) if data.len() >= len) {
124                return Ok(());
125            }
126            // An unused spare is not live output. Release it before growing so
127            // growth does not require two simultaneous staging allocations.
128            *self = Self::default();
129            *self = Self::Admitted(budget.bytes(len)?);
130        } else {
131            let Self::Legacy(data) = self else {
132                unreachable!("storage mode is fixed at construction")
133            };
134            data.clear();
135            data.try_reserve_exact(len)
136                .map_err(|_| CoroutineError::RowAllocation)?;
137            data.resize(len, 0);
138        }
139        Ok(())
140    }
141    pub(crate) fn as_slice(&self) -> &[u8] {
142        match self {
143            Self::Legacy(data) => data,
144            Self::Admitted(data) => data.as_slice(),
145        }
146    }
147    fn as_mut_slice(&mut self) -> &mut [u8] {
148        match self {
149            Self::Legacy(data) => data,
150            Self::Admitted(data) => data.as_mut_slice(),
151        }
152    }
153}
154
155impl From<SourceError> for CoroutineError {
156    fn from(e: SourceError) -> Self {
157        Self::Source(e)
158    }
159}
160impl From<ControlError> for CoroutineError {
161    fn from(error: ControlError) -> Self {
162        Self::Control(error)
163    }
164}
165
166impl From<SinkError> for CoroutineError {
167    fn from(e: SinkError) -> Self {
168        Self::Sink(e)
169    }
170}
171
172impl From<StopReason> for CoroutineError {
173    fn from(r: StopReason) -> Self {
174        Self::Cancelled(r)
175    }
176}
177
178/// Rows the coroutine handed over, waiting for the decoder to copy them
179/// into the sink.
180pub(crate) struct PendingRows {
181    pub(crate) meta: BatchMeta,
182    pub(crate) data: RowStorage,
183    pub(crate) width: u32,
184    pub(crate) rows: u32,
185    pub(crate) stride: usize,
186    pub(crate) descriptor: PixelDescriptor,
187}
188
189/// State shared between the decoder and the coroutine it polls. Only one
190/// side runs at a time, so the lock is never contended.
191pub(crate) struct Shared {
192    pub(crate) controlled: bool,
193    pub(crate) boundary: Option<DecodeBoundary>,
194    pub(crate) reply: Option<(BoundaryId, BoundaryAction)>,
195    pub(crate) adopted: Option<InterpretationFacts>,
196    pub(crate) control_failure: Option<ControlError>,
197    pub(crate) input: InputStore,
198    pub(crate) need: Vec<InputRequest>,
199    /// Scratch for a read's satisfied spans and missing gaps, reused across polls.
200    pub(crate) satisfied: Vec<(u64, usize)>,
201    pub(crate) gaps: Vec<(u64, u64)>,
202    pub(crate) rows: Option<PendingRows>,
203    /// Strip buffer handed back after each `Rows` step, so steady state
204    /// allocates nothing.
205    pub(crate) spare: RowStorage,
206    pub(crate) storage: Option<StorageBudget>,
207    pub(crate) metadata: Option<MetadataUpdate>,
208    pub(crate) info: Option<ImageInfo>,
209    pub(crate) details: Option<zencodec::decode::DecodeDetails>,
210    /// Bumped on every `set_info`, so the decoder refreshes its cached copy.
211    pub(crate) info_generation: u64,
212    pub(crate) descriptor: Option<PixelDescriptor>,
213    pub(crate) accepts_provisional: bool,
214    pub(crate) cancelled: Option<StopReason>,
215}
216
217impl Shared {
218    pub(crate) fn new(cap: usize) -> Self {
219        Self {
220            controlled: false,
221            boundary: None,
222            reply: None,
223            adopted: None,
224            control_failure: None,
225            input: InputStore::new(cap),
226            need: Vec::new(),
227            satisfied: Vec::new(),
228            gaps: Vec::new(),
229            rows: None,
230            spare: RowStorage::default(),
231            storage: None,
232            metadata: None,
233            info: None,
234            details: None,
235            info_generation: 0,
236            descriptor: None,
237            accepts_provisional: false,
238            cancelled: None,
239        }
240    }
241
242    pub(crate) fn with_storage(cap: usize, budget: StorageBudget) -> Result<Self, CoroutineError> {
243        let input = InputStore::with_storage(cap, &budget)?;
244        Ok(Self {
245            input,
246            storage: Some(budget),
247            ..Self::new(cap)
248        })
249    }
250}
251
252pub(crate) type SharedCell = Arc<spin::Mutex<Shared>>;
253
254/// Handle a coroutine decode uses to read input and emit output.
255///
256/// Every `async fn` here suspends the coroutine when it cannot proceed and
257/// resumes it on the step after the driver supplied what was missing.
258pub struct CoroutineContext {
259    shared: SharedCell,
260    cursor: u64,
261}
262
263impl CoroutineContext {
264    pub(crate) fn new(shared: SharedCell) -> Self {
265        Self { shared, cursor: 0 }
266    }
267
268    /// Read exactly `out.len()` bytes at `offset`.
269    ///
270    /// Suspends with `Event::Need` (a `InputRequest::Range`) until the driver
271    /// offers the span. Fails with [`SourceError::UnexpectedEof`] once the
272    /// announced length shows the span can never arrive.
273    pub fn read_exact_at<'a>(
274        &'a mut self,
275        offset: u64,
276        out: &'a mut [u8],
277    ) -> impl Future<Output = Result<(), CoroutineError>> + 'a {
278        ReadMany {
279            shared: &self.shared,
280            reqs: ReqSlot::One(offset, out),
281            release: true,
282        }
283    }
284
285    /// Read several spans at once. The driver sees one `Event::Need` naming
286    /// every missing span, so a batching source fetches them in one round.
287    pub fn read_many_at<'a, 'b: 'a>(
288        &'a mut self,
289        reqs: &'a mut [(u64, &'b mut [u8])],
290    ) -> impl Future<Output = Result<(), CoroutineError>> + 'a {
291        ReadMany {
292            shared: &self.shared,
293            reqs: ReqSlot::Many(reqs),
294            release: true,
295        }
296    }
297
298    /// Read the next `out.len()` bytes at the sequential cursor and
299    /// advance it. Suspends with one `InputRequest::AtLeast` for the rest of the
300    /// read, which stream pumps serve by reading forward. Bytes behind the
301    /// cursor are released.
302    pub async fn read_exact(&mut self, out: &mut [u8]) -> Result<(), CoroutineError> {
303        let offset = self.cursor;
304        ReadMany {
305            shared: &self.shared,
306            reqs: ReqSlot::Sequential(offset, out),
307            release: true,
308        }
309        .await?;
310        self.cursor = offset + out.len() as u64;
311        self.shared.lock().input.release_before(self.cursor);
312        Ok(())
313    }
314
315    /// The sequential cursor used by [`read_exact`](Self::read_exact).
316    pub fn cursor(&self) -> u64 {
317        self.cursor
318    }
319
320    /// Move the sequential cursor (bytes behind it are released).
321    pub fn seek(&mut self, offset: u64) {
322        self.cursor = offset;
323        self.shared.lock().input.release_before(offset);
324    }
325
326    /// Total source length. Suspends with `InputRequest::Len` until a driver
327    /// offers an input carrying it.
328    pub fn len(&mut self) -> impl Future<Output = Result<u64, CoroutineError>> + '_ {
329        Len {
330            shared: &self.shared,
331        }
332    }
333
334    /// Hand a strip of rows to the driver. Suspends until the decoder has
335    /// copied it into the sink. Each `Event::Rows` delivers at most one buffer
336    /// allowed by the sink's `max_buffer_rows`; a larger batch stays retained
337    /// across steps until every row is delivered. This bounds queued sink
338    /// buffers, not the native batch's storage.
339    ///
340    /// Provisional rows (`!meta.is_final`) may be written only when
341    /// [`accepts_provisional`](Self::accepts_provisional) is `true`.
342    pub fn write_rows<'a>(
343        &'a mut self,
344        meta: BatchMeta,
345        rows: &PixelSlice<'_>,
346    ) -> impl Future<Output = Result<(), CoroutineError>> + 'a {
347        let refusal = {
348            let shared = self.shared.lock();
349            shared.control_failure.or_else(|| {
350                (shared.controlled && (shared.adopted.is_none() || shared.boundary.is_some()))
351                    .then_some(ControlError::OutputBeforeInterpretation)
352            })
353        };
354        if let Some(error) = refusal {
355            return WriteRows {
356                shared: &self.shared,
357                pending: None,
358                error: Some(error.into()),
359            };
360        }
361        let width = rows.width();
362        let count = rows.rows();
363        let descriptor = rows.descriptor();
364        let size = (width as usize)
365            .checked_mul(descriptor.bytes_per_pixel())
366            .and_then(|stride| stride.checked_mul(count as usize).map(|len| (stride, len)));
367        let Some((stride, len)) = size else {
368            return WriteRows {
369                shared: &self.shared,
370                pending: None,
371                error: Some(CoroutineError::RowAllocation),
372            };
373        };
374        let (mut data, prepared) = {
375            let mut shared = self.shared.lock();
376            let mut data = core::mem::take(&mut shared.spare);
377            let prepared = data.prepare(len, shared.storage.as_ref());
378            (data, prepared)
379        };
380        if let Err(error) = prepared {
381            return WriteRows {
382                shared: &self.shared,
383                pending: None,
384                error: Some(error),
385            };
386        }
387        for r in 0..count {
388            let at = r as usize * stride;
389            data.as_mut_slice()[at..at + stride].copy_from_slice(&rows.row(r)[..stride]);
390        }
391        WriteRows {
392            shared: &self.shared,
393            error: None,
394            pending: Some(PendingRows {
395                meta,
396                data,
397                width,
398                rows: count,
399                stride,
400                descriptor,
401            }),
402        }
403    }
404
405    /// Report a metadata event to the driver (one `Event::Metadata`).
406    /// `HeaderReady` with dimensions and descriptor set lets the decoder
407    /// call `sink.begin` early.
408    pub fn announce(
409        &mut self,
410        update: MetadataUpdate,
411    ) -> impl Future<Output = Result<(), CoroutineError>> + '_ {
412        Announce {
413            shared: &self.shared,
414            update: Some(update),
415        }
416    }
417
418    /// Set image metadata (visible through `IncrementalDecode::info`).
419    pub fn set_info(&mut self, info: ImageInfo) {
420        let mut shared = self.shared.lock();
421        if shared.adopted.is_some()
422            && shared.info.as_ref().is_some_and(|old| {
423                old.width != info.width
424                    || old.height != info.height
425                    || old.orientation != info.orientation
426                    || old.source_color != info.source_color
427            })
428        {
429            shared.control_failure = Some(ControlError::FinalInterpretationChanged);
430        }
431        shared.info = Some(info);
432        shared.info_generation += 1;
433    }
434
435    /// Supply the owned result details for terminal extraction.
436    ///
437    /// These remain private until the future and sink finish successfully.
438    /// Replacing a pending value drops its old owners. Failure or cancellation
439    /// discards pending details; `take_details` never returns a partial result.
440    /// This is not a metadata-completeness or privacy certificate. The producer
441    /// must include every result detail promised by its configured operation.
442    pub fn set_details(&mut self, details: zencodec::decode::DecodeDetails) {
443        let previous = {
444            let mut shared = self.shared.lock();
445            if shared.adopted.is_some()
446                && shared.info.as_ref().is_some_and(|old| {
447                    old.width != details.info.width
448                        || old.height != details.info.height
449                        || old.orientation != details.info.orientation
450                        || old.source_color != details.info.source_color
451                })
452            {
453                shared.control_failure = Some(ControlError::FinalInterpretationChanged);
454            }
455            shared.details.replace(details)
456        };
457        // Extension destructors run outside the shared-state lock.
458        drop(previous);
459    }
460
461    /// Set the negotiated output descriptor.
462    pub fn set_output_descriptor(&mut self, descriptor: PixelDescriptor) {
463        let mut shared = self.shared.lock();
464        if shared.adopted.is_some() && shared.descriptor != Some(descriptor) {
465            shared.control_failure = Some(ControlError::FinalInterpretationChanged);
466        }
467        shared.descriptor = Some(descriptor);
468    }
469
470    /// Suspend before work affected by these interpretation facts. Await this
471    /// before conversion or output. `ContinueReading` permits discovery only;
472    /// rows stay blocked until final interpretation is adopted. This future is
473    /// served only by the controlled protocol; legacy execution refuses it.
474    pub fn interpretation(
475        &mut self,
476        facts: InterpretationFacts,
477    ) -> impl Future<Output = Result<BoundaryAction, CoroutineError>> + '_ {
478        Interpret {
479            shared: &self.shared,
480            facts: Some(facts),
481            id: None,
482        }
483    }
484
485    /// Whether the driver's sink takes provisional (progressive) rows.
486    /// Known once the header has been announced (the decoder reads it when
487    /// it calls the sink's `begin`); `false` before that.
488    pub fn accepts_provisional(&self) -> bool {
489        self.shared.lock().accepts_provisional
490    }
491
492    /// Check the driver's stop token without suspending.
493    pub fn check_stop(&self) -> Result<(), CoroutineError> {
494        let shared = self.shared.lock();
495        if let Some(error) = shared.control_failure {
496            return Err(error.into());
497        }
498        match shared.cancelled {
499            Some(r) => Err(CoroutineError::Cancelled(r)),
500            None => Ok(()),
501        }
502    }
503}
504
505struct Interpret<'a> {
506    shared: &'a SharedCell,
507    facts: Option<InterpretationFacts>,
508    id: Option<BoundaryId>,
509}
510impl Future for Interpret<'_> {
511    type Output = Result<BoundaryAction, CoroutineError>;
512    fn poll(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Self::Output> {
513        let this = self.get_mut();
514        let mut shared = this.shared.lock();
515        if let Some(reason) = shared.cancelled {
516            return Poll::Ready(Err(reason.into()));
517        }
518        if let Some(error) = shared.control_failure {
519            return Poll::Ready(Err(error.into()));
520        }
521        if !shared.controlled {
522            return Poll::Ready(Err(ControlError::UnhandledBoundary.into()));
523        }
524        if let Some(facts) = this.facts.take() {
525            if shared.boundary.is_some() || shared.reply.is_some() {
526                return Poll::Ready(Err(ControlError::InvalidReply.into()));
527            }
528            if let Some(adopted) = shared.adopted {
529                return Poll::Ready(if adopted == facts {
530                    Ok(BoundaryAction::AdoptInterpretation)
531                } else {
532                    Err(ControlError::FinalInterpretationChanged.into())
533                });
534            }
535            if facts.descriptor != shared.descriptor
536                || shared
537                    .info
538                    .as_ref()
539                    .is_none_or(|info| info.width != facts.width || info.height != facts.height)
540            {
541                return Poll::Ready(Err(ControlError::InconsistentFacts.into()));
542            }
543            let boundary = DecodeBoundary::new(facts)?;
544            this.id = Some(boundary.id());
545            shared.boundary = Some(boundary);
546            return Poll::Pending;
547        }
548        match shared.reply.take() {
549            Some((id, action)) if Some(id) == this.id => Poll::Ready(Ok(action)),
550            Some(_) => Poll::Ready(Err(ControlError::InvalidReply.into())),
551            None => Poll::Pending,
552        }
553    }
554}
555
556enum ReqSlot<'a, 'b> {
557    One(u64, &'a mut [u8]),
558    Sequential(u64, &'a mut [u8]),
559    Many(&'a mut [(u64, &'b mut [u8])]),
560}
561
562struct ReadMany<'a, 'b> {
563    shared: &'a SharedCell,
564    reqs: ReqSlot<'a, 'b>,
565    release: bool,
566}
567
568impl Future for ReadMany<'_, '_> {
569    type Output = Result<(), CoroutineError>;
570
571    fn poll(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Self::Output> {
572        let this = self.get_mut();
573        let mut shared = this.shared.lock();
574        if let Some(r) = shared.cancelled {
575            return Poll::Ready(Err(CoroutineError::Cancelled(r)));
576        }
577        // Reuse the shared scratch vectors: `need` collects the gaps and
578        // `satisfied` the spans to release, so a poll allocates nothing.
579        let mut missing = core::mem::take(&mut shared.need);
580        let mut satisfied = core::mem::take(&mut shared.satisfied);
581        let mut gaps = core::mem::take(&mut shared.gaps);
582        missing.clear();
583        satisfied.clear();
584        gaps.clear();
585        let mut check = |shared: &Shared, offset: u64, out: &mut [u8], sequential: bool| {
586            offset
587                .checked_add(out.len() as u64)
588                .ok_or(CoroutineError::InputRangeOverflow)?;
589            if out.len() > shared.input.cap() {
590                return Err(CoroutineError::ReadExceedsInputCap {
591                    len: out.len(),
592                    cap: shared.input.cap(),
593                });
594            }
595            if shared.input.try_read(offset, out) {
596                satisfied.push((offset, out.len()));
597                Ok(())
598            } else if shared.input.beyond_end(offset, out.len()) {
599                Err(CoroutineError::Source(SourceError::UnexpectedEof {
600                    offset,
601                    len: out.len() as u64,
602                    total: shared.input.total(),
603                }))
604            } else {
605                // Name only the gaps, so a driver that fetched part of the
606                // span (a sequential chunk, a coalesced neighbour) is asked
607                // for the rest rather than the same bytes again.
608                let before = gaps.len();
609                shared.input.missing_into(offset, out.len(), &mut gaps);
610                if sequential {
611                    // One contiguous request from the first missing byte to
612                    // the end of the read, so a stream driver supplies it
613                    // in one input.
614                    let first = gaps.get(before).map(|(o, _)| *o).unwrap_or(offset);
615                    gaps.truncate(before);
616                    let end = offset + out.len() as u64;
617                    missing.push(InputRequest::at_least(first, (end - first) as usize));
618                } else {
619                    missing.extend(
620                        gaps.drain(before..)
621                            .map(|(o, l)| InputRequest::Range(ByteRange::new(o, l))),
622                    );
623                }
624                Ok(())
625            }
626        };
627        let result = match &mut this.reqs {
628            ReqSlot::One(offset, out) => check(&shared, *offset, out, false),
629            ReqSlot::Sequential(offset, out) => check(&shared, *offset, out, true),
630            ReqSlot::Many(reqs) => reqs
631                .iter_mut()
632                .try_for_each(|(offset, out)| check(&shared, *offset, out, false)),
633        };
634        shared.gaps = gaps;
635        if let Err(e) = result {
636            shared.need = missing;
637            shared.satisfied = satisfied;
638            return Poll::Ready(Err(e));
639        }
640        let ready = missing.is_empty();
641        if ready && this.release {
642            for (offset, len) in &satisfied {
643                if let Err(error) = shared.input.release(*offset, *len) {
644                    shared.need = missing;
645                    shared.satisfied = satisfied;
646                    return Poll::Ready(Err(error));
647                }
648            }
649        }
650        shared.need = missing;
651        shared.satisfied = satisfied;
652        if ready {
653            Poll::Ready(Ok(()))
654        } else {
655            Poll::Pending
656        }
657    }
658}
659
660struct Len<'a> {
661    shared: &'a SharedCell,
662}
663
664impl Future for Len<'_> {
665    type Output = Result<u64, CoroutineError>;
666
667    fn poll(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Self::Output> {
668        let mut shared = self.shared.lock();
669        if let Some(r) = shared.cancelled {
670            return Poll::Ready(Err(CoroutineError::Cancelled(r)));
671        }
672        match shared.input.total() {
673            Some(t) => Poll::Ready(Ok(t)),
674            None => {
675                shared.need.clear();
676                shared.need.push(InputRequest::Len);
677                Poll::Pending
678            }
679        }
680    }
681}
682
683struct WriteRows<'a> {
684    shared: &'a SharedCell,
685    pending: Option<PendingRows>,
686    error: Option<CoroutineError>,
687}
688
689impl Future for WriteRows<'_> {
690    type Output = Result<(), CoroutineError>;
691
692    fn poll(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Self::Output> {
693        let this = self.get_mut();
694        if let Some(error) = this.error.take() {
695            return Poll::Ready(Err(error));
696        }
697        let mut shared = this.shared.lock();
698        if let Some(r) = shared.cancelled {
699            return Poll::Ready(Err(CoroutineError::Cancelled(r)));
700        }
701        match this.pending.take() {
702            Some(rows) => {
703                shared.rows = Some(rows);
704                Poll::Pending
705            }
706            // Resumed: the decoder took the rows (or failed; it then never
707            // resumes us).
708            None => Poll::Ready(Ok(())),
709        }
710    }
711}
712
713struct Announce<'a> {
714    shared: &'a SharedCell,
715    update: Option<MetadataUpdate>,
716}
717
718impl Future for Announce<'_> {
719    type Output = Result<(), CoroutineError>;
720
721    fn poll(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Self::Output> {
722        let this = self.get_mut();
723        let mut shared = this.shared.lock();
724        if let Some(r) = shared.cancelled {
725            return Poll::Ready(Err(CoroutineError::Cancelled(r)));
726        }
727        match this.update.take() {
728            Some(update) => {
729                shared.metadata = Some(update);
730                Poll::Pending
731            }
732            None => Poll::Ready(Ok(())),
733        }
734    }
735}
zencodec_coroutine/context.rs

zencodec_coroutine/
context.rs
