Merge #39

39: Replace json std package with github.com/json-iterator/go r=48d90782 a=48d90782

1. Refactor amqp/conn, move redial from config to conn.go
2. Remove dialing function, as not concurrent safe, refactor it
3. Simplify conn.notify function
4. Add a simple threshold limiter to BS/conn.go. Limit located in the constants.go file in beanstalk package

Co-authored-by: Valery Piashchynski <[email protected]>

Author: bors[bot]

.github/workflows/ci-build.yml

@@ -59,6 +59,8 @@ jobs:
         run: |
           docker-compose -f ./tests/docker-compose.yml up -d
           vendor/bin/spiral-cs check src tests
+          go mod vendor
+          composer update
           go test -v -race -cover                    -coverprofile=jobs.txt      -covermode=atomic
           go test -v -race -cover ./broker/amqp      -coverprofile=amqp.txt      -covermode=atomic
           go test -v -race -cover ./broker/ephemeral -coverprofile=ephemeral.txt -covermode=atomic

Makefile

@@ -5,6 +5,7 @@ install: all
 uninstall: 
 	rm -f /usr/local/bin/rr-jobs
 test:
+	go mod vendor
 	composer update
 	go test -v -race -cover
 	go test -v -race -cover ./broker/amqp

broker/amqp/broker.go

