NewRuntime creates new kayak Runtime.
(cfg *kt.RuntimeConfig)
| 160 | |
| 161 | // NewRuntime creates new kayak Runtime. |
| 162 | func NewRuntime(cfg *kt.RuntimeConfig) (rt *Runtime, err error) { |
| 163 | if cfg == nil { |
| 164 | err = errors.Wrap(kt.ErrInvalidConfig, "nil config") |
| 165 | return |
| 166 | } |
| 167 | |
| 168 | peers := cfg.Peers |
| 169 | |
| 170 | if peers == nil { |
| 171 | err = errors.Wrap(kt.ErrInvalidConfig, "nil peers") |
| 172 | return |
| 173 | } |
| 174 | |
| 175 | // verify peers |
| 176 | if err = peers.Verify(); err != nil { |
| 177 | err = errors.Wrap(err, "verify peers during kayak init failed") |
| 178 | return |
| 179 | } |
| 180 | |
| 181 | followers := make([]proto.NodeID, 0, len(peers.Servers)) |
| 182 | exists := false |
| 183 | var role proto.ServerRole |
| 184 | |
| 185 | for _, v := range peers.Servers { |
| 186 | if !v.IsEqual(&peers.Leader) { |
| 187 | followers = append(followers, v) |
| 188 | } |
| 189 | |
| 190 | if v.IsEqual(&cfg.NodeID) { |
| 191 | exists = true |
| 192 | if v.IsEqual(&peers.Leader) { |
| 193 | role = proto.Leader |
| 194 | } else { |
| 195 | role = proto.Follower |
| 196 | } |
| 197 | } |
| 198 | } |
| 199 | |
| 200 | if !exists { |
| 201 | err = errors.Wrapf(kt.ErrNotInPeer, "node %v not in peers %v", cfg.NodeID, peers) |
| 202 | return |
| 203 | } |
| 204 | |
| 205 | // calculate fan-out count according to threshold and peers info |
| 206 | minPreparedFollowers := int(math.Max(math.Ceil(cfg.PrepareThreshold*float64(len(peers.Servers))), 1) - 1) |
| 207 | minCommitFollowers := int(math.Max(math.Ceil(cfg.CommitThreshold*float64(len(peers.Servers))), 1) - 1) |
| 208 | |
| 209 | rt = &Runtime{ |
| 210 | // indexes |
| 211 | pendingPrepares: make(map[uint64]bool, commitWindow*2), |
| 212 | |
| 213 | // handler and logs |
| 214 | sh: cfg.Handler, |
| 215 | wal: cfg.Wal, |
| 216 | instanceID: cfg.InstanceID, |
| 217 | |
| 218 | // peers |
| 219 | peers: cfg.Peers, |