From 35f718bf339fda7c9af0ef28e3a39f5adb2d60d8 Mon Sep 17 00:00:00 2001 From: Todd Gruben Date: Mon, 16 Jun 2014 16:00:59 -0500 Subject: [PATCH] first pass at auto frament allocation --- core/etcd.go | 65 ++++++++++++++++++++++++++++++++++++++++++++++++++ db/topology.go | 46 +++++++++++++++++++---------------- 2 files changed, 91 insertions(+), 20 deletions(-) diff --git a/core/etcd.go b/core/etcd.go index adb0b9c13..63581d650 100644 --- a/core/etcd.go +++ b/core/etcd.go @@ -2,10 +2,12 @@ package core import ( "errors" + "fmt" "log" "pilosa/config" "pilosa/db" "pilosa/util" + "sort" "strconv" "strings" "sync" @@ -63,6 +65,69 @@ func NewTopologyMapper(service *Service, namespace string) *TopologyMapper { return &TopologyMapper{service, namespace} } +type Pair struct { + Key string + Value int +} + +// A slice of Pairs that implements sort.Interface to sort by Value. +type PairList []Pair + +func (p PairList) Swap(i, j int) { p[i], p[j] = p[j], p[i] } +func (p PairList) Len() int { return len(p) } +func (p PairList) Less(i, j int) bool { return p[i].Value < p[j].Value } + +func getLightestProcess(m map[string]int) Pair { + + p := make(PairList, len(m)) + i := 0 + for k, v := range m { + p[i] = Pair{k, v} + } + sort.Sort(p) + return p[0] +} + +func (self *TopologyMapper) AllocateFragment(db, frame string, slice_int int) error { + //get Lock to create the fragment + ttl := uint64(300) //secs to hold the lock + lock_key := fmt.Sprintf("%s-%s-%d", db, frame, slice_int) + _, err := self.service.Etcd.RawCompareAndSwap(lock_key, "0", ttl, "", 0) + + if err == nil { + //figure out least loaded process..possibly check max process + //to create the node, just write off the items to etcd and the watch should spawn + //be nice if something would notify perhaps queue + m := make(map[string]int) + for _, dbs := range self.service.Cluster.GetDatabases() { + for _, fsi := range dbs.GetFramSliceIntersects() { + for _, fragment := range fsi.GetFragments() { + process := fragment.GetProcess().Id().String() + i := m[process] + i++ + m[process] = i + + } + + } + + } + p := getLightestProcess(m) + + // PUT -d "value=5cb315c3-6e1d-4218-89b7-943d1dba985b" http://etcd0:4001/v2/keys/pilosa/0/db/29/frame/d/slice/5/fragment/a2b632fc4001b817/proces + //so i need db, frame, slice , fragment_id + fuid := util.SUUID_to_Hex(util.Id()) + fragment_key := fmt.Sprintf("%s/db/%d/frame/%s/slice/%d/fragment/%s/process", self.namespace, db, frame, slice_int, fuid) + process_guid := p.Key + // need to check value to see how many we have left + _, err = self.service.Etcd.Set(fragment_key, process_guid, 0) + log.Println("fragment created: %s(%s)", fragment_key, process_guid) + + } + return err + +} + func (self *TopologyMapper) handlenode(node *etcd.Node) error { key := node.Key[len(self.namespace)+1:] bits := strings.Split(key, "/") diff --git a/db/topology.go b/db/topology.go index d91bf21de..dd83ea188 100644 --- a/db/topology.go +++ b/db/topology.go @@ -39,12 +39,6 @@ func (self *Process) Id() uuid.UUID { return *self.id } -func (self *Process) SetId(id uuid.UUID) { - self.mutex.Lock() - defer self.mutex.Unlock() - self.id = &id -} - func (self *Process) Host() string { self.mutex.Lock() defer self.mutex.Unlock() @@ -117,6 +111,10 @@ func NewCluster() *Cluster { cluster.databases = make(map[string]*Database) return &cluster } +func (self *Cluster) GetDatabases() map[string]*Database { + return self.databases + +} /////////// DATABASES //////////////////////////////////////////////////////////////////// @@ -129,6 +127,10 @@ type Database struct { mutex sync.Mutex } +func (self *Database) GetFramSliceIntersects() []*FrameSliceIntersect { + return self.frame_slice_intersects +} + // Add a database to a cluster func (c *Cluster) addDatabase(name string) *Database { database := Database{Name: name} @@ -181,7 +183,6 @@ func (d *Database) getFrame(name string) (*Frame, error) { return frame, nil } } - log.Println("Missing frame:", d.Name, name) return nil, FrameDoesNotExistError } @@ -271,8 +272,12 @@ func (d *Database) GetFrameSliceIntersect(frame *Frame, slice *Slice) (*FrameSli return nil, FrameSliceIntersectDoesNotExistError } -func (fsi *FrameSliceIntersect) GetFragment(fragment_id util.SUUID) (*Fragment, error) { - for _, fragment := range fsi.fragments { +func (self *FrameSliceIntersect) GetFragments() []*Fragment { + return self.fragments +} + +func (self *FrameSliceIntersect) GetFragment(fragment_id util.SUUID) (*Fragment, error) { + for _, fragment := range self.fragments { if fragment.id == fragment_id { return fragment, nil } @@ -280,9 +285,9 @@ func (fsi *FrameSliceIntersect) GetFragment(fragment_id util.SUUID) (*Fragment, return nil, FragmentDoesNotExistError } -func (fsi *FrameSliceIntersect) AddFragment(fragment *Fragment) { - fsi.fragments = append(fsi.fragments, fragment) - fsi.hashring.Add(util.SUUID_to_Hex(fragment.id)) +func (self *FrameSliceIntersect) AddFragment(fragment *Fragment) { + self.fragments = append(self.fragments, fragment) + self.hashring.Add(util.SUUID_to_Hex(fragment.id)) } ///////// FRAGMENTS //////////////////////////////////////////////////////////////////////// @@ -293,20 +298,20 @@ type Fragment struct { process *Process } -func (f *Fragment) GetId() util.SUUID { - return f.id +func (self *Fragment) GetId() util.SUUID { + return self.id } -func (f *Fragment) GetProcess() *Process { - return f.process +func (self *Fragment) GetProcess() *Process { + return self.process } -func (f *Fragment) GetProcessId() *uuid.UUID { - return f.process.id +func (self *Fragment) GetProcessId() *uuid.UUID { + return self.process.id } -func (f *Fragment) GetLocation() *Location { - return &Location{f.process.id, f.id} +func (self *Fragment) GetLocation() *Location { + return &Location{self.process.id, self.id} } // rename this one @@ -340,6 +345,7 @@ func (d *Database) GetFragmentForBitmap(slice *Slice, bitmap *Bitmap) (*Fragment func (d *Database) GetFragmentById(fragment_id *uuid.UUID) *Fragment { } */ + func (d *Database) getFragment(frame *Frame, slice *Slice, fragment_id util.SUUID) (*Fragment, error) { fsi, err := d.GetFrameSliceIntersect(frame, slice) if err != nil {