Merge pull request #1906 from jaffee/1905-close-files

implement global open file counter using syswrap

ctl/server.go

@@ -31,6 +31,7 @@ func BuildServerFlags(cmd *cobra.Command, srv *server.Command) {
 	flags.StringVar(&srv.Config.LogPath, "log-path", srv.Config.LogPath, "Log path")
 	flags.BoolVar(&srv.Config.Verbose, "verbose", srv.Config.Verbose, "Enable verbose logging")
 	flags.Uint64Var(&srv.Config.MaxMapCount, "max-map-count", srv.Config.MaxMapCount, "Limits the maximum number of active mmaps. Pilosa will fall back to reading files once this is exhausted. Set below your system's vm.max_map_count.")
+	flags.Uint64Var(&srv.Config.MaxFileCount, "max-file-count", srv.Config.MaxFileCount, "Soft limit on the maximum number of fragment files Pilosa keeps open simultaneously.")
 
 	// TLS
 	SetTLSConfig(flags, &srv.Config.TLS.CertificatePath, &srv.Config.TLS.CertificateKeyPath, &srv.Config.TLS.SkipVerify)

docs/configuration.md

@@ -146,6 +146,19 @@ The config file is in the [toml format](https://github.com/toml-lang/toml) and h
     max-writes-per-request = 5000
     ```
 
+#### Max File Count
+
+* Description: A soft limit on the maximum number of files that Pilosa will keep
+  open simultaneously. When past this limit, Pilosa will only keep files open
+  for as long as it needs to write updates. This will negatively affect
+  performance in cases where Pilosa is doing lots of small updates.
+* Flag: `--max-file-count=500000`
+* Env: `PILOSA_MAX_FILE_COUNT=500000`
+* Config:
+    ```toml
+    max-file-count = 500000
+    ```
+
 #### Gossip Advertise Host
 
 * Description: Host on which memberlist should advertise. Defaults to `advertise` host.

fragment.go

@@ -187,6 +187,18 @@ func (f *fragment) Open() error {
 	return nil
 }
 
+func (f *fragment) reopen() (mustClose bool, err error) {
+	if f.file == nil {
+		// Open the data file to be mmap'd and used as an ops log.
+		f.file, mustClose, err = syswrap.OpenFile(f.path, os.O_RDWR|os.O_CREATE|os.O_APPEND, 0666)
+		if err != nil {
+			return mustClose, fmt.Errorf("open file: %s", err)
+		}
+		f.storage.OpWriter = f.file
+	}
+	return mustClose, nil
+}
+
 // openStorage opens the storage bitmap.
 func (f *fragment) openStorage() error {
 	// Create a roaring bitmap to serve as storage for the shard.
@@ -194,11 +206,14 @@ func (f *fragment) openStorage() error {
 		f.storage = roaring.NewFileBitmap()
 	}
 	// Open the data file to be mmap'd and used as an ops log.
-	file, err := os.OpenFile(f.path, os.O_RDWR|os.O_CREATE|os.O_APPEND, 0666)
+	file, mustClose, err := syswrap.OpenFile(f.path, os.O_RDWR|os.O_CREATE|os.O_APPEND, 0666)
 	if err != nil {
 		return fmt.Errorf("open file: %s", err)
 	}
 	f.file = file
+	if mustClose {
+		defer f.safeClose()
+	}
 
 	// Lock the underlying file.
 	if err := syscall.Flock(int(f.file.Fd()), syscall.LOCK_EX|syscall.LOCK_NB); err != nil {
@@ -322,6 +337,26 @@ func (f *fragment) close() error {
 	return nil
 }
 
+// safeClose is unprotected.
+func (f *fragment) safeClose() error {
+	// Flush file, unlock & close.
+	if f.file != nil {
+		if err := f.file.Sync(); err != nil {
+			return fmt.Errorf("sync: %s", err)
+		}
+		if err := syscall.Flock(int(f.file.Fd()), syscall.LOCK_UN); err != nil {
+			return fmt.Errorf("unlock: %s", err)
+		}
+		if err := syswrap.CloseFile(f.file); err != nil {
+			return fmt.Errorf("close file: %s", err)
+		}
+	}
+	f.file = nil
+	f.storage.OpWriter = nil
+
+	return nil
+}
+
 func (f *fragment) closeStorage() error {
 	// Clear the storage bitmap so it doesn't access the closed mmap.
 
@@ -335,17 +370,8 @@ func (f *fragment) closeStorage() error {
 		f.storageData = nil
 	}
 
-	// Flush file, unlock & close.
-	if f.file != nil {
-		if err := f.file.Sync(); err != nil {
-			return fmt.Errorf("sync: %s", err)
-		}
-		if err := syscall.Flock(int(f.file.Fd()), syscall.LOCK_UN); err != nil {
-			return fmt.Errorf("unlock: %s", err)
-		}
-		if err := f.file.Close(); err != nil {
-			return fmt.Errorf("close file: %s", err)
-		}
+	if err := f.safeClose(); err != nil {
+		return err
 	}
 
 	// opN is determined by how many bit set/clear operations are in the storage
@@ -406,6 +432,13 @@ func (f *fragment) rowFromStorage(rowID uint64) *Row {
 func (f *fragment) setBit(rowID, columnID uint64) (changed bool, err error) {
 	f.mu.Lock()
 	defer f.mu.Unlock()
+	mustClose, err := f.reopen()
+	if err != nil {
+		return false, errors.Wrap(err, "reopening")
+	}
+	if mustClose {
+		defer f.safeClose()
+	}
 
 	// handle mutux field type
 	if f.mutexVector != nil {
@@ -480,6 +513,13 @@ func (f *fragment) unprotectedSetBit(rowID, columnID uint64) (changed bool, err
 func (f *fragment) clearBit(rowID, columnID uint64) (bool, error) {
 	f.mu.Lock()
 	defer f.mu.Unlock()
+	mustClose, err := f.reopen()
+	if err != nil {
+		return false, errors.Wrap(err, "reopening")
+	}
+	if mustClose {
+		defer f.safeClose()
+	}
 	return f.unprotectedClearBit(rowID, columnID)
 }
 
@@ -528,6 +568,13 @@ func (f *fragment) unprotectedClearBit(rowID, columnID uint64) (changed bool, er
 func (f *fragment) setRow(row *Row, rowID uint64) (bool, error) {
 	f.mu.Lock()
 	defer f.mu.Unlock()
+	mustClose, err := f.reopen()
+	if err != nil {
+		return false, errors.Wrap(err, "reopening")
+	}
+	if mustClose {
+		defer f.safeClose()
+	}
 	return f.unprotectedSetRow(row, rowID)
 }
 
@@ -578,6 +625,13 @@ func (f *fragment) unprotectedSetRow(row *Row, rowID uint64) (changed bool, err
 func (f *fragment) clearRow(rowID uint64) (bool, error) {
 	f.mu.Lock()
 	defer f.mu.Unlock()
+	mustClose, err := f.reopen()
+	if err != nil {
+		return false, errors.Wrap(err, "reopening")
+	}
+	if mustClose {
+		defer f.safeClose()
+	}
 	return f.unprotectedClearRow(rowID)
 }
 
@@ -685,6 +739,13 @@ func (f *fragment) positionsForValue(columnID uint64, bitDepth uint, value uint6
 func (f *fragment) setValueBase(columnID uint64, bitDepth uint, value uint64, clear bool) (changed bool, err error) {
 	f.mu.Lock()
 	defer f.mu.Unlock()
+	mustClose, err := f.reopen()
+	if err != nil {
+		return false, errors.Wrap(err, "reopening")
+	}
+	if mustClose {
+		defer f.safeClose()
+	}
 
 	for i := uint(0); i < bitDepth; i++ {
 		if value&(1<<i) != 0 {
@@ -1557,6 +1618,14 @@ func (f *fragment) importPositions(set, clear []uint64, rowSet map[uint64]struct
 	smallWrite := false
 	if len(set)+len(clear)+f.opN < f.MaxOpN {
 		smallWrite = true
+		mustClose, err := f.reopen()
+		if err != nil {
+			return errors.Wrap(err, "reopening")
+		}
+		if mustClose {
+			defer f.safeClose()
+		}
+
 	} else {
 		f.storage.OpWriter = nil
 	}
@@ -1673,6 +1742,7 @@ func (f *fragment) importValue(columnIDs, values []uint64, bitDepth uint, clear
 	smallWrite := false
 	if len(columnIDs)*int(bitDepth+1)+f.opN < f.MaxOpN {
 		smallWrite = true
+
 		// TODO figure out how to avoid re-allocating these each time. Probably
 		// possible to store them on the fragment with a capacity based on
 		// MaxOpN. For now, we know that the total number of bits to be
@@ -1686,6 +1756,7 @@ func (f *fragment) importValue(columnIDs, values []uint64, bitDepth uint, clear
 	if !smallWrite {
 		f.mu.Lock()
 		defer f.mu.Unlock()
+
 		f.storage.OpWriter = nil
 	}
 	// Process every value.

fragment_internal_test.go

@@ -66,7 +66,7 @@ func TestFragment_SetBit(t *testing.T) {
 	}
 
 	// Close and reopen the fragment & verify the data.
-	if err := f.reopen(); err != nil {
+	if err := f.Reopen(); err != nil {
 		t.Fatal(err)
 	} else if n := f.row(120).Count(); n != 2 {
 		t.Fatalf("unexpected count (reopen): %d", n)
@@ -95,7 +95,7 @@ func TestFragment_ClearBit(t *testing.T) {
 	}
 
 	// Close and reopen the fragment & verify the data.
-	if err := f.reopen(); err != nil {
+	if err := f.Reopen(); err != nil {
 		t.Fatal(err)
 	} else if n := f.row(1000).Count(); n != 1 {
 		t.Fatalf("unexpected count (reopen): %d", n)
@@ -122,7 +122,7 @@ func TestFragment_ClearRow(t *testing.T) {
 	}
 
 	// Close and reopen the fragment & verify the data.
-	if err := f.reopen(); err != nil {
+	if err := f.Reopen(); err != nil {
 		t.Fatal(err)
 	} else if n := f.row(1000).Count(); n != 0 {
 		t.Fatalf("unexpected count (reopen): %d", n)
@@ -170,7 +170,7 @@ func TestFragment_SetRow(t *testing.T) {
 	}
 
 	// Close and reopen the fragment & verify the data.
-	if err := f.reopen(); err != nil {
+	if err := f.Reopen(); err != nil {
 		t.Fatal(err)
 	} else if n := f.row(rowID).Count(); n != 3 {
 		t.Fatalf("unexpected count (reopen): %d", n)
@@ -874,7 +874,7 @@ func TestFragment_Snapshot(t *testing.T) {
 	}
 
 	// Close and reopen the fragment & verify the data.
-	if err := f.reopen(); err != nil {
+	if err := f.Reopen(); err != nil {
 		t.Fatal(err)
 	} else if n := f.row(1000).Count(); n != 1 {
 		t.Fatalf("unexpected count (reopen): %d", n)
@@ -1004,12 +1004,22 @@ func TestFragment_TopN_Intersect_Large(t *testing.T) {
 		990, 991, 992, 993, 994, 995, 996, 997, 998, 999,
 	)
 
+	bm := roaring.NewBTreeBitmap()
 	// Set bits on rows 0 - 999. Higher rows have higher bit counts.
 	for i := uint64(0); i < 1000; i++ {
 		for j := uint64(0); j < i; j++ {
-			f.mustSetBits(i, j)
+			addToBitmap(bm, i, j)
 		}
 	}
+	b := &bytes.Buffer{}
+	_, err := bm.WriteTo(b)
+	if err != nil {
+		t.Fatalf("writing to bytes: %v", err)
+	}
+	err = f.importRoaring(b.Bytes(), false)
+	if err != nil {
+		t.Fatalf("importing data: %v", err)
+	}
 	f.RecalculateCache()
 
 	// Retrieve top rows.
@@ -1228,7 +1238,7 @@ func TestFragment_LRUCache_Persistence(t *testing.T) {
 	}
 
 	// Reopen the fragment.
-	if err := f.reopen(); err != nil {
+	if err := f.Reopen(); err != nil {
 		t.Fatal(err)
 	}
 
@@ -1338,7 +1348,7 @@ func TestFragment_WriteTo_ReadFrom(t *testing.T) {
 	}
 
 	// Close and reopen the fragment & verify the data.
-	if err := f1.reopen(); err != nil {
+	if err := f1.Reopen(); err != nil {
 		t.Fatal(err)
 	} else if n := f1.cache.Len(); n != 1 {
 		t.Fatalf("unexpected cache size (reopen): %d", n)
@@ -1466,7 +1476,7 @@ func TestFragment_Snapshot_Run(t *testing.T) {
 	}
 
 	// Close and reopen the fragment & verify the data.
-	if err := f.reopen(); err != nil {
+	if err := f.Reopen(); err != nil {
 		t.Fatal(err)
 	} else if n := f.row(1000).Count(); n != 2 {
 		t.Fatalf("unexpected count (reopen): %d", n)
@@ -2435,7 +2445,7 @@ func mustOpenBoolFragment(index, field, view string, shard uint64, cacheType str
 }
 
 // Reopen closes the fragment and reopens it as a new instance.
-func (f *fragment) reopen() error {
+func (f *fragment) Reopen() error {
 	if err := f.Close(); err != nil {
 		return err
 	}
@@ -2455,6 +2465,14 @@ func (f *fragment) mustSetBits(rowID uint64, columnIDs ...uint64) {
 	}
 }
 
+func addToBitmap(bm *roaring.Bitmap, rowID uint64, columnIDs ...uint64) {
+	// we'll reuse the columnIDs slice and fill it with positions for DirectAddN
+	for i, c := range columnIDs {
+		columnIDs[i] = rowID*ShardWidth + c%ShardWidth
+	}
+	bm.DirectAddN(columnIDs...)
+}
+
 // Test Various methods of retrieving RowIDs
 func TestFragment_RowsIteration(t *testing.T) {
 	t.Run("firstContainer", func(t *testing.T) {

server/config.go

@@ -75,6 +75,13 @@ type Config struct {
 	// normally.
 	MaxMapCount uint64 `toml:"max-map-count"`
 
+	// MaxFileCount puts a soft, in-process limit on the number of open fragment
+	// files. Once this limit is passed, Pilosa will only keep files open while
+	// actively working with them, and will close them afterward. This has a
+	// negative effect on performance for workloads which make small appends to
+	// lots of fragments.
+	MaxFileCount uint64 `toml:"max-file-count"`
+
 	// TLS
 	TLS TLSConfig `toml:"tls"`
 
@@ -130,6 +137,7 @@ func NewConfig() *Config {
 		Bind:                ":10101",
 		MaxWritesPerRequest: 5000,
 		MaxMapCount:         60000,
+		MaxFileCount:        500000,
 		TLS:                 TLSConfig{},
 	}
 

server/server.go

@@ -180,6 +180,7 @@ func (m *Command) Wait() error {
 // SetupServer uses the cluster configuration to set up this server.
 func (m *Command) SetupServer() error {
 	syswrap.SetMaxMapCount(m.Config.MaxMapCount)
+	syswrap.SetMaxFileCount(m.Config.MaxFileCount)
 
 	err := m.setupLogger()
 	if err != nil {

syswrap/os.go

@@ -0,0 +1,41 @@
+package syswrap
+
+import (
+	"os"
+	"sync"
+	"sync/atomic"
+)
+
+var fileCount uint64
+
+// maxFileCount is the soft limit on the number of open files. syswrap.OpenFile
+// will warn when this limit is passed.
+var maxFileCount uint64 = 500000
+var fileMu sync.RWMutex
+
+func SetMaxFileCount(max uint64) {
+	fileMu.Lock()
+	maxFileCount = max
+	fileMu.Unlock()
+}
+
+// OpenFile passes the arguments along to os.OpenFile while incrementing a
+// counter. If the counter is above the maximum, it returns mustClose true to
+// signal the calling function that it should not keep the file open
+// indefinitely. Files opened with this function should be closed by
+// syswrap.CloseFile.
+func OpenFile(name string, flag int, perm os.FileMode) (file *os.File, mustClose bool, err error) {
+	file, err = os.OpenFile(name, flag, perm)
+	fileMu.RLock()
+	defer fileMu.RUnlock()
+	if newCount := atomic.AddUint64(&fileCount, 1); newCount > maxFileCount {
+		mustClose = true
+	}
+	return file, mustClose, err
+}
+
+// CloseFile decrements the global count of open files and closes the file.
+func CloseFile(f *os.File) error {
+	atomic.AddUint64(&fileCount, ^uint64(0)) // decrement
+	return f.Close()
+}