disk buffer: reader can be notified before the data file write completes, delaying the sink's send

Aperta Adatta ai principianti
  • Qualcosa non va?
#26,523 8 commenti 2 reazioni 0 assegnatari Vedi su GitHub

I maintainer di solito rispondono entro 1 giorno

Nessuno ha ancora preso questa issue.

Valutazione

Difficoltà
1/5
Tempo stimato
Meno di un'ora
Idoneità per principianti
88/100
Tipo di issue
Bug
Chiarezza
Specificata chiaramente
Stato di attività
Attiva
Stack tecnologico
rust
Ambito
backend, stream-processing

Direzione di ricerca

Inizia in lib/vector-buffers/src/variants/disk_v2/writer.rs, nel metodo di flush del buffer di TrackingBufWriter. Leggi il flusso write_all esistente e le indicazioni di Tokio sul flush di File, quindi verifica che il reader venga notificato solo dopo il completamento della scrittura del file di dati. Il lavoro è completato quando il flush del buffer di disk_v2 attende il completamento della scrittura senza modificare la gestione degli errori.

Scritto dal modello di indicizzazione a partire dal testo della issue.

Descrizione

A note for the community
  • Please vote on this issue by adding a 👍 reaction to the original issue to help the community and maintainers prioritize this request
  • If you are interested in working on this issue or have submitted a pull request, please leave a comment
Problem

In disk_v2's buffer flush we don't flush the tokio::fs::File, so there is no guarantee the write has finished before the reader is notified. The tokio docs say to call flush (tokio::fs — Using File); to be sure we wait for the write to finish we don't.

Fix:

--- a/lib/vector-buffers/src/variants/disk_v2/writer.rs
+++ b/lib/vector-buffers/src/variants/disk_v2/writer.rs
@@ -408,7 +408,10 @@ impl<W: AsyncWrite + Unpin> TrackingBufWriter<W> {
         let events_flushed = self.unflushed_events as u64;
         let bytes_flushed = self.buf.len() as u64;
 
-        let result = self.inner.write_all(&self.buf[..]).await;
+        let mut result = self.inner.write_all(&self.buf[..]).await;
+        if result.is_ok() {
+            result = self.inner.flush().await;
+        }
Version

vector 0.58.0

Lingua principale
Rust
Stelle
22.8k
Fork
2.4k
Merge medio
1g 13h
PR unite (30g)
273

Preparare l'ambiente

Come iniziare

  1. Leggi tutta la issue e poi la guida ai contributi del progetto.
  2. Commenta sulla issue per dire che te ne occupi tu — evita che due persone facciano lo stesso lavoro.
  3. Fai un fork del repository e lavora su un branch.
  4. Apri una pull request che faccia riferimento al numero della issue.

Altre issue di vectordotdev/vector

Tutte le issue di vectordotdev/vector

Issue simili

Altre issue su Rust

Ricevi le nuove issue nella tua casella

Un breve riepilogo di issue GitHub adatte ai principianti.