Fix flaky race tests

client.go

@@ -3033,7 +3033,10 @@ func (c *pipelineConnClient) writer(conn net.Conn, stopCh <-chan struct{}, chs *
 			select {
 			case w = <-chW:
 			case <-stopTimer.C:
-				return nil
+				if c.canStopPipelineConn(chs) {
+					return nil
+				}
+				goto againChW
 			case <-stopCh:
 				return nil
 			case <-flushTimerCh:
@@ -3102,6 +3105,13 @@ func (c *pipelineConnClient) writer(conn net.Conn, stopCh <-chan struct{}, chs *
 	}
 }
 
+func (c *pipelineConnClient) canStopPipelineConn(chs *pipelineConnChannels) bool {
+	c.chLock.Lock()
+	canStop := c.chs == chs && chs.users == 0
+	c.chLock.Unlock()
+	return canStop && len(chs.chR) == 0 && len(chs.chW) == 0
+}
+
 func (c *pipelineConnClient) reader(conn net.Conn, stopCh <-chan struct{}, chs *pipelineConnChannels) error {
 	readBufferSize := c.ReadBufferSize
 	if readBufferSize <= 0 {

compress_test.go

@@ -216,6 +216,7 @@ func testConcurrent(concurrency int, f func() error) error {
 			err := f()
 			if err != nil {
 				ch <- fmt.Errorf("error in goroutine %d: %w", idx, err)
+				return
 			}
 			ch <- nil
 		}(i)
@@ -226,7 +227,7 @@ func testConcurrent(concurrency int, f func() error) error {
 			if err != nil {
 				return err
 			}
-		case <-time.After(time.Second):
+		case <-time.After(testTimeout(time.Second)):
 			return errors.New("timeout")
 		}
 	}

fs_fs_test.go

@@ -852,7 +852,7 @@ func TestDirFSFSCompressConcurrent(t *testing.T) {
 	for range concurrency {
 		select {
 		case <-ch:
-		case <-time.After(time.Second * 2):
+		case <-time.After(testTimeout(time.Second * 2)):
 			t.Fatalf("timeout")
 		}
 	}

race_disabled_test.go

@@ -0,0 +1,5 @@
+//go:build !race
+
+package fasthttp
+
+const raceEnabled = false

race_enabled_test.go

@@ -0,0 +1,5 @@
+//go:build race
+
+package fasthttp
+
+const raceEnabled = true

test_timeout_test.go

@@ -0,0 +1,16 @@
+package fasthttp
+
+import (
+	"runtime"
+	"time"
+)
+
+func testTimeout(timeout time.Duration) time.Duration {
+	if raceEnabled {
+		timeout *= 5
+	}
+	if runtime.GOOS == "windows" {
+		timeout *= 2
+	}
+	return timeout
+}