| 46 | |
| 47 | |
| 48 | func Main(clusterName, self, buri, rwsk, rosk string, cl *doozer.Conn, udpConn *net.UDPConn, listener, webListener net.Listener, pulseInterval, fillDelay, kickTimeout int64, hi int64) { |
| 49 | listenAddr := listener.Addr().String() |
| 50 | |
| 51 | canWrite := make(chan bool, 1) |
| 52 | in := make(chan consensus.Packet, 50) |
| 53 | out := make(chan consensus.Packet, 50) |
| 54 | |
| 55 | st := store.New() |
| 56 | pr := &proposer{ |
| 57 | seqns: make(chan int64, alpha), |
| 58 | props: make(chan *consensus.Prop), |
| 59 | st: st, |
| 60 | } |
| 61 | |
| 62 | calSrv := func(start int64) { |
| 63 | go gc.Pulse(self, st.Seqns, pr, pulseInterval) |
| 64 | go gc.Clean(st, hi, time.Tick(1e9)) |
| 65 | var m consensus.Manager |
| 66 | m.Self = self |
| 67 | m.DefRev = start |
| 68 | m.Alpha = alpha |
| 69 | m.In = in |
| 70 | m.Out = out |
| 71 | m.Ops = st.Ops |
| 72 | m.PSeqn = pr.seqns |
| 73 | m.Props = pr.props |
| 74 | m.TFill = fillDelay |
| 75 | m.Store = st |
| 76 | m.Ticker = time.Tick(10e6) |
| 77 | go m.Run() |
| 78 | } |
| 79 | |
| 80 | if cl == nil { // we are the only node in a new cluster |
| 81 | set(st, "/ctl/name", clusterName, store.Missing) |
| 82 | set(st, "/ctl/node/"+self+"/addr", listenAddr, store.Missing) |
| 83 | set(st, "/ctl/node/"+self+"/hostname", os.Getenv("HOSTNAME"), store.Missing) |
| 84 | set(st, "/ctl/node/"+self+"/version", Version, store.Missing) |
| 85 | set(st, "/ctl/cal/0", self, store.Missing) |
| 86 | if buri == "" { |
| 87 | set(st, "/ctl/ns/"+clusterName+"/"+self, listenAddr, store.Missing) |
| 88 | } |
| 89 | calSrv(<-st.Seqns) |
| 90 | // Skip ahead alpha steps so that the registrar can provide a |
| 91 | // meaningful cluster. |
| 92 | for i := 0; i < alpha; i++ { |
| 93 | st.Ops <- store.Op{1 + <-st.Seqns, store.Nop} |
| 94 | } |
| 95 | canWrite <- true |
| 96 | go setReady(pr, self) |
| 97 | } else { |
| 98 | setC(cl, "/ctl/node/"+self+"/addr", listenAddr, store.Clobber) |
| 99 | setC(cl, "/ctl/node/"+self+"/hostname", os.Getenv("HOSTNAME"), store.Clobber) |
| 100 | setC(cl, "/ctl/node/"+self+"/version", Version, store.Clobber) |
| 101 | |
| 102 | rev, err := cl.Rev() |
| 103 | if err != nil { |
| 104 | panic(err) |
| 105 | } |