30c9399 · helix · git.iko.soy

git.iko.soy/helix

on master help sign in

Use a single save_queue on the editor

30c9399Blaž Hrastnik4 years ago

helix-term/src/application.rs+19−1

compare with master

unifiedsplit

@@ -1,5 +1,5 @@

1use arc_swap::{access::Map, ArcSwap};

2−use futures_util::Stream;

2+use futures_util::{Stream, StreamExt};

3use helix_core::{

4 diagnostic::{DiagnosticTag, NumberOrString},

5 path::get_relative_path,

@@ -968,6 +968,24 @@ impl Application {

968 // errors along the way

969 let mut errs = Vec::new();

970

971+ // TODO: deduplicate with ctx.block_try_flush_writes

972+ tokio::task::block_in_place(|| {

973+ helix_lsp::block_on(async {

974+ while let Some(save_event) = self.editor.save_queue.next().await {

975+ match &save_event {

976+ Ok(event) => {

977+ let doc = doc_mut!(self.editor, &event.doc_id);

978+ doc.set_last_saved_revision(event.revision);

979+ }

980+ Err(err) => {

981+ log::error!("error saving document: {}", err);

982+ }

983+ };

984+ // TODO: if is_err: break?

985+ }

986+ })

987+ });

988+

971 if let Err(err) = self

989 if let Err(err) = self

972 .jobs

990 .jobs

973 .finish(&mut self.editor, Some(&mut self.compositor))

991 .finish(&mut self.editor, Some(&mut self.compositor))

helix-term/src/commands.rs+2−1

@@ -2541,7 +2541,8 @@ async fn make_format_callback(

2541 }

2542

2543 if let Some((path, force)) = write {

2544+ let id = doc.id();

2544− if let Err(err) = doc.save(path, force) {

2545+ if let Err(err) = editor.save(id, path, force) {

2545 editor.set_error(format!("Error saving: {}", err));

2546 editor.set_error(format!("Error saving: {}", err));

2546 }

2547 }

2548 }

helix-term/src/commands/typed.rs+54−34

@@ -79,12 +79,28 @@ fn buffer_close_by_ids_impl(

79 doc_ids: &[DocumentId],

80 force: bool,

81) -> anyhow::Result<()> {

82+ // TODO: deduplicate with ctx.block_try_flush_writes

83+ tokio::task::block_in_place(|| {

84+ helix_lsp::block_on(async {

85+ while let Some(save_event) = editor.save_queue.next().await {

86+ match &save_event {

87+ Ok(event) => {

88+ let doc = doc_mut!(editor, &event.doc_id);

89+ doc.set_last_saved_revision(event.revision);

90+ }

91+ Err(err) => {

92+ log::error!("error saving document: {}", err);

93+ }

94+ };

95+ // TODO: if is_err: break?

96+ }

97+ })

98+ });

99+

82 let (modified_ids, modified_names): (Vec<_>, Vec<_>) = doc_ids

100 let (modified_ids, modified_names): (Vec<_>, Vec<_>) = doc_ids

83 .iter()

101 .iter()

84 .filter_map(|&doc_id| {

102 .filter_map(|&doc_id| {

85− if let Err(CloseError::BufferModified(name)) = tokio::task::block_in_place(|| {

103+ if let Err(CloseError::BufferModified(name)) = editor.close_document(doc_id, force) {

86− helix_lsp::block_on(editor.close_document(doc_id, force))

87− }) {

88 Some((doc_id, name))

104 Some((doc_id, name))

89 } else {

105 } else {

90 None

106 None

@@ -289,7 +305,8 @@ fn write_impl(

289 };

305 };

290

306

291 if fmt.is_none() {

307 if fmt.is_none() {

308+ let id = doc.id();

292− doc.save(path, force)?;

309+ cx.editor.save(id, path, force)?;

293 }

310 }

294

311

295 Ok(())

312 Ok(())

@@ -569,40 +586,45 @@ fn write_all_impl(

569 return Ok(());

586 return Ok(());

570 }

587 }

571

588

572− let mut errors: Option<String> = None;

589+ let mut errors: Vec<&'static str> = Vec::new();

573 let auto_format = cx.editor.config().auto_format;

590 let auto_format = cx.editor.config().auto_format;

574 let jobs = &mut cx.jobs;

591 let jobs = &mut cx.jobs;

575

592

576 // save all documents

593 // save all documents

577− for doc in &mut cx.editor.documents.values_mut() {

594+ let saves: Vec<_> = cx

595+ .editor

596+ .documents

597+ .values()

598+ .filter_map(|doc| {

578− if doc.path().is_none() {

599+ if doc.path().is_none() {

579− errors = errors

580− .or_else(|| Some(String::new()))

581− .map(|mut errs: String| {

582− errs.push_str("cannot write a buffer without a filename\n");

600+ errors.push("cannot write a buffer without a filename\n");

583− errs

601+ return None;

584− });

585−

586− continue;

587− }

602+ }

588

603

589− if !doc.is_modified() {

604+ if !doc.is_modified() {

590− continue;

605+ return None;

591− }

606+ }

592

607

593− let fmt = if auto_format {

608+ let fmt = if auto_format {

594− doc.auto_format().map(|fmt| {

609+ doc.auto_format().map(|fmt| {

595− let callback =

610+ let callback =

596− make_format_callback(doc.id(), doc.version(), fmt, Some((None, force)));

611+ make_format_callback(doc.id(), doc.version(), fmt, Some((None, force)));

597− jobs.add(Job::with_callback(callback).wait_before_exiting());

612+ jobs.add(Job::with_callback(callback).wait_before_exiting());

598− })

613+ })

599− } else {

614+ } else {

615+ None

616+ };

617+

618+ if fmt.is_none() {

619+ return Some(doc.id());

620+ }

600 None

621 None

601− };

622+ })

623+ .collect();

602

624

603− if fmt.is_none() {

625+ // manually call save for the rest of docs that don't have a formatter

626+ for id in saves {

604− doc.save::<PathBuf>(None, force)?;

627+ cx.editor.save::<PathBuf>(id, None, force)?;

605− }

606 }

628 }

607

629

608 if quit {

630 if quit {

@@ -619,10 +641,8 @@ fn write_all_impl(

619 }

641 }

620 }

642 }

621

643

622− if let Some(errs) = errors {

644+ if !errors.is_empty() && !force {

623− if !force {

624− bail!(errs);

645+ bail!("{:?}", errors);

625− }

626 }

646 }

627

647

628 Ok(())

648 Ok(())

helix-term/src/compositor.rs+17−5

@@ -1,3 +1,4 @@

1+use futures_util::StreamExt;

1// Each component declares it's own size constraints and gets fitted based on it's parent.

2// Each component declares it's own size constraints and gets fitted based on it's parent.

2// Q: how does this work with popups?

3// Q: how does this work with popups?

3// cursive does compositor.screen_mut().add_layer_at(pos::absolute(x, y), <component>)

4// cursive does compositor.screen_mut().add_layer_at(pos::absolute(x, y), <component>)

@@ -33,11 +34,22 @@ impl<'a> Context<'a> {

33 pub fn block_try_flush_writes(&mut self) -> anyhow::Result<()> {

34 pub fn block_try_flush_writes(&mut self) -> anyhow::Result<()> {

34 tokio::task::block_in_place(|| helix_lsp::block_on(self.jobs.finish(self.editor, None)))?;

35 tokio::task::block_in_place(|| helix_lsp::block_on(self.jobs.finish(self.editor, None)))?;

36− for doc in &mut self.editor.documents.values_mut() {

37− tokio::task::block_in_place(|| helix_lsp::block_on(doc.try_flush_saves()))

37+ tokio::task::block_in_place(|| {

38− .map(|result| result.map(|_| ()))

38+ helix_lsp::block_on(async {

39− .unwrap_or(Ok(()))?;

39+ while let Some(save_event) = self.editor.save_queue.next().await {

40+ match &save_event {

41+ Ok(event) => {

42+ let doc = doc_mut!(self.editor, &event.doc_id);

43+ doc.set_last_saved_revision(event.revision);

44+ }

45+ Err(err) => {

46+ log::error!("error saving document: {}", err);

47+ }

48+ };

49+ // TODO: if is_err: break?

50+ }

51+ })

40− }

52+ });

42 Ok(())

54 Ok(())

43 }

55 }

helix-view/src/document.rs+21−119

@@ -13,10 +13,6 @@ use std::future::Future;

13use std::path::{Path, PathBuf};

14use std::str::FromStr;

15use std::sync::Arc;

16−use tokio::sync::mpsc::error::TryRecvError;

17−use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender};

18−

19−use tokio::sync::Mutex;

21use helix_core::{

17use helix_core::{

22 encoding,

18 encoding,

@@ -134,9 +130,6 @@ pub struct Document {

134 last_saved_revision: usize,

130 last_saved_revision: usize,

135 version: i32, // should be usize?

131 version: i32, // should be usize?

136 pub(crate) modified_since_accessed: bool,

132 pub(crate) modified_since_accessed: bool,

137− save_sender: Option<UnboundedSender<DocumentSavedEventFuture>>,

138− save_receiver: Option<UnboundedReceiver<DocumentSavedEventFuture>>,

139− current_save: Arc<Mutex<Option<DocumentSavedEventFuture>>>,

140

133

141 diagnostics: Vec<Diagnostic>,

134 diagnostics: Vec<Diagnostic>,

142 language_server: Option<Arc<helix_lsp::Client>>,

135 language_server: Option<Arc<helix_lsp::Client>>,

@@ -357,7 +350,6 @@ impl Document {

357 let encoding = encoding.unwrap_or(encoding::UTF_8);

350 let encoding = encoding.unwrap_or(encoding::UTF_8);

358 let changes = ChangeSet::new(&text);

351 let changes = ChangeSet::new(&text);

359 let old_state = None;

352 let old_state = None;

360− let (save_sender, save_receiver) = tokio::sync::mpsc::unbounded_channel();

361

353

362 Self {

354 Self {

363 id: DocumentId::default(),

355 id: DocumentId::default(),

@@ -378,9 +370,6 @@ impl Document {

378 savepoint: None,

370 savepoint: None,

379 last_saved_revision: 0,

371 last_saved_revision: 0,

380 modified_since_accessed: false,

372 modified_since_accessed: false,

381− save_sender: Some(save_sender),

382− save_receiver: Some(save_receiver),

383− current_save: Arc::new(Mutex::new(None)),

384 language_server: None,

373 language_server: None,

385 }

374 }

386 }

375 }

@@ -519,21 +508,26 @@ impl Document {

519 &mut self,

508 &mut self,

520 path: Option<P>,

509 path: Option<P>,

521 force: bool,

510 force: bool,

522− ) -> Result<(), anyhow::Error> {

511+ ) -> Result<

512+ impl Future<Output = Result<DocumentSavedEvent, anyhow::Error>> + 'static + Send,

513+ anyhow::Error,

514+ > {

515+ let path = path.map(|path| path.into());

516+ self.save_impl(path, force)

517+

523− self.save_impl::<futures_util::future::Ready<_>, _>(path, force)

518+ // futures_util::future::Ready<_>,

524 }

519 }

525

520

526 /// The `Document`'s text is encoded according to its encoding and written to the file located

521 /// The `Document`'s text is encoded according to its encoding and written to the file located

527 /// at its `path()`.

522 /// at its `path()`.

528− fn save_impl<F, P>(&mut self, path: Option<P>, force: bool) -> Result<(), anyhow::Error>

523+ fn save_impl(

529− where

524+ &mut self,

525+ path: Option<PathBuf>,

526+ force: bool,

527+ ) -> Result<

530− F: Future<Output = Result<Transaction, FormatterError>> + 'static + Send,

528+ impl Future<Output = Result<DocumentSavedEvent, anyhow::Error>> + 'static + Send,

531− P: Into<PathBuf>,

529+ anyhow::Error,

532− {

530+ > {

533− if self.save_sender.is_none() {

534− bail!("saves are closed for this document!");

535− }

536−

537 log::debug!(

531 log::debug!(

538 "submitting save of doc '{:?}'",

532 "submitting save of doc '{:?}'",

539 self.path().map(|path| path.to_string_lossy())

533 self.path().map(|path| path.to_string_lossy())

@@ -544,7 +538,7 @@ impl Document {

544 let text = self.text().clone();

538 let text = self.text().clone();

545

539

546 let path = match path {

540 let path = match path {

547− Some(path) => helix_core::path::get_canonicalized_path(&path.into())?,

541+ Some(path) => helix_core::path::get_canonicalized_path(&path)?,

548 None => {

542 None => {

549 if self.path.is_none() {

543 if self.path.is_none() {

550 bail!("Can't save with no path set!");

544 bail!("Can't save with no path set!");

@@ -564,7 +558,7 @@ impl Document {

564 let encoding = self.encoding;

558 let encoding = self.encoding;

565

559

566 // We encode the file according to the `Document`'s encoding.

560 // We encode the file according to the `Document`'s encoding.

567− let save_event = async move {

561+ let future = async move {

568 use tokio::fs::File;

562 use tokio::fs::File;

569 if let Some(parent) = path.parent() {

563 if let Some(parent) = path.parent() {

570 // TODO: display a prompt asking the user if the directories should be created

564 // TODO: display a prompt asking the user if the directories should be created

@@ -604,107 +598,15 @@ impl Document {

604 Ok(event)

598 Ok(event)

605 };

599 };

606

600

607− self.save_sender

601+ Ok(future)

608− .as_mut()

609− .unwrap()

610− .send(Box::pin(save_event))

611− .map_err(|err| anyhow!("failed to send save event: {}", err))

612− }

613−

614− pub async fn await_save(&mut self) -> Option<DocumentSavedEventResult> {

615− self.await_save_impl(true).await

616− }

617−

618− async fn await_save_impl(&mut self, block: bool) -> Option<DocumentSavedEventResult> {

619− let mut current_save = self.current_save.lock().await;

620− if let Some(ref mut save) = *current_save {

621− log::trace!("reawaiting save of '{:?}'", self.path());

622− let result = save.await;

623− *current_save = None;

624− log::trace!("reawait save of '{:?}' result: {:?}", self.path(), result);

625− return Some(result);

626− }

627−

628− // return early if the receiver is closed

629− let rx = self.save_receiver.as_mut()?;

630−

631− let save_req = if block {

632− rx.recv().await

633− } else {

634− let msg = rx.try_recv();

635−

636− if let Err(err) = msg {

637− match err {

638− TryRecvError::Empty => return None,

639− TryRecvError::Disconnected => None,

640− }

641− } else {

642− msg.ok()

643− }

644− };

645−

646− let save = match save_req {

647− Some(save) => save,

648− None => {

649− self.save_receiver = None;

650− return None;

651− }

652− };

653−

654− // save a handle to the future so that when a poll on this

655− // function gets cancelled, we don't lose it

656− *current_save = Some(save);

657− log::trace!("awaiting save of '{:?}'", self.path());

658−

659− let result = (*current_save).as_mut().unwrap().await;

660− *current_save = None;

661−

662− log::trace!("save of '{:?}' result: {:?}", self.path(), result);

663−

664− Some(result)

665− }

666−

667− /// Flushes the queue of pending writes. If any fail,

668− /// it stops early before emptying the rest of the queue.

669− pub async fn try_flush_saves(&mut self) -> Option<DocumentSavedEventResult> {

670− self.flush_saves_impl(false).await

671− }

672−

673− async fn flush_saves_impl(&mut self, block: bool) -> Option<DocumentSavedEventResult> {

674− let mut final_result = None;

675−

676− while let Some(save_event) = self.await_save_impl(block).await {

677− let is_err = match &save_event {

678− Ok(event) => {

679− self.set_last_saved_revision(event.revision);

680− false

681− }

682− Err(err) => {

683− log::error!(

684− "error saving document {:?}: {}",

685− self.path().map(|path| path.to_string_lossy()),

686− err

687− );

688− true

689− }

690− };

691−

692− final_result = Some(save_event);

693−

694− if is_err {

695− break;

696− }

697− }

698−

699− final_result

700 }

602 }

701

603

702 /// Prepares the Document for being closed by stopping any new writes

604 /// Prepares the Document for being closed by stopping any new writes

703 /// and flushing through the queue of pending writes. If any fail,

605 /// and flushing through the queue of pending writes. If any fail,

704 /// it stops early before emptying the rest of the queue.

606 /// it stops early before emptying the rest of the queue.

705 pub async fn close(&mut self) -> Option<DocumentSavedEventResult> {

607 pub async fn close(&mut self) -> Option<DocumentSavedEventResult> {

706− self.save_sender.take();

608+ // TODO

707− self.flush_saves_impl(true).await

609+ None

708 }

610 }

709

611

710 /// Detect the programming language based on the file type.

612 /// Detect the programming language based on the file type.

helix-view/src/editor.rs+47−32

@@ -1,6 +1,6 @@

1use crate::{

2 clipboard::{get_clipboard_provider, ClipboardProvider},

3− document::{DocumentSavedEventResult, Mode},

3+ document::{DocumentSavedEventFuture, DocumentSavedEventResult, Mode},

4 graphics::{CursorKind, Rect},

5 info::Info,

6 input::KeyEvent,

@@ -9,7 +9,7 @@ use crate::{

9 Document, DocumentId, View, ViewId,

10};

12−use futures_util::stream::{select_all::SelectAll, FuturesUnordered};

12+use futures_util::stream::select_all::SelectAll;

13use futures_util::{future, StreamExt};

14use helix_lsp::Call;

15use tokio_stream::wrappers::UnboundedReceiverStream;

@@ -29,7 +29,7 @@ use tokio::{

29 time::{sleep, Duration, Instant, Sleep},

30};

32−use anyhow::Error;

32+use anyhow::{anyhow, Error};

34pub use helix_core::diagnostic::Severity;

35pub use helix_core::register::Registers;

@@ -644,12 +644,20 @@ pub struct Breakpoint {

644 pub log_message: Option<String>,

645}

646

647+use futures_util::stream::{Flatten, Once};

648+

647pub struct Editor {

649pub struct Editor {

648 /// Current editing mode.

650 /// Current editing mode.

649 pub mode: Mode,

651 pub mode: Mode,

650 pub tree: Tree,

652 pub tree: Tree,

651 pub next_document_id: DocumentId,

653 pub next_document_id: DocumentId,

652 pub documents: BTreeMap<DocumentId, Document>,

654 pub documents: BTreeMap<DocumentId, Document>,

655+

656+ // We Flatten<> to resolve the inner DocumentSavedEventFuture. For that we need a stream of streams, hence the Once<>.

657+ // https://stackoverflow.com/a/66875668

658+ pub saves: HashMap<DocumentId, UnboundedSender<Once<DocumentSavedEventFuture>>>,

659+ pub save_queue: SelectAll<Flatten<UnboundedReceiverStream<Once<DocumentSavedEventFuture>>>>,

660+

653 pub count: Option<std::num::NonZeroUsize>,

661 pub count: Option<std::num::NonZeroUsize>,

654 pub selected_register: Option<char>,

662 pub selected_register: Option<char>,

655 pub registers: Registers,

663 pub registers: Registers,

@@ -751,6 +759,8 @@ impl Editor {

751 tree: Tree::new(area),

759 tree: Tree::new(area),

752 next_document_id: DocumentId::default(),

760 next_document_id: DocumentId::default(),

753 documents: BTreeMap::new(),

761 documents: BTreeMap::new(),

762+ saves: HashMap::new(),

763+ save_queue: SelectAll::new(),

754 count: None,

764 count: None,

755 selected_register: None,

765 selected_register: None,

756 macro_recording: None,

766 macro_recording: None,

@@ -1083,6 +1093,12 @@ impl Editor {

1083 self.new_document(doc)

1093 self.new_document(doc)

1084 };

1094 };

1085

1095

1096+ let (save_sender, save_receiver) = tokio::sync::mpsc::unbounded_channel();

1097+ self.saves.insert(id, save_sender);

1098+

1099+ let stream = UnboundedReceiverStream::new(save_receiver).flatten();

1100+ self.save_queue.push(stream);

1101+

1086 self.switch(id, action);

1102 self.switch(id, action);

1087 Ok(id)

1103 Ok(id)

1088 }

1104 }

@@ -1095,38 +1111,21 @@ impl Editor {

1095 self._refresh();

1111 self._refresh();

1096 }

1112 }

1097

1113

1098− pub async fn close_document(

1114+ pub fn close_document(&mut self, doc_id: DocumentId, force: bool) -> Result<(), CloseError> {

1099− &mut self,

1100− doc_id: DocumentId,

1101− force: bool,

1102− ) -> Result<(), CloseError> {

1103 let doc = match self.documents.get_mut(&doc_id) {

1115 let doc = match self.documents.get_mut(&doc_id) {

1104 Some(doc) => doc,

1116 Some(doc) => doc,

1105 None => return Err(CloseError::DoesNotExist),

1117 None => return Err(CloseError::DoesNotExist),

1106 };

1118 };

1107−

1108− // flush out any pending writes first to clear the modified status

1109− if let Some(Err(err)) = doc.try_flush_saves().await {

1110− return Err(CloseError::SaveError(err));

1111− }

1112−

1113 if !force && doc.is_modified() {

1119 if !force && doc.is_modified() {

1114 return Err(CloseError::BufferModified(doc.display_name().into_owned()));

1120 return Err(CloseError::BufferModified(doc.display_name().into_owned()));

1115 }

1121 }

1116

1122

1117− if let Some(Err(err)) = doc.close().await {

1123+ // This will also disallow any follow-up writes

1118− return Err(CloseError::SaveError(err));

1124+ self.saves.remove(&doc_id);

1119− }

1120

1125

1121− // Don't fail the whole write because the language server could not

1122− // acknowledge the close

1123 if let Some(language_server) = doc.language_server() {

1126 if let Some(language_server) = doc.language_server() {

1124− if let Err(err) = language_server

1127+ // TODO: track error

1125− .text_document_did_close(doc.identifier())

1128+ tokio::spawn(language_server.text_document_did_close(doc.identifier()));

1126− .await

1127− {

1128− log::error!("Error closing doc in language server: {}", err);

1129− }

1130 }

1129 }

1131

1130

1132 enum Action {

1131 enum Action {

@@ -1188,6 +1187,28 @@ impl Editor {

1188 Ok(())

1187 Ok(())

1189 }

1188 }

1190

1189

1190+ pub fn save<P: Into<PathBuf>>(

1191+ &mut self,

1192+ doc_id: DocumentId,

1193+ path: Option<P>,

1194+ force: bool,

1195+ ) -> anyhow::Result<()> {

1196+ // convert a channel of futures to pipe into main queue one by one

1197+ // via stream.then() ? then push into main future

1198+

1199+ let path = path.map(|path| path.into());

1200+ let doc = doc_mut!(self, &doc_id);

1201+ let future = doc.save(path, force)?;

1202+ // TODO: if no self.saves for that doc id then bail

1203+ // bail!("saves are closed for this document!");

1204+ use futures_util::stream;

1205+ self.saves[&doc_id]

1206+ .send(stream::once(Box::pin(future)))

1207+ .map_err(|err| anyhow!("failed to send save event: {}", err))?;

1208+

1209+ Ok(())

1210+ }

1211+

1191 pub fn resize(&mut self, area: Rect) {

1212 pub fn resize(&mut self, area: Rect) {

1192 if self.tree.resize(area) {

1213 if self.tree.resize(area) {

1193 self._refresh();

1214 self._refresh();

@@ -1307,16 +1328,10 @@ impl Editor {

1307 }

1328 }

1308

1329

1309 pub async fn wait_event(&mut self) -> EditorEvent {

1330 pub async fn wait_event(&mut self) -> EditorEvent {

1310− let mut saves: FuturesUnordered<_> = self

1311− .documents

1312− .values_mut()

1313− .map(Document::await_save)

1314− .collect();

1315−

1316 tokio::select! {

1331 tokio::select! {

1317 biased;

1332 biased;

1318

1333

1319− Some(Some(event)) = saves.next() => {

1334+ Some(event) = self.save_queue.next() => {

1320 EditorEvent::DocumentSaved(event)

1335 EditorEvent::DocumentSaved(event)

1321 }

1336 }

1322 Some(config_event) = self.config_events.1.recv() => {

1337 Some(config_event) = self.config_events.1.recv() => {