@@ -103,7 +103,7 @@ type ObjectStorageController struct {
103103
104104 lockerLock sync.Mutex
105105 locker map [types.UID ]* sync.Mutex
106- opMap map [types. UID ] interface {}
106+ opMap * sync. Map
107107}
108108
109109func NewDefaultObjectStorageController (identity string , leaderLockName string , threads int ) (* ObjectStorageController , error ) {
@@ -162,7 +162,7 @@ func NewObjectStorageControllerWithClientset(identity string, leaderLockName str
162162 RenewDeadline : 15 * time .Second ,
163163 RetryPeriod : 5 * time .Second ,
164164
165- opMap : map [types. UID ] interface {} {},
165+ opMap : & sync. Map {},
166166 }, nil
167167}
168168
@@ -250,21 +250,22 @@ func (c *ObjectStorageController) processNextItem(ctx context.Context) bool {
250250 if quit {
251251 return false
252252 }
253-
253+
254254 uuid := uuidInterface .(types.UID )
255255 var err error
256- // With the lock below in place, we can safely tell the queue that we are done
257- // processing this item. The lock will ensure that multiple items of the same
258- // name and kind do not get processed simultaneously
256+
259257 defer c .queue .Done (uuid )
260258
259+ op , ok := c .opMap .Load (uuid )
260+ if ! ok {
261+ panic ("unreachable code" )
262+ }
263+
261264 // Ensure that multiple operations on different versions of the same object
262265 // do not happen in parallel
263266 c .OpLock (uuid )
264267 defer c .OpUnlock (uuid )
265268
266- op := c .opMap [uuid ]
267-
268269 switch o := op .(type ) {
269270 case addOp :
270271 add := * o .AddFunc
@@ -332,12 +333,12 @@ func (c *ObjectStorageController) GetOpLock(op types.UID) *sync.Mutex {
332333}
333334
334335// handleErr checks if an error happened and makes sure we will retry later.
335- func (c * ObjectStorageController ) handleErr (err error , op interface {} ) {
336+ func (c * ObjectStorageController ) handleErr (err error , uuid types. UID ) {
336337 if err == nil {
337- c .queue . Forget ( op )
338+ c .opMap . Delete ( uuid )
338339 return
339340 }
340- c .queue .AddRateLimited (op )
341+ c .queue .AddRateLimited (uuid )
341342}
342343
343344func (c * ObjectStorageController ) runController (ctx context.Context ) {
@@ -348,8 +349,8 @@ func (c *ObjectStorageController) runController(ctx context.Context) {
348349 lw := cache .NewListWatchFromClient (c .bucketClient .ObjectstorageV1alpha1 ().RESTClient (), name , "" , fields .Everything ())
349350 cfg := & cache.Config {
350351 Queue : cache .NewDeltaFIFOWithOptions (cache.DeltaFIFOOptions {
351- KnownObjects : indexer ,
352- EmitDeltaTypeReplaced : true ,
352+ KnownObjects : indexer ,
353+ EmitDeltaTypeReplaced : false ,
353354 }),
354355 ListerWatcher : lw ,
355356 ObjectType : objType ,
@@ -370,15 +371,14 @@ func (c *ObjectStorageController) runController(ctx context.Context) {
370371 }
371372
372373 uuid := d .Object .(metav1.Object ).GetUID ()
373- c .OpLock (uuid )
374- defer c .OpUnlock (uuid )
375- c .opMap [uuid ] = updateOp {
374+
375+ c .opMap .Store (uuid , updateOp {
376376 OldObject : old ,
377377 NewObject : d .Object ,
378378 UpdateFunc : & update ,
379379 Key : key ,
380380 Indexer : indexer ,
381- }
381+ })
382382 c .queue .Add (uuid )
383383 } else {
384384 key , err := cache .MetaNamespaceKeyFunc (d .Object )
@@ -387,21 +387,20 @@ func (c *ObjectStorageController) runController(ctx context.Context) {
387387 }
388388
389389 uuid := d .Object .(metav1.Object ).GetUID ()
390- c .OpLock (uuid )
391- defer c .OpUnlock (uuid )
392-
390+
393391 // If an update to the k8s object happens before add has succeeded
394- if op , ok := c .opMap [ uuid ] ; ok {
392+ if op , ok := c .opMap . Load ( uuid ) ; ok {
395393 if _ , ok := op .(updateOp ); ok {
396- return fmt .Errorf ("cannot add already added object: %s" , key )
394+ err := fmt .Errorf ("cannot add already added object: %s" , key )
395+ return err
397396 }
398397 }
399- c .opMap [ uuid ] = addOp {
398+ c .opMap . Store ( uuid , addOp {
400399 Object : d .Object ,
401400 AddFunc : & add ,
402401 Key : key ,
403402 Indexer : indexer ,
404- }
403+ })
405404 c .queue .Add (uuid )
406405 }
407406 case cache .Deleted :
@@ -411,14 +410,12 @@ func (c *ObjectStorageController) runController(ctx context.Context) {
411410 }
412411
413412 uuid := d .Object .(metav1.Object ).GetUID ()
414- c .OpLock (uuid )
415- defer c .OpUnlock (uuid )
416- c .opMap [uuid ] = deleteOp {
413+ c .opMap .Store (uuid , deleteOp {
417414 Object : d .Object ,
418415 DeleteFunc : & delete ,
419416 Key : key ,
420417 Indexer : indexer ,
421- }
418+ })
422419 c .queue .Add (uuid )
423420 }
424421 }
0 commit comments