🧖🏻 🛍️ 🧖🏽 Rust में async-प्रतीक्षित बुनियादी ढांचे की ओर पलायन ⚙️ 🎎 👐🏻

पक्षियों का प्रवास
img स्रोत

पिछले सप्ताह रस्ट समुदाय के लिए एक बड़ी घटना हुई - संकलक संस्करण 1.39 जारी किया गया था , और इसके स्थान पर एसिंक्स-वेट सुविधाओं का स्थिरीकरण किया गया था । इस पोस्ट में, मैं कंपाइलर और इकोसिस्टम में सभी प्रासंगिक परिवर्तनों को संक्षेप में प्रस्तुत करने की कोशिश करूंगा, साथ ही साथ एसिंक्स-वेट प्रतिमान के लिए माइग्रेशन निर्देश भी प्रदान करूंगा। मैं रस्ट में अतुल्यकालिक विश्लेषण का विस्तृत विश्लेषण नहीं करूंगा, अभी भी हैबे पर प्रासंगिक लेख हैं जो इस विषय पर पहुंचने में मदद करेंगे:

mio पर चैट करें: भाग 1 , भाग 2 ;
टोकियो + वायदा-आरएस 0.1 ;
टोकियो की समीक्षा ;
जनरेटर (यहां वेट के साथ वाक्य रचना का अस्वीकार्य संस्करण है await! मैक्रो का उपयोग किया जाता है, यह पहले से ही काम नहीं करेगा, लेकिन पृष्ठभूमि एक ही है)।

इन लेखों के अलावा, आप मानक पुस्तकालय और आवश्यक बक्से के प्रलेखन का भी उल्लेख कर सकते हैं, साथ ही साथ async-book (अंग्रेजी में) भी पढ़ सकते हैं।

इस आलेख में चर्चा किए गए सभी उदाहरण स्थिर संकलक 1.39 पर काम करते हैं और बाद के सभी संस्करणों पर काम करना चाहिए। अंतिम कोड जीथब पर उपलब्ध है।

अतुल्यकालिक कोड को लागू करने के लिए, वायदा-0.1 पुस्तकालय का उपयोग किया गया था। यह आधार लक्षण futures::Future प्रदान करता है futures::Future और futures::Stream स्थगित कंप्यूटिंग के साथ काम करने के लिए futures::Stream । वे Result<..> प्रकारों पर काम करते हैं और कॉम्बिनेटर का एक सेट प्रदान करते हैं। इसके अलावा, पुस्तकालय कार्यों (कार्यों) के बीच संचार के लिए चैनल प्रदान करता है, निष्पादक और उसकी कार्य प्रणाली के साथ काम करने के लिए विभिन्न इंटरफेस, और बहुत कुछ।

एक उदाहरण पर विचार करें जो उच्चतम 32 बिटकॉन्स से एक श्रृंखला बनाता है और उन्हें Sink भेजता है:

 // futures 0.1.29 use futures::prelude::*; use futures::{stream, futures}; fn sink_fibb_series( sink: impl Sink<SinkItem = u32, SinkError = ()>, ) -> impl Future<Item = (), Error = ()> { stream::unfold((1u32, 1), |(mut fact, n)| { while fact.checked_mul(n).is_none() { fact >>= 1; } fact *= n; Some(future::ok((fact, (fact, n + 1)))) }) .forward(sink) .map(|_v| ()) }

नोट: कोरटाइन पर सीपीयू-बाउंड कार्यों को ध्यान में रखते हुए सबसे अच्छा अनुप्रयोग नहीं है, लेकिन उदाहरण आत्मनिर्भर और सरल है।

