diff --git a/ctl/server.go b/ctl/server.go index 10496c933..2f727e9e0 100644 --- a/ctl/server.go +++ b/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) diff --git a/docs/configuration.md b/docs/configuration.md index 2a8c31c3f..2d046a961 100644 --- a/docs/configuration.md +++ b/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. diff --git a/fragment.go b/fragment.go index a89702528..2b2925e16 100644 --- a/fragment.go +++ b/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< 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() +}