store is sort of working

git-svn-id: https://weed-fs.googlecode.com/svn/trunk@3 282b0af5-e82d-9cf1-ede4-77906d7719d0

6 files changed, 297 insertions, 0 deletions

diff --git a/weed-fs/src/pkg/directory/volume_mapping.go b/weed-fs/src/pkg/directory/volume_mapping.go
new file mode 100644
index 000000000..60fef7d7d
--- /dev/null
+++ b/weed-fs/src/pkg/directory/volume_mapping.go
@@ -0,0 +1,47 @@
+package directory
+
+import (
+	"gob"
+	"os"
+	"log"
+)
+
+type Mapper struct {
+	dir              string
+	fileName         string
+	Virtual2physical map[uint32][]uint32
+}
+
+func NewMapper(dirname string, filename string) (m *Mapper) {
+	m = new(Mapper)
+	m.dir = dirname
+	m.fileName = filename
+	log.Println("Loading virtual to physical:", m.dir, "/", m.fileName)
+	dataFile, e := os.OpenFile(m.dir+string(os.PathSeparator)+m.fileName+".map", os.O_RDONLY, 0644)
+	if e != nil {
+		log.Fatalf("Mapping File Read [ERROR] %s\n", e)
+	} else {
+		m.Virtual2physical = make(map[uint32][]uint32)
+		decoder := gob.NewDecoder(dataFile)
+		decoder.Decode(m.Virtual2physical)
+        dataFile.Close()
+	}
+	return
+}
+func (m *Mapper) Get(vid uint32) []uint32 {
+	return m.Virtual2physical[vid]
+}
+func (m *Mapper) Add(vid uint32, pids ...uint32) {
+	m.Virtual2physical[vid] = append(m.Virtual2physical[vid], pids...)
+}
+func (m *Mapper) Save() {
+	log.Println("Saving virtual to physical:", m.dir, "/", m.fileName)
+	dataFile, e := os.OpenFile(m.dir+string(os.PathSeparator)+m.fileName+".map", os.O_WRONLY, 0644)
+	if e != nil {
+		log.Fatalf("Mapping File Save [ERROR] %s\n", e)
+	}
+	defer dataFile.Close()
+	m.Virtual2physical = make(map[uint32][]uint32)
+	encoder := gob.NewEncoder(dataFile)
+	encoder.Encode(m.Virtual2physical)
+}
diff --git a/weed-fs/src/pkg/store/needle.go b/weed-fs/src/pkg/store/needle.go
new file mode 100644
index 000000000..7f1f51d90
--- /dev/null
+++ b/weed-fs/src/pkg/store/needle.go
@@ -0,0 +1,67 @@
+package store
+
+import (
+    "io"
+	"io/ioutil"
+	"http"
+	"log"
+	"strconv"
+	"strings"
+)
+
+type Needle struct{
+  Cookie uint8 "random number to mitigate brute force lookups"
+  Key uint64 "file id"
+  AlternateKey uint32 "supplemental id"
+  Size uint32 "Data size"
+  Data []byte "The actual file data"
+  Checksum int32 "CRC32 to check integrity"
+  Padding []byte "Aligned to 8 bytes"
+}
+func NewNeedle(r *http.Request)(n *Needle){
+  n = new(Needle)
+  form,fe:=r.MultipartReader()
+  if fe!=nil {
+    log.Fatalf("MultipartReader [ERROR] %s\n", fe)
+  }
+  part,_:=form.NextPart()
+  data,_:=ioutil.ReadAll(part)
+  n.Data = data
+
+  n.ParsePath(r.URL.Path[1:strings.LastIndex(r.URL.Path,".")])
+
+  return
+}
+func (n *Needle) ParsePath(path string){
+  a := strings.Split(path,"_")
+  log.Println("cookie",a[0],"key",a[1],"altKey",a[2])
+  cookie,_ := strconv.Atoi(a[0])
+  n.Cookie = uint8(cookie)
+  n.Key,_ = strconv.Atoui64(a[1])
+  altKey,_ := strconv.Atoui64(a[2])
+  n.AlternateKey = uint32(altKey)
+}
+func (n *Needle) Append(w io.Writer){
+  header := make([]byte,17)
+  header[0] = n.Cookie
+  uint64toBytes(header[1:9],n.Key)
+  uint32toBytes(header[9:13],n.AlternateKey)
+  n.Size = uint32(len(n.Data))
+  uint32toBytes(header[13:17],n.Size)
+  w.Write(header)
+  w.Write(n.Data)
+  rest := 8-((n.Size+17+4)%8)
+  uint32toBytes(header[0:4],uint32(n.Checksum))
+  w.Write(header[0:rest+4])
+}
+func (n *Needle) Read(r io.Reader, size uint32){
+  bytes := make([]byte,size+17+4)
+  r.Read(bytes)
+  n.Cookie = bytes[0]
+  n.Key = bytesToUint64(bytes[1:9])
+  n.AlternateKey = bytesToUint32(bytes[9:13])
+  n.Size = bytesToUint32(bytes[13:17])
+  n.Data = bytes[17:17+size]
+  n.Checksum = int32(bytesToUint32(bytes[17+size:17+size+4]))
+}
+
diff --git a/weed-fs/src/pkg/store/needle_map.go b/weed-fs/src/pkg/store/needle_map.go
new file mode 100644
index 000000000..df50dfb0e
--- /dev/null
+++ b/weed-fs/src/pkg/store/needle_map.go
@@ -0,0 +1,53 @@
+package store
+
+import (
+    "os"
+)
+
+type NeedleKey struct{
+  Key uint64 "file id"
+  AlternateKey uint32 "supplemental id"
+}
+func (k *NeedleKey) String() string {
+  var tmp [12]byte
+  for i :=uint(0);i<8;i++{
+    tmp[i] = byte(k.Key >> (8*i));
+  }
+  for i :=uint(0);i<4;i++{
+    tmp[i+8] = byte(k.AlternateKey >> (8*i));
+  }
+  return string(tmp[:])
+}
+
+type NeedleValue struct{
+  Offset uint32 "Volume offset" //since aligned to 8 bytes, range is 4G*8=32G
+  Size uint32 "Size of the data portion"
+}
+
+type NeedleMap struct{
+  m map[string]*NeedleValue //mapping NeedleKey(Key,AlternateKey) to NeedleValue
+}
+func NewNeedleMap() (nm *NeedleMap){
+  nm = new(NeedleMap)
+  nm.m = make(map[string]*NeedleValue)
+  return
+}
+func (nm *NeedleMap) load(file *os.File){
+}
+func makeKey(key uint64, altKey uint32) string {
+  var tmp [12]byte
+  for i :=uint(0);i<8;i++{
+    tmp[i] = byte(key >> (8*i));
+  }
+  for i :=uint(0);i<4;i++{
+    tmp[i+8] = byte(altKey >> (8*i));
+  }
+  return string(tmp[:])
+}
+func (nm *NeedleMap) put(key uint64, altKey uint32, offset uint32, size uint32){
+  nm.m[makeKey(key,altKey)] = &NeedleValue{Offset:offset, Size:size}
+}
+func (nm *NeedleMap) get(key uint64, altKey uint32) (element *NeedleValue, ok bool){
+  element, ok = nm.m[makeKey(key,altKey)]
+  return
+}
diff --git a/weed-fs/src/pkg/store/store.go b/weed-fs/src/pkg/store/store.go
new file mode 100644
index 000000000..a568fb199
--- /dev/null
+++ b/weed-fs/src/pkg/store/store.go
@@ -0,0 +1,39 @@
+package store
+
+import (
+    "log"
+    "strconv"
+)
+type Store struct{
+  volumes []*Volume
+  dir string
+  
+  freeVolumeChannel chan int
+}
+func NewStore(dirname string, count int) (s *Store){
+  s = new(Store)
+  s.dir = dirname
+  s.volumes = make([]*Volume,count)
+  s.freeVolumeChannel = make(chan int, count)
+  for i:=0;i<count;i++{
+    s.volumes[i] = NewVolume(s.dir, strconv.Itob(i,16))
+    s.freeVolumeChannel <- i
+  }
+  log.Println("Store started on dir:", dirname, "with", count,"volumes");
+  return
+}
+func (s *Store)Close(){
+  close(s.freeVolumeChannel)
+  for _, v := range s.volumes{
+    v.Close()
+  }
+}
+func (s *Store)Write(n *Needle)(int){
+  i := <- s.freeVolumeChannel
+  s.volumes[i].write(n)
+  s.freeVolumeChannel <- i
+  return i
+}
+func (s *Store)Read(i int, n *Needle){
+  s.volumes[i].read(n)
+}
diff --git a/weed-fs/src/pkg/store/util.go b/weed-fs/src/pkg/store/util.go
new file mode 100644
index 000000000..258dec482
--- /dev/null
+++ b/weed-fs/src/pkg/store/util.go
@@ -0,0 +1,28 @@
+package store
+
+func bytesToUint64(b []byte)(v uint64){
+  for i :=uint(7);i>0;i-- {
+    v += uint64(b[i])
+    v <<= 8
+  }
+  v+=uint64(b[0])
+  return
+}
+func bytesToUint32(b []byte)(v uint32){
+  for i :=uint(3);i>0;i-- {
+    v += uint32(b[i])
+    v <<= 8
+  }
+  v+=uint32(b[0])
+  return
+}
+func uint64toBytes(b []byte, v uint64){
+  for i :=uint(0);i<8;i++ {
+    b[i] = byte(v>>(i*8))
+  }
+}
+func uint32toBytes(b []byte, v uint32){
+  for i :=uint(0);i<4;i++ {
+    b[i] = byte(v>>(i*8))
+  }
+}
diff --git a/weed-fs/src/pkg/store/volume.go b/weed-fs/src/pkg/store/volume.go
new file mode 100644
index 000000000..33b010e62
--- /dev/null
+++ b/weed-fs/src/pkg/store/volume.go
@@ -0,0 +1,63 @@
+package store
+
+import (
+	"os"
+	"log"
+)
+
+type Volume struct {
+	dir                 string
+	fileName            string
+	dataFile, indexFile *os.File
+	nm                  *NeedleMap
+
+	accessChannel chan int
+}
+
+func NewVolume(dirname string, filename string) (v *Volume) {
+	var e os.Error
+	v = new(Volume)
+	v.dir = dirname
+	v.fileName = filename
+	log.Println("file", v.dir, "/", v.fileName)
+	v.dataFile, e = os.OpenFile(v.dir+string(os.PathSeparator)+v.fileName+".dat", os.O_RDWR|os.O_CREATE, 0644)
+	if e != nil {
+		log.Fatalf("New Volume [ERROR] %s\n", e)
+	}
+	v.indexFile, e = os.OpenFile(v.dir+string(os.PathSeparator)+v.fileName+".idx", os.O_RDWR|os.O_CREATE, 0644)
+	if e != nil {
+		log.Fatalf("New Volume [ERROR] %s\n", e)
+	}
+	v.nm = NewNeedleMap()
+	v.nm.load(v.indexFile)
+
+	v.accessChannel = make(chan int, 1)
+	v.accessChannel <- 0
+
+	return
+}
+func (v *Volume) Close() {
+	close(v.accessChannel)
+	v.dataFile.Close()
+	v.indexFile.Close()
+}
+
+func (v *Volume) write(n *Needle) {
+	counter := <-v.accessChannel
+	offset, _ := v.dataFile.Seek(0, 2)
+	n.Append(v.dataFile)
+	nv, ok := v.nm.get(n.Key, n.AlternateKey)
+	if !ok || int64(nv.Offset)*8 < offset {
+		v.nm.put(n.Key, n.AlternateKey, uint32(offset/8), n.Size)
+	}
+	v.accessChannel <- counter + 1
+}
+func (v *Volume) read(n *Needle) {
+	counter := <-v.accessChannel
+	nv, ok := v.nm.get(n.Key, n.AlternateKey)
+	if ok && nv.Offset > 0 {
+		v.dataFile.Seek(int64(nv.Offset)*8, 0)
+		n.Read(v.dataFile, nv.Size)
+	}
+	v.accessChannel <- counter + 1
+}