public void should_test_that_an_item_is_contained()
        {
            var setToRemoveFrom = new ConcurrentSet<int>(Enumerable.Range(1, 5));

            setToRemoveFrom.Contains(3).ShouldBeTrue();
            setToRemoveFrom.Contains(7).ShouldBeFalse();
        }

        private async Task<ProjectMap> CreateProjectMapAsync(ConcurrentSet<SymbolAndProjectId> symbols)
        {
            using (Logger.LogBlock(FunctionId.FindReference_CreateProjectMapAsync, _cancellationToken))
            {
                var projectMap = new ProjectMap();

                var scope = _documents?.Select(d => d.Project).ToImmutableHashSet();
                foreach (var symbolAndProjectId in symbols)
                {
                    foreach (var finder in _finders)
                    {
                        _cancellationToken.ThrowIfCancellationRequested();

                        var projects = await finder.DetermineProjectsToSearchAsync(symbolAndProjectId.Symbol, _solution, scope, _cancellationToken).ConfigureAwait(false);
                        foreach (var project in projects.Distinct().WhereNotNull())
                        {
                            if (scope == null || scope.Contains(project))
                            {
                                projectMap.Add(project, (symbolAndProjectId, finder));
                            }
                        }
                    }
                }

                Contract.ThrowIfTrue(projectMap.Any(kvp => kvp.Value.Count != kvp.Value.ToSet().Count));
                return projectMap;
            }
        }

		protected void HandleReduceForIndex(IndexToWorkOn indexToWorkOn)
		{
			var viewGenerator = context.IndexDefinitionStorage.GetViewGenerator(indexToWorkOn.IndexId);
			if (viewGenerator == null)
				return;

			bool operationCanceled = false;
			var itemsToDelete = new ConcurrentSet<object>();

			IList<ReduceTypePerKey> mappedResultsInfo = null;
			transactionalStorage.Batch(actions =>
			{
				mappedResultsInfo = actions.MapReduce.GetReduceTypesPerKeys(indexToWorkOn.IndexId,
					context.CurrentNumberOfItemsToReduceInSingleBatch,
					context.NumberOfItemsToExecuteReduceInSingleStep).ToList();
			});

			var singleStepReduceKeys = mappedResultsInfo.Where(x => x.OperationTypeToPerform == ReduceType.SingleStep).Select(x => x.ReduceKey).ToArray();
			var multiStepsReduceKeys = mappedResultsInfo.Where(x => x.OperationTypeToPerform == ReduceType.MultiStep).Select(x => x.ReduceKey).ToArray();

			currentlyProcessedIndexes.TryAdd(indexToWorkOn.IndexId, indexToWorkOn.Index);
			try
			{
				if (singleStepReduceKeys.Length > 0)
				{
					Log.Debug("SingleStep reduce for keys: {0}",singleStepReduceKeys.Select(x => x + ","));
					SingleStepReduce(indexToWorkOn, singleStepReduceKeys, viewGenerator, itemsToDelete);
				}

				if (multiStepsReduceKeys.Length > 0)
				{
					Log.Debug("MultiStep reduce for keys: {0}", singleStepReduceKeys.Select(x => x + ","));
					MultiStepReduce(indexToWorkOn, multiStepsReduceKeys, viewGenerator, itemsToDelete);
				}
			}
			catch (OperationCanceledException)
			{
				operationCanceled = true;
			}
			finally
			{
				if (operationCanceled == false)
				{
					// whatever we succeeded in indexing or not, we have to update this
					// because otherwise we keep trying to re-index failed mapped results
					transactionalStorage.Batch(actions =>
					{
						var latest = actions.MapReduce.DeleteScheduledReduction(itemsToDelete);

						if (latest == null)
							return;
						actions.Indexing.UpdateLastReduced(indexToWorkOn.Index.indexId, latest.Etag, latest.Timestamp);
					});
				}

				Index _;
				currentlyProcessedIndexes.TryRemove(indexToWorkOn.IndexId, out _);
			}
		}

		/// <summary>
		/// Create a work item queue that will try to pull items from a named RabbitMQ endpoint
		/// </summary>
		/// <param name="endpoint">Destination endpoint to pull messages from</param>
		/// <param name="messagingBase">RabbitMQ connection provider</param>
		/// <param name="sleeper">Sleeper to rate limit polling</param>
		public RabbitMqPollingNode(IRoutingEndpoint endpoint,
			IMessagingBase messagingBase, ISleepWrapper sleeper)
		{
			_endpoint = endpoint.ToString();
			_messagingBase = messagingBase;
			_sleeper = sleeper;
			_boundMessageTypes = new ConcurrentSet<Type>();
		}

        public void should_add_an_item()
        {
            var setToRemoveFrom = new ConcurrentSet<int>(Enumerable.Range(1, 5));

            setToRemoveFrom.Add(6);

            setToRemoveFrom.ShouldEqual(new[] { 1, 2, 3, 4, 5, 6 });
        }

        public void should_clear_itself()
        {
            var setToClear = new ConcurrentSet<int>(Enumerable.Range(1, 5));

            setToClear.Clear();

            setToClear.Count.ShouldEqual(0);
        }

        public void should_remove_an_item()
        {
            var setToRemoveFrom = new ConcurrentSet<int>(Enumerable.Range(1, 5));

            setToRemoveFrom.Remove(3);

            setToRemoveFrom.ShouldEqual(new[] { 1, 2, 4, 5 });
        }

        public WatcherInfo(FileSystemWatcher watcher, Boolean isDirectory)
            : this()
        {
            Watcher = watcher;

            if (isDirectory)
                WatchedFiles = new ConcurrentSet<String>();
        }

		/// <summary>
		/// Create a local polling node.
		/// <para>You should not use this yourself. Use:</para>
		/// <para>MessagingSystem.Configure.WithLocalQueue(...);</para>
		/// and receive messages as normal.
		/// </summary>
		public LocalQueuePollingNode(string dispatchPath, string incomingPath,
		                             IMessageSerialiser serialiser, ISleepWrapper sleeper)
		{
			_dispatchPath = dispatchPath;
			_incomingPath = incomingPath;
			_serialiser = serialiser;
			_sleeper = sleeper;
			_boundMessageTypes = new ConcurrentSet<Type>();
		}

        public void should_copy_itself_to_an_array()
        {
            var setToCopy = new ConcurrentSet<int>(Enumerable.Range(1, 5));
            var destinationArray = new int[5];

            setToCopy.CopyTo(destinationArray, 0);

            destinationArray.ShouldEqual(new[] { 1, 2, 3, 4, 5 });
        }

		public GetItemsToReduceParams(string index, IEnumerable<string> reduceKeys, int level, bool loadData, ConcurrentSet<object> itemsToDelete)
		{
			Index = index;
			Level = level;
			LoadData = loadData;
			ItemsToDelete = itemsToDelete;
			ItemsAlreadySeen = new HashSet<Tuple<string, int>>();
			ReduceKeys = new HashSet<string>(reduceKeys);
		}

      public async Task RunAsync() {
         Console.WriteLine(DateTime.Now);
         const int kMessagesPerWorker = 200000;
         var sink = courierFacades[0];
         var senders = courierFacades.Skip(1).ToArray();
         var counter = kMessagesPerWorker * senders.Length;
         var doneSignal = new AsyncLatch();
         int upCounter = 0;
         var set = new ConcurrentSet<int>();
         sink.InboundMessageRouter.RegisterHandler<string>(
            x => {
               set.AddOrThrow(int.Parse(x.Body));
               var newCounter = Interlocked.Decrement(ref counter);
               Interlocked.Increment(ref upCounter);
               if (upCounter % 500 == 0)
                  Console.WriteLine(newCounter + " " + upCounter);
               if (newCounter == 0) {
                  doneSignal.Set();
               }
               return Task.FromResult(false);
            });

         var sync = new AsyncCountdownLatch(senders.Length);
         var senderTasks = senders.Select((s, id) => Go(async () => {
            await s.PeerTable.GetOrAdd(sink.Identity.Id).WaitForDiscoveryAsync().ConfigureAwait(false);
            sync.Signal();
            await sync.WaitAsync().ConfigureAwait(false);
            Console.WriteLine("Sink discovered: " + DateTime.Now);

            const int kBatchFactor = 1;
            for (var batch = 0; batch < kBatchFactor; batch++) {
               var batchSize = kMessagesPerWorker / kBatchFactor;
               await Task.WhenAll(Util.Generate(
                  batchSize,
                  i => s.Messenger.SendReliableAsync(
                     "" + (batch * batchSize + i + id * kMessagesPerWorker),
                     sink.Identity.Id))
                  ).ConfigureAwait(false);
            }

            Console.WriteLine("Worker Done: " + DateTime.Now);
         }));
         await Task.WhenAll(senderTasks).ConfigureAwait(false);
         Console.WriteLine("Senders Done: " + DateTime.Now);

         await doneSignal.WaitAsync().ConfigureAwait(false);
         Console.WriteLine("Done Signalled: " + DateTime.Now);

         AssertCollectionDeepEquals(set, new ConcurrentSet<int>(Enumerable.Range(0, kMessagesPerWorker * senders.Length)));

         while (true) {
            GC.Collect();
         }
      }

        /// <summary>
        /// Creates a new compilation by attaching diagnostic analyzers to an existing compilation.
        /// </summary>
        /// <param name="compilation">The original compilation.</param>
        /// <param name="analyzers">The set of analyzers to include in future analyses.</param>
        /// <param name="options">Options that are passed to analyzers.</param>
        /// <param name="cancellationToken">A cancellation token that can be used to abort analysis.</param>
        public CompilationWithAnalyzers(Compilation compilation, ImmutableArray<DiagnosticAnalyzer> analyzers, AnalyzerOptions options, CancellationToken cancellationToken)
        {
            if (compilation == null)
            {
                throw new ArgumentNullException(nameof(compilation));
            }

            VerifyAnalyzersArgument(analyzers);

            _cancellationToken = cancellationToken;
            _exceptionDiagnostics = new ConcurrentSet<Diagnostic>();
            _driver = AnalyzerDriver.Create(compilation, analyzers, options, AnalyzerManager.Instance, AddExceptionDiagnostic, false, out _compilation, _cancellationToken);
        }

        public CodeLensFindReferencesProgress(
            ISymbol queriedDefinition,
            SyntaxNode queriedNode,
            int searchCap,
            CancellationToken cancellationToken)
        {
            _queriedSymbol = queriedDefinition;
            _queriedNode = queriedNode;
            _aggregateCancellationTokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
            _locations = new ConcurrentSet<Location>();

            SearchCap = searchCap;
        }

        private async Task<ConcurrentDictionary<Project, ConcurrentQueue<ValueTuple<SymbolAndProjectId, IReferenceFinder>>>> CreateProjectMapAsync(
            ConcurrentSet<SymbolAndProjectId> symbols)
        {
            using (Logger.LogBlock(FunctionId.FindReference_CreateProjectMapAsync, _cancellationToken))
            {
                Func<Project, ConcurrentQueue<ValueTuple<SymbolAndProjectId, IReferenceFinder>>> createQueue = 
                    p => new ConcurrentQueue<ValueTuple<SymbolAndProjectId, IReferenceFinder>>();

                var projectMap = new ConcurrentDictionary<Project, ConcurrentQueue<ValueTuple<SymbolAndProjectId, IReferenceFinder>>>();

#if PARALLEL
            Roslyn.Utilities.TaskExtensions.RethrowIncorrectAggregateExceptions(cancellationToken, () =>
                {
                    symbols.AsParallel().WithCancellation(cancellationToken).ForAll(s =>
                    {
                        finders.AsParallel().WithCancellation(cancellationToken).ForAll(f =>
                        {
                            var projects = f.DetermineProjectsToSearch(s, solution, cancellationToken) ?? SpecializedCollections.EmptyEnumerable<Project>();
                            foreach (var project in projects.Distinct())
                            {
                                projectMap.GetOrAdd(project, createQueue).Enqueue(ValueTuple.Create(s, f));
                            }
                        });
                    });
                });
#else

                var scope = _documents != null ? _documents.Select(d => d.Project).ToImmutableHashSet() : null;
                foreach (var s in symbols)
                {
                    foreach (var f in _finders)
                    {
                        _cancellationToken.ThrowIfCancellationRequested();

                        var projects = await f.DetermineProjectsToSearchAsync(s.Symbol, _solution, scope, _cancellationToken).ConfigureAwait(false);
                        foreach (var project in projects.Distinct().WhereNotNull())
                        {
                            if (scope == null || scope.Contains(project))
                            {
                                projectMap.GetOrAdd(project, createQueue).Enqueue(ValueTuple.Create(s, f));
                            }
                        }
                    }
                }
#endif

                Contract.ThrowIfTrue(projectMap.Any(kvp => kvp.Value.Count != kvp.Value.ToSet().Count));
                return projectMap;
            }
        }

        public void ConcurrentTest()
        {
            var cset = new ConcurrentSet<int>();
            Action a1 = () => { for (int i = 0; i < 1000000; i++) cset.Add(i); };
            Action a2 = () => { for (int i = 1000000; i < 2000000; i++) cset.Add(i); };
            Action a3 = () => { for (int i = 2000000; i < 3000000; i++) cset.Add(i); };
            Action a4 = () => { for (int i = 3000000; i < 4000000; i++) cset.Add(i); };
            bool b1 = false;
            bool b2 = false;
            bool b3 = false;
            bool b4 = false;
            a1.BeginInvoke(iar =>
            {
                a1.EndInvoke(iar);
                b1 = true;
            }, null);
            a2.BeginInvoke(iar =>
            {
                a2.EndInvoke(iar);
                b2 = true;
            }, null);
            a3.BeginInvoke(iar =>
            {
                a3.EndInvoke(iar);
                b3 = true;
            }, null);
            a4.BeginInvoke(iar =>
            {
                a4.EndInvoke(iar);
                b4 = true;
            }, null);

            while (!(b1 && b2 && b3 && b4))
            {
                Thread.Sleep(10);
            }

            Assert.AreEqual(4000000, cset.Count());
        }

        private Task CreateTask(int baseInterval, ConcurrentSet<ICounterMetric> concurrentSet)
        {
            return Task.Run(async () =>
            {
                long elapsed = 0;

                while (true)
                {
                    try
                    {
                        if (cts.IsCancellationRequested)
                            return;

                        var milliseconds = baseInterval - elapsed;
                        if (milliseconds > 0)
                        {
                            await Task.Delay(TimeSpan.FromMilliseconds(milliseconds), cts.Token).ConfigureAwait(false);
                        }

                        var sp = Stopwatch.StartNew();
                        foreach (var ticker in concurrentSet)
                        {
                            if (cts.IsCancellationRequested)
                                return;

                            ticker.Tick();
                        }
                        elapsed = sp.ElapsedMilliseconds;
                    }
                    catch (TaskCanceledException)
                    {
                        return;
                    }
                }
            }, cts.Token);
        }

        private int RunCore(TextWriter consoleOutput, ErrorLogger errorLogger, CancellationToken cancellationToken)
        {
            Debug.Assert(!Arguments.IsScriptRunner);

            cancellationToken.ThrowIfCancellationRequested();

            if (Arguments.DisplayLogo)
            {
                PrintLogo(consoleOutput);
            }

            if (Arguments.DisplayHelp)
            {
                PrintHelp(consoleOutput);
                return Succeeded;
            }

            if (ReportErrors(Arguments.Errors, consoleOutput, errorLogger))
            {
                return Failed;
            }

            var touchedFilesLogger = (Arguments.TouchedFilesPath != null) ? new TouchedFileLogger() : null;

            Compilation compilation = CreateCompilation(consoleOutput, touchedFilesLogger, errorLogger);
            if (compilation == null)
            {
                return Failed;
            }


            var diagnostics = new List<DiagnosticInfo>();
            var analyzers = ResolveAnalyzersFromArguments(diagnostics, MessageProvider, touchedFilesLogger);
            var additionalTextFiles = ResolveAdditionalFilesFromArguments(diagnostics, MessageProvider, touchedFilesLogger);
            if (ReportErrors(diagnostics, consoleOutput, errorLogger))
            {
                return Failed;
            }

            cancellationToken.ThrowIfCancellationRequested();

            CancellationTokenSource analyzerCts = null;
            AnalyzerManager analyzerManager = null;
            AnalyzerDriver analyzerDriver = null;
            try
            {
                Func<ImmutableArray<Diagnostic>> getAnalyzerDiagnostics = null;
                ConcurrentSet<Diagnostic> analyzerExceptionDiagnostics = null;
                if (!analyzers.IsDefaultOrEmpty)
                {
                    analyzerCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
                    analyzerManager = new AnalyzerManager();
                    analyzerExceptionDiagnostics = new ConcurrentSet<Diagnostic>();
                    Action<Diagnostic> addExceptionDiagnostic = diagnostic => analyzerExceptionDiagnostics.Add(diagnostic);
                    var analyzerOptions = new AnalyzerOptions(ImmutableArray<AdditionalText>.CastUp(additionalTextFiles));
                    
                    analyzerDriver = AnalyzerDriver.CreateAndAttachToCompilation(compilation, analyzers, analyzerOptions, analyzerManager, addExceptionDiagnostic, Arguments.ReportAnalyzer, out compilation, analyzerCts.Token);
                    getAnalyzerDiagnostics = () => analyzerDriver.GetDiagnosticsAsync(compilation).Result;
                }

                // Print the diagnostics produced during the parsing stage and exit if there were any errors.
                if (ReportErrors(compilation.GetParseDiagnostics(), consoleOutput, errorLogger))
                {
                    return Failed;
                }

                if (ReportErrors(compilation.GetDeclarationDiagnostics(), consoleOutput, errorLogger))
                {
                    return Failed;
                }

                EmitResult emitResult;

                // NOTE: as native compiler does, we generate the documentation file
                // NOTE: 'in place', replacing the contents of the file if it exists

                string finalPeFilePath;
                string finalPdbFilePath;
                string finalXmlFilePath;

                Stream xmlStreamOpt = null;

                cancellationToken.ThrowIfCancellationRequested();

                finalXmlFilePath = Arguments.DocumentationPath;
                if (finalXmlFilePath != null)
                {
                    xmlStreamOpt = OpenFile(finalXmlFilePath, consoleOutput, PortableShim.FileMode.OpenOrCreate, PortableShim.FileAccess.Write, PortableShim.FileShare.ReadWriteBitwiseOrDelete);
                    if (xmlStreamOpt == null)
                    {
                        return Failed;
                    }

                    xmlStreamOpt.SetLength(0);
                }

                cancellationToken.ThrowIfCancellationRequested();

                IEnumerable<DiagnosticInfo> errors;
                using (var win32ResourceStreamOpt = GetWin32Resources(Arguments, compilation, out errors))
//.........这里部分代码省略.........

		private ReducingPerformanceStats SingleStepReduce(IndexToWorkOn index, List<string> keysToReduce, AbstractViewGenerator viewGenerator,
												          ConcurrentSet<object> itemsToDelete, CancellationToken token)
		{
			var needToMoveToSingleStepQueue = new ConcurrentQueue<HashSet<string>>();

            if ( Log.IsDebugEnabled )
			    Log.Debug(() => string.Format("Executing single step reducing for {0} keys [{1}]", keysToReduce.Count, string.Join(", ", keysToReduce)));

			var batchTimeWatcher = Stopwatch.StartNew();

			var reducingBatchThrottlerId = Guid.NewGuid();
			var reducePerformanceStats = new ReducingPerformanceStats(ReduceType.SingleStep);
			var reduceLevelStats = new ReduceLevelPeformanceStats
			{
				Started = SystemTime.UtcNow,
				Level = 2
			};

			try
			{
				var parallelOperations = new ConcurrentQueue<ParallelBatchStats>();

				var parallelProcessingStart = SystemTime.UtcNow;

				BackgroundTaskExecuter.Instance.ExecuteAllBuffered(context, keysToReduce, enumerator =>
				{
					var parallelStats = new ParallelBatchStats
					{
						StartDelay = (long)(SystemTime.UtcNow - parallelProcessingStart).TotalMilliseconds
					};

					var localNeedToMoveToSingleStep = new HashSet<string>();
					needToMoveToSingleStepQueue.Enqueue(localNeedToMoveToSingleStep);
					var localKeys = new HashSet<string>();
					while (enumerator.MoveNext())
					{
						token.ThrowIfCancellationRequested();

						localKeys.Add(enumerator.Current);
					}

					transactionalStorage.Batch(actions =>
					{
						var getItemsToReduceParams = new GetItemsToReduceParams(index: index.IndexId, reduceKeys: localKeys, level: 0, loadData: false, itemsToDelete: itemsToDelete)
						{
							Take = int.MaxValue // just get all, we do the rate limit when we load the number of keys to reduce, anyway
						};

						var getItemsToReduceDuration = Stopwatch.StartNew();

                        int scheduledItemsSum = 0;
                        int scheduledItemsCount = 0;
                        List<int> scheduledItemsMappedBuckets = new List<int>();
						using (StopwatchScope.For(getItemsToReduceDuration))
						{
                            foreach (var item in actions.MapReduce.GetItemsToReduce(getItemsToReduceParams, token))
                            {
                                scheduledItemsMappedBuckets.Add(item.Bucket);
                                scheduledItemsSum += item.Size;
                                scheduledItemsCount++;
                            }
						}

						parallelStats.Operations.Add(PerformanceStats.From(IndexingOperation.Reduce_GetItemsToReduce, getItemsToReduceDuration.ElapsedMilliseconds));

						autoTuner.CurrentlyUsedBatchSizesInBytes.GetOrAdd(reducingBatchThrottlerId, scheduledItemsSum);

                        if (scheduledItemsCount == 0)
						{						    
							// Here we have an interesting issue. We have scheduled reductions, because GetReduceTypesPerKeys() returned them
							// and at the same time, we don't have any at level 0. That probably means that we have them at level 1 or 2.
							// They shouldn't be here, and indeed, we remove them just a little down from here in this function.
							// That said, they might have smuggled in between versions, or something happened to cause them to be here.
							// In order to avoid that, we forcibly delete those extra items from the scheduled reductions, and move on

                            Log.Warn("Found single reduce items ({0}) that didn't have any items to reduce. Deleting level 1 & level 2 items for those keys. (If you can reproduce this, please contact [email protected])", string.Join(", ", keysToReduce));

							var deletingScheduledReductionsDuration = Stopwatch.StartNew();

							using (StopwatchScope.For(deletingScheduledReductionsDuration))
							{
								foreach (var reduceKey in keysToReduce)
								{
									token.ThrowIfCancellationRequested();

									actions.MapReduce.DeleteScheduledReduction(index.IndexId, 1, reduceKey);
									actions.MapReduce.DeleteScheduledReduction(index.IndexId, 2, reduceKey);
								}
							}

							parallelStats.Operations.Add(PerformanceStats.From(IndexingOperation.Reduce_DeleteScheduledReductions, deletingScheduledReductionsDuration.ElapsedMilliseconds));
						}

						var removeReduceResultsDuration = new Stopwatch();

						foreach (var reduceKey in localKeys)
						{
							token.ThrowIfCancellationRequested();

							var lastPerformedReduceType = actions.MapReduce.GetLastPerformedReduceType(index.IndexId, reduceKey);
//.........这里部分代码省略.........

		protected ReducingPerformanceStats[] HandleReduceForIndex(IndexToWorkOn indexToWorkOn, CancellationToken token)
		{
			var viewGenerator = context.IndexDefinitionStorage.GetViewGenerator(indexToWorkOn.IndexId);
			if (viewGenerator == null)
				return null;

			bool operationCanceled = false;
			var itemsToDelete = new ConcurrentSet<object>();

			var singleStepReduceKeys = new List<string>();
			var multiStepsReduceKeys = new List<string>();

			transactionalStorage.Batch(actions =>
			{
                var mappedResultsInfo = actions.MapReduce.GetReduceTypesPerKeys(indexToWorkOn.IndexId, 
                                                                    context.CurrentNumberOfItemsToReduceInSingleBatch, 
                                                                    context.NumberOfItemsToExecuteReduceInSingleStep, token);

			    foreach (var key in mappedResultsInfo)
			    {
				    token.ThrowIfCancellationRequested();

				    switch (key.OperationTypeToPerform)
				    {
					    case ReduceType.SingleStep:
						    singleStepReduceKeys.Add(key.ReduceKey);
						    break;
					    case ReduceType.MultiStep:
						    multiStepsReduceKeys.Add(key.ReduceKey);
						    break;
				    }
			    }
			});

			currentlyProcessedIndexes.TryAdd(indexToWorkOn.IndexId, indexToWorkOn.Index);

			var performanceStats = new List<ReducingPerformanceStats>();

			try
			{
				if (singleStepReduceKeys.Count > 0)
				{
                    if ( Log.IsDebugEnabled )
                        Log.Debug("SingleStep reduce for keys: {0}", singleStepReduceKeys.Select(x => x + ","));
                    
					var singleStepStats = SingleStepReduce(indexToWorkOn, singleStepReduceKeys, viewGenerator, itemsToDelete, token);

					performanceStats.Add(singleStepStats);
				}

				if (multiStepsReduceKeys.Count > 0)
				{
                    if ( Log.IsDebugEnabled )
                        Log.Debug("MultiStep reduce for keys: {0}", multiStepsReduceKeys.Select(x => x + ","));

					var multiStepStats = MultiStepReduce(indexToWorkOn, multiStepsReduceKeys, viewGenerator, itemsToDelete, token);

					performanceStats.Add(multiStepStats);
				}
			}
			catch (OperationCanceledException)
			{
				operationCanceled = true;
			}
			catch (AggregateException e)
			{
				var anyOperationsCanceled = e
					.InnerExceptions
					.OfType<OperationCanceledException>()
					.Any();

				if (anyOperationsCanceled == false) 
					throw;

				operationCanceled = true;
			}
			finally
			{
				var postReducingOperations = new ReduceLevelPeformanceStats
				{
					Level = -1,
					Started = SystemTime.UtcNow
				};

				if (operationCanceled == false)
				{
					var deletingScheduledReductionsDuration = new Stopwatch();
					var storageCommitDuration = new Stopwatch();

					// whatever we succeeded in indexing or not, we have to update this
					// because otherwise we keep trying to re-index failed mapped results
					transactionalStorage.Batch(actions =>
					{
						actions.BeforeStorageCommit += storageCommitDuration.Start;
						actions.AfterStorageCommit += storageCommitDuration.Stop;

						ScheduledReductionInfo latest;

						using (StopwatchScope.For(deletingScheduledReductionsDuration))
						{
//.........这里部分代码省略.........