adjusting for types
This commit is contained in:
@@ -5,7 +5,7 @@ import (
|
|||||||
"encoding/json"
|
"encoding/json"
|
||||||
"log"
|
"log"
|
||||||
"net/http"
|
"net/http"
|
||||||
"pkg/topology"
|
"pkg/storage"
|
||||||
"strconv"
|
"strconv"
|
||||||
"strings"
|
"strings"
|
||||||
)
|
)
|
||||||
@@ -39,7 +39,7 @@ func dirLookupHandler(w http.ResponseWriter, r *http.Request) {
|
|||||||
vid = vid[0:commaSep]
|
vid = vid[0:commaSep]
|
||||||
}
|
}
|
||||||
volumeId, _ := strconv.ParseUint(vid, 10, 64)
|
volumeId, _ := strconv.ParseUint(vid, 10, 64)
|
||||||
machine, e := mapper.Get(topology.VolumeId(volumeId))
|
machine, e := mapper.Get(storage.VolumeId(volumeId))
|
||||||
if e == nil {
|
if e == nil {
|
||||||
writeJson(w, r, machine.Server)
|
writeJson(w, r, machine.Server)
|
||||||
} else {
|
} else {
|
||||||
@@ -60,7 +60,7 @@ func dirAssignHandler(w http.ResponseWriter, r *http.Request) {
|
|||||||
func dirJoinHandler(w http.ResponseWriter, r *http.Request) {
|
func dirJoinHandler(w http.ResponseWriter, r *http.Request) {
|
||||||
s := r.RemoteAddr[0:strings.Index(r.RemoteAddr, ":")+1] + r.FormValue("port")
|
s := r.RemoteAddr[0:strings.Index(r.RemoteAddr, ":")+1] + r.FormValue("port")
|
||||||
publicUrl := r.FormValue("publicUrl")
|
publicUrl := r.FormValue("publicUrl")
|
||||||
volumes := new([]topology.VolumeInfo)
|
volumes := new([]storage.VolumeInfo)
|
||||||
json.Unmarshal([]byte(r.FormValue("volumes")), volumes)
|
json.Unmarshal([]byte(r.FormValue("volumes")), volumes)
|
||||||
if *IsDebug {
|
if *IsDebug {
|
||||||
log.Println(s, "volumes", r.FormValue("volumes"))
|
log.Println(s, "volumes", r.FormValue("volumes"))
|
||||||
|
|||||||
@@ -4,19 +4,18 @@ import (
|
|||||||
"encoding/hex"
|
"encoding/hex"
|
||||||
"log"
|
"log"
|
||||||
"pkg/storage"
|
"pkg/storage"
|
||||||
"pkg/topology"
|
|
||||||
"strconv"
|
"strconv"
|
||||||
"strings"
|
"strings"
|
||||||
"pkg/util"
|
"pkg/util"
|
||||||
)
|
)
|
||||||
|
|
||||||
type FileId struct {
|
type FileId struct {
|
||||||
VolumeId topology.VolumeId
|
VolumeId storage.VolumeId
|
||||||
Key uint64
|
Key uint64
|
||||||
Hashcode uint32
|
Hashcode uint32
|
||||||
}
|
}
|
||||||
|
|
||||||
func NewFileId(VolumeId topology.VolumeId, Key uint64, Hashcode uint32) *FileId {
|
func NewFileId(VolumeId storage.VolumeId, Key uint64, Hashcode uint32) *FileId {
|
||||||
return &FileId{VolumeId: VolumeId, Key: Key, Hashcode: Hashcode}
|
return &FileId{VolumeId: VolumeId, Key: Key, Hashcode: Hashcode}
|
||||||
}
|
}
|
||||||
func ParseFileId(fid string) *FileId{
|
func ParseFileId(fid string) *FileId{
|
||||||
@@ -28,7 +27,7 @@ func ParseFileId(fid string) *FileId{
|
|||||||
vid_string, key_hash_string := a[0], a[1]
|
vid_string, key_hash_string := a[0], a[1]
|
||||||
vid, _ := strconv.ParseUint(vid_string, 10, 64)
|
vid, _ := strconv.ParseUint(vid_string, 10, 64)
|
||||||
key, hash := storage.ParseKeyHash(key_hash_string)
|
key, hash := storage.ParseKeyHash(key_hash_string)
|
||||||
return &FileId{VolumeId: topology.VolumeId(vid), Key: key, Hashcode: hash}
|
return &FileId{VolumeId: storage.VolumeId(vid), Key: key, Hashcode: hash}
|
||||||
}
|
}
|
||||||
func (n *FileId) String() string {
|
func (n *FileId) String() string {
|
||||||
bytes := make([]byte, 12)
|
bytes := make([]byte, 12)
|
||||||
|
|||||||
@@ -7,7 +7,7 @@ import (
|
|||||||
"math/rand"
|
"math/rand"
|
||||||
"os"
|
"os"
|
||||||
"path"
|
"path"
|
||||||
"pkg/topology"
|
"pkg/storage"
|
||||||
"strconv"
|
"strconv"
|
||||||
"sync"
|
"sync"
|
||||||
)
|
)
|
||||||
@@ -22,7 +22,7 @@ type MachineInfo struct {
|
|||||||
}
|
}
|
||||||
type Machine struct {
|
type Machine struct {
|
||||||
Server MachineInfo
|
Server MachineInfo
|
||||||
Volumes []topology.VolumeInfo
|
Volumes []storage.VolumeInfo
|
||||||
}
|
}
|
||||||
|
|
||||||
type Mapper struct {
|
type Mapper struct {
|
||||||
@@ -32,8 +32,8 @@ type Mapper struct {
|
|||||||
volumeLock sync.Mutex
|
volumeLock sync.Mutex
|
||||||
sequenceLock sync.Mutex
|
sequenceLock sync.Mutex
|
||||||
Machines []*Machine
|
Machines []*Machine
|
||||||
vid2machineId map[topology.VolumeId]int //machineId is +1 of the index of []*Machine, to detect not found entries
|
vid2machineId map[storage.VolumeId]int //machineId is +1 of the index of []*Machine, to detect not found entries
|
||||||
Writers []topology.VolumeId // transient array of Writers volume id
|
Writers []storage.VolumeId // transient array of Writers volume id
|
||||||
|
|
||||||
FileIdSequence uint64
|
FileIdSequence uint64
|
||||||
fileIdCounter uint64
|
fileIdCounter uint64
|
||||||
@@ -41,15 +41,15 @@ type Mapper struct {
|
|||||||
volumeSizeLimit uint64
|
volumeSizeLimit uint64
|
||||||
}
|
}
|
||||||
|
|
||||||
func NewMachine(server, publicUrl string, volumes []topology.VolumeInfo) *Machine {
|
func NewMachine(server, publicUrl string, volumes []storage.VolumeInfo) *Machine {
|
||||||
return &Machine{Server: MachineInfo{Url: server, PublicUrl: publicUrl}, Volumes: volumes}
|
return &Machine{Server: MachineInfo{Url: server, PublicUrl: publicUrl}, Volumes: volumes}
|
||||||
}
|
}
|
||||||
|
|
||||||
func NewMapper(dirname string, filename string, volumeSizeLimit uint64) (m *Mapper) {
|
func NewMapper(dirname string, filename string, volumeSizeLimit uint64) (m *Mapper) {
|
||||||
m = &Mapper{dir: dirname, fileName: filename}
|
m = &Mapper{dir: dirname, fileName: filename}
|
||||||
m.vid2machineId = make(map[topology.VolumeId]int)
|
m.vid2machineId = make(map[storage.VolumeId]int)
|
||||||
m.volumeSizeLimit = volumeSizeLimit
|
m.volumeSizeLimit = volumeSizeLimit
|
||||||
m.Writers = *new([]topology.VolumeId)
|
m.Writers = *new([]storage.VolumeId)
|
||||||
m.Machines = *new([]*Machine)
|
m.Machines = *new([]*Machine)
|
||||||
|
|
||||||
seqFile, se := os.OpenFile(path.Join(m.dir, m.fileName+".seq"), os.O_RDONLY, 0644)
|
seqFile, se := os.OpenFile(path.Join(m.dir, m.fileName+".seq"), os.O_RDONLY, 0644)
|
||||||
@@ -102,7 +102,7 @@ func (m *Mapper) NextFileId(c string) (uint64,int) {
|
|||||||
m.fileIdCounter = m.fileIdCounter - count
|
m.fileIdCounter = m.fileIdCounter - count
|
||||||
return m.FileIdSequence - m.fileIdCounter, int(count)
|
return m.FileIdSequence - m.fileIdCounter, int(count)
|
||||||
}
|
}
|
||||||
func (m *Mapper) Get(vid topology.VolumeId) (*Machine, error) {
|
func (m *Mapper) Get(vid storage.VolumeId) (*Machine, error) {
|
||||||
machineId := m.vid2machineId[vid]
|
machineId := m.vid2machineId[vid]
|
||||||
if machineId <= 0 {
|
if machineId <= 0 {
|
||||||
return nil, errors.New("invalid volume id " + strconv.FormatUint(uint64(vid), 10))
|
return nil, errors.New("invalid volume id " + strconv.FormatUint(uint64(vid), 10))
|
||||||
@@ -134,7 +134,7 @@ func (m *Mapper) Add(machine Machine) {
|
|||||||
m.vid2machineId[v.Id] = machineId + 1 //use base 1 indexed, to detect not found cases
|
m.vid2machineId[v.Id] = machineId + 1 //use base 1 indexed, to detect not found cases
|
||||||
}
|
}
|
||||||
//setting Writers, copy-on-write because of possible updating, this needs some future work!
|
//setting Writers, copy-on-write because of possible updating, this needs some future work!
|
||||||
var writers []topology.VolumeId
|
var writers []storage.VolumeId
|
||||||
for _, machine_entry := range m.Machines {
|
for _, machine_entry := range m.Machines {
|
||||||
for _, v := range machine_entry.Volumes {
|
for _, v := range machine_entry.Volumes {
|
||||||
if uint64(v.Size) < m.volumeSizeLimit {
|
if uint64(v.Size) < m.volumeSizeLimit {
|
||||||
|
|||||||
@@ -7,7 +7,6 @@ import (
|
|||||||
"net/url"
|
"net/url"
|
||||||
"strconv"
|
"strconv"
|
||||||
"strings"
|
"strings"
|
||||||
"pkg/topology"
|
|
||||||
"pkg/util"
|
"pkg/util"
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -60,20 +59,20 @@ func (s *Store) addVolume(vid VolumeId) error {
|
|||||||
s.volumes[vid] = NewVolume(s.dir, vid)
|
s.volumes[vid] = NewVolume(s.dir, vid)
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
func (s *Store) Status() *[]*topology.VolumeInfo {
|
func (s *Store) Status() *[]*VolumeInfo {
|
||||||
stats := new([]*topology.VolumeInfo)
|
stats := new([]*VolumeInfo)
|
||||||
for k, v := range s.volumes {
|
for k, v := range s.volumes {
|
||||||
s := new(topology.VolumeInfo)
|
s := new(VolumeInfo)
|
||||||
s.Id, s.Size = topology.VolumeId(k), v.Size()
|
s.Id, s.Size = VolumeId(k), v.Size()
|
||||||
*stats = append(*stats, s)
|
*stats = append(*stats, s)
|
||||||
}
|
}
|
||||||
return stats
|
return stats
|
||||||
}
|
}
|
||||||
func (s *Store) Join(mserver string) {
|
func (s *Store) Join(mserver string) {
|
||||||
stats := new([]*topology.VolumeInfo)
|
stats := new([]*VolumeInfo)
|
||||||
for k, v := range s.volumes {
|
for k, v := range s.volumes {
|
||||||
s := new(topology.VolumeInfo)
|
s := new(VolumeInfo)
|
||||||
s.Id, s.Size = topology.VolumeId(k), v.Size()
|
s.Id, s.Size = VolumeId(k), v.Size()
|
||||||
*stats = append(*stats, s)
|
*stats = append(*stats, s)
|
||||||
}
|
}
|
||||||
bytes, _ := json.Marshal(stats)
|
bytes, _ := json.Marshal(stats)
|
||||||
|
|||||||
@@ -14,6 +14,10 @@ const (
|
|||||||
)
|
)
|
||||||
|
|
||||||
type VolumeId uint32
|
type VolumeId uint32
|
||||||
|
type VolumeInfo struct {
|
||||||
|
Id VolumeId
|
||||||
|
Size int64
|
||||||
|
}
|
||||||
func NewVolumeId(vid string) (VolumeId,error) {
|
func NewVolumeId(vid string) (VolumeId,error) {
|
||||||
volumeId, err := strconv.ParseUint(vid, 10, 64)
|
volumeId, err := strconv.ParseUint(vid, 10, 64)
|
||||||
return VolumeId(volumeId), err
|
return VolumeId(volumeId), err
|
||||||
|
|||||||
@@ -1,15 +1,12 @@
|
|||||||
package topology
|
package topology
|
||||||
|
|
||||||
import ()
|
import (
|
||||||
|
"pkg/storage"
|
||||||
|
)
|
||||||
|
|
||||||
type NodeId uint32
|
type NodeId uint32
|
||||||
type VolumeId uint32
|
|
||||||
type VolumeInfo struct {
|
|
||||||
Id VolumeId
|
|
||||||
Size int64
|
|
||||||
}
|
|
||||||
type Node struct {
|
type Node struct {
|
||||||
volumes map[VolumeId]VolumeInfo
|
volumes map[storage.VolumeId]storage.VolumeInfo
|
||||||
volumeLimit int
|
volumeLimit int
|
||||||
Ip string
|
Ip string
|
||||||
Port int
|
Port int
|
||||||
|
|||||||
Reference in New Issue
Block a user