v1.4.2

directory/dispatcher.go

@@ -70,7 +70,6 @@ func (d *Dispatcher) Update(group map[int][]string,
 		if !write {
 			continue
 		}
-
 		// calc score
 		for _, sid = range stores {
 			totalAdd, totalAddDelay, restSpace, minScore = 0, 0, 0, 0

proxy/bfs/bfs.go

@@ -109,12 +109,16 @@ func (b *Bfs) Get(bucket, filename string) (src io.ReadCloser, ctlen int, mtime
 		}
 		td.Stop()
 		if resp.StatusCode != http.StatusOK {
+			resp.Body.Close()
 			continue
 		}
 		src = resp.Body
 		ctlen = int(resp.ContentLength)
 		break
 	}
+	if err == nil && resp.StatusCode == http.StatusServiceUnavailable {
+		err = errors.ErrStoreNotAvailable
+	}
 	return
 }
 
@@ -140,6 +144,11 @@ func (b *Bfs) Upload(bucket, filename, mine, sha1 string, buf []byte) (err error
 		err = errors.ErrInternal
 		return
 	}
+	// same sha1sum.
+	if strings.HasPrefix(filename, sha1) && res.Ret == errors.RetNeedleExist {
+		err = errors.ErrNeedleExist
+		return
+	}
 
 	params = url.Values{}
 	for _, host = range res.Stores {

proxy/http_api.go

@@ -190,6 +190,8 @@ func (s *server) download(item *ibucket.Item, bucket, file string, wr http.Respo
 	} else {
 		if err == errors.ErrNeedleNotExist {
 			status = http.StatusNotFound
+		} else if err == errors.ErrStoreNotAvailable {
+			status = http.StatusServiceUnavailable
 		} else {
 			status = http.StatusInternalServerError
 		}

proxy/proxy.toml

@@ -11,3 +11,11 @@ Domain = "http://localhost:2232/"
 Prefix = "/bfs/"
 
 MaxFileSize = 20971520
+
+AliyunKeyId = "xxxxx"
+AliyunKeySecret = "xxxxx"
+
+NetUserName = "xxxx"
+NetPasswd = "xxxx"
+
+PurgeMaxSize = 100000

store/block/supper_block.go

@@ -319,6 +319,7 @@ func (b *SuperBlock) Scan(r *os.File, offset uint32, fn func(*needle.Needle, uin
 
 // Recovery recovery needles map from super block.
 func (b *SuperBlock) Recovery(offset uint32, fn func(*needle.Needle, uint32, uint32) error) (err error) {
+	var rsize int64
 	// WARN block may be no left data, must update block offset first
 	if offset == 0 {
 		offset = needle.NeedleOffset(_headerOffset)
@@ -340,9 +341,21 @@ func (b *SuperBlock) Recovery(offset uint32, fn func(*needle.Needle, uint32, uin
 		log.Errorf("block: %s Fadvise() error(%v)", b.File)
 		return
 	}
+	rsize = needle.BlockOffset(b.Offset)
 	// reset b.w offset, discard left space which can't parse to a needle
-	if _, err = b.w.Seek(needle.BlockOffset(b.Offset), os.SEEK_SET); err != nil {
+	if _, err = b.w.Seek(rsize, os.SEEK_SET); err != nil {
 		log.Errorf("block: %s Seek() error(%v)", b.File, err)
+		return
+	}
+	// recheck offset, keep size and offset consistency
+	if b.Size != rsize {
+		log.Warningf("block: %s [real size: %d, offset: %d] but [size: %d, offset: %d] not consistency, truncate file for force recovery, this may lost data",
+			b.File, b.Size, needle.NeedleOffset(b.Size),
+			rsize, b.Offset)
+		// truncate file
+		if err = b.w.Truncate(rsize); err != nil {
+			log.Errorf("block: %s Truncate() error(%v)", b.File, err)
+		}
 	}
 	return
 }

store/http.go

@@ -8,6 +8,7 @@ import (
 	log "github.com/golang/glog"
 	"golang.org/x/time/rate"
 	"mime/multipart"
+	"net"
 	"net/http"
 	"os"
 	"strconv"
@@ -18,21 +19,56 @@ type Server struct {
 	store *Store
 	conf  *conf.Config
 	info  *stat.Info
-
+	// server
+	statSvr  net.Listener
+	adminSvr net.Listener
+	apiSvr   net.Listener
+	// limit
 	rl *rate.Limiter
 	wl *rate.Limiter
 	dl *rate.Limiter
 }
 
-func NewServer(s *Store, c *conf.Config) *Server {
-	svr := &Server{
+func NewServer(s *Store, c *conf.Config) (svr *Server, err error) {
+	svr = &Server{
 		store: s,
 		conf:  c,
 		rl:    rate.NewLimiter(rate.Limit(c.Limit.Read.Rate), c.Limit.Read.Brust),
 		wl:    rate.NewLimiter(rate.Limit(c.Limit.Write.Rate), c.Limit.Write.Brust),
 		dl:    rate.NewLimiter(rate.Limit(c.Limit.Delete.Rate), c.Limit.Delete.Brust),
 	}
-	return svr
+	if svr.statSvr, err = net.Listen("tcp", c.StatListen); err != nil {
+		log.Errorf("net.Listen(%s) error(%v)", c.StatListen, err)
+		return
+	}
+	if svr.apiSvr, err = net.Listen("tcp", c.ApiListen); err != nil {
+		log.Errorf("net.Listen(%s) error(%v)", c.ApiListen, err)
+		return
+	}
+	if svr.adminSvr, err = net.Listen("tcp", c.AdminListen); err != nil {
+		log.Errorf("net.Listen(%s) error(%v)", c.AdminListen, err)
+		return
+	}
+	go svr.startStat()
+	go svr.startApi()
+	go svr.startAdmin()
+	if c.Pprof {
+		go StartPprof(c.PprofListen)
+	}
+	return
+}
+
+func (s *Server) Close() {
+	if s.statSvr != nil {
+		s.statSvr.Close()
+	}
+	if s.adminSvr != nil {
+		s.adminSvr.Close()
+	}
+	if s.apiSvr != nil {
+		s.apiSvr.Close()
+	}
+	return
 }
 
 type sizer interface {

store/http_admin.go

@@ -9,23 +9,26 @@ import (
 	"time"
 )
 
-// StartAdmin start admin http listen.
-func StartAdmin(addr string, s *Server) {
-	go func() {
-		var (
-			err      error
-			serveMux = http.NewServeMux()
-		)
-		serveMux.HandleFunc("/probe", s.probe)
-		serveMux.HandleFunc("/bulk_volume", s.bulkVolume)
-		serveMux.HandleFunc("/compact_volume", s.compactVolume)
-		serveMux.HandleFunc("/add_volume", s.addVolume)
-		serveMux.HandleFunc("/add_free_volume", s.addFreeVolume)
-		if err = http.ListenAndServe(addr, serveMux); err != nil {
-			log.Errorf("http.ListenAndServe(\"%s\") error(%v)", addr, err)
-			return
+// startAdmin start admin http listen.
+func (s *Server) startAdmin() {
+	var (
+		err      error
+		serveMux = http.NewServeMux()
+		server   = &http.Server{
+			Addr:    s.conf.AdminListen,
+			Handler: serveMux,
+			// TODO read/write timeout
 		}
-	}()
+	)
+	serveMux.HandleFunc("/probe", s.probe)
+	serveMux.HandleFunc("/bulk_volume", s.bulkVolume)
+	serveMux.HandleFunc("/compact_volume", s.compactVolume)
+	serveMux.HandleFunc("/add_volume", s.addVolume)
+	serveMux.HandleFunc("/add_free_volume", s.addFreeVolume)
+	if err = server.Serve(s.adminSvr); err != nil {
+		log.Errorf("server.Serve() error(%v)", err)
+	}
+	log.Info("http admin stop")
 	return
 }
 

store/http_api.go

@@ -11,22 +11,25 @@ import (
 	"time"
 )
 
-// StartApi start api http listen.
-func StartApi(addr string, s *Server) {
-	go func() {
-		var (
-			err      error
-			serveMux = http.NewServeMux()
-		)
-		serveMux.HandleFunc("/get", s.get)
-		serveMux.HandleFunc("/upload", s.upload)
-		serveMux.HandleFunc("/uploads", s.uploads)
-		serveMux.HandleFunc("/del", s.del)
-		if err = http.ListenAndServe(addr, serveMux); err != nil {
-			log.Errorf("http.ListenAndServe(\"%s\") error(%v)", addr, err)
-			return
+// startApi start api http listen.
+func (s *Server) startApi() {
+	var (
+		err      error
+		serveMux = http.NewServeMux()
+		server   = &http.Server{
+			Addr:    s.conf.ApiListen,
+			Handler: serveMux,
+			// TODO read/write timeout
 		}
-	}()
+	)
+	serveMux.HandleFunc("/get", s.get)
+	serveMux.HandleFunc("/upload", s.upload)
+	serveMux.HandleFunc("/uploads", s.uploads)
+	serveMux.HandleFunc("/del", s.del)
+	if err = server.Serve(s.apiSvr); err != nil {
+		log.Errorf("server.Serve() error(%v)", err)
+	}
+	log.Info("http api stop")
 	return
 }