Телеграмм чат группы rust

А есть какой-то аналог select!, который ветки исполняет в бэкграунде/отдельных фьючах?

У меня есть поток входящих сообщений, и я хожу их обрабатывать конкурентно, fanout по потокам или как-нибудь так

источник

18:21пожаловаться #4

folex in Rust Async

Я могу руками конечно запускать task::spawn, но это же нужно писать бойлерплейт чтобы эти фьючи поллились, накосячу еще где-нибудь.

источник

18:22пожаловаться #5

folex in Rust Async

Вижу есть parallel-stream и rayon

источник

18:24пожаловаться #6

pl 🦑 hk in Rust Async

folex

Какой бойлерплейт?

источник

18:24пожаловаться #7

folex in Rust Async

Не уверен. Что-то типа

let futures = select! {
   msg = fused_stream.next() => {
     task::spawn(f(msg))
   }
}

poll_futures(futures)

но я не пробовал еще так делать, так что это псевдокод

источник

18:26пожаловаться #8

folex in Rust Async

Хотя, хм, task::spawn же не надо поллить, оно само запускается, а возвращает лишь JoinHandler.

источник

18:26пожаловаться #9

folex in Rust Async

Еще из минусов у такого подхода – нет backpressure

источник

18:29пожаловаться #10

folex in Rust Async

у меня fused_stream это mpsc::channel, и у него есть backpressure, и это хорошо. А если читать в task::spawn, то оно будет бесконечно читать, и забивать экзекьютор фьючами. Это мб роскошь, но хотелось бы как-то получше обращаться с экзекьютором

источник

18:30пожаловаться #11

folex in Rust Async

По хорошему хотелось бы ограничить кол-во одновременно сущствующих future, запущенных таким образом. Но как за этим следить в подходе с msg = next() => { task::spawn(..) }? 🤔

источник

18:34пожаловаться #12

pl 🦑 hk in Rust Async

https://docs.rs/tokio/1.0.1/tokio/sync/struct.Semaphore.html

docs.rs

tokio::sync::Semaphore - Rust

API documentation for the Rust `Semaphore` struct in crate `tokio`.

источник

18:58пожаловаться #13

pl 🦑 hk in Rust Async

https://docs.rs/futures/0.3.9/futures/prelude/stream/trait.StreamExt.html#method.buffer_unordered

docs.rs

futures::prelude::stream::StreamExt - Rust

API documentation for the Rust `StreamExt` trait in crate `futures`.

источник

18:59пожаловаться #14

folex in Rust Async

Спасибо! Я пока остановился на for_each_concurrent.

источник

23:22пожаловаться #15

folex in Rust Async

Правда, набрел на непонятную ошибку – никогда не встречал такую.
https://play.rust-lang.org/?version=stable&mode=debug&edition=2018&gist=fcacd1e1a21b594688681c502a3c4a90

> move occurs due to use in generator

источник

23:22пожаловаться #16

folex in Rust Async

т.е. вот такой вот паттерн

let value: Arc<Mutex<..>> = ..;
let cloned_value = value.clone();
FnMut: || { async move { 
  let value = cloned_value.clone();
  foo(value)
} }

почему-то говорит, что move случается лишний. Хотя вроде бы везде всё клонируется

источник

23:24пожаловаться #17

Denis Nevmerzhitskii in Rust Async

folex

т.е. вот такой вот паттерн

let value: Arc<Mutex<..>> = ..;
let cloned_value = value.clone();
FnMut: || { async move { 
  let value = cloned_value.clone();
  foo(value)
} }

почему-то говорит, что move случается лишний. Хотя вроде бы везде всё клонируется

Попробуй клонировать в замыкании, но перед async блоком

источник

23:42пожаловаться #18

folex in Rust Async

хм, кажется сработало! спасибо

источник

23:44пожаловаться #19

Denis Nevmerzhitskii in Rust Async

В первом случае клон вызывался один раз, и первый же async блок забрал бы его

источник

23:47пожаловаться #20