2.94

avoid too much memory used also

k8s/helm_charts2/Chart.yaml

@@ -1,5 +1,5 @@
 apiVersion: v1
 description: SeaweedFS
 name: seaweedfs
-appVersion: "2.93"
-version: "2.93"
+appVersion: "2.94"
+version: "2.94"

weed/command/filer.go

@@ -2,6 +2,7 @@ package command
 
 import (
 	"fmt"
+	"net"
 	"net/http"
 	"os"
 	"time"
@@ -51,6 +52,7 @@ type FilerOptions struct {
 	concurrentUploadLimitMB *int
 	debug                   *bool
 	debugPort               *int
+	localSocket             *string
 }
 
 func init() {
@@ -58,7 +60,7 @@ func init() {
 	f.mastersString = cmdFiler.Flag.String("master", "localhost:9333", "comma-separated master servers")
 	f.collection = cmdFiler.Flag.String("collection", "", "all data will be stored in this default collection")
 	f.ip = cmdFiler.Flag.String("ip", util.DetectedHostAddress(), "filer server http listen ip address")
-	f.bindIp = cmdFiler.Flag.String("ip.bind", "", "ip address to bind to")
+	f.bindIp = cmdFiler.Flag.String("ip.bind", "", "ip address to bind to. If empty, default to same as -ip option.")
 	f.port = cmdFiler.Flag.Int("port", 8888, "filer server http listen port")
 	f.portGrpc = cmdFiler.Flag.Int("port.grpc", 0, "filer server grpc listen port")
 	f.publicPort = cmdFiler.Flag.Int("port.readonly", 0, "readonly port opened to public")
@@ -76,6 +78,7 @@ func init() {
 	f.concurrentUploadLimitMB = cmdFiler.Flag.Int("concurrentUploadLimitMB", 128, "limit total concurrent upload size")
 	f.debug = cmdFiler.Flag.Bool("debug", false, "serves runtime profiling data, e.g., http://localhost:<debug.port>/debug/pprof/goroutine?debug=2")
 	f.debugPort = cmdFiler.Flag.Int("debug.port", 6060, "http port for debugging")
+	f.localSocket = cmdFiler.Flag.String("localSocket", "", "default to /tmp/seaweedfs-filer-<port>.sock")
 
 	// start s3 on filer
 	filerStartS3 = cmdFiler.Flag.Bool("s3", false, "whether to start S3 gateway")
@@ -139,11 +142,14 @@ func runFiler(cmd *Command, args []string) bool {
 	if *filerStartS3 {
 		filerS3Options.filer = &filerAddress
 		filerS3Options.bindIp = f.bindIp
+		filerS3Options.localFilerSocket = f.localSocket
 		go func() {
 			time.Sleep(startDelay * time.Second)
 			filerS3Options.startS3Server()
 		}()
 		startDelay++
+	} else {
+		*f.localSocket = ""
 	}
 
 	if *filerStartWebDav {
@@ -182,6 +188,9 @@ func (fo *FilerOptions) startFiler() {
 	if *fo.portGrpc == 0 {
 		*fo.portGrpc = 10000 + *fo.port
 	}
+	if *fo.bindIp == "" {
+		*fo.bindIp = *fo.ip
+	}
 
 	defaultLevelDbDirectory := util.ResolvePath(*fo.defaultLevelDbDirectory + "/filerldb2")
 
@@ -230,6 +239,18 @@ func (fo *FilerOptions) startFiler() {
 		glog.Fatalf("Filer listener error: %v", e)
 	}
 
+	// start on local unix socket
+	if *fo.localSocket == "" {
+		*fo.localSocket = fmt.Sprintf("/tmp/seaweefs-filer-%d.sock", *fo.port)
+		if err := os.Remove(*fo.localSocket); err != nil && !os.IsNotExist(err) {
+			glog.Fatalf("Failed to remove %s, error: %s", *fo.localSocket, err.Error())
+		}
+	}
+	filerSocketListener, err := net.Listen("unix", *fo.localSocket)
+	if err != nil {
+		glog.Fatalf("Failed to listen on %s: %v", *fo.localSocket, err)
+	}
+
 	// starting grpc server
 	grpcPort := *fo.portGrpc
 	grpcL, err := util.NewListener(util.JoinHostPort(*fo.bindIp, grpcPort), 0)
@@ -242,6 +263,9 @@ func (fo *FilerOptions) startFiler() {
 	go grpcS.Serve(grpcL)
 
 	httpS := &http.Server{Handler: defaultMux}
+	go func() {
+		httpS.Serve(filerSocketListener)
+	}()
 	if err := httpS.Serve(filerListener); err != nil {
 		glog.Fatalf("Filer Fail to serve: %v", e)
 	}

weed/command/master.go

@@ -53,7 +53,7 @@ func init() {
 	m.port = cmdMaster.Flag.Int("port", 9333, "http listen port")
 	m.portGrpc = cmdMaster.Flag.Int("port.grpc", 0, "grpc listen port")
 	m.ip = cmdMaster.Flag.String("ip", util.DetectedHostAddress(), "master <ip>|<server> address, also used as identifier")
-	m.ipBind = cmdMaster.Flag.String("ip.bind", "", "ip address to bind to")
+	m.ipBind = cmdMaster.Flag.String("ip.bind", "", "ip address to bind to. If empty, default to same as -ip option.")
 	m.metaFolder = cmdMaster.Flag.String("mdir", os.TempDir(), "data directory to store meta data")
 	m.peers = cmdMaster.Flag.String("peers", "", "all master nodes in comma separated ip:port list, example: 127.0.0.1:9093,127.0.0.1:9094,127.0.0.1:9095")
 	m.volumeSizeLimitMB = cmdMaster.Flag.Uint("volumeSizeLimitMB", 30*1000, "Master stops directing writes to oversized volumes.")
@@ -124,6 +124,9 @@ func startMaster(masterOption MasterOptions, masterWhiteList []string) {
 	if *masterOption.portGrpc == 0 {
 		*masterOption.portGrpc = 10000 + *masterOption.port
 	}
+	if *masterOption.ipBind == "" {
+		*masterOption.ipBind = "localhost"
+	}
 
 	myMasterAddress, peers := checkPeers(*masterOption.ip, *masterOption.port, *masterOption.portGrpc, *masterOption.peers)
 

weed/command/master_follower.go

@@ -24,7 +24,7 @@ func init() {
 	cmdMasterFollower.Run = runMasterFollower // break init cycle
 	mf.port = cmdMasterFollower.Flag.Int("port", 9334, "http listen port")
 	mf.portGrpc = cmdMasterFollower.Flag.Int("port.grpc", 0, "grpc listen port")
-	mf.ipBind = cmdMasterFollower.Flag.String("ip.bind", "", "ip address to bind to")
+	mf.ipBind = cmdMasterFollower.Flag.String("ip.bind", "", "ip address to bind to. Default to localhost.")
 	mf.peers = cmdMasterFollower.Flag.String("masters", "localhost:9333", "all master nodes in comma separated ip:port list, example: 127.0.0.1:9093,127.0.0.1:9094,127.0.0.1:9095")
 
 	mf.ip = aws.String(util.DetectedHostAddress())
@@ -111,6 +111,10 @@ func startMasterFollower(masterOptions MasterOptions) {
 	option := masterOptions.toMasterOption(nil)
 	option.IsFollower = true
 
+	if *masterOptions.ipBind == "" {
+		*masterOptions.ipBind = "localhost"
+	}
+
 	r := mux.NewRouter()
 	ms := weed_server.NewMasterServer(r, option, masters)
 	listeningAddress := util.JoinHostPort(*masterOptions.ipBind, *masterOptions.port)

weed/command/s3.go

@@ -34,12 +34,13 @@ type S3Options struct {
 	metricsHttpPort  *int
 	allowEmptyFolder *bool
 	auditLogConfig   *string
+	localFilerSocket *string
 }
 
 func init() {
 	cmdS3.Run = runS3 // break init cycle
 	s3StandaloneOptions.filer = cmdS3.Flag.String("filer", "localhost:8888", "filer server address")
-	s3StandaloneOptions.bindIp = cmdS3.Flag.String("ip.bind", "", "ip address to bind to")
+	s3StandaloneOptions.bindIp = cmdS3.Flag.String("ip.bind", "", "ip address to bind to. Default to localhost.")
 	s3StandaloneOptions.port = cmdS3.Flag.Int("port", 8333, "s3 server http listen port")
 	s3StandaloneOptions.domainName = cmdS3.Flag.String("domainName", "", "suffix of the host name in comma separated list, {bucket}.{domainName}")
 	s3StandaloneOptions.config = cmdS3.Flag.String("config", "", "path to the config file")
@@ -184,6 +185,7 @@ func (s3opt *S3Options) startS3Server() bool {
 		BucketsPath:      filerBucketsPath,
 		GrpcDialOption:   grpcDialOption,
 		AllowEmptyFolder: *s3opt.allowEmptyFolder,
+		LocalFilerSocket: s3opt.localFilerSocket,
 	})
 	if s3ApiServer_err != nil {
 		glog.Fatalf("S3 API Server startup error: %v", s3ApiServer_err)
@@ -191,6 +193,10 @@ func (s3opt *S3Options) startS3Server() bool {
 
 	httpS := &http.Server{Handler: router}
 
+	if *s3opt.bindIp == "" {
+		*s3opt.bindIp = "localhost"
+	}
+
 	listenAddress := fmt.Sprintf("%s:%d", *s3opt.bindIp, *s3opt.port)
 	s3ApiListener, err := util.NewListener(listenAddress, time.Duration(10)*time.Second)
 	if err != nil {

weed/command/server.go

@@ -55,7 +55,7 @@ var cmdServer = &Command{
 
 var (
 	serverIp                  = cmdServer.Flag.String("ip", util.DetectedHostAddress(), "ip or server name, also used as identifier")
-	serverBindIp              = cmdServer.Flag.String("ip.bind", "", "ip address to bind to")
+	serverBindIp              = cmdServer.Flag.String("ip.bind", "", "ip address to bind to. If empty, default to same as -ip option.")
 	serverTimeout             = cmdServer.Flag.Int("idleTimeout", 30, "connection idle seconds")
 	serverDataCenter          = cmdServer.Flag.String("dataCenter", "", "current volume server's data center name")
 	serverRack                = cmdServer.Flag.String("rack", "", "current volume server's rack name")
@@ -112,6 +112,7 @@ func init() {
 	filerOptions.cipher = cmdServer.Flag.Bool("filer.encryptVolumeData", false, "encrypt data on volume servers")
 	filerOptions.saveToFilerLimit = cmdServer.Flag.Int("filer.saveToFilerLimit", 0, "Small files smaller than this limit can be cached in filer store.")
 	filerOptions.concurrentUploadLimitMB = cmdServer.Flag.Int("filer.concurrentUploadLimitMB", 64, "limit total concurrent upload size")
+	filerOptions.localSocket = cmdServer.Flag.String("filer.localSocket", "", "default to /tmp/seaweedfs-filer-<port>.sock")
 
 	serverOptions.v.port = cmdServer.Flag.Int("volume.port", 8080, "volume server http listen port")
 	serverOptions.v.portGrpc = cmdServer.Flag.Int("volume.port.grpc", 0, "volume server grpc listen port")
@@ -183,6 +184,10 @@ func runServer(cmd *Command, args []string) bool {
 		masterOptions.peers = &peers
 	}
 
+	if *serverBindIp == "" {
+		serverBindIp = serverIp
+	}
+
 	// ip address
 	masterOptions.ip = serverIp
 	masterOptions.ipBind = serverBindIp
@@ -245,6 +250,7 @@ func runServer(cmd *Command, args []string) bool {
 	if *isStartingS3 {
 		go func() {
 			time.Sleep(2 * time.Second)
+			s3Options.localFilerSocket = filerOptions.localSocket
 			s3Options.startS3Server()
 		}()
 	}

weed/command/volume.go

@@ -74,7 +74,7 @@ func init() {
 	v.publicPort = cmdVolume.Flag.Int("port.public", 0, "port opened to public")
 	v.ip = cmdVolume.Flag.String("ip", util.DetectedHostAddress(), "ip or server name, also used as identifier")
 	v.publicUrl = cmdVolume.Flag.String("publicUrl", "", "Publicly accessible address")
-	v.bindIp = cmdVolume.Flag.String("ip.bind", "", "ip address to bind to")
+	v.bindIp = cmdVolume.Flag.String("ip.bind", "", "ip address to bind to. If empty, default to same as -ip option.")
 	v.mastersString = cmdVolume.Flag.String("mserver", "localhost:9333", "comma-separated master servers")
 	v.preStopSeconds = cmdVolume.Flag.Int("preStopSeconds", 10, "number of seconds between stop send heartbeats and stop volume server")
 	// v.pulseSeconds = cmdVolume.Flag.Int("pulseSeconds", 5, "number of seconds between heartbeats, must be smaller than or equal to the master's setting")
@@ -193,6 +193,9 @@ func (v VolumeServerOptions) startVolumeServer(volumeFolders, maxVolumeCounts, v
 		*v.ip = util.DetectedHostAddress()
 		glog.V(0).Infof("detected volume server ip address: %v", *v.ip)
 	}
+	if *v.bindIp == "" {
+		*v.bindIp = *v.ip
+	}
 
 	if *v.publicPort == 0 {
 		*v.publicPort = *v.port

weed/filer/filechunk_manifest.go

@@ -68,12 +68,12 @@ func ResolveChunkManifest(lookupFileIdFn wdclient.LookupFileIdFunctionType, chun
 
 		manifestChunks = append(manifestChunks, chunk)
 		// recursive
-		dchunks, mchunks, subErr := ResolveChunkManifest(lookupFileIdFn, resolvedChunks, startOffset, stopOffset)
+		dataChunks, manifestChunks, subErr := ResolveChunkManifest(lookupFileIdFn, resolvedChunks, startOffset, stopOffset)
 		if subErr != nil {
 			return chunks, nil, subErr
 		}
-		dataChunks = append(dataChunks, dchunks...)
-		manifestChunks = append(manifestChunks, mchunks...)
+		dataChunks = append(dataChunks, dataChunks...)
+		manifestChunks = append(manifestChunks, manifestChunks...)
 	}
 	return
 }
@@ -115,6 +115,15 @@ func fetchWholeChunk(bytesBuffer *bytes.Buffer, lookupFileIdFn wdclient.LookupFi
 	return nil
 }
 
+func fetchChunkRange(buffer []byte, lookupFileIdFn wdclient.LookupFileIdFunctionType, fileId string, cipherKey []byte, isGzipped bool, offset int64) (int, error) {
+	urlStrings, err := lookupFileIdFn(fileId)
+	if err != nil {
+		glog.Errorf("operation LookupFileId %s failed, err: %v", fileId, err)
+		return 0, err
+	}
+	return retriedFetchChunkData(buffer, urlStrings, cipherKey, isGzipped, false, offset)
+}
+
 func retriedFetchChunkData(buffer []byte, urlStrings []string, cipherKey []byte, isGzipped bool, isFullChunk bool, offset int64) (n int, err error) {
 
 	var shouldRetry bool

weed/filer/filer_deletion.go

@@ -1,6 +1,7 @@
 package filer
 
 import (
+	"math"
 	"strings"
 	"time"
 
@@ -129,6 +130,12 @@ func (f *Filer) DeleteChunks(chunks []*filer_pb.FileChunk) {
 	}
 }
 
+func (f *Filer) DeleteChunksNotRecursive(chunks []*filer_pb.FileChunk) {
+	for _, chunk := range chunks {
+		f.fileIdDeletionQueue.EnQueue(chunk.GetFileIdString())
+	}
+}
+
 func (f *Filer) deleteChunksIfNotNew(oldEntry, newEntry *Entry) {
 
 	if oldEntry == nil {
@@ -141,14 +148,36 @@ func (f *Filer) deleteChunksIfNotNew(oldEntry, newEntry *Entry) {
 
 	var toDelete []*filer_pb.FileChunk
 	newChunkIds := make(map[string]bool)
-	for _, newChunk := range newEntry.Chunks {
+	newDataChunks, newManifestChunks, err := ResolveChunkManifest(f.MasterClient.GetLookupFileIdFunction(),
+		newEntry.Chunks, 0, math.MaxInt64)
+	if err != nil {
+		glog.Errorf("Failed to resolve new entry chunks when delete old entry chunks. new: %s, old: %s",
+			newEntry.Chunks, oldEntry.Chunks)
+		return
+	}
+	for _, newChunk := range newDataChunks {
+		newChunkIds[newChunk.GetFileIdString()] = true
+	}
+	for _, newChunk := range newManifestChunks {
 		newChunkIds[newChunk.GetFileIdString()] = true
 	}
 
-	for _, oldChunk := range oldEntry.Chunks {
+	oldDataChunks, oldManifestChunks, err := ResolveChunkManifest(f.MasterClient.GetLookupFileIdFunction(),
+		oldEntry.Chunks, 0, math.MaxInt64)
+	if err != nil {
+		glog.Errorf("Failed to resolve old entry chunks when delete old entry chunks. new: %s, old: %s",
+			newEntry.Chunks, oldEntry.Chunks)
+		return
+	}
+	for _, oldChunk := range oldDataChunks {
 		if _, found := newChunkIds[oldChunk.GetFileIdString()]; !found {
 			toDelete = append(toDelete, oldChunk)
 		}
 	}
-	f.DeleteChunks(toDelete)
+	for _, oldChunk := range oldManifestChunks {
+		if _, found := newChunkIds[oldChunk.GetFileIdString()]; !found {
+			toDelete = append(toDelete, oldChunk)
+		}
+	}
+	f.DeleteChunksNotRecursive(toDelete)
 }

weed/filer/mysql2/mysql2_store.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"database/sql"
 	"fmt"
+	"strings"
 	"time"
 
 	"github.com/chrislusf/seaweedfs/weed/filer"
@@ -82,7 +83,7 @@ func (store *MysqlStore2) initialize(createTable, upsertQuery string, enableUpse
 		return fmt.Errorf("connect to %s error:%v", sqlUrl, err)
 	}
 
-	if err = store.CreateTable(context.Background(), abstract_sql.DEFAULT_TABLE); err != nil {
+	if err = store.CreateTable(context.Background(), abstract_sql.DEFAULT_TABLE); err != nil && !strings.Contains(err.Error(), "table already exist") {
 		return fmt.Errorf("init table %s: %v", abstract_sql.DEFAULT_TABLE, err)
 	}
 

weed/filer/reader_at.go

@@ -164,7 +164,7 @@ func (c *ChunkReadAt) doReadAt(p []byte, offset int64) (n int, err error) {
 func (c *ChunkReadAt) readChunkSliceAt(buffer []byte, chunkView *ChunkView, nextChunkViews []*ChunkView, offset uint64) (n int, err error) {
 
 	if c.readerPattern.IsRandomMode() {
-		return c.readerCache.ReadChunkAt(buffer, chunkView.FileId, chunkView.CipherKey, chunkView.IsGzipped, int64(offset), int(chunkView.ChunkSize), chunkView.LogicOffset == 0)
+		return fetchChunkRange(buffer, c.readerCache.lookupFileIdFn, chunkView.FileId, chunkView.CipherKey, chunkView.IsGzipped, int64(offset))
 	}
 
 	n, err = c.readerCache.ReadChunkAt(buffer, chunkView.FileId, chunkView.CipherKey, chunkView.IsGzipped, int64(offset), int(chunkView.ChunkSize), chunkView.LogicOffset == 0)

weed/filer/reader_at_test.go

@@ -84,7 +84,7 @@ func TestReaderAt(t *testing.T) {
 
 func testReadAt(t *testing.T, readerAt *ChunkReadAt, offset int64, size int, expected int, expectedErr error) {
 	data := make([]byte, size)
-	n, err := readerAt.ReadAt(data, offset)
+	n, err := readerAt.doReadAt(data, offset)
 
 	for _, d := range data {
 		fmt.Printf("%x", d)

weed/mount/dirty_pages_chunked.go

@@ -30,16 +30,19 @@ func newMemoryChunkPages(fh *FileHandle, chunkSize int64) *ChunkedDirtyPages {
 		fh: fh,
 	}
 
-	dirtyPages.uploadPipeline = page_writer.NewUploadPipeline(fh.wfs.concurrentWriters, chunkSize, dirtyPages.saveChunkedFileIntevalToStorage, fh.wfs.option.ConcurrentWriters)
+	swapFileDir := fh.wfs.option.getTempFilePageDir()
+
+	dirtyPages.uploadPipeline = page_writer.NewUploadPipeline(fh.wfs.concurrentWriters, chunkSize,
+		dirtyPages.saveChunkedFileIntevalToStorage, fh.wfs.option.ConcurrentWriters, swapFileDir)
 
 	return dirtyPages
 }
 
-func (pages *ChunkedDirtyPages) AddPage(offset int64, data []byte) {
+func (pages *ChunkedDirtyPages) AddPage(offset int64, data []byte, isSequential bool) {
 	pages.hasWrites = true
 
 	glog.V(4).Infof("%v memory AddPage [%d, %d)", pages.fh.fh, offset, offset+int64(len(data)))
-	pages.uploadPipeline.SaveDataAt(data, offset)
+	pages.uploadPipeline.SaveDataAt(data, offset, isSequential)
 
 	return
 }

weed/mount/filehandle.go

@@ -88,6 +88,10 @@ func (fh *FileHandle) addChunks(chunks []*filer_pb.FileChunk) {
 	fh.chunkAddLock.Unlock()
 }
 
+func (fh *FileHandle) Release() {
+	fh.dirtyPages.Destroy()
+}
+
 func lessThan(a, b *filer_pb.FileChunk) bool {
 	if a.Mtime == b.Mtime {
 		return a.Fid.FileKey < b.Fid.FileKey

weed/mount/filehandle_map.go

@@ -61,6 +61,7 @@ func (i *FileHandleToInode) ReleaseByInode(inode uint64) {
 		if fh.counter <= 0 {
 			delete(i.inode2fh, inode)
 			delete(i.fh2inode, fh.fh)
+			fh.Release()
 		}
 	}
 }
@@ -77,6 +78,7 @@ func (i *FileHandleToInode) ReleaseByHandle(fh FileHandleId) {
 			if fhHandle.counter <= 0 {
 				delete(i.inode2fh, inode)
 				delete(i.fh2inode, fhHandle.fh)
+				fhHandle.Release()
 			}
 		}
 

weed/mount/inode_to_path.go

@@ -4,7 +4,6 @@ import (
 	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/chrislusf/seaweedfs/weed/util"
 	"github.com/hanwen/go-fuse/v2/fuse"
-	"os"
 	"sync"
 )
 
@@ -31,13 +30,13 @@ func NewInodeToPath(root util.FullPath) *InodeToPath {
 	return t
 }
 
-func (i *InodeToPath) Lookup(path util.FullPath, mode os.FileMode, isHardlink bool, possibleInode uint64, isLookup bool) uint64 {
+func (i *InodeToPath) Lookup(path util.FullPath, unixTime int64, isDirectory bool, isHardlink bool, possibleInode uint64, isLookup bool) uint64 {
 	i.Lock()
 	defer i.Unlock()
 	inode, found := i.path2inode[path]
 	if !found {
 		if possibleInode == 0 {
-			inode = path.AsInode(mode)
+			inode = path.AsInode(unixTime)
 		} else {
 			inode = possibleInode
 		}
@@ -55,22 +54,22 @@ func (i *InodeToPath) Lookup(path util.FullPath, mode os.FileMode, isHardlink bo
 		}
 	} else {
 		if !isLookup {
-			i.inode2path[inode] = &InodeEntry{path, 0, mode&os.ModeDir > 0, false}
+			i.inode2path[inode] = &InodeEntry{path, 0, isDirectory, false}
 		} else {
-			i.inode2path[inode] = &InodeEntry{path, 1, mode&os.ModeDir > 0, false}
+			i.inode2path[inode] = &InodeEntry{path, 1, isDirectory, false}
 		}
 	}
 
 	return inode
 }
 
-func (i *InodeToPath) AllocateInode(path util.FullPath, mode os.FileMode) uint64 {
+func (i *InodeToPath) AllocateInode(path util.FullPath, unixTime int64) uint64 {
 	if path == "/" {
 		return 1
 	}
 	i.Lock()
 	defer i.Unlock()
-	inode := path.AsInode(mode)
+	inode := path.AsInode(unixTime)
 	for _, found := i.inode2path[inode]; found; inode++ {
 		_, found = i.inode2path[inode]
 	}

weed/mount/page_writer.go

@@ -6,10 +6,11 @@ import (
 )
 
 type PageWriter struct {
-	fh          *FileHandle
-	collection  string
-	replication string
-	chunkSize   int64
+	fh            *FileHandle
+	collection    string
+	replication   string
+	chunkSize     int64
+	writerPattern *WriterPattern
 
 	randomWriter page_writer.DirtyPages
 }
@@ -20,29 +21,29 @@ var (
 
 func newPageWriter(fh *FileHandle, chunkSize int64) *PageWriter {
 	pw := &PageWriter{
-		fh:           fh,
-		chunkSize:    chunkSize,
-		randomWriter: newMemoryChunkPages(fh, chunkSize),
-		// randomWriter: newTempFileDirtyPages(fh.f, chunkSize),
+		fh:            fh,
+		chunkSize:     chunkSize,
+		writerPattern: NewWriterPattern(chunkSize),
+		randomWriter:  newMemoryChunkPages(fh, chunkSize),
 	}
 	return pw
 }
 
-func (pw *PageWriter) AddPage(offset int64, data []byte) {
+func (pw *PageWriter) AddPage(offset int64, data []byte, isSequentail bool) {
 
 	glog.V(4).Infof("%v AddPage [%d, %d)", pw.fh.fh, offset, offset+int64(len(data)))
 
 	chunkIndex := offset / pw.chunkSize
 	for i := chunkIndex; len(data) > 0; i++ {
 		writeSize := min(int64(len(data)), (i+1)*pw.chunkSize-offset)
-		pw.addToOneChunk(i, offset, data[:writeSize])
+		pw.addToOneChunk(i, offset, data[:writeSize], isSequentail)
 		offset += writeSize
 		data = data[writeSize:]
 	}
 }
 
-func (pw *PageWriter) addToOneChunk(chunkIndex, offset int64, data []byte) {
-	pw.randomWriter.AddPage(offset, data)
+func (pw *PageWriter) addToOneChunk(chunkIndex, offset int64, data []byte, isSequential bool) {
+	pw.randomWriter.AddPage(offset, data, isSequential)
 }
 
 func (pw *PageWriter) FlushData() error {

weed/mount/page_writer/dirty_pages.go

@@ -1,7 +1,7 @@
 package page_writer
 
 type DirtyPages interface {
-	AddPage(offset int64, data []byte)
+	AddPage(offset int64, data []byte, isSequential bool)
 	FlushData() error
 	ReadDirtyDataAt(data []byte, startOffset int64) (maxStop int64)
 	GetStorageOptions() (collection, replication string)

weed/mount/page_writer/page_chunk_mem.go

@@ -3,10 +3,13 @@ package page_writer
 import (
 	"github.com/chrislusf/seaweedfs/weed/util"
 	"github.com/chrislusf/seaweedfs/weed/util/mem"
+	"sync/atomic"
 )
 
 var (
 	_ = PageChunk(&MemChunk{})
+
+	memChunkCounter int64
 )
 
 type MemChunk struct {
@@ -17,6 +20,7 @@ type MemChunk struct {
 }
 
 func NewMemChunk(logicChunkIndex LogicChunkIndex, chunkSize int64) *MemChunk {
+	atomic.AddInt64(&memChunkCounter, 1)
 	return &MemChunk{
 		logicChunkIndex: logicChunkIndex,
 		chunkSize:       chunkSize,
@@ -26,6 +30,7 @@ func NewMemChunk(logicChunkIndex LogicChunkIndex, chunkSize int64) *MemChunk {
 }
 
 func (mc *MemChunk) FreeResource() {
+	atomic.AddInt64(&memChunkCounter, -1)
 	mem.Free(mc.buf)
 }
 

weed/mount/page_writer/page_chunk_swapfile.go

@@ -18,6 +18,7 @@ type SwapFile struct {
 	file                    *os.File
 	logicToActualChunkIndex map[LogicChunkIndex]ActualChunkIndex
 	chunkSize               int64
+	freeActualChunkList     []ActualChunkIndex
 }
 
 type SwapFileChunk struct {
@@ -53,7 +54,12 @@ func (sf *SwapFile) NewTempFileChunk(logicChunkIndex LogicChunkIndex) (tc *SwapF
 	}
 	actualChunkIndex, found := sf.logicToActualChunkIndex[logicChunkIndex]
 	if !found {
-		actualChunkIndex = ActualChunkIndex(len(sf.logicToActualChunkIndex))
+		if len(sf.freeActualChunkList) > 0 {
+			actualChunkIndex = sf.freeActualChunkList[0]
+			sf.freeActualChunkList = sf.freeActualChunkList[1:]
+		} else {
+			actualChunkIndex = ActualChunkIndex(len(sf.logicToActualChunkIndex))
+		}
 		sf.logicToActualChunkIndex[logicChunkIndex] = actualChunkIndex
 	}
 
@@ -66,6 +72,8 @@ func (sf *SwapFile) NewTempFileChunk(logicChunkIndex LogicChunkIndex) (tc *SwapF
 }
 
 func (sc *SwapFileChunk) FreeResource() {
+	sc.swapfile.freeActualChunkList = append(sc.swapfile.freeActualChunkList, sc.actualChunkIndex)
+	delete(sc.swapfile.logicToActualChunkIndex, sc.logicChunkIndex)
 }
 
 func (sc *SwapFileChunk) WriteDataAt(src []byte, offset int64) (n int) {

weed/mount/page_writer/upload_pipeline.go

@@ -24,7 +24,8 @@ type UploadPipeline struct {
 	saveToStorageFn      SaveToStorageFunc
 	activeReadChunks     map[LogicChunkIndex]int
 	activeReadChunksLock sync.Mutex
-	bufferChunkLimit     int
+	writableChunkLimit   int
+	swapFile             *SwapFile
 }
 
 type SealedChunk struct {
@@ -40,30 +41,30 @@ func (sc *SealedChunk) FreeReference(messageOnFree string) {
 	}
 }
 
-func NewUploadPipeline(writers *util.LimitedConcurrentExecutor, chunkSize int64, saveToStorageFn SaveToStorageFunc, bufferChunkLimit int) *UploadPipeline {
+func NewUploadPipeline(writers *util.LimitedConcurrentExecutor, chunkSize int64, saveToStorageFn SaveToStorageFunc, bufferChunkLimit int, swapFileDir string) *UploadPipeline {
 	return &UploadPipeline{
-		ChunkSize:         chunkSize,
-		writableChunks:    make(map[LogicChunkIndex]PageChunk),
-		sealedChunks:      make(map[LogicChunkIndex]*SealedChunk),
-		uploaders:         writers,
-		uploaderCountCond: sync.NewCond(&sync.Mutex{}),
-		saveToStorageFn:   saveToStorageFn,
-		activeReadChunks:  make(map[LogicChunkIndex]int),
-		bufferChunkLimit:  bufferChunkLimit,
+		ChunkSize:          chunkSize,
+		writableChunks:     make(map[LogicChunkIndex]PageChunk),
+		sealedChunks:       make(map[LogicChunkIndex]*SealedChunk),
+		uploaders:          writers,
+		uploaderCountCond:  sync.NewCond(&sync.Mutex{}),
+		saveToStorageFn:    saveToStorageFn,
+		activeReadChunks:   make(map[LogicChunkIndex]int),
+		writableChunkLimit: bufferChunkLimit,
+		swapFile:           NewSwapFile(swapFileDir, chunkSize),
 	}
 }
 
-func (up *UploadPipeline) SaveDataAt(p []byte, off int64) (n int) {
+func (up *UploadPipeline) SaveDataAt(p []byte, off int64, isSequential bool) (n int) {
 	up.writableChunksLock.Lock()
 	defer up.writableChunksLock.Unlock()
 
 	logicChunkIndex := LogicChunkIndex(off / up.ChunkSize)
 
-	memChunk, found := up.writableChunks[logicChunkIndex]
+	pageChunk, found := up.writableChunks[logicChunkIndex]
 	if !found {
-		if len(up.writableChunks) < up.bufferChunkLimit {
-			memChunk = NewMemChunk(logicChunkIndex, up.ChunkSize)
-		} else {
+		if len(up.writableChunks) > up.writableChunkLimit {
+			// if current file chunks is over the per file buffer count limit
 			fullestChunkIndex, fullness := LogicChunkIndex(-1), int64(0)
 			for lci, mc := range up.writableChunks {
 				chunkFullness := mc.WrittenSize()
@@ -74,13 +75,19 @@ func (up *UploadPipeline) SaveDataAt(p []byte, off int64) (n int) {
 			}
 			up.moveToSealed(up.writableChunks[fullestChunkIndex], fullestChunkIndex)
 			delete(up.writableChunks, fullestChunkIndex)
-			fmt.Printf("flush chunk %d with %d bytes written", logicChunkIndex, fullness)
-			memChunk = NewMemChunk(logicChunkIndex, up.ChunkSize)
+			// fmt.Printf("flush chunk %d with %d bytes written\n", logicChunkIndex, fullness)
 		}
-		up.writableChunks[logicChunkIndex] = memChunk
+		if isSequential &&
+			len(up.writableChunks) < up.writableChunkLimit &&
+			atomic.LoadInt64(&memChunkCounter) < 4*int64(up.writableChunkLimit) {
+			pageChunk = NewMemChunk(logicChunkIndex, up.ChunkSize)
+		} else {
+			pageChunk = up.swapFile.NewTempFileChunk(logicChunkIndex)
+		}
+		up.writableChunks[logicChunkIndex] = pageChunk
 	}
-	n = memChunk.WriteDataAt(p, off)
-	up.maybeMoveToSealed(memChunk, logicChunkIndex)
+	n = pageChunk.WriteDataAt(p, off)
+	up.maybeMoveToSealed(pageChunk, logicChunkIndex)
 
 	return
 }
@@ -179,4 +186,8 @@ func (up *UploadPipeline) moveToSealed(memChunk PageChunk, logicChunkIndex Logic
 }
 
 func (up *UploadPipeline) Shutdown() {
+	up.swapFile.FreeResource()
+	for logicChunkIndex, sealedChunk := range up.sealedChunks {
+		sealedChunk.FreeReference(fmt.Sprintf("%s uploadpipeline shutdown chunk %d", up.filepath, logicChunkIndex))
+	}
 }

weed/mount/page_writer/upload_pipeline_test.go

@@ -7,7 +7,7 @@ import (
 
 func TestUploadPipeline(t *testing.T) {
 
-	uploadPipeline := NewUploadPipeline(nil, 2*1024*1024, nil, 16)
+	uploadPipeline := NewUploadPipeline(nil, 2*1024*1024, nil, 16, "")
 
 	writeRange(uploadPipeline, 0, 131072)
 	writeRange(uploadPipeline, 131072, 262144)
@@ -31,7 +31,7 @@ func writeRange(uploadPipeline *UploadPipeline, startOff, stopOff int64) {
 	p := make([]byte, 4)
 	for i := startOff / 4; i < stopOff/4; i += 4 {
 		util.Uint32toBytes(p, uint32(i))
-		uploadPipeline.SaveDataAt(p, i)
+		uploadPipeline.SaveDataAt(p, i, false)
 	}
 }
 

weed/mount/page_writer_pattern.go

@@ -0,0 +1,44 @@
+package mount
+
+type WriterPattern struct {
+	isStreaming     bool
+	lastWriteOffset int64
+	chunkSize       int64
+}
+
+// For streaming write: only cache the first chunk
+// For random write: fall back to temp file approach
+// writes can only change from streaming mode to non-streaming mode
+
+func NewWriterPattern(chunkSize int64) *WriterPattern {
+	return &WriterPattern{
+		isStreaming:     true,
+		lastWriteOffset: -1,
+		chunkSize:       chunkSize,
+	}
+}
+
+func (rp *WriterPattern) MonitorWriteAt(offset int64, size int) {
+	if rp.lastWriteOffset > offset {
+		rp.isStreaming = false
+	}
+	if rp.lastWriteOffset == -1 {
+		if offset != 0 {
+			rp.isStreaming = false
+		}
+	}
+	rp.lastWriteOffset = offset
+}
+
+func (rp *WriterPattern) IsStreamingMode() bool {
+	return rp.isStreaming
+}
+
+func (rp *WriterPattern) IsRandomMode() bool {
+	return !rp.isStreaming
+}
+
+func (rp *WriterPattern) Reset() {
+	rp.isStreaming = true
+	rp.lastWriteOffset = -1
+}

weed/mount/weedfs.go

@@ -16,6 +16,7 @@ import (
 	"math/rand"
 	"os"
 	"path"
+	"path/filepath"
 	"time"
 
 	"github.com/hanwen/go-fuse/v2/fs"
@@ -50,10 +51,12 @@ type Option struct {
 	Cipher             bool   // whether encrypt data on volume server
 	UidGidMapper       *meta_cache.UidGidMapper
 
-	uniqueCacheDir string
+	uniqueCacheDir         string
+	uniqueCacheTempPageDir string
 }
 
 type WFS struct {
+	// https://dl.acm.org/doi/fullHtml/10.1145/3310148
 	// follow https://github.com/hanwen/go-fuse/blob/master/fuse/api.go
 	fuse.RawFileSystem
 	fs.Inode
@@ -177,7 +180,12 @@ func (wfs *WFS) getCurrentFiler() pb.ServerAddress {
 func (option *Option) setupUniqueCacheDirectory() {
 	cacheUniqueId := util.Md5String([]byte(option.MountDirectory + string(option.FilerAddresses[0]) + option.FilerMountRootPath + util.Version()))[0:8]
 	option.uniqueCacheDir = path.Join(option.CacheDir, cacheUniqueId)
-	os.MkdirAll(option.uniqueCacheDir, os.FileMode(0777)&^option.Umask)
+	option.uniqueCacheTempPageDir = filepath.Join(option.uniqueCacheDir, "swap")
+	os.MkdirAll(option.uniqueCacheTempPageDir, os.FileMode(0777)&^option.Umask)
+}
+
+func (option *Option) getTempFilePageDir() string {
+	return option.uniqueCacheTempPageDir
 }
 
 func (option *Option) getUniqueCacheDir() string {

weed/mount/weedfs_attr.go

@@ -79,14 +79,23 @@ func (wfs *WFS) SetAttr(cancel <-chan struct{}, input *fuse.SetAttrIn, out *fuse
 	if mode, ok := input.GetMode(); ok {
 		// glog.V(4).Infof("setAttr mode %o", mode)
 		entry.Attributes.FileMode = chmod(entry.Attributes.FileMode, mode)
+		if input.NodeId == 1 {
+			wfs.option.MountMode = os.FileMode(chmod(uint32(wfs.option.MountMode), mode))
+		}
 	}
 
 	if uid, ok := input.GetUID(); ok {
 		entry.Attributes.Uid = uid
+		if input.NodeId == 1 {
+			wfs.option.MountUid = uid
+		}
 	}
 
 	if gid, ok := input.GetGID(); ok {
 		entry.Attributes.Gid = gid
+		if input.NodeId == 1 {
+			wfs.option.MountGid = gid
+		}
 	}
 
 	if mtime, ok := input.GetMTime(); ok {

weed/mount/weedfs_dir_lookup.go

@@ -53,9 +53,9 @@ func (wfs *WFS) Lookup(cancel <-chan struct{}, header *fuse.InHeader, name strin
 		return fuse.ENOENT
 	}
 
-	inode := wfs.inodeToPath.Lookup(fullFilePath, localEntry.Mode, len(localEntry.HardLinkId) > 0, localEntry.Inode, true)
+	inode := wfs.inodeToPath.Lookup(fullFilePath, localEntry.Crtime.Unix(), localEntry.IsDirectory(), len(localEntry.HardLinkId) > 0, localEntry.Inode, true)
 
-	if fh, found := wfs.fhmap.FindFileHandle(inode); found {
+	if fh, found := wfs.fhmap.FindFileHandle(inode); found && fh.entry != nil {
 		glog.V(4).Infof("lookup opened file %s size %d", dirPath.Child(localEntry.Name()), filer.FileSize(fh.entry))
 		localEntry = filer.FromPbEntry(string(dirPath), fh.entry)
 	}

weed/mount/weedfs_dir_mkrm.go

@@ -78,7 +78,7 @@ func (wfs *WFS) Mkdir(cancel <-chan struct{}, in *fuse.MkdirIn, name string, out
 		return fuse.EIO
 	}
 
-	inode := wfs.inodeToPath.Lookup(entryFullPath, os.ModeDir, false, 0, true)
+	inode := wfs.inodeToPath.Lookup(entryFullPath, newEntry.Attributes.Crtime, true, false, 0, true)
 
 	wfs.outputPbEntry(out, inode, newEntry)
 

weed/mount/weedfs_dir_read.go

@@ -162,14 +162,14 @@ func (wfs *WFS) doReadDirectory(input *fuse.ReadIn, out *fuse.DirEntryList, isPl
 		dirEntry.Name = entry.Name()
 		dirEntry.Mode = toSyscallMode(entry.Mode)
 		if !isPlusMode {
-			inode := wfs.inodeToPath.Lookup(dirPath.Child(dirEntry.Name), entry.Mode, len(entry.HardLinkId) > 0, entry.Inode, false)
+			inode := wfs.inodeToPath.Lookup(dirPath.Child(dirEntry.Name), entry.Crtime.Unix(), entry.IsDirectory(), len(entry.HardLinkId) > 0, entry.Inode, false)
 			dirEntry.Ino = inode
 			if !out.AddDirEntry(dirEntry) {
 				isEarlyTerminated = true
 				return false
 			}
 		} else {
-			inode := wfs.inodeToPath.Lookup(dirPath.Child(dirEntry.Name), entry.Mode, len(entry.HardLinkId) > 0, entry.Inode, true)
+			inode := wfs.inodeToPath.Lookup(dirPath.Child(dirEntry.Name), entry.Crtime.Unix(), entry.IsDirectory(), len(entry.HardLinkId) > 0, entry.Inode, true)
 			dirEntry.Ino = inode
 			entryOut := out.AddDirLookupEntry(dirEntry)
 			if entryOut == nil {

weed/mount/weedfs_file_io.go

@@ -61,12 +61,13 @@ import (
 	 * @param fi file information
 */
 func (wfs *WFS) Open(cancel <-chan struct{}, in *fuse.OpenIn, out *fuse.OpenOut) (status fuse.Status) {
-	fileHandle, code := wfs.AcquireHandle(in.NodeId, in.Uid, in.Gid)
-	if code == fuse.OK {
+	var fileHandle *FileHandle
+	fileHandle, status = wfs.AcquireHandle(in.NodeId, in.Uid, in.Gid)
+	if status == fuse.OK {
 		out.Fh = uint64(fileHandle.fh)
 		// TODO https://github.com/libfuse/libfuse/blob/master/include/fuse_common.h#L64
 	}
-	return code
+	return status
 }
 
 /**

weed/mount/weedfs_file_mkrm.go

@@ -51,14 +51,15 @@ func (wfs *WFS) Mknod(cancel <-chan struct{}, in *fuse.MknodIn, name string, out
 
 	entryFullPath := dirFullPath.Child(name)
 	fileMode := toOsFileMode(in.Mode)
-	inode := wfs.inodeToPath.AllocateInode(entryFullPath, fileMode)
+	now := time.Now().Unix()
+	inode := wfs.inodeToPath.AllocateInode(entryFullPath, now)
 
 	newEntry := &filer_pb.Entry{
 		Name:        name,
 		IsDirectory: false,
 		Attributes: &filer_pb.FuseAttributes{
-			Mtime:       time.Now().Unix(),
-			Crtime:      time.Now().Unix(),
+			Mtime:       now,
+			Crtime:      now,
 			FileMode:    uint32(fileMode),
 			Uid:         in.Uid,
 			Gid:         in.Gid,
@@ -101,7 +102,7 @@ func (wfs *WFS) Mknod(cancel <-chan struct{}, in *fuse.MknodIn, name string, out
 	}
 
 	// this is to increase nlookup counter
-	inode = wfs.inodeToPath.Lookup(entryFullPath, fileMode, false, inode, true)
+	inode = wfs.inodeToPath.Lookup(entryFullPath, newEntry.Attributes.Crtime, false, false, inode, true)
 
 	wfs.outputPbEntry(out, inode, newEntry)
 

weed/mount/weedfs_file_write.go

@@ -43,6 +43,8 @@ func (wfs *WFS) Write(cancel <-chan struct{}, in *fuse.WriteIn, data []byte) (wr
 		return 0, fuse.ENOENT
 	}
 
+	fh.dirtyPages.writerPattern.MonitorWriteAt(int64(in.Offset), int(in.Size))
+
 	fh.Lock()
 	defer fh.Unlock()
 
@@ -56,7 +58,7 @@ func (wfs *WFS) Write(cancel <-chan struct{}, in *fuse.WriteIn, data []byte) (wr
 	entry.Attributes.FileSize = uint64(max(offset+int64(len(data)), int64(entry.Attributes.FileSize)))
 	// glog.V(4).Infof("%v write [%d,%d) %d", fh.f.fullpath(), req.Offset, req.Offset+int64(len(req.Data)), len(req.Data))
 
-	fh.dirtyPages.AddPage(offset, data)
+	fh.dirtyPages.AddPage(offset, data, fh.dirtyPages.writerPattern.IsStreamingMode())
 
 	written = uint32(len(data))
 

weed/mount/weedfs_filehandle.go

@@ -1,9 +1,13 @@
 package mount
 
-import "github.com/hanwen/go-fuse/v2/fuse"
+import (
+	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
+	"github.com/hanwen/go-fuse/v2/fuse"
+)
 
-func (wfs *WFS) AcquireHandle(inode uint64, uid, gid uint32) (fileHandle *FileHandle, code fuse.Status) {
-	_, _, entry, status := wfs.maybeReadEntry(inode)
+func (wfs *WFS) AcquireHandle(inode uint64, uid, gid uint32) (fileHandle *FileHandle, status fuse.Status) {
+	var entry *filer_pb.Entry
+	_, _, entry, status = wfs.maybeReadEntry(inode)
 	if status == fuse.OK {
 		fileHandle = wfs.fhmap.AcquireFileHandle(wfs, inode, entry)
 		fileHandle.entry = entry

weed/mount/weedfs_link.go

@@ -102,7 +102,7 @@ func (wfs *WFS) Link(cancel <-chan struct{}, in *fuse.LinkIn, name string, out *
 		return fuse.EIO
 	}
 
-	inode := wfs.inodeToPath.Lookup(newEntryPath, oldEntry.FileMode(), true, oldEntry.Attributes.Inode, true)
+	inode := wfs.inodeToPath.Lookup(newEntryPath, oldEntry.Attributes.Crtime, oldEntry.IsDirectory, true, oldEntry.Attributes.Inode, true)
 
 	wfs.outputPbEntry(out, inode, request.Entry)
 

weed/mount/weedfs_symlink.go

@@ -63,7 +63,7 @@ func (wfs *WFS) Symlink(cancel <-chan struct{}, header *fuse.InHeader, target st
 		return fuse.EIO
 	}
 
-	inode := wfs.inodeToPath.Lookup(entryFullPath, os.ModeSymlink, false, 0, true)
+	inode := wfs.inodeToPath.Lookup(entryFullPath, request.Entry.Attributes.Crtime, false, false, 0, true)
 
 	wfs.outputPbEntry(out, inode, request.Entry)
 

weed/s3api/s3api_object_handlers.go

@@ -27,17 +27,6 @@ import (
 	"github.com/chrislusf/seaweedfs/weed/util"
 )
 
-var (
-	client *http.Client
-)
-
-func init() {
-	client = &http.Client{Transport: &http.Transport{
-		MaxIdleConns:        1024,
-		MaxIdleConnsPerHost: 1024,
-	}}
-}
-
 func mimeDetect(r *http.Request, dataReader io.Reader) io.ReadCloser {
 	mimeBuffer := make([]byte, 512)
 	size, _ := dataReader.Read(mimeBuffer)
@@ -105,8 +94,7 @@ func (s3a *S3ApiServer) PutObjectHandler(w http.ResponseWriter, r *http.Request)
 			return
 		}
 	} else {
-		uploadUrl := fmt.Sprintf("http://%s%s/%s%s", s3a.option.Filer.ToHttpAddress(), s3a.option.BucketsPath, bucket, urlPathEscape(object))
-
+		uploadUrl := s3a.toFilerUrl(bucket, object)
 		if r.Header.Get("Content-Type") == "" {
 			dataReader = mimeDetect(r, dataReader)
 		}
@@ -132,6 +120,12 @@ func urlPathEscape(object string) string {
 	return strings.Join(escapedParts, "/")
 }
 
+func (s3a *S3ApiServer) toFilerUrl(bucket, object string) string {
+	destUrl := fmt.Sprintf("http://%s%s/%s%s",
+		s3a.option.Filer.ToHttpAddress(), s3a.option.BucketsPath, bucket, urlPathEscape(object))
+	return destUrl
+}
+
 func (s3a *S3ApiServer) GetObjectHandler(w http.ResponseWriter, r *http.Request) {
 
 	bucket, object := xhttp.GetBucketAndObject(r)
@@ -142,8 +136,7 @@ func (s3a *S3ApiServer) GetObjectHandler(w http.ResponseWriter, r *http.Request)
 		return
 	}
 
-	destUrl := fmt.Sprintf("http://%s%s/%s%s",
-		s3a.option.Filer.ToHttpAddress(), s3a.option.BucketsPath, bucket, urlPathEscape(object))
+	destUrl := s3a.toFilerUrl(bucket, object)
 
 	s3a.proxyToFiler(w, r, destUrl, false, passThroughResponse)
 }
@@ -153,8 +146,7 @@ func (s3a *S3ApiServer) HeadObjectHandler(w http.ResponseWriter, r *http.Request
 	bucket, object := xhttp.GetBucketAndObject(r)
 	glog.V(3).Infof("HeadObjectHandler %s %s", bucket, object)
 
-	destUrl := fmt.Sprintf("http://%s%s/%s%s",
-		s3a.option.Filer.ToHttpAddress(), s3a.option.BucketsPath, bucket, urlPathEscape(object))
+	destUrl := s3a.toFilerUrl(bucket, object)
 
 	s3a.proxyToFiler(w, r, destUrl, false, passThroughResponse)
 }
@@ -164,8 +156,7 @@ func (s3a *S3ApiServer) DeleteObjectHandler(w http.ResponseWriter, r *http.Reque
 	bucket, object := xhttp.GetBucketAndObject(r)
 	glog.V(3).Infof("DeleteObjectHandler %s %s", bucket, object)
 
-	destUrl := fmt.Sprintf("http://%s%s/%s%s?recursive=true",
-		s3a.option.Filer.ToHttpAddress(), s3a.option.BucketsPath, bucket, urlPathEscape(object))
+	destUrl := s3a.toFilerUrl(bucket, object)
 
 	s3a.proxyToFiler(w, r, destUrl, true, func(proxyResponse *http.Response, w http.ResponseWriter) (statusCode int) {
 		statusCode = http.StatusNoContent
@@ -333,7 +324,7 @@ func (s3a *S3ApiServer) proxyToFiler(w http.ResponseWriter, r *http.Request, des
 	// ensure that the Authorization header is overriding any previous
 	// Authorization header which might be already present in proxyReq
 	s3a.maybeAddFilerJwtAuthorization(proxyReq, isWrite)
-	resp, postErr := client.Do(proxyReq)
+	resp, postErr := s3a.client.Do(proxyReq)
 
 	if postErr != nil {
 		glog.Errorf("post to filer: %v", postErr)
@@ -399,7 +390,7 @@ func (s3a *S3ApiServer) putToFiler(r *http.Request, uploadUrl string, dataReader
 	// ensure that the Authorization header is overriding any previous
 	// Authorization header which might be already present in proxyReq
 	s3a.maybeAddFilerJwtAuthorization(proxyReq, true)
-	resp, postErr := client.Do(proxyReq)
+	resp, postErr := s3a.client.Do(proxyReq)
 
 	if postErr != nil {
 		glog.Errorf("post to filer: %v", postErr)

weed/s3api/s3api_server.go

@@ -1,7 +1,9 @@
 package s3api
 
 import (
+	"context"
 	"fmt"
+	"net"
 	"net/http"
 	"strings"
 	"time"
@@ -24,6 +26,7 @@ type S3ApiServerOption struct {
 	BucketsPath      string
 	GrpcDialOption   grpc.DialOption
 	AllowEmptyFolder bool
+	LocalFilerSocket *string
 }
 
 type S3ApiServer struct {
@@ -31,6 +34,7 @@ type S3ApiServer struct {
 	iam            *IdentityAccessManagement
 	randomClientId int32
 	filerGuard     *security.Guard
+	client         *http.Client
 }
 
 func NewS3ApiServer(router *mux.Router, option *S3ApiServerOption) (s3ApiServer *S3ApiServer, err error) {
@@ -49,6 +53,20 @@ func NewS3ApiServer(router *mux.Router, option *S3ApiServerOption) (s3ApiServer
 		randomClientId: util.RandomInt32(),
 		filerGuard:     security.NewGuard([]string{}, signingKey, expiresAfterSec, readSigningKey, readExpiresAfterSec),
 	}
+	if option.LocalFilerSocket == nil {
+		s3ApiServer.client = &http.Client{Transport: &http.Transport{
+			MaxIdleConns:        1024,
+			MaxIdleConnsPerHost: 1024,
+		}}
+	} else {
+		s3ApiServer.client = &http.Client{
+			Transport: &http.Transport{
+				DialContext: func(_ context.Context, _, _ string) (net.Conn, error) {
+					return net.Dial("unix", *option.LocalFilerSocket)
+				},
+			},
+		}
+	}
 
 	s3ApiServer.registerRouter(router)
 

weed/server/common.go

@@ -1,12 +1,12 @@
 package weed_server
 
 import (
+	"bufio"
 	"bytes"
 	"encoding/json"
 	"errors"
 	"fmt"
 	xhttp "github.com/chrislusf/seaweedfs/weed/s3api/http"
-	"github.com/chrislusf/seaweedfs/weed/util/mem"
 	"io"
 	"io/fs"
 	"mime/multipart"
@@ -278,10 +278,12 @@ func adjustHeaderContentDisposition(w http.ResponseWriter, r *http.Request, file
 
 func processRangeRequest(r *http.Request, w http.ResponseWriter, totalSize int64, mimeType string, writeFn func(writer io.Writer, offset int64, size int64) error) {
 	rangeReq := r.Header.Get("Range")
+	bufferedWriter := bufio.NewWriterSize(w, 128*1024)
+	defer bufferedWriter.Flush()
 
 	if rangeReq == "" {
 		w.Header().Set("Content-Length", strconv.FormatInt(totalSize, 10))
-		if err := writeFn(w, 0, totalSize); err != nil {
+		if err := writeFn(bufferedWriter, 0, totalSize); err != nil {
 			http.Error(w, err.Error(), http.StatusInternalServerError)
 			return
 		}
@@ -322,7 +324,7 @@ func processRangeRequest(r *http.Request, w http.ResponseWriter, totalSize int64
 		w.Header().Set("Content-Range", ra.contentRange(totalSize))
 
 		w.WriteHeader(http.StatusPartialContent)
-		err = writeFn(w, ra.start, ra.length)
+		err = writeFn(bufferedWriter, ra.start, ra.length)
 		if err != nil {
 			http.Error(w, err.Error(), http.StatusInternalServerError)
 			return
@@ -362,9 +364,7 @@ func processRangeRequest(r *http.Request, w http.ResponseWriter, totalSize int64
 		w.Header().Set("Content-Length", strconv.FormatInt(sendSize, 10))
 	}
 	w.WriteHeader(http.StatusPartialContent)
-	buf := mem.Allocate(128 * 1024)
-	defer mem.Free(buf)
-	if _, err := io.CopyBuffer(w, io.LimitReader(sendContent, sendSize), buf); err != nil {
+	if _, err := io.CopyN(bufferedWriter, sendContent, sendSize); err != nil {
 		http.Error(w, "Internal Error", http.StatusInternalServerError)
 		return
 	}

weed/server/filer_server_handlers_read.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"github.com/chrislusf/seaweedfs/weed/util/mem"
 	"io"
+	"math"
 	"mime"
 	"net/http"
 	"path/filepath"
@@ -124,7 +125,7 @@ func (fs *FilerServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request)
 		if query.Get("resolveManifest") == "true" {
 			if entry.Chunks, _, err = filer.ResolveChunkManifest(
 				fs.filer.MasterClient.GetLookupFileIdFunction(),
-				entry.Chunks, 0, int64(entry.Size())); err != nil {
+				entry.Chunks, 0, math.MaxInt64); err != nil {
 				err = fmt.Errorf("failed to resolve chunk manifest, err: %s", err.Error())
 				writeJsonError(w, r, http.StatusInternalServerError, err)
 			}

weed/server/filer_server_handlers_write_upload.go

@@ -79,6 +79,7 @@ func (fs *FilerServer) uploadReaderToChunks(w http.ResponseWriter, r *http.Reque
 			bufPool.Put(bytesBuffer)
 			atomic.AddInt64(&bytesBufferCounter, -1)
 			bytesBufferLimitCond.Signal()
+			uploadErr = err
 			break
 		}
 		if chunkOffset == 0 && !isAppend {
@@ -126,6 +127,7 @@ func (fs *FilerServer) uploadReaderToChunks(w http.ResponseWriter, r *http.Reque
 	wg.Wait()
 
 	if uploadErr != nil {
+		fs.filer.DeleteChunks(fileChunks)
 		return nil, md5Hash, 0, uploadErr, nil
 	}
 

weed/shell/command_volume_fsck.go

@@ -153,12 +153,12 @@ func (c *commandVolumeFsck) collectFilerFileIdAndPaths(volumeIdToServer map[uint
 		if verbose && entry.Entry.IsDirectory {
 			fmt.Fprintf(writer, "checking directory %s\n", util.NewFullPath(entry.Dir, entry.Entry.Name))
 		}
-		dChunks, mChunks, resolveErr := filer.ResolveChunkManifest(filer.LookupFn(c.env), entry.Entry.Chunks, 0, math.MaxInt64)
+		dataChunks, manifestChunks, resolveErr := filer.ResolveChunkManifest(filer.LookupFn(c.env), entry.Entry.Chunks, 0, math.MaxInt64)
 		if resolveErr != nil {
 			return nil
 		}
-		dChunks = append(dChunks, mChunks...)
-		for _, chunk := range dChunks {
+		dataChunks = append(dataChunks, manifestChunks...)
+		for _, chunk := range dataChunks {
 			outputChan <- &Item{
 				vid:     chunk.Fid.VolumeId,
 				fileKey: chunk.Fid.FileKey,
@@ -332,15 +332,15 @@ func (c *commandVolumeFsck) collectFilerFileIds(volumeIdToServer map[uint32]VInf
 		fileKey uint64
 	}
 	return doTraverseBfsAndSaving(c.env, nil, "/", false, func(entry *filer_pb.FullEntry, outputChan chan interface{}) (err error) {
-		dChunks, mChunks, resolveErr := filer.ResolveChunkManifest(filer.LookupFn(c.env), entry.Entry.Chunks, 0, math.MaxInt64)
+		dataChunks, manifestChunks, resolveErr := filer.ResolveChunkManifest(filer.LookupFn(c.env), entry.Entry.Chunks, 0, math.MaxInt64)
 		if resolveErr != nil {
 			if verbose {
 				fmt.Fprintf(writer, "resolving manifest chunks in %s: %v\n", util.NewFullPath(entry.Dir, entry.Entry.Name), resolveErr)
 			}
 			return nil
 		}
-		dChunks = append(dChunks, mChunks...)
-		for _, chunk := range dChunks {
+		dataChunks = append(dataChunks, manifestChunks...)
+		for _, chunk := range dataChunks {
 			outputChan <- &Item{
 				vid:     chunk.Fid.VolumeId,
 				fileKey: chunk.Fid.FileKey,

weed/storage/backend/disk_file.go

@@ -70,7 +70,10 @@ func (df *DiskFile) Close() error {
 }
 
 func (df *DiskFile) GetStat() (datSize int64, modTime time.Time, err error) {
-	return df.fileSize, df.modTime, nil
+	if df.File == nil {
+		err = os.ErrInvalid
+	}
+	return df.fileSize, df.modTime, err
 }
 
 func (df *DiskFile) Name() string {

weed/util/constants.go

@@ -5,7 +5,7 @@ import (
 )
 
 var (
-	VERSION_NUMBER = fmt.Sprintf("%.02f", 2.93)
+	VERSION_NUMBER = fmt.Sprintf("%.02f", 2.94)
 	VERSION        = sizeLimit + " " + VERSION_NUMBER
 	COMMIT         = ""
 )

weed/util/fullpath.go

@@ -1,7 +1,6 @@
 package util
 
 import (
-	"os"
 	"path/filepath"
 	"strings"
 )
@@ -43,29 +42,9 @@ func (fp FullPath) Child(name string) FullPath {
 }
 
 // AsInode an in-memory only inode representation
-func (fp FullPath) AsInode(fileMode os.FileMode) uint64 {
+func (fp FullPath) AsInode(unixTime int64) uint64 {
 	inode := uint64(HashStringToLong(string(fp)))
-	inode = inode - inode%16
-	if fileMode == 0 {
-	} else if fileMode&os.ModeDir > 0 {
-		inode += 1
-	} else if fileMode&os.ModeSymlink > 0 {
-		inode += 2
-	} else if fileMode&os.ModeDevice > 0 {
-		if fileMode&os.ModeCharDevice > 0 {
-			inode += 6
-		} else {
-			inode += 3
-		}
-	} else if fileMode&os.ModeNamedPipe > 0 {
-		inode += 4
-	} else if fileMode&os.ModeSocket > 0 {
-		inode += 5
-	} else if fileMode&os.ModeCharDevice > 0 {
-		inode += 6
-	} else if fileMode&os.ModeIrregular > 0 {
-		inode += 7
-	}
+	inode = inode + uint64(unixTime)*37
 	return inode
 }