…lines 1–92 unchanged (92 lines)
93 await item.Key.CancelAsync().ConfigureAwait(false);
94 _taskRunners.Remove(item.Key);
95 }
96 }
97 }
98
99 private bool ShouldForceSequentialOperation()
100 {
101 // if the user either set the setting to 1 or it's unset and we have fewer than 4 cores it's better to run sequentially.
102 var fanoutSetting = _serverConfigurationManager.Configuration.LibraryScanFanoutConcurrency;
103 return fanoutSetting == 1 || (fanoutSetting <= 0 && Environment.ProcessorCount <= 3);
104 }
105
106 private int CalculateScanConcurrencyLimit()
107 {
108 // when this is invoked, we already checked ShouldForceSequentialOperation for the sequential check.
109 var fanoutConcurrency = _serverConfigurationManager.Configuration.LibraryScanFanoutConcurrency;
110 if (fanoutConcurrency <= 0)
111 {
112 // in case the user did not set a limit manually, we can assume he has 3 or more cores as already checked by ShouldForceSequentialOperation.
113 return Environment.ProcessorCount - 3;
114 }
115
116 return fanoutConcurrency;
117 }
118
119 private void Worker()
120 {
121 lock (_taskLock)
122 {
123 var operationFanout = Math.Max(0, CalculateScanConcurrencyLimit() - _taskRunners.Count);
124 _logger.LogDebug("Spawn {NumberRunners} new runners.", operationFanout);
125 for (int i = 0; i < operationFanout; i++)
126 {
127 var stopToken = new CancellationTokenSource();
128 var combinedSource = CancellationTokenSource.CreateLinkedTokenSource(stopToken.Token, _hostApplicationLifetime.ApplicationStopping);
129 _taskRunners.Add(
130 combinedSource,
131 Task.Factory.StartNew(
…lines 132–238 unchanged (107 lines)
239 }),
240 Worker = (val) => worker((T)val, queueItem.Progress),
241 CancellationToken = cancellationToken
242 };
243 }).ToArray();
244
245 if (ShouldForceSequentialOperation())
246 {
247 _logger.LogDebug("Process sequentially.");
248 try
249 {
250 foreach (var item in workItems)
251 {
…lines 252–336 unchanged (85 lines)