Skip to content

Commit

Permalink
Making static instances thread safe (#161)
Browse files Browse the repository at this point in the history 
* Making static instances thread safe

* reverting changes that were already protected by locks
  • Loading branch information
@peterfreiling
peterfreiling authored Feb 11, 2022
1 parent c6d4e2e commit 2039f16
Show file tree
Hide file tree
Showing 5 changed files with 107 additions and 104 deletions.
99 changes: 51 additions & 48 deletions Sources/Core/Microsoft.StreamProcessing/Serializer/Serializers/ReflectionSchemaBuilder.cs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
// Licensed under the MIT License
// *********************************************************************
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Reflection;
Expand All @@ -13,59 +14,61 @@ namespace Microsoft.StreamProcessing.Serializer.Serializers
{
internal sealed class ReflectionSchemaBuilder
{
private static readonly Dictionary<Type, Func<ObjectSerializerBase>> RuntimeTypeToSerializer =
new Dictionary<Type, Func<ObjectSerializerBase>>
{
{ typeof(char), () => PrimitiveSerializer.Char },
{ typeof(byte), () => PrimitiveSerializer.Byte },
{ typeof(sbyte), () => PrimitiveSerializer.SByte },
{ typeof(short), () => PrimitiveSerializer.Short },
{ typeof(ushort), () => PrimitiveSerializer.UShort },
{ typeof(int), () => PrimitiveSerializer.Int },
{ typeof(uint), () => PrimitiveSerializer.UInt },
{ typeof(bool), () => PrimitiveSerializer.Boolean },
{ typeof(long), () => PrimitiveSerializer.Long },
{ typeof(ulong), () => PrimitiveSerializer.ULong },
{ typeof(float), () => PrimitiveSerializer.Float },
{ typeof(double), () => PrimitiveSerializer.Double },
{ typeof(decimal), () => PrimitiveSerializer.Decimal },
{ typeof(string), () => PrimitiveSerializer.String },
{ typeof(Uri), () => PrimitiveSerializer.Uri },
{ typeof(TimeSpan), () => PrimitiveSerializer.TimeSpan },
{ typeof(DateTime), () => PrimitiveSerializer.DateTime },
{ typeof(DateTimeOffset), () => PrimitiveSerializer.DateTimeOffset },
{ typeof(Guid), () => PrimitiveSerializer.Guid },

{ typeof(char[]), () => PrimitiveSerializer.CreateForArray<char>() },
{ typeof(byte[]), () => PrimitiveSerializer.ByteArray },
{ typeof(short[]), () => PrimitiveSerializer.CreateForArray<short>() },
{ typeof(ushort[]), () => PrimitiveSerializer.CreateForArray<ushort>() },
{ typeof(int[]), () => PrimitiveSerializer.CreateForArray<int>() },
{ typeof(uint[]), () => PrimitiveSerializer.CreateForArray<uint>() },
{ typeof(long[]), () => PrimitiveSerializer.CreateForArray<long>() },
{ typeof(ulong[]), () => PrimitiveSerializer.CreateForArray<ulong>() },
{ typeof(float[]), () => PrimitiveSerializer.CreateForArray<float>() },
{ typeof(double[]), () => PrimitiveSerializer.CreateForArray<double>() },
{ typeof(string[]), () => PrimitiveSerializer.StringArray },

{ typeof(ColumnBatch<char>), () => PrimitiveSerializer.CreateForColumnBatch<char>() },
{ typeof(ColumnBatch<short>), () => PrimitiveSerializer.CreateForColumnBatch<short>() },
{ typeof(ColumnBatch<ushort>), () => PrimitiveSerializer.CreateForColumnBatch<ushort>() },
{ typeof(ColumnBatch<int>), () => PrimitiveSerializer.CreateForColumnBatch<int>() },
{ typeof(ColumnBatch<uint>), () => PrimitiveSerializer.CreateForColumnBatch<uint>() },
{ typeof(ColumnBatch<long>), () => PrimitiveSerializer.CreateForColumnBatch<long>() },
{ typeof(ColumnBatch<ulong>), () => PrimitiveSerializer.CreateForColumnBatch<ulong>() },
{ typeof(ColumnBatch<float>), () => PrimitiveSerializer.CreateForColumnBatch<float>() },
{ typeof(ColumnBatch<double>), () => PrimitiveSerializer.CreateForColumnBatch<double>() },
{ typeof(ColumnBatch<string>), () => PrimitiveSerializer.StringColumnBatch },

{ typeof(CharArrayWrapper), () => PrimitiveSerializer.CharArray },
};
private static readonly ConcurrentDictionary<Type, Func<ObjectSerializerBase>> RuntimeTypeToSerializer =
new ConcurrentDictionary<Type, Func<ObjectSerializerBase>>();

private readonly SerializerSettings settings;
private readonly HashSet<Type> knownTypes;
private readonly Dictionary<Type, ObjectSerializerBase> seenTypes = new Dictionary<Type, ObjectSerializerBase>();

static ReflectionSchemaBuilder()
{
RuntimeTypeToSerializer[typeof(char)] = () => PrimitiveSerializer.Char;
RuntimeTypeToSerializer[typeof(byte)] = () => PrimitiveSerializer.Byte;
RuntimeTypeToSerializer[typeof(sbyte)] = () => PrimitiveSerializer.SByte;
RuntimeTypeToSerializer[typeof(short)] = () => PrimitiveSerializer.Short;
RuntimeTypeToSerializer[typeof(ushort)] = () => PrimitiveSerializer.UShort;
RuntimeTypeToSerializer[typeof(int)] = () => PrimitiveSerializer.Int;
RuntimeTypeToSerializer[typeof(uint)] = () => PrimitiveSerializer.UInt;
RuntimeTypeToSerializer[typeof(bool)] = () => PrimitiveSerializer.Boolean;
RuntimeTypeToSerializer[typeof(long)] = () => PrimitiveSerializer.Long;
RuntimeTypeToSerializer[typeof(ulong)] = () => PrimitiveSerializer.ULong;
RuntimeTypeToSerializer[typeof(float)] = () => PrimitiveSerializer.Float;
RuntimeTypeToSerializer[typeof(double)] = () => PrimitiveSerializer.Double;
RuntimeTypeToSerializer[typeof(decimal)] = () => PrimitiveSerializer.Decimal;
RuntimeTypeToSerializer[typeof(string)] = () => PrimitiveSerializer.String;
RuntimeTypeToSerializer[typeof(Uri)] = () => PrimitiveSerializer.Uri;
RuntimeTypeToSerializer[typeof(TimeSpan)] = () => PrimitiveSerializer.TimeSpan;
RuntimeTypeToSerializer[typeof(DateTime)] = () => PrimitiveSerializer.DateTime;
RuntimeTypeToSerializer[typeof(DateTimeOffset)] = () => PrimitiveSerializer.DateTimeOffset;
RuntimeTypeToSerializer[typeof(Guid)] = () => PrimitiveSerializer.Guid;

RuntimeTypeToSerializer[typeof(char[])] = () => PrimitiveSerializer.CreateForArray<char>();
RuntimeTypeToSerializer[typeof(byte[])] = () => PrimitiveSerializer.ByteArray;
RuntimeTypeToSerializer[typeof(short[])] = () => PrimitiveSerializer.CreateForArray<short>();
RuntimeTypeToSerializer[typeof(ushort[])] = () => PrimitiveSerializer.CreateForArray<ushort>();
RuntimeTypeToSerializer[typeof(int[])] = () => PrimitiveSerializer.CreateForArray<int>();
RuntimeTypeToSerializer[typeof(uint[])] = () => PrimitiveSerializer.CreateForArray<uint>();
RuntimeTypeToSerializer[typeof(long[])] = () => PrimitiveSerializer.CreateForArray<long>();
RuntimeTypeToSerializer[typeof(ulong[])] = () => PrimitiveSerializer.CreateForArray<ulong>();
RuntimeTypeToSerializer[typeof(float[])] = () => PrimitiveSerializer.CreateForArray<float>();
RuntimeTypeToSerializer[typeof(double[])] = () => PrimitiveSerializer.CreateForArray<double>();
RuntimeTypeToSerializer[typeof(string[])] = () => PrimitiveSerializer.StringArray;

RuntimeTypeToSerializer[typeof(ColumnBatch<char>)] = () => PrimitiveSerializer.CreateForColumnBatch<char>();
RuntimeTypeToSerializer[typeof(ColumnBatch<short>)] = () => PrimitiveSerializer.CreateForColumnBatch<short>();
RuntimeTypeToSerializer[typeof(ColumnBatch<ushort>)] = () => PrimitiveSerializer.CreateForColumnBatch<ushort>();
RuntimeTypeToSerializer[typeof(ColumnBatch<int>)] = () => PrimitiveSerializer.CreateForColumnBatch<int>();
RuntimeTypeToSerializer[typeof(ColumnBatch<uint>)] = () => PrimitiveSerializer.CreateForColumnBatch<uint>();
RuntimeTypeToSerializer[typeof(ColumnBatch<long>)] = () => PrimitiveSerializer.CreateForColumnBatch<long>();
RuntimeTypeToSerializer[typeof(ColumnBatch<ulong>)] = () => PrimitiveSerializer.CreateForColumnBatch<ulong>();
RuntimeTypeToSerializer[typeof(ColumnBatch<float>)] = () => PrimitiveSerializer.CreateForColumnBatch<float>();
RuntimeTypeToSerializer[typeof(ColumnBatch<double>)] = () => PrimitiveSerializer.CreateForColumnBatch<double>();
RuntimeTypeToSerializer[typeof(ColumnBatch<string>)] = () => PrimitiveSerializer.StringColumnBatch;

RuntimeTypeToSerializer[typeof(CharArrayWrapper)] = () => PrimitiveSerializer.CharArray;
}

public ReflectionSchemaBuilder(SerializerSettings settings)
{
this.settings = settings ?? throw new ArgumentNullException(nameof(settings));
Expand Down
8 changes: 4 additions & 4 deletions Sources/Core/Microsoft.StreamProcessing/Serializer/StreamSerializer.cs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@
// Licensed under the MIT License
// *********************************************************************
using System;
using System.Collections.Generic;
using System.Collections.Concurrent;
using Microsoft.StreamProcessing.Serializer.Serializers;

namespace Microsoft.StreamProcessing.Serializer
Expand All @@ -13,8 +13,8 @@ namespace Microsoft.StreamProcessing.Serializer
/// </summary>
public static class StreamSerializer
{
private static readonly Dictionary<Tuple<Type, SerializerSettings>, object> TypedSerializers
= new Dictionary<Tuple<Type, SerializerSettings>, object>();
private static readonly ConcurrentDictionary<Tuple<Type, SerializerSettings>, object> TypedSerializers
= new ConcurrentDictionary<Tuple<Type, SerializerSettings>, object>();

/// <summary>
/// Create instance of serializer for given object type
Expand Down Expand Up @@ -48,7 +48,7 @@ public static StateSerializer<T> Create<T>(SerializerSettings settings)
var reader = new ReflectionSchemaBuilder(settings).BuildSchema(typeof(T));
var serializerTyped = new StateSerializer<T>(reader);

if (settings.UseCache) TypedSerializers.Add(key, serializerTyped);
if (settings.UseCache) TypedSerializers.TryAdd(key, serializerTyped);
return serializerTyped;
}
}
Expand Down
35 changes: 18 additions & 17 deletions Sources/Core/Microsoft.StreamProcessing/Utilities/ComparerExpression.cs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@
// *********************************************************************
using System;
using System.Collections;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Linq.Expressions;
Expand All @@ -26,25 +27,25 @@ public interface IComparerExpression<T>

internal sealed class ComparerExpressionCache
{
private static readonly Dictionary<Type, object> typeComparerCache = new Dictionary<Type, object>();
private static readonly ConcurrentDictionary<Type, object> typeComparerCache = new ConcurrentDictionary<Type, object>();

static ComparerExpressionCache()
{
typeComparerCache.Add(typeof(byte), new PrimitiveComparerExpression<byte>((x, y) => x < y ? -1 : x == y ? 0 : 1));
typeComparerCache.Add(typeof(sbyte), new PrimitiveComparerExpression<sbyte>((x, y) => x < y ? -1 : x == y ? 0 : 1));
typeComparerCache.Add(typeof(char), new PrimitiveComparerExpression<char>((x, y) => x < y ? -1 : x == y ? 0 : 1));
typeComparerCache.Add(typeof(short), new PrimitiveComparerExpression<short>((x, y) => x < y ? -1 : x == y ? 0 : 1));
typeComparerCache.Add(typeof(ushort), new PrimitiveComparerExpression<ushort>((x, y) => x < y ? -1 : x == y ? 0 : 1));
typeComparerCache.Add(typeof(int), new PrimitiveComparerExpression<int>((x, y) => x < y ? -1 : x == y ? 0 : 1));
typeComparerCache.Add(typeof(uint), new PrimitiveComparerExpression<uint>((x, y) => x < y ? -1 : x == y ? 0 : 1));
typeComparerCache.Add(typeof(long), new PrimitiveComparerExpression<long>((x, y) => x < y ? -1 : x == y ? 0 : 1));
typeComparerCache.Add(typeof(ulong), new PrimitiveComparerExpression<ulong>((x, y) => x < y ? -1 : x == y ? 0 : 1));
typeComparerCache.Add(typeof(decimal), new PrimitiveComparerExpression<decimal>((x, y) => x < y ? -1 : x == y ? 0 : 1));
typeComparerCache.Add(typeof(string), new GenericComparableExpression<string>());
typeComparerCache.Add(typeof(TimeSpan), new GenericComparableExpression<TimeSpan>());
typeComparerCache.Add(typeof(DateTime), new GenericComparableExpression<DateTime>());
typeComparerCache.Add(typeof(DateTimeOffset), new GenericComparableExpression<DateTimeOffset>());
typeComparerCache.Add(typeof(Empty), new PrimitiveComparerExpression<Empty>((x, y) => 0));
typeComparerCache.TryAdd(typeof(byte), new PrimitiveComparerExpression<byte>((x, y) => x < y ? -1 : x == y ? 0 : 1));
typeComparerCache.TryAdd(typeof(sbyte), new PrimitiveComparerExpression<sbyte>((x, y) => x < y ? -1 : x == y ? 0 : 1));
typeComparerCache.TryAdd(typeof(char), new PrimitiveComparerExpression<char>((x, y) => x < y ? -1 : x == y ? 0 : 1));
typeComparerCache.TryAdd(typeof(short), new PrimitiveComparerExpression<short>((x, y) => x < y ? -1 : x == y ? 0 : 1));
typeComparerCache.TryAdd(typeof(ushort), new PrimitiveComparerExpression<ushort>((x, y) => x < y ? -1 : x == y ? 0 : 1));
typeComparerCache.TryAdd(typeof(int), new PrimitiveComparerExpression<int>((x, y) => x < y ? -1 : x == y ? 0 : 1));
typeComparerCache.TryAdd(typeof(uint), new PrimitiveComparerExpression<uint>((x, y) => x < y ? -1 : x == y ? 0 : 1));
typeComparerCache.TryAdd(typeof(long), new PrimitiveComparerExpression<long>((x, y) => x < y ? -1 : x == y ? 0 : 1));
typeComparerCache.TryAdd(typeof(ulong), new PrimitiveComparerExpression<ulong>((x, y) => x < y ? -1 : x == y ? 0 : 1));
typeComparerCache.TryAdd(typeof(decimal), new PrimitiveComparerExpression<decimal>((x, y) => x < y ? -1 : x == y ? 0 : 1));
typeComparerCache.TryAdd(typeof(string), new GenericComparableExpression<string>());
typeComparerCache.TryAdd(typeof(TimeSpan), new GenericComparableExpression<TimeSpan>());
typeComparerCache.TryAdd(typeof(DateTime), new GenericComparableExpression<DateTime>());
typeComparerCache.TryAdd(typeof(DateTimeOffset), new GenericComparableExpression<DateTimeOffset>());
typeComparerCache.TryAdd(typeof(Empty), new PrimitiveComparerExpression<Empty>((x, y) => 0));
}

public static bool TryGetCachedComparer<T>(out IComparerExpression<T> comparer)
Expand All @@ -59,7 +60,7 @@ public static bool TryGetCachedComparer<T>(out IComparerExpression<T> comparer)
return false;
}

public static void Add<T>(IComparerExpression<T> comparer) => typeComparerCache.Add(typeof(T), comparer);
public static void Add<T>(IComparerExpression<T> comparer) => typeComparerCache.TryAdd(typeof(T), comparer);
}

internal class ComparerExpression<T> : IComparerExpression<T>
Expand Down
Loading

0 comments on commit 2039f16

Please sign in to comment.