implement global open file counter using syswrap

close files after using them if global max is passed.

I originally implemented this without the global count—just always closing files
when done with them, and reopening for new writes. This was crazy slow for that
one test that uses mustSetBits in a big loop. I modified the test to use
importRoaring and everything worked better (though much more slowly).

After adding the global counter, I ran the tests with that one test using
mustSetBits again, and the performance was similar to master. After completing
this PR, I ran the tests with the max limit set to 5—they still passed but were
much slower.

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) {
+	for i, c := range columnIDs {
+		columnIDs[i] = rowID*ShardWidth + c%ShardWidth
+	}
+	positions := columnIDs
+	bm.DirectAddN(positions...)
+}
+
 // 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()
+}