जैसा कि आप देख सकते हैं, कोड बल्कि बोझिल लग रहा है: आपको इस तथ्य के बावजूद कि कोई उपयोगी मूल्य नहीं है, वापसी मूल्य निर्दिष्ट करना होगा। वायदा 0.3 में, कोड थोड़ा आसान हो जाता है:

 // futures 0.3.1 use futures::prelude::*; use futures::stream; async fn sink_fibb_series(sink: impl Sink<u32>) { stream::unfold((1u32, 1), |(mut fact, n)| { async move { while fact.checked_mul(n).is_none() { fact >>= 1; } fact *= n; Some((fact, (fact, n + 1))) } }) .map(Ok) .forward(sink) .map(|_v| ()) .await; }

यहां, फ़ंक्शन में async कीवर्ड जोड़ा जाता है, जो Future में फ़ंक्शन के रिटर्न मान को लपेटता है। चूंकि हमारे मामले में यह शून्य आकार का एक टपल है, इसलिए इसे साधारण कार्यों की तरह ही छोड़ा जा सकता है।

निष्पादन के लिए प्रतीक्षा करने के लिए कॉल श्रृंखला के अंत में await कीवर्ड का उपयोग किया जाता है। यह कॉल वर्तमान async संदर्भ में निष्पादन को रोकती है और जब तक कि Future का अपेक्षित मान तैयार नहीं हो जाता है, तब तक इसे शेड्यूलर को नियंत्रित करता है। फिर, निष्पादन अंतिम await (हमारे उदाहरण में फ़ंक्शन को समाप्त करना) के साथ फिर से शुरू होता है, अर्थात। नियंत्रण प्रवाह समान समकालिक कोड की तुलना में गैर-रैखिक हो जाता है।

एक अन्य महत्वपूर्ण अंतर stream::unfold अंदर बंद होने के शरीर में एक async ब्लॉक की उपस्थिति है stream::unfold । यह आवरण एक ही शरीर के साथ एक नया async फ़ंक्शन घोषित करने और एक async ब्लॉक के बजाय कॉल करने के लिए एक पूर्ण एनालॉग है।

# सुविधा (async_closure)

शायद यह क्लोजर जल्द ही async_closure सुविधा का उपयोग करके लिखा जा सकता है, लेकिन अफसोस, इसे अभी तक लागू नहीं किया गया है:

 async |(mut fact, n)| { while fact.checked_mul(n).is_none() { fact >>= 1; } fact *= n; Some((fact, (fact, n + 1))) }

जैसा कि आप देख सकते हैं, नया Stream प्रकार न केवल Result<..> तत्वों के साथ काम करता है Result<..> , जैसा कि पहले था। Future लक्षण के समान परिवर्तन किए गए थे, संस्करण की परिभाषाएं इस प्रकार हैं:

 // futures 0.1 trait Future { type Item; type Error; fn poll(&mut self) -> Result<Async<Self::Item>, Self::Error>; } enum Async<T> { Ready(T), NotReady } // futures 0.3 trait Future { type Output; fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output>; } enum Poll<T> { Ready(T), Pending }

इस तथ्य के अलावा कि रिटर्न प्रकार मनमाना हो सकता है, Future::poll के इनपुट पैरामीटर भी बदल गए हैं। एक नया Context पैरामीटर दिखाई दिया है, जो वर्तमान कार्य को जागने के लिए एक स्पष्ट इंटरफ़ेस प्रदान करता है। इससे पहले, एक ही चीज़ को किसी विशेष निष्पादक के वैश्विक चर के माध्यम से प्राप्त किया जा सकता है (उदाहरण के लिए, tokio::prelude::task::current().notify() को कॉल करके tokio::prelude::task::current().notify() )।

इंटरफ़ेस के बीच अधिक मौलिक अंतर यह है कि आपको अपने आप को Pin में लिंक लपेटना होगा। पॉइंटर पर यह आवरण स्मृति में डेटा की "गतिहीनता" की गारंटी देता है ( Pin का अधिक विस्तृत विवरण हब पर कंपाइलर की 1.33 रिलीज में है , या अंग्रेजी में, मानक std :: pin पुस्तकालय के प्रलेखन में)।

