| 1 | // SPDX-License-Identifier: GPL-3.0-or-later |
| 2 | |
| 3 | package chartengine |
| 4 | |
| 5 | import ( |
| 6 | "errors" |
| 7 | "fmt" |
| 8 | "sync" |
| 9 | |
| 10 | "github.com/netdata/netdata/go/plugins/pkg/metrix" |
| 11 | ) |
| 12 | |
| 13 | var ( |
| 14 | ErrOutstandingPlanAttempt = errors.New("chartengine: plan attempt already outstanding") |
| 15 | ErrStalePlanAttempt = errors.New("chartengine: stale plan attempt") |
| 16 | ErrFinishedPlanAttempt = errors.New("chartengine: plan attempt already finished") |
| 17 | ) |
| 18 | |
| 19 | type PlanAttempt struct { |
| 20 | state *planAttemptState |
| 21 | } |
| 22 | |
| 23 | type planAttemptState struct { |
| 24 | mu sync.Mutex |
| 25 | |
| 26 | engine *Engine |
| 27 | plan Plan |
| 28 | materialized materializedState |
| 29 | epoch uint64 |
| 30 | commitSeq uint64 |
| 31 | attemptID uint64 |
| 32 | reserved bool |
| 33 | finished bool |
| 34 | } |
| 35 | |
| 36 | func (a PlanAttempt) Plan() Plan { |
| 37 | if a.state == nil { |
| 38 | return Plan{} |
| 39 | } |
| 40 | return a.state.plan |
| 41 | } |
| 42 | |
| 43 | func (a PlanAttempt) Commit() error { |
| 44 | if a.state == nil { |
| 45 | return nil |
| 46 | } |
| 47 | |
| 48 | a.state.mu.Lock() |
| 49 | if a.state.finished { |
| 50 | a.state.mu.Unlock() |
| 51 | return ErrFinishedPlanAttempt |
| 52 | } |
| 53 | a.state.finished = true |
| 54 | reserved := a.state.reserved |
| 55 | engine := a.state.engine |
| 56 | materialized := a.state.materialized |
| 57 | epoch := a.state.epoch |
| 58 | commitSeq := a.state.commitSeq |
| 59 | attemptID := a.state.attemptID |
| 60 | a.state.mu.Unlock() |
| 61 | |
| 62 | if !reserved { |
| 63 | return nil |
| 64 | } |
| 65 | if engine == nil { |
| 66 | return fmt.Errorf("chartengine: nil engine on commit") |
| 67 | } |
| 68 | return engine.commitAttempt(materialized, epoch, commitSeq, attemptID) |
| 69 | } |
| 70 | |
| 71 | func (a PlanAttempt) Abort() { |
| 72 | if a.state == nil { |
| 73 | return |
| 74 | } |
| 75 | |
| 76 | a.state.mu.Lock() |
| 77 | if a.state.finished { |
| 78 | a.state.mu.Unlock() |
| 79 | return |
| 80 | } |
| 81 | a.state.finished = true |
| 82 | reserved := a.state.reserved |
| 83 | engine := a.state.engine |
| 84 | attemptID := a.state.attemptID |
| 85 | a.state.mu.Unlock() |
| 86 | |
| 87 | if !reserved || engine == nil { |
| 88 | return |
| 89 | } |
| 90 | engine.abortAttempt(attemptID) |
| 91 | } |
| 92 | |
| 93 | func newPreparedAttempt( |
| 94 | engine *Engine, |
| 95 | plan Plan, |
| 96 | materialized materializedState, |
| 97 | epoch uint64, |
| 98 | commitSeq uint64, |
| 99 | attemptID uint64, |
| 100 | ) PlanAttempt { |
| 101 | return PlanAttempt{ |
| 102 | state: &planAttemptState{ |
| 103 | engine: engine, |
| 104 | plan: plan, |
| 105 | materialized: materialized, |
| 106 | epoch: epoch, |
| 107 | commitSeq: commitSeq, |
| 108 | attemptID: attemptID, |
| 109 | reserved: true, |
| 110 | }, |
| 111 | } |
| 112 | } |
| 113 | |
| 114 | func newNoopAttempt(plan Plan) PlanAttempt { |
| 115 | return PlanAttempt{ |
| 116 | state: &planAttemptState{ |
| 117 | plan: plan, |
| 118 | }, |
| 119 | } |
| 120 | } |
| 121 | |
| 122 | func (e *Engine) PreparePlan(reader metrix.Reader) (PlanAttempt, error) { |
| 123 | plan, materialized, epoch, commitSeq, attemptID, reserved, err := e.preparePlan(reader) |
| 124 | if err != nil { |
| 125 | return PlanAttempt{}, err |
| 126 | } |
| 127 | if !reserved { |
| 128 | return newNoopAttempt(plan), nil |
| 129 | } |
| 130 | return newPreparedAttempt(e, plan, materialized, epoch, commitSeq, attemptID), nil |
| 131 | } |
| 132 | |
| 133 | func (e *Engine) nextAttemptIDLocked() uint64 { |
| 134 | e.state.nextAttempt++ |
| 135 | if e.state.nextAttempt == 0 { |
| 136 | e.state.nextAttempt++ |
| 137 | } |
| 138 | return e.state.nextAttempt |
| 139 | } |
| 140 | |
| 141 | func (e *Engine) commitAttempt(materialized materializedState, epoch, commitSeq, attemptID uint64) error { |
| 142 | if e == nil { |
| 143 | return fmt.Errorf("chartengine: nil engine") |
| 144 | } |
| 145 | e.mu.Lock() |
| 146 | defer e.mu.Unlock() |
| 147 | |
| 148 | if e.state.outstanding != attemptID || e.state.outstanding == 0 { |
| 149 | return ErrStalePlanAttempt |
| 150 | } |
| 151 | if e.state.engineEpoch != epoch || e.state.commitSeq != commitSeq { |
| 152 | e.state.outstanding = 0 |
| 153 | return ErrStalePlanAttempt |
| 154 | } |
| 155 | |
| 156 | e.state.materialized = materialized |
| 157 | e.state.commitSeq++ |
| 158 | e.state.outstanding = 0 |
| 159 | return nil |
| 160 | } |
| 161 | |
| 162 | func (e *Engine) abortAttempt(attemptID uint64) { |
| 163 | if e == nil { |
| 164 | return |
| 165 | } |
| 166 | e.mu.Lock() |
| 167 | if e.state.outstanding == attemptID { |
| 168 | e.state.outstanding = 0 |
| 169 | } |
| 170 | e.mu.Unlock() |
| 171 | } |
| 172 | |
| 173 | func prepareAndCommitPlan(engine *Engine, reader metrix.Reader) (Plan, error) { |
| 174 | if engine == nil { |
| 175 | return Plan{}, fmt.Errorf("chartengine: nil engine") |
| 176 | } |
| 177 | attempt, err := engine.PreparePlan(reader) |
| 178 | if err != nil { |
| 179 | return Plan{}, err |
| 180 | } |
| 181 | |
| 182 | plan := attempt.Plan() |
| 183 | if err := attempt.Commit(); err != nil { |
| 184 | return Plan{}, err |
| 185 | } |
| 186 | return plan, nil |
| 187 | } |