volume: add fsync threshold

volume server count how many object has
been written, if the number of objects is
bigger than threshold, sync the kernel cache.

Author: xushuxun

weed/command/volume.go

@@ -14,7 +14,7 @@ import (
 
 	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/chrislusf/seaweedfs/weed/pb/volume_server_pb"
-	"github.com/chrislusf/seaweedfs/weed/server"
+	weed_server "github.com/chrislusf/seaweedfs/weed/server"
 	"github.com/chrislusf/seaweedfs/weed/storage"
 	"github.com/chrislusf/seaweedfs/weed/util"
 	"google.golang.org/grpc/reflection"
@@ -44,6 +44,7 @@ type VolumeServerOptions struct {
 	cpuProfile            *string
 	memProfile            *string
 	compactionMBPerSecond *int
+	fsyncThreshold        *int
 }
 
 func init() {
@@ -64,6 +65,7 @@ func init() {
 	v.cpuProfile = cmdVolume.Flag.String("cpuprofile", "", "cpu profile output file")
 	v.memProfile = cmdVolume.Flag.String("memprofile", "", "memory profile output file")
 	v.compactionMBPerSecond = cmdVolume.Flag.Int("compactionMBps", 0, "limit background compaction or copying speed in mega bytes per second")
+	v.fsyncThreshold = cmdVolume.Flag.Int("fsyncThreshold", 0, "call fsync when have written fsyncThreshold objects, set to 0 will not call fsync")
 }
 
 var cmdVolume = &Command{
@@ -156,6 +158,7 @@ func (v VolumeServerOptions) startVolumeServer(volumeFolders, maxVolumeCounts, v
 		v.whiteList,
 		*v.fixJpgOrientation, *v.readRedirect,
 		*v.compactionMBPerSecond,
+		*v.fsyncThreshold,
 	)
 
 	listeningAddress := *v.bindIp + ":" + strconv.Itoa(*v.port)

weed/server/volume_server.go

@@ -29,6 +29,7 @@ type VolumeServer struct {
 	compactionBytePerSecond int64
 	MetricsAddress          string
 	MetricsIntervalSec      int
+	fsyncThreshold          int
 }
 
 func NewVolumeServer(adminMux, publicMux *http.ServeMux, ip string,
@@ -41,6 +42,7 @@ func NewVolumeServer(adminMux, publicMux *http.ServeMux, ip string,
 	fixJpgOrientation bool,
 	readRedirect bool,
 	compactionMBPerSecond int,
+	fsyncThreshold int,
 ) *VolumeServer {
 
 	v := viper.GetViper()
@@ -62,9 +64,10 @@ func NewVolumeServer(adminMux, publicMux *http.ServeMux, ip string,
 		ReadRedirect:            readRedirect,
 		grpcDialOption:          security.LoadClientTLS(viper.Sub("grpc"), "volume"),
 		compactionBytePerSecond: int64(compactionMBPerSecond) * 1024 * 1024,
+		fsyncThreshold:          fsyncThreshold,
 	}
 	vs.SeedMasterNodes = masterNodes
-	vs.store = storage.NewStore(vs.grpcDialOption, port, ip, publicUrl, folders, maxCounts, vs.needleMapKind)
+	vs.store = storage.NewStore(vs.grpcDialOption, port, ip, publicUrl, folders, maxCounts, vs.needleMapKind, vs.fsyncThreshold)
 
 	vs.guard = security.NewGuard(whiteList, signingKey, expiresAfterSec, readSigningKey, readExpiresAfterSec)
 

weed/storage/store.go

@@ -2,6 +2,7 @@ package storage
 
 import (
 	"fmt"
+	"sync"
 	"sync/atomic"
 
 	"github.com/chrislusf/seaweedfs/weed/glog"
@@ -35,14 +36,18 @@ type Store struct {
 	DeletedVolumesChan  chan master_pb.VolumeShortInformationMessage
 	NewEcShardsChan     chan master_pb.VolumeEcShardInformationMessage
 	DeletedEcShardsChan chan master_pb.VolumeEcShardInformationMessage
+
+	fsyncThreshold   int
+	writtenFileCount int
+	fsyncLock        sync.Mutex
 }
 
 func (s *Store) String() (str string) {
 	str = fmt.Sprintf("Ip:%s, Port:%d, PublicUrl:%s, dataCenter:%s, rack:%s, connected:%v, volumeSizeLimit:%d", s.Ip, s.Port, s.PublicUrl, s.dataCenter, s.rack, s.connected, s.GetVolumeSizeLimit())
 	return
 }
 
-func NewStore(grpcDialOption grpc.DialOption, port int, ip, publicUrl string, dirnames []string, maxVolumeCounts []int, needleMapKind NeedleMapType) (s *Store) {
+func NewStore(grpcDialOption grpc.DialOption, port int, ip, publicUrl string, dirnames []string, maxVolumeCounts []int, needleMapKind NeedleMapType, fsyncThreshold int) (s *Store) {
 	s = &Store{grpcDialOption: grpcDialOption, Port: port, Ip: ip, PublicUrl: publicUrl, NeedleMapType: needleMapKind}
 	s.Locations = make([]*DiskLocation, 0)
 	for i := 0; i < len(dirnames); i++ {
@@ -57,6 +62,8 @@ func NewStore(grpcDialOption grpc.DialOption, port int, ip, publicUrl string, di
 	s.NewEcShardsChan = make(chan master_pb.VolumeEcShardInformationMessage, 3)
 	s.DeletedEcShardsChan = make(chan master_pb.VolumeEcShardInformationMessage, 3)
 
+	s.fsyncThreshold = fsyncThreshold
+
 	return
 }
 func (s *Store) AddVolume(volumeId needle.VolumeId, collection string, needleMapKind NeedleMapType, replicaPlacement string, ttlString string, preallocate int64) error {
@@ -219,6 +226,15 @@ func (s *Store) WriteVolumeNeedle(i needle.VolumeId, n *needle.Needle) (size uin
 		}
 		if MaxPossibleVolumeSize >= v.ContentSize()+uint64(needle.GetActualSize(size, v.version)) {
 			_, size, isUnchanged, err = v.writeNeedle(n)
+			if s.fsyncThreshold != 0 {
+				s.fsyncLock.Lock()
+				s.writtenFileCount++
+				if s.writtenFileCount > s.fsyncThreshold {
+					v.dataFile.Sync()
+					s.writtenFileCount = 0
+				}
+				s.fsyncLock.Unlock()
+			}
 		} else {
 			err = fmt.Errorf("volume size limit %d exceeded! current size is %d", s.GetVolumeSizeLimit(), v.ContentSize())
 		}