uppal/cgrates
1
0
Fork 1
mirror of https://github.com/cgrates/cgrates.git synced 2026-02-22 07:38:45 +05:00
Code Issues Packages Projects Releases Wiki Activity

Update concurrent requests mechanism to work with BiJSON server

Browse Source
This commit is contained in:
TeoV
2020-07-09 11:08:29 +03:00
parent d2aa9d6581
commit cd81ab9449
6 changed files with 357 additions and 212 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
apier/v1/concreqs_it_test.go
View File

@@ -252,7 +252,6 @@ func testConcReqsOnBiJSONBusy(t *testing.T) {
if err := concReqsBiRPC.Call(utils.SessionSv1Sleep,
&DurationArgs{DurationTime: time.Duration(10 * time.Millisecond)},
&resp); err != nil {
fmt.Println(err)
lock.Lock()
failedAPIs++
lock.Unlock()

297
apier/v1/sessionsbirpc.go
View File

@@ -67,137 +67,434 @@ func (ssv1 *SessionSv1) Handlers() map[string]interface{} {
func (ssv1 *SessionSv1) BiRPCv1AuthorizeEvent(clnt *rpc2.Client, args *sessions.V1AuthorizeArgs,
rply *sessions.V1AuthorizeReply) error {
err := utils.ConReqs.Allocate()
if err != nil {
return err
}
defer func() {
if err != nil {
utils.ConReqs.Deallocate(err.Error())
} else {
utils.ConReqs.Deallocate(utils.EmptyString)
}
}()
return ssv1.Ss.BiRPCv1AuthorizeEvent(clnt, args, rply)
}
func (ssv1 *SessionSv1) BiRPCv1AuthorizeEventWithDigest(clnt *rpc2.Client, args *sessions.V1AuthorizeArgs,
rply *sessions.V1AuthorizeReplyWithDigest) error {
err := utils.ConReqs.Allocate()
if err != nil {
return err
}
defer func() {
if err != nil {
utils.ConReqs.Deallocate(err.Error())
} else {
utils.ConReqs.Deallocate(utils.EmptyString)
}
}()
return ssv1.Ss.BiRPCv1AuthorizeEventWithDigest(clnt, args, rply)
}
func (ssv1 *SessionSv1) BiRPCv1InitiateSession(clnt *rpc2.Client, args *sessions.V1InitSessionArgs,
rply *sessions.V1InitSessionReply) error {
err := utils.ConReqs.Allocate()
if err != nil {
return err
}
defer func() {
if err != nil {
utils.ConReqs.Deallocate(err.Error())
} else {
utils.ConReqs.Deallocate(utils.EmptyString)
}
}()
return ssv1.Ss.BiRPCv1InitiateSession(clnt, args, rply)
}
func (ssv1 *SessionSv1) BiRPCv1InitiateSessionWithDigest(clnt *rpc2.Client, args *sessions.V1InitSessionArgs,
rply *sessions.V1InitReplyWithDigest) error {
err := utils.ConReqs.Allocate()
if err != nil {
return err
}
defer func() {
if err != nil {
utils.ConReqs.Deallocate(err.Error())
} else {
utils.ConReqs.Deallocate(utils.EmptyString)
}
}()
return ssv1.Ss.BiRPCv1InitiateSessionWithDigest(clnt, args, rply)
}
func (ssv1 *SessionSv1) BiRPCv1UpdateSession(clnt *rpc2.Client, args *sessions.V1UpdateSessionArgs,
rply *sessions.V1UpdateSessionReply) error {
err := utils.ConReqs.Allocate()
if err != nil {
return err
}
defer func() {
if err != nil {
utils.ConReqs.Deallocate(err.Error())
} else {
utils.ConReqs.Deallocate(utils.EmptyString)
}
}()
return ssv1.Ss.BiRPCv1UpdateSession(clnt, args, rply)
}
func (ssv1 *SessionSv1) BiRPCv1SyncSessions(clnt *rpc2.Client, args *utils.TenantWithArgDispatcher,
rply *string) error {
err := utils.ConReqs.Allocate()
if err != nil {
return err
}
defer func() {
if err != nil {
utils.ConReqs.Deallocate(err.Error())
} else {
utils.ConReqs.Deallocate(utils.EmptyString)
}
}()
return ssv1.Ss.BiRPCv1SyncSessions(clnt, &utils.TenantWithArgDispatcher{}, rply)
}
func (ssv1 *SessionSv1) BiRPCv1TerminateSession(clnt *rpc2.Client, args *sessions.V1TerminateSessionArgs,
rply *string) error {
err := utils.ConReqs.Allocate()
if err != nil {
return err
}
defer func() {
if err != nil {
utils.ConReqs.Deallocate(err.Error())
} else {
utils.ConReqs.Deallocate(utils.EmptyString)
}
}()
return ssv1.Ss.BiRPCv1TerminateSession(clnt, args, rply)
}
func (ssv1 *SessionSv1) BiRPCv1ProcessCDR(clnt *rpc2.Client, cgrEv *utils.CGREventWithArgDispatcher, rply *string) error {
err := utils.ConReqs.Allocate()
if err != nil {
return err
}
defer func() {
if err != nil {
utils.ConReqs.Deallocate(err.Error())
} else {
utils.ConReqs.Deallocate(utils.EmptyString)
}
}()
return ssv1.Ss.BiRPCv1ProcessCDR(clnt, cgrEv, rply)
}
func (ssv1 *SessionSv1) BiRPCv1ProcessMessage(clnt *rpc2.Client, args *sessions.V1ProcessMessageArgs,
rply *sessions.V1ProcessMessageReply) error {
err := utils.ConReqs.Allocate()
if err != nil {
return err
}
defer func() {
if err != nil {
utils.ConReqs.Deallocate(err.Error())
} else {
utils.ConReqs.Deallocate(utils.EmptyString)
}
}()
return ssv1.Ss.BiRPCv1ProcessMessage(clnt, args, rply)
}
func (ssv1 *SessionSv1) BiRPCv1ProcessEvent(clnt *rpc2.Client, args *sessions.V1ProcessEventArgs,
rply *sessions.V1ProcessEventReply) error {
err := utils.ConReqs.Allocate()
if err != nil {
return err
}
defer func() {
if err != nil {
utils.ConReqs.Deallocate(err.Error())
} else {
utils.ConReqs.Deallocate(utils.EmptyString)
}
}()
return ssv1.Ss.BiRPCv1ProcessEvent(clnt, args, rply)
}
func (ssv1 *SessionSv1) BiRPCv1GetCost(clnt *rpc2.Client, args *sessions.V1ProcessEventArgs,
rply *sessions.V1GetCostReply) error {
err := utils.ConReqs.Allocate()
if err != nil {
return err
}
defer func() {
if err != nil {
utils.ConReqs.Deallocate(err.Error())
} else {
utils.ConReqs.Deallocate(utils.EmptyString)
}
}()
return ssv1.Ss.BiRPCv1GetCost(clnt, args, rply)
}
func (ssv1 *SessionSv1) BiRPCv1GetActiveSessions(clnt *rpc2.Client, args *utils.SessionFilter,
rply *[]*sessions.ExternalSession) error {
err := utils.ConReqs.Allocate()
if err != nil {
return err
}
defer func() {
if err != nil {
utils.ConReqs.Deallocate(err.Error())
} else {
utils.ConReqs.Deallocate(utils.EmptyString)
}
}()
return ssv1.Ss.BiRPCv1GetActiveSessions(clnt, args, rply)
}
func (ssv1 *SessionSv1) BiRPCv1GetActiveSessionsCount(clnt *rpc2.Client, args *utils.SessionFilter,
rply *int) error {
err := utils.ConReqs.Allocate()
if err != nil {
return err
}
defer func() {
if err != nil {
utils.ConReqs.Deallocate(err.Error())
} else {
utils.ConReqs.Deallocate(utils.EmptyString)
}
}()
return ssv1.Ss.BiRPCv1GetActiveSessionsCount(clnt, args, rply)
}
func (ssv1 *SessionSv1) BiRPCv1GetPassiveSessions(clnt *rpc2.Client, args *utils.SessionFilter,
rply *[]*sessions.ExternalSession) error {
err := utils.ConReqs.Allocate()
if err != nil {
return err
}
defer func() {
if err != nil {
utils.ConReqs.Deallocate(err.Error())
} else {
utils.ConReqs.Deallocate(utils.EmptyString)
}
}()
return ssv1.Ss.BiRPCv1GetPassiveSessions(clnt, args, rply)
}
func (ssv1 *SessionSv1) BiRPCv1GetPassiveSessionsCount(clnt *rpc2.Client, args *utils.SessionFilter,
rply *int) error {
err := utils.ConReqs.Allocate()
if err != nil {
return err
}
defer func() {
if err != nil {
utils.ConReqs.Deallocate(err.Error())
} else {
utils.ConReqs.Deallocate(utils.EmptyString)
}
}()
return ssv1.Ss.BiRPCv1GetPassiveSessionsCount(clnt, args, rply)
}
func (ssv1 *SessionSv1) BiRPCv1ForceDisconnect(clnt *rpc2.Client, args *utils.SessionFilter,
rply *string) error {
err := utils.ConReqs.Allocate()
if err != nil {
return err
}
defer func() {
if err != nil {
utils.ConReqs.Deallocate(err.Error())
} else {
utils.ConReqs.Deallocate(utils.EmptyString)
}
}()
return ssv1.Ss.BiRPCv1ForceDisconnect(clnt, args, rply)
}
func (ssv1 *SessionSv1) BiRPCv1RegisterInternalBiJSONConn(clnt *rpc2.Client, args string,
rply *string) error {
err := utils.ConReqs.Allocate()
if err != nil {
return err
}
defer func() {
if err != nil {
utils.ConReqs.Deallocate(err.Error())
} else {
utils.ConReqs.Deallocate(utils.EmptyString)
}
}()
return ssv1.Ss.BiRPCv1RegisterInternalBiJSONConn(clnt, args, rply)
}
func (ssv1 *SessionSv1) BiRPCPing(clnt *rpc2.Client, ign *utils.CGREventWithArgDispatcher, reply *string) error {
err := utils.ConReqs.Allocate()
if err != nil {
return err
}
defer func() {
if err != nil {
utils.ConReqs.Deallocate(err.Error())
} else {
utils.ConReqs.Deallocate(utils.EmptyString)
}
}()
return ssv1.Ping(ign, reply)
}
func (ssv1 *SessionSv1) BiRPCv1ReplicateSessions(clnt *rpc2.Client,
args sessions.ArgsReplicateSessions, reply *string) error {
err := utils.ConReqs.Allocate()
if err != nil {
return err
}
defer func() {
if err != nil {
utils.ConReqs.Deallocate(err.Error())
} else {
utils.ConReqs.Deallocate(utils.EmptyString)
}
}()
return ssv1.BiRPCv1ReplicateSessions(clnt, args, reply)
}
func (ssv1 *SessionSv1) BiRPCv1SetPassiveSession(clnt *rpc2.Client,
args *sessions.Session, reply *string) error {
err := utils.ConReqs.Allocate()
if err != nil {
return err
}
defer func() {
if err != nil {
utils.ConReqs.Deallocate(err.Error())
} else {
utils.ConReqs.Deallocate(utils.EmptyString)
}
}()
return ssv1.Ss.BiRPCv1SetPassiveSession(clnt, args, reply)
}
func (ssv1 *SessionSv1) BiRPCv1ActivateSessions(clnt *rpc2.Client,
args *utils.SessionIDsWithArgsDispatcher, reply *string) error {
err := utils.ConReqs.Allocate()
if err != nil {
return err
}
defer func() {
if err != nil {
utils.ConReqs.Deallocate(err.Error())
} else {
utils.ConReqs.Deallocate(utils.EmptyString)
}
}()
return ssv1.Ss.BiRPCv1ActivateSessions(clnt, args, reply)
}
func (ssv1 *SessionSv1) BiRPCv1DeactivateSessions(clnt *rpc2.Client,
args *utils.SessionIDsWithArgsDispatcher, reply *string) error {
err := utils.ConReqs.Allocate()
if err != nil {
return err
}
defer func() {
if err != nil {
utils.ConReqs.Deallocate(err.Error())
} else {
utils.ConReqs.Deallocate(utils.EmptyString)
}
}()
return ssv1.Ss.BiRPCv1DeactivateSessions(clnt, args, reply)
}
// BiRPCV1ReAuthorize sends the RAR for filterd sessions
func (ssv1 *SessionSv1) BiRPCV1ReAuthorize(clnt *rpc2.Client,
args *utils.SessionFilter, reply *string) error {
err := utils.ConReqs.Allocate()
if err != nil {
return err
}
defer func() {
if err != nil {
utils.ConReqs.Deallocate(err.Error())
} else {
utils.ConReqs.Deallocate(utils.EmptyString)
}
}()
return ssv1.Ss.BiRPCv1ReAuthorize(clnt, args, reply)
}
// BiRPCV1DisconnectPeer sends the DPR for the OriginHost and OriginRealm
func (ssv1 *SessionSv1) BiRPCV1DisconnectPeer(clnt *rpc2.Client,
args *utils.DPRArgs, reply *string) error {
err := utils.ConReqs.Allocate()
if err != nil {
return err
}
defer func() {
if err != nil {
utils.ConReqs.Deallocate(err.Error())
} else {
utils.ConReqs.Deallocate(utils.EmptyString)
}
}()
return ssv1.Ss.BiRPCv1DisconnectPeer(clnt, args, reply)
}
// BiRPCV1STIRAuthenticate checks the identity using STIR/SHAKEN
func (ssv1 *SessionSv1) BiRPCV1STIRAuthenticate(clnt *rpc2.Client,
args *sessions.V1STIRAuthenticateArgs, reply *string) error {
err := utils.ConReqs.Allocate()
if err != nil {
return err
}
defer func() {
if err != nil {
utils.ConReqs.Deallocate(err.Error())
} else {
utils.ConReqs.Deallocate(utils.EmptyString)
}
}()
return ssv1.Ss.BiRPCv1STIRAuthenticate(clnt, args, reply)
}
// BiRPCV1STIRIdentity creates the identity for STIR/SHAKEN
func (ssv1 *SessionSv1) BiRPCV1STIRIdentity(clnt *rpc2.Client,
args *sessions.V1STIRIdentityArgs, reply *string) error {
err := utils.ConReqs.Allocate()
if err != nil {
return err
}
defer func() {
if err != nil {
utils.ConReqs.Deallocate(err.Error())
} else {
utils.ConReqs.Deallocate(utils.EmptyString)
}
}()
return ssv1.Ss.BiRPCv1STIRIdentity(nil, args, reply)
}
func (ssv1 *SessionSv1) BiRPCV1Sleep(clnt *rpc2.Client, arg *DurationArgs, reply *string) error {
err := utils.ConReqs.Allocate()
if err != nil {
return err
}
defer func() {
if err != nil {
utils.ConReqs.Deallocate(err.Error())
} else {
utils.ConReqs.Deallocate(utils.EmptyString)
}
}()
time.Sleep(arg.DurationTime)
*reply = utils.OK
return nil

56
apier/v1/smgbirpc.go
View File

@@ -20,6 +20,7 @@ package v1
import (
"github.com/cenkalti/rpc2"
"github.com/cgrates/cgrates/utils"
)
// Publishes methods exported by SMGenericV1 as SMGenericV1 (so we can handle standard RPC methods via birpc socket)
@@ -36,29 +37,84 @@ func (smgv1 *SMGenericV1) Handlers() map[string]interface{} {
/// Returns MaxUsage (for calls in seconds), -1 for no limit
func (smgv1 *SMGenericV1) BiRPCV1GetMaxUsage(clnt *rpc2.Client,
ev map[string]interface{}, maxUsage *float64) error {
err := utils.ConReqs.Allocate()
if err != nil {
return err
}
defer func() {
if err != nil {
utils.ConReqs.Deallocate(err.Error())
} else {
utils.ConReqs.Deallocate(utils.EmptyString)
}
}()
return smgv1.Ss.BiRPCV1GetMaxUsage(clnt, ev, maxUsage)
}
// Called on session start, returns the maximum number of seconds the session can last
func (smgv1 *SMGenericV1) BiRPCV1InitiateSession(clnt *rpc2.Client,
ev map[string]interface{}, maxUsage *float64) error {
err := utils.ConReqs.Allocate()
if err != nil {
return err
}
defer func() {
if err != nil {
utils.ConReqs.Deallocate(err.Error())
} else {
utils.ConReqs.Deallocate(utils.EmptyString)
}
}()
return smgv1.Ss.BiRPCV1InitiateSession(clnt, ev, maxUsage)
}
// Interim updates, returns remaining duration from the rater
func (smgv1 *SMGenericV1) BiRPCV1UpdateSession(clnt *rpc2.Client,
ev map[string]interface{}, maxUsage *float64) error {
err := utils.ConReqs.Allocate()
if err != nil {
return err
}
defer func() {
if err != nil {
utils.ConReqs.Deallocate(err.Error())
} else {
utils.ConReqs.Deallocate(utils.EmptyString)
}
}()
return smgv1.Ss.BiRPCV1UpdateSession(clnt, ev, maxUsage)
}
// Called on session end, should stop debit loop
func (smgv1 *SMGenericV1) BiRPCV1TerminateSession(clnt *rpc2.Client,
ev map[string]interface{}, reply *string) error {
err := utils.ConReqs.Allocate()
if err != nil {
return err
}
defer func() {
if err != nil {
utils.ConReqs.Deallocate(err.Error())
} else {
utils.ConReqs.Deallocate(utils.EmptyString)
}
}()
return smgv1.Ss.BiRPCV1TerminateSession(clnt, ev, reply)
}
// Called on session end, should send the CDR to CDRS
func (smgv1 *SMGenericV1) BiRPCV1ProcessCDR(clnt *rpc2.Client,
ev map[string]interface{}, reply *string) error {
err := utils.ConReqs.Allocate()
if err != nil {
return err
}
defer func() {
if err != nil {
utils.ConReqs.Deallocate(err.Error())
} else {
utils.ConReqs.Deallocate(utils.EmptyString)
}
}()
return smgv1.Ss.BiRPCV1ProcessCDR(clnt, ev, reply)
}

210
utils/concreqs_bijson_codec.go
View File

@@ -1,210 +0,0 @@
/*
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/>
*/
// Most of the logic follows standard library implementation in this file
package utils
import (
"encoding/json"
"errors"
"fmt"
"io"
"sync"
"github.com/cenkalti/rpc2"
)
type concReqsBiJSONCoded struct {
dec *json.Decoder // for reading JSON values
enc *json.Encoder // for writing JSON values
c io.Closer
// temporary work space
msg message
serverRequest serverRequest
clientResponse clientResponse
// JSON-RPC clients can use arbitrary json values as request IDs.
// Package rpc expects uint64 request IDs.
// We assign uint64 sequence numbers to incoming requests
// but save the original request ID in the pending map.
// When rpc responds, we use the sequence number in
// the response to find the original request ID.
mutex sync.Mutex // protects seq, pending
pending map[uint64]*json.RawMessage
seq uint64
}
// NewConcReqsBiJSONCoded returns a new rpc2.Codec using JSON-RPC on conn.
func NewConcReqsBiJSONCoded(conn io.ReadWriteCloser) rpc2.Codec {
return &concReqsBiJSONCoded{
dec: json.NewDecoder(conn),
enc: json.NewEncoder(conn),
c: conn,
pending: make(map[uint64]*json.RawMessage),
}
}
// serverRequest and clientResponse combined
type message struct {
Method string `json:"method"`
Params *json.RawMessage `json:"params"`
Id *json.RawMessage `json:"id"`
Result *json.RawMessage `json:"result"`
Error interface{} `json:"error"`
}
type clientResponse struct {
Id uint64 `json:"id"`
Result *json.RawMessage `json:"result"`
Error interface{} `json:"error"`
}
type clientRequest struct {
Method string `json:"method"`
Params []interface{} `json:"params"`
Id *uint64 `json:"id"`
}
func (c *concReqsBiJSONCoded) ReadHeader(req *rpc2.Request, resp *rpc2.Response) error {
c.msg = message{}
if err := c.dec.Decode(&c.msg); err != nil {
return err
}
if c.msg.Method != "" {
// request comes to server
c.serverRequest.Id = c.msg.Id
c.serverRequest.Method = c.msg.Method
c.serverRequest.Params = c.msg.Params
req.Method = c.serverRequest.Method
// JSON request id can be any JSON value;
// RPC package expects uint64. Translate to
// internal uint64 and save JSON on the side.
if c.serverRequest.Id == nil {
// Notification
} else {
c.mutex.Lock()
c.seq++
c.pending[c.seq] = c.serverRequest.Id
c.serverRequest.Id = nil
req.Seq = c.seq
c.mutex.Unlock()
}
} else {
// response comes to client
err := json.Unmarshal(*c.msg.Id, &c.clientResponse.Id)
if err != nil {
return err
}
c.clientResponse.Result = c.msg.Result
c.clientResponse.Error = c.msg.Error
resp.Error = ""
resp.Seq = c.clientResponse.Id
if c.clientResponse.Error != nil || c.clientResponse.Result == nil {
x, ok := c.clientResponse.Error.(string)
if !ok {
return fmt.Errorf("invalid error %v", c.clientResponse.Error)
}
if x == "" {
x = "unspecified error"
}
resp.Error = x
}
}
return nil
}
func (c *concReqsBiJSONCoded) ReadRequestBody(x interface{}) error {
if err := ConReqs.Allocate(); err != nil {
return err
}
if x == nil {
return nil
}
if c.serverRequest.Params == nil {
return errMissingParams
}
var params *[]interface{}
switch x := x.(type) {
case *[]interface{}:
params = x
default:
params = &[]interface{}{x}
}
return json.Unmarshal(*c.serverRequest.Params, params)
}
func (c *concReqsBiJSONCoded) ReadResponseBody(x interface{}) error {
if err := ConReqs.Allocate(); err != nil {
return err
}
if x == nil {
return nil
}
return json.Unmarshal(*c.clientResponse.Result, x)
}
func (c *concReqsBiJSONCoded) WriteRequest(r *rpc2.Request, param interface{}) error {
req := &clientRequest{Method: r.Method}
switch param := param.(type) {
case []interface{}:
req.Params = param
default:
req.Params = []interface{}{param}
}
if r.Seq == 0 {
// Notification
req.Id = nil
} else {
seq := r.Seq
req.Id = &seq
}
return c.enc.Encode(req)
}
func (c *concReqsBiJSONCoded) WriteResponse(r *rpc2.Response, x interface{}) error {
defer ConReqs.Deallocate(r.Error)
c.mutex.Lock()
b, ok := c.pending[r.Seq]
if !ok {
c.mutex.Unlock()
return errors.New("invalid sequence number in response")
}
delete(c.pending, r.Seq)
c.mutex.Unlock()
if b == nil {
// Invalid request so no id. Use JSON null.
b = &null
}
resp := serverResponse{Id: b}
if r.Error == "" {
resp.Result = x
} else {
resp.Error = r.Error
}
return c.enc.Encode(resp)
}
func (c *concReqsBiJSONCoded) Close() error {
return c.c.Close()
}

1
utils/concureqs.go
View File

@@ -45,6 +45,7 @@ func NewConReqs(reqs int, strategy string) *ConcReqs {
var errDeny = fmt.Errorf("denying request due to maximum active requests reached")
func (cR *ConcReqs) Allocate() (err error) {
fmt.Println("ENTER IN ALLOCATE ")
if cR.nAReqs == 0 {
return
}

4
utils/server.go
View File

@@ -37,6 +37,8 @@ import (
"sync"
"time"
rpc2_jsonrpc "github.com/cenkalti/rpc2/jsonrpc"
"github.com/cenkalti/rpc2"
"golang.org/x/net/websocket"
)
@@ -327,7 +329,7 @@ func (s *Server) ServeBiJSON(addr string, onConn func(*rpc2.Client), onDis func(
log.Fatal(err)
return // stop if we get Accept error
}
go s.birpcSrv.ServeCodec(NewConcReqsBiJSONCoded(conn))
go s.birpcSrv.ServeCodec(rpc2_jsonrpc.NewJSONCodec(conn))
}
}(lBiJSON)
<-s.stopbiRPCServer // wait until server is stoped to close the listener