lib/logstorage: close files in parallel inside parts opened for reading

This should reduce the time needed for closing the parts opened for reading
on high-latency storage systems such as NFS of Ceph.

Updates #517

Author: valyala

go.mod

@@ -3,7 +3,7 @@ module github.com/VictoriaMetrics/VictoriaLogs
 go 1.24.5
 
 require (
-	github.com/VictoriaMetrics/VictoriaMetrics v0.0.0-20250727161922-a05e4cf67b09
+	github.com/VictoriaMetrics/VictoriaMetrics v0.0.0-20250727172355-8e155da0ac00
 	github.com/VictoriaMetrics/easyproto v0.1.4
 	github.com/VictoriaMetrics/metrics v1.38.0
 	github.com/cespare/xxhash/v2 v2.3.0

go.sum

@@ -1,5 +1,5 @@
-github.com/VictoriaMetrics/VictoriaMetrics v0.0.0-20250727161922-a05e4cf67b09 h1:+V7+2wKZMngpKjnW/n9eK7V3R0J97D/Y5rttiQYJ53g=
-github.com/VictoriaMetrics/VictoriaMetrics v0.0.0-20250727161922-a05e4cf67b09/go.mod h1:NQGojsgYDAf4cR0k1us+EzDnsZSsHTun1Hp/f3+S6p4=
+github.com/VictoriaMetrics/VictoriaMetrics v0.0.0-20250727172355-8e155da0ac00 h1:eMRy88id0ZRmg6GxTuLrF8sBYNzdAoJa1MIx5+HjnxE=
+github.com/VictoriaMetrics/VictoriaMetrics v0.0.0-20250727172355-8e155da0ac00/go.mod h1:NQGojsgYDAf4cR0k1us+EzDnsZSsHTun1Hp/f3+S6p4=
 github.com/VictoriaMetrics/easyproto v0.1.4 h1:r8cNvo8o6sR4QShBXQd1bKw/VVLSQma/V2KhTBPf+Sc=
 github.com/VictoriaMetrics/easyproto v0.1.4/go.mod h1:QlGlzaJnDfFd8Lk6Ci/fuLxfTo3/GThPs2KH23mv710=
 github.com/VictoriaMetrics/metrics v1.38.0 h1:1d0dRgVH8Nnu8dKMfisKefPC3q7gqf3/odyO0quAvyA=

lib/logstorage/block_stream_reader.go

@@ -95,9 +95,10 @@ func (r *bloomValuesReader) totalBytesRead() uint64 {
 	return r.bloom.bytesRead + r.values.bytesRead
 }
 
-func (r *bloomValuesReader) MustClose() {
-	r.bloom.MustClose()
-	r.values.MustClose()
+func (r *bloomValuesReader) appendClosers(dst []fs.MustCloser) []fs.MustCloser {
+	dst = append(dst, &r.bloom)
+	dst = append(dst, &r.values)
+	return dst
 }
 
 type bloomValuesStreamReader struct {
@@ -178,19 +179,25 @@ func (sr *streamReaders) totalBytesRead() uint64 {
 }
 
 func (sr *streamReaders) MustClose() {
-	sr.columnNamesReader.MustClose()
-	sr.columnIdxsReader.MustClose()
-	sr.metaindexReader.MustClose()
-	sr.indexReader.MustClose()
-	sr.columnsHeaderIndexReader.MustClose()
-	sr.columnsHeaderReader.MustClose()
-	sr.timestampsReader.MustClose()
-
-	sr.messageBloomValuesReader.MustClose()
-	sr.oldBloomValuesReader.MustClose()
+	// Close files in parallel in order to reduce the time needed for this operation
+	// on high-latency storage systems such as NFS or Ceph.
+	cs := []fs.MustCloser{
+		&sr.columnNamesReader,
+		&sr.columnIdxsReader,
+		&sr.metaindexReader,
+		&sr.indexReader,
+		&sr.columnsHeaderIndexReader,
+		&sr.columnsHeaderReader,
+		&sr.timestampsReader,
+	}
+
+	cs = sr.messageBloomValuesReader.appendClosers(cs)
+	cs = sr.oldBloomValuesReader.appendClosers(cs)
 	for i := range sr.bloomValuesShards {
-		sr.bloomValuesShards[i].MustClose()
+		cs = sr.bloomValuesShards[i].appendClosers(cs)
 	}
+
+	fs.MustCloseParallel(cs)
 }
 
 func (sr *streamReaders) getBloomValuesReaderForColumnName(name string) *bloomValuesReader {

lib/logstorage/block_stream_writer.go

@@ -84,9 +84,9 @@ func (w *bloomValuesWriter) totalBytesWritten() uint64 {
 	return w.bloom.bytesWritten + w.values.bytesWritten
 }
 
-func (w *bloomValuesWriter) appendWriteClosers(dst []filestream.WriteCloser) []filestream.WriteCloser {
-	dst = append(dst, w.bloom.w)
-	dst = append(dst, w.values.w)
+func (w *bloomValuesWriter) appendClosers(dst []fs.MustCloser) []fs.MustCloser {
+	dst = append(dst, &w.bloom)
+	dst = append(dst, &w.values)
 	return dst
 }
 
@@ -156,22 +156,24 @@ func (sw *streamWriters) totalBytesWritten() uint64 {
 }
 
 func (sw *streamWriters) MustClose() {
-	wcs := []filestream.WriteCloser{
-		sw.columnNamesWriter.w,
-		sw.columnIdxsWriter.w,
-		sw.metaindexWriter.w,
-		sw.indexWriter.w,
-		sw.columnsHeaderIndexWriter.w,
-		sw.columnsHeaderWriter.w,
-		sw.timestampsWriter.w,
+	// Flush and close files in parallel in order to reduce the time needed for this operation
+	// on high-latency storage systems such as NFS or Ceph.
+	cs := []fs.MustCloser{
+		&sw.columnNamesWriter,
+		&sw.columnIdxsWriter,
+		&sw.metaindexWriter,
+		&sw.indexWriter,
+		&sw.columnsHeaderIndexWriter,
+		&sw.columnsHeaderWriter,
+		&sw.timestampsWriter,
 	}
 
-	wcs = sw.messageBloomValuesWriter.appendWriteClosers(wcs)
+	cs = sw.messageBloomValuesWriter.appendClosers(cs)
 	for i := range sw.bloomValuesShards {
-		wcs = sw.bloomValuesShards[i].appendWriteClosers(wcs)
+		cs = sw.bloomValuesShards[i].appendClosers(cs)
 	}
 
-	filestream.MustCloseWritersParallel(wcs)
+	fs.MustCloseParallel(cs)
 }
 
 func (sw *streamWriters) getBloomValuesWriterForColumnName(name string) *bloomValuesWriter {

lib/logstorage/part.go

@@ -54,9 +54,10 @@ type bloomValuesReaderAt struct {
 	values fs.MustReadAtCloser
 }
 
-func (r *bloomValuesReaderAt) MustClose() {
-	r.bloom.MustClose()
-	r.values.MustClose()
+func (r *bloomValuesReaderAt) appendClosers(dst []fs.MustCloser) []fs.MustCloser {
+	dst = append(dst, r.bloom)
+	dst = append(dst, r.values)
+	return dst
 }
 
 func mustOpenInmemoryPart(pt *partition, mp *inmemoryPart) *part {
@@ -173,22 +174,28 @@ func mustOpenFilePart(pt *partition, path string) *part {
 }
 
 func mustClosePart(p *part) {
-	p.indexFile.MustClose()
+	// Close files in parallel in order to speed up this operation
+	// on high-latency storage systems such as NFS and Ceph.
+	var cs []fs.MustCloser
+
+	cs = append(cs, p.indexFile)
 	if p.ph.FormatVersion >= 1 {
-		p.columnsHeaderIndexFile.MustClose()
+		cs = append(cs, p.columnsHeaderIndexFile)
 	}
-	p.columnsHeaderFile.MustClose()
-	p.timestampsFile.MustClose()
-	p.messageBloomValues.MustClose()
+	cs = append(cs, p.columnsHeaderFile)
+	cs = append(cs, p.timestampsFile)
+	cs = p.messageBloomValues.appendClosers(cs)
 
 	if p.ph.FormatVersion < 1 {
-		p.oldBloomValues.MustClose()
+		cs = p.oldBloomValues.appendClosers(cs)
 	} else {
 		for i := range p.bloomValuesShards {
-			p.bloomValuesShards[i].MustClose()
+			cs = p.bloomValuesShards[i].appendClosers(cs)
 		}
 	}
 
+	fs.MustCloseParallel(cs)
+
 	p.pt = nil
 }