@@ -56,13 +56,13 @@ func (b *Broker) Register(pipe *jobs.Pipeline) error {
 func (b *Broker) Serve() (err error) {
 	b.mu.Lock()
 
-	if b.publish, err = newConn(b.cfg.dial, b.cfg.TimeoutDuration()); err != nil {
+	if b.publish, err = newConn(b.cfg.Addr, b.cfg.TimeoutDuration()); err != nil {
 		b.mu.Unlock()
 		return err
 	}
 	defer b.publish.Close()
 
-	if b.consume, err = newConn(b.cfg.dial, b.cfg.TimeoutDuration()); err != nil {
+	if b.consume, err = newConn(b.cfg.Addr, b.cfg.TimeoutDuration()); err != nil {
 		b.mu.Unlock()
 		return err
 	}

broker/amqp/config.go

@@ -3,7 +3,6 @@ package amqp
 import (
 	"fmt"
 	"github.com/spiral/roadrunner/service"
-	"github.com/streadway/amqp"
 	"time"
 )
 
@@ -29,7 +28,7 @@ func (c *Config) Hydrate(cfg service.Config) error {
 	return nil
 }
 
-// TimeoutDuration returns number of seconds allowed to allocate the publish.
+// TimeoutDuration returns number of seconds allowed to redial
 func (c *Config) TimeoutDuration() time.Duration {
 	timeout := c.Timeout
 	if timeout == 0 {
@@ -38,8 +37,3 @@ func (c *Config) TimeoutDuration() time.Duration {
 
 	return time.Duration(timeout) * time.Second
 }
-
-// dial dials to AMQP.
-func (c *Config) dial() (*amqp.Connection, error) {
-	return amqp.Dial(c.Addr)
-}

broker/amqp/config_test.go

@@ -1,7 +1,7 @@
 package amqp
 
 import (
-	"encoding/json"
+	json "github.com/json-iterator/go"
 	"github.com/spiral/roadrunner/service"
 	"github.com/stretchr/testify/assert"
 	"testing"

broker/amqp/conn.go

@@ -10,7 +10,12 @@ import (
 
 // manages set of AMQP channels
 type chanPool struct {
-	mu        sync.Mutex
+	// timeout to backoff redial
+	tout time.Duration
+	url  string
+
+	mu sync.Mutex
+
 	conn      *amqp.Connection
 	channels  map[string]*channel
 	wait      chan interface{}
@@ -19,35 +24,39 @@ type chanPool struct {
 
 // manages single channel
 type channel struct {
-	ch       *amqp.Channel
+	ch *amqp.Channel
 	// todo unused
 	//consumer string
-	confirm  chan amqp.Confirmation
-	signal   chan error
+	confirm chan amqp.Confirmation
+	signal  chan error
 }
 
-type dialer func() (*amqp.Connection, error)
-
 // newConn creates new watched AMQP connection
-func newConn(dial dialer, tout time.Duration) (*chanPool, error) {
-	conn, err := dial()
+func newConn(url string, tout time.Duration) (*chanPool, error) {
+	conn, err := dial(url)
 	if err != nil {
 		return nil, err
 	}
 
 	cp := &chanPool{
+		url:       url,
+		tout:      tout,
 		conn:      conn,
 		channels:  make(map[string]*channel),
 		wait:      make(chan interface{}),
 		connected: make(chan interface{}),
 	}
 
 	close(cp.connected)
-	go cp.watch(dial, conn.NotifyClose(make(chan *amqp.Error)))
-
+	go cp.watch()
 	return cp, nil
 }
 
+// dial dials to AMQP.
+func dial(url string) (*amqp.Connection, error) {
+	return amqp.Dial(url)
+}
+
 // Close gracefully closes all underlying channels and connection.
 func (cp *chanPool) Close() error {
 	cp.mu.Lock()
@@ -91,14 +100,16 @@ func (cp *chanPool) waitConnected() chan interface{} {
 }
 
 // watch manages connection state and reconnects if needed
-func (cp *chanPool) watch(dial dialer, errors chan *amqp.Error) {
+func (cp *chanPool) watch() {
 	for {
 		select {
 		case <-cp.wait:
 			// connection has been closed
 			return
-		case err := <-errors:
+			// here we are waiting for the errors from amqp connection
+		case err := <-cp.conn.NotifyClose(make(chan *amqp.Error)):
 			cp.mu.Lock()
+			// clear connected, since connections are dead
 			cp.connected = make(chan interface{})
 
 			// broadcast error to all consume to let them for the tryReconnect
@@ -109,20 +120,20 @@ func (cp *chanPool) watch(dial dialer, errors chan *amqp.Error) {
 			// disable channel allocation while server is dead
 			cp.conn = nil
 			cp.channels = nil
-			cp.mu.Unlock()
 
 			// initialize the backoff
 			expb := backoff.NewExponentialBackOff()
-			expb.MaxInterval = DefaultMaxInterval
+			expb.MaxInterval = cp.tout
+			cp.mu.Unlock()
 
-			//reconnect function
+			// reconnect function
 			reconnect := func() error {
 				cp.mu.Lock()
-				defer cp.mu.Unlock()
-				conn, err := dial()
+				conn, err := dial(cp.url)
 				if err != nil {
 					// still failing
 					fmt.Println(fmt.Sprintf("error during the amqp dialing, %s", err.Error()))
+					cp.mu.Unlock()
 					return err
 				}
 
@@ -134,18 +145,16 @@ func (cp *chanPool) watch(dial dialer, errors chan *amqp.Error) {
 				cp.conn = conn
 				// re-init the channels
 				cp.channels = make(map[string]*channel)
-				errors = cp.conn.NotifyClose(make(chan *amqp.Error))
+				cp.mu.Unlock()
 				return nil
 			}
 
-
+			// start backoff retry
 			errb := backoff.Retry(reconnect, expb)
 			if errb != nil {
 				fmt.Println(fmt.Sprintf("backoff Retry error, %s", errb.Error()))
 				// reconnection failed
-				cp.mu.Lock()
 				close(cp.connected)
-				cp.mu.Unlock()
 				return
 			}
 			close(cp.connected)
@@ -162,7 +171,7 @@ func (cp *chanPool) channel(name string) (*channel, error) {
 	if dead {
 		// wait for connection restoration (doubled the timeout duration)
 		select {
-		case <-time.NewTimer(DefaultMaxInterval * 2).C:
+		case <-time.NewTimer(cp.tout * 2).C:
 			return nil, fmt.Errorf("connection is dead")
 		case <-cp.connected:
 			// connected

broker/amqp/constants.go

@@ -1,7 +1,7 @@
 package beanstalk
 
 import (
-	"encoding/json"
+	json "github.com/json-iterator/go"
 	"github.com/spiral/roadrunner/service"
 	"github.com/stretchr/testify/assert"
 	"testing"

broker/beanstalk/config_test.go

@@ -111,13 +111,17 @@ func (cn *conn) release(err error) error {
 // watch and reconnect if dead
 func (cn *conn) watch(network, addr string) {
 	cn.free <- nil
+	t := time.NewTicker(WatchThrottleLimit)
+	defer t.Stop()
 	for {
 		select {
 		case <-cn.dead:
+			// simple throttle limiter
+			<-t.C
 			// try to reconnect
 			// TODO add logging here
 			expb := backoff.NewExponentialBackOff()
-			expb.MaxInterval = time.Second * 5
+			expb.MaxInterval = cn.tout
 
 			reconnect := func() error {
 				conn, err := beanstalk.Dial(network, addr)

broker/beanstalk/conn.go

@@ -2,5 +2,5 @@ package beanstalk
 
 import "time"
 
-// DefaultMaxInterval is the max reconnect time interval
-const DefaultMaxInterval = 30 * time.Second
+// WatchThrottleLimit is used to limit reconnection occurrence in watch function
+const WatchThrottleLimit = time.Second

broker/beanstalk/constants.go

@@ -1,7 +1,7 @@
 package sqs
 
 import (
-	"encoding/json"
+	json "github.com/json-iterator/go"
 	"github.com/spiral/roadrunner/service"
 	"github.com/stretchr/testify/assert"
 	"testing"

broker/sqs/config_test.go

@@ -1,7 +1,7 @@
 package jobs
 
 import (
-	"encoding/json"
+	json "github.com/json-iterator/go"
 	"github.com/spiral/roadrunner/service"
 	"github.com/stretchr/testify/assert"
 	"testing"

config_test.go

@@ -1,6 +1,6 @@
 module github.com/spiral/jobs/v2
 
-go 1.13
+go 1.14
 
 require (
 	github.com/aws/aws-sdk-go v1.16.14
@@ -12,6 +12,7 @@ require (
 	github.com/cpuguy83/go-md2man v1.0.10 // indirect
 	github.com/dustin/go-humanize v1.0.0
 	github.com/gofrs/uuid v3.1.0+incompatible
+	github.com/json-iterator/go v1.1.9
 	github.com/kr/beanstalk v0.0.0-20180818045031-cae1762e4858 // indirect
 	github.com/olekukonko/tablewriter v0.0.4
 	github.com/prometheus/client_golang v1.5.0

go.mod

@@ -1,6 +1,6 @@
 package jobs
 
-import "encoding/json"
+import json "github.com/json-iterator/go"
 
 // Handler handles job execution.
 type Handler func(id string, j *Job) error