Merge pull request #4 from chrislusf/master

Syncing  to the original repository

7 files changed, 532 insertions, 24 deletions

diff --git a/weed/util/config.go b/weed/util/config.go
index 1ea833d1f..4ba68b800 100644
--- a/weed/util/config.go
+++ b/weed/util/config.go
@@ -9,9 +9,8 @@ type Configuration interface {
 	GetString(key string) string
 	GetBool(key string) bool
 	GetInt(key string) int
-	GetInt64(key string) int64
-	GetFloat64(key string) float64
 	GetStringSlice(key string) []string
+	SetDefault(key string, value interface{})
 }
 
 func LoadConfiguration(configFileName string, required bool) (loaded bool) {
@@ -40,5 +39,4 @@ func LoadConfiguration(configFileName string, required bool) (loaded bool) {
 	}
 
 	return true
-
 }
diff --git a/weed/util/constants.go b/weed/util/constants.go
index 0f29e3388..0916850ef 100644
--- a/weed/util/constants.go
+++ b/weed/util/constants.go
@@ -5,5 +5,5 @@ import (
 )
 
 var (
-	VERSION = fmt.Sprintf("%s %d.%d", sizeLimit, 1, 41)
+	VERSION = fmt.Sprintf("%s %d.%d", sizeLimit, 1, 47)
 )
diff --git a/weed/util/file_util.go b/weed/util/file_util.go
index 78add6724..bef9f7cd6 100644
--- a/weed/util/file_util.go
+++ b/weed/util/file_util.go
@@ -3,6 +3,7 @@ package util
 import (
 	"errors"
 	"os"
+	"time"
 
 	"github.com/chrislusf/seaweedfs/weed/glog"
 )
@@ -40,3 +41,21 @@ func FileExists(filename string) bool {
 	return true
 
 }
