From 68be5a1ffdf58b1350fec2b29e4809186592deaf Mon Sep 17 00:00:00 2001 From: Andrew M Date: Thu, 30 Jul 2020 15:29:53 -0400 Subject: [PATCH] fixes openziti/fabric#109, adds pools of workers for select handlers - adds woker pool support in routers for dial and route handers - adds configuration for pool/queue under the forwarder section --- router/forwarder/forwarder.go | 4 +- router/forwarder/options.go | 63 ++++++++++++++++- router/handler_ctrl/bind.go | 2 +- router/handler_ctrl/dial.go | 19 +++-- router/handler_ctrl/pool.go | 55 +++++++++++++++ router/handler_ctrl/route.go | 128 ++++++++++++++++++---------------- router/router.go | 2 +- 7 files changed, 205 insertions(+), 68 deletions(-) create mode 100644 router/handler_ctrl/pool.go diff --git a/router/forwarder/forwarder.go b/router/forwarder/forwarder.go index eda7f2bc8..e2ba32e2e 100644 --- a/router/forwarder/forwarder.go +++ b/router/forwarder/forwarder.go @@ -35,6 +35,7 @@ type Forwarder struct { payloadBufferController *xgress.PayloadBufferController metricsRegistry metrics.Registry traceController trace.Controller + Options *Options } type Destination interface { @@ -50,12 +51,13 @@ type XgressDestination interface { Label() string } -func NewForwarder(metricsRegistry metrics.Registry) *Forwarder { +func NewForwarder(metricsRegistry metrics.Registry, options *Options) *Forwarder { forwarder := &Forwarder{ sessions: newSessionTable(), destinations: newDestinationTable(), metricsRegistry: metricsRegistry, traceController: trace.NewController(), + Options: options, } forwarder.payloadBufferController = xgress.NewPayloadBufferController(forwarder) diff --git a/router/forwarder/options.go b/router/forwarder/options.go index fd2656dc6..094a97f53 100644 --- a/router/forwarder/options.go +++ b/router/forwarder/options.go @@ -23,10 +23,27 @@ import ( type Options struct { LatencyProbeInterval time.Duration + XgressDial WorkerPoolOptions + LinkDial WorkerPoolOptions +} + +type WorkerPoolOptions struct { + QueueLength uint16 + WorkerCount uint16 } func DefaultOptions() *Options { - return &Options{LatencyProbeInterval: time.Duration(10) * time.Second} + return &Options{ + LatencyProbeInterval: time.Duration(10) * time.Second, + XgressDial: WorkerPoolOptions{ + QueueLength: 1000, + WorkerCount: 10, + }, + LinkDial: WorkerPoolOptions{ + QueueLength: 1000, + WorkerCount: 10, + }, + } } func LoadOptions(src map[interface{}]interface{}) (*Options, error) { @@ -40,5 +57,49 @@ func LoadOptions(src map[interface{}]interface{}) (*Options, error) { } } + if value, found := src["xgressDialQueueLength "]; found { + if length, ok := value.(int); ok { + if length <= 0 || length > 10000 { + return nil, errors.New("invalid value for 'xgressDialQueueLength ', expected integer between 1 and 1000") + } + options.XgressDial.QueueLength = uint16(length) + } else { + return nil, errors.New("invalid value for 'xgressDialQueueLength ', expected integer between 1 and 1000") + } + } + + if value, found := src["xgressDialWorkerCount"]; found { + if workers, ok := value.(int); ok { + if workers <= 0 || workers > 10000 { + return nil, errors.New("invalid value for 'xgressDialWorkerCount', expected integer between 1 and 1000") + } + options.XgressDial.WorkerCount = uint16(workers) + } else { + return nil, errors.New("invalid value for 'xgressDialWorkerCount', expected integer between 1 and 1000") + } + } + + if value, found := src["linkDialQueueLength"]; found { + if length, ok := value.(int); ok { + if length <= 0 || length > 10000 { + return nil, errors.New("invalid value for 'linkDialQueueLength', expected integer between 1 and 1000") + } + options.LinkDial.QueueLength = uint16(length) + } else { + return nil, errors.New("invalid value for 'linkDialQueueLength', expected integer between 1 and 1000") + } + } + + if value, found := src["linkDialWorkerCount"]; found { + if workers, ok := value.(int); ok { + if workers <= 0 || workers > 10000 { + return nil, errors.New("invalid value for 'linkDialWorkerCount', expected integer between 10 and 1000") + } + options.LinkDial.WorkerCount = uint16(workers) + } else { + return nil, errors.New("invalid value for 'linkDialWorkerCount', expected integer between 10 and 1000") + } + } + return options, nil } diff --git a/router/handler_ctrl/bind.go b/router/handler_ctrl/bind.go index 2420604a3..a93fc09c6 100644 --- a/router/handler_ctrl/bind.go +++ b/router/handler_ctrl/bind.go @@ -52,7 +52,7 @@ func NewBindHandler(id *identity.TokenId, } func (self *bindHandler) BindChannel(ch channel2.Channel) error { - ch.AddReceiveHandler(newDialHandler(self.id, self.ctrl, self.xlinkDialers)) + ch.AddReceiveHandler(newDialHandler(self.id, self.ctrl, self.xlinkDialers, self.forwarder)) ch.AddReceiveHandler(newRouteHandler(self.id, self.ctrl, self.dialerCfg, self.forwarder)) ch.AddReceiveHandler(newValidateTerminatorsHandler(self.ctrl, self.dialerCfg)) ch.AddReceiveHandler(newUnrouteHandler(self.forwarder)) diff --git a/router/handler_ctrl/dial.go b/router/handler_ctrl/dial.go index 9323894a6..89f45272b 100644 --- a/router/handler_ctrl/dial.go +++ b/router/handler_ctrl/dial.go @@ -20,6 +20,7 @@ import ( "fmt" "github.com/golang/protobuf/proto" "github.com/openziti/fabric/pb/ctrl_pb" + "github.com/openziti/fabric/router/forwarder" "github.com/openziti/fabric/router/xgress" "github.com/openziti/fabric/router/xlink" "github.com/openziti/foundation/channel2" @@ -31,23 +32,33 @@ type dialHandler struct { id *identity.TokenId ctrl xgress.CtrlChannel dialers []xlink.Dialer + pool handlerPool } -func newDialHandler(id *identity.TokenId, ctrl xgress.CtrlChannel, dialers []xlink.Dialer) *dialHandler { - return &dialHandler{ +func newDialHandler(id *identity.TokenId, ctrl xgress.CtrlChannel, dialers []xlink.Dialer, forwarder *forwarder.Forwarder) *dialHandler { + handler := &dialHandler{ id: id, ctrl: ctrl, dialers: dialers, + pool: handlerPool{options: forwarder.Options.LinkDial}, } + handler.pool.Start() + + return handler } func (self *dialHandler) ContentType() int32 { return int32(ctrl_pb.ContentType_DialType) } -func (self *dialHandler) HandleReceive(msg *channel2.Message, _ channel2.Channel) { - logrus.Info("received link connect request") +func (self *dialHandler) HandleReceive(msg *channel2.Message, ch channel2.Channel) { + self.pool.Queue(func(){ + self.handle(msg, ch) + }) +} +func (self *dialHandler) handle(msg *channel2.Message, _ channel2.Channel) { + logrus.Info("received link connect request") dial := &ctrl_pb.Dial{} if err := proto.Unmarshal(msg.Body, dial); err == nil { linkId := self.id.ShallowCloneWithNewToken(dial.Id) diff --git a/router/handler_ctrl/pool.go b/router/handler_ctrl/pool.go new file mode 100644 index 000000000..f9f05206b --- /dev/null +++ b/router/handler_ctrl/pool.go @@ -0,0 +1,55 @@ +/* + Copyright NetFoundry, Inc. + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + https://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +*/ + +package handler_ctrl + +import ( + "github.com/michaelquigley/pfxlog" + "github.com/openziti/fabric/router/forwarder" + "sync" +) + +type handlerPool struct { + options forwarder.WorkerPoolOptions + startOnce sync.Once + queue chan func() +} + +func (pool *handlerPool) Start() { + pool.startOnce.Do(func() { + pool.queue = make(chan func(), pool.options.QueueLength) + for i := uint16(0); i < pool.options.WorkerCount; i++ { + go func() { + for handler := range pool.queue { + if handler != nil { + func() { + defer func() { + if err := recover(); err != nil { + pfxlog.Logger().Errorf("worker error: %v", err) + } + }() + handler() + }() + } + } + }() + } + }) +} + +func (pool *handlerPool) Queue(handler func()) { + pool.queue <- handler +} diff --git a/router/handler_ctrl/route.go b/router/handler_ctrl/route.go index 38261b2aa..158879e76 100644 --- a/router/handler_ctrl/route.go +++ b/router/handler_ctrl/route.go @@ -17,7 +17,6 @@ package handler_ctrl import ( - "fmt" "github.com/golang/protobuf/proto" "github.com/michaelquigley/pfxlog" "github.com/openziti/fabric/controller/xt" @@ -34,15 +33,21 @@ type routeHandler struct { ctrl xgress.CtrlChannel dialerCfg map[string]xgress.OptionsData forwarder *forwarder.Forwarder + pool handlerPool } func newRouteHandler(id *identity.TokenId, ctrl xgress.CtrlChannel, dialerCfg map[string]xgress.OptionsData, forwarder *forwarder.Forwarder) *routeHandler { - return &routeHandler{ + handler := &routeHandler{ id: id, ctrl: ctrl, dialerCfg: dialerCfg, forwarder: forwarder, + pool: handlerPool{options: forwarder.Options.XgressDial}, } + + handler.pool.Start() + + return handler } func (rh *routeHandler) ContentType() int32 { @@ -54,67 +59,70 @@ func (rh *routeHandler) HandleReceive(msg *channel2.Message, ch channel2.Channel route := &ctrl_pb.Route{} if err := proto.Unmarshal(msg.Body, route); err == nil { - if peerData, err := rh.createRoute(route); err == nil { - response := channel2.NewResult(true, "") - - for k, v := range peerData { - response.Headers[int32(k)] = v - } - - response.ReplyTo(msg) - if err := rh.ctrl.Channel().Send(response); err == nil { - log.Debugf("handled route for [s/%s]", route.SessionId) + if route.Egress != nil { + if rh.forwarder.HasDestination(xgress.Address(route.Egress.Address)) { + pfxlog.Logger().Warnf("destination exists for [%s]", route.Egress.Address) + rh.success(msg, ch, route, nil) + return } else { - log.Errorf("send response failed for [s/%s] (%s)", route.SessionId, err) + rh.connectEgress(msg, ch, route) + return } + } else { + rh.success(msg, ch, route, nil) + } + } else { + log.Errorf("error unmarshaling (%s)", err) + } +} + +func (rh *routeHandler) success(msg *channel2.Message, ch channel2.Channel, route *ctrl_pb.Route, peerData xt.PeerData) { + rh.forwarder.Route(route) + + log := pfxlog.ContextLogger(ch.Label()) + response := channel2.NewResult(true, "") + + for k, v := range peerData { + response.Headers[int32(k)] = v + } + + response.ReplyTo(msg) + + if err := rh.ctrl.Channel().Send(response); err == nil { + log.Debugf("handled route for [s/%s]", route.SessionId) + } else { + log.Errorf("send response failed for [s/%s] (%s)", route.SessionId, err) + } +} + +func (rh *routeHandler) connectEgress(msg *channel2.Message, ch channel2.Channel, route *ctrl_pb.Route) { + rh.pool.Queue(func() { + log := pfxlog.Logger().WithField("sessionId", route.SessionId) + log.Debugf("route request received. binding: %v, destination: %v, address: %v", + route.Egress.Binding, route.Egress.Destination, route.Egress.Address) + if factory, err := xgress.GlobalRegistry().Factory(route.Egress.Binding); err == nil { + if dialer, err := factory.CreateDialer(rh.dialerCfg[route.Egress.Binding]); err == nil { + sessionId := &identity.TokenId{Token: route.SessionId, Data: route.Egress.PeerData} + + bindHandler := handler_xgress.NewBindHandler( + handler_xgress.NewReceiveHandler(rh.ctrl, rh.forwarder), + handler_xgress.NewCloseHandler(rh.ctrl, rh.forwarder), + rh.forwarder) + + if peerData, err := dialer.Dial(route.Egress.Destination, + sessionId, + xgress.Address(route.Egress.Address), + bindHandler); err == nil { + rh.success(msg, ch, route, peerData) + } else { + pfxlog.ContextLogger(ch.Label()).Errorf("error creating route for [s/%s] (%s)", route.SessionId, err) + } + } else { + log.Errorf("unable to create dialer (%s)", err) + } + } else { pfxlog.ContextLogger(ch.Label()).Errorf("error creating route for [s/%s] (%s)", route.SessionId, err) } - - } else { - pfxlog.ContextLogger(ch.Label()).Errorf("error unmarshaling (%s)", err) - } -} - -func (rh *routeHandler) createRoute(route *ctrl_pb.Route) (xt.PeerData, error) { - var peerData xt.PeerData - - if route.Egress != nil { - if rh.forwarder.HasDestination(xgress.Address(route.Egress.Address)) { - pfxlog.Logger().Warnf("destination exists for [%s]", route.Egress.Address) - } else { - var err error - if peerData, err = rh.connectEgress(route); err != nil { - return nil, err - } - } - } - rh.forwarder.Route(route) - return peerData, nil -} - -func (rh *routeHandler) connectEgress(route *ctrl_pb.Route) (xt.PeerData, error) { - log := pfxlog.Logger().WithField("sessionId", route.SessionId) - log.Debugf("route request received. binding: %v, destination: %v, address: %v", - route.Egress.Binding, route.Egress.Destination, route.Egress.Address) - if factory, err := xgress.GlobalRegistry().Factory(route.Egress.Binding); err == nil { - if dialer, err := factory.CreateDialer(rh.dialerCfg[route.Egress.Binding]); err == nil { - sessionId := &identity.TokenId{Token: route.SessionId, Data: route.Egress.PeerData} - - bindHandler := handler_xgress.NewBindHandler( - handler_xgress.NewReceiveHandler(rh.ctrl, rh.forwarder), - handler_xgress.NewCloseHandler(rh.ctrl, rh.forwarder), - rh.forwarder) - - return dialer.Dial(route.Egress.Destination, - sessionId, - xgress.Address(route.Egress.Address), - bindHandler) - } else { - return nil, fmt.Errorf("unable to create dialer (%s)", err) - } - - } else { - return nil, err - } + }) } diff --git a/router/router.go b/router/router.go index db65b5e88..67bf60eea 100644 --- a/router/router.go +++ b/router/router.go @@ -71,7 +71,7 @@ func Create(config *Config) *Router { return &Router{ config: config, - forwarder: forwarder.NewForwarder(metricsRegistry), + forwarder: forwarder.NewForwarder(metricsRegistry, config.Forwarder), metricsRegistry: metricsRegistry, shutdownC: make(chan struct{}), }