mirror of
https://github.com/cgrates/cgrates.git
synced 2026-02-11 18:16:24 +05:00
moved balancer out as a general library
This commit is contained in:
Radu Ioan Fericean
@@ -1,113 +0,0 @@
|
||||
/*
|
||||
Rating system designed to be used in VoIP Carriers Wobld
|
||||
Copyright (C) 2012 Radu Ioan Fericean
|
||||
|
||||
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 balancer
|
||||
|
||||
import (
|
||||
"log"
|
||||
"sync"
|
||||
)
|
||||
|
||||
type Balancer struct {
|
||||
sync.RWMutex
|
||||
clients map[string]Worker
|
||||
balancerChannel chan Worker
|
||||
}
|
||||
|
||||
type Worker interface {
|
||||
Call(serviceMethod string, args interface{}, reply interface{}) error
|
||||
Close() error
|
||||
}
|
||||
|
||||
/*
|
||||
Constructor for RateList holding one slice for addreses and one slice for connections.
|
||||
*/
|
||||
func NewBalancer() *Balancer {
|
||||
r := &Balancer{clients: make(map[string]Worker), balancerChannel: make(chan Worker)} // leaving both slices to nil
|
||||
go func() {
|
||||
for {
|
||||
if len(r.clients) > 0 {
|
||||
for _, c := range r.clients {
|
||||
r.balancerChannel <- c
|
||||
}
|
||||
} else {
|
||||
r.balancerChannel <- nil
|
||||
}
|
||||
}
|
||||
}()
|
||||
return r
|
||||
}
|
||||
|
||||
/*
|
||||
Adds a client to the two internal slices.
|
||||
*/
|
||||
func (bl *Balancer) AddClient(address string, client Worker) {
|
||||
bl.Lock()
|
||||
defer bl.Unlock()
|
||||
bl.clients[address] = client
|
||||
return
|
||||
}
|
||||
|
||||
/*
|
||||
Removes a client from the slices locking the readers and reseting the balancer index.
|
||||
*/
|
||||
func (bl *Balancer) RemoveClient(address string) {
|
||||
bl.Lock()
|
||||
defer bl.Unlock()
|
||||
delete(bl.clients, address)
|
||||
<-bl.balancerChannel
|
||||
}
|
||||
|
||||
/*
|
||||
Returns a client for the specifed address.
|
||||
*/
|
||||
func (bl *Balancer) GetClient(address string) (c Worker, exists bool) {
|
||||
bl.RLock()
|
||||
defer bl.RUnlock()
|
||||
c, exists = bl.clients[address]
|
||||
return
|
||||
}
|
||||
|
||||
/*
|
||||
Returns the next available connection at each call looping at the end of connections.
|
||||
*/
|
||||
func (bl *Balancer) Balance() (result Worker) {
|
||||
bl.RLock()
|
||||
defer bl.RUnlock()
|
||||
return <-bl.balancerChannel
|
||||
}
|
||||
|
||||
func (bl *Balancer) Shutdown() {
|
||||
bl.Lock()
|
||||
defer bl.Unlock()
|
||||
var reply string
|
||||
for address, client := range bl.clients {
|
||||
client.Call("Responder.Shutdown", "", &reply)
|
||||
log.Printf("Shutdown rater %v: %v ", address, reply)
|
||||
}
|
||||
}
|
||||
|
||||
func (bl *Balancer) GetClientAddresses() []string {
|
||||
bl.RLock()
|
||||
defer bl.RUnlock()
|
||||
var addresses []string
|
||||
for a, _ := range bl.clients {
|
||||
addresses = append(addresses, a)
|
||||
}
|
||||
return addresses
|
||||
}
|
||||
@@ -1,94 +0,0 @@
|
||||
/*
|
||||
Rating system designed to be used in VoIP Carriers World
|
||||
Copyright (C) 2012 Radu Ioan Fericean
|
||||
|
||||
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 balancer
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"net/rpc"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func BenchmarkBalance(b *testing.B) {
|
||||
balancer := NewBalancer()
|
||||
balancer.AddClient("client 1", new(rpc.Client))
|
||||
balancer.AddClient("client 2", new(rpc.Client))
|
||||
balancer.AddClient("client 3", new(rpc.Client))
|
||||
for i := 0; i < b.N; i++ {
|
||||
balancer.Balance()
|
||||
}
|
||||
}
|
||||
|
||||
func TestRemoving(t *testing.T) {
|
||||
balancer := NewBalancer()
|
||||
c1 := new(rpc.Client)
|
||||
c2 := new(rpc.Client)
|
||||
c3 := new(rpc.Client)
|
||||
balancer.AddClient("client 1", c1)
|
||||
balancer.AddClient("client 2", c2)
|
||||
balancer.AddClient("client 3", c3)
|
||||
balancer.RemoveClient("client 2")
|
||||
if balancer.clients["client 1"] != c1 ||
|
||||
balancer.clients["client 3"] != c3 ||
|
||||
len(balancer.clients) != 2 {
|
||||
t.Error("Failed removing rater")
|
||||
}
|
||||
}
|
||||
|
||||
func TestGet(t *testing.T) {
|
||||
balancer := NewBalancer()
|
||||
c1 := new(rpc.Client)
|
||||
balancer.AddClient("client 1", c1)
|
||||
result, ok := balancer.GetClient("client 1")
|
||||
if !ok || c1 != result {
|
||||
t.Error("Get failed")
|
||||
}
|
||||
}
|
||||
|
||||
func TestOneBalancer(t *testing.T) {
|
||||
balancer := NewBalancer()
|
||||
balancer.AddClient("client 1", new(rpc.Client))
|
||||
c1 := balancer.Balance()
|
||||
c2 := balancer.Balance()
|
||||
if c1 != c2 {
|
||||
t.Error("With only one rater these shoud be equal")
|
||||
}
|
||||
}
|
||||
|
||||
func Test100Balancer(t *testing.T) {
|
||||
balancer := NewBalancer()
|
||||
var clients []Worker
|
||||
for i := 0; i < 100; i++ {
|
||||
c := new(rpc.Client)
|
||||
balancer.AddClient(fmt.Sprintf("client%v", i), c)
|
||||
}
|
||||
for i := 0; i < 100; i++ {
|
||||
c := balancer.Balance()
|
||||
if c == nil {
|
||||
t.Error("Retuned nil client!")
|
||||
}
|
||||
for _, o := range clients {
|
||||
if c == o {
|
||||
t.Error("Balance did not iterate all the available clients")
|
||||
break
|
||||
}
|
||||
}
|
||||
clients = append(clients, c)
|
||||
}
|
||||
|
||||
}
|
||||
@@ -23,9 +23,9 @@ import (
|
||||
"errors"
|
||||
"flag"
|
||||
"fmt"
|
||||
"github.com/cgrates/cgrates/balancer"
|
||||
"github.com/cgrates/cgrates/sessionmanager"
|
||||
"github.com/cgrates/cgrates/timespans"
|
||||
"github.com/rif/balancer2go"
|
||||
"io"
|
||||
"net"
|
||||
"net/http"
|
||||
@@ -95,7 +95,7 @@ var (
|
||||
freeswitch_server = "localhost:8021" // freeswitch address host:port
|
||||
freeswitch_pass = "ClueCon" // reeswitch address host:port
|
||||
|
||||
bal = balancer.NewBalancer()
|
||||
bal = balancer2go.NewBalancer()
|
||||
exitChan = make(chan bool)
|
||||
)
|
||||
|
||||
|
||||
@@ -37,7 +37,7 @@ func stopBalancerSingnalHandler() {
|
||||
|
||||
sig := <-c
|
||||
timespans.Logger.Info(fmt.Sprintf("Caught signal %v, sending shutdownto raters\n", sig))
|
||||
bal.Shutdown()
|
||||
bal.Shutdown("Responder.Shutdown")
|
||||
exitChan <- true
|
||||
}
|
||||
|
||||
|
||||
@@ -21,7 +21,7 @@ package timespans
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"github.com/cgrates/cgrates/balancer"
|
||||
"github.com/rif/balancer2go"
|
||||
"net/rpc"
|
||||
"reflect"
|
||||
"runtime"
|
||||
@@ -30,7 +30,7 @@ import (
|
||||
)
|
||||
|
||||
type Responder struct {
|
||||
Bal *balancer.Balancer
|
||||
Bal *balancer2go.Balancer
|
||||
ExitChan chan bool
|
||||
}
|
||||
|
||||
@@ -167,7 +167,7 @@ func (rs *Responder) Status(arg string, reply *string) (err error) {
|
||||
|
||||
func (rs *Responder) Shutdown(arg string, reply *string) (err error) {
|
||||
if rs.Bal != nil {
|
||||
rs.Bal.Shutdown()
|
||||
rs.Bal.Shutdown("Responder.Shutdown")
|
||||
}
|
||||
storageGetter.Close()
|
||||
defer func() { rs.ExitChan <- true }()
|
||||
|
||||
Reference in New Issue
Block a user