work on backpressure

Author: nathanverse

.gitignore

@@ -1 +1,2 @@
-.idea
+.idea
+.vendor

queue/Makefile

@@ -0,0 +1,19 @@
+.PHONY: server client
+
+PORT ?= 8080
+CAPACITY ?= 100
+WORKERS ?= 4
+TOTAL ?= 100
+CONCURRENCY ?= 4
+ITERATIONS ?= 1
+PPROF_PORT ?= 8081
+PPROF_FILE ?= mem.pprof
+
+server:
+	go run main.go -mode=server -addr=:$(PORT) -capacity=$(CAPACITY) -workers=$(WORKERS)
+
+client:
+	go run main.go -mode=client -addr=:$(PORT) -total=$(TOTAL) -concurrency=$(CONCURRENCY) -iterations=$(ITERATIONS)
+
+profmem:
+	go tool pprof -http=:$(PPROF_PORT) $(PPROF_FILE)

queue/main.go

@@ -8,6 +8,8 @@ import (
 	"github.com/nathanverse/go-concurrency-exercises/queue/runner"
 )
 
+//go run main.go -mode=server -workers=4 -capacity=10 -addr=:8082
+
 func main() {
 	mode := flag.String("mode", "server", "choose server or client mode")
 	addr := flag.String("addr", ":8080", "tcp listen address")

queue/mem.pprof

@@ -39,7 +39,8 @@ func RunClient(cfg ClientConfig) error {
 
 	for i := 0; i < cfg.Concurrency; i++ {
 		wg.Add(1)
-		go func() {
+		go func(index int) {
+			fmt.Printf("Starting goroutine %d\n", index+1)
 			defer wg.Done()
 
 			conn, err := net.Dial("tcp", cfg.Addr)
@@ -54,6 +55,7 @@ func RunClient(cfg ClientConfig) error {
 
 			for {
 				idx := atomic.AddInt64(&sent, 1)
+				fmt.Printf("Goroutine %d runs task %d\n", index+1, int(idx))
 				if idx > int64(cfg.Total) {
 					return
 				}
@@ -75,13 +77,14 @@ func RunClient(cfg ClientConfig) error {
 					return
 				}
 				if resp.Error != "" {
+					fmt.Printf("Error received: %s", resp.Error)
 					recordError(errCh, &once, errors.New(resp.Error))
 					return
 				}
 
 				atomic.AddInt64(&completed, 1)
 			}
-		}()
+		}(i)
 	}
 
 	wg.Wait()

queue/runner/client.go

@@ -4,13 +4,14 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"github.com/nathanverse/go-concurrency-exercises/queue/internal"
+	"github.com/nathanverse/go-concurrency-exercises/queue/tasks"
 	"github.com/pkg/profile"
 	"io"
 	"net"
 	"sync"
-
-	"github.com/nathanverse/go-concurrency-exercises/queue/internal"
-	"github.com/nathanverse/go-concurrency-exercises/queue/tasks"
+	"sync/atomic"
+	"time"
 )
 
 type response struct {
@@ -19,9 +20,11 @@ type response struct {
 	Error  string `json:"error,omitempty"`
 }
 
+var waitingGoroutines int64
+
 // Serve listens for TCP connections and forwards incoming tasks to the queue.
 func Serve(addr string, queue internal.IQueue, done <-chan struct{}) error {
-	defer profile.Start(profile.TraceProfile, profile.ProfilePath(".")).Stop()
+	defer profile.Start(profile.MemProfile, profile.ProfilePath(".")).Stop()
 
 	listener, err := net.Listen("tcp", addr)
 	if err != nil {
@@ -37,6 +40,20 @@ func Serve(addr string, queue internal.IQueue, done <-chan struct{}) error {
 		listener.Close()
 	}()
 
+	go func() {
+		for {
+			select {
+			case <-done:
+				{
+					break
+				}
+			default:
+				time.Sleep(5 * time.Second)
+				fmt.Println("Goroutines count: ", atomic.LoadInt64(&waitingGoroutines))
+			}
+		}
+	}()
+
 	for {
 		conn, err := listener.Accept()
 		if err != nil {
@@ -58,18 +75,28 @@ func Serve(addr string, queue internal.IQueue, done <-chan struct{}) error {
 
 		wg.Add(1)
 		go func(c net.Conn) {
-			defer wg.Done()
+			idx := atomic.AddInt64(&waitingGoroutines, 1)
+			defer func() {
+				wg.Done()
+				fmt.Printf("Goroutine %d exits\n", idx+1)
+				atomic.AddInt64(&waitingGoroutines, -1)
+			}()
+			fmt.Printf("Goroutine %d accpet connection\n", idx+1)
 			handleConnection(c, queue, done)
 		}(conn)
 	}
 }
 
 func handleConnection(conn net.Conn, queue internal.IQueue, done <-chan struct{}) {
-	defer conn.Close()
+	isFull := false
+
+	defer func() {
+		conn.Close()
+		fmt.Println("Connection closed, isFull=", isFull)
+	}()
 
 	decoder := json.NewDecoder(conn)
 	encoder := json.NewEncoder(conn)
-
 	for {
 		var task tasks.Task
 		if err := decoder.Decode(&task); err != nil {
@@ -84,6 +111,8 @@ func handleConnection(conn net.Conn, queue internal.IQueue, done <-chan struct{}
 		ch, err := queue.Put(&task)
 		if err != nil {
 			_ = encoder.Encode(response{ID: task.Id, Error: err.Error()})
+			fmt.Printf("Error putting task to queue: %s, retry \n", err.Error())
+			isFull = true
 			continue
 		}
 

queue/server/server.go

@@ -1,9 +1,9 @@
 package tasks
 
 import (
-	"crypto/sha256"
 	"encoding/json"
 	"fmt"
+	"time"
 )
 
 const (
@@ -48,10 +48,13 @@ type HashTaskOutput struct {
 
 func HashTask(iterations int) []byte {
 	data := []byte("benchmark")
-	var sum []byte
-	for i := 0; i < iterations; i++ {
-		h := sha256.Sum256(data)
-		sum = h[:]
-	}
-	return sum
+	//var sum []byte
+	//for i := 0; i < iterations; i++ {
+	//	h := sha256.Sum256(data)
+	//	sum = h[:]
+	//}
+	//return sum
+
+	time.Sleep(time.Duration(iterations) * time.Second)
+	return data
 }