You are an expert in writing code in many different languages. Your goal is to perform code completion for the following code, keeping in mind the rest of the code and the file meta data.
    Metadata details:
    filename: windows.rs
    path of file: ./repos/wasmtime/crates/wasi-common/src/sync/sched
    
    the code of the file until where you have to start completion: // The windows scheduler is unmaintained and due for a rewrite.
//
// Rather than use a polling mechanism for file read/write readiness,
// it checks readiness just once, before sleeping for any timer subscriptions.
// Checking stdin readiness uses a worker thread which, once started, lives for the
// lifetime of the process.
//
// We suspect there are bugs in this scheduler, however, we have not
// taken the time to improve it. See bug #2880.

use crate::sched::subscription::{RwEventFlags, Subscription};
use crate::{file::WasiFile, sched::Poll, Error, ErrorExt};
use once_cell::sync::Lazy;
use std::sync::mpsc::{self, Receiver, RecvTimeoutError, Sender, TryRecvError};
use std::sync::Mutex;
use std::thread;
use std::time::Duration;

pub async fn poll_oneoff<'a>(poll: &mut Poll<'a>) -> Result<(), Error> {
    poll_oneoff_(poll, wasi_file_is_stdin).await
}

pub async fn poll_oneoff_<'a>(
    poll: &mut Poll<'a>,
    file_is_stdin: impl Fn(&dyn WasiFile) -> bool,
) -> Result<(), Error> {
    if poll.is_empty() {
        return Ok(());
    }

    let mut ready = false;
    let waitmode = if let Some(t) = poll.earliest_clock_deadline() {
        if let Some(duration) = t.duration_until() {
            WaitMode::Timeout(duration)
        } else {
            WaitMode::Immediate
        }
    } else {
        if ready {
            WaitMode::Immediate
        } else {
            WaitMode::Infinite
        }
    };

    let mut stdin_read_subs = Vec::new();
    let mut immediate_reads = Vec::new();
    let mut immediate_writes = Vec::new();
    for s in poll.rw_subscriptions() {
        matc
    
    fn poll(&self, wait_mode: WaitMode) -> Result<PollState, Error> {
        match self.notify_rx.try_recv() {
            // Clean up possibly unread result from previous poll.
            Ok(_) | Err(TryRecvError::Empty) => {}
            Err(TryRecvError::Disconnected) => {
                return Err(Error::trap(anyhow::Error::msg(
                    "StdinPoll notify_rx channel closed",
                )))
            }
        }

        // Notify the worker thread to poll stdin
        self.request_tx
            .send(())
            .map_err(|_| Error::trap(anyhow::Error::msg("request_tx channel closed")))?;

        // Wait for the worker thread to send a readiness notification
        match wait_mode {
            WaitMode::Timeout(timeout) => match self.notify_rx.recv_timeout(timeout) {
                Ok(r) => Ok(r),
                Err(RecvTimeoutError::Timeout) => Ok(PollState::TimedOut),
                Err(RecvTimeoutError::Disconnected) => Err(Error::trap(anyhow::Error::msg(
                    "StdinPoll notify_rx channel closed",
                ))),
            },
            WaitMode::Infinite => self
                .notify_rx
                .recv()
                .map_err(|_| Error::trap(anyhow::Error::msg("StdinPoll notify_rx channel closed"))),
            WaitMode::Immediate => match self.notify_rx.try_recv() {
                Ok(r) => Ok(r),
                Err(TryRecvError::Empty) => Ok(PollState::NotReady),
                Err(TryRecvError::Disconnected) => Err(Error::trap(anyhow::Error::msg(
                    "StdinPoll notify_rx channel closed",
                ))),
            },
        }
    }