Add per-service mutex to prevent concurrent reconciliation

[mweibel]

Prevents duplicate LB creation when multiple goroutines process the same service concurrently (e.g. EnsureLoadBalancer + UpdateLoadBalancer triggered by node sync).

Uses sync.Map with service.UID keys.

Includes a failing unit test that reproduces the concurrent creation race.

[alakae]

I think there are two problems:

1. I fear there is never anything removed: keys are strings but here we remove UID

2. I think there is the possibility that two callers end up holding different locks for the same UID.

Example interleaving (please challenge my understanding):

                                               muMap[uid]  Mutex 1           Mutex 2
A  LoadOrStore finds no entry, stores Mutex 1  Mutex 1     A acquired           -
B  LoadOrStore finds Mutex 1, waits on it      Mutex 1     A acquired, B waits  -
A  unlock()                                    Mutex 1     free, B waits     -
A  Delete removes the entry                    -           free, B waits     -
B  wakes on the mutex it already had           -           B acquired           -
C  LoadOrStore finds no entry, stores Mutex 2  Mutex 2     B acquired           C acquired

in the final line: there are two mutexes for the same UID acquired by different callers. only one of them is in map.

not sure about the best fix - maybe revert to the more trivial variant, that leaks a few bytes? :/

[mweibel]

I fear there is never anything removed: keys are strings but here we remove UID

UID is a string type alias (type UID string) but fair, the original code wouldn't need the string or we should have them in both places.

I think there is the possibility that two callers end up holding different locks for the same UID.

You mean this case:

package main

import (
	"fmt"
	"sync"
	"time"
)

var muMap sync.Map

func acquire(prefix, s string) func() {
	rawMu, _ := muMap.LoadOrStore(s, new(sync.Mutex))
	mu := rawMu.(*sync.Mutex)
	start := time.Now()
	fmt.Println(prefix, "acquiring lock")
	mu.Lock()

	return func() {
		fmt.Println(prefix, "releasing service lock", "uid", s, "duration", time.Since(start))
		mu.Unlock()
	}
}

func main() {
	s := "test"

	var wg sync.WaitGroup

	wg.Go(func() {
		unlock := acquire("a", s)
		time.Sleep(1 * time.Second)
		unlock()
		fmt.Println("deleting lock")
		muMap.Delete(s)
	})
	wg.Go(func() {
		time.Sleep(10*time.Millisecond)
		unlock := acquire("b", s)
		time.Sleep(2 * time.Second)
		unlock()
	})
	wg.Go(func() {
		time.Sleep(1500 * time.Millisecond)
		unlock := acquire("c", s)
		defer unlock()
		// critical section
	})

	wg.Wait()
	fmt.Println("done")
}

Right?

This is a valid concern in general, but I'm not sure if it's valid in this case.

The goroutine C you mention is IMO only possible in theory.
Reason: when EnsureLoadbalancerDeleted is called, the service is either deleted or doesn't need a LoadBalancer anymore.
In both cases, nobody would call EnsureLoadbalancer or UpdateLoadbalancer or EnsureLoadbalancerDeleted anymore.
The case where a goroutine C would happen, might be in laggy scenarios. For example: when CCM's informer (the watcher on the service resource) hasn't seen the deletionTimestamp yet and would queue a reconciliation. But even in this case, the race window is small because the rest would need to match as well.

That doesn't mean however, that the issue doesn't exist. To be safe, we'd need to get rid of the cleanup again and live with the fact that we have a small leak there. If that proves to be an issue, we could still add a GC on schedule which cleans up not locks in the map which don't exist anymore.

[mweibel]

actually, you're right about the first issue - I haven't considered the comparison it does. Will fix.