Merge pull request #2 from chrislusf/master

sync

other/java/client/pom.xml

@@ -5,7 +5,7 @@
 
     <groupId>com.github.chrislusf</groupId>
     <artifactId>seaweedfs-client</artifactId>
-    <version>1.3.6</version>
+    <version>1.3.9</version>
 
     <parent>
         <groupId>org.sonatype.oss</groupId>

other/java/client/pom.xml.deploy

@@ -5,7 +5,7 @@
 
     <groupId>com.github.chrislusf</groupId>
     <artifactId>seaweedfs-client</artifactId>
-    <version>1.3.6</version>
+    <version>1.3.9</version>
 
     <parent>
         <groupId>org.sonatype.oss</groupId>

other/java/client/pom_debug.xml

@@ -5,7 +5,7 @@
 
     <groupId>com.github.chrislusf</groupId>
     <artifactId>seaweedfs-client</artifactId>
-    <version>1.3.6</version>
+    <version>1.3.9</version>
 
     <parent>
         <groupId>org.sonatype.oss</groupId>

other/java/hdfs2/dependency-reduced-pom.xml

@@ -15,8 +15,8 @@
       <plugin>
         <artifactId>maven-compiler-plugin</artifactId>
         <configuration>
-          <source>7</source>
+          <source>8</source>
-          <target>7</target>
+          <target>8</target>
         </configuration>
       </plugin>
       <plugin>
@@ -127,7 +127,7 @@
     </snapshotRepository>
   </distributionManagement>
   <properties>
-    <seaweedfs.client.version>1.3.6</seaweedfs.client.version>
+    <seaweedfs.client.version>1.3.9</seaweedfs.client.version>
     <hadoop.version>2.9.2</hadoop.version>
   </properties>
 </project>

other/java/hdfs2/pom.xml

@@ -5,7 +5,7 @@
     <modelVersion>4.0.0</modelVersion>
 
     <properties>
-        <seaweedfs.client.version>1.3.6</seaweedfs.client.version>
+        <seaweedfs.client.version>1.3.9</seaweedfs.client.version>
         <hadoop.version>2.9.2</hadoop.version>
     </properties>
 
@@ -31,8 +31,8 @@
                 <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-compiler-plugin</artifactId>
                 <configuration>
-                    <source>7</source>
+                    <source>8</source>
-                    <target>7</target>
+                    <target>8</target>
                 </configuration>
             </plugin>
             <plugin>

other/java/hdfs2/src/main/java/seaweed/hdfs/SeaweedOutputStream.java

@@ -169,12 +169,35 @@ public class SeaweedOutputStream extends OutputStream {
             return;
         }
 
-        buffer.flip();
+        position += submitWriteBufferToService(buffer, position);
-        int bytesLength = buffer.limit() - buffer.position();
+
-        SeaweedWrite.writeData(entry, replication, filerGrpcClient, position, buffer.array(), buffer.position(), buffer.limit());
+        buffer = ByteBufferPool.request(bufferSize);
-        // System.out.println(path + " saved [" + (position) + "," + ((position) + bytesLength) + ")");
+
-        position += bytesLength;
+    }
-        buffer.clear();
+
+    private synchronized int submitWriteBufferToService(final ByteBuffer bufferToWrite, final long writePosition) throws IOException {
+
+        bufferToWrite.flip();
+        int bytesLength = bufferToWrite.limit() - bufferToWrite.position();
+
+        if (threadExecutor.getQueue().size() >= maxConcurrentRequestCount * 2) {
+            waitForTaskToComplete();
+        }
+        final Future<Void> job = completionService.submit(() -> {
+            // System.out.println(path + " is going to save [" + (writePosition) + "," + ((writePosition) + bytesLength) + ")");
+            SeaweedWrite.writeData(entry, replication, filerGrpcClient, writePosition, bufferToWrite.array(), bufferToWrite.position(), bufferToWrite.limit());
+            // System.out.println(path + " saved [" + (writePosition) + "," + ((writePosition) + bytesLength) + ")");
+            bufferToWrite.clear();
+            ByteBufferPool.release(bufferToWrite);
+            return null;
+        });
+
+        writeOperations.add(new WriteOperation(job, writePosition, bytesLength));
+
+        // Try to shrink the queue
+        shrinkWriteOperationQueue();
+
+        return bytesLength;
 
     }
 

other/java/hdfs3/dependency-reduced-pom.xml

