rust学习（tokio future分析）-编程知识

rust学习（tokio future分析）

自己编写一个impl future来看一下Tokio的是如何实现的。

第一步：

代码：

struct TExecuteTask {count:u32
}impl Future for TExecuteTask {type Output = ();fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {println!("future require");if self.count == 2 {return Ready(());}unsafe {self.get_unchecked_mut().count += 1;}Pending}
}pub fn test1() {let v = TExecuteTask{count:0};let rt = tokio::runtime::Builder::new_multi_thread().enable_all().build().unwrap();println!("test trace1");rt.block_on(v);println!("test trace2");
}

运行结果：

貌似调用block_on之后会回调一次future的poll，如果这个时候还没有计算出结果，那就直接卡死了，所以需要找个地方能唤醒。

第二步：

原来代码上添加上wake

impl Future for TExecuteTask {type Output = ();fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {println!("future require");let mut v = self.core.lock().unwrap();v.waker = Some(cx.waker().clone());Pending}
}pub fn test1() {let waker: Arc<Mutex<TExecuteTaskCore>> = Arc::new(Mutex::new(TExecuteTaskCore {waker:None}));let task: TExecuteTask = TExecuteTask {core:waker.clone()};let rt = tokio::runtime::Builder::new_multi_thread().enable_all().build().unwrap();let closure1 = waker.clone();thread::spawn(move|| {thread::sleep(Duration::from_secs(1));let mut core: std::sync::MutexGuard<'_, TExecuteTaskCore> = closure1.lock().unwrap();if let Some(waker) = core.waker.take() {println!("detect future is ready, wakeup the future task to executor.");waker.wake()    // wakeup the future task to executor.}});println!("test trace1");rt.block_on(task);println!("test trace2");
}

关键是在poll返回pending的时候，获取waker。然后等计算完成后再wake一下哈哈。

然后结果呢：

还是卡住了。。。这个是为啥呢？原来是我poll的代码有问题，无论如何处理都是返回pending..

所以从目前的试验来看有如下结论：

1.当调用block_on的时候，会直接调用一次future的poll确认是否运算结束。

2.当调用waker.wake的时候貌似还会调用一次poll确认是否运算结束。

所以我们现在要加一个标志位来表示运算结束，这样block_on应该就能正常走下去了吧。

第三步：

impl Future for TExecuteTask {type Output = ();fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {println!("future require");let mut v = self.core.lock().unwrap();if v.complete == true {return Ready(())}v.waker = Some(cx.waker().clone());Pending}
}pub fn test1() {let waker: Arc<Mutex<TExecuteTaskCore>> = Arc::new(Mutex::new(TExecuteTaskCore {waker:None,complete:false}));let task: TExecuteTask = TExecuteTask {core:waker.clone()};let rt = tokio::runtime::Builder::new_multi_thread().enable_all().build().unwrap();let closure1 = waker.clone();thread::spawn(move|| {thread::sleep(Duration::from_secs(1));let mut core: std::sync::MutexGuard<'_, TExecuteTaskCore> = closure1.lock().unwrap();if let Some(waker) = core.waker.take() {println!("detect future is ready, wakeup the future task to executor.");core.complete = true;waker.wake()    // wakeup the future task to executor.}});println!("test trace1");rt.block_on(task);println!("test trace2");
}

添加了一个complete标志运算是否结束。