mirror of
https://github.com/cgrates/cgrates.git
synced 2026-02-11 10:06:24 +05:00
Moving Guardian into it's own package so we can use it from other independent packages
This commit is contained in:
DanB
142
guardian/guardian.go
Normal file
142
guardian/guardian.go
Normal file
@@ -0,0 +1,142 @@
|
||||
/*
|
||||
Real-time Online/Offline Charging System (OCS) for Telecom & ISP environments
|
||||
Copyright (C) ITsysCOM GmbH
|
||||
|
||||
This program is free software: you can redistribute it and/or modify
|
||||
it under the terms of the GNU General Public License as published by
|
||||
the Free Software Foundation, either version 3 of the License, or
|
||||
(at your option) any later version.
|
||||
|
||||
This program is distributed in the hope that it will be useful,
|
||||
but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
GNU General Public License for more details.
|
||||
|
||||
You should have received a copy of the GNU General Public License
|
||||
along with this program. If not, see <http://www.gnu.org/licenses/>
|
||||
*/
|
||||
package guardian
|
||||
|
||||
import (
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// global package variable
|
||||
var Guardian = &GuardianLock{locksMap: make(map[string]*itemLock)}
|
||||
|
||||
func newItemLock(keyID string) *itemLock {
|
||||
return &itemLock{keyID: keyID}
|
||||
}
|
||||
|
||||
// itemLock represents one lock with key autodestroy
|
||||
type itemLock struct {
|
||||
keyID string // store it so we know what to destroy
|
||||
cnt int
|
||||
sync.Mutex
|
||||
}
|
||||
|
||||
// unlock() executes combined lock with autoremoving lock from Guardian
|
||||
func (il *itemLock) unlock() {
|
||||
il.cnt--
|
||||
if il.cnt == 0 { // last lock in the queue
|
||||
Guardian.Lock()
|
||||
delete(Guardian.locksMap, il.keyID)
|
||||
Guardian.Unlock()
|
||||
}
|
||||
il.Unlock()
|
||||
}
|
||||
|
||||
// GuardianLock is an optimized locking system per locking key
|
||||
type GuardianLock struct {
|
||||
locksMap map[string]*itemLock
|
||||
sync.Mutex // protects the maps
|
||||
}
|
||||
|
||||
// lockItems locks a set of lockIDs
|
||||
// returning the lock structs so they can be later unlocked
|
||||
func (guard *GuardianLock) lockItems(lockIDs []string) (itmLocks []*itemLock) {
|
||||
guard.Lock()
|
||||
for _, lockID := range lockIDs {
|
||||
var itmLock *itemLock
|
||||
itmLock, exists := Guardian.locksMap[lockID]
|
||||
if !exists {
|
||||
itmLock = newItemLock(lockID)
|
||||
Guardian.locksMap[lockID] = itmLock
|
||||
}
|
||||
itmLock.cnt++
|
||||
itmLocks = append(itmLocks, itmLock)
|
||||
}
|
||||
guard.Unlock()
|
||||
for _, itmLock := range itmLocks {
|
||||
itmLock.Lock()
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
// unlockItems will unlock the items provided
|
||||
func (guard *GuardianLock) unlockItems(itmLocks []*itemLock) {
|
||||
for _, itmLock := range itmLocks {
|
||||
itmLock.unlock()
|
||||
}
|
||||
}
|
||||
|
||||
func (guard *GuardianLock) Guard(handler func() (interface{}, error), timeout time.Duration, lockIDs ...string) (reply interface{}, err error) {
|
||||
itmLocks := guard.lockItems(lockIDs)
|
||||
|
||||
rplyChan := make(chan interface{})
|
||||
errChan := make(chan error)
|
||||
go func(rplyChan chan interface{}, errChan chan error) {
|
||||
// execute
|
||||
if rply, err := handler(); err != nil {
|
||||
errChan <- err
|
||||
} else {
|
||||
rplyChan <- rply
|
||||
}
|
||||
}(rplyChan, errChan)
|
||||
|
||||
if timeout > 0 { // wait with timeout
|
||||
select {
|
||||
case err = <-errChan:
|
||||
case reply = <-rplyChan:
|
||||
case <-time.After(timeout):
|
||||
}
|
||||
} else { // a bit dangerous but wait till handler finishes
|
||||
select {
|
||||
case err = <-errChan:
|
||||
case reply = <-rplyChan:
|
||||
}
|
||||
}
|
||||
|
||||
guard.unlockItems(itmLocks)
|
||||
return
|
||||
}
|
||||
|
||||
// GuardTimed aquires a lock for duration, returning an identifier which can be used later to cancel the lock or find it's status
|
||||
func (guard *GuardianLock) GuardIDs(timeout time.Duration, lockIDs ...string) {
|
||||
guard.lockItems(lockIDs)
|
||||
if timeout != 0 {
|
||||
go func(timeout time.Duration, lockIDs ...string) {
|
||||
time.Sleep(timeout)
|
||||
guard.UnguardIDs(lockIDs...)
|
||||
}(timeout, lockIDs...)
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
// UnguardTimed attempts to unlock a set of locks based on their locksUUID
|
||||
// Returns false if locks were not found
|
||||
func (guard *GuardianLock) UnguardIDs(lockIDs ...string) {
|
||||
var itmLocks []*itemLock
|
||||
guard.Lock()
|
||||
for _, lockID := range lockIDs {
|
||||
var itmLock *itemLock
|
||||
itmLock, exists := Guardian.locksMap[lockID]
|
||||
if exists {
|
||||
itmLocks = append(itmLocks, itmLock)
|
||||
}
|
||||
}
|
||||
guard.Unlock()
|
||||
guard.unlockItems(itmLocks)
|
||||
return
|
||||
}
|
||||
169
guardian/guardian_test.go
Normal file
169
guardian/guardian_test.go
Normal file
@@ -0,0 +1,169 @@
|
||||
/*
|
||||
Real-time Online/Offline Charging System (OCS) for Telecom & ISP environments
|
||||
Copyright (C) ITsysCOM GmbH
|
||||
|
||||
This program is free software: you can redistribute it and/or modify
|
||||
it under the terms of the GNU General Public License as published by
|
||||
the Free Software Foundation, either version 3 of the License, or
|
||||
(at your option) any later version.
|
||||
|
||||
This program is distributed in the hope that it will be useful,
|
||||
but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
GNU General Public License for more details.
|
||||
|
||||
You should have received a copy of the GNU General Public License
|
||||
along with this program. If not, see <http://www.gnu.org/licenses/>
|
||||
*/
|
||||
package guardian
|
||||
|
||||
import (
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func delayHandler() (interface{}, error) {
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
// Forks 3 groups of workers and makes sure that the time for execution is the one we expect for all 15 goroutines (with 100ms )
|
||||
func TestGuardianMultipleKeys(t *testing.T) {
|
||||
tStart := time.Now()
|
||||
maxIter := 5
|
||||
sg := new(sync.WaitGroup)
|
||||
keys := []string{"test1", "test2", "test3"}
|
||||
for i := 0; i < maxIter; i++ {
|
||||
for _, key := range keys {
|
||||
sg.Add(1)
|
||||
go func(key string) {
|
||||
Guardian.Guard(delayHandler, 0, key)
|
||||
sg.Done()
|
||||
}(key)
|
||||
}
|
||||
}
|
||||
sg.Wait()
|
||||
mustExecDur := time.Duration(maxIter*100) * time.Millisecond
|
||||
if execTime := time.Now().Sub(tStart); execTime < mustExecDur || execTime > mustExecDur+time.Duration(20*time.Millisecond) {
|
||||
t.Errorf("Execution took: %v", execTime)
|
||||
}
|
||||
for _, key := range keys {
|
||||
if _, hasKey := Guardian.locksMap[key]; hasKey {
|
||||
t.Error("Possible memleak")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestGuardianTimeout(t *testing.T) {
|
||||
tStart := time.Now()
|
||||
maxIter := 5
|
||||
sg := new(sync.WaitGroup)
|
||||
keys := []string{"test1", "test2", "test3"}
|
||||
for i := 0; i < maxIter; i++ {
|
||||
for _, key := range keys {
|
||||
sg.Add(1)
|
||||
go func(key string) {
|
||||
Guardian.Guard(delayHandler, time.Duration(10*time.Millisecond), key)
|
||||
sg.Done()
|
||||
}(key)
|
||||
}
|
||||
}
|
||||
sg.Wait()
|
||||
mustExecDur := time.Duration(maxIter*10) * time.Millisecond
|
||||
if execTime := time.Now().Sub(tStart); execTime < mustExecDur || execTime > mustExecDur+time.Duration(20*time.Millisecond) {
|
||||
t.Errorf("Execution took: %v", execTime)
|
||||
}
|
||||
for _, key := range keys {
|
||||
if _, hasKey := Guardian.locksMap[key]; hasKey {
|
||||
t.Error("Possible memleak")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestGuardianGuardIDs(t *testing.T) {
|
||||
lockIDs := []string{"test1", "test2", "test3"}
|
||||
for _, lockID := range lockIDs {
|
||||
if _, hasKey := Guardian.locksMap[lockID]; hasKey {
|
||||
t.Errorf("Unexpected lockID found: %s", lockID)
|
||||
}
|
||||
}
|
||||
tStart := time.Now()
|
||||
lockDur := 2 * time.Millisecond
|
||||
Guardian.GuardIDs(lockDur, lockIDs...)
|
||||
for _, lockID := range lockIDs {
|
||||
if itmLock, hasKey := Guardian.locksMap[lockID]; !hasKey {
|
||||
t.Errorf("Cannot find lock for lockID: %s", lockID)
|
||||
} else if itmLock.cnt != 1 {
|
||||
t.Errorf("Unexpected itmLock found: %+v", itmLock)
|
||||
}
|
||||
}
|
||||
go Guardian.GuardIDs(time.Duration(1*time.Millisecond), lockIDs[1:]...) // to test counter
|
||||
time.Sleep(20 * time.Microsecond) // give time for goroutine to lock
|
||||
if itmLock, hasKey := Guardian.locksMap["test1"]; !hasKey {
|
||||
t.Errorf("Cannot find lock for lockID: %s", "test1")
|
||||
} else if itmLock.cnt != 1 {
|
||||
t.Errorf("Unexpected itmLock found: %+v", itmLock)
|
||||
}
|
||||
if itmLock, hasKey := Guardian.locksMap["test2"]; !hasKey {
|
||||
t.Errorf("Cannot find lock for lockID: %s", "test2")
|
||||
} else if itmLock.cnt != 2 {
|
||||
t.Errorf("Unexpected itmLock found: %+v", itmLock)
|
||||
}
|
||||
if itmLock, hasKey := Guardian.locksMap["test3"]; !hasKey {
|
||||
t.Errorf("Cannot find lock for lockID: %s", "test3")
|
||||
} else if itmLock.cnt != 2 {
|
||||
t.Errorf("Unexpected itmLock found: %+v", itmLock)
|
||||
}
|
||||
Guardian.GuardIDs(0, lockIDs...)
|
||||
if totalLockDur := time.Now().Sub(tStart); totalLockDur < lockDur {
|
||||
t.Errorf("Lock duration too small")
|
||||
}
|
||||
//time.Sleep(1000 * time.Microsecond)
|
||||
if len(Guardian.locksMap) != 3 {
|
||||
t.Errorf("locksMap should be have 3 elements, have: %+v", Guardian.locksMap)
|
||||
} else if itmLock, hasKey := Guardian.locksMap["test1"]; !hasKey {
|
||||
t.Errorf("Cannot find lock for lockID: %s", "test1")
|
||||
} else if itmLock.cnt != 1 {
|
||||
t.Errorf("Unexpected itmLock found: %+v", itmLock)
|
||||
} else if itmLock, hasKey := Guardian.locksMap["test2"]; !hasKey {
|
||||
t.Errorf("Cannot find lock for lockID: %s", "test2")
|
||||
} else if itmLock.cnt != 1 {
|
||||
t.Errorf("Unexpected itmLock found: %+v", itmLock)
|
||||
} else if itmLock, hasKey := Guardian.locksMap["test3"]; !hasKey {
|
||||
t.Errorf("Cannot find lock for lockID: %s", "test2")
|
||||
} else if itmLock.cnt != 1 {
|
||||
t.Errorf("Unexpected itmLock found: %+v", itmLock)
|
||||
}
|
||||
Guardian.UnguardIDs(lockIDs...)
|
||||
if len(Guardian.locksMap) != 0 {
|
||||
t.Errorf("locksMap should be have 0 elements, have: %+v", Guardian.locksMap)
|
||||
}
|
||||
}
|
||||
|
||||
func BenchmarkGuard(b *testing.B) {
|
||||
for i := 0; i < 100; i++ {
|
||||
go Guardian.Guard(func() (interface{}, error) {
|
||||
time.Sleep(1 * time.Millisecond)
|
||||
return 0, nil
|
||||
}, 0, "1")
|
||||
go Guardian.Guard(func() (interface{}, error) {
|
||||
time.Sleep(1 * time.Millisecond)
|
||||
return 0, nil
|
||||
}, 0, "2")
|
||||
go Guardian.Guard(func() (interface{}, error) {
|
||||
time.Sleep(1 * time.Millisecond)
|
||||
return 0, nil
|
||||
}, 0, "1")
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
func BenchmarkGuardian(b *testing.B) {
|
||||
for i := 0; i < 100; i++ {
|
||||
go Guardian.Guard(func() (interface{}, error) {
|
||||
time.Sleep(1 * time.Millisecond)
|
||||
return 0, nil
|
||||
}, 0, "1")
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user