@@ -15,8 +15,8 @@
       <plugin>
         <artifactId>maven-compiler-plugin</artifactId>
         <configuration>
-          <source>7</source>
+          <source>8</source>
-          <target>7</target>
+          <target>8</target>
         </configuration>
       </plugin>
       <plugin>
@@ -127,7 +127,7 @@
     </snapshotRepository>
   </distributionManagement>
   <properties>
-    <seaweedfs.client.version>1.3.6</seaweedfs.client.version>
+    <seaweedfs.client.version>1.3.9</seaweedfs.client.version>
     <hadoop.version>3.1.1</hadoop.version>
   </properties>
 </project>

other/java/hdfs3/pom.xml

@@ -5,7 +5,7 @@
     <modelVersion>4.0.0</modelVersion>
 
     <properties>
-        <seaweedfs.client.version>1.3.6</seaweedfs.client.version>
+        <seaweedfs.client.version>1.3.9</seaweedfs.client.version>
         <hadoop.version>3.1.1</hadoop.version>
     </properties>
 
@@ -31,8 +31,8 @@
                 <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-compiler-plugin</artifactId>
                 <configuration>
-                    <source>7</source>
+                    <source>8</source>
-                    <target>7</target>
+                    <target>8</target>
                 </configuration>
             </plugin>
             <plugin>

other/java/hdfs3/src/main/java/seaweed/hdfs/SeaweedOutputStream.java

@@ -216,12 +216,35 @@ public class SeaweedOutputStream extends OutputStream implements Syncable, Strea
             return;
         }
 
-        buffer.flip();
+        position += submitWriteBufferToService(buffer, position);
-        int bytesLength = buffer.limit() - buffer.position();
+
-        SeaweedWrite.writeData(entry, replication, filerGrpcClient, position, buffer.array(), buffer.position(), buffer.limit());
+        buffer = ByteBufferPool.request(bufferSize);
-        // System.out.println(path + " saved [" + (position) + "," + ((position) + bytesLength) + ")");
+
-        position += bytesLength;
+    }
-        buffer.clear();
+
+    private synchronized int submitWriteBufferToService(final ByteBuffer bufferToWrite, final long writePosition) throws IOException {
+
+        bufferToWrite.flip();
+        int bytesLength = bufferToWrite.limit() - bufferToWrite.position();
+
+        if (threadExecutor.getQueue().size() >= maxConcurrentRequestCount * 2) {
+            waitForTaskToComplete();
+        }
+        final Future<Void> job = completionService.submit(() -> {
+            // System.out.println(path + " is going to save [" + (writePosition) + "," + ((writePosition) + bytesLength) + ")");
+            SeaweedWrite.writeData(entry, replication, filerGrpcClient, writePosition, bufferToWrite.array(), bufferToWrite.position(), bufferToWrite.limit());
+            // System.out.println(path + " saved [" + (writePosition) + "," + ((writePosition) + bytesLength) + ")");
+            bufferToWrite.clear();
+            ByteBufferPool.release(bufferToWrite);
+            return null;
+        });
+
+        writeOperations.add(new WriteOperation(job, writePosition, bytesLength));
+
+        // Try to shrink the queue
+        shrinkWriteOperationQueue();
+
+        return bytesLength;
 
     }
 

weed/command/backup.go

@@ -112,7 +112,7 @@ func runBackup(cmd *Command, args []string) bool {
 			return true
 		}
 	}
