better data structures for matches & actions maps
This commit is contained in:
parent
b6d7e5a946
commit
d26c7f24f2
@ -111,12 +111,7 @@ func ActionsManager() {
|
||||
go action.exec(match)
|
||||
} else {
|
||||
actionsLock.Lock()
|
||||
// make sure map exists
|
||||
if actions[action] == nil {
|
||||
actions[action] = make(PatternTimes)
|
||||
}
|
||||
// append() to nil is valid go
|
||||
actions[action][match] = append(actions[action][match], then)
|
||||
actions[PAT{match, action, then}] = struct{}{}
|
||||
actionsLock.Unlock()
|
||||
go func(pat PAT, now time.Time) {
|
||||
time.Sleep(pat.t.Sub(now))
|
||||
@ -124,21 +119,18 @@ func ActionsManager() {
|
||||
}(pat, now)
|
||||
}
|
||||
case pat = <-pendingActionsC:
|
||||
match = pat.p
|
||||
action = pat.a
|
||||
match, action, then = pat.p, pat.a, pat.t
|
||||
actionsLock.Lock()
|
||||
actions[action][match] = actions[action][match][1:]
|
||||
delete(actions, PAT{match, action, then})
|
||||
actionsLock.Unlock()
|
||||
wgActions.Add(1)
|
||||
go action.exec(match)
|
||||
case _, _ = <-stopActions:
|
||||
actionsLock.Lock()
|
||||
for action := range actions {
|
||||
if action.OnExit {
|
||||
for match := range actions[action] {
|
||||
wgActions.Add(1)
|
||||
go action.exec(match)
|
||||
}
|
||||
for pat := range actions {
|
||||
if pat.a.OnExit {
|
||||
wgActions.Add(1)
|
||||
go action.exec(match)
|
||||
}
|
||||
}
|
||||
actionsLock.Unlock()
|
||||
@ -156,7 +148,7 @@ func MatchesManager() {
|
||||
for !end {
|
||||
select {
|
||||
case pf = <-cleanMatchesC:
|
||||
delete(matches[pf.f], pf.p)
|
||||
delete(matches, pf)
|
||||
case pft, ok := <-startupMatchesC:
|
||||
if !ok {
|
||||
end = true
|
||||
@ -170,7 +162,7 @@ func MatchesManager() {
|
||||
select {
|
||||
case pf = <-cleanMatchesC:
|
||||
matchesLock.Lock()
|
||||
delete(matches[pf.f], pf.p)
|
||||
delete(matches, pf)
|
||||
matchesLock.Unlock()
|
||||
case pft = <-matchesC:
|
||||
|
||||
@ -187,26 +179,25 @@ func MatchesManager() {
|
||||
|
||||
func matchesManagerHandleMatch(pft PFT) bool {
|
||||
filter, match, then := pft.f, pft.p, pft.t
|
||||
pf := PF{pft.p, pft.f}
|
||||
|
||||
if filter.Retry > 1 {
|
||||
// make sure map exists
|
||||
if matches[filter] == nil {
|
||||
matches[filter] = make(PatternTimes)
|
||||
if matches[pf] == nil {
|
||||
matches[pf] = make(map[time.Time]struct{})
|
||||
}
|
||||
// clean old matches
|
||||
newMatches := make([]time.Time, 0, len(matches[filter][match]))
|
||||
for _, old := range matches[filter][match] {
|
||||
if old.Add(filter.retryDuration).After(then) {
|
||||
newMatches = append(newMatches, old)
|
||||
for old := range matches[pf] {
|
||||
if !old.Add(filter.retryDuration).After(then) {
|
||||
delete(matches[pf], old)
|
||||
}
|
||||
}
|
||||
// add new match
|
||||
newMatches = append(newMatches, then)
|
||||
matches[filter][match] = newMatches
|
||||
matches[pf][then] = struct{}{}
|
||||
}
|
||||
|
||||
if filter.Retry <= 1 || len(matches[filter][match]) >= filter.Retry {
|
||||
delete(matches[filter], match)
|
||||
if filter.Retry <= 1 || len(matches[pf]) >= filter.Retry {
|
||||
delete(matches, pf)
|
||||
filter.sendActions(match, then)
|
||||
return true
|
||||
}
|
||||
|
32
app/pipe.go
32
app/pipe.go
@ -14,45 +14,37 @@ func genClientStatus() ClientStatus {
|
||||
matchesLock.Lock()
|
||||
|
||||
// Painful data manipulation
|
||||
for filter, filterMatches := range matches {
|
||||
for pf, times := range matches {
|
||||
pattern, filter := pf.p, pf.f
|
||||
if cs[filter.stream.name] == nil {
|
||||
cs[filter.stream.name] = make(map[string]MapPatternStatus)
|
||||
}
|
||||
if cs[filter.stream.name][filter.name] == nil {
|
||||
cs[filter.stream.name][filter.name] = make(MapPatternStatus)
|
||||
}
|
||||
for pattern, patternMatches := range filterMatches {
|
||||
var ps PatternStatus
|
||||
cs[filter.stream.name][filter.name][pattern] = &ps
|
||||
|
||||
ps.Matches = len(patternMatches)
|
||||
}
|
||||
cs[filter.stream.name][filter.name][pattern] = &PatternStatus{len(times), nil}
|
||||
}
|
||||
|
||||
matchesLock.Unlock()
|
||||
actionsLock.Lock()
|
||||
|
||||
// Painful data manipulation
|
||||
for action, pendingActions := range actions {
|
||||
for pat := range actions {
|
||||
pattern, action, then := pat.p, pat.a, pat.t
|
||||
if cs[action.filter.stream.name] == nil {
|
||||
cs[action.filter.stream.name] = make(map[string]MapPatternStatus)
|
||||
}
|
||||
if cs[action.filter.stream.name][action.filter.name] == nil {
|
||||
cs[action.filter.stream.name][action.filter.name] = make(MapPatternStatus)
|
||||
}
|
||||
for pattern, patternPendingActions := range pendingActions {
|
||||
if cs[action.filter.stream.name][action.filter.name][pattern] == nil {
|
||||
var ps PatternStatus
|
||||
cs[action.filter.stream.name][action.filter.name][pattern] = &ps
|
||||
}
|
||||
var ps *PatternStatus
|
||||
ps = cs[action.filter.stream.name][action.filter.name][pattern]
|
||||
ps.Actions = make(map[string][]string)
|
||||
|
||||
for _, t := range patternPendingActions {
|
||||
ps.Actions[action.name] = append(ps.Actions[action.name], t.Format(time.DateTime))
|
||||
}
|
||||
if cs[action.filter.stream.name][action.filter.name][pattern] == nil {
|
||||
cs[action.filter.stream.name][action.filter.name][pattern] = new(PatternStatus)
|
||||
}
|
||||
ps := cs[action.filter.stream.name][action.filter.name][pattern]
|
||||
if ps.Actions == nil {
|
||||
ps.Actions = make(map[string][]string)
|
||||
}
|
||||
ps.Actions[action.name] = append(ps.Actions[action.name], then.Format(time.DateTime))
|
||||
}
|
||||
actionsLock.Unlock()
|
||||
return cs
|
||||
|
@ -75,13 +75,10 @@ type WriteDB struct {
|
||||
enc *gob.Encoder
|
||||
}
|
||||
|
||||
type PatternTimes map[string][]time.Time
|
||||
type MatchesMap map[PF]map[time.Time]struct{}
|
||||
type ActionsMap map[PAT]struct{}
|
||||
|
||||
type MatchesMap map[*Filter]PatternTimes
|
||||
|
||||
type ActionsMap map[*Action]PatternTimes
|
||||
|
||||
// Helper structs made to carry information across channels
|
||||
// Helper structs made to carry information
|
||||
type SF struct{ s, f string }
|
||||
type PSF struct{ p, s, f string }
|
||||
type PF struct {
|
||||
|
Loading…
Reference in New Issue
Block a user