आइए अब हमारे उदाहरण को चलाने का प्रयास करें। Sink रूप में हम वायदा से आधा चैनल लेते हैं और आउटपुट पक्ष पर हम पुनरावृत्तियों के बीच कुछ देरी के साथ परिणाम प्रिंट करेंगे। वायदा-0.1 पर, इस तरह के कोड को निम्नानुसार लिखा जा सकता है:

 use std::time::{Duration, Instant}; // futures 0.1.29 use futures::prelude::*; use futures::sync::mpsc; // tokio 0.1.22 use tokio::runtime::Runtime; use tokio::timer::Delay; fn main() { let mut rt = Runtime::new().unwrap(); let (tx, rx) = mpsc::channel(32); rt.spawn(Box::new(sink_fibb_series(tx.sink_map_err(|_e| ())))); let fut = rx.take(100).for_each(|val| { println!("{}", val); Delay::new(Instant::now() + Duration::from_millis(50)) .map(|_| ()) .map_err(|_| ()) }); rt.spawn(Box::new(fut)); rt.shutdown_on_idle().wait().unwrap(); }

नए टोकिओ के साथ समान कोड (जो लेखन के समय अभी भी अल्फ़ा है) और वायदा-0.3 इस तरह दिख सकता है:

 use std::time::Duration; // futures 0.3.1 use futures::channel::mpsc; use futures::prelude::*; // tokio 0.2.0-alpha.5 use tokio::timer; #[tokio::main] async fn main() { let (tx, rx) = mpsc::channel(32); tokio::spawn(sink_fibb_series(tx)); rx.take(100) .for_each(|val| { println!("{}", val); timer::delay_for(Duration::from_millis(50)) }) .await; }

जैसा कि आप देख सकते हैं, नए वायदा के साथ कोड बहुत कम हो गया है। लेखक के अनुभव के अनुसार, लाइनों की संख्या हमेशा काफी कम निकलती है (कभी-कभी समकालिक कोड को फिर से लिखते समय भी)। लेकिन यह मुझे लगता है कि पठनीयता और map / map_err मिश्रण की कमी में बहुत अधिक अंतर है, जो कि Result<..> में मानक प्रकारों में त्रुटियों की परिवर्तनशीलता के कारण आवश्यक थे।

Result<..> तत्वों पर संयोजक Result<..> प्रकार अभी भी बने हुए हैं और अलग-अलग प्रकारों में हैं, कुछ थोड़ा अद्यतन नाम के साथ। अब वे दो अलग-अलग प्रकारों में विभाजित हैं; जिन्हें लागू किया गया है:

सभी तत्व: futures::FuturesExt और futures::StreamExt ;
Result<..> प्रकार के तत्व Result<..> : futures::TryFuturesExt और futures::TryStreamExt ।

Future और Stream लक्षणों का कार्यान्वयन थोड़ा अधिक जटिल है। एक उदाहरण के रूप में, आइए पहले से ही विचार की गई संख्या श्रृंखला के लिए Stream को लागू करने का प्रयास करें। वायदा के दोनों संस्करणों के लिए सामान्य प्रकार निम्नानुसार होंगे:

 struct FactStream { fact: u32, n: u32, } impl FactStream { fn new() -> Self { Self { fact: 1, n: 1 } } }

वायदा -१.० के लिए, कार्यान्वयन निम्नानुसार होगा:

 impl Stream for FactStream { type Item = u32; type Error = (); fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> { while self.fact.checked_mul(self.n).is_none() { self.fact >>= 1; } self.fact *= self.n; self.n += 1; Ok(Async::Ready(Some(self.fact))) } }