-	v, err := storage.NewVolume(*s.dir, *s.collection, vid, storage.NeedleMapInMemory, replication, ttl, 0, 0)
+	v, err := storage.NewVolume(util.ResolvePath(*s.dir), *s.collection, vid, storage.NeedleMapInMemory, replication, ttl, 0, 0)
 	if err != nil {
 		fmt.Printf("Error creating or reading from volume %d: %v\n", vid, err)
 		return true
@@ -137,7 +137,7 @@ func runBackup(cmd *Command, args []string) bool {
 		// remove the old data
 		v.Destroy()
 		// recreate an empty volume
-		v, err = storage.NewVolume(*s.dir, *s.collection, vid, storage.NeedleMapInMemory, replication, ttl, 0, 0)
+		v, err = storage.NewVolume(util.ResolvePath(*s.dir), *s.collection, vid, storage.NeedleMapInMemory, replication, ttl, 0, 0)
 		if err != nil {
 			fmt.Printf("Error creating or reading from volume %d: %v\n", vid, err)
 			return true

weed/command/compact.go

@@ -4,6 +4,7 @@ import (
 	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/chrislusf/seaweedfs/weed/storage"
 	"github.com/chrislusf/seaweedfs/weed/storage/needle"
+	"github.com/chrislusf/seaweedfs/weed/util"
 )
 
 func init() {
@@ -40,7 +41,7 @@ func runCompact(cmd *Command, args []string) bool {
 	preallocate := *compactVolumePreallocate * (1 << 20)
 
 	vid := needle.VolumeId(*compactVolumeId)
-	v, err := storage.NewVolume(*compactVolumePath, *compactVolumeCollection, vid,
+	v, err := storage.NewVolume(util.ResolvePath(*compactVolumePath), *compactVolumeCollection, vid,
 		storage.NeedleMapInMemory, nil, nil, preallocate, 0)
 	if err != nil {
 		glog.Fatalf("Load Volume [ERROR] %s\n", err)

weed/command/download.go

@@ -43,7 +43,7 @@ var cmdDownload = &Command{
 
 func runDownload(cmd *Command, args []string) bool {
 	for _, fid := range args {
-		if e := downloadToFile(*d.server, fid, *d.dir); e != nil {
+		if e := downloadToFile(*d.server, fid, util.ResolvePath(*d.dir)); e != nil {
 			fmt.Println("Download Error: ", fid, e)
 		}
 	}

weed/command/export.go

@@ -198,7 +198,7 @@ func runExport(cmd *Command, args []string) bool {
 	needleMap := needle_map.NewMemDb()
 	defer needleMap.Close()
 
-	if err := needleMap.LoadFromIdx(path.Join(*export.dir, fileName+".idx")); err != nil {
+	if err := needleMap.LoadFromIdx(path.Join(util.ResolvePath(*export.dir), fileName+".idx")); err != nil {
 		glog.Fatalf("cannot load needle map from %s.idx: %s", fileName, err)
 	}
 
@@ -211,7 +211,7 @@ func runExport(cmd *Command, args []string) bool {
 		fmt.Printf("key\tname\tsize\tgzip\tmime\tmodified\tttl\tdeleted\n")
 	}
 
-	err = storage.ScanVolumeFile(*export.dir, *export.collection, vid, storage.NeedleMapInMemory, volumeFileScanner)
+	err = storage.ScanVolumeFile(util.ResolvePath(*export.dir), *export.collection, vid, storage.NeedleMapInMemory, volumeFileScanner)
 	if err != nil && err != io.EOF {
 		glog.Fatalf("Export Volume File [ERROR] %s\n", err)
 	}

weed/command/filer.go

@@ -100,7 +100,7 @@ func (fo *FilerOptions) startFiler() {
 
 	defaultLevelDbDirectory := "./filerldb2"
 	if fo.defaultLevelDbDirectory != nil {
-		defaultLevelDbDirectory = *fo.defaultLevelDbDirectory + "/filerldb2"
+		defaultLevelDbDirectory = util.ResolvePath(*fo.defaultLevelDbDirectory + "/filerldb2")
 	}
 
 	var peers []string

weed/command/fix.go

@@ -11,6 +11,7 @@ import (
 	"github.com/chrislusf/seaweedfs/weed/storage/needle_map"
 	"github.com/chrislusf/seaweedfs/weed/storage/super_block"
 	"github.com/chrislusf/seaweedfs/weed/storage/types"
+	"github.com/chrislusf/seaweedfs/weed/util"
 )
 
 func init() {
@@ -67,7 +68,7 @@ func runFix(cmd *Command, args []string) bool {
 	if *fixVolumeCollection != "" {
 		baseFileName = *fixVolumeCollection + "_" + baseFileName
 	}
-	indexFileName := path.Join(*fixVolumePath, baseFileName+".idx")
+	indexFileName := path.Join(util.ResolvePath(*fixVolumePath), baseFileName+".idx")
 
 	nm := needle_map.NewMemDb()
 	defer nm.Close()
@@ -77,7 +78,7 @@ func runFix(cmd *Command, args []string) bool {
 		nm: nm,
 	}
 
-	if err := storage.ScanVolumeFile(*fixVolumePath, *fixVolumeCollection, vid, storage.NeedleMapInMemory, scanner); err != nil {
+	if err := storage.ScanVolumeFile(util.ResolvePath(*fixVolumePath), *fixVolumeCollection, vid, storage.NeedleMapInMemory, scanner); err != nil {
 		glog.Fatalf("scan .dat File: %v", err)
 		os.Remove(indexFileName)
 	}

weed/command/master.go

@@ -8,10 +8,11 @@ import (
 	"strings"
 
 	"github.com/chrislusf/raft/protobuf"
-	"github.com/chrislusf/seaweedfs/weed/util/grace"
 	"github.com/gorilla/mux"
 	"google.golang.org/grpc/reflection"
 
+	"github.com/chrislusf/seaweedfs/weed/util/grace"
+
 	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/chrislusf/seaweedfs/weed/pb"
 	"github.com/chrislusf/seaweedfs/weed/pb/master_pb"
@@ -85,7 +86,7 @@ func runMaster(cmd *Command, args []string) bool {
 	runtime.GOMAXPROCS(runtime.NumCPU())
 	grace.SetupProfiling(*masterCpuProfile, *masterMemProfile)
 
-	if err := util.TestFolderWritable(*m.metaFolder); err != nil {
+	if err := util.TestFolderWritable(util.ResolvePath(*m.metaFolder)); err != nil {
 		glog.Fatalf("Check Meta Folder (-mdir) Writable %s : %s", *m.metaFolder, err)
 	}
 
@@ -118,7 +119,7 @@ func startMaster(masterOption MasterOptions, masterWhiteList []string) {
 	}
 	// start raftServer
 	raftServer := weed_server.NewRaftServer(security.LoadClientTLS(util.GetViper(), "grpc.master"),
-		peers, myMasterAddress, *masterOption.metaFolder, ms.Topo, 5)
+		peers, myMasterAddress, util.ResolvePath(*masterOption.metaFolder), ms.Topo, 5)
 	if raftServer == nil {
 		glog.Fatalf("please verify %s is writable, see https://github.com/chrislusf/seaweedfs/issues/717", *masterOption.metaFolder)
 	}

weed/command/mount_std.go

@@ -69,7 +69,7 @@ func RunMount(option *MountOptions, umask os.FileMode) bool {
 	}
 
 	filerMountRootPath := *option.filerMountRootPath
-	dir := *option.dir
+	dir := util.ResolvePath(*option.dir)
 	chunkSizeLimitMB := *mountOptions.chunkSizeLimitMB
 
 	util.LoadConfiguration("security", false)

weed/command/server.go

@@ -176,7 +176,7 @@ func runServer(cmd *Command, args []string) bool {
 	if *masterOptions.metaFolder == "" {
 		*masterOptions.metaFolder = folders[0]
 	}
-	if err := util.TestFolderWritable(*masterOptions.metaFolder); err != nil {
+	if err := util.TestFolderWritable(util.ResolvePath(*masterOptions.metaFolder)); err != nil {
 		glog.Fatalf("Check Meta Folder (-mdir=\"%s\") Writable: %s", *masterOptions.metaFolder, err)
 	}
 	filerOptions.defaultLevelDbDirectory = masterOptions.metaFolder

weed/command/upload.go

@@ -69,7 +69,7 @@ func runUpload(cmd *Command, args []string) bool {
 		if *upload.dir == "" {
 			return false
 		}
-		filepath.Walk(*upload.dir, func(path string, info os.FileInfo, err error) error {
+		filepath.Walk(util.ResolvePath(*upload.dir), func(path string, info os.FileInfo, err error) error {
 			if err == nil {
 				if !info.IsDir() {
 					if *upload.include != "" {

weed/command/volume.go

@@ -117,7 +117,7 @@ func (v VolumeServerOptions) startVolumeServer(volumeFolders, maxVolumeCounts, v
 	// Set multiple folders and each folder's max volume count limit'
 	v.folders = strings.Split(volumeFolders, ",")
 	for _, folder := range v.folders {
-		if err := util.TestFolderWritable(folder); err != nil {
+		if err := util.TestFolderWritable(util.ResolvePath(folder)); err != nil {
 			glog.Fatalf("Check Data Folder(-dir) Writable %s : %s", folder, err)
 		}
 	}

weed/command/webdav.go

@@ -110,7 +110,7 @@ func (wo *WebDavOption) startWebDav() bool {
 		Uid:              uid,
 		Gid:              gid,
 		Cipher:           cipher,
-		CacheDir:         *wo.cacheDir,
+		CacheDir:         util.ResolvePath(*wo.cacheDir),
 		CacheSizeMB:      *wo.cacheSizeMB,
 	})
 	if webdavServer_err != nil {

weed/server/volume_server_handlers_read.go

@@ -208,7 +208,9 @@ func (vs *VolumeServer) tryHandleChunkedFile(n *needle.Needle, fileName string,
 
 func conditionallyResizeImages(originalDataReaderSeeker io.ReadSeeker, ext string, r *http.Request) io.ReadSeeker {
 	rs := originalDataReaderSeeker
-
+	if len(ext) > 0 {
+		ext = strings.ToLower(ext)
+	}
 	width, height, mode, shouldResize := shouldResizeImages(ext, r)
 	if shouldResize {
 		rs, _, _ = images.Resized(ext, originalDataReaderSeeker, width, height, mode)
@@ -217,9 +219,6 @@ func conditionallyResizeImages(originalDataReaderSeeker io.ReadSeeker, ext strin
 }
 
 func shouldResizeImages(ext string, r *http.Request) (width, height int, mode string, shouldResize bool) {
-	if len(ext) > 0 {
-		ext = strings.ToLower(ext)
-	}
 	if ext == ".png" || ext == ".jpg" || ext == ".jpeg" || ext == ".gif" {
 		if r.FormValue("width") != "" {
 			width, _ = strconv.Atoi(r.FormValue("width"))

weed/storage/store.go

@@ -16,6 +16,7 @@ import (
 	"github.com/chrislusf/seaweedfs/weed/storage/needle"
 	"github.com/chrislusf/seaweedfs/weed/storage/super_block"
 	. "github.com/chrislusf/seaweedfs/weed/storage/types"
+	"github.com/chrislusf/seaweedfs/weed/util"
 )
 
 const (
@@ -28,13 +29,13 @@ const (
 type Store struct {
 	MasterAddress       string
 	grpcDialOption      grpc.DialOption
-	volumeSizeLimit     uint64 //read from the master
+	volumeSizeLimit     uint64 // read from the master
 	Ip                  string
 	Port                int
 	PublicUrl           string
 	Locations           []*DiskLocation
-	dataCenter          string //optional informaton, overwriting master setting if exists
+	dataCenter          string // optional informaton, overwriting master setting if exists
-	rack                string //optional information, overwriting master setting if exists
+	rack                string // optional information, overwriting master setting if exists
 	connected           bool
 	NeedleMapType       NeedleMapType
 	NewVolumesChan      chan master_pb.VolumeShortInformationMessage
@@ -52,7 +53,7 @@ func NewStore(grpcDialOption grpc.DialOption, port int, ip, publicUrl string, di
 	s = &Store{grpcDialOption: grpcDialOption, Port: port, Ip: ip, PublicUrl: publicUrl, NeedleMapType: needleMapKind}
 	s.Locations = make([]*DiskLocation, 0)
 	for i := 0; i < len(dirnames); i++ {
-		location := NewDiskLocation(dirnames[i], maxVolumeCounts[i], minFreeSpacePercents[i])
+		location := NewDiskLocation(util.ResolvePath(dirnames[i]), maxVolumeCounts[i], minFreeSpacePercents[i])
 		location.loadExistingVolumes(needleMapKind)
 		s.Locations = append(s.Locations, location)
 		stats.VolumeServerMaxVolumeCounter.Add(float64(maxVolumeCounts[i]))

weed/util/file_util.go

@@ -3,6 +3,9 @@ package util
 import (
 	"errors"
 	"os"
+	"os/user"
+	"path/filepath"
+	"strings"
 	"time"
 
 	"github.com/chrislusf/seaweedfs/weed/glog"
@@ -63,3 +66,20 @@ func CheckFile(filename string) (exists, canRead, canWrite bool, modTime time.Ti
 	fileSize = fi.Size()
 	return
 }
+
+func ResolvePath(path string) string {
+
+	usr, _ := user.Current()
+	dir := usr.HomeDir
+
+	if path == "~" {
+		// In case of "~", which won't be caught by the "else if"
+		path = dir
+	} else if strings.HasPrefix(path, "~/") {
+		// Use strings.HasPrefix so we don't match paths like
+		// "/something/~/something/"
+		path = filepath.Join(dir, path[2:])
+	}
+
+	return path
+}