Yehowshua/ZIm
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Split out a foreground and background executor

Browse source
This commit is contained in:
Nathan Sobo 2023-11-01 13:53:08 -06:00
parent dd1a2a9e44
commit 11b6d9e33a
26 changed files with 165 additions and 150 deletions
Download patch file Download diff file Expand all files Collapse all files

1
crates/ai2/src/auth.rs
View file

@ -1,4 +1,3 @@
use async_trait::async_trait;
use gpui2::AppContext;
#[derive(Clone, Debug)]

4
crates/ai2/src/providers/open_ai/completion.rs
View file

@ -1,5 +1,4 @@
use anyhow::{anyhow, Result};
use async_trait::async_trait;
use futures::{
future::BoxFuture, io::BufReader, stream::BoxStream, AsyncBufReadExt, AsyncReadExt, FutureExt,
Stream, StreamExt,
@ -258,8 +257,7 @@ impl CredentialProvider for OpenAICompletionProvider {
}
fn delete_credentials(&self, cx: &mut AppContext) {
cx.run_on_main(move |cx| cx.delete_credentials(OPENAI_API_URL).log_err())
.await;
cx.delete_credentials(OPENAI_API_URL).log_err();
*self.credential.write() = ProviderCredential::NoCredentials;
}
}

52
crates/ai2/src/providers/open_ai/embedding.rs
View file