इस उदाहरण में, Stream::poll कार्यान्वयन वास्तव में क्लोजर stream::unfold की पूरी प्रतिलिपि है stream::unfold । वायदा-0.3 के मामले में, कार्यान्वयन समान है:

 impl Stream for FactStream { type Item = u32; fn poll_next(mut self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Option<Self::Item>> { while self.fact.checked_mul(self.n).is_none() { self.fact >>= 1; } self.fact *= self.n; self.n += 1; Poll::Ready(Some(self.fact)) } }

हालाँकि, यदि संरचना क्षेत्र का प्रकार Unpin लागू नहीं करता है, तो std::ops::DerefMut को Pin<&mut T> पर लागू नहीं किया जाएगा और इस प्रकार सभी क्षेत्रों में कोई परिवर्तनशील पहुंच नहीं होगी:

 use std::marker::PhantomPinned; struct Fact { inner: u32, //    Unpin   _pin: PhantomPinned, } struct FactStream { fact: Fact, n: u32, } impl Stream for FactStream { type Item = u32; fn poll_next(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Option<Self::Item>> { while self.fact.inner.checked_mul(self.n).is_none() { self.fact.inner >>= 1; // <-   // trait `DerefMut` is required to modify // through a dereference, but it is not // implemented for `std::pin::Pin<&mut FactStream>` } self.fact.inner *= self.n; // <-   self.n += 1; // <- Poll::Ready(Some(self.fact.inner)) } }

इस मामले में, एक या दूसरे रूप में, आपको "प्रोजेक्शन" प्राप्त करने के लिए असुरक्षित कार्यों Pin::get_unchecked_mut और Pin::map_unchecked_mut का उपयोग करना होगा ( दस्तावेज़ में अधिक विस्तृत विवरण है )। सौभाग्य से, ऐसे मामलों के लिए, पिन_प्रोजेक्ट टोकरे में एक सुरक्षित आवरण लागू किया गया है (कार्यान्वयन का विवरण पुस्तकालय प्रलेखन में पाया जा सकता है)।

 use pin_project::pin_project; #[pin_project] struct FactStream { fact: Fact, n: u32, } impl Stream for FactStream { type Item = u32; fn poll_next(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Option<Self::Item>> { let mut this = self.project(); while this.fact.inner.checked_mul(*this.n).is_none() { this.fact.inner >>= 1; } this.fact.inner *= *this.n; *this.n += 1; Poll::Ready(Some(this.fact.inner)) } }

अंतिम बिंदु जिसे मैं उजागर करना चाहूंगा वह है विभिन्न संस्करणों के प्रकारों के बीच अंतर-स्थिरता। ऐसा करने के लिए, एक मॉड्यूल फ्यूचर्स :: कंपेटर्स है , जो आपको पुराने प्रकारों से नए में बदलने और इसके विपरीत करने की अनुमति देता है। उदाहरण के लिए, आप async-wait का उपयोग करके वायदा-0.1 से Stream पर पुनरावृति कर सकते हैं:

 use std::fmt::Display; // futures 0.3 use new_futures::compat::Compat01As03 as Compat; use new_futures::StreamExt as _; // futures 0.1 use old_futures::Stream as OldStream; async fn stream_iterate<E>( old_stream: impl OldStream<Item = impl Display, Error = E>, ) -> Result<(), E> { let stream = Compat::new(old_stream); let mut stream = Box::pin(stream); while let Some(item) = stream.as_mut().next().await.transpose()? { println!("{}", item); } Ok(()) }

नोट: केवल टोकियो निष्पादक को लेख में सबसे लंबे समय तक रहने वाला और व्यापक माना जाता है। फिर भी, दुनिया वहां समाप्त नहीं होती है, उदाहरण के लिए, एक वैकल्पिक async-std , जो इसके अलावा मानक पुस्तकालय के प्रकारों के लिए फ्यूचरिस्टिक रैपर प्रदान करता है, साथ ही LocalPool और LocalPool को माना गया वायदा-0.3 पुस्तकालय से।

Rust में async-प्रतीक्षित बुनियादी ढांचे की ओर पलायन

More articles: