Skip to content

Commit 08c370a

Browse files
committed
draft
1 parent 8213cc2 commit 08c370a

File tree

4 files changed

+66
-5
lines changed

4 files changed

+66
-5
lines changed

RecordParser/Engines/ExpressionHelper.cs

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -9,13 +9,13 @@ public static Expression Call(Delegate f, params Expression[] args) =>
99
Expression.Call(f.Target is null ? null : Expression.Constant(f.Target), f.Method, args);
1010

1111
public static Expression StringAsSpan(Expression str) =>
12-
Expression.Call(typeof(MemoryExtensions), "AsSpan", Type.EmptyTypes, str);
12+
Expression.Call(typeof(MemoryExtensions), nameof(MemoryExtensions.AsSpan), Type.EmptyTypes, str);
1313

1414
public static Expression SpanAsString(Expression span) =>
1515
Expression.Call(span, "ToString", Type.EmptyTypes);
1616

1717
public static Expression Trim(Expression str) =>
18-
Expression.Call(typeof(MemoryExtensions), "Trim", Type.EmptyTypes, str);
18+
Expression.Call(typeof(MemoryExtensions), nameof(MemoryExtensions.Trim), Type.EmptyTypes, str);
1919

2020
public static Expression Slice(Expression span, Expression start) =>
2121
Expression.Call(span, "Slice", Type.EmptyTypes, start);
@@ -30,6 +30,6 @@ public static Expression Slice(Expression span, Expression start, Expression len
3030
Expression.Call(span, "Slice", Type.EmptyTypes, start, length);
3131

3232
public static Expression IsWhiteSpace(Expression valueText) =>
33-
Expression.Call(typeof(MemoryExtensions), "IsWhiteSpace", Type.EmptyTypes, valueText);
33+
Expression.Call(typeof(MemoryExtensions), nameof(MemoryExtensions.IsWhiteSpace), Type.EmptyTypes, valueText);
3434
}
3535
}

RecordParser/Engines/Reader/PrimitiveTypeReaderEngine.cs

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -78,7 +78,7 @@ private static Expression GetEnumFromSpanParseExpression(Type type, Expression s
7878
{
7979
var enumText = color.ToString();
8080

81-
var compareTo = Expression.Call(typeof(MemoryExtensions), "CompareTo", Type.EmptyTypes,
81+
var compareTo = Expression.Call(typeof(MemoryExtensions), nameof(MemoryExtensions.CompareTo), Type.EmptyTypes,
8282
StringAsSpan(Expression.Constant(enumText)),
8383
trim,
8484
Expression.Constant(StringComparison.OrdinalIgnoreCase));

RecordParser/Extensions/FileReader/ReaderCommon.cs

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@
33
using System.Collections.Generic;
44
using System.Linq;
55
using System.Threading;
6+
using System.Threading.Tasks;
67

78
namespace RecordParser.Extensions
89
{
@@ -44,6 +45,19 @@ private static IEnumerable<T> Skip<T>(this IEnumerable<T> source, bool hasHeader
4445
? source.Skip(1)
4546
: source;
4647

48+
public static ParallelOptions AsParallel(this ParallelismOptions option)
49+
{
50+
var query = new ParallelOptions();
51+
52+
if (option.MaxDegreeOfParallelism is { } degree)
53+
query.MaxDegreeOfParallelism = degree;
54+
55+
if (option.CancellationToken is { } token)
56+
query.CancellationToken = token;
57+
58+
return query;
59+
}
60+
4761
public static ParallelQuery<T> AsParallel<T>(this IEnumerable<T> source, ParallelismOptions option)
4862
{
4963
var query = source.AsParallel();

RecordParser/Extensions/FileWriter/WriterExtensions.cs

Lines changed: 48 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@
33
using System.Collections.Generic;
44
using System.IO;
55
using System.Linq;
6+
using System.Threading.Tasks;
67

78
namespace RecordParser.Extensions
89
{
@@ -47,14 +48,60 @@ public static void WriteRecords<T>(this TextWriter textWriter, IEnumerable<T> it
4748
{
4849
if (options.Enabled)
4950
{
50-
WriteParallel(textWriter, items, tryFormat, options);
51+
if (options.EnsureOriginalOrdering)
52+
WriteParallel(textWriter, items, tryFormat, options);
53+
else
54+
WriteParallelUnordered(textWriter, items, tryFormat, options);
5155
}
5256
else
5357
{
5458
WriteSequential(textWriter, items, tryFormat);
5559
}
5660
}
5761

62+
private class BufferContext
63+
{
64+
public char[] buffer;
65+
public object lockObj;
66+
}
67+
68+
private static void WriteParallelUnordered<T>(TextWriter textWriter, IEnumerable<T> items, TryFormat<T> tryFormat, ParallelismOptions options)
69+
{
70+
var poolSize = 1_000;
71+
var textWriterLock = new object();
72+
var opt = options.AsParallel();
73+
var pool = Enumerable
74+
.Range(0, poolSize)
75+
.Select(_ => new BufferContext
76+
{
77+
buffer = new char[(int)Math.Pow(2, initialPow)],
78+
lockObj = new object()
79+
})
80+
.ToArray();
81+
82+
Parallel.ForEach(items, opt, (item, _, i) =>
83+
{
84+
var x = pool[i % poolSize];
85+
86+
lock (x.lockObj)
87+
{
88+
x = pool[i % poolSize];
89+
retry:
90+
91+
if (tryFormat(item, x.buffer, out var charsWritten))
92+
{
93+
lock (textWriterLock)
94+
textWriter.WriteLine(x.buffer, 0, charsWritten);
95+
}
96+
else
97+
{
98+
x.buffer = new char[x.buffer.Length * 2];
99+
goto retry;
100+
}
101+
}
102+
});
103+
}
104+
58105
private static void WriteParallel<T>(TextWriter textWriter, IEnumerable<T> items, TryFormat<T> tryFormat, ParallelismOptions options)
59106
{
60107
var poolSize = 10_000;

0 commit comments

Comments
 (0)