Adds rideshare memory leak

examples/golang-push/rideshare/docker-compose.yml

@@ -6,8 +6,8 @@ services:
     environment:
       - REGION=us-east
       - PYROSCOPE_SERVER_ADDRESS=http://pyroscope:4040
-      - PARAMETERS_POOL_SIZE=1000
-      - PARAMETERS_POOL_BUFFER_SIZE_KB=1000
+      - PARAMETERS_POOL_SIZE=10000
+      - PARAMETERS_POOL_BUFFER_SIZE_KB=10000
     build:
       context: .
 

examples/golang-push/rideshare/utility/pool.go

@@ -3,15 +3,13 @@ package utility
 import (
 	"os"
 	"sync"
-
-	"rideshare/rideshare"
 )
 
 type workerPool struct {
-	poolLock   *sync.Mutex
-	pool       []chan struct{}
-	limit      int
-	bufferSize int
+	poolLock     *sync.Mutex
+	pool         []chan struct{}
+	retainedData [][]byte // Slice to retain data to simulate memory leak.
+	limit        int
 }
 
 // Run a function using a pool.
@@ -30,15 +28,18 @@ func (c *workerPool) Run(fn func()) {
 	c.poolLock.Lock()
 	size := len(c.pool)
 	if c.limit != 0 && size >= c.limit {
-		// We're at max pool limit, reset the pool.
-		c.resetWithoutLock()
+		// We're at max pool limit, release a resource.
+		last := c.pool[size-1]
+		last <- struct{}{}
+		close(last)
+		c.pool = c.pool[:size-1]
 	}
 	c.pool = append(c.pool, stop)
 	c.poolLock.Unlock()
 
 	// Create a goroutine to run the function. It will write to done when the
 	// work is over, but won't clean up until it receives a signal from stop.
-	go c.doWork(fn, stop, done)
+	go cacheVehicleLocations(fn, stop, done, c)
 
 	// Block until the worker signals it's done.
 	<-done
@@ -50,30 +51,31 @@ func (c *workerPool) Close() {
 	c.poolLock.Lock()
 	defer c.poolLock.Unlock()
 
-	c.resetWithoutLock()
-}
-
-func (c *workerPool) resetWithoutLock() {
 	for _, c := range c.pool {
 		c <- struct{}{}
 		close(c)
 	}
-	c.pool = c.pool[:0]
+	c.pool = nil         // Fix here to properly clean up the slice.
+	c.retainedData = nil // Ensure that we also clear the retained data.
 }
 
-func (c *workerPool) doWork(fn func(), stop <-chan struct{}, done chan<- struct{}) {
+func cacheVehicleLocations(fn func(), stop <-chan struct{}, done chan<- struct{}, pool *workerPool) {
 	buf := make([]byte, 0)
 
 	// Do work.
 	fn()
 
 	// Simulate the work in fn requiring some data to be added to a buffer.
-	for i := 0; i < c.bufferSize; i++ {
-		buf = append(buf, byte(i))
+	// Increase buffer size to leak more memory.
+	const mb = 1 << 20
+	for i := 0; i < 1*mb; i++ {
+		buf = append(buf, byte(i%256))
 	}
 
-	// Don't let the compiler optimize away the buf.
-	var _ = buf
+	// Retain the buffer in the pool to prevent it from being garbage collected.
+	pool.poolLock.Lock()
+	pool.retainedData = append(pool.retainedData, buf)
+	pool.poolLock.Unlock()
 
 	// Signal we're done working.
 	done <- struct{}{}
@@ -82,11 +84,11 @@ func (c *workerPool) doWork(fn func(), stop <-chan struct{}, done chan<- struct{
 	<-stop
 }
 
-func newPool(c rideshare.Config) *workerPool {
+func newPool(n int) *workerPool {
 	return &workerPool{
-		poolLock:   &sync.Mutex{},
-		pool:       make([]chan struct{}, 0, c.ParametersPoolSize),
-		limit:      c.ParametersPoolSize,
-		bufferSize: c.ParametersPoolBufferSize,
+		poolLock:     &sync.Mutex{},
+		pool:         make([]chan struct{}, 0, n),
+		retainedData: make([][]byte, 0), // Initialize the slice to retain data.
+		limit:        n,
 	}
 }

examples/golang-push/rideshare/utility/utility.go

@@ -16,9 +16,10 @@ const durationConstant = time.Duration(200 * time.Millisecond)
 
 var pool *workerPool
 
-// InitWorkPool initializes the worker pool and returns a clean up function.
+// InitWorkerPool initializes the worker pool and returns a clean up function.
 func InitWorkerPool(c rideshare.Config) func() {
-	pool = newPool(c)
+	// Use the ParametersPoolSize from the config to initialize the pool.
+	pool = newPool(c.ParametersPoolSize)
 	return pool.Close
 }