Skip to content
Snippets Groups Projects
Commit eeb99207 authored by AlexStocks's avatar AlexStocks
Browse files

Rem: delete out of date files

parent fb9c5ed5
No related branches found
No related tags found
No related merge requests found
Hide whitespace changes
Inline Side-by-side
dubbo/client.go deleted 100644 → 0
+ 0
− 274
View file @ fb9c5ed5
package rpc
import (
"math/rand"
"sync"
"time"
)
import (
"github.com/AlexStocks/getty"
"github.com/AlexStocks/goext/sync/atomic"
jerrors "github.com/juju/errors"
)
var (
errInvalidCodecType = jerrors.New("illegal CodecType")
errInvalidAddress = jerrors.New("remote address invalid or empty")
errSessionNotExist = jerrors.New("session not exist")
errClientClosed = jerrors.New("client closed")
errClientReadTimeout = jerrors.New("client read timeout")
)
func init() {
rand.Seed(time.Now().UnixNano())
}
type CallOptions struct {
// request timeout
RequestTimeout time.Duration
// response timeout
ResponseTimeout time.Duration
Meta map[interface{}]interface{}
}
type CallOption func(*CallOptions)
func WithCallRequestTimeout(d time.Duration) CallOption {
return func(o *CallOptions) {
o.RequestTimeout = d
}
}
func WithCallResponseTimeout(d time.Duration) CallOption {
return func(o *CallOptions) {
o.ResponseTimeout = d
}
}
func WithCallMeta(k, v interface{}) CallOption {
return func(o *CallOptions) {
if o.Meta == nil {
o.Meta = make(map[interface{}]interface{})
}
o.Meta[k] = v
}
}
type CallResponse struct {
Opts CallOptions
Cause error
Start time.Time // invoke(call) start time == write start time
ReadStart time.Time // read start time, write duration = ReadStart - Start
Reply interface{}
}
type AsyncCallback func(response CallResponse)
type Client struct {
conf ClientConfig
pool *gettyRPCClientPool
sequence gxatomic.Uint64
pendingLock sync.RWMutex
pendingResponses map[SequenceType]*PendingResponse
}
func NewClient(conf *ClientConfig) (*Client, error) {
if err := conf.CheckValidity(); err != nil {
return nil, jerrors.Trace(err)
}
c := &Client{
pendingResponses: make(map[SequenceType]*PendingResponse),
conf: *conf,
}
c.pool = newGettyRPCClientConnPool(c, conf.PoolSize, time.Duration(int(time.Second)*conf.PoolTTL))
return c, nil
}
// call one way
func (c *Client) CallOneway(typ CodecType, addr, service, method string, args interface{}, opts ...CallOption) error {
var copts CallOptions
for _, o := range opts {
o(&copts)
}
return jerrors.Trace(c.call(CT_OneWay, typ, addr, service, method, args, nil, nil, copts))
}
// if @reply is nil, the transport layer will get the response without notify the invoker.
func (c *Client) Call(typ CodecType, addr, service, method string, args, reply interface{}, opts ...CallOption) error {
var copts CallOptions
for _, o := range opts {
o(&copts)
}
ct := CT_TwoWay
if reply == nil {
ct = CT_TwoWayNoReply
}
return jerrors.Trace(c.call(ct, typ, addr, service, method, args, reply, nil, copts))
}
func (c *Client) AsyncCall(typ CodecType, addr, service, method string, args interface{},
callback AsyncCallback, reply interface{}, opts ...CallOption) error {
var copts CallOptions
for _, o := range opts {
o(&copts)
}
return jerrors.Trace(c.call(CT_TwoWay, typ, addr, service, method, args, reply, callback, copts))
}
func (c *Client) call(ct CallType, typ CodecType, addr, service, method string,
args, reply interface{}, callback AsyncCallback, opts CallOptions) error {
if opts.RequestTimeout == 0 {
opts.RequestTimeout = c.conf.GettySessionParam.tcpWriteTimeout
}
if opts.ResponseTimeout == 0 {
opts.ResponseTimeout = c.conf.GettySessionParam.tcpReadTimeout
}
if !typ.CheckValidity() {
return errInvalidCodecType
}
b := &GettyRPCRequest{}
b.header.Service = service
b.header.Method = method
b.header.CallType = ct
b.body = args
var rsp *PendingResponse
if ct != CT_OneWay {
rsp = NewPendingResponse()
rsp.reply = reply
rsp.callback = callback
rsp.opts = opts
}
var (
err error
session getty.Session
conn *gettyRPCClient
)
conn, session, err = c.selectSession(typ, addr)
if err != nil || session == nil {
return errSessionNotExist
}
defer c.pool.release(conn, err)
if err = c.transfer(session, typ, b, rsp, opts); err != nil {
return jerrors.Trace(err)
}
if ct == CT_OneWay || callback != nil {
return nil
}
select {
case <-getty.GetTimeWheel().After(opts.ResponseTimeout):
err = errClientReadTimeout
c.removePendingResponse(SequenceType(rsp.seq))
case <-rsp.done:
err = rsp.err
}
return jerrors.Trace(err)
}
func (c *Client) Close() {
if c.pool != nil {
c.pool.close()
}
c.pool = nil
}
func (c *Client) selectSession(typ CodecType, addr string) (*gettyRPCClient, getty.Session, error) {
rpcConn, err := c.pool.getConn(typ.String(), addr)
if err != nil {
return nil, nil, jerrors.Trace(err)
}
return rpcConn, rpcConn.selectSession(), nil
}
func (c *Client) heartbeat(session getty.Session, typ CodecType) error {
return c.transfer(session, typ, nil, NewPendingResponse(), CallOptions{})
}
func (c *Client) transfer(session getty.Session, typ CodecType, req *GettyRPCRequest,
rsp *PendingResponse, opts CallOptions) error {
var (
sequence uint64
err error
pkg GettyPackage
)
sequence = c.sequence.Add(1)
pkg.H.Magic = MagicType(gettyPackageMagic)
pkg.H.LogID = LogIDType(randomID())
pkg.H.Sequence = SequenceType(sequence)
pkg.H.Command = gettyCmdHbRequest
pkg.H.CodecType = typ
if req != nil {
pkg.H.Command = gettyCmdRPCRequest
pkg.B = req
}
// cond1
if rsp != nil {
rsp.seq = sequence
c.addPendingResponse(rsp)
}
err = session.WritePkg(pkg, opts.RequestTimeout)
if err != nil {
c.removePendingResponse(SequenceType(rsp.seq))
} else if rsp != nil { // cond2
// cond2 should not merged with cond1. cause the response package may be returned very
// soon and it will be handled by other goroutine.
rsp.readStart = time.Now()
}
return jerrors.Trace(err)
}
// func (c *Client) PendingResponseCount() int {
// c.pendingLock.RLock()
// defer c.pendingLock.RUnlock()
// return len(c.pendingResponses)
// }
func (c *Client) addPendingResponse(pr *PendingResponse) {
c.pendingLock.Lock()
defer c.pendingLock.Unlock()
c.pendingResponses[SequenceType(pr.seq)] = pr
}
func (c *Client) removePendingResponse(seq SequenceType) *PendingResponse {
c.pendingLock.Lock()
defer c.pendingLock.Unlock()
if c.pendingResponses == nil {
return nil
}
if presp, ok := c.pendingResponses[seq]; ok {
delete(c.pendingResponses, seq)
return presp
}
return nil
}
// func (c *Client) ClearPendingResponses() map[SequenceType]*PendingResponse {
// c.pendingLock.Lock()
// defer c.pendingLock.Unlock()
// presps := c.pendingResponses
// c.pendingResponses = nil
// return presps
// }
dubbo/config.go deleted 100644 → 0
+ 0
− 101
View file @ fb9c5ed5
package rpc
import (
"time"
)
import (
jerrors "github.com/juju/errors"
)
type (
GettySessionParam struct {
CompressEncoding bool `default:"false" yaml:"compress_encoding" json:"compress_encoding,omitempty"`
TcpNoDelay bool `default:"true" yaml:"tcp_no_delay" json:"tcp_no_delay,omitempty"`
TcpKeepAlive bool `default:"true" yaml:"tcp_keep_alive" json:"tcp_keep_alive,omitempty"`
KeepAlivePeriod string `default:"180s" yaml:"keep_alive_period" json:"keep_alive_period,omitempty"`
keepAlivePeriod time.Duration
TcpRBufSize int `default:"262144" yaml:"tcp_r_buf_size" json:"tcp_r_buf_size,omitempty"`
TcpWBufSize int `default:"65536" yaml:"tcp_w_buf_size" json:"tcp_w_buf_size,omitempty"`
PkgRQSize int `default:"1024" yaml:"pkg_rq_size" json:"pkg_rq_size,omitempty"`
PkgWQSize int `default:"1024" yaml:"pkg_wq_size" json:"pkg_wq_size,omitempty"`
TcpReadTimeout string `default:"1s" yaml:"tcp_read_timeout" json:"tcp_read_timeout,omitempty"`
tcpReadTimeout time.Duration
TcpWriteTimeout string `default:"5s" yaml:"tcp_write_timeout" json:"tcp_write_timeout,omitempty"`
tcpWriteTimeout time.Duration
WaitTimeout string `default:"7s" yaml:"wait_timeout" json:"wait_timeout,omitempty"`
waitTimeout time.Duration
MaxMsgLen int `default:"1024" yaml:"max_msg_len" json:"max_msg_len,omitempty"`
SessionName string `default:"rpc" yaml:"session_name" json:"session_name,omitempty"`
}
// Config holds supported types by the multiconfig package
ClientConfig struct {
// local address
AppName string `default:"rpc-client" yaml:"app_name" json:"app_name,omitempty"`
Host string `default:"127.0.0.1" yaml:"host" json:"host,omitempty"`
ProfilePort int `default:"10086" yaml:"profile_port" json:"profile_port,omitempty"`
// session pool
ConnectionNum int `default:"16" yaml:"connection_number" json:"connection_number,omitempty"`
// heartbeat
HeartbeatPeriod string `default:"15s" yaml:"heartbeat_period" json:"heartbeat_period,omitempty"`
heartbeatPeriod time.Duration
// session
SessionTimeout string `default:"60s" yaml:"session_timeout" json:"session_timeout,omitempty"`
sessionTimeout time.Duration
// app
FailFastTimeout string `default:"5s" yaml:"fail_fast_timeout" json:"fail_fast_timeout,omitempty"`
failFastTimeout time.Duration
// Connection Pool
PoolSize int `default:"2" yaml:"pool_size" json:"pool_size,omitempty"`
PoolTTL int `default:"180" yaml:"pool_ttl" json:"pool_ttl,omitempty"`
// session tcp parameters
GettySessionParam GettySessionParam `required:"true" yaml:"getty_session_param" json:"getty_session_param,omitempty"`
}
)
func (c *GettySessionParam) CheckValidity() error {
var err error
if c.keepAlivePeriod, err = time.ParseDuration(c.KeepAlivePeriod); err != nil {
return jerrors.Annotatef(err, "time.ParseDuration(KeepAlivePeriod{%#v})", c.KeepAlivePeriod)
}
if c.tcpReadTimeout, err = time.ParseDuration(c.TcpReadTimeout); err != nil {
return jerrors.Annotatef(err, "time.ParseDuration(TcpReadTimeout{%#v})", c.TcpReadTimeout)
}
if c.tcpWriteTimeout, err = time.ParseDuration(c.TcpWriteTimeout); err != nil {
return jerrors.Annotatef(err, "time.ParseDuration(TcpWriteTimeout{%#v})", c.TcpWriteTimeout)
}
if c.waitTimeout, err = time.ParseDuration(c.WaitTimeout); err != nil {
return jerrors.Annotatef(err, "time.ParseDuration(WaitTimeout{%#v})", c.WaitTimeout)
}
return nil
}
func (c *ClientConfig) CheckValidity() error {
var err error
if c.heartbeatPeriod, err = time.ParseDuration(c.HeartbeatPeriod); err != nil {
return jerrors.Annotatef(err, "time.ParseDuration(HeartbeatPeroid{%#v})", c.HeartbeatPeriod)
}
if c.sessionTimeout, err = time.ParseDuration(c.SessionTimeout); err != nil {
return jerrors.Annotatef(err, "time.ParseDuration(SessionTimeout{%#v})", c.SessionTimeout)
}
if c.failFastTimeout, err = time.ParseDuration(c.FailFastTimeout); err != nil {
return jerrors.Annotatef(err, "time.ParseDuration(FailFastTimeout{%#v})", c.FailFastTimeout)
}
return jerrors.Trace(c.GettySessionParam.CheckValidity())
}
dubbo/pool.go deleted 100644 → 0
+ 0
− 351
View file @ fb9c5ed5
package rpc
import (
"fmt"
"math/rand"
"net"
"strings"
"sync"
"time"
)
import (
"github.com/AlexStocks/getty"
log "github.com/AlexStocks/log4go"
jerrors "github.com/juju/errors"
)
type rpcSession struct {
session getty.Session
reqNum int32
}
type gettyRPCClient struct {
once sync.Once
protocol string
addr string
created int64 // 为0，则说明没有被创建或者被销毁了
pool *gettyRPCClientPool
lock sync.RWMutex
gettyClient getty.Client
sessions []*rpcSession
}
var (
errClientPoolClosed = jerrors.New("client pool closed")
)
func newGettyRPCClientConn(pool *gettyRPCClientPool, protocol, addr string) (*gettyRPCClient, error) {
c := &gettyRPCClient{
protocol: protocol,
addr: addr,
pool: pool,
gettyClient: getty.NewTCPClient(
getty.WithServerAddress(addr),
getty.WithConnectionNumber((int)(pool.rpcClient.conf.ConnectionNum)),
),
}
c.gettyClient.RunEventLoop(c.newSession)
idx := 1
for {
idx++
if c.isAvailable() {
break
}
if idx > 5000 {
return nil, jerrors.New(fmt.Sprintf("failed to create client connection to %s in 5 seconds", addr))
}
time.Sleep(1e6)
}
log.Info("client init ok")
c.created = time.Now().Unix()
return c, nil
}
func (c *gettyRPCClient) newSession(session getty.Session) error {
var (
ok bool
tcpConn *net.TCPConn
conf ClientConfig
)
conf = c.pool.rpcClient.conf
if conf.GettySessionParam.CompressEncoding {
session.SetCompressType(getty.CompressZip)
}
if tcpConn, ok = session.Conn().(*net.TCPConn); !ok {
panic(fmt.Sprintf("%s, session.conn{%#v} is not tcp connection\n", session.Stat(), session.Conn()))
}
tcpConn.SetNoDelay(conf.GettySessionParam.TcpNoDelay)
tcpConn.SetKeepAlive(conf.GettySessionParam.TcpKeepAlive)
if conf.GettySessionParam.TcpKeepAlive {
tcpConn.SetKeepAlivePeriod(conf.GettySessionParam.keepAlivePeriod)
}
tcpConn.SetReadBuffer(conf.GettySessionParam.TcpRBufSize)
tcpConn.SetWriteBuffer(conf.GettySessionParam.TcpWBufSize)
session.SetName(conf.GettySessionParam.SessionName)
session.SetMaxMsgLen(conf.GettySessionParam.MaxMsgLen)
session.SetPkgHandler(NewRpcClientPackageHandler())
session.SetEventListener(NewRpcClientHandler(c))
session.SetRQLen(conf.GettySessionParam.PkgRQSize)
session.SetWQLen(conf.GettySessionParam.PkgWQSize)
session.SetReadTimeout(conf.GettySessionParam.tcpReadTimeout)
session.SetWriteTimeout(conf.GettySessionParam.tcpWriteTimeout)
session.SetCronPeriod((int)(conf.heartbeatPeriod.Nanoseconds() / 1e6))
session.SetWaitTime(conf.GettySessionParam.waitTimeout)
log.Debug("client new session:%s\n", session.Stat())
return nil
}
func (c *gettyRPCClient) selectSession() getty.Session {
c.lock.RLock()
defer c.lock.RUnlock()
if c.sessions == nil {
return nil
}
count := len(c.sessions)
if count == 0 {
return nil
}
return c.sessions[rand.Int31n(int32(count))].session
}
func (c *gettyRPCClient) addSession(session getty.Session) {
log.Debug("add session{%s}", session.Stat())
if session == nil {
return
}
c.lock.Lock()
c.sessions = append(c.sessions, &rpcSession{session: session})
c.lock.Unlock()
}
func (c *gettyRPCClient) removeSession(session getty.Session) {
if session == nil {
return
}
c.lock.Lock()
defer c.lock.Unlock()
if c.sessions == nil {
return
}
for i, s := range c.sessions {
if s.session == session {
c.sessions = append(c.sessions[:i], c.sessions[i+1:]...)
log.Debug("delete session{%s}, its index{%d}", session.Stat(), i)
break
}
}
log.Info("after remove session{%s}, left session number:%d", session.Stat(), len(c.sessions))
if len(c.sessions) == 0 {
c.close() // -> pool.remove(c)
}
}
func (c *gettyRPCClient) updateSession(session getty.Session) {
if session == nil {
return
}
c.lock.Lock()
defer c.lock.Unlock()
if c.sessions == nil {
return
}
for i, s := range c.sessions {
if s.session == session {
c.sessions[i].reqNum++
break
}
}
}
func (c *gettyRPCClient) getClientRpcSession(session getty.Session) (rpcSession, error) {
var (
err error
rpcSession rpcSession
)
c.lock.Lock()
defer c.lock.Unlock()
if c.sessions == nil {
return rpcSession, errClientClosed
}
err = errSessionNotExist
for _, s := range c.sessions {
if s.session == session {
rpcSession = *s
err = nil
break
}
}
return rpcSession, jerrors.Trace(err)
}
func (c *gettyRPCClient) isAvailable() bool {
if c.selectSession() == nil {
return false
}
return true
}
func (c *gettyRPCClient) close() error {
err := jerrors.Errorf("close gettyRPCClient{%#v} again", c)
c.once.Do(func() {
// delete @c from client pool
c.pool.remove(c)
for _, s := range c.sessions {
log.Info("close client session{%s, last active:%s, request number:%d}",
s.session.Stat(), s.session.GetActive().String(), s.reqNum)
s.session.Close()
}
c.gettyClient.Close()
c.gettyClient = nil
c.sessions = c.sessions[:0]
c.created = 0
err = nil
})
return err
}
type gettyRPCClientPool struct {
rpcClient *Client
size int // []*gettyRPCClient数组的size
ttl int64 // 每个gettyRPCClient的有效期时间. pool对象会在getConn时执行ttl检查
sync.Mutex
connMap map[string][]*gettyRPCClient // 从[]*gettyRPCClient 可见key是连接地址，而value是对应这个地址的连接数组
}
func newGettyRPCClientConnPool(rpcClient *Client, size int, ttl time.Duration) *gettyRPCClientPool {
return &gettyRPCClientPool{
rpcClient: rpcClient,
size: size,
ttl: int64(ttl.Seconds()),
connMap: make(map[string][]*gettyRPCClient),
}
}
func (p *gettyRPCClientPool) close() {
p.Lock()
connMap := p.connMap
p.connMap = nil
p.Unlock()
for _, connArray := range connMap {
for _, conn := range connArray {
conn.close()
}
}
}
func (p *gettyRPCClientPool) getConn(protocol, addr string) (*gettyRPCClient, error) {
var builder strings.Builder
builder.WriteString(addr)
builder.WriteString("@")
builder.WriteString(protocol)
key := builder.String()
p.Lock()
defer p.Unlock()
if p.connMap == nil {
return nil, errClientPoolClosed
}
connArray := p.connMap[key]
now := time.Now().Unix()
for len(connArray) > 0 {
conn := connArray[len(connArray)-1]
connArray = connArray[:len(connArray)-1]
p.connMap[key] = connArray
if d := now - conn.created; d > p.ttl {
conn.close() // -> pool.remove(c)
continue
}
return conn, nil
}
// create new conn
return newGettyRPCClientConn(p, protocol, addr)
}
func (p *gettyRPCClientPool) release(conn *gettyRPCClient, err error) {
if conn == nil || conn.created == 0 {
return
}
if err != nil {
conn.close()
return
}
var builder strings.Builder
builder.WriteString(conn.addr)
builder.WriteString("@")
builder.WriteString(conn.protocol)
key := builder.String()
p.Lock()
defer p.Unlock()
if p.connMap == nil {
return
}
connArray := p.connMap[key]
if len(connArray) >= p.size {
conn.close()
return
}
p.connMap[key] = append(connArray, conn)
}
func (p *gettyRPCClientPool) remove(conn *gettyRPCClient) {
if conn == nil || conn.created == 0 {
return
}
var builder strings.Builder
builder.WriteString(conn.addr)
builder.WriteString("@")
builder.WriteString(conn.protocol)
key := builder.String()
p.Lock()
defer p.Unlock()
if p.connMap == nil {
return
}
connArray := p.connMap[key]
if len(connArray) > 0 {
for idx, c := range connArray {
if conn == c {
p.connMap[key] = append(connArray[:idx], connArray[idx+1:]...)
break
}
}
}
}
dubbo/readwriter.go deleted 100644 → 0
+ 0
− 63
View file @ fb9c5ed5
package rpc
import (
"bytes"
"reflect"
)
import (
"github.com/AlexStocks/getty"
log "github.com/AlexStocks/log4go"
jerrors "github.com/juju/errors"
)
////////////////////////////////////////////
// RpcClientPackageHandler
////////////////////////////////////////////
type RpcClientPackageHandler struct {
}
func NewRpcClientPackageHandler() *RpcClientPackageHandler {
return &RpcClientPackageHandler{}
}
func (p *RpcClientPackageHandler) Read(ss getty.Session, data []byte) (interface{}, int, error) {
pkg := &GettyPackage{
B: NewGettyRPCResponse(),
}
buf := bytes.NewBuffer(data)
length, err := pkg.Unmarshal(buf)
if err != nil {
if err == ErrNotEnoughStream {
return nil, 0, nil
}
return nil, 0, jerrors.Trace(err)
}
resp := &GettyRPCResponsePackage{
H: pkg.H,
header: pkg.B.GetHeader().(GettyRPCResponseHeader),
}
if pkg.H.Command != gettyCmdHbResponse {
resp.body = pkg.B.GetBody()
}
return resp, length, nil
}
func (p *RpcClientPackageHandler) Write(ss getty.Session, pkg interface{}) error {
req, ok := pkg.(GettyPackage)
if !ok {
log.Error("illegal pkg:%+v\n", pkg)
return jerrors.New("invalid rpc request")
}
buf, err := req.Marshal()
if err != nil {
log.Warn("binary.Write(req{%#v}) = err{%#v}", req, jerrors.ErrorStack(err))
return jerrors.Trace(err)
}
return jerrors.Trace(ss.WriteBytes(buf.Bytes()))
}
dubbo/util.go deleted 100644 → 0
+ 0
− 19
View file @ fb9c5ed5
package rpc
import (
"math/rand"
"sync"
"time"
)
var (
seededIDGen = rand.New(rand.NewSource(time.Now().UnixNano()))
// The golang rand generators are *not* intrinsically thread-safe.
seededIDLock sync.Mutex
)
func randomID() uint64 {
seededIDLock.Lock()
defer seededIDLock.Unlock()
return uint64(seededIDGen.Int63())
}
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment