Merge pull request openziti/fabric#111 from openziti/fix.109.handler.pools

fixes openziti/fabric#109, adds pools of workers for select handlers

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)

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
 }

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))

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)

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
+}

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
-	}
+	})
 }

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{}),
 	}