From e7f65cf7be3eb6701f27e6b0a3c851bff5ceff69 Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Wed, 20 Mar 2019 16:05:10 -0500 Subject: [PATCH 1/2] implement global open file counter using syswrap MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 1 + docs/configuration.md | 13 ++++++ fragment.go | 95 ++++++++++++++++++++++++++++++++++----- fragment_internal_test.go | 38 +++++++++++----- server/config.go | 8 ++++ server/server.go | 1 + syswrap/os.go | 41 +++++++++++++++++ 7 files changed, 175 insertions(+), 22 deletions(-) create mode 100644 syswrap/os.go 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() +} From 53dfa9b7f2bc8c8249a704551fb139fc2c6696f6 Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Fri, 22 Mar 2019 16:24:08 -0500 Subject: [PATCH 2/2] remove rename of columnIDs and add comment --- fragment_internal_test.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/fragment_internal_test.go b/fragment_internal_test.go index 219934439..b726d4fe5 100644 --- a/fragment_internal_test.go +++ b/fragment_internal_test.go @@ -2466,11 +2466,11 @@ 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 } - positions := columnIDs - bm.DirectAddN(positions...) + bm.DirectAddN(columnIDs...) } // Test Various methods of retrieving RowIDs