@@ -333,11 +333,27 @@ private async Task updateParameters()
333333 try
334334 {
335335 await requestQueuedMessages ( ) ;
336+ var loopDetectionCache = new HashSet < ParameterUpdatedBag > ( ) ;
336337 while ( ParameterUpdatedBag . TryPeek ( out ParameterUpdatedBag bag ) )
337338 {
338- if ( DateTime . UtcNow - bag . Timestamp < TimeSpan . FromMilliseconds ( GlobalTimers . Instance . NonQueuedUpdateTime ) )
339+ TimeSpan updateTime = TimeSpan . FromMilliseconds ( GlobalTimers . Instance . NonQueuedUpdateTime ) ;
340+ if ( bag . Parameter . GetAttribute < ParameterUpdateTimeAttribute > ( ) is ParameterUpdateTimeAttribute attribute )
341+ updateTime = TimeSpan . FromMilliseconds ( attribute . Milliseconds ) ;
342+
343+ if ( loopDetectionCache . Any ( b => b . Parameter == bag . Parameter && b . Index == bag . Index ) )
339344 return ;
340345
346+ if ( DateTime . UtcNow - bag . Timestamp < updateTime )
347+ {
348+ ParameterUpdatedBag . TryDequeue ( out _ ) ;
349+ ParameterUpdatedBag . Enqueue ( bag ) ;
350+
351+ loopDetectionCache . Add ( bag ) ;
352+ continue ;
353+ }
354+
355+ loopDetectionCache . Add ( bag ) ;
356+
341357 var cts = new CancellationTokenSource ( ) ;
342358 cts . CancelAfter ( TimeSpan . FromMilliseconds ( GlobalTimers . Instance . ParameterUpdateTimerInterval ) ) ;
343359 var task = Task . Run ( async ( ) =>
@@ -393,12 +409,7 @@ private async Task requestQueuedMessages()
393409 mc = result . MessageCounter ;
394410 sw ? . Stop ( ) ;
395411 if ( result . State == PeerToPeerProcess . EPeerToPeerProcessState . Failed )
396- {
397- if ( sw . Elapsed . TotalSeconds > 3 )
398- QueuedSupported = false ;
399- Logger ? . LogWarning ( $ "Queued Parameter failed after { sw . ElapsedMilliseconds } ms, Queued seems not supported, and wil be disabled") ;
400412 return ;
401- }
402413 Logger ? . LogTrace ( $ "Queued Parameter update took { sw . ElapsedMilliseconds } ms for { mc } messages.") ;
403414 } , cts . Token ) ;
404415 await task ;
0 commit comments