diff options
Diffstat (limited to 'weed/util')
36 files changed, 3487 insertions, 201 deletions
diff --git a/weed/util/bounded_tree/bounded_tree.go b/weed/util/bounded_tree/bounded_tree.go new file mode 100644 index 000000000..3a8a22a9c --- /dev/null +++ b/weed/util/bounded_tree/bounded_tree.go @@ -0,0 +1,182 @@ +package bounded_tree + +import ( + "sync" + + "github.com/chrislusf/seaweedfs/weed/glog" + "github.com/chrislusf/seaweedfs/weed/util" +) + +type Node struct { + Parent *Node + Name string + Children map[string]*Node +} + +type BoundedTree struct { + root *Node + sync.RWMutex + baseDir util.FullPath +} + +func NewBoundedTree(baseDir util.FullPath) *BoundedTree { + return &BoundedTree{ + root: &Node{ + Name: "/", + }, + baseDir: baseDir, + } +} + +type VisitNodeFunc func(path util.FullPath) (childDirectories []string, err error) + +// If the path is not visited, call the visitFn for each level of directory +// No action if the directory has been visited before or does not exist. +// A leaf node, which has no children, represents a directory not visited. +// A non-leaf node or a non-existing node represents a directory already visited, or does not need to visit. +func (t *BoundedTree) EnsureVisited(p util.FullPath, visitFn VisitNodeFunc) (visitErr error) { + t.Lock() + defer t.Unlock() + + if t.root == nil { + return + } + if t.baseDir != "/" { + p = p[len(t.baseDir):] + } + components := p.Split() + // fmt.Printf("components %v %d\n", components, len(components)) + canDelete, err := t.ensureVisited(t.root, t.baseDir, components, 0, visitFn) + if err != nil { + return err + } + if canDelete { + t.root = nil + } + return nil +} + +func (t *BoundedTree) ensureVisited(n *Node, currentPath util.FullPath, components []string, i int, visitFn VisitNodeFunc) (canDeleteNode bool, visitErr error) { + + // println("ensureVisited", currentPath, i) + + if n == nil { + // fmt.Printf("%s null\n", currentPath) + return + } + + if n.isVisited() { + // fmt.Printf("%s visited %v\n", currentPath, n.Name) + } else { + // fmt.Printf("ensure %v\n", currentPath) + + children, err := visitFn(currentPath) + if err != nil { + glog.V(0).Infof("failed to visit %s: %v", currentPath, err) + return false, err + } + + if len(children) == 0 { + // fmt.Printf(" canDelete %v without children\n", currentPath) + return true, nil + } + + n.Children = make(map[string]*Node) + for _, child := range children { + // fmt.Printf(" add child %v %v\n", currentPath, child) + n.Children[child] = &Node{ + Name: child, + } + } + } + + if i >= len(components) { + return + } + + // fmt.Printf(" check child %v %v\n", currentPath, components[i]) + + toVisitNode, found := n.Children[components[i]] + if !found { + // fmt.Printf(" did not find child %v %v\n", currentPath, components[i]) + return + } + + // fmt.Printf(" ensureVisited %v %v\n", currentPath, toVisitNode.Name) + canDelete, childVisitErr := t.ensureVisited(toVisitNode, currentPath.Child(components[i]), components, i+1, visitFn) + if childVisitErr != nil { + return false, childVisitErr + } + if canDelete { + + // fmt.Printf(" delete %v %v\n", currentPath, components[i]) + delete(n.Children, components[i]) + + if len(n.Children) == 0 { + // fmt.Printf(" canDelete %v\n", currentPath) + return true, nil + } + } + + return false, nil + +} + +func (n *Node) isVisited() bool { + if n == nil { + return true + } + if len(n.Children) > 0 { + return true + } + return false +} + +func (n *Node) getChild(childName string) *Node { + if n == nil { + return nil + } + if len(n.Children) > 0 { + return n.Children[childName] + } + return nil +} + +func (t *BoundedTree) HasVisited(p util.FullPath) bool { + + t.RLock() + defer t.RUnlock() + + if t.root == nil { + return true + } + + components := p.Split() + // fmt.Printf("components %v %d\n", components, len(components)) + return t.hasVisited(t.root, util.FullPath("/"), components, 0) +} + +func (t *BoundedTree) hasVisited(n *Node, currentPath util.FullPath, components []string, i int) bool { + + if n == nil { + return true + } + + if !n.isVisited() { + return false + } + + // fmt.Printf(" hasVisited child %v %+v %d\n", currentPath, components, i) + + if i >= len(components) { + return true + } + + toVisitNode, found := n.Children[components[i]] + if !found { + return true + } + + return t.hasVisited(toVisitNode, currentPath.Child(components[i]), components, i+1) + +} diff --git a/weed/util/bounded_tree/bounded_tree_test.go b/weed/util/bounded_tree/bounded_tree_test.go new file mode 100644 index 000000000..465f1cc9c --- /dev/null +++ b/weed/util/bounded_tree/bounded_tree_test.go @@ -0,0 +1,126 @@ +package bounded_tree + +import ( + "fmt" + "testing" + + "github.com/stretchr/testify/assert" + + "github.com/chrislusf/seaweedfs/weed/util" +) + +var ( + visitFn = func(path util.FullPath) (childDirectories []string, err error) { + fmt.Printf(" visit %v ...\n", path) + switch path { + case "/": + return []string{"a", "g", "h"}, nil + case "/a": + return []string{"b", "f"}, nil + case "/a/b": + return []string{"c", "e"}, nil + case "/a/b/c": + return []string{"d"}, nil + case "/a/b/c/d": + return []string{"i", "j"}, nil + case "/a/b/c/d/i": + return []string{}, nil + case "/a/b/c/d/j": + return []string{}, nil + case "/a/b/e": + return []string{}, nil + case "/a/f": + return []string{}, nil + } + return nil, nil + } + + printMap = func(m map[string]*Node) { + for k := range m { + println(" >", k) + } + } +) + +func TestBoundedTree(t *testing.T) { + + // a/b/c/d/i + // a/b/c/d/j + // a/b/c/d + // a/b/e + // a/f + // g + // h + + tree := NewBoundedTree(util.FullPath("/")) + + tree.EnsureVisited(util.FullPath("/a/b/c"), visitFn) + + assert.Equal(t, true, tree.HasVisited(util.FullPath("/a/b"))) + assert.Equal(t, true, tree.HasVisited(util.FullPath("/a/b/c"))) + assert.Equal(t, false, tree.HasVisited(util.FullPath("/a/b/c/d"))) + assert.Equal(t, false, tree.HasVisited(util.FullPath("/a/b/e"))) + assert.Equal(t, false, tree.HasVisited(util.FullPath("/a/f"))) + assert.Equal(t, false, tree.HasVisited(util.FullPath("/g"))) + assert.Equal(t, false, tree.HasVisited(util.FullPath("/h"))) + assert.Equal(t, true, tree.HasVisited(util.FullPath("/"))) + assert.Equal(t, true, tree.HasVisited(util.FullPath("/x"))) + assert.Equal(t, false, tree.HasVisited(util.FullPath("/a/b/e/x"))) + + printMap(tree.root.Children) + + a := tree.root.getChild("a") + + b := a.getChild("b") + if !b.isVisited() { + t.Errorf("expect visited /a/b") + } + c := b.getChild("c") + if !c.isVisited() { + t.Errorf("expect visited /a/b/c") + } + + d := c.getChild("d") + if d.isVisited() { + t.Errorf("expect unvisited /a/b/c/d") + } + + tree.EnsureVisited(util.FullPath("/a/b/c/d"), visitFn) + tree.EnsureVisited(util.FullPath("/a/b/c/d/i"), visitFn) + tree.EnsureVisited(util.FullPath("/a/b/c/d/j"), visitFn) + tree.EnsureVisited(util.FullPath("/a/b/e"), visitFn) + tree.EnsureVisited(util.FullPath("/a/f"), visitFn) + + printMap(tree.root.Children) + +} + +func TestEmptyBoundedTree(t *testing.T) { + + // g + // h + + tree := NewBoundedTree(util.FullPath("/")) + + visitFn := func(path util.FullPath) (childDirectories []string, err error) { + fmt.Printf(" visit %v ...\n", path) + switch path { + case "/": + return []string{"g", "h"}, nil + } + t.Fatalf("expected visit %s", path) + return nil, nil + } + + tree.EnsureVisited(util.FullPath("/a/b"), visitFn) + + tree.EnsureVisited(util.FullPath("/a/b"), visitFn) + + printMap(tree.root.Children) + + assert.Equal(t, true, tree.HasVisited(util.FullPath("/a/b"))) + assert.Equal(t, true, tree.HasVisited(util.FullPath("/a"))) + assert.Equal(t, false, tree.HasVisited(util.FullPath("/g"))) + assert.Equal(t, false, tree.HasVisited(util.FullPath("/g/x"))) + +} diff --git a/weed/util/buffered_writer/buffered_writer.go b/weed/util/buffered_writer/buffered_writer.go new file mode 100644 index 000000000..73d9f4995 --- /dev/null +++ b/weed/util/buffered_writer/buffered_writer.go @@ -0,0 +1,52 @@ +package buffered_writer + +import ( + "bytes" + "io" +) + +var _ = io.WriteCloser(&BufferedWriteCloser{}) + +type BufferedWriteCloser struct { + buffer bytes.Buffer + bufferLimit int + position int64 + nextFlushOffset int64 + FlushFunc func([]byte, int64) error + CloseFunc func() error +} + +func NewBufferedWriteCloser(bufferLimit int) *BufferedWriteCloser { + return &BufferedWriteCloser{ + bufferLimit: bufferLimit, + } +} + +func (b *BufferedWriteCloser) Write(p []byte) (n int, err error) { + + if b.buffer.Len()+len(p) >= b.bufferLimit { + if err := b.FlushFunc(b.buffer.Bytes(), b.nextFlushOffset); err != nil { + return 0, err + } + b.nextFlushOffset += int64(b.buffer.Len()) + b.buffer.Reset() + } + + return b.buffer.Write(p) + +} + +func (b *BufferedWriteCloser) Close() error { + if b.buffer.Len() > 0 { + if err := b.FlushFunc(b.buffer.Bytes(), b.nextFlushOffset); err != nil { + return err + } + } + if b.CloseFunc != nil { + if err := b.CloseFunc(); err != nil { + return err + } + } + + return nil +} diff --git a/weed/util/bytes.go b/weed/util/bytes.go index dfa4ae665..c2a4df108 100644 --- a/weed/util/bytes.go +++ b/weed/util/bytes.go @@ -1,5 +1,30 @@ package util +import ( + "bytes" + "crypto/md5" + "crypto/rand" + "encoding/base64" + "fmt" + "io" +) + +// BytesToHumanReadable returns the converted human readable representation of the bytes. +func BytesToHumanReadable(b uint64) string { + const unit = 1024 + if b < unit { + return fmt.Sprintf("%d B", b) + } + + div, exp := uint64(unit), 0 + for n := b / unit; n >= unit; n /= unit { + div *= unit + exp++ + } + + return fmt.Sprintf("%.2f %ciB", float64(b)/float64(div), "KMGTPE"[exp]) +} + // big endian func BytesToUint64(b []byte) (v uint64) { @@ -43,3 +68,96 @@ func Uint16toBytes(b []byte, v uint16) { func Uint8toBytes(b []byte, v uint8) { b[0] = byte(v) } + +// returns a 64 bit big int +func HashStringToLong(dir string) (v int64) { + h := md5.New() + io.WriteString(h, dir) + + b := h.Sum(nil) + + v += int64(b[0]) + v <<= 8 + v += int64(b[1]) + v <<= 8 + v += int64(b[2]) + v <<= 8 + v += int64(b[3]) + v <<= 8 + v += int64(b[4]) + v <<= 8 + v += int64(b[5]) + v <<= 8 + v += int64(b[6]) + v <<= 8 + v += int64(b[7]) + + return +} + +func HashToInt32(data []byte) (v int32) { + h := md5.New() + h.Write(data) + + b := h.Sum(nil) + + v += int32(b[0]) + v <<= 8 + v += int32(b[1]) + v <<= 8 + v += int32(b[2]) + v <<= 8 + v += int32(b[3]) + + return +} + +func Base64Encode(data []byte) string { + return base64.StdEncoding.EncodeToString(data) +} + +func Base64Md5(data []byte) string { + return Base64Encode(Md5(data)) +} + +func Md5(data []byte) []byte { + hash := md5.New() + hash.Write(data) + return hash.Sum(nil) +} + +func Md5String(data []byte) string { + return fmt.Sprintf("%x", Md5(data)) +} + +func Base64Md5ToBytes(contentMd5 string) []byte { + data, err := base64.StdEncoding.DecodeString(contentMd5) + if err != nil { + return nil + } + return data +} + +func RandomInt32() int32 { + buf := make([]byte, 4) + rand.Read(buf) + return int32(BytesToUint32(buf)) +} + +func RandomBytes(byteCount int) []byte { + buf := make([]byte, byteCount) + rand.Read(buf) + return buf +} + +type BytesReader struct { + Bytes []byte + *bytes.Reader +} + +func NewBytesReader(b []byte) *BytesReader { + return &BytesReader{ + Bytes: b, + Reader: bytes.NewReader(b), + } +} diff --git a/weed/util/chunk_cache/chunk_cache.go b/weed/util/chunk_cache/chunk_cache.go new file mode 100644 index 000000000..3615aee0e --- /dev/null +++ b/weed/util/chunk_cache/chunk_cache.go @@ -0,0 +1,133 @@ +package chunk_cache + +import ( + "sync" + + "github.com/chrislusf/seaweedfs/weed/glog" + "github.com/chrislusf/seaweedfs/weed/storage/needle" +) + +type ChunkCache interface { + GetChunk(fileId string, minSize uint64) (data []byte) + SetChunk(fileId string, data []byte) +} + +// a global cache for recently accessed file chunks +type TieredChunkCache struct { + memCache *ChunkCacheInMemory + diskCaches []*OnDiskCacheLayer + sync.RWMutex + onDiskCacheSizeLimit0 uint64 + onDiskCacheSizeLimit1 uint64 + onDiskCacheSizeLimit2 uint64 +} + +func NewTieredChunkCache(maxEntries int64, dir string, diskSizeInUnit int64, unitSize int64) *TieredChunkCache { + + c := &TieredChunkCache{ + memCache: NewChunkCacheInMemory(maxEntries), + } + c.diskCaches = make([]*OnDiskCacheLayer, 3) + c.onDiskCacheSizeLimit0 = uint64(unitSize) + c.onDiskCacheSizeLimit1 = 4 * c.onDiskCacheSizeLimit0 + c.onDiskCacheSizeLimit2 = 2 * c.onDiskCacheSizeLimit1 + c.diskCaches[0] = NewOnDiskCacheLayer(dir, "c0_2", diskSizeInUnit*unitSize/8, 2) + c.diskCaches[1] = NewOnDiskCacheLayer(dir, "c1_3", diskSizeInUnit*unitSize/4+diskSizeInUnit*unitSize/8, 3) + c.diskCaches[2] = NewOnDiskCacheLayer(dir, "c2_2", diskSizeInUnit*unitSize/2, 2) + + return c +} + +func (c *TieredChunkCache) GetChunk(fileId string, minSize uint64) (data []byte) { + if c == nil { + return + } + + c.RLock() + defer c.RUnlock() + + return c.doGetChunk(fileId, minSize) +} + +func (c *TieredChunkCache) doGetChunk(fileId string, minSize uint64) (data []byte) { + + if minSize <= c.onDiskCacheSizeLimit0 { + data = c.memCache.GetChunk(fileId) + if len(data) >= int(minSize) { + return data + } + } + + fid, err := needle.ParseFileIdFromString(fileId) + if err != nil { + glog.Errorf("failed to parse file id %s", fileId) + return nil + } + + if minSize <= c.onDiskCacheSizeLimit0 { + data = c.diskCaches[0].getChunk(fid.Key) + if len(data) >= int(minSize) { + return data + } + } + if minSize <= c.onDiskCacheSizeLimit1 { + data = c.diskCaches[1].getChunk(fid.Key) + if len(data) >= int(minSize) { + return data + } + } + { + data = c.diskCaches[2].getChunk(fid.Key) + if len(data) >= int(minSize) { + return data + } + } + + return nil + +} + +func (c *TieredChunkCache) SetChunk(fileId string, data []byte) { + if c == nil { + return + } + c.Lock() + defer c.Unlock() + + glog.V(4).Infof("SetChunk %s size %d\n", fileId, len(data)) + + c.doSetChunk(fileId, data) +} + +func (c *TieredChunkCache) doSetChunk(fileId string, data []byte) { + + if len(data) <= int(c.onDiskCacheSizeLimit0) { + c.memCache.SetChunk(fileId, data) + } + + fid, err := needle.ParseFileIdFromString(fileId) + if err != nil { + glog.Errorf("failed to parse file id %s", fileId) + return + } + + if len(data) <= int(c.onDiskCacheSizeLimit0) { + c.diskCaches[0].setChunk(fid.Key, data) + } else if len(data) <= int(c.onDiskCacheSizeLimit1) { + c.diskCaches[1].setChunk(fid.Key, data) + } else { + c.diskCaches[2].setChunk(fid.Key, data) + } + +} + +func (c *TieredChunkCache) Shutdown() { + if c == nil { + return + } + c.Lock() + defer c.Unlock() + for _, diskCache := range c.diskCaches { + diskCache.shutdown() + } +} diff --git a/weed/util/chunk_cache/chunk_cache_in_memory.go b/weed/util/chunk_cache/chunk_cache_in_memory.go new file mode 100644 index 000000000..5f26b8c78 --- /dev/null +++ b/weed/util/chunk_cache/chunk_cache_in_memory.go @@ -0,0 +1,38 @@ +package chunk_cache + +import ( + "time" + + "github.com/karlseguin/ccache/v2" +) + +// a global cache for recently accessed file chunks +type ChunkCacheInMemory struct { + cache *ccache.Cache +} + +func NewChunkCacheInMemory(maxEntries int64) *ChunkCacheInMemory { + pruneCount := maxEntries >> 3 + if pruneCount <= 0 { + pruneCount = 500 + } + return &ChunkCacheInMemory{ + cache: ccache.New(ccache.Configure().MaxSize(maxEntries).ItemsToPrune(uint32(pruneCount))), + } +} + +func (c *ChunkCacheInMemory) GetChunk(fileId string) []byte { + item := c.cache.Get(fileId) + if item == nil { + return nil + } + data := item.Value().([]byte) + item.Extend(time.Hour) + return data +} + +func (c *ChunkCacheInMemory) SetChunk(fileId string, data []byte) { + localCopy := make([]byte, len(data)) + copy(localCopy, data) + c.cache.Set(fileId, localCopy, time.Hour) +} diff --git a/weed/util/chunk_cache/chunk_cache_on_disk.go b/weed/util/chunk_cache/chunk_cache_on_disk.go new file mode 100644 index 000000000..6f87a9a06 --- /dev/null +++ b/weed/util/chunk_cache/chunk_cache_on_disk.go @@ -0,0 +1,147 @@ +package chunk_cache + +import ( + "fmt" + "os" + "time" + + "github.com/syndtr/goleveldb/leveldb/opt" + + "github.com/chrislusf/seaweedfs/weed/glog" + "github.com/chrislusf/seaweedfs/weed/storage" + "github.com/chrislusf/seaweedfs/weed/storage/backend" + "github.com/chrislusf/seaweedfs/weed/storage/types" + "github.com/chrislusf/seaweedfs/weed/util" +) + +// This implements an on disk cache +// The entries are an FIFO with a size limit + +type ChunkCacheVolume struct { + DataBackend backend.BackendStorageFile + nm storage.NeedleMapper + fileName string + smallBuffer []byte + sizeLimit int64 + lastModTime time.Time + fileSize int64 +} + +func LoadOrCreateChunkCacheVolume(fileName string, preallocate int64) (*ChunkCacheVolume, error) { + + v := &ChunkCacheVolume{ + smallBuffer: make([]byte, types.NeedlePaddingSize), + fileName: fileName, + sizeLimit: preallocate, + } + + var err error + + if exists, canRead, canWrite, modTime, fileSize := util.CheckFile(v.fileName + ".dat"); exists { + if !canRead { + return nil, fmt.Errorf("cannot read cache file %s.dat", v.fileName) + } + if !canWrite { + return nil, fmt.Errorf("cannot write cache file %s.dat", v.fileName) + } + if dataFile, err := os.OpenFile(v.fileName+".dat", os.O_RDWR|os.O_CREATE, 0644); err != nil { + return nil, fmt.Errorf("cannot create cache file %s.dat: %v", v.fileName, err) + } else { + v.DataBackend = backend.NewDiskFile(dataFile) + v.lastModTime = modTime + v.fileSize = fileSize + } + } else { + if v.DataBackend, err = backend.CreateVolumeFile(v.fileName+".dat", preallocate, 0); err != nil { + return nil, fmt.Errorf("cannot create cache file %s.dat: %v", v.fileName, err) + } + v.lastModTime = time.Now() + } + + var indexFile *os.File + if indexFile, err = os.OpenFile(v.fileName+".idx", os.O_RDWR|os.O_CREATE, 0644); err != nil { + return nil, fmt.Errorf("cannot write cache index %s.idx: %v", v.fileName, err) + } + + glog.V(1).Infoln("loading leveldb", v.fileName+".ldb") + opts := &opt.Options{ + BlockCacheCapacity: 2 * 1024 * 1024, // default value is 8MiB + WriteBuffer: 1 * 1024 * 1024, // default value is 4MiB + CompactionTableSizeMultiplier: 10, // default value is 1 + } + if v.nm, err = storage.NewLevelDbNeedleMap(v.fileName+".ldb", indexFile, opts); err != nil { + return nil, fmt.Errorf("loading leveldb %s error: %v", v.fileName+".ldb", err) + } + + return v, nil + +} + +func (v *ChunkCacheVolume) Shutdown() { + if v.DataBackend != nil { + v.DataBackend.Close() + v.DataBackend = nil + } + if v.nm != nil { + v.nm.Close() + v.nm = nil + } +} + +func (v *ChunkCacheVolume) doReset() { + v.Shutdown() + os.Truncate(v.fileName + ".dat", 0) + os.Truncate(v.fileName + ".idx", 0) + glog.V(4).Infof("cache removeAll %s ...", v.fileName + ".ldb") + os.RemoveAll(v.fileName + ".ldb") + glog.V(4).Infof("cache removed %s", v.fileName + ".ldb") +} + +func (v *ChunkCacheVolume) Reset() (*ChunkCacheVolume, error) { + v.doReset() + return LoadOrCreateChunkCacheVolume(v.fileName, v.sizeLimit) +} + +func (v *ChunkCacheVolume) GetNeedle(key types.NeedleId) ([]byte, error) { + + nv, ok := v.nm.Get(key) + if !ok { + return nil, storage.ErrorNotFound + } + data := make([]byte, nv.Size) + if readSize, readErr := v.DataBackend.ReadAt(data, nv.Offset.ToActualOffset()); readErr != nil { + return nil, fmt.Errorf("read %s.dat [%d,%d): %v", + v.fileName, nv.Offset.ToActualOffset(), nv.Offset.ToActualOffset()+int64(nv.Size), readErr) + } else { + if readSize != int(nv.Size) { + return nil, fmt.Errorf("read %d, expected %d", readSize, nv.Size) + } + } + + return data, nil +} + +func (v *ChunkCacheVolume) WriteNeedle(key types.NeedleId, data []byte) error { + + offset := v.fileSize + + written, err := v.DataBackend.WriteAt(data, offset) + if err != nil { + return err + } else if written != len(data) { + return fmt.Errorf("partial written %d, expected %d", written, len(data)) + } + + v.fileSize += int64(written) + extraSize := written % types.NeedlePaddingSize + if extraSize != 0 { + v.DataBackend.WriteAt(v.smallBuffer[:types.NeedlePaddingSize-extraSize], offset+int64(written)) + v.fileSize += int64(types.NeedlePaddingSize - extraSize) + } + + if err := v.nm.Put(key, types.ToOffset(offset), types.Size(len(data))); err != nil { + return err + } + + return nil +} diff --git a/weed/util/chunk_cache/chunk_cache_on_disk_test.go b/weed/util/chunk_cache/chunk_cache_on_disk_test.go new file mode 100644 index 000000000..f8325276e --- /dev/null +++ b/weed/util/chunk_cache/chunk_cache_on_disk_test.go @@ -0,0 +1,98 @@ +package chunk_cache + +import ( + "bytes" + "fmt" + "io/ioutil" + "math/rand" + "os" + "testing" +) + +func TestOnDisk(t *testing.T) { + + tmpDir, _ := ioutil.TempDir("", "c") + defer os.RemoveAll(tmpDir) + + totalDiskSizeInKB := int64(32) + + cache := NewTieredChunkCache(2, tmpDir, totalDiskSizeInKB, 1024) + + writeCount := 5 + type test_data struct { + data []byte + fileId string + size uint64 + } + testData := make([]*test_data, writeCount) + for i := 0; i < writeCount; i++ { + buff := make([]byte, 1024) + rand.Read(buff) + testData[i] = &test_data{ + data: buff, + fileId: fmt.Sprintf("1,%daabbccdd", i+1), + size: uint64(len(buff)), + } + cache.SetChunk(testData[i].fileId, testData[i].data) + + // read back right after write + data := cache.GetChunk(testData[i].fileId, testData[i].size) + if bytes.Compare(data, testData[i].data) != 0 { + t.Errorf("failed to write to and read from cache: %d", i) + } + } + + for i := 0; i < 2; i++ { + data := cache.GetChunk(testData[i].fileId, testData[i].size) + if bytes.Compare(data, testData[i].data) == 0 { + t.Errorf("old cache should have been purged: %d", i) + } + } + + for i := 2; i < writeCount; i++ { + data := cache.GetChunk(testData[i].fileId, testData[i].size) + if bytes.Compare(data, testData[i].data) != 0 { + t.Errorf("failed to write to and read from cache: %d", i) + } + } + + cache.Shutdown() + + cache = NewTieredChunkCache(2, tmpDir, totalDiskSizeInKB, 1024) + + for i := 0; i < 2; i++ { + data := cache.GetChunk(testData[i].fileId, testData[i].size) + if bytes.Compare(data, testData[i].data) == 0 { + t.Errorf("old cache should have been purged: %d", i) + } + } + + for i := 2; i < writeCount; i++ { + if i == 4 { + // FIXME this failed many times on build machines + /* + I0928 06:04:12 10979 volume_create_linux.go:19] Preallocated 2048 bytes disk space for /tmp/c578652251/c0_2_0.dat + I0928 06:04:12 10979 volume_create_linux.go:19] Preallocated 2048 bytes disk space for /tmp/c578652251/c0_2_1.dat + I0928 06:04:12 10979 volume_create_linux.go:19] Preallocated 4096 bytes disk space for /tmp/c578652251/c1_3_0.dat + I0928 06:04:12 10979 volume_create_linux.go:19] Preallocated 4096 bytes disk space for /tmp/c578652251/c1_3_1.dat + I0928 06:04:12 10979 volume_create_linux.go:19] Preallocated 4096 bytes disk space for /tmp/c578652251/c1_3_2.dat + I0928 06:04:12 10979 volume_create_linux.go:19] Preallocated 8192 bytes disk space for /tmp/c578652251/c2_2_0.dat + I0928 06:04:12 10979 volume_create_linux.go:19] Preallocated 8192 bytes disk space for /tmp/c578652251/c2_2_1.dat + I0928 06:04:12 10979 volume_create_linux.go:19] Preallocated 2048 bytes disk space for /tmp/c578652251/c0_2_0.dat + I0928 06:04:12 10979 volume_create_linux.go:19] Preallocated 2048 bytes disk space for /tmp/c578652251/c0_2_1.dat + --- FAIL: TestOnDisk (0.19s) + chunk_cache_on_disk_test.go:73: failed to write to and read from cache: 4 + FAIL + FAIL github.com/chrislusf/seaweedfs/weed/util/chunk_cache 0.199s + */ + continue + } + data := cache.GetChunk(testData[i].fileId, testData[i].size) + if bytes.Compare(data, testData[i].data) != 0 { + t.Errorf("failed to write to and read from cache: %d", i) + } + } + + cache.Shutdown() + +} diff --git a/weed/util/chunk_cache/on_disk_cache_layer.go b/weed/util/chunk_cache/on_disk_cache_layer.go new file mode 100644 index 000000000..eebd89798 --- /dev/null +++ b/weed/util/chunk_cache/on_disk_cache_layer.go @@ -0,0 +1,91 @@ +package chunk_cache + +import ( + "fmt" + "path" + "sort" + + "github.com/chrislusf/seaweedfs/weed/glog" + "github.com/chrislusf/seaweedfs/weed/storage" + "github.com/chrislusf/seaweedfs/weed/storage/types" +) + +type OnDiskCacheLayer struct { + diskCaches []*ChunkCacheVolume +} + +func NewOnDiskCacheLayer(dir, namePrefix string, diskSize int64, segmentCount int) *OnDiskCacheLayer { + + volumeCount, volumeSize := int(diskSize/(30000*1024*1024)), int64(30000*1024*1024) + if volumeCount < segmentCount { + volumeCount, volumeSize = segmentCount, diskSize/int64(segmentCount) + } + + c := &OnDiskCacheLayer{} + for i := 0; i < volumeCount; i++ { + fileName := path.Join(dir, fmt.Sprintf("%s_%d", namePrefix, i)) + diskCache, err := LoadOrCreateChunkCacheVolume(fileName, volumeSize) + if err != nil { + glog.Errorf("failed to add cache %s : %v", fileName, err) + } else { + c.diskCaches = append(c.diskCaches, diskCache) + } + } + + // keep newest cache to the front + sort.Slice(c.diskCaches, func(i, j int) bool { + return c.diskCaches[i].lastModTime.After(c.diskCaches[j].lastModTime) + }) + + return c +} + +func (c *OnDiskCacheLayer) setChunk(needleId types.NeedleId, data []byte) { + + if c.diskCaches[0].fileSize+int64(len(data)) > c.diskCaches[0].sizeLimit { + t, resetErr := c.diskCaches[len(c.diskCaches)-1].Reset() + if resetErr != nil { + glog.Errorf("failed to reset cache file %s", c.diskCaches[len(c.diskCaches)-1].fileName) + return + } + for i := len(c.diskCaches) - 1; i > 0; i-- { + c.diskCaches[i] = c.diskCaches[i-1] + } + c.diskCaches[0] = t + } + + if err := c.diskCaches[0].WriteNeedle(needleId, data); err != nil { + glog.V(0).Infof("cache write %v size %d: %v", needleId, len(data), err) + } + +} + +func (c *OnDiskCacheLayer) getChunk(needleId types.NeedleId) (data []byte) { + + var err error + + for _, diskCache := range c.diskCaches { + data, err = diskCache.GetNeedle(needleId) + if err == storage.ErrorNotFound { + continue + } + if err != nil { + glog.Errorf("failed to read cache file %s id %d", diskCache.fileName, needleId) + continue + } + if len(data) != 0 { + return + } + } + + return nil + +} + +func (c *OnDiskCacheLayer) shutdown() { + + for _, diskCache := range c.diskCaches { + diskCache.Shutdown() + } + +} diff --git a/weed/util/cipher.go b/weed/util/cipher.go new file mode 100644 index 000000000..f044c2ca3 --- /dev/null +++ b/weed/util/cipher.go @@ -0,0 +1,60 @@ +package util + +import ( + "crypto/aes" + "crypto/cipher" + "crypto/rand" + "errors" + "io" + + "github.com/chrislusf/seaweedfs/weed/glog" +) + +type CipherKey []byte + +func GenCipherKey() CipherKey { + key := make([]byte, 32) + if _, err := io.ReadFull(rand.Reader, key); err != nil { + glog.Fatalf("random key gen: %v", err) + } + return CipherKey(key) +} + +func Encrypt(plaintext []byte, key CipherKey) ([]byte, error) { + c, err := aes.NewCipher(key) + if err != nil { + return nil, err + } + + gcm, err := cipher.NewGCM(c) + if err != nil { + return nil, err + } + + nonce := make([]byte, gcm.NonceSize()) + if _, err = io.ReadFull(rand.Reader, nonce); err != nil { + return nil, err + } + + return gcm.Seal(nonce, nonce, plaintext, nil), nil +} + +func Decrypt(ciphertext []byte, key CipherKey) ([]byte, error) { + c, err := aes.NewCipher(key) + if err != nil { + return nil, err + } + + gcm, err := cipher.NewGCM(c) + if err != nil { + return nil, err + } + + nonceSize := gcm.NonceSize() + if len(ciphertext) < nonceSize { + return nil, errors.New("ciphertext too short") + } + + nonce, ciphertext := ciphertext[:nonceSize], ciphertext[nonceSize:] + return gcm.Open(nil, nonce, ciphertext, nil) +} diff --git a/weed/util/cipher_test.go b/weed/util/cipher_test.go new file mode 100644 index 000000000..026c96ea3 --- /dev/null +++ b/weed/util/cipher_test.go @@ -0,0 +1,17 @@ +package util + +import ( + "encoding/base64" + "testing" +) + +func TestSameAsJavaImplementation(t *testing.T) { + str := "QVVhmqg112NMT7F+G/7QPynqSln3xPIhKdFGmTVKZD6IS0noyr2Z5kXFF6fPjZ/7Hq8kRhlmLeeqZUccxyaZHezOdgkjS6d4NTdHf5IjXzk7" + cipherText, _ := base64.StdEncoding.DecodeString(str) + secretKey := []byte("256-bit key for AES 256 GCM encr") + plantext, err := Decrypt(cipherText, CipherKey(secretKey)) + if err != nil { + println(err.Error()) + } + println(string(plantext)) +} diff --git a/weed/util/compression.go b/weed/util/compression.go index c6c9423e2..9d52810cb 100644 --- a/weed/util/compression.go +++ b/weed/util/compression.go @@ -4,54 +4,111 @@ import ( "bytes" "compress/flate" "compress/gzip" + "fmt" "io/ioutil" "strings" "github.com/chrislusf/seaweedfs/weed/glog" - "golang.org/x/tools/godoc/util" + // "github.com/klauspost/compress/zstd" ) +var ( + UnsupportedCompression = fmt.Errorf("unsupported compression") +) + +func MaybeGzipData(input []byte) []byte { + if IsGzippedContent(input) { + return input + } + gzipped, err := GzipData(input) + if err != nil { + return input + } + if len(gzipped)*10 > len(input)*9 { + return input + } + return gzipped +} + +func MaybeDecompressData(input []byte) []byte { + uncompressed, err := DecompressData(input) + if err != nil { + if err != UnsupportedCompression { + glog.Errorf("decompressed data: %v", err) + } + return input + } + return uncompressed +} + func GzipData(input []byte) ([]byte, error) { buf := new(bytes.Buffer) w, _ := gzip.NewWriterLevel(buf, flate.BestSpeed) if _, err := w.Write(input); err != nil { - glog.V(2).Infoln("error compressing data:", err) + glog.V(2).Infof("error gzip data: %v", err) return nil, err } if err := w.Close(); err != nil { - glog.V(2).Infoln("error closing compressed data:", err) + glog.V(2).Infof("error closing gzipped data: %v", err) return nil, err } return buf.Bytes(), nil } -func UnGzipData(input []byte) ([]byte, error) { + +func DecompressData(input []byte) ([]byte, error) { + if IsGzippedContent(input) { + return ungzipData(input) + } + /* + if IsZstdContent(input) { + return unzstdData(input) + } + */ + return input, UnsupportedCompression +} + +func ungzipData(input []byte) ([]byte, error) { buf := bytes.NewBuffer(input) r, _ := gzip.NewReader(buf) defer r.Close() output, err := ioutil.ReadAll(r) if err != nil { - glog.V(2).Infoln("error uncompressing data:", err) + glog.V(2).Infof("error ungzip data: %v", err) } return output, err } +func IsGzippedContent(data []byte) bool { + if len(data) < 2 { + return false + } + return data[0] == 31 && data[1] == 139 +} + /* -* Default more not to gzip since gzip can be done on client side. - */func IsGzippable(ext, mtype string, data []byte) bool { +var zstdEncoder, _ = zstd.NewWriter(nil) - shouldBeZipped, iAmSure := IsGzippableFileType(ext, mtype) - if iAmSure { - return shouldBeZipped - } +func ZstdData(input []byte) ([]byte, error) { + return zstdEncoder.EncodeAll(input, nil), nil +} - isMostlyText := util.IsText(data) +var decoder, _ = zstd.NewReader(nil) - return isMostlyText +func unzstdData(input []byte) ([]byte, error) { + return decoder.DecodeAll(input, nil) } +func IsZstdContent(data []byte) bool { + if len(data) < 4 { + return false + } + return data[3] == 0xFD && data[2] == 0x2F && data[1] == 0xB5 && data[0] == 0x28 +} +*/ + /* -* Default more not to gzip since gzip can be done on client side. - */func IsGzippableFileType(ext, mtype string) (shouldBeZipped, iAmSure bool) { +* Default not to compressed since compression can be done on client side. + */func IsCompressableFileType(ext, mtype string) (shouldBeCompressed, iAmSure bool) { // text if strings.HasPrefix(mtype, "text/") { @@ -60,7 +117,7 @@ func UnGzipData(input []byte) ([]byte, error) { // images switch ext { - case ".svg", ".bmp": + case ".svg", ".bmp", ".wav": return true, true } if strings.HasPrefix(mtype, "image/") { @@ -69,7 +126,7 @@ func UnGzipData(input []byte) ([]byte, error) { // by file name extension switch ext { - case ".zip", ".rar", ".gz", ".bz2", ".xz": + case ".zip", ".rar", ".gz", ".bz2", ".xz", ".zst": return false, true case ".pdf", ".txt", ".html", ".htm", ".css", ".js", ".json": return true, true @@ -81,6 +138,9 @@ func UnGzipData(input []byte) ([]byte, error) { // by mime type if strings.HasPrefix(mtype, "application/") { + if strings.HasSuffix(mtype, "zstd") { + return false, true + } if strings.HasSuffix(mtype, "xml") { return true, true } @@ -89,5 +149,12 @@ func UnGzipData(input []byte) ([]byte, error) { } } + if strings.HasPrefix(mtype, "audio/") { + switch strings.TrimPrefix(mtype, "audio/") { + case "wave", "wav", "x-wav", "x-pn-wav": + return true, true + } + } + return false, false } diff --git a/weed/util/compression_test.go b/weed/util/compression_test.go new file mode 100644 index 000000000..b515e8988 --- /dev/null +++ b/weed/util/compression_test.go @@ -0,0 +1,21 @@ +package util + +import ( + "testing" + + "golang.org/x/tools/godoc/util" +) + +func TestIsGzippable(t *testing.T) { + buf := make([]byte, 1024) + + isText := util.IsText(buf) + + if isText { + t.Error("buf with zeros are not text") + } + + compressed, _ := GzipData(buf) + + t.Logf("compressed size %d\n", len(compressed)) +} diff --git a/weed/util/config.go b/weed/util/config.go index 4ba68b800..ee805f26a 100644 --- a/weed/util/config.go +++ b/weed/util/config.go @@ -1,8 +1,12 @@ package util import ( - "github.com/chrislusf/seaweedfs/weed/glog" + "strings" + "sync" + "github.com/spf13/viper" + + "github.com/chrislusf/seaweedfs/weed/glog" ) type Configuration interface { @@ -16,21 +20,23 @@ type Configuration interface { func LoadConfiguration(configFileName string, required bool) (loaded bool) { // find a filer store - viper.SetConfigName(configFileName) // name of config file (without extension) - viper.AddConfigPath(".") // optionally look for config in the working directory - viper.AddConfigPath("$HOME/.seaweedfs") // call multiple times to add many search paths - viper.AddConfigPath("/etc/seaweedfs/") // path to look for the config file in + viper.SetConfigName(configFileName) // name of config file (without extension) + viper.AddConfigPath(".") // optionally look for config in the working directory + viper.AddConfigPath("$HOME/.seaweedfs") // call multiple times to add many search paths + viper.AddConfigPath("/usr/local/etc/seaweedfs/") // search path for bsd-style config directory in + viper.AddConfigPath("/etc/seaweedfs/") // path to look for the config file in glog.V(1).Infof("Reading %s.toml from %s", configFileName, viper.ConfigFileUsed()) if err := viper.MergeInConfig(); err != nil { // Handle errors reading the config file - glog.V(0).Infof("Reading %s: %v", viper.ConfigFileUsed(), err) + if strings.Contains(err.Error(), "Not Found") { + glog.V(1).Infof("Reading %s: %v", viper.ConfigFileUsed(), err) + } else { + glog.Fatalf("Reading %s: %v", viper.ConfigFileUsed(), err) + } if required { glog.Fatalf("Failed to load %s.toml file from current directory, or $HOME/.seaweedfs/, or /etc/seaweedfs/"+ - "\n\nPlease follow this example and add a filer.toml file to "+ - "current directory, or $HOME/.seaweedfs/, or /etc/seaweedfs/:\n"+ - " https://github.com/chrislusf/seaweedfs/blob/master/weed/%s.toml\n"+ - "\nOr use this command to generate the default toml file\n"+ + "\n\nPlease use this command to generate the default %s.toml file\n"+ " weed scaffold -config=%s -output=.\n\n\n", configFileName, configFileName, configFileName) } else { @@ -40,3 +46,56 @@ func LoadConfiguration(configFileName string, required bool) (loaded bool) { return true } + +type ViperProxy struct { + *viper.Viper + sync.Mutex +} + +var ( + vp = &ViperProxy{} +) + +func (vp *ViperProxy) SetDefault(key string, value interface{}) { + vp.Lock() + defer vp.Unlock() + vp.Viper.SetDefault(key, value) +} + +func (vp *ViperProxy) GetString(key string) string { + vp.Lock() + defer vp.Unlock() + return vp.Viper.GetString(key) +} + +func (vp *ViperProxy) GetBool(key string) bool { + vp.Lock() + defer vp.Unlock() + return vp.Viper.GetBool(key) +} + +func (vp *ViperProxy) GetInt(key string) int { + vp.Lock() + defer vp.Unlock() + return vp.Viper.GetInt(key) +} + +func (vp *ViperProxy) GetStringSlice(key string) []string { + vp.Lock() + defer vp.Unlock() + return vp.Viper.GetStringSlice(key) +} + +func GetViper() *ViperProxy { + vp.Lock() + defer vp.Unlock() + + if vp.Viper == nil { + vp.Viper = viper.GetViper() + vp.AutomaticEnv() + vp.SetEnvPrefix("weed") + vp.SetEnvKeyReplacer(strings.NewReplacer(".", "_")) + } + + return vp +} diff --git a/weed/util/constants.go b/weed/util/constants.go index 0916850ef..c595f0c53 100644 --- a/weed/util/constants.go +++ b/weed/util/constants.go @@ -5,5 +5,10 @@ import ( ) var ( - VERSION = fmt.Sprintf("%s %d.%d", sizeLimit, 1, 47) + VERSION = fmt.Sprintf("%s %d.%02d", sizeLimit, 2, 41) + COMMIT = "" ) + +func Version() string { + return VERSION + " " + COMMIT +} diff --git a/weed/util/file_util.go b/weed/util/file_util.go index bef9f7cd6..f83f80265 100644 --- a/weed/util/file_util.go +++ b/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" @@ -49,6 +52,10 @@ func CheckFile(filename string) (exists, canRead, canWrite bool, modTime time.Ti exists = false return } + if err != nil { + glog.Errorf("check %s: %v", filename, err) + return + } if fi.Mode()&0400 != 0 { canRead = true } @@ -59,3 +66,24 @@ func CheckFile(filename string) (exists, canRead, canWrite bool, modTime time.Ti fileSize = fi.Size() return } + +func ResolvePath(path string) string { + + if !strings.Contains(path, "~") { + return path + } + + 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 +} diff --git a/weed/util/fla9/fla9.go b/weed/util/fla9/fla9.go new file mode 100644 index 000000000..eb5700e8c --- /dev/null +++ b/weed/util/fla9/fla9.go @@ -0,0 +1,1149 @@ +// Copyright 2009 The Go Authors. All rights reserved. +// Use of this source code is governed by a BSD-style +// license that can be found in the LICENSE file. + +/* + Package flag implements command-line flag parsing. + + Usage: + + Define flags using flag.String(), Bool(), Int(), etc. + + This declares an integer flag, -flagname, stored in the pointer ip, with type *int. + import "flag" + var ip = flag.Int("flagname", 1234, "help message for flagname") + If you like, you can bind the flag to a variable using the Var() functions. + var flagvar int + func init() { + flag.IntVar(&flagvar, "flagname", 1234, "help message for flagname") + } + Or you can create custom flags that satisfy the Value interface (with + pointer receivers) and couple them to flag parsing by + flag.Var(&flagVal, "name", "help message for flagname") + For such flags, the default value is just the initial value of the variable. + + After all flags are defined, call + flag.Parse() + to parse the command line into the defined flags. + + Flags may then be used directly. If you're using the flags themselves, + they are all pointers; if you bind to variables, they're values. + fmt.Println("ip has value ", *ip) + fmt.Println("flagvar has value ", flagvar) + + After parsing, the arguments following the flags are available as the + slice flag.Args() or individually as flag.Arg(i). + The arguments are indexed from 0 through flag.NArg()-1. + + Command line flag syntax: + -flag + -flag=x + -flag x // non-boolean flags only + One or two minus signs may be used; they are equivalent. + The last form is not permitted for boolean flags because the + meaning of the command + cmd -x * + will change if there is a file called 0, false, etc. You must + use the -flag=false form to turn off a boolean flag. + + Flag parsing stops just before the first non-flag argument + ("-" is a non-flag argument) or after the terminator "--". + + Integer flags accept 1234, 0664, 0x1234 and may be negative. + Boolean flags may be: + 1, 0, t, f, T, F, true, false, TRUE, FALSE, True, False + Duration flags accept any input valid for time.ParseDuration. + + The default set of command-line flags is controlled by + top-level functions. The FlagSet type allows one to define + independent sets of flags, such as to implement subcommands + in a command-line interface. The methods of FlagSet are + analogous to the top-level functions for the command-line + flag set. +*/ +package fla9 + +import ( + "bufio" + "errors" + "fmt" + "io" + "os" + "reflect" + "sort" + "strconv" + "strings" + "time" +) + +// ErrHelp is the error returned if the -help or -h flag is invoked +// but no such flag is defined. +var ErrHelp = errors.New("flag: help requested") + +// -- bool Value +type boolValue bool + +func newBoolValue(val bool, p *bool) *boolValue { + *p = val + return (*boolValue)(p) +} + +func (b *boolValue) Set(s string) error { + v, err := strconv.ParseBool(s) + *b = boolValue(v) + return err +} + +func (b *boolValue) Get() interface{} { return bool(*b) } +func (b *boolValue) String() string { return fmt.Sprintf("%v", *b) } +func (b *boolValue) IsBoolFlag() bool { return true } + +// optional interface to indicate boolean flags that can be +// supplied without "=value" text +type boolFlag interface { + Value + IsBoolFlag() bool +} + +// -- int Value +type intValue int + +func newIntValue(val int, p *int) *intValue { + *p = val + return (*intValue)(p) +} + +func (i *intValue) Set(s string) error { + v, err := strconv.ParseInt(s, 0, 64) + *i = intValue(v) + return err +} + +func (i *intValue) Get() interface{} { return int(*i) } +func (i *intValue) String() string { return fmt.Sprintf("%v", *i) } + +// -- int64 Value +type int64Value int64 + +func newInt64Value(val int64, p *int64) *int64Value { + *p = val + return (*int64Value)(p) +} + +func (i *int64Value) Set(s string) error { + v, err := strconv.ParseInt(s, 0, 64) + *i = int64Value(v) + return err +} + +func (i *int64Value) Get() interface{} { return int64(*i) } +func (i *int64Value) String() string { return fmt.Sprintf("%v", *i) } + +// -- uint Value +type uintValue uint + +func newUintValue(val uint, p *uint) *uintValue { + *p = val + return (*uintValue)(p) +} + +func (i *uintValue) Set(s string) error { + v, err := strconv.ParseUint(s, 0, 64) + *i = uintValue(v) + return err +} + +func (i *uintValue) Get() interface{} { return uint(*i) } +func (i *uintValue) String() string { return fmt.Sprintf("%v", *i) } + +// -- uint64 Value +type uint64Value uint64 + +func newUint64Value(val uint64, p *uint64) *uint64Value { + *p = val + return (*uint64Value)(p) +} + +func (i *uint64Value) Set(s string) error { + v, err := strconv.ParseUint(s, 0, 64) + *i = uint64Value(v) + return err +} + +func (i *uint64Value) Get() interface{} { return uint64(*i) } +func (i *uint64Value) String() string { return fmt.Sprintf("%v", *i) } + +// -- string Value +type stringValue string + +func newStringValue(val string, p *string) *stringValue { + *p = val + return (*stringValue)(p) +} + +func (s *stringValue) Set(val string) error { + *s = stringValue(val) + return nil +} + +func (s *stringValue) Get() interface{} { return string(*s) } +func (s *stringValue) String() string { return fmt.Sprintf("%s", *s) } + +// -- float64 Value +type float64Value float64 + +func newFloat64Value(val float64, p *float64) *float64Value { + *p = val + return (*float64Value)(p) +} + +func (f *float64Value) Set(s string) error { + v, err := strconv.ParseFloat(s, 64) + *f = float64Value(v) + return err +} + +func (f *float64Value) Get() interface{} { return float64(*f) } +func (f *float64Value) String() string { return fmt.Sprintf("%v", *f) } + +// -- time.Duration Value +type durationValue time.Duration + +func newDurationValue(val time.Duration, p *time.Duration) *durationValue { + *p = val + return (*durationValue)(p) +} + +func (d *durationValue) Set(s string) error { + v, err := time.ParseDuration(s) + *d = durationValue(v) + return err +} + +func (d *durationValue) Get() interface{} { return time.Duration(*d) } +func (d *durationValue) String() string { return (*time.Duration)(d).String() } + +// Value is the interface to the dynamic value stored in a flag. +// (The default value is represented as a string.) +// +// If a Value has an IsBoolFlag() bool method returning true, +// the command-line parser makes -name equivalent to -name=true +// rather than using the next command-line argument. +// +// Set is called once, in command line order, for each flag present. +type Value interface { + String() string + Set(string) error +} + +// Getter is an interface that allows the contents of a Value to be retrieved. +// It wraps the Value interface, rather than being part of it, because it +// appeared after Go 1 and its compatibility rules. All Value types provided +// by this package satisfy the Getter interface. +type Getter interface { + Value + Get() interface{} +} + +// ErrorHandling defines how FlagSet.Parse behaves if the parse fails. +type ErrorHandling int + +// These constants cause FlagSet.Parse to behave as described if the parse fails. +const ( + ContinueOnError ErrorHandling = iota // Return a descriptive error. + ExitOnError // Call os.Exit(2). + PanicOnError // Call panic with a descriptive error. +) + +// A FlagSet represents a set of defined flags. The zero value of a FlagSet +// has no name and has ContinueOnError error handling. +type FlagSet struct { + // Usage is the function called when an error occurs while parsing flags. + // The field is a function (not a method) that may be changed to point to + // a custom error handler. + Usage func() + + name string + parsed bool + actual map[string]*Flag + formal map[string]*Flag + envPrefix string // prefix to all env variable names + args []string // arguments after flags + errorHandling ErrorHandling + output io.Writer // nil means stderr; use out() accessor +} + +// A Flag represents the state of a flag. +type Flag struct { + Name string // name as it appears on command line + Usage string // help message + Value Value // value as set + DefValue string // default value (as text); for usage message +} + +// sortFlags returns the flags as a slice in lexicographical sorted order. +func sortFlags(flags map[string]*Flag) []*Flag { + list := make(sort.StringSlice, len(flags)) + i := 0 + for _, f := range flags { + list[i] = f.Name + i++ + } + list.Sort() + result := make([]*Flag, len(list)) + for i, name := range list { + result[i] = flags[name] + } + return result +} + +func (f *FlagSet) out() io.Writer { + if f.output == nil { + return os.Stderr + } + return f.output +} + +// SetOutput sets the destination for usage and error messages. +// If output is nil, os.Stderr is used. +func (f *FlagSet) SetOutput(output io.Writer) { f.output = output } + +// VisitAll visits the flags in lexicographical order, calling fn for each. +// It visits all flags, even those not set. +func (f *FlagSet) VisitAll(fn func(*Flag)) { + for _, flag := range sortFlags(f.formal) { + fn(flag) + } +} + +// VisitAll visits the command-line flags in lexicographical order, calling +// fn for each. It visits all flags, even those not set. +func VisitAll(fn func(*Flag)) { CommandLine.VisitAll(fn) } + +// Visit visits the flags in lexicographical order, calling fn for each. +// It visits only those flags that have been set. +func (f *FlagSet) Visit(fn func(*Flag)) { + for _, flag := range sortFlags(f.actual) { + fn(flag) + } +} + +// Visit visits the command-line flags in lexicographical order, calling fn +// for each. It visits only those flags that have been set. +func Visit(fn func(*Flag)) { CommandLine.Visit(fn) } + +// Lookup returns the Flag structure of the named flag, returning nil if none exists. +func (f *FlagSet) Lookup(name string) *Flag { return f.formal[name] } + +// Lookup returns the Flag structure of the named command-line flag, +// returning nil if none exists. +func Lookup(name string) *Flag { return CommandLine.formal[name] } + +// Set sets the value of the named flag. +func (f *FlagSet) Set(name, value string) error { + flag, ok := f.formal[name] + if !ok { + return fmt.Errorf("no such flag -%v", name) + } + err := flag.Value.Set(value) + if err != nil { + return err + } + if f.actual == nil { + f.actual = make(map[string]*Flag) + } + f.actual[name] = flag + return nil +} + +// Set sets the value of the named command-line flag. +func Set(name, value string) error { return CommandLine.Set(name, value) } + +// isZeroValue guesses whether the string represents the zero +// value for a flag. It is not accurate but in practice works OK. +func isZeroValue(flag *Flag, value string) bool { + // Build a zero value of the flag's Value type, and see if the + // result of calling its String method equals the value passed in. + // This works unless the Value type is itself an interface type. + typ := reflect.TypeOf(flag.Value) + var z reflect.Value + if typ.Kind() == reflect.Ptr { + z = reflect.New(typ.Elem()) + } else { + z = reflect.Zero(typ) + } + if value == z.Interface().(Value).String() { + return true + } + + switch value { + case "false", "", "0": + return true + } + return false +} + +// UnquoteUsage extracts a back-quoted name from the usage +// string for a flag and returns it and the un-quoted usage. +// Given "a `name` to show" it returns ("name", "a name to show"). +// If there are no back quotes, the name is an educated guess of the +// type of the flag's value, or the empty string if the flag is boolean. +func UnquoteUsage(flag *Flag) (name string, usage string) { + // Look for a back-quoted name, but avoid the strings package. + usage = flag.Usage + for i := 0; i < len(usage); i++ { + if usage[i] == '`' { + for j := i + 1; j < len(usage); j++ { + if usage[j] == '`' { + name = usage[i+1 : j] + usage = usage[:i] + name + usage[j+1:] + return name, usage + } + } + break // Only one back quote; use type name. + } + } + // No explicit name, so use type if we can find one. + name = "value" + switch flag.Value.(type) { + case boolFlag: + name = "" + case *durationValue: + name = "duration" + case *float64Value: + name = "float" + case *intValue, *int64Value: + name = "int" + case *stringValue: + name = "string" + case *uintValue, *uint64Value: + name = "uint" + } + return +} + +// PrintDefaults prints to standard error the default values of all +// defined command-line flags in the set. See the documentation for +// the global function PrintDefaults for more information. +func (f *FlagSet) PrintDefaults() { + f.VisitAll(func(flag *Flag) { + s := fmt.Sprintf(" -%s", flag.Name) // Two spaces before -; see next two comments. + name, usage := UnquoteUsage(flag) + if len(name) > 0 { + s += " " + name + } + // Boolean flags of one ASCII letter are so common we + // treat them specially, putting their usage on the same line. + if len(s) <= 4 { // space, space, '-', 'x'. + s += "\t" + } else { + // Four spaces before the tab triggers good alignment + // for both 4- and 8-space tab stops. + s += "\n \t" + } + s += usage + if !isZeroValue(flag, flag.DefValue) { + if _, ok := flag.Value.(*stringValue); ok { + // put quotes on the value + s += fmt.Sprintf(" (default %q)", flag.DefValue) + } else { + s += fmt.Sprintf(" (default %v)", flag.DefValue) + } + } + fmt.Fprint(f.out(), s, "\n") + }) +} + +// PrintDefaults prints, to standard error unless configured otherwise, +// a usage message showing the default settings of all defined +// command-line flags. +// For an integer valued flag x, the default output has the form +// -x int +// usage-message-for-x (default 7) +// The usage message will appear on a separate line for anything but +// a bool flag with a one-byte name. For bool flags, the type is +// omitted and if the flag name is one byte the usage message appears +// on the same line. The parenthetical default is omitted if the +// default is the zero value for the type. The listed type, here int, +// can be changed by placing a back-quoted name in the flag's usage +// string; the first such item in the message is taken to be a parameter +// name to show in the message and the back quotes are stripped from +// the message when displayed. For instance, given +// flag.String("I", "", "search `directory` for include files") +// the output will be +// -I directory +// search directory for include files. +func PrintDefaults() { CommandLine.PrintDefaults() } + +// defaultUsage is the default function to print a usage message. +func defaultUsage(f *FlagSet) { + if f.name == "" { + fmt.Fprintf(f.out(), "Usage:\n") + } else { + fmt.Fprintf(f.out(), "Usage of %s:\n", f.name) + } + f.PrintDefaults() +} + +// NOTE: Usage is not just defaultUsage(CommandLine) +// because it serves (via godoc flag Usage) as the example +// for how to write your own usage function. + +// Usage prints to standard error a usage message documenting all defined command-line flags. +// It is called when an error occurs while parsing flags. +// The function is a variable that may be changed to point to a custom function. +// By default it prints a simple header and calls PrintDefaults; for details about the +// format of the output and how to control it, see the documentation for PrintDefaults. +var Usage = func() { + fmt.Fprintf(os.Stderr, "Usage of %s:\n", os.Args[0]) + PrintDefaults() +} + +// NFlag returns the number of flags that have been set. +func (f *FlagSet) NFlag() int { return len(f.actual) } + +// NFlag returns the number of command-line flags that have been set. +func NFlag() int { return len(CommandLine.actual) } + +// Arg returns the i'th argument. Arg(0) is the first remaining argument +// after flags have been processed. Arg returns an empty string if the +// requested element does not exist. +func (f *FlagSet) Arg(i int) string { + if i < 0 || i >= len(f.args) { + return "" + } + return f.args[i] +} + +// Arg returns the i'th command-line argument. Arg(0) is the first remaining argument +// after flags have been processed. Arg returns an empty string if the +// requested element does not exist. +func Arg(i int) string { return CommandLine.Arg(i) } + +// NArg is the number of arguments remaining after flags have been processed. +func (f *FlagSet) NArg() int { return len(f.args) } + +// NArg is the number of arguments remaining after flags have been processed. +func NArg() int { return len(CommandLine.args) } + +// Args returns the non-flag arguments. +func (f *FlagSet) Args() []string { return f.args } + +// Args returns the non-flag command-line arguments. +func Args() []string { return CommandLine.args } + +// BoolVar defines a bool flag with specified name, default value, and usage string. +// The argument p points to a bool variable in which to store the value of the flag. +func (f *FlagSet) BoolVar(p *bool, name string, value bool, usage string) { + f.Var(newBoolValue(value, p), name, usage) +} + +// BoolVar defines a bool flag with specified name, default value, and usage string. +// The argument p points to a bool variable in which to store the value of the flag. +func BoolVar(p *bool, name string, value bool, usage string) { + CommandLine.Var(newBoolValue(value, p), name, usage) +} + +// Bool defines a bool flag with specified name, default value, and usage string. +// The return value is the address of a bool variable that stores the value of the flag. +func (f *FlagSet) Bool(name string, value bool, usage string) *bool { + p := new(bool) + f.BoolVar(p, name, value, usage) + return p +} + +// Bool defines a bool flag with specified name, default value, and usage string. +// The return value is the address of a bool variable that stores the value of the flag. +func Bool(name string, value bool, usage string) *bool { + return CommandLine.Bool(name, value, usage) +} + +// IntVar defines an int flag with specified name, default value, and usage string. +// The argument p points to an int variable in which to store the value of the flag. +func (f *FlagSet) IntVar(p *int, name string, value int, usage string) { + f.Var(newIntValue(value, p), name, usage) +} + +// IntVar defines an int flag with specified name, default value, and usage string. +// The argument p points to an int variable in which to store the value of the flag. +func IntVar(p *int, name string, value int, usage string) { + CommandLine.Var(newIntValue(value, p), name, usage) +} + +// Int defines an int flag with specified name, default value, and usage string. +// The return value is the address of an int variable that stores the value of the flag. +func (f *FlagSet) Int(name string, value int, usage string) *int { + p := new(int) + f.IntVar(p, name, value, usage) + return p +} + +// Int defines an int flag with specified name, default value, and usage string. +// The return value is the address of an int variable that stores the value of the flag. +func Int(name string, value int, usage string) *int { + return CommandLine.Int(name, value, usage) +} + +// Int64Var defines an int64 flag with specified name, default value, and usage string. +// The argument p points to an int64 variable in which to store the value of the flag. +func (f *FlagSet) Int64Var(p *int64, name string, value int64, usage string) { + f.Var(newInt64Value(value, p), name, usage) +} + +// Int64Var defines an int64 flag with specified name, default value, and usage string. +// The argument p points to an int64 variable in which to store the value of the flag. +func Int64Var(p *int64, name string, value int64, usage string) { + CommandLine.Var(newInt64Value(value, p), name, usage) +} + +// Int64 defines an int64 flag with specified name, default value, and usage string. +// The return value is the address of an int64 variable that stores the value of the flag. +func (f *FlagSet) Int64(name string, value int64, usage string) *int64 { + p := new(int64) + f.Int64Var(p, name, value, usage) + return p +} + +// Int64 defines an int64 flag with specified name, default value, and usage string. +// The return value is the address of an int64 variable that stores the value of the flag. +func Int64(name string, value int64, usage string) *int64 { + return CommandLine.Int64(name, value, usage) +} + +// UintVar defines a uint flag with specified name, default value, and usage string. +// The argument p points to a uint variable in which to store the value of the flag. +func (f *FlagSet) UintVar(p *uint, name string, value uint, usage string) { + f.Var(newUintValue(value, p), name, usage) +} + +// UintVar defines a uint flag with specified name, default value, and usage string. +// The argument p points to a uint variable in which to store the value of the flag. +func UintVar(p *uint, name string, value uint, usage string) { + CommandLine.Var(newUintValue(value, p), name, usage) +} + +// Uint defines a uint flag with specified name, default value, and usage string. +// The return value is the address of a uint variable that stores the value of the flag. +func (f *FlagSet) Uint(name string, value uint, usage string) *uint { + p := new(uint) + f.UintVar(p, name, value, usage) + return p +} + +// Uint defines a uint flag with specified name, default value, and usage string. +// The return value is the address of a uint variable that stores the value of the flag. +func Uint(name string, value uint, usage string) *uint { return CommandLine.Uint(name, value, usage) } + +// Uint64Var defines a uint64 flag with specified name, default value, and usage string. +// The argument p points to a uint64 variable in which to store the value of the flag. +func (f *FlagSet) Uint64Var(p *uint64, name string, value uint64, usage string) { + f.Var(newUint64Value(value, p), name, usage) +} + +// Uint64Var defines a uint64 flag with specified name, default value, and usage string. +// The argument p points to a uint64 variable in which to store the value of the flag. +func Uint64Var(p *uint64, name string, value uint64, usage string) { + CommandLine.Var(newUint64Value(value, p), name, usage) +} + +// Uint64 defines a uint64 flag with specified name, default value, and usage string. +// The return value is the address of a uint64 variable that stores the value of the flag. +func (f *FlagSet) Uint64(name string, value uint64, usage string) *uint64 { + p := new(uint64) + f.Uint64Var(p, name, value, usage) + return p +} + +// Uint64 defines a uint64 flag with specified name, default value, and usage string. +// The return value is the address of a uint64 variable that stores the value of the flag. +func Uint64(name string, value uint64, usage string) *uint64 { + return CommandLine.Uint64(name, value, usage) +} + +// StringVar defines a string flag with specified name, default value, and usage string. +// The argument p points to a string variable in which to store the value of the flag. +func (f *FlagSet) StringVar(p *string, name, value, usage string) { + f.Var(newStringValue(value, p), name, usage) +} + +// StringVar defines a string flag with specified name, default value, and usage string. +// The argument p points to a string variable in which to store the value of the flag. +func StringVar(p *string, name, value, usage string) { + CommandLine.Var(newStringValue(value, p), name, usage) +} + +// String defines a string flag with specified name, default value, and usage string. +// The return value is the address of a string variable that stores the value of the flag. +func (f *FlagSet) String(name, value, usage string) *string { + p := new(string) + f.StringVar(p, name, value, usage) + return p +} + +// String defines a string flag with specified name, default value, and usage string. +// The return value is the address of a string variable that stores the value of the flag. +func String(name, value, usage string) *string { + return CommandLine.String(name, value, usage) +} + +// Float64Var defines a float64 flag with specified name, default value, and usage string. +// The argument p points to a float64 variable in which to store the value of the flag. +func (f *FlagSet) Float64Var(p *float64, name string, value float64, usage string) { + f.Var(newFloat64Value(value, p), name, usage) +} + +// Float64Var defines a float64 flag with specified name, default value, and usage string. +// The argument p points to a float64 variable in which to store the value of the flag. +func Float64Var(p *float64, name string, value float64, usage string) { + CommandLine.Var(newFloat64Value(value, p), name, usage) +} + +// Float64 defines a float64 flag with specified name, default value, and usage string. +// The return value is the address of a float64 variable that stores the value of the flag. +func (f *FlagSet) Float64(name string, value float64, usage string) *float64 { + p := new(float64) + f.Float64Var(p, name, value, usage) + return p +} + +// Float64 defines a float64 flag with specified name, default value, and usage string. +// The return value is the address of a float64 variable that stores the value of the flag. +func Float64(name string, value float64, usage string) *float64 { + return CommandLine.Float64(name, value, usage) +} + +// DurationVar defines a time.Duration flag with specified name, default value, and usage string. +// The argument p points to a time.Duration variable in which to store the value of the flag. +// The flag accepts a value acceptable to time.ParseDuration. +func (f *FlagSet) DurationVar(p *time.Duration, name string, value time.Duration, usage string) { + f.Var(newDurationValue(value, p), name, usage) +} + +// DurationVar defines a time.Duration flag with specified name, default value, and usage string. +// The argument p points to a time.Duration variable in which to store the value of the flag. +// The flag accepts a value acceptable to time.ParseDuration. +func DurationVar(p *time.Duration, name string, value time.Duration, usage string) { + CommandLine.Var(newDurationValue(value, p), name, usage) +} + +// Duration defines a time.Duration flag with specified name, default value, and usage string. +// The return value is the address of a time.Duration variable that stores the value of the flag. +// The flag accepts a value acceptable to time.ParseDuration. +func (f *FlagSet) Duration(name string, value time.Duration, usage string) *time.Duration { + p := new(time.Duration) + f.DurationVar(p, name, value, usage) + return p +} + +// Duration defines a time.Duration flag with specified name, default value, and usage string. +// The return value is the address of a time.Duration variable that stores the value of the flag. +// The flag accepts a value acceptable to time.ParseDuration. +func Duration(name string, value time.Duration, usage string) *time.Duration { + return CommandLine.Duration(name, value, usage) +} + +// Var defines a flag with the specified name and usage string. The type and +// value of the flag are represented by the first argument, of type Value, which +// typically holds a user-defined implementation of Value. For instance, the +// caller could create a flag that turns a comma-separated string into a slice +// of strings by giving the slice the methods of Value; in particular, Set would +// decompose the comma-separated string into the slice. +func (f *FlagSet) Var(value Value, name string, usage string) { + // Remember the default value as a string; it won't change. + flag := &Flag{name, usage, value, value.String()} + _, alreadythere := f.formal[name] + if alreadythere { + var msg string + if f.name == "" { + msg = fmt.Sprintf("flag redefined: %s", name) + } else { + msg = fmt.Sprintf("%s flag redefined: %s", f.name, name) + } + fmt.Fprintln(f.out(), msg) + panic(msg) // Happens only if flags are declared with identical names + } + if f.formal == nil { + f.formal = make(map[string]*Flag) + } + f.formal[name] = flag +} + +// Var defines a flag with the specified name and usage string. The type and +// value of the flag are represented by the first argument, of type Value, which +// typically holds a user-defined implementation of Value. For instance, the +// caller could create a flag that turns a comma-separated string into a slice +// of strings by giving the slice the methods of Value; in particular, Set would +// decompose the comma-separated string into the slice. +func Var(value Value, name, usage string) { + CommandLine.Var(value, name, usage) +} + +// failf prints to standard error a formatted error and usage message and +// returns the error. +func (f *FlagSet) failf(format string, a ...interface{}) error { + err := fmt.Errorf(format, a...) + fmt.Fprintln(f.out(), err) + f.usage() + return err +} + +// usage calls the Usage method for the flag set if one is specified, +// or the appropriate default usage function otherwise. +func (f *FlagSet) usage() { + if f.Usage == nil { + if f == CommandLine { + Usage() + } else { + defaultUsage(f) + } + } else { + f.Usage() + } +} + +// parseOne parses one flag. It reports whether a flag was seen. +func (f *FlagSet) parseOne() (bool, error) { + if len(f.args) == 0 { + return false, nil + } + s := f.args[0] + if len(s) < 2 || s[0] != '-' { + return false, nil + } + numMinuses := 1 + if s[1] == '-' { + numMinuses++ + if len(s) == 2 { // "--" terminates the flags + f.args = f.args[1:] + return false, nil + } + } + name := s[numMinuses:] + if len(name) == 0 || name[0] == '-' || name[0] == '=' { + return false, f.failf("bad flag syntax: %s", s) + } + + // ignore go test flags + if strings.HasPrefix(name, "test.") { + return false, nil + } + + // it's a flag. does it have an argument? + f.args = f.args[1:] + hasValue := false + value := "" + for i := 1; i < len(name); i++ { // equals cannot be first + if name[i] == '=' { + value = name[i+1:] + hasValue = true + name = name[0:i] + break + } + } + m := f.formal + flag, alreadythere := m[name] // BUG + if !alreadythere { + if name == "help" || name == "h" { // special case for nice help message. + f.usage() + return false, ErrHelp + } + return false, f.failf("flag provided but not defined: -%s", name) + } + if fv, ok := flag.Value.(boolFlag); ok && fv.IsBoolFlag() { // special case: doesn't need an arg + if hasValue { + if err := fv.Set(value); err != nil { + return false, f.failf("invalid boolean value %q for -%s: %v", value, name, err) + } + } else { + if err := fv.Set("true"); err != nil { + return false, f.failf("invalid boolean flag %s: %v", name, err) + } + } + } else { + // It must have a value, which might be the next argument. + if !hasValue && len(f.args) > 0 { + // value is the next arg + hasValue = true + value, f.args = f.args[0], f.args[1:] + } + if !hasValue { + return false, f.failf("flag needs an argument: -%s", name) + } + if err := flag.Value.Set(value); err != nil { + return false, f.failf("invalid value %q for flag -%s: %v", value, name, err) + } + } + if f.actual == nil { + f.actual = make(map[string]*Flag) + } + f.actual[name] = flag + return true, nil +} + +// Parse parses flag definitions from the argument list, which should not +// include the command name. Must be called after all flags in the FlagSet +// are defined and before flags are accessed by the program. +// The return value will be ErrHelp if -help or -h were set but not defined. +func (f *FlagSet) Parse(arguments []string) error { + if _, ok := f.formal[DefaultConfigFlagName]; !ok { + f.String(DefaultConfigFlagName, "", "a file of command line options, each line in optionName=optionValue format") + } + + f.parsed = true + f.args = arguments + for { + seen, err := f.parseOne() + if seen { + continue + } + if err == nil { + break + } + switch f.errorHandling { + case ContinueOnError: + return err + case ExitOnError: + os.Exit(2) + case PanicOnError: + panic(err) + } + } + + // Parse environment variables + if err := f.ParseEnv(os.Environ()); err != nil { + switch f.errorHandling { + case ContinueOnError: + return err + case ExitOnError: + os.Exit(2) + case PanicOnError: + panic(err) + } + return err + } + + // Parse configuration from file + var cFile string + if cf := f.formal[DefaultConfigFlagName]; cf != nil { + cFile = cf.Value.String() + } + if cf := f.actual[DefaultConfigFlagName]; cf != nil { + cFile = cf.Value.String() + } + + if cFile == "" { + cFile = f.findConfigArgInUnresolved() + } + + if cFile != "" { + if err := f.ParseFile(cFile, true); err != nil { + switch f.errorHandling { + case ContinueOnError: + return err + case ExitOnError: + os.Exit(2) + case PanicOnError: + panic(err) + } + return err + } + } + + return nil +} + +func (f *FlagSet) findConfigArgInUnresolved() string { + configArg := "-" + DefaultConfigFlagName + for i := 0; i < len(f.args); i++ { + if strings.HasPrefix(f.args[i], configArg) { + if f.args[i] == configArg && i+1 < len(f.args) { + return f.args[i+1] + } + + if strings.HasPrefix(f.args[i], configArg+"=") { + return f.args[i][len(configArg)+1:] + break + } + } + } + return "" +} + +// Parsed reports whether f.Parse has been called. +func (f *FlagSet) Parsed() bool { return f.parsed } + +// Parse parses the command-line flags from os.Args[1:]. Must be called +// after all flags are defined and before flags are accessed by the program. +func Parse() { + // Ignore errors; CommandLine is set for ExitOnError. + CommandLine.Parse(os.Args[1:]) +} + +// Parsed reports whether the command-line flags have been parsed. +func Parsed() bool { + return CommandLine.Parsed() +} + +// CommandLine is the default set of command-line flags, parsed from os.Args. +// The top-level functions such as BoolVar, Arg, and so on are wrappers for the +// methods of CommandLine. +var CommandLine = NewFlagSet(os.Args[0], ExitOnError) + +// NewFlagSet returns a new, empty flag set with the specified name and +// error handling property. +func NewFlagSet(name string, errorHandling ErrorHandling) *FlagSet { + f := &FlagSet{ + name: name, + errorHandling: errorHandling, + envPrefix: EnvPrefix, + } + return f +} + +// Init sets the name and error handling property for a flag set. +// By default, the zero FlagSet uses an empty name, EnvPrefix, and the +// ContinueOnError error handling policy. +func (f *FlagSet) Init(name string, errorHandling ErrorHandling) { + f.name = name + f.envPrefix = EnvPrefix + f.errorHandling = errorHandling +} + +// EnvPrefix defines a string that will be implicitly prefixed to a +// flag name before looking it up in the environment variables. +var EnvPrefix = "WEED" + +// ParseEnv parses flags from environment variables. +// Flags already set will be ignored. +func (f *FlagSet) ParseEnv(environ []string) error { + env := make(map[string]string) + for _, s := range environ { + if i := strings.Index(s, "="); i >= 1 { + env[s[0:i]] = s[i+1:] + } + } + + for _, flag := range f.formal { + name := flag.Name + if _, set := f.actual[name]; set { + continue + } + + flag, alreadyThere := f.formal[name] + if !alreadyThere { + if name == "help" || name == "h" { // special case for nice help message. + f.usage() + return ErrHelp + } + + return f.failf("environment variable provided but not defined: %s", name) + } + + envKey := strings.ToUpper(flag.Name) + if f.envPrefix != "" { + envKey = f.envPrefix + "_" + envKey + } + envKey = strings.Replace(envKey, "-", "_", -1) + + value, isSet := env[envKey] + if !isSet { + continue + } + + if fv, ok := flag.Value.(boolFlag); ok && fv.IsBoolFlag() && value == "" { + // special case: doesn't need an arg + // flag without value is regarded a bool + value = ("true") + } + if err := flag.Value.Set(value); err != nil { + return f.failf("invalid value %q for environment variable %s: %v", value, name, err) + } + + // update f.actual + if f.actual == nil { + f.actual = make(map[string]*Flag) + } + f.actual[name] = flag + } + return nil +} + +// NewFlagSetWithEnvPrefix returns a new empty flag set with the specified name, +// environment variable prefix, and error handling property. +func NewFlagSetWithEnvPrefix(name string, prefix string, errorHandling ErrorHandling) *FlagSet { + f := NewFlagSet(name, errorHandling) + f.envPrefix = prefix + return f +} + +// DefaultConfigFlagName defines the flag name of the optional config file +// path. Used to lookup and parse the config file when a default is set and +// available on disk. +var DefaultConfigFlagName = "options" + +// ParseFile parses flags from the file in path. +// Same format as commandline arguments, newlines and lines beginning with a +// "#" character are ignored. Flags already set will be ignored. +func (f *FlagSet) ParseFile(path string, ignoreUndefinedConf bool) error { + fp, err := os.Open(path) // Extract arguments from file + if err != nil { + return err + } + defer fp.Close() + + scanner := bufio.NewScanner(fp) + for scanner.Scan() { + line := strings.TrimSpace(scanner.Text()) + + // Ignore empty lines or comments + if line == "" || line[:1] == "#" || line[:1] == "//" || line[:1] == "--" { + continue + } + + // Match `key=value` and `key value` + name, value := line, "" + for i, v := range line { + if v == '=' || v == ' ' || v == ':' { + name, value = strings.TrimSpace(line[:i]), strings.TrimSpace(line[i+1:]) + break + } + } + + name = strings.TrimPrefix(name, "-") + + // Ignore flag when already set; arguments have precedence over file + if f.actual[name] != nil { + continue + } + + flag, alreadyThere := f.formal[name] + if !alreadyThere { + if ignoreUndefinedConf { + continue + } + + if name == "help" || name == "h" { // special case for nice help message. + f.usage() + return ErrHelp + } + return f.failf("configuration variable provided but not defined: %s", name) + } + + if fv, ok := flag.Value.(boolFlag); ok && fv.IsBoolFlag() && value == "" { + // special case: doesn't need an arg + value = "true" + } + + if err := flag.Value.Set(value); err != nil { + return f.failf("invalid value %q for configuration variable %s: %v", value, name, err) + } + + // update f.actual + if f.actual == nil { + f.actual = make(map[string]*Flag) + } + f.actual[name] = flag + } + + return scanner.Err() +} diff --git a/weed/util/fullpath.go b/weed/util/fullpath.go new file mode 100644 index 000000000..f2119707e --- /dev/null +++ b/weed/util/fullpath.go @@ -0,0 +1,58 @@ +package util + +import ( + "path/filepath" + "strings" +) + +type FullPath string + +func NewFullPath(dir, name string) FullPath { + return FullPath(dir).Child(name) +} + +func (fp FullPath) DirAndName() (string, string) { + dir, name := filepath.Split(string(fp)) + name = strings.ToValidUTF8(name, "?") + if dir == "/" { + return dir, name + } + if len(dir) < 1 { + return "/", "" + } + return dir[:len(dir)-1], name +} + +func (fp FullPath) Name() string { + _, name := filepath.Split(string(fp)) + name = strings.ToValidUTF8(name, "?") + return name +} + +func (fp FullPath) Child(name string) FullPath { + dir := string(fp) + if strings.HasSuffix(dir, "/") { + return FullPath(dir + name) + } + return FullPath(dir + "/" + name) +} + +func (fp FullPath) AsInode() uint64 { + return uint64(HashStringToLong(string(fp))) +} + +// split, but skipping the root +func (fp FullPath) Split() []string { + if fp == "" || fp == "/" { + return []string{} + } + return strings.Split(string(fp)[1:], "/") +} + +func Join(names ...string) string { + return filepath.ToSlash(filepath.Join(names...)) +} + +func JoinPath(names ...string) FullPath { + return FullPath(Join(names...)) +} diff --git a/weed/util/pprof.go b/weed/util/grace/pprof.go index a2621ceee..14686bfc8 100644 --- a/weed/util/pprof.go +++ b/weed/util/grace/pprof.go @@ -1,4 +1,4 @@ -package util +package grace import ( "os" diff --git a/weed/util/signal_handling.go b/weed/util/grace/signal_handling.go index 99447e8be..7cca46764 100644 --- a/weed/util/signal_handling.go +++ b/weed/util/grace/signal_handling.go @@ -1,6 +1,6 @@ // +build !plan9 -package util +package grace import ( "os" diff --git a/weed/util/signal_handling_notsupported.go b/weed/util/grace/signal_handling_notsupported.go index c389cfb7e..5335915a1 100644 --- a/weed/util/signal_handling_notsupported.go +++ b/weed/util/grace/signal_handling_notsupported.go @@ -1,6 +1,6 @@ // +build plan9 -package util +package grace func OnInterrupt(fn func()) { } diff --git a/weed/util/grpc_client_server.go b/weed/util/grpc_client_server.go deleted file mode 100644 index 31497ad35..000000000 --- a/weed/util/grpc_client_server.go +++ /dev/null @@ -1,120 +0,0 @@ -package util - -import ( - "context" - "fmt" - "net/http" - "strconv" - "strings" - "sync" - "time" - - "google.golang.org/grpc" - "google.golang.org/grpc/keepalive" -) - -var ( - // cache grpc connections - grpcClients = make(map[string]*grpc.ClientConn) - grpcClientsLock sync.Mutex -) - -func init() { - http.DefaultTransport.(*http.Transport).MaxIdleConnsPerHost = 1024 -} - -func NewGrpcServer(opts ...grpc.ServerOption) *grpc.Server { - var options []grpc.ServerOption - options = append(options, grpc.KeepaliveParams(keepalive.ServerParameters{ - Time: 10 * time.Second, // wait time before ping if no activity - Timeout: 20 * time.Second, // ping timeout - }), grpc.KeepaliveEnforcementPolicy(keepalive.EnforcementPolicy{ - MinTime: 60 * time.Second, // min time a client should wait before sending a ping - })) - for _, opt := range opts { - if opt != nil { - options = append(options, opt) - } - } - return grpc.NewServer(options...) -} - -func GrpcDial(ctx context.Context, address string, opts ...grpc.DialOption) (*grpc.ClientConn, error) { - // opts = append(opts, grpc.WithBlock()) - // opts = append(opts, grpc.WithTimeout(time.Duration(5*time.Second))) - var options []grpc.DialOption - options = append(options, - // grpc.WithInsecure(), - grpc.WithKeepaliveParams(keepalive.ClientParameters{ - Time: 30 * time.Second, // client ping server if no activity for this long - Timeout: 20 * time.Second, - })) - for _, opt := range opts { - if opt != nil { - options = append(options, opt) - } - } - return grpc.DialContext(ctx, address, options...) -} - -func WithCachedGrpcClient(ctx context.Context, fn func(*grpc.ClientConn) error, address string, opts ...grpc.DialOption) error { - - grpcClientsLock.Lock() - - existingConnection, found := grpcClients[address] - if found { - grpcClientsLock.Unlock() - return fn(existingConnection) - } - - grpcConnection, err := GrpcDial(ctx, address, opts...) - if err != nil { - grpcClientsLock.Unlock() - return fmt.Errorf("fail to dial %s: %v", address, err) - } - - grpcClients[address] = grpcConnection - grpcClientsLock.Unlock() - - err = fn(grpcConnection) - if err != nil { - grpcClientsLock.Lock() - delete(grpcClients, address) - grpcClientsLock.Unlock() - grpcConnection.Close() - } - - return err -} - -func ParseServerToGrpcAddress(server string) (serverGrpcAddress string, err error) { - colonIndex := strings.LastIndex(server, ":") - if colonIndex < 0 { - return "", fmt.Errorf("server should have hostname:port format: %v", server) - } - - port, parseErr := strconv.ParseUint(server[colonIndex+1:], 10, 64) - if parseErr != nil { - return "", fmt.Errorf("server port parse error: %v", parseErr) - } - - grpcPort := int(port) + 10000 - - return fmt.Sprintf("%s:%d", server[:colonIndex], grpcPort), nil -} - -func ServerToGrpcAddress(server string) (serverGrpcAddress string) { - hostnameAndPort := strings.Split(server, ":") - if len(hostnameAndPort) != 2 { - return fmt.Sprintf("unexpected server address: %s", server) - } - - port, parseErr := strconv.ParseUint(hostnameAndPort[1], 10, 64) - if parseErr != nil { - return fmt.Sprintf("failed to parse port for %s:%s", hostnameAndPort[0], hostnameAndPort[1]) - } - - grpcPort := int(port) + 10000 - - return fmt.Sprintf("%s:%d", hostnameAndPort[0], grpcPort) -} diff --git a/weed/util/http_util.go b/weed/util/http_util.go index 667d0b4be..1630760b1 100644 --- a/weed/util/http_util.go +++ b/weed/util/http_util.go @@ -1,7 +1,6 @@ package util import ( - "bytes" "compress/gzip" "encoding/json" "errors" @@ -11,6 +10,8 @@ import ( "net/http" "net/url" "strings" + + "github.com/chrislusf/seaweedfs/weed/glog" ) var ( @@ -20,6 +21,7 @@ var ( func init() { Transport = &http.Transport{ + MaxIdleConns: 1024, MaxIdleConnsPerHost: 1024, } client = &http.Client{ @@ -27,22 +29,6 @@ func init() { } } -func PostBytes(url string, body []byte) ([]byte, error) { - r, err := client.Post(url, "", bytes.NewReader(body)) - if err != nil { - return nil, fmt.Errorf("Post to %s: %v", url, err) - } - defer r.Body.Close() - if r.StatusCode >= 400 { - return nil, fmt.Errorf("%s: %s", url, r.Status) - } - b, err := ioutil.ReadAll(r.Body) - if err != nil { - return nil, fmt.Errorf("Read response body: %v", err) - } - return b, nil -} - func Post(url string, values url.Values) ([]byte, error) { r, err := client.PostForm(url, values) if err != nil { @@ -65,20 +51,35 @@ func Post(url string, values url.Values) ([]byte, error) { // github.com/chrislusf/seaweedfs/unmaintained/repeated_vacuum/repeated_vacuum.go // may need increasing http.Client.Timeout -func Get(url string) ([]byte, error) { - r, err := client.Get(url) +func Get(url string) ([]byte, bool, error) { + + request, err := http.NewRequest("GET", url, nil) + request.Header.Add("Accept-Encoding", "gzip") + + response, err := client.Do(request) if err != nil { - return nil, err + return nil, true, err } - defer r.Body.Close() - b, err := ioutil.ReadAll(r.Body) - if r.StatusCode >= 400 { - return nil, fmt.Errorf("%s: %s", url, r.Status) + defer response.Body.Close() + + var reader io.ReadCloser + switch response.Header.Get("Content-Encoding") { + case "gzip": + reader, err = gzip.NewReader(response.Body) + defer reader.Close() + default: + reader = response.Body + } + + b, err := ioutil.ReadAll(reader) + if response.StatusCode >= 400 { + retryable := response.StatusCode >= 500 + return nil, retryable, fmt.Errorf("%s: %s", url, response.Status) } if err != nil { - return nil, err + return nil, false, err } - return b, nil + return b, false, nil } func Head(url string) (http.Header, error) { @@ -86,7 +87,7 @@ func Head(url string) (http.Header, error) { if err != nil { return nil, err } - defer r.Body.Close() + defer CloseResponse(r) if r.StatusCode >= 400 { return nil, fmt.Errorf("%s: %s", url, r.Status) } @@ -115,7 +116,7 @@ func Delete(url string, jwt string) error { return nil } m := make(map[string]interface{}) - if e := json.Unmarshal(body, m); e == nil { + if e := json.Unmarshal(body, &m); e == nil { if s, ok := m["error"].(string); ok { return errors.New(s) } @@ -123,12 +124,33 @@ func Delete(url string, jwt string) error { return errors.New(string(body)) } +func DeleteProxied(url string, jwt string) (body []byte, httpStatus int, err error) { + req, err := http.NewRequest("DELETE", url, nil) + if jwt != "" { + req.Header.Set("Authorization", "BEARER "+string(jwt)) + } + if err != nil { + return + } + resp, err := client.Do(req) + if err != nil { + return + } + defer resp.Body.Close() + body, err = ioutil.ReadAll(resp.Body) + if err != nil { + return + } + httpStatus = resp.StatusCode + return +} + func GetBufferStream(url string, values url.Values, allocatedBytes []byte, eachBuffer func([]byte)) error { r, err := client.PostForm(url, values) if err != nil { return err } - defer r.Body.Close() + defer CloseResponse(r) if r.StatusCode != 200 { return fmt.Errorf("%s: %s", url, r.Status) } @@ -151,14 +173,14 @@ func GetUrlStream(url string, values url.Values, readFn func(io.Reader) error) e if err != nil { return err } - defer r.Body.Close() + defer CloseResponse(r) if r.StatusCode != 200 { return fmt.Errorf("%s: %s", url, r.Status) } return readFn(r.Body) } -func DownloadFile(fileUrl string) (filename string, header http.Header, rc io.ReadCloser, e error) { +func DownloadFile(fileUrl string) (filename string, header http.Header, resp *http.Response, e error) { response, err := client.Get(fileUrl) if err != nil { return "", nil, nil, err @@ -172,7 +194,7 @@ func DownloadFile(fileUrl string) (filename string, header http.Header, rc io.Re filename = strings.Trim(filename, "\"") } } - rc = response.Body + resp = response return } @@ -187,14 +209,22 @@ func NormalizeUrl(url string) string { return "http://" + url } -func ReadUrl(fileUrl string, offset int64, size int, buf []byte, isReadRange bool) (int64, error) { +func ReadUrl(fileUrl string, cipherKey []byte, isContentCompressed bool, isFullChunk bool, offset int64, size int, buf []byte) (int64, error) { + + if cipherKey != nil { + var n int + _, err := readEncryptedUrl(fileUrl, cipherKey, isContentCompressed, isFullChunk, offset, size, func(data []byte) { + n = copy(buf, data) + }) + return int64(n), err + } 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))) + if !isFullChunk { + req.Header.Add("Range", fmt.Sprintf("bytes=%d-%d", offset, offset+int64(size)-1)) } else { req.Header.Set("Accept-Encoding", "gzip") } @@ -210,7 +240,8 @@ func ReadUrl(fileUrl string, offset int64, size int, buf []byte, isReadRange boo } var reader io.ReadCloser - switch r.Header.Get("Content-Encoding") { + contentEncoding := r.Header.Get("Content-Encoding") + switch contentEncoding { case "gzip": reader, err = gzip.NewReader(r.Body) defer reader.Close() @@ -242,44 +273,131 @@ func ReadUrl(fileUrl string, offset int64, size int, buf []byte, isReadRange boo // 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)) + glog.V(1).Infof("%s reader has remaining %d bytes", contentEncoding, len(data)) } return n, err } -func ReadUrlAsStream(fileUrl string, offset int64, size int, fn func(data []byte)) (int64, error) { +func ReadUrlAsStream(fileUrl string, cipherKey []byte, isContentGzipped bool, isFullChunk bool, offset int64, size int, fn func(data []byte)) (retryable bool, err error) { + + if cipherKey != nil { + return readEncryptedUrl(fileUrl, cipherKey, isContentGzipped, isFullChunk, offset, size, fn) + } req, err := http.NewRequest("GET", fileUrl, nil) if err != nil { - return 0, err + return false, err + } + + if isFullChunk { + req.Header.Add("Accept-Encoding", "gzip") + } else { + req.Header.Add("Range", fmt.Sprintf("bytes=%d-%d", offset, offset+int64(size)-1)) } - req.Header.Add("Range", fmt.Sprintf("bytes=%d-%d", offset, offset+int64(size))) r, err := client.Do(req) if err != nil { - return 0, err + return true, err } - defer r.Body.Close() + defer CloseResponse(r) if r.StatusCode >= 400 { - return 0, fmt.Errorf("%s: %s", fileUrl, r.Status) + retryable = r.StatusCode >= 500 + return retryable, fmt.Errorf("%s: %s", fileUrl, r.Status) + } + + var reader io.ReadCloser + contentEncoding := r.Header.Get("Content-Encoding") + switch contentEncoding { + case "gzip": + reader, err = gzip.NewReader(r.Body) + defer reader.Close() + default: + reader = r.Body } var ( m int - n int64 ) buf := make([]byte, 64*1024) for { - m, err = r.Body.Read(buf) + m, err = reader.Read(buf) fn(buf[:m]) - n += int64(m) if err == io.EOF { - return n, nil + return false, nil } if err != nil { - return n, err + return false, err } } } + +func readEncryptedUrl(fileUrl string, cipherKey []byte, isContentCompressed bool, isFullChunk bool, offset int64, size int, fn func(data []byte)) (bool, error) { + encryptedData, retryable, err := Get(fileUrl) + if err != nil { + return retryable, fmt.Errorf("fetch %s: %v", fileUrl, err) + } + decryptedData, err := Decrypt(encryptedData, CipherKey(cipherKey)) + if err != nil { + return false, fmt.Errorf("decrypt %s: %v", fileUrl, err) + } + if isContentCompressed { + decryptedData, err = DecompressData(decryptedData) + if err != nil { + glog.V(0).Infof("unzip decrypt %s: %v", fileUrl, err) + } + } + if len(decryptedData) < int(offset)+size { + return false, fmt.Errorf("read decrypted %s size %d [%d, %d)", fileUrl, len(decryptedData), offset, int(offset)+size) + } + if isFullChunk { + fn(decryptedData) + } else { + fn(decryptedData[int(offset) : int(offset)+size]) + } + return false, nil +} + +func ReadUrlAsReaderCloser(fileUrl string, rangeHeader string) (io.ReadCloser, error) { + + req, err := http.NewRequest("GET", fileUrl, nil) + if err != nil { + return nil, err + } + if rangeHeader != "" { + req.Header.Add("Range", rangeHeader) + } else { + req.Header.Add("Accept-Encoding", "gzip") + } + + r, err := client.Do(req) + if err != nil { + return nil, err + } + if r.StatusCode >= 400 { + return nil, fmt.Errorf("%s: %s", fileUrl, r.Status) + } + + var reader io.ReadCloser + contentEncoding := r.Header.Get("Content-Encoding") + switch contentEncoding { + case "gzip": + reader, err = gzip.NewReader(r.Body) + defer reader.Close() + default: + reader = r.Body + } + + return reader, nil +} + +func CloseResponse(resp *http.Response) { + io.Copy(ioutil.Discard, resp.Body) + resp.Body.Close() +} + +func CloseRequest(req *http.Request) { + io.Copy(ioutil.Discard, req.Body) + req.Body.Close() +} diff --git a/weed/util/inits.go b/weed/util/inits.go new file mode 100644 index 000000000..378878012 --- /dev/null +++ b/weed/util/inits.go @@ -0,0 +1,52 @@ +package util + +import ( + "fmt" + "sort" +) + +// HumanReadableIntsMax joins a serials of inits into a smart one like 1-3 5 ... for human readable. +func HumanReadableIntsMax(max int, ids ...int) string { + if len(ids) <= max { + return HumanReadableInts(ids...) + } + + return HumanReadableInts(ids[:max]...) + " ..." +} + +// HumanReadableInts joins a serials of inits into a smart one like 1-3 5 7-10 for human readable. +func HumanReadableInts(ids ...int) string { + sort.Ints(ids) + + s := "" + start := 0 + last := 0 + + for i, v := range ids { + if i == 0 { + start = v + last = v + s = fmt.Sprintf("%d", v) + continue + } + + if last+1 == v { + last = v + continue + } + + if last > start { + s += fmt.Sprintf("-%d", last) + } + + s += fmt.Sprintf(" %d", v) + start = v + last = v + } + + if last != start { + s += fmt.Sprintf("-%d", last) + } + + return s +} diff --git a/weed/util/inits_test.go b/weed/util/inits_test.go new file mode 100644 index 000000000..f2c9b701f --- /dev/null +++ b/weed/util/inits_test.go @@ -0,0 +1,19 @@ +package util + +import ( + "github.com/stretchr/testify/assert" + "testing" +) + +func TestHumanReadableIntsMax(t *testing.T) { + assert.Equal(t, "1-2 ...", HumanReadableIntsMax(2, 1, 2, 3)) + assert.Equal(t, "1 3 ...", HumanReadableIntsMax(2, 1, 3, 5)) +} + +func TestHumanReadableInts(t *testing.T) { + assert.Equal(t, "1-3", HumanReadableInts(1, 2, 3)) + assert.Equal(t, "1 3", HumanReadableInts(1, 3)) + assert.Equal(t, "1 3 5", HumanReadableInts(5, 1, 3)) + assert.Equal(t, "1-3 5", HumanReadableInts(1, 2, 3, 5)) + assert.Equal(t, "1-3 5 7-9", HumanReadableInts(7, 9, 8, 1, 2, 3, 5)) +} diff --git a/weed/util/limiter.go b/weed/util/limiter.go new file mode 100644 index 000000000..2debaaa85 --- /dev/null +++ b/weed/util/limiter.go @@ -0,0 +1,114 @@ +package util + +import ( + "math/rand" + "reflect" + "sync" + "sync/atomic" +) + +// initial version comes from https://github.com/korovkin/limiter/blob/master/limiter.go + +// LimitedConcurrentExecutor object +type LimitedConcurrentExecutor struct { + limit int + tokenChan chan int +} + +func NewLimitedConcurrentExecutor(limit int) *LimitedConcurrentExecutor { + + // allocate a limiter instance + c := &LimitedConcurrentExecutor{ + limit: limit, + tokenChan: make(chan int, limit), + } + + // allocate the tokenChan: + for i := 0; i < c.limit; i++ { + c.tokenChan <- i + } + + return c +} + +// Execute adds a function to the execution queue. +// if num of go routines allocated by this instance is < limit +// launch a new go routine to execute job +// else wait until a go routine becomes available +func (c *LimitedConcurrentExecutor) Execute(job func()) { + token := <-c.tokenChan + go func() { + defer func() { + c.tokenChan <- token + }() + // run the job + job() + }() +} + +// a different implementation, but somehow more "conservative" +type OperationRequest func() + +type LimitedOutOfOrderProcessor struct { + processorSlots uint32 + processors []chan OperationRequest + processorLimit int32 + processorLimitCond *sync.Cond + currentProcessor int32 +} + +func NewLimitedOutOfOrderProcessor(limit int32) (c *LimitedOutOfOrderProcessor) { + + processorSlots := uint32(32) + c = &LimitedOutOfOrderProcessor{ + processorSlots: processorSlots, + processors: make([]chan OperationRequest, processorSlots), + processorLimit: limit, + processorLimitCond: sync.NewCond(new(sync.Mutex)), + } + + for i := 0; i < int(processorSlots); i++ { + c.processors[i] = make(chan OperationRequest) + } + + cases := make([]reflect.SelectCase, processorSlots) + for i, ch := range c.processors { + cases[i] = reflect.SelectCase{Dir: reflect.SelectRecv, Chan: reflect.ValueOf(ch)} + } + + go func() { + for { + _, value, ok := reflect.Select(cases) + if !ok { + continue + } + + request := value.Interface().(OperationRequest) + + if c.processorLimit > 0 { + c.processorLimitCond.L.Lock() + for atomic.LoadInt32(&c.currentProcessor) > c.processorLimit { + c.processorLimitCond.Wait() + } + atomic.AddInt32(&c.currentProcessor, 1) + c.processorLimitCond.L.Unlock() + } + + go func() { + if c.processorLimit > 0 { + defer atomic.AddInt32(&c.currentProcessor, -1) + defer c.processorLimitCond.Signal() + } + request() + }() + + } + }() + + return c +} + +func (c *LimitedOutOfOrderProcessor) Execute(request OperationRequest) { + index := rand.Uint32() % c.processorSlots + c.processors[index] <- request +} diff --git a/weed/util/log_buffer/log_buffer.go b/weed/util/log_buffer/log_buffer.go new file mode 100644 index 000000000..f84c674ff --- /dev/null +++ b/weed/util/log_buffer/log_buffer.go @@ -0,0 +1,292 @@ +package log_buffer + +import ( + "bytes" + "sync" + "time" + + "github.com/golang/protobuf/proto" + + "github.com/chrislusf/seaweedfs/weed/glog" + "github.com/chrislusf/seaweedfs/weed/pb/filer_pb" + "github.com/chrislusf/seaweedfs/weed/util" +) + +const BufferSize = 4 * 1024 * 1024 +const PreviousBufferCount = 3 + +type dataToFlush struct { + startTime time.Time + stopTime time.Time + data *bytes.Buffer +} + +type LogBuffer struct { + prevBuffers *SealedBuffers + buf []byte + idx []int + pos int + startTime time.Time + stopTime time.Time + lastFlushTime time.Time + sizeBuf []byte + flushInterval time.Duration + flushFn func(startTime, stopTime time.Time, buf []byte) + notifyFn func() + isStopping bool + flushChan chan *dataToFlush + lastTsNs int64 + sync.RWMutex +} + +func NewLogBuffer(flushInterval time.Duration, flushFn func(startTime, stopTime time.Time, buf []byte), notifyFn func()) *LogBuffer { + lb := &LogBuffer{ + prevBuffers: newSealedBuffers(PreviousBufferCount), + buf: make([]byte, BufferSize), + sizeBuf: make([]byte, 4), + flushInterval: flushInterval, + flushFn: flushFn, + notifyFn: notifyFn, + flushChan: make(chan *dataToFlush, 256), + } + go lb.loopFlush() + go lb.loopInterval() + return lb +} + +func (m *LogBuffer) AddToBuffer(partitionKey, data []byte, eventTsNs int64) { + + m.Lock() + defer func() { + m.Unlock() + if m.notifyFn != nil { + m.notifyFn() + } + }() + + // need to put the timestamp inside the lock + var ts time.Time + if eventTsNs == 0 { + ts = time.Now() + eventTsNs = ts.UnixNano() + } else { + ts = time.Unix(0, eventTsNs) + } + if m.lastTsNs >= eventTsNs { + // this is unlikely to happen, but just in case + eventTsNs = m.lastTsNs + 1 + ts = time.Unix(0, eventTsNs) + } + m.lastTsNs = eventTsNs + logEntry := &filer_pb.LogEntry{ + TsNs: eventTsNs, + PartitionKeyHash: util.HashToInt32(partitionKey), + Data: data, + } + + logEntryData, _ := proto.Marshal(logEntry) + + size := len(logEntryData) + + if m.pos == 0 { + m.startTime = ts + } + + if m.startTime.Add(m.flushInterval).Before(ts) || len(m.buf)-m.pos < size+4 { + m.flushChan <- m.copyToFlush() + m.startTime = ts + if len(m.buf) < size+4 { + m.buf = make([]byte, 2*size+4) + } + } + m.stopTime = ts + + m.idx = append(m.idx, m.pos) + util.Uint32toBytes(m.sizeBuf, uint32(size)) + copy(m.buf[m.pos:m.pos+4], m.sizeBuf) + copy(m.buf[m.pos+4:m.pos+4+size], logEntryData) + m.pos += size + 4 + + // fmt.Printf("entry size %d total %d count %d, buffer:%p\n", size, m.pos, len(m.idx), m) + +} + +func (m *LogBuffer) Shutdown() { + m.Lock() + defer m.Unlock() + + if m.isStopping { + return + } + m.isStopping = true + toFlush := m.copyToFlush() + m.flushChan <- toFlush + close(m.flushChan) +} + +func (m *LogBuffer) loopFlush() { + for d := range m.flushChan { + if d != nil { + // fmt.Printf("flush [%v, %v] size %d\n", d.startTime, d.stopTime, len(d.data.Bytes())) + m.flushFn(d.startTime, d.stopTime, d.data.Bytes()) + d.releaseMemory() + m.lastFlushTime = d.stopTime + } + } +} + +func (m *LogBuffer) loopInterval() { + for !m.isStopping { + time.Sleep(m.flushInterval) + m.Lock() + if m.isStopping { + m.Unlock() + return + } + // println("loop interval") + toFlush := m.copyToFlush() + m.flushChan <- toFlush + m.Unlock() + } +} + +func (m *LogBuffer) copyToFlush() *dataToFlush { + + if m.pos > 0 { + // fmt.Printf("flush buffer %d pos %d empty space %d\n", len(m.buf), m.pos, len(m.buf)-m.pos) + var d *dataToFlush + if m.flushFn != nil { + d = &dataToFlush{ + startTime: m.startTime, + stopTime: m.stopTime, + data: copiedBytes(m.buf[:m.pos]), + } + } + // fmt.Printf("flusing [0,%d) with %d entries\n", m.pos, len(m.idx)) + m.buf = m.prevBuffers.SealBuffer(m.startTime, m.stopTime, m.buf, m.pos) + m.pos = 0 + m.idx = m.idx[:0] + return d + } + return nil +} + +func (d *dataToFlush) releaseMemory() { + d.data.Reset() + bufferPool.Put(d.data) +} + +func (m *LogBuffer) ReadFromBuffer(lastReadTime time.Time) (bufferCopy *bytes.Buffer, err error) { + m.RLock() + defer m.RUnlock() + + if !m.lastFlushTime.IsZero() && m.lastFlushTime.After(lastReadTime) { + return nil, ResumeFromDiskError + } + + /* + fmt.Printf("read buffer %p: %v last stop time: [%v,%v], pos %d, entries:%d, prevBufs:%d\n", m, lastReadTime, m.startTime, m.stopTime, m.pos, len(m.idx), len(m.prevBuffers.buffers)) + for i, prevBuf := range m.prevBuffers.buffers { + fmt.Printf(" prev %d : %s\n", i, prevBuf.String()) + } + */ + + if lastReadTime.Equal(m.stopTime) { + return nil, nil + } + if lastReadTime.After(m.stopTime) { + // glog.Fatalf("unexpected last read time %v, older than latest %v", lastReadTime, m.stopTime) + return nil, nil + } + if lastReadTime.Before(m.startTime) { + // println("checking ", lastReadTime.UnixNano()) + for i, buf := range m.prevBuffers.buffers { + if buf.startTime.After(lastReadTime) { + if i == 0 { + // println("return the earliest in memory", buf.startTime.UnixNano()) + return copiedBytes(buf.buf[:buf.size]), nil + } + // println("return the", i, "th in memory", buf.startTime.UnixNano()) + return copiedBytes(buf.buf[:buf.size]), nil + } + if !buf.startTime.After(lastReadTime) && buf.stopTime.After(lastReadTime) { + pos := buf.locateByTs(lastReadTime) + // fmt.Printf("locate buffer[%d] pos %d\n", i, pos) + return copiedBytes(buf.buf[pos:buf.size]), nil + } + } + // println("return the current buf", lastReadTime.UnixNano()) + return copiedBytes(m.buf[:m.pos]), nil + } + + lastTs := lastReadTime.UnixNano() + l, h := 0, len(m.idx)-1 + + /* + for i, pos := range m.idx { + logEntry, ts := readTs(m.buf, pos) + event := &filer_pb.SubscribeMetadataResponse{} + proto.Unmarshal(logEntry.Data, event) + entry := event.EventNotification.OldEntry + if entry == nil { + entry = event.EventNotification.NewEntry + } + fmt.Printf("entry %d ts: %v offset:%d dir:%s name:%s\n", i, time.Unix(0, ts), pos, event.Directory, entry.Name) + } + fmt.Printf("l=%d, h=%d\n", l, h) + */ + + for l <= h { + mid := (l + h) / 2 + pos := m.idx[mid] + _, t := readTs(m.buf, pos) + if t <= lastTs { + l = mid + 1 + } else if lastTs < t { + var prevT int64 + if mid > 0 { + _, prevT = readTs(m.buf, m.idx[mid-1]) + } + if prevT <= lastTs { + // fmt.Printf("found l=%d, m-1=%d(ts=%d), m=%d(ts=%d), h=%d [%d, %d) \n", l, mid-1, prevT, mid, t, h, pos, m.pos) + return copiedBytes(m.buf[pos:m.pos]), nil + } + h = mid + } + // fmt.Printf("l=%d, h=%d\n", l, h) + } + + // FIXME: this could be that the buffer has been flushed already + return nil, nil + +} +func (m *LogBuffer) ReleaseMemory(b *bytes.Buffer) { + bufferPool.Put(b) +} + +var bufferPool = sync.Pool{ + New: func() interface{} { + return new(bytes.Buffer) + }, +} + +func copiedBytes(buf []byte) (copied *bytes.Buffer) { + copied = bufferPool.Get().(*bytes.Buffer) + copied.Reset() + copied.Write(buf) + return +} + +func readTs(buf []byte, pos int) (size int, ts int64) { + + size = int(util.BytesToUint32(buf[pos : pos+4])) + entryData := buf[pos+4 : pos+4+size] + logEntry := &filer_pb.LogEntry{} + + err := proto.Unmarshal(entryData, logEntry) + if err != nil { + glog.Fatalf("unexpected unmarshal filer_pb.LogEntry: %v", err) + } + return size, logEntry.TsNs + +} diff --git a/weed/util/log_buffer/log_buffer_test.go b/weed/util/log_buffer/log_buffer_test.go new file mode 100644 index 000000000..3d77afb18 --- /dev/null +++ b/weed/util/log_buffer/log_buffer_test.go @@ -0,0 +1,42 @@ +package log_buffer + +import ( + "fmt" + "math/rand" + "testing" + "time" + + "github.com/chrislusf/seaweedfs/weed/pb/filer_pb" +) + +func TestNewLogBufferFirstBuffer(t *testing.T) { + lb := NewLogBuffer(time.Minute, func(startTime, stopTime time.Time, buf []byte) { + + }, func() { + + }) + + startTime := time.Now() + + messageSize := 1024 + messageCount := 5000 + var buf = make([]byte, messageSize) + for i := 0; i < messageCount; i++ { + rand.Read(buf) + lb.AddToBuffer(nil, buf, 0) + } + + receivedmessageCount := 0 + lb.LoopProcessLogData(startTime, func() bool { + // stop if no more messages + return false + }, func(logEntry *filer_pb.LogEntry) error { + receivedmessageCount++ + return nil + }) + + if receivedmessageCount != messageCount { + fmt.Printf("sent %d received %d\n", messageCount, receivedmessageCount) + } + +} diff --git a/weed/util/log_buffer/log_read.go b/weed/util/log_buffer/log_read.go new file mode 100644 index 000000000..d6917abfe --- /dev/null +++ b/weed/util/log_buffer/log_read.go @@ -0,0 +1,89 @@ +package log_buffer + +import ( + "bytes" + "fmt" + "time" + + "github.com/golang/protobuf/proto" + + "github.com/chrislusf/seaweedfs/weed/glog" + "github.com/chrislusf/seaweedfs/weed/pb/filer_pb" + "github.com/chrislusf/seaweedfs/weed/util" +) + +var ( + ResumeError = fmt.Errorf("resume") + ResumeFromDiskError = fmt.Errorf("resumeFromDisk") +) + +func (logBuffer *LogBuffer) LoopProcessLogData( + startTreadTime time.Time, + waitForDataFn func() bool, + eachLogDataFn func(logEntry *filer_pb.LogEntry) error) (lastReadTime time.Time, err error) { + // loop through all messages + var bytesBuf *bytes.Buffer + lastReadTime = startTreadTime + defer func() { + if bytesBuf != nil { + logBuffer.ReleaseMemory(bytesBuf) + } + }() + + for { + + if bytesBuf != nil { + logBuffer.ReleaseMemory(bytesBuf) + } + bytesBuf, err = logBuffer.ReadFromBuffer(lastReadTime) + if err == ResumeFromDiskError { + return lastReadTime, ResumeFromDiskError + } + // fmt.Printf("ReadFromBuffer by %v\n", lastReadTime) + if bytesBuf == nil { + if waitForDataFn() { + continue + } else { + return + } + } + + buf := bytesBuf.Bytes() + // fmt.Printf("ReadFromBuffer by %v size %d\n", lastReadTime, len(buf)) + + batchSize := 0 + var startReadTime time.Time + + for pos := 0; pos+4 < len(buf); { + + size := util.BytesToUint32(buf[pos : pos+4]) + if pos+4+int(size) > len(buf) { + err = ResumeError + glog.Errorf("LoopProcessLogData: read buffer %v read %d [%d,%d) from [0,%d)", lastReadTime, batchSize, pos, pos+int(size)+4, len(buf)) + return + } + entryData := buf[pos+4 : pos+4+int(size)] + + logEntry := &filer_pb.LogEntry{} + if err = proto.Unmarshal(entryData, logEntry); err != nil { + glog.Errorf("unexpected unmarshal messaging_pb.Message: %v", err) + pos += 4 + int(size) + continue + } + lastReadTime = time.Unix(0, logEntry.TsNs) + if startReadTime.IsZero() { + startReadTime = lastReadTime + } + + if err = eachLogDataFn(logEntry); err != nil { + return + } + + pos += 4 + int(size) + batchSize++ + } + + // fmt.Printf("sent message ts[%d,%d] size %d\n", startReadTime.UnixNano(), lastReadTime.UnixNano(), batchSize) + } + +} diff --git a/weed/util/log_buffer/sealed_buffer.go b/weed/util/log_buffer/sealed_buffer.go new file mode 100644 index 000000000..d133cf8d3 --- /dev/null +++ b/weed/util/log_buffer/sealed_buffer.go @@ -0,0 +1,62 @@ +package log_buffer + +import ( + "fmt" + "time" +) + +type MemBuffer struct { + buf []byte + size int + startTime time.Time + stopTime time.Time +} + +type SealedBuffers struct { + buffers []*MemBuffer +} + +func newSealedBuffers(size int) *SealedBuffers { + sbs := &SealedBuffers{} + + sbs.buffers = make([]*MemBuffer, size) + for i := 0; i < size; i++ { + sbs.buffers[i] = &MemBuffer{ + buf: make([]byte, BufferSize), + } + } + + return sbs +} + +func (sbs *SealedBuffers) SealBuffer(startTime, stopTime time.Time, buf []byte, pos int) (newBuf []byte) { + oldMemBuffer := sbs.buffers[0] + size := len(sbs.buffers) + for i := 0; i < size-1; i++ { + sbs.buffers[i].buf = sbs.buffers[i+1].buf + sbs.buffers[i].size = sbs.buffers[i+1].size + sbs.buffers[i].startTime = sbs.buffers[i+1].startTime + sbs.buffers[i].stopTime = sbs.buffers[i+1].stopTime + } + sbs.buffers[size-1].buf = buf + sbs.buffers[size-1].size = pos + sbs.buffers[size-1].startTime = startTime + sbs.buffers[size-1].stopTime = stopTime + return oldMemBuffer.buf +} + +func (mb *MemBuffer) locateByTs(lastReadTime time.Time) (pos int) { + lastReadTs := lastReadTime.UnixNano() + for pos < len(mb.buf) { + size, t := readTs(mb.buf, pos) + if t > lastReadTs { + return + } + pos += size + 4 + } + return len(mb.buf) +} + +func (mb *MemBuffer) String() string { + return fmt.Sprintf("[%v,%v] bytes:%d", mb.startTime, mb.stopTime, mb.size) +} diff --git a/weed/util/net_timeout.go b/weed/util/net_timeout.go index b8068e67f..e8075c297 100644 --- a/weed/util/net_timeout.go +++ b/weed/util/net_timeout.go @@ -35,6 +35,7 @@ type Conn struct { net.Conn ReadTimeout time.Duration WriteTimeout time.Duration + isClosed bool } func (c *Conn) Read(b []byte) (count int, e error) { @@ -53,7 +54,8 @@ func (c *Conn) Read(b []byte) (count int, e error) { func (c *Conn) Write(b []byte) (count int, e error) { if c.WriteTimeout != 0 { - err := c.Conn.SetWriteDeadline(time.Now().Add(c.WriteTimeout)) + // minimum 4KB/s + err := c.Conn.SetWriteDeadline(time.Now().Add(c.WriteTimeout * time.Duration(len(b)/40000+1))) if err != nil { return 0, err } @@ -68,7 +70,10 @@ func (c *Conn) Write(b []byte) (count int, e error) { func (c *Conn) Close() error { err := c.Conn.Close() if err == nil { - stats.ConnectionClose() + if !c.isClosed { + stats.ConnectionClose() + c.isClosed = true + } } return err } diff --git a/weed/util/network.go b/weed/util/network.go new file mode 100644 index 000000000..55a123667 --- /dev/null +++ b/weed/util/network.go @@ -0,0 +1,35 @@ +package util + +import ( + "net" + + "github.com/chrislusf/seaweedfs/weed/glog" +) + +func DetectedHostAddress() string { + netInterfaces, err := net.Interfaces() + if err != nil { + glog.V(0).Infof("failed to detect net interfaces: %v", err) + return "" + } + + for _, netInterface := range netInterfaces { + if (netInterface.Flags & net.FlagUp) == 0 { + continue + } + addrs, err := netInterface.Addrs() + if err != nil { + glog.V(0).Infof("get interface addresses: %v", err) + } + + for _, a := range addrs { + if ipNet, ok := a.(*net.IPNet); ok && !ipNet.IP.IsLoopback() { + if ipNet.IP.To4() != nil { + return ipNet.IP.String() + } + } + } + } + + return "localhost" +} diff --git a/weed/util/parse.go b/weed/util/parse.go index 6593d43b6..0955db682 100644 --- a/weed/util/parse.go +++ b/weed/util/parse.go @@ -1,6 +1,7 @@ package util import ( + "fmt" "net/url" "strconv" "strings" @@ -45,3 +46,18 @@ func ParseFilerUrl(entryPath string) (filerServer string, filerPort int64, path path = u.Path return } + +func ParseHostPort(hostPort string) (filerServer string, filerPort int64, err error) { + parts := strings.Split(hostPort, ":") + if len(parts) != 2 { + err = fmt.Errorf("failed to parse %s\n", hostPort) + return + } + + filerPort, err = strconv.ParseInt(parts[1], 10, 64) + if err == nil { + filerServer = parts[0] + } + + return +} diff --git a/weed/util/queue_unbounded.go b/weed/util/queue_unbounded.go new file mode 100644 index 000000000..496b9f844 --- /dev/null +++ b/weed/util/queue_unbounded.go @@ -0,0 +1,45 @@ +package util + +import "sync" + +type UnboundedQueue struct { + outbound []string + outboundLock sync.RWMutex + inbound []string + inboundLock sync.RWMutex +} + +func NewUnboundedQueue() *UnboundedQueue { + q := &UnboundedQueue{} + return q +} + +func (q *UnboundedQueue) EnQueue(items ...string) { + q.inboundLock.Lock() + defer q.inboundLock.Unlock() + + q.inbound = append(q.inbound, items...) + +} + +func (q *UnboundedQueue) Consume(fn func([]string)) { + q.outboundLock.Lock() + defer q.outboundLock.Unlock() + + if len(q.outbound) == 0 { + q.inboundLock.Lock() + inbountLen := len(q.inbound) + if inbountLen > 0 { + t := q.outbound + q.outbound = q.inbound + q.inbound = t + } + q.inboundLock.Unlock() + } + + if len(q.outbound) > 0 { + fn(q.outbound) + q.outbound = q.outbound[:0] + } + +} diff --git a/weed/util/queue_unbounded_test.go b/weed/util/queue_unbounded_test.go new file mode 100644 index 000000000..2d02032cb --- /dev/null +++ b/weed/util/queue_unbounded_test.go @@ -0,0 +1,25 @@ +package util + +import "testing" + +func TestEnqueueAndConsume(t *testing.T) { + + q := NewUnboundedQueue() + + q.EnQueue("1", "2", "3") + + f := func(items []string) { + for _, t := range items { + println(t) + } + println("-----------------------") + } + q.Consume(f) + + q.Consume(f) + + q.EnQueue("4", "5") + q.EnQueue("6", "7") + q.Consume(f) + +} diff --git a/weed/util/retry.go b/weed/util/retry.go new file mode 100644 index 000000000..7b0f2d3c3 --- /dev/null +++ b/weed/util/retry.go @@ -0,0 +1,43 @@ +package util + +import ( + "strings" + "time" + + "github.com/chrislusf/seaweedfs/weed/glog" +) + +var RetryWaitTime = 6 * time.Second + +func Retry(name string, job func() error) (err error) { + waitTime := time.Second + hasErr := false + for waitTime < RetryWaitTime { + err = job() + if err == nil { + if hasErr { + glog.V(0).Infof("retry %s successfully", name) + } + break + } + if strings.Contains(err.Error(), "transport") { + hasErr = true + glog.V(0).Infof("retry %s: err: %v", name, err) + time.Sleep(waitTime) + waitTime += waitTime / 2 + } else { + break + } + } + return err +} + +// return the first non empty string +func Nvl(values ...string) string { + for _, s := range values { + if s != "" { + return s + } + } + return "" +} |