Merge pull request #5 from wrongerror/main

adjust buffer/retry mechanism

VERSION

@@ -1 +1 @@
-latest
+latest

go.mod

@@ -4,6 +4,7 @@ go 1.18
 
 require (
  github.com/OpenFunction/functions-framework-go v0.5.0
+ github.com/cenkalti/backoff/v4 v4.1.2
  github.com/dapr/components-contrib v1.8.1-rc.1
  github.com/dapr/dapr v1.8.3
  github.com/dapr/go-sdk v1.5.0
@@ -25,7 +26,6 @@ require (
  github.com/andybalholm/brotli v1.0.4 // indirect
  github.com/antlr/antlr4/runtime/Go/antlr v0.0.0-20210826220005-b48c857c3a0e // indirect
  github.com/beorn7/perks v1.0.1 // indirect
- github.com/cenkalti/backoff/v4 v4.1.2 // indirect
  github.com/cespare/xxhash/v2 v2.1.2 // indirect
  github.com/cloudevents/sdk-go/v2 v2.4.1 // indirect
  github.com/dapr/kit v0.0.2-0.20210614175626-b9074b64d233 // indirect

main.go

@@ -8,11 +8,9 @@ import (
 
  ofctx "github.com/OpenFunction/functions-framework-go/context"
  "github.com/OpenFunction/functions-framework-go/framework"
- "github.com/cenkalti/backoff/v4"
  diag "github.com/dapr/dapr/pkg/diagnostics"
  "github.com/dapr/dapr/pkg/modes"
  "github.com/dapr/dapr/pkg/runtime"
- "github.com/pkg/errors"
  "k8s.io/klog/v2"
 
  proxyruntime "github.com/OpenFunction/dapr-proxy/pkg/runtime"
@@ -77,36 +75,32 @@ func EventHandler(ctx ofctx.Context, in []byte) (ofctx.Out, error) {
  start := time.Now()
  defer func() {
  elapsed := diag.ElapsedSince(start)
+ pendingEventsCount := FuncRuntime.GetPendingEventsCount()
+ klog.V(4).Infof("Input: %s - Pending Events Count: %v", ctx.GetInputName(), pendingEventsCount)
  klog.V(4).Infof("Input: %s - Event Forwarding Elapsed: %vms", ctx.GetInputName(), elapsed)
  }()
 
  c := ctx.GetNativeContext()
+ respCh := make(chan *proxyruntime.EventResponse, 1)
 
  // Handle BindingEvent
  bindingEvent := ctx.GetBindingEvent()
  if bindingEvent != nil {
- FuncRuntime.EnqueueBindingEvent(&c, bindingEvent)
+ event := proxyruntime.NewEvent(&c, bindingEvent, nil, respCh)
+ FuncRuntime.EnqueueEvent(&event)
  }
 
  // Handle TopicEvent
  topicEvent := ctx.GetTopicEvent()
  if topicEvent != nil {
- FuncRuntime.EnqueueTopicEvent(&c, topicEvent)
+ event := proxyruntime.NewEvent(&c, nil, topicEvent, respCh)
+ FuncRuntime.EnqueueEvent(&event)
  }
 
- var resp *proxyruntime.EventResponse
- err := backoff.Retry(func() error {
- resp = FuncRuntime.GetEventResponse(&c)
- if resp == nil {
- return errors.New("Failed to get event response")
- }
- return nil
- }, utils.NewExponentialBackOff())
-
- if err != nil {
- e := errors.New("Processing event timeout")
- klog.Error(e)
- return ctx.ReturnOnInternalError(), e
+ resp := <-respCh
+ if resp.Error != nil {
+ klog.Error(resp.Error)
+ return ctx.ReturnOnInternalError(), resp.Error
  } else {
  out := new(ofctx.FunctionOut)
  out.WithData(resp.Data)

pkg/grpc/grpc.go

@@ -100,14 +100,16 @@ func (g *Manager) StartEndpointsDetection() {
  }
  }
 
- for ep, _ := range endpoints {
+ for ep := range endpoints {
  conn, teardown, err := g.getGRPCConnection(context.TODO(), ep.String(), "", "", true, false, g.sslEnabled)
  teardown()
  state := conn.GetState()
  if err != nil {
  klog.Error(err)
  } else if state == connectivity.Ready || state == connectivity.Idle {
  g.balancer.Add(ep)
+ } else {
+ g.balancer.Remove(ep)
  }
  }
  time.Sleep(200 * time.Millisecond)
@@ -131,7 +133,9 @@ func (g *Manager) GetGRPCConnection() (*grpc.ClientConn, func(), error) {
  state := conn.GetState()
  if state != connectivity.Ready && state != connectivity.Idle {
  g.balancer.Remove(address)
- delete(g.connectionPool.pool, address.String())
+ g.lock.Lock()
+ defer g.lock.Unlock()
+ g.connectionPool.Remove(address.String())
  teardown()
  }
  }, nil
@@ -257,6 +261,13 @@ func (p *connectionPool) Register(address string, conn *grpc.ClientConn) {
  p.referenceCount[conn] = 2
 }
 
+func (p *connectionPool) Remove(address string) {
+ if conn, ok := p.pool[address]; ok {
+ delete(p.pool, address)
+ conn.Close()
+ }
+}
+
 func (p *connectionPool) Share(address string) (*grpc.ClientConn, bool) {
  conn, ok := p.pool[address]
  if !ok {

pkg/runtime/runtime.go

@@ -5,7 +5,6 @@ import (
  "context"
  "encoding/json"
  nethttp "net/http"
- "sync"
 
  ofctx "github.com/OpenFunction/functions-framework-go/context"
  "github.com/cenkalti/backoff/v4"
@@ -33,40 +32,43 @@ type Config struct {
  MaxBufferSize int
 }
 
-type EventRequest struct {
- ctx *context.Context
- ofctx.EventRequest
+type Event struct {
+ ctx *context.Context
+ bindingEvent *common.BindingEvent
+ topicEvent *common.TopicEvent
+ respCh chan *EventResponse
+}
+
+func NewEvent(ctx *context.Context,
+ bindingEvent *common.BindingEvent,
+ topicEvent *common.TopicEvent,
+ respCh chan *EventResponse) Event {
+ return Event{
+ ctx: ctx,
+ bindingEvent: bindingEvent,
+ topicEvent: topicEvent,
+ respCh: respCh,
+ }
 }
 
 type EventResponse struct {
  Data []byte
  Error error
 }
 
-type ResponseMap struct {
- l *sync.RWMutex
- m map[*context.Context]*EventResponse
-}
-
 type Runtime struct {
  config *Config
  ctx *ofctx.FunctionContext
  grpc *grpc.Manager
  funcChannel channel.AppChannel
- reqChan chan *EventRequest
- respMap *ResponseMap
+ events chan *Event
 }
 
 func NewFuncRuntime(config *Config, ctx *ofctx.FunctionContext) *Runtime {
- lock := new(sync.RWMutex)
  return &Runtime{
- config: config,
- ctx: ctx,
- reqChan: make(chan *EventRequest, config.MaxBufferSize),
- respMap: &ResponseMap{
- l: lock,
- m: make(map[*context.Context]*EventResponse),
- },
+ config: config,
+ ctx: ctx,
+ events: make(chan *Event, config.MaxBufferSize),
  }
 }
 
@@ -89,76 +91,59 @@ func (r *Runtime) CreateFuncChannel() error {
  return nil
 }
 
-func (r *Runtime) ProcessEvents() {
- for e := range r.reqChan {
- if e.BindingEvent != nil {
- var data []byte
- // Retry on connection error.
- err := backoff.Retry(func() error {
- var err error
- data, err = r.OnBindingEvent(e.ctx, e.BindingEvent)
- if err != nil {
- klog.V(4).Info(err)
- return err
- }
- return nil
- }, utils.NewExponentialBackOff())
+func (r *Runtime) GetPendingEventsCount() int {
+ return len(r.events)
+}
 
- resp := EventResponse{
- Data: data,
- Error: err,
- }
- r.respMap.l.Lock()
- r.respMap.m[e.ctx] = &resp
- r.respMap.l.Unlock()
- }
+func (r *Runtime) ProcessEvents() {
+ for e := range r.events {
+ if e.bindingEvent != nil {
+ go func() {
+ var data []byte
+ // Retry on connection error.
+ err := backoff.Retry(func() error {
+ var err error
+ data, err = r.OnBindingEvent(e.ctx, e.bindingEvent)
+ if err != nil {
+ klog.V(4).Info(err)
+ return err
+ }
+ return nil
+ }, utils.NewExponentialBackOff())
 
- if e.TopicEvent != nil {
- // Retry on connection error.
- err := backoff.Retry(func() error {
- var err error
- err = r.OnTopicEvent(e.ctx, e.TopicEvent)
- if err != nil {
- return err
+ resp := EventResponse{
+ Data: data,
+ Error: err,
  }
- return nil
- }, utils.NewExponentialBackOff())
-
- resp := EventResponse{
- Data: nil,
- Error: err,
- }
- r.respMap.l.Lock()
- r.respMap.m[e.ctx] = &resp
- r.respMap.l.Unlock()
+ e.respCh <- &resp
+ }()
  }
- }
-}
 
-func (r *Runtime) EnqueueBindingEvent(ctx *context.Context, event *common.BindingEvent) {
- req := EventRequest{
- ctx: ctx,
- EventRequest: ofctx.EventRequest{BindingEvent: event},
- }
- r.reqChan <- &req
-}
+ if e.topicEvent != nil {
+ go func() {
+ // Retry on connection error.
+ err := backoff.Retry(func() error {
+ var err error
+ err = r.OnTopicEvent(e.ctx, e.topicEvent)
+ if err != nil {
+ klog.V(4).Info(err)
+ return err
+ }
+ return nil
+ }, utils.NewExponentialBackOff())
 
-func (r *Runtime) EnqueueTopicEvent(ctx *context.Context, event *common.TopicEvent) {
- req := EventRequest{
- ctx: ctx,
- EventRequest: ofctx.EventRequest{TopicEvent: event},
+ resp := EventResponse{
+ Data: nil,
+ Error: err,
+ }
+ e.respCh <- &resp
+ }()
+ }
  }
- r.reqChan <- &req
 }
 
-func (r *Runtime) GetEventResponse(ctx *context.Context) *EventResponse {
- defer r.respMap.l.Unlock()
- r.respMap.l.Lock()
- if resp, ok := r.respMap.m[ctx]; ok {
- delete(r.respMap.m, ctx)
- return resp
- }
- return nil
+func (r *Runtime) EnqueueEvent(event *Event) {
+ r.events <- event
 }
 
 func (r *Runtime) OnBindingEvent(ctx *context.Context, event *common.BindingEvent) ([]byte, error) {

pkg/utils/utils.go

@@ -60,9 +60,9 @@ func GetEnvVar(key, fallbackValue string) string {
 func NewExponentialBackOff() *backoff.ExponentialBackOff {
  b := &backoff.ExponentialBackOff{
  InitialInterval: 5 * time.Millisecond,
- RandomizationFactor: 0.5,
- Multiplier: 1.5,
- MaxInterval: 100 * time.Millisecond,
+ RandomizationFactor: 0.2,
+ Multiplier: 1,
+ MaxInterval: 5 * time.Millisecond,
  MaxElapsedTime: 60 * time.Second,
  Stop: backoff.Stop,
  Clock: backoff.SystemClock,