Merge pull request #22 from smola/fix-scaling-policy

pool: use a new scaling policy instance per pool

pool.go

@@ -11,8 +11,6 @@ import (
 )
 
 var (
-	// DefaultScalingPolicy is the default ScalingPolicy.
-	DefaultScalingPolicy = MovingAverage(10, MinMax(1, 10, AIMD(1, 0.5)))
 	// DefaultPoolTimeout is the time a request to the DriverPool can wait
 	// before getting a driver assigned.
 	DefaultPoolTimeout = time.Second * 5
@@ -234,6 +232,12 @@ type ScalingPolicy interface {
 	Scale(total, load int) int
 }
 
+// DefaultScalingPolicy returns a new instance of the default scaling policy.
+// Instances returned by this function should not be reused.
+func DefaultScalingPolicy() ScalingPolicy {
+	return MovingAverage(10, MinMax(1, 10, AIMD(1, 0.5)))
+}
+
 type movingAverage struct {
 	ScalingPolicy
 	loads  []float64
@@ -242,7 +246,8 @@ type movingAverage struct {
 }
 
 // MovingAverage computes a moving average of the load and forwards it to the
-// underlying scaling policy.
+// underlying scaling policy. This policy is stateful and not thread-safe, do not
+// reuse its instances for multiple pools.
 func MovingAverage(window int, p ScalingPolicy) ScalingPolicy {
 	return &movingAverage{
 		ScalingPolicy: p,

pool_test.go

@@ -33,7 +33,7 @@ func TestDriverPoolStartNoopClose(t *testing.T) {
 		return &mockDriver{}, nil
 	}
 
-	dp, err := StartDriverPool(DefaultScalingPolicy, DefaultPoolTimeout, new)
+	dp, err := StartDriverPool(DefaultScalingPolicy(), DefaultPoolTimeout, new)
 	require.NoError(err)
 	require.NotNil(dp)
 
@@ -56,7 +56,7 @@ func TestDriverPoolStartFailingDriver(t *testing.T) {
 		return nil, fmt.Errorf("driver error")
 	}
 
-	dp, err := StartDriverPool(DefaultScalingPolicy, DefaultPoolTimeout, new)
+	dp, err := StartDriverPool(DefaultScalingPolicy(), DefaultPoolTimeout, new)
 	require.EqualError(err, "driver error")
 	require.Nil(dp)
 }
@@ -74,7 +74,7 @@ func TestDriverPoolSequential(t *testing.T) {
 		}, nil
 	}
 
-	dp, err := StartDriverPool(DefaultScalingPolicy, DefaultPoolTimeout, new)
+	dp, err := StartDriverPool(DefaultScalingPolicy(), DefaultPoolTimeout, new)
 	require.NoError(err)
 	require.NotNil(dp)
 
@@ -103,7 +103,7 @@ func TestDriverPoolParallel(t *testing.T) {
 		}, nil
 	}
 
-	dp, err := StartDriverPool(DefaultScalingPolicy, time.Second*10, new)
+	dp, err := StartDriverPool(DefaultScalingPolicy(), time.Second*10, new)
 	require.NoError(err)
 	require.NotNil(dp)
 

server.go

@@ -64,7 +64,7 @@ func (s *Server) AddDriver(lang string, img string) error {
 		return ErrRuntime.Wrap(err)
 	}
 
-	dp, err := StartDriverPool(DefaultScalingPolicy, DefaultPoolTimeout, func() (Driver, error) {
+	dp, err := StartDriverPool(DefaultScalingPolicy(), DefaultPoolTimeout, func() (Driver, error) {
 		return ExecDriver(s.rt, image)
 	})
 	if err != nil {

server_test.go

@@ -46,7 +46,7 @@ func TestNewServerMockedDriverParallelClients(t *testing.T) {
 	require.NoError(err)
 
 	s := NewServer(r)
-	dp, err := StartDriverPool(DefaultScalingPolicy, DefaultPoolTimeout, func() (Driver, error) {
+	dp, err := StartDriverPool(DefaultScalingPolicy(), DefaultPoolTimeout, func() (Driver, error) {
 		return &echoDriver{}, nil
 	})
 	require.NoError(err)