< Summary

Class:Microsoft.Azure.Batch.ODATAMonitor
Assembly:Microsoft.Azure.Batch
File(s):C:\Git\azure-sdk-for-net\sdk\batch\Microsoft.Azure.Batch\src\ODATAMonitor.cs
Covered lines:74
Uncovered lines:14
Coverable lines:88
Total lines:387
Line coverage:84% (74 of 88)
Covered branches:32
Total branches:38
Branch coverage:84.2% (32 of 38)

Metrics

MethodCyclomatic complexity Line coverage Branch coverage
WhenAllAsync()-86.36%87.5%
.ctor()-75%100%
.ctor(...)-100%100%
get_CancellationToken()-100%100%
ProcessOneInstance()-80.77%78.57%
SwapQueues()-100%100%
CallListAndProcessResults()-93.75%87.5%

File(s)

C:\Git\azure-sdk-for-net\sdk\batch\Microsoft.Azure.Batch\src\ODATAMonitor.cs

#LineLine coverage
 1// Copyright (c) Microsoft Corporation. All rights reserved.
 2// Licensed under the MIT License. See License.txt in the project root for license information.
 3
 4using System;
 5using System.Collections.Generic;
 6using System.Linq;
 7using System.Text;
 8using System.Threading.Tasks;
 9
 10namespace Microsoft.Azure.Batch
 11{
 12    using System.Threading;
 13
 14    /// <summary>
 15    /// Contains control settings used for optimal retrieval of state data via OData predicates.
 16    /// </summary>
 17    public class ODATAMonitorControl
 18    {
 19        private TimeSpan _delayBetweenDataFetch = new TimeSpan(0, 0, seconds: 2);
 20        private static TimeSpan _lowerBoundDelayBetweenRefresh = new TimeSpan(0, 0, 0, 0, milliseconds : 500);
 21
 22        /// <summary>
 23        /// The minimum time between attempts to fetch data for a monitored instance.
 24        /// </summary>
 25        public TimeSpan DelayBetweenDataFetch
 26        {
 27            get
 28            {
 29                return _delayBetweenDataFetch;
 30            }
 31            set
 32            {
 33                // forbid values that are too small... avoid DOS of server
 34                if (value < _lowerBoundDelayBetweenRefresh)
 35                {
 36                    value = _lowerBoundDelayBetweenRefresh;
 37                }
 38
 39                _delayBetweenDataFetch = value;
 40            }
 41        }
 42
 43        /// <summary>
 44        /// The maximum number of characters allowed for a single OData predicate.
 45        /// </summary>
 46        public int ODATAPredicateLimit = 500;
 47    }
 48
 49    /// <summary>
 50    /// Use a lambda to construct the correct ListFoo calls with correct params.
 51    /// </summary>
 52    /// <typeparam name="T"></typeparam>
 53    /// <returns></returns>
 54    internal delegate IPagedEnumerable<T> ListDelegate<T>();
 55
 56    /// <summary>
 57    /// A class that leverages OData predicates to poll the states of instances.
 58    /// </summary>
 59    internal class ODATAMonitor
 60    {
 61        /// <summary>
 62        /// Polls the collection of instances until each has passed the condition at least once.
 63        /// </summary>
 64        /// <typeparam name="T"></typeparam>
 65        /// <param name="collectionToMonitor"></param>
 66        /// <param name="condition"></param>
 67        /// <param name="getName"></param>
 68        /// <param name="listObjects"></param>
 69        /// <param name="cancellationToken"></param>
 70        /// <param name="detailLevel">Controls the detail level of the data returned by a call to the Azure Batch Servic
 71        /// <param name="control"></param>
 72        /// <returns></returns>
 73        public static async Task WhenAllAsync<T>(
 74            IEnumerable<T> collectionToMonitor,
 75            Func<T, bool> condition,
 76            Func<T, string> getName,
 77            ListDelegate<T> listObjects,
 78            CancellationToken cancellationToken,
 79            ODATADetailLevel detailLevel,
 80            ODATAMonitorControl control) where T : IRefreshable
 81        {
 382            if (null == collectionToMonitor)
 83            {
 084                throw new ArgumentNullException("collectionToMonitor");
 85            }
 86
 387            if (null == condition)
 88            {
 089                throw new ArgumentNullException("condition");
 90            }
 91
 392            RefreshViaODATAFilterList<T> odataRefresher = new RefreshViaODATAFilterList<T>(cancellationToken, detailLeve
 93
 94            // populate for first pass
 1895            foreach (T curT in collectionToMonitor)
 96            {
 97                // filter out the instances that already meet the condition
 698                if (!condition(curT))
 99                {
 6100                    MonitorLastCall<T> box = new MonitorLastCall<T>(curT, /* flags each instance as available to refresh
 101
 6102                    odataRefresher.CurrentPassQueue.Enqueue(box);
 103                }
 104            }
 105
 106            // process the instances in the current pass... swap queues to begin next pass
 12107            while (!odataRefresher.CancellationToken.IsCancellationRequested &&
 12108                odataRefresher.CurrentPassQueue.Count > 0)
 109            {
 110                // get next instance
 9111                MonitorLastCall<T> nextInstanceToProcess = odataRefresher.CurrentPassQueue.Dequeue();
 112
 113                // build an odata filter with as many names as the limit will allow and call refresh instances as needed
 9114                Task asyncProcessOneInstance = odataRefresher.ProcessOneInstance(nextInstanceToProcess, getName);
 115
 116                // a-wait for completion
 9117                await asyncProcessOneInstance.ConfigureAwait(continueOnCapturedContext: false);
 118
 119                // if the current pass queue is empty, swap the queues to begin next pass
 9120                if (0 == odataRefresher.CurrentPassQueue.Count)
 121                {
 3122                    odataRefresher.SwapQueues();
 123
 124                    // if we appear to be done, the stringbuilder might have the last predicate in it so flush that
 3125                    if (0 == odataRefresher.CurrentPassQueue.Count)
 126                    {
 127                        // start the call to process the last predicate
 3128                        Task asyncListTask = odataRefresher.CallListAndProcessResults();
 129
 130                        // a-wait for completion
 3131                        await asyncListTask.ConfigureAwait(continueOnCapturedContext: false);
 132
 133                        // if the last predicate created new work... the swap will bring it into a new pass
 3134                        odataRefresher.SwapQueues();
 135                    }
 136                }
 137            }
 138
 139            //Were we cancelled?
 3140            odataRefresher.CancellationToken.ThrowIfCancellationRequested();
 0141        }
 142
 143        /// <summary>
 144        /// Will accept instances of T and construct a filter predicate.
 145        /// The predicate will be used (and reset) when it is full or if
 146        /// the given instance was to recently refreshed (DOS avoidance).
 147        /// </summary>
 148        /// <typeparam name="T"></typeparam>
 149        internal class RefreshViaODATAFilterList<T>
 150        {
 151#region constructors
 152
 0153            private RefreshViaODATAFilterList()
 154            {
 0155            }
 156
 3157            internal RefreshViaODATAFilterList(
 3158                            CancellationToken cancellationToken,
 3159                            ODATADetailLevel odataDetail,
 3160                            Func<T, bool> condition,
 3161                            ListDelegate<T> listObjects,
 3162                            ODATAMonitorControl odataMonitorControl)
 163            {
 3164                this.CancellationToken = cancellationToken;
 3165                _odataDetailLevel = odataDetail;
 3166                _condition = condition;
 3167                _listObjects = listObjects;
 3168                _odataMonitorControl = odataMonitorControl;
 169
 3170                CurrentPassQueue = new Queue<MonitorLastCall<T>>();
 3171                NextPassQueue = new Queue<MonitorLastCall<T>>();
 3172            }
 173
 174#endregion constructors
 175
 176            // queue that holds the instances being refreshed on this pass
 0177            public Queue<MonitorLastCall<T>> CurrentPassQueue = new Queue<MonitorLastCall<T>>();
 178
 179            // queue that holds the instances that have been refreshed and failed the condition...
 180            // to be refreshed again on the next pass
 0181            public Queue<MonitorLastCall<T>> NextPassQueue = new Queue<MonitorLastCall<T>>();
 182
 27183            public CancellationToken CancellationToken { get; private set; }
 184
 185            /// <summary>
 186            /// Holds the delegate that determines of the given instance no longer needs to be polled.
 187            /// </summary>
 188            private readonly Func<T, bool> _condition;
 189
 190            /// <summary>
 191            /// Delegate to fetch new instances fresh state data.
 192            /// </summary>
 193            private readonly ListDelegate<T> _listObjects;
 194
 195            /// <summary>
 196            /// To be updated with filter predicate for refreshing instance state data.
 197            /// </summary>
 198            private readonly ODATADetailLevel _odataDetailLevel;
 199
 200            /// <summary>
 201            /// Holds control data for this odata monitor.
 202            /// </summary>
 203            private readonly ODATAMonitorControl _odataMonitorControl;
 204
 205            // ODATA predicated will be constructed in this
 0206            private readonly StringBuilder _odataFilterSB = new StringBuilder();
 207
 208            private const string IdPrefix = "id eq '";
 209            private const string IdPostfix = "'";
 210            private const string OrOperator = " or ";
 211
 212            internal async Task ProcessOneInstance(MonitorLastCall<T> nextInstance, Func<T, string> getName)
 213            {
 214                // we will loop until this is null
 9215                MonitorLastCall<T> processThisInstance = nextInstance;
 216
 21217                while (null != processThisInstance)
 218                {
 12219                    bool usePredicateToCallList = false;
 12220                    DateTime utcNow = DateTime.UtcNow;
 221
 222                    // if it is not too early, we can use this instance
 12223                    if (utcNow >= nextInstance.DoNotRefreshUntilUTC)
 224                    {
 225                        // now check to see if it will fit
 226                        // remember the limit on the name is 64 and the limit on the predicate is 500
 227                        // the assumption is that even if these #s evolve at least one max-sized-name will fit.
 228
 9229                        StringBuilder possibleAdditionalExpression = new StringBuilder();
 230
 231                        // if this is not the first name then we must "or" this name in
 9232                        if (0 != _odataFilterSB.Length)
 233                        {
 3234                            possibleAdditionalExpression.Append(OrOperator);
 235                        }
 236
 237                        // add in the name prefix
 9238                        possibleAdditionalExpression.Append(IdPrefix);
 239
 240                        // get the name of the object
 9241                        string name = getName(nextInstance.Instance);
 242
 243                        // add in the name
 9244                        possibleAdditionalExpression.Append(name);
 245
 246                        // add the name postfix
 9247                        possibleAdditionalExpression.Append(IdPostfix);
 248
 249                        // if it will fit then append the clause to the predicate
 9250                        if ((_odataFilterSB.Length + possibleAdditionalExpression.Length) < _odataMonitorControl.ODATAPr
 251                        {
 252                            // amend the predicate to refresh the object
 9253                            _odataFilterSB.Append(possibleAdditionalExpression.ToString());
 254
 9255                            processThisInstance = null;  // we are done
 256                        }
 257                        else
 258                        {
 259                            // it will not fit so we are done with this predicate
 0260                            usePredicateToCallList = true;
 261                        }
 262                    }
 263                    else // if the next instance cannot be refreshed yet we may need to wait a bit
 264                    {
 265                        // if we have work to do... return to process that work and use up time
 3266                        if (_odataFilterSB.Length > 0)
 267                        {
 0268                            usePredicateToCallList = true;
 269                        }
 270                        else  // if we have no work to do we should delay until the instance can be refreshed
 271                        {
 3272                            TimeSpan delayThisMuch = processThisInstance.DoNotRefreshUntilUTC - utcNow;
 273
 3274                            if (delayThisMuch.Ticks > 0)
 275                            {
 3276                                await System.Threading.Tasks.Task.Delay(delayThisMuch).ConfigureAwait(continueOnCaptured
 277                            }
 278                        }
 279                    }
 280
 281                    // should we call the server with the current predicate data?
 12282                    if (usePredicateToCallList)
 283                    {
 0284                        usePredicateToCallList = false;
 285
 286                        // start the new list operation
 0287                        Task asyncListTask = CallListAndProcessResults();
 288
 289                        // wait for completion
 0290                        await asyncListTask.ConfigureAwait(continueOnCapturedContext: false);
 291                    }
 292                }
 9293            }
 294
 295            /// <summary>
 296            /// Will swap the queues.  This is the transition from one pass to the next.
 297            /// </summary>
 298            internal void SwapQueues()
 299            {
 6300                Queue<MonitorLastCall<T>> tmp = this.CurrentPassQueue;
 301
 6302                this.CurrentPassQueue = this.NextPassQueue;
 6303                this.NextPassQueue = tmp;
 6304            }
 305
 306            /// <summary>
 307            /// Uses list func to fetch fresh data on the instances in the predicate.
 308            /// Instances that fail the condition are enqueued for the next pass.
 309            /// </summary>
 310            /// <returns></returns>
 311            internal async Task CallListAndProcessResults()
 312            {
 313                // extract the predicate that restricts the list call
 3314                string predicate = _odataFilterSB.ToString();
 315
 316                // clear the sb for the next batch
 3317                _odataFilterSB.Clear();
 318
 319                // early exit if there is no work to do
 3320                if (string.IsNullOrWhiteSpace(predicate))
 321                {
 0322                    return;
 323                }
 324
 325                // update the detail level
 3326                _odataDetailLevel.FilterClause = predicate;
 327
 328                // get the enumerable to refresh the instances
 3329                IPagedEnumerable<T> tEnumberable = _listObjects();
 330
 331                // get the enumerator for asycn walk of the collection
 3332                IPagedEnumerator<T> tEnumerator = tEnumberable.GetPagedEnumerator();
 333
 334                // used to enumerate until out of data
 335                bool thereIsMoreData;
 336
 337                do
 338                {
 339                    // move to next instance, possibley make call to server to get next batch
 9340                    Task<bool> asyncTask = tEnumerator.MoveNextAsync(this.CancellationToken);
 341
 9342                    thereIsMoreData = await asyncTask.ConfigureAwait(continueOnCapturedContext: false);
 343
 9344                    if (thereIsMoreData)
 345                    {
 346                        // get the current instance
 6347                        T refreshedInstance = tEnumerator.Current;
 348
 349                        // test it to see if it is done
 6350                        if (!_condition(refreshedInstance))
 351                        {
 352                            // we will have to refresh it again so put in queue for next pass
 353
 354                            // box it up
 6355                            MonitorLastCall<T> mlc = new MonitorLastCall<T>(refreshedInstance, DateTime.UtcNow + _odataM
 356
 357                            // enqueue it for next pass
 6358                            this.NextPassQueue.Enqueue(mlc);
 359                        }
 360                    }
 361                }
 9362                while (thereIsMoreData);
 3363            }
 364        }
 365    }
 366
 367    /// <summary>
 368    /// A class to track the last time an instance was refreshed/monitored.
 369    /// </summary>
 370    /// <typeparam name="T"></typeparam>
 371    internal class MonitorLastCall<T>
 372    {
 373        private MonitorLastCall()
 374        {
 375        }
 376
 377        internal MonitorLastCall(T instance, DateTime timestamp)
 378        {
 379            this.DoNotRefreshUntilUTC = timestamp;
 380            this.Instance = instance;
 381        }
 382
 383        public DateTime DoNotRefreshUntilUTC { get; internal set; }
 384
 385        public T Instance { get; internal set; }
 386    }
 387}
Generated by: ReportGenerator 4.6.4.0
8/12/2020 - 3:47:10 PM
GitHub | www.palmmedia.de

Methods/Properties

WhenAllAsync()
.ctor()
.ctor(...)
get_CancellationToken()
ProcessOneInstance()
SwapQueues()
CallListAndProcessResults()