@ -146,7 +146,6 @@ impl OpenAIEmbeddingProvider {
}
}
#[async_trait]
impl CredentialProvider for OpenAIEmbeddingProvider {
fn has_credentials(&self) -> bool {
match *self.credential.read() {
@ -154,52 +153,45 @@ impl CredentialProvider for OpenAIEmbeddingProvider {
_ => false,
}
}
async fn retrieve_credentials(&self, cx: &mut AppContext) -> ProviderCredential {
fn retrieve_credentials(&self, cx: &mut AppContext) -> ProviderCredential {
let existing_credential = self.credential.read().clone();
let retrieved_credential = cx
.run_on_main(move |cx| match existing_credential {
ProviderCredential::Credentials { .. } => {
return existing_credential.clone();
}
_ => {
if let Some(api_key) = env::var("OPENAI_API_KEY").log_err() {
return ProviderCredential::Credentials { api_key };
}
if let Some(Some((_, api_key))) = cx.read_credentials(OPENAI_API_URL).log_err()
{
if let Some(api_key) = String::from_utf8(api_key).log_err() {
return ProviderCredential::Credentials { api_key };
} else {
return ProviderCredential::NoCredentials;
}
let retrieved_credential = match existing_credential {
ProviderCredential::Credentials { .. } => existing_credential.clone(),
_ => {
if let Some(api_key) = env::var("OPENAI_API_KEY").log_err() {
ProviderCredential::Credentials { api_key }
} else if let Some(Some((_, api_key))) =
cx.read_credentials(OPENAI_API_URL).log_err()
{
if let Some(api_key) = String::from_utf8(api_key).log_err() {
ProviderCredential::Credentials { api_key }
} else {
return ProviderCredential::NoCredentials;
ProviderCredential::NoCredentials
}
} else {
ProviderCredential::NoCredentials
}
})
.await;
}
};
*self.credential.write() = retrieved_credential.clone();
retrieved_credential
}
async fn save_credentials(&self, cx: &mut AppContext, credential: ProviderCredential) {
fn save_credentials(&self, cx: &mut AppContext, credential: ProviderCredential) {
*self.credential.write() = credential.clone();
let credential = credential.clone();
cx.run_on_main(move |cx| match credential {
match credential {
ProviderCredential::Credentials { api_key } => {
cx.write_credentials(OPENAI_API_URL, "Bearer", api_key.as_bytes())
.log_err();
}
_ => {}
})
.await;
}
}
async fn delete_credentials(&self, cx: &mut AppContext) {
cx.run_on_main(move |cx| cx.delete_credentials(OPENAI_API_URL).log_err())
.await;
fn delete_credentials(&self, cx: &mut AppContext) {
cx.delete_credentials(OPENAI_API_URL).log_err();
*self.credential.write() = ProviderCredential::NoCredentials;
}
}

14
crates/ai2/src/test.rs
View file

@ -100,16 +100,15 @@ impl FakeEmbeddingProvider {
}
}
#[async_trait]
impl CredentialProvider for FakeEmbeddingProvider {
fn has_credentials(&self) -> bool {
true
}
async fn retrieve_credentials(&self, _cx: &mut AppContext) -> ProviderCredential {
fn retrieve_credentials(&self, _cx: &mut AppContext) -> ProviderCredential {
ProviderCredential::NotNeeded
}
async fn save_credentials(&self, _cx: &mut AppContext, _credential: ProviderCredential) {}
async fn delete_credentials(&self, _cx: &mut AppContext) {}
fn save_credentials(&self, _cx: &mut AppContext, _credential: ProviderCredential) {}
fn delete_credentials(&self, _cx: &mut AppContext) {}
}
#[async_trait]
@ -162,16 +161,15 @@ impl FakeCompletionProvider {
}
}
#[async_trait]
impl CredentialProvider for FakeCompletionProvider {
fn has_credentials(&self) -> bool {
true
}
async fn retrieve_credentials(&self, _cx: &mut AppContext) -> ProviderCredential {
fn retrieve_credentials(&self, _cx: &mut AppContext) -> ProviderCredential {
ProviderCredential::NotNeeded
}
async fn save_credentials(&self, _cx: &mut AppContext, _credential: ProviderCredential) {}
async fn delete_credentials(&self, _cx: &mut AppContext) {}
fn save_credentials(&self, _cx: &mut AppContext, _credential: ProviderCredential) {}
fn delete_credentials(&self, _cx: &mut AppContext) {}
}
impl CompletionProvider for FakeCompletionProvider {

4
crates/call2/src/call2.rs
View file

@ -196,7 +196,7 @@ impl ActiveCall {
})
.shared();
self.pending_room_creation = Some(room.clone());
cx.executor().spawn(async move {
cx.background_executor().spawn(async move {
room.await.map_err(|err| anyhow!("{:?}", err))?;
anyhow::Ok(())
})
@ -230,7 +230,7 @@ impl ActiveCall {
};
let client = self.client.clone();
cx.executor().spawn(async move {
cx.background_executor().spawn(async move {
client
.request(proto::CancelCall {
room_id,

9
crates/call2/src/room.rs
View file

@ -322,7 +322,7 @@ impl Room {
fn app_will_quit(&mut self, cx: &mut ModelContext<Self>) -> impl Future<Output = ()> {
let task = if self.status.is_online() {
let leave = self.leave_internal(cx);
Some(cx.executor().spawn(async move {
Some(cx.background_executor().spawn(async move {
leave.await.log_err();
}))
} else {
@ -390,7 +390,7 @@ impl Room {
self.clear_state(cx);
let leave_room = self.client.request(proto::LeaveRoom {});
cx.executor().spawn(async move {
cx.background_executor().spawn(async move {
leave_room.await?;
anyhow::Ok(())
})
@ -1202,7 +1202,7 @@ impl Room {
};
cx.notify();
cx.executor().spawn_on_main(move || async move {
cx.background_executor().spawn(async move {
client
.request(proto::UpdateParticipantLocation {
room_id,
@ -1569,7 +1569,8 @@ impl LiveKitRoom {
*muted = should_mute;
cx.notify();
Ok((
cx.executor().spawn(track_publication.set_mute(*muted)),
cx.background_executor()
.spawn(track_publication.set_mute(*muted)),
old_muted,
))
}

32
crates/client2/src/client2.rs
View file

@ -234,12 +234,14 @@ struct ClientState {
message_handlers: HashMap<
TypeId,
Arc<
dyn Fn(
AnyModel,
Box<dyn AnyTypedEnvelope>,
&Arc<Client>,
AsyncAppContext,
) -> LocalBoxFuture<'static, Result<()>>,
dyn Send
+ Sync
+ Fn(
AnyModel,
Box<dyn AnyTypedEnvelope>,
&Arc<Client>,
AsyncAppContext,
) -> LocalBoxFuture<'static, Result<()>>,
>,
>,
}
@ -551,7 +553,11 @@ impl Client {
where
M: EnvelopedMessage,
E: 'static,
H: 'static + Sync + Fn(Model<E>, TypedEnvelope<M>, Arc<Self>, AsyncAppContext) -> F,
H: 'static
+ Sync
+ Fn(Model<E>, TypedEnvelope<M>, Arc<Self>, AsyncAppContext) -> F
+ Send
+ Sync,
F: 'static + Future<Output = Result<()>>,
{
let message_type_id = TypeId::of::<M>();
@ -593,7 +599,11 @@ impl Client {
where
M: RequestMessage,
E: 'static,
H: 'static + Sync + Fn(Model<E>, TypedEnvelope<M>, Arc<Self>, AsyncAppContext) -> F,
H: 'static
+ Sync
+ Fn(Model<E>, TypedEnvelope<M>, Arc<Self>, AsyncAppContext) -> F
+ Send
+ Sync,
F: 'static + Future<Output = Result<M::Response>>,
{
self.add_message_handler(model, move |handle, envelope, this, cx| {
@ -609,7 +619,7 @@ impl Client {
where
M: EntityMessage,
E: 'static,
H: 'static + Fn(Model<E>, TypedEnvelope<M>, Arc<Self>, AsyncAppContext) -> F,
H: 'static + Fn(Model<E>, TypedEnvelope<M>, Arc<Self>, AsyncAppContext) -> F + Send + Sync,
F: 'static + Future<Output = Result<()>>,
{
self.add_entity_message_handler::<M, E, _, _>(move |subscriber, message, client, cx| {
@ -621,7 +631,7 @@ impl Client {
where
M: EntityMessage,
E: 'static,
H: 'static + Fn(AnyModel, TypedEnvelope<M>, Arc<Self>, AsyncAppContext) -> F,
H: 'static + Fn(AnyModel, TypedEnvelope<M>, Arc<Self>, AsyncAppContext) -> F + Send + Sync,
F: 'static + Future<Output = Result<()>>,
{
let model_type_id = TypeId::of::<E>();
@ -660,7 +670,7 @@ impl Client {
where
M: EntityMessage + RequestMessage,
E: 'static,
H: 'static + Fn(Model<E>, TypedEnvelope<M>, Arc<Self>, AsyncAppContext) -> F,
H: 'static + Fn(Model<E>, TypedEnvelope<M>, Arc<Self>, AsyncAppContext) -> F + Send + Sync,
F: 'static + Future<Output = Result<M::Response>>,
{
self.add_model_message_handler(move |entity, envelope, client, cx| {

2
crates/client2/src/telemetry.rs
View file

@ -123,7 +123,7 @@ impl Telemetry {
// TODO: Replace all hardware stuff with nested SystemSpecs json
let this = Arc::new(Self {
http_client: client,
executor: cx.executor().clone(),
executor: cx.background_executor().clone(),
state: Mutex::new(TelemetryState {
app_metadata: cx.app_metadata(),
architecture: env::consts::ARCH,

12
crates/copilot2/src/copilot2.rs
View file

@ -535,7 +535,7 @@ impl Copilot {
}
};
cx.executor()
cx.background_executor()
.spawn(task.map_err(|err| anyhow!("{:?}", err)))
} else {
// If we're downloading, wait until download is finished
@ -549,7 +549,7 @@ impl Copilot {
self.update_sign_in_status(request::SignInStatus::NotSignedIn, cx);
if let CopilotServer::Running(RunningCopilotServer { lsp: server, .. }) = &self.server {
let server = server.clone();
cx.executor().spawn(async move {
cx.background_executor().spawn(async move {
server
.request::<request::SignOut>(request::SignOutParams {})
.await?;
@ -579,7 +579,7 @@ impl Copilot {
cx.notify();
cx.executor().spawn(start_task)
cx.background_executor().spawn(start_task)
}
pub fn language_server(&self) -> Option<(&LanguageServerName, &Arc<LanguageServer>)> {
@ -760,7 +760,7 @@ impl Copilot {
.request::<request::NotifyAccepted>(request::NotifyAcceptedParams {
uuid: completion.uuid.clone(),
});
cx.executor().spawn(async move {
cx.background_executor().spawn(async move {
request.await?;
Ok(())
})
@ -784,7 +784,7 @@ impl Copilot {
.map(|completion| completion.uuid.clone())
.collect(),
});
cx.executor().spawn(async move {
cx.background_executor().spawn(async move {
request.await?;
Ok(())
})
@ -827,7 +827,7 @@ impl Copilot {
.map(|file| file.path().to_path_buf())
.unwrap_or_default();
cx.executor().spawn(async move {
cx.background_executor().spawn(async move {
let (version, snapshot) = snapshot.await?;
let result = lsp
.request::<R>(request::GetCompletionsParams {

2
crates/db2/src/db2.rs
View file

@ -185,7 +185,7 @@ pub fn write_and_log<F>(cx: &mut AppContext, db_write: impl FnOnce() -> F + Send
where
F: Future<Output = anyhow::Result<()>> + Send,
{
cx.executor()
cx.background_executor()
.spawn(async move { db_write().await.log_err() })
.detach()
}

15
crates/gpui2/src/app.rs
View file

@ -21,7 +21,7 @@ use crate::{
};
use anyhow::{anyhow, Result};
use collections::{HashMap, HashSet, VecDeque};
use futures::{future::BoxFuture, Future};
use futures::{future::LocalBoxFuture, Future};
use parking_lot::Mutex;
use slotmap::SlotMap;
use std::{
@ -101,6 +101,10 @@ impl App {
self.0.borrow().background_executor.clone()
}
pub fn foreground_executor(&self) -> ForegroundExecutor {
self.0.borrow().foreground_executor.clone()
}
pub fn text_system(&self) -> Arc<TextSystem> {
self.0.borrow().text_system.clone()
}
@ -110,7 +114,7 @@ type ActionBuilder = fn(json: Option<serde_json::Value>) -> anyhow::Result<Box<d
type FrameCallback = Box<dyn FnOnce(&mut WindowContext)>;
type Handler = Box<dyn FnMut(&mut AppContext) -> bool + 'static>;
type Listener = Box<dyn FnMut(&dyn Any, &mut AppContext) -> bool + 'static>;
type QuitHandler = Box<dyn FnMut(&mut AppContext) -> BoxFuture<'static, ()> + 'static>;
type QuitHandler = Box<dyn FnMut(&mut AppContext) -> LocalBoxFuture<'static, ()> + 'static>;
type ReleaseListener = Box<dyn FnMut(&mut dyn Any, &mut AppContext) + 'static>;
pub struct AppContext {
@ -535,10 +539,15 @@ impl AppContext {
}
/// Obtains a reference to the executor, which can be used to spawn futures.
pub fn executor(&self) -> &BackgroundExecutor {
pub fn background_executor(&self) -> &BackgroundExecutor {
&self.background_executor
}
/// Obtains a reference to the executor, which can be used to spawn futures.
pub fn foreground_executor(&self) -> &ForegroundExecutor {
&self.foreground_executor
}
/// Spawns the future returned by the given function on the thread pool. The closure will be invoked
/// with AsyncAppContext, which allows the application state to be accessed across await points.
pub fn spawn<Fut, R>(&self, f: impl FnOnce(AsyncAppContext) -> Fut) -> Task<R>

28
crates/gpui2/src/app/model_context.rs
View file

@ -43,10 +43,10 @@ impl<'a, T: 'static> ModelContext<'a, T> {
pub fn observe<T2, E>(
&mut self,
entity: &E,
mut on_notify: impl FnMut(&mut T, E, &mut ModelContext<'_, T>) + Send + 'static,
mut on_notify: impl FnMut(&mut T, E, &mut ModelContext<'_, T>) + 'static,
) -> Subscription
where
T: 'static + Send,
T: 'static,
T2: 'static,
E: Entity<T2>,
{
@ -69,10 +69,10 @@ impl<'a, T: 'static> ModelContext<'a, T> {
pub fn subscribe<T2, E>(
&mut self,
entity: &E,
mut on_event: impl FnMut(&mut T, E, &T2::Event, &mut ModelContext<'_, T>) + Send + 'static,
mut on_event: impl FnMut(&mut T, E, &T2::Event, &mut ModelContext<'_, T>) + 'static,
) -> Subscription
where
T: 'static + Send,
T: 'static,
T2: 'static + EventEmitter,
E: Entity<T2>,
{
@ -95,7 +95,7 @@ impl<'a, T: 'static> ModelContext<'a, T> {
pub fn on_release(
&mut self,
mut on_release: impl FnMut(&mut T, &mut AppContext) + Send + 'static,
mut on_release: impl FnMut(&mut T, &mut AppContext) + 'static,
) -> Subscription
where
T: 'static,
@ -112,10 +112,10 @@ impl<'a, T: 'static> ModelContext<'a, T> {
pub fn observe_release<T2, E>(
&mut self,
entity: &E,
mut on_release: impl FnMut(&mut T, &mut T2, &mut ModelContext<'_, T>) + Send + 'static,
mut on_release: impl FnMut(&mut T, &mut T2, &mut ModelContext<'_, T>) + 'static,
) -> Subscription
where
T: Any + Send,
T: Any,
T2: 'static,
E: Entity<T2>,
{
@ -134,10 +134,10 @@ impl<'a, T: 'static> ModelContext<'a, T> {
pub fn observe_global<G: 'static>(
&mut self,
mut f: impl FnMut(&mut T, &mut ModelContext<'_, T>) + Send + 'static,
mut f: impl FnMut(&mut T, &mut ModelContext<'_, T>) + 'static,
) -> Subscription
where
T: 'static + Send,
T: 'static,
{
let handle = self.weak_model();
self.global_observers.insert(
@ -148,11 +148,11 @@ impl<'a, T: 'static> ModelContext<'a, T> {
pub fn on_app_quit<Fut>(
&mut self,
mut on_quit: impl FnMut(&mut T, &mut ModelContext<T>) -> Fut + Send + 'static,
mut on_quit: impl FnMut(&mut T, &mut ModelContext<T>) -> Fut + 'static,
) -> Subscription
where
Fut: 'static + Future<Output = ()> + Send,
T: 'static + Send,
Fut: 'static + Future<Output = ()>,
T: 'static,
{
let handle = self.weak_model();
self.app.quit_observers.insert(
@ -164,7 +164,7 @@ impl<'a, T: 'static> ModelContext<'a, T> {
future.await;
}
}
.boxed()
.boxed_local()
}),
)
}
@ -183,7 +183,7 @@ impl<'a, T: 'static> ModelContext<'a, T> {
pub fn update_global<G, R>(&mut self, f: impl FnOnce(&mut G, &mut Self) -> R) -> R
where
G: 'static + Send,
G: 'static,
{
let mut global = self.app.lease_global::<G>();
let result = f(&mut global, self);

8
crates/gpui2/src/app/test_context.rs
View file

@ -135,18 +135,20 @@ impl TestAppContext {
}
}
pub fn subscribe<T: 'static + EventEmitter + Send>(
pub fn subscribe<T: 'static + EventEmitter>(
&mut self,
entity: &Model<T>,
) -> futures::channel::mpsc::UnboundedReceiver<T::Event>
where
T::Event: 'static + Send + Clone,
T::Event: 'static + Clone,
{
let (mut tx, rx) = futures::channel::mpsc::unbounded();
entity
.update(self, |_, cx: &mut ModelContext<T>| {
cx.subscribe(entity, move |_, _, event, cx| {
cx.executor().block(tx.send(event.clone())).unwrap();
cx.background_executor()
.block(tx.send(event.clone()))
.unwrap();
})
})
.detach();

2
crates/gpui2/src/assets.rs
View file

@ -8,7 +8,7 @@ use std::{
sync::atomic::{AtomicUsize, Ordering::SeqCst},
};
pub trait AssetSource: 'static + Sync {
pub trait AssetSource: 'static + Send + Sync {
fn load(&self, path: &str) -> Result<Cow<[u8]>>;
fn list(&self, path: &str) -> Result<Vec<SharedString>>;
}

2
crates/gpui2/src/executor.rs
View file

@ -53,7 +53,7 @@ where
E: 'static + Send + Debug,
{
pub fn detach_and_log_err(self, cx: &mut AppContext) {
cx.executor().spawn(self.log_err()).detach();
cx.background_executor().spawn(self.log_err()).detach();
}
}

2
crates/journal2/src/journal2.rs
View file

@ -77,7 +77,7 @@ pub fn new_journal_entry(_: Arc<AppState>, cx: &mut AppContext) {
let now = now.time();
let _entry_heading = heading_entry(now, &settings.hour_format);
let _create_entry = cx.executor().spawn(async move {
let _create_entry = cx.background_executor().spawn(async move {
std::fs::create_dir_all(month_dir)?;
OpenOptions::new()
.create(true)

16
crates/language2/src/buffer.rs
View file

@ -652,7 +652,7 @@ impl Buffer {
if !self.is_dirty() {
let reload = self.reload(cx).log_err().map(drop);
task = cx.executor().spawn(reload);
task = cx.background_executor().spawn(reload);
}
}
}
@ -684,7 +684,7 @@ impl Buffer {
let snapshot = self.snapshot();
let mut diff = self.git_diff.clone();
let diff = cx.executor().spawn(async move {
let diff = cx.background_executor().spawn(async move {
diff.update(&diff_base, &snapshot).await;
diff
});
@ -793,7 +793,7 @@ impl Buffer {
let mut syntax_snapshot = syntax_map.snapshot();
drop(syntax_map);
let parse_task = cx.executor().spawn({
let parse_task = cx.background_executor().spawn({
let language = language.clone();
let language_registry = language_registry.clone();
async move {
@ -803,7 +803,7 @@ impl Buffer {
});
match cx
.executor()
.background_executor()
.block_with_timeout(self.sync_parse_timeout, parse_task)
{
Ok(new_syntax_snapshot) => {
@ -866,9 +866,9 @@ impl Buffer {
fn request_autoindent(&mut self, cx: &mut ModelContext<Self>) {
if let Some(indent_sizes) = self.compute_autoindents() {
let indent_sizes = cx.executor().spawn(indent_sizes);
let indent_sizes = cx.background_executor().spawn(indent_sizes);
match cx
.executor()
.background_executor()
.block_with_timeout(Duration::from_micros(500), indent_sizes)
{
Ok(indent_sizes) => self.apply_autoindents(indent_sizes, cx),
@ -1117,7 +1117,7 @@ impl Buffer {
pub fn diff(&self, mut new_text: String, cx: &AppContext) -> Task<Diff> {
let old_text = self.as_rope().clone();
let base_version = self.version();
cx.executor().spawn(async move {
cx.background_executor().spawn(async move {
let old_text = old_text.to_string();
let line_ending = LineEnding::detect(&new_text);
LineEnding::normalize(&mut new_text);
@ -1155,7 +1155,7 @@ impl Buffer {
let old_text = self.as_rope().clone();
let line_ending = self.line_ending();
let base_version = self.version();
cx.executor().spawn(async move {
cx.background_executor().spawn(async move {
let ranges = trailing_whitespace_ranges(&old_text);
let empty = Arc::<str>::from("");
Diff {

10
crates/language2/src/buffer_tests.rs
View file

@ -559,7 +559,7 @@ async fn test_outline(cx: &mut gpui2::TestAppContext) {
cx: &'a gpui2::TestAppContext,
) -> Vec<(&'a str, Vec<usize>)> {
let matches = cx
.update(|cx| outline.search(query, cx.executor().clone()))
.update(|cx| outline.search(query, cx.background_executor().clone()))
.await;
matches
.into_iter()
@ -1879,7 +1879,7 @@ fn test_serialization(cx: &mut gpui2::AppContext) {
let state = buffer1.read(cx).to_proto();
let ops = cx
.executor()
.background_executor()
.block(buffer1.read(cx).serialize_ops(None, cx));
let buffer2 = cx.build_model(|cx| {
let mut buffer = Buffer::from_proto(1, state, None).unwrap();
@ -1921,7 +1921,7 @@ fn test_random_collaboration(cx: &mut AppContext, mut rng: StdRng) {
let buffer = cx.build_model(|cx| {
let state = base_buffer.read(cx).to_proto();
let ops = cx
.executor()
.background_executor()
.block(base_buffer.read(cx).serialize_ops(None, cx));
let mut buffer = Buffer::from_proto(i as ReplicaId, state, None).unwrap();
buffer
@ -2025,7 +2025,9 @@ fn test_random_collaboration(cx: &mut AppContext, mut rng: StdRng) {
}
50..=59 if replica_ids.len() < max_peers => {
let old_buffer_state = buffer.read(cx).to_proto();
let old_buffer_ops = cx.executor().block(buffer.read(cx).serialize_ops(None, cx));
let old_buffer_ops = cx
.background_executor()
.block(buffer.read(cx).serialize_ops(None, cx));
let new_replica_id = (0..=replica_ids.len() as ReplicaId)
.filter(|replica_id| *replica_id != buffer.read(cx).replica_id())
.choose(&mut rng)

10
crates/lsp2/src/lsp2.rs
View file

@ -595,8 +595,8 @@ impl LanguageServer {
where
T: request::Request,
T::Params: 'static + Send,
F: 'static + Send + FnMut(T::Params, AsyncAppContext) -> Fut,
Fut: 'static + Future<Output = Result<T::Result>> + Send,
F: 'static + FnMut(T::Params, AsyncAppContext) -> Fut + Send,
Fut: 'static + Future<Output = Result<T::Result>>,
{
self.on_custom_request(T::METHOD, f)
}
@ -629,7 +629,7 @@ impl LanguageServer {
#[must_use]
pub fn on_custom_notification<Params, F>(&self, method: &'static str, mut f: F) -> Subscription
where
F: 'static + Send + FnMut(Params, AsyncAppContext),
F: 'static + FnMut(Params, AsyncAppContext) + Send,
Params: DeserializeOwned,
{
let prev_handler = self.notification_handlers.lock().insert(
@ -657,8 +657,8 @@ impl LanguageServer {
mut f: F,
) -> Subscription
where
F: 'static + Send + FnMut(Params, AsyncAppContext) -> Fut,
Fut: 'static + Future<Output = Result<Res>> + Send,
F: 'static + FnMut(Params, AsyncAppContext) -> Fut + Send,
Fut: 'static + Future<Output = Result<Res>>,
Params: DeserializeOwned + Send + 'static,
Res: Serialize,
{

2
crates/prettier2/src/prettier2.rs
View file

@ -143,7 +143,7 @@ impl Prettier {
) -> anyhow::Result<Self> {
use lsp2::LanguageServerBinary;
let executor = cx.executor().clone();
let executor = cx.background_executor().clone();
anyhow::ensure!(
prettier_dir.is_dir(),
"Prettier dir {prettier_dir:?} is not a directory"

24
crates/project2/src/lsp_command.rs
View file

@ -32,7 +32,7 @@ pub fn lsp_formatting_options(tab_size: u32) -> lsp2::FormattingOptions {
}
}
#[async_trait]
#[async_trait(?Send)]
pub(crate) trait LspCommand: 'static + Sized + Send {
type Response: 'static + Default + Send;
type LspRequest: 'static + Send + lsp2::request::Request;
@ -148,7 +148,7 @@ impl From<lsp2::FormattingOptions> for FormattingOptions {
}
}
#[async_trait]
#[async_trait(?Send)]
impl LspCommand for PrepareRename {
type Response = Option<Range<Anchor>>;
type LspRequest = lsp2::request::PrepareRenameRequest;
@ -279,7 +279,7 @@ impl LspCommand for PrepareRename {
}
}
#[async_trait]
#[async_trait(?Send)]
impl LspCommand for PerformRename {
type Response = ProjectTransaction;
type LspRequest = lsp2::request::Rename;
@ -398,7 +398,7 @@ impl LspCommand for PerformRename {
}
}
#[async_trait]
#[async_trait(?Send)]
impl LspCommand for GetDefinition {
type Response = Vec<LocationLink>;
type LspRequest = lsp2::request::GotoDefinition;
@ -491,7 +491,7 @@ impl LspCommand for GetDefinition {
}
}
#[async_trait]
#[async_trait(?Send)]
impl LspCommand for GetTypeDefinition {
type Response = Vec<LocationLink>;
type LspRequest = lsp2::request::GotoTypeDefinition;
@ -783,7 +783,7 @@ fn location_links_to_proto(
.collect()
}
#[async_trait]
#[async_trait(?Send)]
impl LspCommand for GetReferences {
type Response = Vec<Location>;
type LspRequest = lsp2::request::References;
@ -945,7 +945,7 @@ impl LspCommand for GetReferences {
}
}
#[async_trait]
#[async_trait(?Send)]
impl LspCommand for GetDocumentHighlights {
type Response = Vec<DocumentHighlight>;
type LspRequest = lsp2::request::DocumentHighlightRequest;
@ -1096,7 +1096,7 @@ impl LspCommand for GetDocumentHighlights {
}
}
#[async_trait]
#[async_trait(?Send)]
impl LspCommand for GetHover {
type Response = Option<Hover>;
type LspRequest = lsp2::request::HoverRequest;
@ -1314,7 +1314,7 @@ impl LspCommand for GetHover {
}
}
#[async_trait]
#[async_trait(?Send)]
impl LspCommand for GetCompletions {
type Response = Vec<Completion>;
type LspRequest = lsp2::request::Completion;
@ -1545,7 +1545,7 @@ impl LspCommand for GetCompletions {
}
}
#[async_trait]
#[async_trait(?Send)]
impl LspCommand for GetCodeActions {
type Response = Vec<CodeAction>;
type LspRequest = lsp2::request::CodeActionRequest;
@ -1684,7 +1684,7 @@ impl LspCommand for GetCodeActions {
}
}
#[async_trait]
#[async_trait(?Send)]
impl LspCommand for OnTypeFormatting {
type Response = Option<Transaction>;
type LspRequest = lsp2::request::OnTypeFormatting;
@ -2192,7 +2192,7 @@ impl InlayHints {
}
}
#[async_trait]
#[async_trait(?Send)]
impl LspCommand for InlayHints {
type Response = Vec<InlayHint>;
type LspRequest = lsp2::InlayHintRequest;

24
crates/project2/src/project2.rs
View file

@ -1758,7 +1758,7 @@ impl Project {
}
};
cx.executor().spawn(async move {
cx.background_executor().spawn(async move {
wait_for_loading_buffer(loading_watch)
.await
.map_err(|error| anyhow!("{}", error))
@ -5593,7 +5593,7 @@ impl Project {
})
.collect::<Vec<_>>();
let background = cx.executor().clone();
let background = cx.background_executor().clone();
let path_count: usize = snapshots.iter().map(|s| s.visible_file_count()).sum();
if path_count == 0 {
let (_, rx) = smol::channel::bounded(1024);
@ -5616,11 +5616,11 @@ impl Project {
}
})
.collect();
cx.executor()
cx.background_executor()
.spawn(Self::background_search(
unnamed_files,
opened_buffers,
cx.executor().clone(),
cx.background_executor().clone(),
self.fs.clone(),
workers,
query.clone(),
@ -5631,9 +5631,9 @@ impl Project {
.detach();
let (buffers, buffers_rx) = Self::sort_candidates_and_open_buffers(matching_paths_rx, cx);
let background = cx.executor().clone();
let background = cx.background_executor().clone();
let (result_tx, result_rx) = smol::channel::bounded(1024);
cx.executor()
cx.background_executor()
.spawn(async move {
let Ok(buffers) = buffers.await else {
return;
@ -5993,7 +5993,7 @@ impl Project {
Task::ready(Ok((tree, relative_path)))
} else {
let worktree = self.create_local_worktree(abs_path, visible, cx);
cx.executor()
cx.background_executor()
.spawn(async move { Ok((worktree.await?, PathBuf::new())) })
}
}
@ -6064,7 +6064,7 @@ impl Project {
.shared()
})
.clone();
cx.executor().spawn(async move {
cx.background_executor().spawn(async move {
match task.await {
Ok(worktree) => Ok(worktree),
Err(err) => Err(anyhow!("{}", err)),
@ -6519,7 +6519,7 @@ impl Project {
})
.collect::<Vec<_>>();
cx.executor()
cx.background_executor()
.spawn(async move {
for task_result in future::join_all(prettiers_to_reload.into_iter().map(|(worktree_id, prettier_path, prettier_task)| {
async move {
@ -7358,7 +7358,7 @@ impl Project {
})
.log_err();
cx.executor()
cx.background_executor()
.spawn(
async move {
let operations = operations.await;
@ -7960,7 +7960,7 @@ impl Project {
if let Some(buffer) = this.buffer_for_id(buffer_id) {
let operations =
buffer.read(cx).serialize_ops(Some(remote_version), cx);
cx.executor().spawn(async move {
cx.background_executor().spawn(async move {
let operations = operations.await;
for chunk in split_operations(operations) {
client
@ -8198,7 +8198,7 @@ impl Project {
cx: &mut ModelContext<Self>,
) -> Task<Result<Vec<(Range<Anchor>, String)>>> {
let snapshot = self.buffer_snapshot_for_lsp_version(buffer, server_id, version, cx);
cx.executor().spawn(async move {
cx.background_executor().spawn(async move {
let snapshot = snapshot?;
let mut lsp_edits = lsp_edits
.into_iter()

25
crates/project2/src/worktree.rs
View file

@ -365,10 +365,10 @@ impl Worktree {
})
.detach();
let background_scanner_task = cx.executor().spawn({
let background_scanner_task = cx.background_executor().spawn({
let fs = fs.clone();
let snapshot = snapshot.clone();
let background = cx.executor().clone();
let background = cx.background_executor().clone();
async move {
let events = fs.watch(&abs_path, Duration::from_millis(100)).await;
BackgroundScanner::new(
@ -429,7 +429,7 @@ impl Worktree {
let background_snapshot = Arc::new(Mutex::new(snapshot.clone()));
let (mut snapshot_updated_tx, mut snapshot_updated_rx) = watch::channel();
cx.executor()
cx.background_executor()
.spawn({
let background_snapshot = background_snapshot.clone();
async move {
@ -1008,7 +1008,7 @@ impl LocalWorktree {
let lowest_ancestor = self.lowest_ancestor(&path);
let abs_path = self.absolutize(&path);
let fs = self.fs.clone();
let write = cx.executor().spawn(async move {
let write = cx.background_executor().spawn(async move {
if is_dir {
fs.create_dir(&abs_path).await
} else {
@ -1058,7 +1058,7 @@ impl LocalWorktree {
let abs_path = self.absolutize(&path);
let fs = self.fs.clone();
let write = cx
.executor()
.background_executor()
.spawn(async move { fs.save(&abs_path, &text, line_ending).await });
cx.spawn(|this, mut cx| async move {
@ -1079,7 +1079,7 @@ impl LocalWorktree {
let abs_path = self.absolutize(&entry.path);
let fs = self.fs.clone();
let delete = cx.executor().spawn(async move {
let delete = cx.background_executor().spawn(async move {
if entry.is_file() {
fs.remove_file(&abs_path, Default::default()).await?;
} else {
@ -1119,7 +1119,7 @@ impl LocalWorktree {
let abs_old_path = self.absolutize(&old_path);
let abs_new_path = self.absolutize(&new_path);
let fs = self.fs.clone();
let rename = cx.executor().spawn(async move {
let rename = cx.background_executor().spawn(async move {
fs.rename(&abs_old_path, &abs_new_path, Default::default())
.await
});
@ -1146,7 +1146,7 @@ impl LocalWorktree {
let abs_old_path = self.absolutize(&old_path);
let abs_new_path = self.absolutize(&new_path);
let fs = self.fs.clone();
let copy = cx.executor().spawn(async move {
let copy = cx.background_executor().spawn(async move {
copy_recursive(
fs.as_ref(),
&abs_old_path,
@ -1174,7 +1174,7 @@ impl LocalWorktree {
) -> Option<Task<Result<()>>> {
let path = self.entry_for_id(entry_id)?.path.clone();
let mut refresh = self.refresh_entries_for_paths(vec![path]);
Some(cx.executor().spawn(async move {
Some(cx.background_executor().spawn(async move {
refresh.next().await;
Ok(())
}))
@ -1248,7 +1248,7 @@ impl LocalWorktree {
.ok();
let worktree_id = cx.entity_id().as_u64();
let _maintain_remote_snapshot = cx.executor().spawn(async move {
let _maintain_remote_snapshot = cx.background_executor().spawn(async move {
let mut is_first = true;
while let Some((snapshot, entry_changes, repo_changes)) = snapshots_rx.next().await {
let update;
@ -1306,7 +1306,7 @@ impl LocalWorktree {
let rx = self.observe_updates(project_id, cx, move |update| {
client.request(update).map(|result| result.is_ok())
});
cx.executor()
cx.background_executor()
.spawn(async move { rx.await.map_err(|_| anyhow!("share ended")) })
}
@ -2672,7 +2672,8 @@ impl language2::LocalFile for File {
let worktree = self.worktree.read(cx).as_local().unwrap();
let abs_path = worktree.absolutize(&self.path);
let fs = worktree.fs.clone();
cx.executor().spawn(async move { fs.load(&abs_path).await })
cx.background_executor()
.spawn(async move { fs.load(&abs_path).await })
}
fn buffer_reloaded(

5
crates/settings2/src/settings_file.rs
View file

@ -63,7 +63,10 @@ pub fn handle_settings_file_changes(
mut user_settings_file_rx: mpsc::UnboundedReceiver<String>,
cx: &mut AppContext,
) {
let user_settings_content = cx.executor().block(user_settings_file_rx.next()).unwrap();
let user_settings_content = cx
.background_executor()
.block(user_settings_file_rx.next())
.unwrap();
cx.update_global(|store: &mut SettingsStore, cx| {
store
.set_user_settings(&user_settings_content, cx)

4
crates/terminal2/src/terminal2.rs
View file

@ -984,7 +984,7 @@ impl Terminal {
term.lock_unfair() //It's been too long, force block
} else if let None = self.sync_task {
//Skip this frame
let delay = cx.executor().timer(Duration::from_millis(16));
let delay = cx.background_executor().timer(Duration::from_millis(16));
self.sync_task = Some(cx.spawn(|weak_handle, mut cx| async move {
delay.await;
if let Some(handle) = weak_handle.upgrade() {
@ -1302,7 +1302,7 @@ impl Terminal {
cx: &mut ModelContext<Self>,
) -> Task<Vec<RangeInclusive<AlacPoint>>> {
let term = self.term.clone();
cx.executor().spawn(async move {
cx.background_executor().spawn(async move {
let term = term.lock();
all_search_matches(&term, &searcher).collect()

6
crates/zed2/src/main.rs
View file

@ -117,7 +117,7 @@ fn main() {
let client = client2::Client::new(http.clone(), cx);
let mut languages = LanguageRegistry::new(login_shell_env_loaded);
let copilot_language_server_id = languages.next_language_server_id();
languages.set_executor(cx.executor().clone());
languages.set_executor(cx.background_executor().clone());
languages.set_language_server_download_dir(paths::LANGUAGES_DIR.clone());
let languages = Arc::new(languages);
let node_runtime = RealNodeRuntime::new(http.clone());
@ -514,7 +514,7 @@ fn init_panic_hook(app: &App, installation_id: Option<String>, session_id: Strin
fn upload_previous_panics(http: Arc<dyn HttpClient>, cx: &mut AppContext) {
let telemetry_settings = *client2::TelemetrySettings::get_global(cx);
cx.executor()
cx.background_executor()
.spawn(async move {
let panic_report_url = format!("{}/api/panic", &*client2::ZED_SERVER_URL);
let mut children = smol::fs::read_dir(&*paths::LOGS_DIR).await?;
@ -644,7 +644,7 @@ fn load_embedded_fonts(cx: &AppContext) {
let asset_source = cx.asset_source();
let font_paths = asset_source.list("fonts").unwrap();
let embedded_fonts = Mutex::new(Vec::new());
let executor = cx.executor();
let executor = cx.background_executor();
executor.block(executor.scoped(|scope| {
for font_path in &font_paths {