seaweedfs/go/filer/embedded_filer/directory_in_map.go
2015-01-13 17:04:41 -08:00

260 lines
6.3 KiB
Go

package embedded_filer
import (
"bufio"
"fmt"
"io"
"os"
"path/filepath"
"strconv"
"strings"
"sync"
"github.com/chrislusf/weed-fs/go/filer"
"github.com/chrislusf/weed-fs/go/util"
)
var writeLock sync.Mutex //serialize changes to dir.log
type DirectoryEntryInMap struct {
Name string
Parent *DirectoryEntryInMap
SubDirectories map[string]*DirectoryEntryInMap
Id filer.DirectoryId
}
type DirectoryManagerInMap struct {
Root *DirectoryEntryInMap
max filer.DirectoryId
logFile *os.File
isLoading bool
}
func (dm *DirectoryManagerInMap) NewDirectoryEntryInMap(parent *DirectoryEntryInMap, name string) (d *DirectoryEntryInMap, err error) {
writeLock.Lock()
defer writeLock.Unlock()
d = &DirectoryEntryInMap{Name: name, Parent: parent, SubDirectories: make(map[string]*DirectoryEntryInMap)}
parts := make([]string, 0)
for p := d; p != nil && p.Name != ""; p = p.Parent {
parts = append(parts, p.Name)
}
n := len(parts)
if n <= 0 {
return nil, fmt.Errorf("Failed to create folder %s/%s", parent.Name, name)
}
for i := 0; i < n/2; i++ {
parts[i], parts[n-1-i] = parts[n-1-i], parts[i]
}
dm.max++
d.Id = dm.max
dm.log("add", "/"+strings.Join(parts, "/"), strconv.Itoa(int(d.Id)))
return d, nil
}
func (dm *DirectoryManagerInMap) log(words ...string) {
if !dm.isLoading {
dm.logFile.WriteString(strings.Join(words, "\t") + "\n")
}
}
func NewDirectoryManagerInMap(dirLogFile string) (dm *DirectoryManagerInMap, err error) {
dm = &DirectoryManagerInMap{}
//dm.Root do not use NewDirectoryEntryInMap, since dm.max will be changed
dm.Root = &DirectoryEntryInMap{SubDirectories: make(map[string]*DirectoryEntryInMap)}
if dm.logFile, err = os.OpenFile(dirLogFile, os.O_RDWR|os.O_CREATE, 0644); err != nil {
return nil, fmt.Errorf("cannot write directory log file %s.idx: %v", dirLogFile, err)
}
return dm, dm.load()
}
func (dm *DirectoryManagerInMap) processEachLine(line string) error {
if strings.HasPrefix(line, "#") {
return nil
}
if line == "" {
return nil
}
parts := strings.Split(line, "\t")
if len(parts) == 0 {
return nil
}
switch parts[0] {
case "add":
v, pe := strconv.Atoi(parts[2])
if pe != nil {
return pe
}
if e := dm.loadDirectory(parts[1], filer.DirectoryId(v)); e != nil {
return e
}
case "mov":
newName := ""
if len(parts) >= 4 {
newName = parts[3]
}
if e := dm.MoveUnderDirectory(parts[1], parts[2], newName); e != nil {
return e
}
case "del":
if e := dm.DeleteDirectory(parts[1]); e != nil {
return e
}
default:
fmt.Printf("line %s has %s!\n", line, parts[0])
return nil
}
return nil
}
func (dm *DirectoryManagerInMap) load() error {
dm.max = 0
lines := bufio.NewReader(dm.logFile)
dm.isLoading = true
defer func() { dm.isLoading = false }()
for {
line, err := util.Readln(lines)
if err != nil && err != io.EOF {
return err
}
if pe := dm.processEachLine(string(line)); pe != nil {
return pe
}
if err == io.EOF {
return nil
}
}
}
func (dm *DirectoryManagerInMap) findDirectory(dirPath string) (*DirectoryEntryInMap, error) {
if dirPath == "" {
return dm.Root, nil
}
dirPath = filepath.Clean(dirPath)
if dirPath == "/" {
return dm.Root, nil
}
parts := strings.Split(dirPath, "/")
dir := dm.Root
for i := 1; i < len(parts); i++ {
if sub, ok := dir.SubDirectories[parts[i]]; ok {
dir = sub
} else {
return dm.Root, fmt.Errorf("Directory %s Not Found", dirPath)
}
}
return dir, nil
}
func (dm *DirectoryManagerInMap) FindDirectory(dirPath string) (filer.DirectoryId, error) {
d, e := dm.findDirectory(dirPath)
if e == nil {
return d.Id, nil
}
return dm.Root.Id, e
}
func (dm *DirectoryManagerInMap) loadDirectory(dirPath string, dirId filer.DirectoryId) error {
dirPath = filepath.Clean(dirPath)
if dirPath == "/" {
return nil
}
parts := strings.Split(dirPath, "/")
dir := dm.Root
for i := 1; i < len(parts); i++ {
sub, ok := dir.SubDirectories[parts[i]]
if !ok {
if i != len(parts)-1 {
return fmt.Errorf("%s should be created after parent %s!", dirPath, parts[i])
}
var err error
sub, err = dm.NewDirectoryEntryInMap(dir, parts[i])
if err != nil {
return err
}
if sub.Id != dirId {
return fmt.Errorf("%s should be have id %v instead of %v!", dirPath, sub.Id, dirId)
}
dir.SubDirectories[parts[i]] = sub
}
dir = sub
}
return nil
}
func (dm *DirectoryManagerInMap) makeDirectory(dirPath string) (dir *DirectoryEntryInMap, created bool) {
dirPath = filepath.Clean(dirPath)
if dirPath == "/" {
return dm.Root, false
}
parts := strings.Split(dirPath, "/")
dir = dm.Root
for i := 1; i < len(parts); i++ {
sub, ok := dir.SubDirectories[parts[i]]
if !ok {
var err error
sub, err = dm.NewDirectoryEntryInMap(dir, parts[i])
if err != nil {
return nil, false
}
dir.SubDirectories[parts[i]] = sub
created = true
}
dir = sub
}
return dir, created
}
func (dm *DirectoryManagerInMap) MakeDirectory(dirPath string) (filer.DirectoryId, error) {
dir, _ := dm.makeDirectory(dirPath)
return dir.Id, nil
}
func (dm *DirectoryManagerInMap) MoveUnderDirectory(oldDirPath string, newParentDirPath string, newName string) error {
writeLock.Lock()
defer writeLock.Unlock()
oldDir, oe := dm.findDirectory(oldDirPath)
if oe != nil {
return oe
}
parentDir, pe := dm.findDirectory(newParentDirPath)
if pe != nil {
return pe
}
dm.log("mov", oldDirPath, newParentDirPath, newName)
delete(oldDir.Parent.SubDirectories, oldDir.Name)
if newName == "" {
newName = oldDir.Name
}
parentDir.SubDirectories[newName] = oldDir
oldDir.Name = newName
oldDir.Parent = parentDir
return nil
}
func (dm *DirectoryManagerInMap) ListDirectories(dirPath string) (dirNames []filer.DirectoryEntry, err error) {
d, e := dm.findDirectory(dirPath)
if e != nil {
return dirNames, e
}
for k, v := range d.SubDirectories {
dirNames = append(dirNames, filer.DirectoryEntry{Name: k, Id: v.Id})
}
return dirNames, nil
}
func (dm *DirectoryManagerInMap) DeleteDirectory(dirPath string) error {
writeLock.Lock()
defer writeLock.Unlock()
if dirPath == "/" {
return fmt.Errorf("Can not delete %s", dirPath)
}
d, e := dm.findDirectory(dirPath)
if e != nil {
return e
}
if len(d.SubDirectories) != 0 {
return fmt.Errorf("dir %s still has sub directories", dirPath)
}
delete(d.Parent.SubDirectories, d.Name)
d.Parent = nil
dm.log("del", dirPath)
return nil
}