Pool first attempt

This commit is contained in:
Asim 2016-05-13 15:58:53 +01:00
parent c0eac7dda8
commit 2e9f4271a8
3 changed files with 127 additions and 46 deletions

View File

@ -19,16 +19,16 @@ import (
type rpcClient struct { type rpcClient struct {
once sync.Once once sync.Once
opts Options opts Options
pool *pool
} }
func newRpcClient(opt ...Option) Client { func newRpcClient(opt ...Option) Client {
var once sync.Once
opts := newOptions(opt...) opts := newOptions(opt...)
rc := &rpcClient{ rc := &rpcClient{
once: once, once: sync.Once{},
opts: opts, opts: opts,
pool: newPool(),
} }
c := Client(rc) c := Client(rc)
@ -73,10 +73,15 @@ func (r *rpcClient) call(ctx context.Context, address string, req Request, resp
return errors.InternalServerError("go.micro.client", err.Error()) return errors.InternalServerError("go.micro.client", err.Error())
} }
c, err := r.opts.Transport.Dial(address, transport.WithTimeout(opts.DialTimeout)) var grr error
c, err := r.pool.getConn(address, r.opts.Transport, transport.WithTimeout(opts.DialTimeout))
if err != nil { if err != nil {
return errors.InternalServerError("go.micro.client", fmt.Sprintf("Error sending request: %v", err)) return errors.InternalServerError("go.micro.client", fmt.Sprintf("Error sending request: %v", err))
} }
defer func() {
// defer execution of release
r.pool.release(address, c, grr)
}()
stream := &rpcStream{ stream := &rpcStream{
context: ctx, context: ctx,
@ -107,8 +112,10 @@ func (r *rpcClient) call(ctx context.Context, address string, req Request, resp
select { select {
case err := <-ch: case err := <-ch:
grr = err
return err return err
case <-ctx.Done(): case <-ctx.Done():
grr = ctx.Err()
return errors.New("go.micro.client", fmt.Sprintf("%v", ctx.Err()), 408) return errors.New("go.micro.client", fmt.Sprintf("%v", ctx.Err()), 408)
} }
} }

71
client/rpc_pool.go Normal file
View File

@ -0,0 +1,71 @@
package client
import (
"sync"
"github.com/micro/go-micro/transport"
)
type pool struct {
tr transport.Transport
sync.Mutex
conns map[string][]*poolConn
}
type poolConn struct {
transport.Client
}
var (
maxIdleConn = 2
)
func newPool() *pool {
return &pool{
conns: make(map[string][]*poolConn),
}
}
// NoOp the Close since we manage it
func (p *poolConn) Close() error {
return nil
}
func (p *pool) getConn(addr string, tr transport.Transport, opts ...transport.DialOption) (*poolConn, error) {
p.Lock()
conns, ok := p.conns[addr]
// no free conn
if !ok || len(conns) == 0 {
p.Unlock()
// create new conn
c, err := tr.Dial(addr, opts...)
if err != nil {
return nil, err
}
return &poolConn{c}, nil
}
conn := conns[len(conns)-1]
p.conns[addr] = conns[:len(conns)-1]
p.Unlock()
return conn, nil
}
func (p *pool) release(addr string, conn *poolConn, err error) {
// don't store the conn
if err != nil {
conn.Client.Close()
return
}
// otherwise put it back
p.Lock()
conns := p.conns[addr]
if len(conns) >= maxIdleConn {
conn.Client.Close()
return
}
p.conns[addr] = append(conns, conn)
p.Unlock()
}

View File

@ -55,50 +55,53 @@ func (s *rpcServer) accept(sock transport.Socket) {
} }
}() }()
var msg transport.Message for {
if err := sock.Recv(&msg); err != nil { var msg transport.Message
return if err := sock.Recv(&msg); err != nil {
} return
// we use this Timeout header to set a server deadline
to := msg.Header["Timeout"]
// we use this Content-Type header to identify the codec needed
ct := msg.Header["Content-Type"]
cf, err := s.newCodec(ct)
// TODO: needs better error handling
if err != nil {
sock.Send(&transport.Message{
Header: map[string]string{
"Content-Type": "text/plain",
},
Body: []byte(err.Error()),
})
return
}
codec := newRpcPlusCodec(&msg, sock, cf)
// strip our headers
hdr := make(map[string]string)
for k, v := range msg.Header {
hdr[k] = v
}
delete(hdr, "Content-Type")
delete(hdr, "Timeout")
ctx := metadata.NewContext(context.Background(), hdr)
// set the timeout if we have it
if len(to) > 0 {
if n, err := strconv.ParseUint(to, 10, 64); err == nil {
ctx, _ = context.WithTimeout(ctx, time.Duration(n))
} }
}
// TODO: needs better error handling // we use this Timeout header to set a server deadline
if err := s.rpc.serveRequest(ctx, codec, ct); err != nil { to := msg.Header["Timeout"]
log.Printf("Unexpected error serving request, closing socket: %v", err) // we use this Content-Type header to identify the codec needed
ct := msg.Header["Content-Type"]
cf, err := s.newCodec(ct)
// TODO: needs better error handling
if err != nil {
sock.Send(&transport.Message{
Header: map[string]string{
"Content-Type": "text/plain",
},
Body: []byte(err.Error()),
})
return
}
codec := newRpcPlusCodec(&msg, sock, cf)
// strip our headers
hdr := make(map[string]string)
for k, v := range msg.Header {
hdr[k] = v
}
delete(hdr, "Content-Type")
delete(hdr, "Timeout")
ctx := metadata.NewContext(context.Background(), hdr)
// set the timeout if we have it
if len(to) > 0 {
if n, err := strconv.ParseUint(to, 10, 64); err == nil {
ctx, _ = context.WithTimeout(ctx, time.Duration(n))
}
}
// TODO: needs better error handling
if err := s.rpc.serveRequest(ctx, codec, ct); err != nil {
log.Printf("Unexpected error serving request, closing socket: %v", err)
return
}
} }
} }