Parallelize results parser pipeline
This commit is contained in:
Marek Lemańczyk committed
1 parent
1d19904239
commit
18b6505281
1 file changed
+381
-326
@@ -1,6 +1,4 @@
|
||||
using System.Collections.Concurrent;
|
||||
using System.Globalization;
|
||||
using System.Runtime.CompilerServices;
|
||||
using System.Text;
|
||||
using System.Threading.Tasks;
|
||||
using Open.Numeric.Primes;
|
||||
@@ -51,7 +49,12 @@ internal static class Program
|
||||
return 1;
|
||||
}
|
||||
|
||||
if (!TryParseRangeArguments(args.AsSpan(1), out ulong pMin, out ulong pMax))
|
||||
if (!TryParseProcessingArguments(
|
||||
args.AsSpan(1),
|
||||
out ulong pMin,
|
||||
out ulong pMax,
|
||||
out int threadCount,
|
||||
out int bufferSize))
|
||||
{
|
||||
return 1;
|
||||
}
|
||||
@@ -62,9 +65,21 @@ internal static class Program
|
||||
return 1;
|
||||
}
|
||||
|
||||
if (threadCount < 1)
|
||||
{
|
||||
Console.Error.WriteLine("The --threads value must be a positive integer.");
|
||||
return 1;
|
||||
}
|
||||
|
||||
if (bufferSize < 1)
|
||||
{
|
||||
Console.Error.WriteLine("The --buffer-size value must be a positive integer.");
|
||||
return 1;
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
ProcessFile(inputPath, pMin, pMax);
|
||||
ProcessFile(inputPath, pMin, pMax, threadCount, bufferSize);
|
||||
return 0;
|
||||
}
|
||||
catch (Exception exception)
|
||||
@@ -76,31 +91,32 @@ internal static class Program
|
||||
|
||||
private static bool IsHelpOption(string value) => HelpOptions.Contains(value);
|
||||
|
||||
private static bool TryParseRangeArguments(ReadOnlySpan<string> arguments, out ulong pMin, out ulong pMax)
|
||||
private static bool TryParseProcessingArguments(
|
||||
ReadOnlySpan<string> arguments,
|
||||
out ulong pMin,
|
||||
out ulong pMax,
|
||||
out int threadCount,
|
||||
out int bufferSize)
|
||||
{
|
||||
pMin = 0UL;
|
||||
pMax = ulong.MaxValue;
|
||||
threadCount = Environment.ProcessorCount;
|
||||
bufferSize = 4096;
|
||||
|
||||
if (arguments.Length == 0)
|
||||
{
|
||||
return true;
|
||||
}
|
||||
|
||||
if ((arguments.Length & 1) == 1)
|
||||
int index = 0;
|
||||
while (index < arguments.Length)
|
||||
{
|
||||
string argument = arguments[index];
|
||||
if (!TryReadOptionValue(arguments, ref index, out string option, out string value))
|
||||
{
|
||||
Console.Error.WriteLine("Each range option must be followed by a numeric value.");
|
||||
PrintHelp();
|
||||
return false;
|
||||
}
|
||||
|
||||
int index;
|
||||
string option, optionAndValue, value;
|
||||
for (int i = 0; i < arguments.Length; i += 2)
|
||||
{
|
||||
optionAndValue = arguments[i];
|
||||
index = optionAndValue.IndexOf('=');
|
||||
option = optionAndValue[..index];
|
||||
value = optionAndValue[(index + 1)..];
|
||||
|
||||
try
|
||||
{
|
||||
switch (option)
|
||||
@@ -111,6 +127,12 @@ internal static class Program
|
||||
case "--p-max":
|
||||
pMax = ParseRangeValue(option, value);
|
||||
break;
|
||||
case "--threads":
|
||||
threadCount = ParsePositiveInt(option, value);
|
||||
break;
|
||||
case "--buffer-size":
|
||||
bufferSize = ParsePositiveInt(option, value);
|
||||
break;
|
||||
default:
|
||||
Console.Error.WriteLine($"Unsupported option '{option}'.");
|
||||
PrintHelp();
|
||||
@@ -138,13 +160,62 @@ internal static class Program
|
||||
return parsed;
|
||||
}
|
||||
|
||||
private static int ParsePositiveInt(in string option, in string value)
|
||||
{
|
||||
if (!int.TryParse(value, NumberStyles.Integer, CultureInfo.InvariantCulture, out int parsed) || parsed < 1)
|
||||
{
|
||||
throw new FormatException($"Value '{value}' for option '{option}' is not a valid positive integer.");
|
||||
}
|
||||
|
||||
return parsed;
|
||||
}
|
||||
|
||||
private static bool TryReadOptionValue(
|
||||
ReadOnlySpan<string> arguments,
|
||||
ref int index,
|
||||
out string option,
|
||||
out string value)
|
||||
{
|
||||
option = string.Empty;
|
||||
value = string.Empty;
|
||||
string current = arguments[index];
|
||||
int equalsIndex = current.IndexOf('=');
|
||||
if (equalsIndex > 0)
|
||||
{
|
||||
option = current[..equalsIndex];
|
||||
value = current[(equalsIndex + 1)..];
|
||||
index++;
|
||||
if (value.Length == 0)
|
||||
{
|
||||
Console.Error.WriteLine($"Option '{option}' must specify a value.");
|
||||
PrintHelp();
|
||||
return false;
|
||||
}
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
option = current;
|
||||
if (index + 1 >= arguments.Length)
|
||||
{
|
||||
Console.Error.WriteLine($"Option '{option}' must be followed by a value.");
|
||||
PrintHelp();
|
||||
return false;
|
||||
}
|
||||
|
||||
value = arguments[index + 1];
|
||||
index += 2;
|
||||
return true;
|
||||
}
|
||||
|
||||
private static void PrintHelp()
|
||||
{
|
||||
Console.WriteLine("EvenPerfectBitScanner.ResultsParser");
|
||||
Console.WriteLine("Parses CSV results produced by the candidate scanner.");
|
||||
Console.WriteLine();
|
||||
Console.WriteLine("Usage:");
|
||||
Console.WriteLine(" EvenPerfectBitScanner.ResultsParser <results.csv> [--p-min <value>] [--p-max <value>]");
|
||||
Console.WriteLine(
|
||||
" EvenPerfectBitScanner.ResultsParser <results.csv> [--p-min <value>] [--p-max <value>] [--threads <value>] [--buffer-size <value>]");
|
||||
Console.WriteLine();
|
||||
Console.WriteLine("The program accepts the following help switches:");
|
||||
Console.WriteLine(" --?, -?, /?, -help, --help, --h, -h, /h, /help");
|
||||
@@ -153,6 +224,11 @@ internal static class Program
|
||||
Console.WriteLine(" --p-min <value> - include entries with p >= value");
|
||||
Console.WriteLine(" --p-max <value> - include entries with p <= value");
|
||||
Console.WriteLine();
|
||||
Console.WriteLine("Parallel processing options:");
|
||||
Console.WriteLine(" --threads <value> - number of worker threads used for parsing and splitting (default: logical cores)");
|
||||
Console.WriteLine(" --buffer-size <value> - number of records read into each parsing batch (default: 4096)");
|
||||
Console.WriteLine("Values can also be provided using the --option=value syntax.");
|
||||
Console.WriteLine();
|
||||
Console.WriteLine("Outputs prime results into four files placed next to the source file:");
|
||||
Console.WriteLine(" raw-primes-<name> - prime entries in the original order");
|
||||
Console.WriteLine(" sorted-primes-<name> - prime entries sorted by p");
|
||||
@@ -163,12 +239,11 @@ internal static class Program
|
||||
Console.WriteLine(" EvenPerfectBitScanner.ResultsParser results.csv --p-min 89 --p-max 107");
|
||||
}
|
||||
|
||||
private const int CandidateBufferCapacity = 512;
|
||||
|
||||
private static void ProcessFile(in string inputPath, ulong pMin, ulong pMax)
|
||||
private static void ProcessFile(in string inputPath, ulong pMin, ulong pMax, int threadCount, int bufferSize)
|
||||
{
|
||||
using IEnumerator<string> enumerator = File.ReadLines(inputPath).GetEnumerator();
|
||||
if (!enumerator.MoveNext())
|
||||
using StreamReader reader = new(inputPath);
|
||||
string? headerLine = reader.ReadLine();
|
||||
if (headerLine is null)
|
||||
{
|
||||
Console.WriteLine("Input file did not contain any data. Created empty outputs using the default header.");
|
||||
string fallbackHeader = DefaultHeader;
|
||||
@@ -178,66 +253,18 @@ internal static class Program
|
||||
return;
|
||||
}
|
||||
|
||||
string headerLine = enumerator.Current;
|
||||
string header = string.IsNullOrWhiteSpace(headerLine) ? DefaultHeader : headerLine.Trim();
|
||||
|
||||
using IEnumerator<ulong> primeEnumerator = Prime.Numbers.GetEnumerator();
|
||||
if (!primeEnumerator.MoveNext())
|
||||
{
|
||||
throw new InvalidOperationException("Prime generator did not provide any values.");
|
||||
}
|
||||
List<CandidateResult> candidates = LoadCandidates(reader, pMin, pMax, threadCount, bufferSize);
|
||||
|
||||
ulong currentPrime = primeEnumerator.Current;
|
||||
PendingCandidateList pendingResults = new();
|
||||
List<CandidateResult> primeResults = [];
|
||||
int consoleProgress = 0;
|
||||
ulong lastReportedCandidate = 0UL;
|
||||
CandidateResult[] candidateBuffer = new CandidateResult[CandidateBufferCapacity];
|
||||
int bufferedCount = 0;
|
||||
ulong pEmpty = CandidateResult.Empty.P;
|
||||
while (enumerator.MoveNext())
|
||||
{
|
||||
string currentLine = enumerator.Current;
|
||||
if (string.IsNullOrWhiteSpace(currentLine))
|
||||
{
|
||||
continue;
|
||||
}
|
||||
|
||||
CandidateResult result = ParseLine(currentLine, pMin, pMax);
|
||||
if (result.P == pEmpty)
|
||||
{
|
||||
continue;
|
||||
}
|
||||
|
||||
candidateBuffer[bufferedCount++] = result;
|
||||
lastReportedCandidate = result.P;
|
||||
if (bufferedCount == CandidateBufferCapacity)
|
||||
{
|
||||
ProcessCandidateBuffer(candidateBuffer.AsSpan(0, bufferedCount), pendingResults);
|
||||
DrainPrimeMatches(pendingResults, primeResults, ref currentPrime, primeEnumerator, false);
|
||||
bufferedCount = 0;
|
||||
}
|
||||
|
||||
if (++consoleProgress == 10_000)
|
||||
{
|
||||
Console.WriteLine($"Processed {lastReportedCandidate}");
|
||||
consoleProgress = 0;
|
||||
}
|
||||
}
|
||||
|
||||
if (bufferedCount > 0)
|
||||
{
|
||||
ProcessCandidateBuffer(candidateBuffer.AsSpan(0, bufferedCount), pendingResults);
|
||||
DrainPrimeMatches(pendingResults, primeResults, ref currentPrime, primeEnumerator, false);
|
||||
}
|
||||
|
||||
DrainPrimeMatches(pendingResults, primeResults, ref currentPrime, primeEnumerator, true);
|
||||
List<CandidateResult> sortedPrimeResults = ExtractSortedPrimeCandidates(candidates);
|
||||
List<CandidateResult> primeResults = BuildRawPrimeResults(candidates, sortedPrimeResults);
|
||||
|
||||
string rawOutputPath = BuildOutputPath(inputPath, "raw-primes-");
|
||||
bool append = File.Exists(rawOutputPath);
|
||||
WriteCsv(rawOutputPath, header, append, primeResults);
|
||||
|
||||
if (primeResults.Count == 0)
|
||||
if (sortedPrimeResults.Count == 0)
|
||||
{
|
||||
Console.WriteLine("No prime entries were found in the input file.");
|
||||
Console.WriteLine($"Created an empty prime snapshot at: {rawOutputPath}");
|
||||
@@ -245,12 +272,11 @@ internal static class Program
|
||||
return;
|
||||
}
|
||||
|
||||
List<CandidateResult> sortedResults = [.. primeResults];
|
||||
Console.WriteLine("Sorting candidates...");
|
||||
SortByPrime(sortedResults);
|
||||
SortByPrime(sortedPrimeResults);
|
||||
|
||||
Console.WriteLine("Splitting results...");
|
||||
SplitResults(sortedResults, out List<CandidateResult> passedResults, out List<CandidateResult> rejectedResults);
|
||||
SplitResults(sortedPrimeResults, threadCount, out List<CandidateResult> passedResults, out List<CandidateResult> rejectedResults);
|
||||
|
||||
Console.WriteLine("Saving files...");
|
||||
string sortedOutputPath = BuildOutputPath(inputPath, "sorted-primes-");
|
||||
@@ -258,13 +284,13 @@ internal static class Program
|
||||
string rejectedOutputPath = BuildOutputPath(inputPath, "sorted-primes-rejected-");
|
||||
|
||||
append = File.Exists(sortedOutputPath);
|
||||
WriteCsv(sortedOutputPath, header, append, sortedResults);
|
||||
WriteCsv(sortedOutputPath, header, append, sortedPrimeResults);
|
||||
append = File.Exists(passedOutputPath);
|
||||
WriteCsv(passedOutputPath, header, append, passedResults);
|
||||
append = File.Exists(rejectedOutputPath);
|
||||
WriteCsv(rejectedOutputPath, header, append, rejectedResults);
|
||||
|
||||
Console.WriteLine($"Processed {primeResults.Count} prime entr{(primeResults.Count == 1 ? "y" : "ies")}.");
|
||||
Console.WriteLine($"Processed {sortedPrimeResults.Count} prime entr{(sortedPrimeResults.Count == 1 ? "y" : "ies")}.");
|
||||
Console.WriteLine($"Prime results (raw order): {rawOutputPath}");
|
||||
Console.WriteLine($"Prime results (sorted): {sortedOutputPath}");
|
||||
Console.WriteLine($"Prime results (passed): {passedOutputPath}");
|
||||
@@ -329,36 +355,237 @@ internal static class Program
|
||||
}
|
||||
}
|
||||
|
||||
private static void ProcessCandidateBuffer(Span<CandidateResult> buffer, PendingCandidateList pendingResults)
|
||||
private static List<CandidateResult> LoadCandidates(
|
||||
StreamReader reader,
|
||||
ulong pMin,
|
||||
ulong pMax,
|
||||
int threadCount,
|
||||
int bufferSize)
|
||||
{
|
||||
if (buffer.Length == 0)
|
||||
List<CandidateResult> candidates = [];
|
||||
if (bufferSize < 1)
|
||||
{
|
||||
return candidates;
|
||||
}
|
||||
|
||||
int workerLimit = Math.Max(1, threadCount);
|
||||
List<Task<ChunkResult>> activeChunks = new(workerLimit);
|
||||
Dictionary<int, ChunkResult> completedChunks = new();
|
||||
int nextChunkToFlush = 0;
|
||||
int consoleProgress = 0;
|
||||
ulong lastReportedCandidate = 0UL;
|
||||
|
||||
string[] buffer = new string[bufferSize];
|
||||
int buffered = 0;
|
||||
int chunkIndex = 0;
|
||||
|
||||
void FlushChunk(in ChunkResult chunk)
|
||||
{
|
||||
if (chunk.Count == 0)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
buffer.Sort(CandidateResultComparer.Instance);
|
||||
|
||||
int start = 0;
|
||||
int length = buffer.Length;
|
||||
while (start < length)
|
||||
CandidateResult[] items = chunk.Items;
|
||||
int count = chunk.Count;
|
||||
candidates.EnsureCapacity(candidates.Count + count);
|
||||
for (int i = 0; i < count; i++)
|
||||
{
|
||||
int end = start + 1;
|
||||
ulong previous = buffer[start].P;
|
||||
while (end < length)
|
||||
CandidateResult candidate = items[i];
|
||||
candidates.Add(candidate);
|
||||
lastReportedCandidate = candidate.P;
|
||||
if (++consoleProgress == 10_000)
|
||||
{
|
||||
ulong current = buffer[end].P;
|
||||
if (current > previous + 1)
|
||||
{
|
||||
break;
|
||||
Console.WriteLine($"Processed {lastReportedCandidate}");
|
||||
consoleProgress = 0;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
previous = current;
|
||||
end++;
|
||||
void TryFlush()
|
||||
{
|
||||
while (completedChunks.Remove(nextChunkToFlush, out ChunkResult ready))
|
||||
{
|
||||
FlushChunk(in ready);
|
||||
nextChunkToFlush++;
|
||||
}
|
||||
}
|
||||
|
||||
pendingResults.AddRange(buffer[start..end]);
|
||||
start = end;
|
||||
void WaitForAny()
|
||||
{
|
||||
Task<ChunkResult> completed = Task.WhenAny(activeChunks).GetAwaiter().GetResult();
|
||||
activeChunks.Remove(completed);
|
||||
ChunkResult chunk = completed.GetAwaiter().GetResult();
|
||||
completedChunks.Add(chunk.Index, chunk);
|
||||
TryFlush();
|
||||
}
|
||||
|
||||
void ScheduleChunk(string[] chunkLines)
|
||||
{
|
||||
int currentIndex = chunkIndex++;
|
||||
Task<ChunkResult> task = Task.Run(() => ProcessChunk(chunkLines, currentIndex, pMin, pMax));
|
||||
activeChunks.Add(task);
|
||||
if (activeChunks.Count == workerLimit)
|
||||
{
|
||||
WaitForAny();
|
||||
}
|
||||
}
|
||||
|
||||
void DispatchBuffer(int length)
|
||||
{
|
||||
string[] chunkLines;
|
||||
if (length == buffer.Length)
|
||||
{
|
||||
chunkLines = buffer;
|
||||
buffer = new string[bufferSize];
|
||||
}
|
||||
else
|
||||
{
|
||||
chunkLines = new string[length];
|
||||
Array.Copy(buffer, chunkLines, length);
|
||||
}
|
||||
|
||||
buffered = 0;
|
||||
ScheduleChunk(chunkLines);
|
||||
}
|
||||
|
||||
string? line;
|
||||
while ((line = reader.ReadLine()) is not null)
|
||||
{
|
||||
if (string.IsNullOrWhiteSpace(line))
|
||||
{
|
||||
continue;
|
||||
}
|
||||
|
||||
buffer[buffered++] = line;
|
||||
if (buffered == bufferSize)
|
||||
{
|
||||
DispatchBuffer(buffered);
|
||||
}
|
||||
}
|
||||
|
||||
if (buffered > 0)
|
||||
{
|
||||
DispatchBuffer(buffered);
|
||||
}
|
||||
|
||||
while (activeChunks.Count > 0)
|
||||
{
|
||||
WaitForAny();
|
||||
}
|
||||
|
||||
TryFlush();
|
||||
return candidates;
|
||||
}
|
||||
|
||||
private static ChunkResult ProcessChunk(string[] lines, int index, ulong pMin, ulong pMax)
|
||||
{
|
||||
CandidateResult[] items = new CandidateResult[lines.Length];
|
||||
int count = 0;
|
||||
ulong pEmpty = CandidateResult.Empty.P;
|
||||
for (int i = 0; i < lines.Length; i++)
|
||||
{
|
||||
string line = lines[i];
|
||||
if (string.IsNullOrWhiteSpace(line))
|
||||
{
|
||||
continue;
|
||||
}
|
||||
|
||||
CandidateResult candidate = ParseLine(line, pMin, pMax);
|
||||
if (candidate.P == pEmpty)
|
||||
{
|
||||
continue;
|
||||
}
|
||||
|
||||
items[count++] = candidate;
|
||||
}
|
||||
|
||||
return new ChunkResult(index, items, count);
|
||||
}
|
||||
|
||||
private static List<CandidateResult> ExtractSortedPrimeCandidates(List<CandidateResult> candidates)
|
||||
{
|
||||
if (candidates.Count == 0)
|
||||
{
|
||||
return [];
|
||||
}
|
||||
|
||||
List<CandidateResult> sortedCandidates = [.. candidates];
|
||||
sortedCandidates.Sort(CandidateResultComparer.Instance);
|
||||
|
||||
List<CandidateResult> primeCandidates = [];
|
||||
using IEnumerator<ulong> primeEnumerator = Prime.Numbers.GetEnumerator();
|
||||
if (!primeEnumerator.MoveNext())
|
||||
{
|
||||
throw new InvalidOperationException("Prime generator did not provide any values.");
|
||||
}
|
||||
|
||||
ulong currentPrime = primeEnumerator.Current;
|
||||
foreach (CandidateResult candidate in sortedCandidates)
|
||||
{
|
||||
ulong candidateP = candidate.P;
|
||||
while (currentPrime < candidateP)
|
||||
{
|
||||
if (!primeEnumerator.MoveNext())
|
||||
{
|
||||
return primeCandidates;
|
||||
}
|
||||
|
||||
currentPrime = primeEnumerator.Current;
|
||||
}
|
||||
|
||||
if (currentPrime == candidateP)
|
||||
{
|
||||
primeCandidates.Add(candidate);
|
||||
}
|
||||
}
|
||||
|
||||
return primeCandidates;
|
||||
}
|
||||
|
||||
private static List<CandidateResult> BuildRawPrimeResults(
|
||||
List<CandidateResult> candidates,
|
||||
List<CandidateResult> sortedPrimeResults)
|
||||
{
|
||||
int primeCount = sortedPrimeResults.Count;
|
||||
if (primeCount == 0)
|
||||
{
|
||||
return [];
|
||||
}
|
||||
|
||||
Dictionary<ulong, int> remainingCounts = new(primeCount);
|
||||
for (int i = 0; i < primeCount; i++)
|
||||
{
|
||||
CandidateResult candidate = sortedPrimeResults[i];
|
||||
if (remainingCounts.TryGetValue(candidate.P, out int count))
|
||||
{
|
||||
remainingCounts[candidate.P] = count + 1;
|
||||
continue;
|
||||
}
|
||||
|
||||
remainingCounts.Add(candidate.P, 1);
|
||||
}
|
||||
|
||||
List<CandidateResult> rawPrimes = new(primeCount);
|
||||
for (int i = 0; i < candidates.Count; i++)
|
||||
{
|
||||
CandidateResult candidate = candidates[i];
|
||||
if (!remainingCounts.TryGetValue(candidate.P, out int count))
|
||||
{
|
||||
continue;
|
||||
}
|
||||
|
||||
rawPrimes.Add(candidate);
|
||||
if (count == 1)
|
||||
{
|
||||
remainingCounts.Remove(candidate.P);
|
||||
continue;
|
||||
}
|
||||
|
||||
remainingCounts[candidate.P] = count - 1;
|
||||
}
|
||||
|
||||
return rawPrimes;
|
||||
}
|
||||
|
||||
private static void SortByPrime(List<CandidateResult> results)
|
||||
@@ -373,6 +600,7 @@ internal static class Program
|
||||
|
||||
private static void SplitResults(
|
||||
List<CandidateResult> sortedResults,
|
||||
int threadCount,
|
||||
out List<CandidateResult> passedResults,
|
||||
out List<CandidateResult> rejectedResults)
|
||||
{
|
||||
@@ -384,7 +612,7 @@ internal static class Program
|
||||
return;
|
||||
}
|
||||
|
||||
if (count < Environment.ProcessorCount * 4)
|
||||
if (threadCount <= 1 || count < threadCount * 4)
|
||||
{
|
||||
passedResults = new List<CandidateResult>(count);
|
||||
rejectedResults = new List<CandidateResult>(count);
|
||||
@@ -403,18 +631,34 @@ internal static class Program
|
||||
return;
|
||||
}
|
||||
|
||||
List<CandidateResult> sharedPassed = new(count);
|
||||
List<CandidateResult> sharedRejected = new(count);
|
||||
object passedLock = new();
|
||||
object rejectedLock = new();
|
||||
Parallel.ForEach(
|
||||
Partitioner.Create(0, count),
|
||||
() => new SplitAccumulator(),
|
||||
(range, _, local) =>
|
||||
int workerCount = Math.Min(threadCount, count);
|
||||
List<CandidateResult>[] passedSegments = new List<CandidateResult>[workerCount];
|
||||
List<CandidateResult>[] rejectedSegments = new List<CandidateResult>[workerCount];
|
||||
Task[] tasks = new Task[workerCount];
|
||||
|
||||
int start = 0;
|
||||
int baseChunk = count / workerCount;
|
||||
int remainder = count % workerCount;
|
||||
for (int worker = 0; worker < workerCount; worker++)
|
||||
{
|
||||
List<CandidateResult> localPassed = local.Passed;
|
||||
List<CandidateResult> localRejected = local.Rejected;
|
||||
for (int i = range.Item1; i < range.Item2; i++)
|
||||
int length = baseChunk + (worker < remainder ? 1 : 0);
|
||||
int chunkStart = start;
|
||||
start += length;
|
||||
|
||||
if (length == 0)
|
||||
{
|
||||
passedSegments[worker] = [];
|
||||
rejectedSegments[worker] = [];
|
||||
tasks[worker] = Task.CompletedTask;
|
||||
continue;
|
||||
}
|
||||
|
||||
tasks[worker] = Task.Run(() =>
|
||||
{
|
||||
List<CandidateResult> localPassed = new(length);
|
||||
List<CandidateResult> localRejected = new(length);
|
||||
int end = chunkStart + length;
|
||||
for (int i = chunkStart; i < end; i++)
|
||||
{
|
||||
CandidateResult candidate = sortedResults[i];
|
||||
if (candidate.PassedAllTests)
|
||||
@@ -426,29 +670,27 @@ internal static class Program
|
||||
localRejected.Add(candidate);
|
||||
}
|
||||
|
||||
return local;
|
||||
},
|
||||
local =>
|
||||
{
|
||||
if (local.Passed.Count > 0)
|
||||
{
|
||||
lock (passedLock)
|
||||
{
|
||||
sharedPassed.AddRange(local.Passed);
|
||||
}
|
||||
}
|
||||
|
||||
if (local.Rejected.Count > 0)
|
||||
{
|
||||
lock (rejectedLock)
|
||||
{
|
||||
sharedRejected.AddRange(local.Rejected);
|
||||
}
|
||||
}
|
||||
passedSegments[worker] = localPassed;
|
||||
rejectedSegments[worker] = localRejected;
|
||||
});
|
||||
}
|
||||
|
||||
passedResults = sharedPassed;
|
||||
rejectedResults = sharedRejected;
|
||||
Task.WaitAll(tasks);
|
||||
|
||||
passedResults = new List<CandidateResult>(count);
|
||||
rejectedResults = new List<CandidateResult>(count);
|
||||
for (int worker = 0; worker < workerCount; worker++)
|
||||
{
|
||||
if (passedSegments[worker].Count > 0)
|
||||
{
|
||||
passedResults.AddRange(passedSegments[worker]);
|
||||
}
|
||||
|
||||
if (rejectedSegments[worker].Count > 0)
|
||||
{
|
||||
rejectedResults.AddRange(rejectedSegments[worker]);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private sealed class CandidateResultComparer : IComparer<CandidateResult>
|
||||
@@ -466,204 +708,17 @@ internal static class Program
|
||||
}
|
||||
}
|
||||
|
||||
private static void DrainPrimeMatches(
|
||||
PendingCandidateList pendingResults,
|
||||
List<CandidateResult> primeResults,
|
||||
ref ulong currentPrime,
|
||||
IEnumerator<ulong> primeEnumerator,
|
||||
bool inputCompleted)
|
||||
private readonly struct ChunkResult
|
||||
{
|
||||
if (pendingResults.Count == 0)
|
||||
{
|
||||
return;
|
||||
}
|
||||
public readonly int Index;
|
||||
public readonly CandidateResult[] Items;
|
||||
public readonly int Count;
|
||||
|
||||
pendingResults.RemoveLessThan(currentPrime);
|
||||
|
||||
bool extractedPrime;
|
||||
while (pendingResults.Count > 0)
|
||||
public ChunkResult(int index, CandidateResult[] items, int count)
|
||||
{
|
||||
extractedPrime = ExtractPrimeEntries(pendingResults, primeResults, currentPrime);
|
||||
if (extractedPrime)
|
||||
{
|
||||
if (!TryAdvancePrime(primeEnumerator, ref currentPrime, pendingResults))
|
||||
{
|
||||
pendingResults.Clear();
|
||||
return;
|
||||
}
|
||||
|
||||
pendingResults.RemoveLessThan(currentPrime);
|
||||
continue;
|
||||
}
|
||||
|
||||
if (!inputCompleted)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
if (!TryAdvancePrime(primeEnumerator, ref currentPrime, pendingResults))
|
||||
{
|
||||
pendingResults.Clear();
|
||||
return;
|
||||
}
|
||||
|
||||
pendingResults.RemoveLessThan(currentPrime);
|
||||
}
|
||||
}
|
||||
|
||||
private static bool ExtractPrimeEntries(
|
||||
PendingCandidateList pendingResults,
|
||||
List<CandidateResult> primeResults,
|
||||
ulong prime)
|
||||
{
|
||||
if (pendingResults.Count == 0)
|
||||
{
|
||||
return false;
|
||||
}
|
||||
|
||||
if (!pendingResults.TryPeek(out CandidateResult next) || next.P != prime)
|
||||
{
|
||||
return false;
|
||||
}
|
||||
|
||||
do
|
||||
{
|
||||
primeResults.Add(pendingResults.Dequeue());
|
||||
}
|
||||
while (pendingResults.TryPeek(out CandidateResult candidate) && candidate.P == prime);
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
private static bool TryAdvancePrime(
|
||||
IEnumerator<ulong> primeEnumerator,
|
||||
ref ulong currentPrime,
|
||||
PendingCandidateList pendingResults)
|
||||
{
|
||||
ulong minimumPending = pendingResults.Count > 0 && pendingResults.TryPeek(out CandidateResult pending)
|
||||
? pending.P
|
||||
: 0UL;
|
||||
do
|
||||
{
|
||||
if (!primeEnumerator.MoveNext())
|
||||
{
|
||||
currentPrime = ulong.MaxValue;
|
||||
return false;
|
||||
}
|
||||
|
||||
currentPrime = primeEnumerator.Current;
|
||||
}
|
||||
while (pendingResults.Count > 0 && currentPrime < minimumPending);
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
private sealed class PendingCandidateList
|
||||
{
|
||||
private readonly List<CandidateResult> _items = [];
|
||||
|
||||
public int Count => _items.Count;
|
||||
|
||||
[MethodImpl(MethodImplOptions.AggressiveInlining)]
|
||||
public void AddRange(ReadOnlySpan<CandidateResult> candidates)
|
||||
{
|
||||
if (candidates.Length == 0)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
List<CandidateResult> items = _items;
|
||||
items.EnsureCapacity(items.Count + candidates.Length);
|
||||
int insertIndex = FindInsertIndex(items, candidates[0].P);
|
||||
int candidateIndex = 0;
|
||||
int candidateCount = candidates.Length;
|
||||
while (candidateIndex < candidateCount)
|
||||
{
|
||||
CandidateResult candidate = candidates[candidateIndex];
|
||||
while (insertIndex < items.Count && items[insertIndex].P <= candidate.P)
|
||||
{
|
||||
insertIndex++;
|
||||
}
|
||||
|
||||
items.Insert(insertIndex, candidate);
|
||||
insertIndex++;
|
||||
candidateIndex++;
|
||||
}
|
||||
}
|
||||
|
||||
[MethodImpl(MethodImplOptions.AggressiveInlining)]
|
||||
public bool TryPeek(out CandidateResult result)
|
||||
{
|
||||
if (_items.Count == 0)
|
||||
{
|
||||
result = default;
|
||||
return false;
|
||||
}
|
||||
|
||||
result = _items[0];
|
||||
return true;
|
||||
}
|
||||
|
||||
[MethodImpl(MethodImplOptions.AggressiveInlining)]
|
||||
public CandidateResult Dequeue()
|
||||
{
|
||||
CandidateResult value = _items[0];
|
||||
_items.RemoveAt(0);
|
||||
return value;
|
||||
}
|
||||
|
||||
[MethodImpl(MethodImplOptions.AggressiveInlining)]
|
||||
public void RemoveLessThan(ulong prime)
|
||||
{
|
||||
int removeCount = 0;
|
||||
List<CandidateResult> items = _items;
|
||||
int count = items.Count;
|
||||
while (removeCount < count && items[removeCount].P < prime)
|
||||
{
|
||||
removeCount++;
|
||||
}
|
||||
|
||||
if (removeCount > 0)
|
||||
{
|
||||
items.RemoveRange(0, removeCount);
|
||||
}
|
||||
}
|
||||
|
||||
[MethodImpl(MethodImplOptions.AggressiveInlining)]
|
||||
public void Clear()
|
||||
{
|
||||
_items.Clear();
|
||||
}
|
||||
|
||||
private static int FindInsertIndex(List<CandidateResult> items, ulong value)
|
||||
{
|
||||
int low = 0;
|
||||
int high = items.Count;
|
||||
while (low < high)
|
||||
{
|
||||
int mid = (low + high) >> 1;
|
||||
if (items[mid].P < value)
|
||||
{
|
||||
low = mid + 1;
|
||||
continue;
|
||||
}
|
||||
|
||||
high = mid;
|
||||
}
|
||||
|
||||
return low;
|
||||
}
|
||||
}
|
||||
|
||||
private readonly struct SplitAccumulator
|
||||
{
|
||||
public readonly List<CandidateResult> Passed;
|
||||
public readonly List<CandidateResult> Rejected;
|
||||
|
||||
public SplitAccumulator()
|
||||
{
|
||||
Passed = [];
|
||||
Rejected = [];
|
||||
Index = index;
|
||||
Items = items;
|
||||
Count = count;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in new issue
Block a user