Merge pull request #1903 from jaffee/mmap-fail

Mmap fail

ctl/server.go

@@ -30,6 +30,7 @@ func BuildServerFlags(cmd *cobra.Command, srv *server.Command) {
 	flags.IntVarP(&srv.Config.MaxWritesPerRequest, "max-writes-per-request", "", srv.Config.MaxWritesPerRequest, "Number of write commands per request.")
 	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.")
 
 	// TLS
 	SetTLSConfig(flags, &srv.Config.TLS.CertificatePath, &srv.Config.TLS.CertificateKeyPath, &srv.Config.TLS.SkipVerify)

fragment.go

@@ -41,6 +41,7 @@ import (
 	"github.com/pilosa/pilosa/pql"
 	"github.com/pilosa/pilosa/roaring"
 	"github.com/pilosa/pilosa/stats"
+	"github.com/pilosa/pilosa/syswrap"
 	"github.com/pilosa/pilosa/tracing"
 	"github.com/pkg/errors"
 )
@@ -214,28 +215,33 @@ func (f *fragment) openStorage() error {
 			return fmt.Errorf("init storage file: %s", err)
 		}
 		bi.Flush()
-		fi, err = f.file.Stat()
+		_, err = f.file.Stat()
 		if err != nil {
 			return errors.Wrap(err, "statting file after")
 		}
-	}
+	} else {
+		// Mmap the underlying file so it can be zero copied.
+		data, err := syswrap.Mmap(int(f.file.Fd()), 0, int(fi.Size()), syscall.PROT_READ, syscall.MAP_SHARED)
+		if err == syswrap.ErrMaxMapCountReached {
+			f.Logger.Debugf("maximum number of maps reached, reading file instead")
+			data, err = ioutil.ReadAll(file)
+			if err != nil {
+				return errors.Wrap(err, "failure file readall")
+			}
+		} else if err != nil {
+			return errors.Wrap(err, "mmap failed")
+		} else {
+			f.storageData = data
+			// Advise the kernel that the mmap is accessed randomly.
+			if err := madvise(f.storageData, syscall.MADV_RANDOM); err != nil {
+				return fmt.Errorf("madvise: %s", err)
+			}
+		}
 
-	// Mmap the underlying file so it can be zero copied.
-	storageData, err := syscall.Mmap(int(f.file.Fd()), 0, int(fi.Size()), syscall.PROT_READ, syscall.MAP_SHARED)
-	if err != nil {
-		return fmt.Errorf("mmap: %s", err)
-	}
-	f.storageData = storageData
+		if err := f.storage.UnmarshalBinary(data); err != nil {
+			return fmt.Errorf("unmarshal storage: file=%s, err=%s", f.file.Name(), err)
+		}
 
-	// Advise the kernel that the mmap is accessed randomly.
-	if err := madvise(f.storageData, syscall.MADV_RANDOM); err != nil {
-		return fmt.Errorf("madvise: %s", err)
-	}
-
-	// Attach the mmap file to the bitmap.
-	data := f.storageData
-	if err := f.storage.UnmarshalBinary(data); err != nil {
-		return fmt.Errorf("unmarshal storage: file=%s, err=%s", f.file.Name(), err)
 	}
 
 	f.opN = f.storage.Info().OpN
@@ -323,7 +329,7 @@ func (f *fragment) closeStorage() error {
 
 	// Unmap the file.
 	if f.storageData != nil {
-		if err := syscall.Munmap(f.storageData); err != nil {
+		if err := syswrap.Munmap(f.storageData); err != nil {
 			return fmt.Errorf("munmap: %s", err)
 		}
 		f.storageData = nil

server/config.go

@@ -70,6 +70,11 @@ type Config struct {
 		AllowedOrigins []string `toml:"allowed-origins"`
 	} `toml:"handler"`
 
+	// MaxMapCount puts an in-process limit on the number of mmaps. After this
+	// is exhausted, Pilosa will fall back to reading the file into memory
+	// normally.
+	MaxMapCount uint64 `toml:"max-map-count"`
+
 	// TLS
 	TLS TLSConfig `toml:"tls"`
 
@@ -124,6 +129,7 @@ func NewConfig() *Config {
 		DataDir:             "~/.pilosa",
 		Bind:                ":10101",
 		MaxWritesPerRequest: 5000,
+		MaxMapCount:         60000,
 		TLS:                 TLSConfig{},
 	}
 

server/server.go

@@ -45,6 +45,7 @@ import (
 	"github.com/pilosa/pilosa/logger"
 	"github.com/pilosa/pilosa/stats"
 	"github.com/pilosa/pilosa/statsd"
+	"github.com/pilosa/pilosa/syswrap"
 	"github.com/pkg/errors"
 )
 
@@ -178,6 +179,8 @@ 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)
+
 	err := m.setupLogger()
 	if err != nil {
 		return errors.Wrap(err, "setting up logger")

syswrap/mmap.go

@@ -0,0 +1,54 @@
+// Package syswrap wraps syscalls (just mmap right now) in order to impose a
+// global in-process limit on the maximum number of active mmaps.
+package syswrap
+
+import (
+	"sync"
+	"sync/atomic"
+	"syscall"
+
+	"github.com/pkg/errors"
+)
+
+var mapCount uint64
+
+var ErrMaxMapCountReached = errors.New("maximum map count reached")
+
+// maxMapCount default to slightly less than the typical
+// default on Linux (65K). We want to leave some
+// overhead for (e.g.) the Go runtime.
+var maxMapCount uint64 = 60000
+var mu sync.RWMutex
+
+func SetMaxMapCount(max uint64) {
+	mu.Lock()
+	maxMapCount = max
+	mu.Unlock()
+}
+
+// Mmap increments the global map count, and then calls syscall.Mmap. It
+// decrements the map count and returns an error if the count was over the
+// limit. If syscall.Mmap returns an error it also decrements the count.
+func Mmap(fd int, offset int64, length int, prot int, flags int) (data []byte, err error) {
+	mu.RLock()
+	defer mu.RUnlock()
+	if newCount := atomic.AddUint64(&mapCount, 1); newCount > maxMapCount {
+		atomic.AddUint64(&mapCount, ^uint64(0)) // decrement
+		return nil, ErrMaxMapCountReached
+	}
+	data, err = syscall.Mmap(fd, offset, length, prot, flags)
+	if err != nil {
+		atomic.AddUint64(&mapCount, ^uint64(0)) // decrement
+	}
+	return data, err
+}
+
+// Munmap calls sycall.Munmap, and then decrements the global map count if there
+// was no error.
+func Munmap(b []byte) (err error) {
+	err = syscall.Munmap(b)
+	if err == nil {
+		atomic.AddUint64(&mapCount, ^uint64(0)) // decrement
+	}
+	return err
+}