+
+func CheckFile(filename string) (exists, canRead, canWrite bool, modTime time.Time, fileSize int64) {
+	exists = true
+	fi, err := os.Stat(filename)
+	if os.IsNotExist(err) {
+		exists = false
+		return
+	}
+	if fi.Mode()&0400 != 0 {
+		canRead = true
+	}
+	if fi.Mode()&0200 != 0 {
+		canWrite = true
+	}
+	modTime = fi.ModTime()
+	fileSize = fi.Size()
+	return
+}
diff --git a/weed/util/http_util.go b/weed/util/http_util.go
index a465a7b7f..667d0b4be 100644
--- a/weed/util/http_util.go
+++ b/weed/util/http_util.go
@@ -11,7 +11,6 @@ import (
 	"net/http"
 	"net/url"
 	"strings"
-	"time"
 )
 
 var (
@@ -25,12 +24,11 @@ func init() {
 	}
 	client = &http.Client{
 		Transport: Transport,
-		Timeout:   5 * time.Second,
 	}
 }
 
 func PostBytes(url string, body []byte) ([]byte, error) {
-	r, err := client.Post(url, "application/octet-stream", bytes.NewReader(body))
+	r, err := client.Post(url, "", bytes.NewReader(body))
 	if err != nil {
 		return nil, fmt.Errorf("Post to %s: %v", url, err)
 	}
@@ -189,9 +187,12 @@ func NormalizeUrl(url string) string {
 	return "http://" + url
 }
 
-func ReadUrl(fileUrl string, offset int64, size int, buf []byte, isReadRange bool) (n int64, e error) {
+func ReadUrl(fileUrl string, offset int64, size int, buf []byte, isReadRange bool) (int64, error) {
 
-	req, _ := http.NewRequest("GET", fileUrl, nil)
+	req, err := http.NewRequest("GET", fileUrl, nil)
+	if err != nil {
+		return 0, err
+	}
 	if isReadRange {
 		req.Header.Add("Range", fmt.Sprintf("bytes=%d-%d", offset, offset+int64(size)))
 	} else {
@@ -217,28 +218,41 @@ func ReadUrl(fileUrl string, offset int64, size int, buf []byte, isReadRange boo
 		reader = r.Body
 	}
 
-	var i, m int
+	var (
+		i, m int
+		n    int64
+	)
 
+	// refers to https://github.com/golang/go/blob/master/src/bytes/buffer.go#L199
+	// commit id c170b14c2c1cfb2fd853a37add92a82fd6eb4318
 	for {
 		m, err = reader.Read(buf[i:])
-		if m == 0 {
-			return
-		}
 		i += m
 		n += int64(m)
 		if err == io.EOF {
 			return n, nil
 		}
-		if e != nil {
-			return n, e
+		if err != nil {
+			return n, err
+		}
+		if n == int64(len(buf)) {
+			break
 		}
 	}
-
+	// drains the response body to avoid memory leak
+	data, _ := ioutil.ReadAll(reader)
+	if len(data) != 0 {
+		err = fmt.Errorf("buffer size is too small. remains %d", len(data))
+	}
+	return n, err
 }
 
-func ReadUrlAsStream(fileUrl string, offset int64, size int, fn func(data []byte)) (n int64, e error) {
+func ReadUrlAsStream(fileUrl string, offset int64, size int, fn func(data []byte)) (int64, error) {
 
-	req, _ := http.NewRequest("GET", fileUrl, nil)
+	req, err := http.NewRequest("GET", fileUrl, nil)
+	if err != nil {
+		return 0, err
+	}
 	req.Header.Add("Range", fmt.Sprintf("bytes=%d-%d", offset, offset+int64(size)))
 
 	r, err := client.Do(req)
@@ -250,21 +264,21 @@ func ReadUrlAsStream(fileUrl string, offset int64, size int, fn func(data []byte
 		return 0, fmt.Errorf("%s: %s", fileUrl, r.Status)
 	}
 
-	var m int
+	var (
+		m int
+		n int64
+	)
 	buf := make([]byte, 64*1024)
 
 	for {
 		m, err = r.Body.Read(buf)
-		if m == 0 {
-			return
-		}
 		fn(buf[:m])
 		n += int64(m)
 		if err == io.EOF {
 			return n, nil
 		}
-		if e != nil {
-			return n, e
+		if err != nil {
+			return n, err
 		}
 	}
 
diff --git a/weed/util/httpdown/http_down.go b/weed/util/httpdown/http_down.go
new file mode 100644
index 000000000..5cbd9611c
--- /dev/null
+++ b/weed/util/httpdown/http_down.go
@@ -0,0 +1,395 @@
+// Package httpdown provides http.ConnState enabled graceful termination of
+// http.Server.
+// based on github.com/facebookarchive/httpdown, who's licence is MIT-licence,
+// we add a feature of supporting for http TLS
+package httpdown
+
+import (
+	"crypto/tls"
+	"fmt"
+	"net"
+	"net/http"
+	"os"
+	"os/signal"
+	"sync"
+	"syscall"
+	"time"
+
+	"github.com/facebookgo/clock"
+	"github.com/facebookgo/stats"
+)
+
+const (
+	defaultStopTimeout = time.Minute
+	defaultKillTimeout = time.Minute
+)
+
+// A Server allows encapsulates the process of accepting new connections and
+// serving them, and gracefully shutting down the listener without dropping
+// active connections.
+type Server interface {
+	// Wait waits for the serving loop to finish. This will happen when Stop is
+	// called, at which point it returns no error, or if there is an error in the
+	// serving loop. You must call Wait after calling Serve or ListenAndServe.
+	Wait() error
+
+	// Stop stops the listener. It will block until all connections have been
+	// closed.
+	Stop() error
+}
+
+// HTTP defines the configuration for serving a http.Server. Multiple calls to
+// Serve or ListenAndServe can be made on the same HTTP instance. The default
+// timeouts of 1 minute each result in a maximum of 2 minutes before a Stop()
+// returns.
+type HTTP struct {
+	// StopTimeout is the duration before we begin force closing connections.
+	// Defaults to 1 minute.
+	StopTimeout time.Duration
+
+	// KillTimeout is the duration before which we completely give up and abort
+	// even though we still have connected clients. This is useful when a large
+	// number of client connections exist and closing them can take a long time.
+	// Note, this is in addition to the StopTimeout. Defaults to 1 minute.
+	KillTimeout time.Duration
+
+	// Stats is optional. If provided, it will be used to record various metrics.
+	Stats stats.Client
+
+	// Clock allows for testing timing related functionality. Do not specify this
+	// in production code.
+	Clock clock.Clock
+
+	// when set CertFile and KeyFile, the httpDown will start a http with TLS.
+	// Files containing a certificate and matching private key for the
+	// server must be provided if neither the Server's
+	// TLSConfig.Certificates nor TLSConfig.GetCertificate are populated.
+	// If the certificate is signed by a certificate authority, the
+	// certFile should be the concatenation of the server's certificate,
+	// any intermediates, and the CA's certificate.
+	CertFile, KeyFile string
+}
+
+// Serve provides the low-level API which is useful if you're creating your own
+// net.Listener.
+func (h HTTP) Serve(s *http.Server, l net.Listener) Server {
+	stopTimeout := h.StopTimeout
+	if stopTimeout == 0 {
+		stopTimeout = defaultStopTimeout
+	}
+	killTimeout := h.KillTimeout
+	if killTimeout == 0 {
+		killTimeout = defaultKillTimeout
+	}
+	klock := h.Clock
+	if klock == nil {
+		klock = clock.New()
+	}
+
+	ss := &server{
+		stopTimeout:  stopTimeout,
+		killTimeout:  killTimeout,
+		stats:        h.Stats,
+		clock:        klock,
+		oldConnState: s.ConnState,
+		listener:     l,
+		server:       s,
+		serveDone:    make(chan struct{}),
+		serveErr:     make(chan error, 1),
+		new:          make(chan net.Conn),
+		active:       make(chan net.Conn),
+		idle:         make(chan net.Conn),
+		closed:       make(chan net.Conn),
+		stop:         make(chan chan struct{}),
+		kill:         make(chan chan struct{}),
+		certFile:     h.CertFile,
+		keyFile:      h.KeyFile,
+	}
+	s.ConnState = ss.connState
+	go ss.manage()
+	go ss.serve()
+	return ss
+}
+
+// ListenAndServe returns a Server for the given http.Server. It is equivalent
+// to ListenAndServe from the standard library, but returns immediately.
+// Requests will be accepted in a background goroutine. If the http.Server has
+// a non-nil TLSConfig, a TLS enabled listener will be setup.
+func (h HTTP) ListenAndServe(s *http.Server) (Server, error) {
+	addr := s.Addr
+	if addr == "" {
+		if s.TLSConfig == nil {
+			addr = ":http"
+		} else {
+			addr = ":https"
+		}
+	}
+	l, err := net.Listen("tcp", addr)
+	if err != nil {
+		stats.BumpSum(h.Stats, "listen.error", 1)
+		return nil, err
+	}
+	if s.TLSConfig != nil {
+		l = tls.NewListener(l, s.TLSConfig)
+	}
+	return h.Serve(s, l), nil
+}
+
+// server manages the serving process and allows for gracefully stopping it.
+type server struct {
+	stopTimeout time.Duration
+	killTimeout time.Duration
+	stats       stats.Client
+	clock       clock.Clock
+
+	oldConnState func(net.Conn, http.ConnState)
+	server       *http.Server
+	serveDone    chan struct{}
+	serveErr     chan error
+	listener     net.Listener
+
+	new    chan net.Conn
+	active chan net.Conn
+	idle   chan net.Conn
+	closed chan net.Conn
+	stop   chan chan struct{}
+	kill   chan chan struct{}
+
+	stopOnce sync.Once
+	stopErr  error
+
+	certFile, keyFile string
+}
+
+func (s *server) connState(c net.Conn, cs http.ConnState) {
+	if s.oldConnState != nil {
+		s.oldConnState(c, cs)
+	}
+
+	switch cs {
+	case http.StateNew:
+		s.new <- c
+	case http.StateActive:
+		s.active <- c
+	case http.StateIdle:
+		s.idle <- c
+	case http.StateHijacked, http.StateClosed:
+		s.closed <- c
+	}
+}
+
+func (s *server) manage() {
+	defer func() {
+		close(s.new)
+		close(s.active)
+		close(s.idle)
+		close(s.closed)
+		close(s.stop)
+		close(s.kill)
+	}()
+
+	var stopDone chan struct{}
+
+	conns := map[net.Conn]http.ConnState{}
+	var countNew, countActive, countIdle float64
+
+	// decConn decrements the count associated with the current state of the
+	// given connection.
+	decConn := func(c net.Conn) {
+		switch conns[c] {
+		default:
+			panic(fmt.Errorf("unknown existing connection: %s", c))
+		case http.StateNew:
+			countNew--
+		case http.StateActive:
+			countActive--
+		case http.StateIdle:
+			countIdle--
+		}
+	}
+
+	// setup a ticker to report various values every minute. if we don't have a
+	// Stats implementation provided, we Stop it so it never ticks.
+	statsTicker := s.clock.Ticker(time.Minute)
+	if s.stats == nil {
+		statsTicker.Stop()
+	}
+
+	for {
+		select {
+		case <-statsTicker.C:
+			// we'll only get here when s.stats is not nil
+			s.stats.BumpAvg("http-state.new", countNew)
+			s.stats.BumpAvg("http-state.active", countActive)
+			s.stats.BumpAvg("http-state.idle", countIdle)
+			s.stats.BumpAvg("http-state.total", countNew+countActive+countIdle)
+		case c := <-s.new:
+			conns[c] = http.StateNew
+			countNew++
+		case c := <-s.active:
+			decConn(c)
+			countActive++
+
+			conns[c] = http.StateActive
+		case c := <-s.idle:
+			decConn(c)
+			countIdle++
+
+			conns[c] = http.StateIdle
+
+			// if we're already stopping, close it
+			if stopDone != nil {
+				c.Close()
+			}
+		case c := <-s.closed:
+			stats.BumpSum(s.stats, "conn.closed", 1)
+			decConn(c)
+			delete(conns, c)
+
+			// if we're waiting to stop and are all empty, we just closed the last
+			// connection and we're done.
+			if stopDone != nil && len(conns) == 0 {
+				close(stopDone)
+				return
+			}
+		case stopDone = <-s.stop:
+			// if we're already all empty, we're already done
+			if len(conns) == 0 {
+				close(stopDone)
+				return
+			}
+
+			// close current idle connections right away
+			for c, cs := range conns {
+				if cs == http.StateIdle {
+					c.Close()
+				}
+			}
+
+			// continue the loop and wait for all the ConnState updates which will
+			// eventually close(stopDone) and return from this goroutine.
+
+		case killDone := <-s.kill:
+			// force close all connections
+			stats.BumpSum(s.stats, "kill.conn.count", float64(len(conns)))
+			for c := range conns {
+				c.Close()
+			}
+
+			// don't block the kill.
+			close(killDone)
+
+			// continue the loop and we wait for all the ConnState updates and will
+			// return from this goroutine when we're all done. otherwise we'll try to
+			// send those ConnState updates on closed channels.
+
+		}
+	}
+}
+
+func (s *server) serve() {
+	stats.BumpSum(s.stats, "serve", 1)
+	if s.certFile == "" && s.keyFile == "" {
+		s.serveErr <- s.server.Serve(s.listener)
+	} else {
+		s.serveErr <- s.server.ServeTLS(s.listener, s.certFile, s.keyFile)
+	}
+	close(s.serveDone)
+	close(s.serveErr)
+}
+
+func (s *server) Wait() error {
+	if err := <-s.serveErr; !isUseOfClosedError(err) {
+		return err
+	}
+	return nil
+}
+
+func (s *server) Stop() error {
+	s.stopOnce.Do(func() {
+		defer stats.BumpTime(s.stats, "stop.time").End()
+		stats.BumpSum(s.stats, "stop", 1)
+
+		// first disable keep-alive for new connections
+		s.server.SetKeepAlivesEnabled(false)
+
+		// then close the listener so new connections can't connect come thru
+		closeErr := s.listener.Close()
+		<-s.serveDone
+
+		// then trigger the background goroutine to stop and wait for it
+		stopDone := make(chan struct{})
+		s.stop <- stopDone
+
+		// wait for stop
+		select {
+		case <-stopDone:
+		case <-s.clock.After(s.stopTimeout):
+			defer stats.BumpTime(s.stats, "kill.time").End()
+			stats.BumpSum(s.stats, "kill", 1)
+
+			// stop timed out, wait for kill
+			killDone := make(chan struct{})
+			s.kill <- killDone
+			select {
+			case <-killDone:
+			case <-s.clock.After(s.killTimeout):
+				// kill timed out, give up
+				stats.BumpSum(s.stats, "kill.timeout", 1)
+			}
+		}
+
+		if closeErr != nil && !isUseOfClosedError(closeErr) {
+			stats.BumpSum(s.stats, "listener.close.error", 1)
+			s.stopErr = closeErr
+		}
+	})
+	return s.stopErr
+}
+
+func isUseOfClosedError(err error) bool {
+	if err == nil {
+		return false
+	}
+	if opErr, ok := err.(*net.OpError); ok {
+		err = opErr.Err
+	}
+	return err.Error() == "use of closed network connection"
+}
+
+// ListenAndServe is a convenience function to serve and wait for a SIGTERM
+// or SIGINT before shutting down.
+func ListenAndServe(s *http.Server, hd *HTTP) error {
+	if hd == nil {
+		hd = &HTTP{}
+	}
+	hs, err := hd.ListenAndServe(s)
+	if err != nil {
+		return err
+	}
+
+	waiterr := make(chan error, 1)
+	go func() {
+		defer close(waiterr)
+		waiterr <- hs.Wait()
+	}()
+
+	signals := make(chan os.Signal, 10)
+	signal.Notify(signals, syscall.SIGTERM, syscall.SIGINT)
+
+	select {
+	case err := <-waiterr:
+		if err != nil {
+			return err
+		}
+	case <-signals:
+		signal.Stop(signals)
+		if err := hs.Stop(); err != nil {
+			return err
+		}
+		if err := <-waiterr; err != nil {
+			return err
+		}
+	}
+	return nil
+}
diff --git a/weed/util/parse.go b/weed/util/parse.go
index 0a8317c19..6593d43b6 100644
--- a/weed/util/parse.go
+++ b/weed/util/parse.go
@@ -1,7 +1,9 @@
 package util
 
 import (
+	"net/url"
 	"strconv"
+	"strings"
 )
 
 func ParseInt(text string, defaultValue int) int {
@@ -24,3 +26,22 @@ func ParseUint64(text string, defaultValue uint64) uint64 {
 	}
 	return count
 }
+
+func ParseFilerUrl(entryPath string) (filerServer string, filerPort int64, path string, err error) {
+	if !strings.HasPrefix(entryPath, "http://") && !strings.HasPrefix(entryPath, "https://") {
+		entryPath = "http://" + entryPath
+	}
+
+	var u *url.URL
+	u, err = url.Parse(entryPath)
+	if err != nil {
+		return
+	}
+	filerServer = u.Hostname()
+	portString := u.Port()
+	if portString != "" {
+		filerPort, err = strconv.ParseInt(portString, 10, 32)
+	}
+	path = u.Path
+	return
+}
diff --git a/weed/util/queue.go b/weed/util/queue.go
new file mode 100644
index 000000000..1e6211e0d
--- /dev/null
+++ b/weed/util/queue.go
@@ -0,0 +1,61 @@
+package util
+
+import "sync"
+
+type node struct {
+	data interface{}
+	next *node
+}
+
+type Queue struct {
+	head  *node
+	tail  *node
+	count int
+	sync.RWMutex
+}
+
+func NewQueue() *Queue {
+	q := &Queue{}
+	return q
+}
+
+func (q *Queue) Len() int {
+	q.RLock()
+	defer q.RUnlock()
+	return q.count
+}
+
+func (q *Queue) Enqueue(item interface{}) {
+	q.Lock()
+	defer q.Unlock()
+
+	n := &node{data: item}
+
+	if q.tail == nil {
+		q.tail = n
+		q.head = n
+	} else {
+		q.tail.next = n
+		q.tail = n
+	}
+	q.count++
+}
+
+func (q *Queue) Dequeue() interface{} {
+	q.Lock()
+	defer q.Unlock()
+
+	if q.head == nil {
+		return nil
+	}
+
+	n := q.head
+	q.head = n.next
+
+	if q.head == nil {
+		q.tail = nil
+	}
+	q.count--
+
+	return n.data
+}