From 531cf5f858805f5358c209fae6ad7b5eb751ef3f Mon Sep 17 00:00:00 2001 From: Todd Gruben Date: Tue, 3 Dec 2013 18:39:10 -0600 Subject: [PATCH] prevent dup ids in non-present properties --- commands/pilosa-nexter/nexter.go | 75 +++++++++++++++++++++++--------- 1 file changed, 54 insertions(+), 21 deletions(-) diff --git a/commands/pilosa-nexter/nexter.go b/commands/pilosa-nexter/nexter.go index e06673f83..52abc33bb 100644 --- a/commands/pilosa-nexter/nexter.go +++ b/commands/pilosa-nexter/nexter.go @@ -1,43 +1,57 @@ package main import ( + "encoding/json" "github.com/coreos/go-etcd/etcd" "log" "net/http" - "encoding/json" - "strings" "strconv" + "strings" "time" ) const blocksize = 64 -type Req struct { - id int +type Req interface{} + +type IncReq struct { + id int ret chan uint64 } +type DelReq struct { + id int +} + type Nexter struct { - reqchan chan *Req + reqchan chan Req + done chan bool } func (self *Nexter) countloop(ch chan uint64, id int, client *etcd.Client) { var start uint64 var end uint64 + path := "nexter/" + strconv.Itoa(id) + for { - path := "nexter/" + strconv.Itoa(id) - node, err := client.Get(path, false) + node, err := client.Get(path, false,false) if err != nil { ee, ok := err.(etcd.EtcdError) if ok && ee.ErrorCode == 100 { // node does not exist - start = 0 - end = blocksize - 1 - client.Set(path, "0", 0) // TODO: error check + /* start = 0 + end = blocksize - 1 + client.Set(path, "0", 0) // TODO: error check + _ , err := client.CompareAndSwap(path, "0", 0, "0", 0) + + */ + //_,_ := c.put(path, "0", 0, options) + client.RawCreate(path, "0", 0) + continue } else { log.Fatal(err) } } else { // No error, get start of series from etcd node - start, err = strconv.ParseUint(node.Kvs[0].Value, 10, 0) + start, err = strconv.ParseUint(node.Node.Value, 10, 0) end = start + blocksize if err != nil { log.Fatal(err) @@ -50,7 +64,7 @@ func (self *Nexter) countloop(ch chan uint64, id int, client *etcd.Client) { } else { log.Println("Error with CompareAndSet! Trying again in 1 second...") time.Sleep(time.Second) - start, err = strconv.ParseUint(newval.Value, 10, 0) + start, err = strconv.ParseUint(newval.Node.Value, 10, 0) if err != nil { log.Fatal(err) } @@ -68,25 +82,44 @@ func (self *Nexter) loop() { client := etcd.NewClient(nil) for { select { - case countreq := <-self.reqchan: - counter, ok := counters[countreq.id] - if !ok { - counter = make(chan uint64) - counters[countreq.id] = counter - go self.countloop(counter, countreq.id, client) + case req := <-self.reqchan: + switch req.(type) { + case IncReq: + countreq := req.(*IncReq) + counter, ok := counters[countreq.id] + if !ok { + counter = make(chan uint64) + counters[countreq.id] = counter + go self.countloop(counter, countreq.id, client) + } + go func() { countreq.ret <- <-counter }() + case DelReq: + delreq := req.(*DelReq) + delete(counters, delreq.id) + path := "nexter/" + strconv.Itoa(delreq.id) + client.Delete(path, true) + } - go func() {countreq.ret <- <-counter}() + case <-self.done: + break } } } + func (self *Nexter) GetCount(id int) uint64 { ret := make(chan uint64) - self.reqchan <- &Req{id, ret} + self.reqchan <- &IncReq{id, ret} return <-ret } +func (self *Nexter) Delete(id int) { + self.reqchan <- &DelReq{id} +} +func (self *Nexter) Stop() { + self.done <- true +} func NewNexter() *Nexter { - nexter := &Nexter{make(chan *Req)} + nexter := &Nexter{make(chan Req), make(chan bool)} go nexter.loop() return nexter }