Superlinear: An Investigation into Concurrent Speedup

#define UNSAFE
#define MOCK

#define LINEARX
#define STRIDE

using System;
using System.Diagnostics;
using System.Threading;
using System.Collections.Generic;
using System.Linq;
using System.IO;
using System.Runtime.Serialization.Formatters.Binary;

namespace Superlinear
{
	[Serializable]
	class Results
	{
		public Repeat Repeat = new Repeat();
		public RepeatDetails RepeatDetails = new RepeatDetails();
	}

	[Serializable]
	class Repeat : SortedList<int, Cache> { }

	[Serializable]
	class RepeatDetails : SortedList<int, Steps> { }

	[Serializable]
	class Cache : SortedList<int, Efficiencies>
	{
		public Cache() { }

		public Cache( SortedList<int, List<double>> effs )
		{
			foreach ( var eff in effs ) Add( eff.Key, new Efficiencies( eff.Value ) );
		}
	}

	[Serializable]
	class Steps : SortedList<int, Details> { }

	[Serializable]
	class Efficiencies : List<double>
	{
		public Efficiencies() { }
		public Efficiencies( List<double> list ) : base( list ) { }
	}

	[Serializable]
	class Details : List<Detail> { }

	[Serializable]
	class Detail
	{
		public TimeSpan Seq;
		public TimeSpan Par;

		public double Eff { get { return Seq.TotalSeconds / Par.TotalSeconds; } }

		public Detail( TimeSpan seq, TimeSpan par )
		{
			Seq = seq;
			Par = par;
		}
	}


	unsafe class Worker
	{
#if LINEAR
		const int REPEAT = 3;
#endif
#if STRIDE
		const int REPEAT = 10;
#endif

		static readonly int CORES = Environment.ProcessorCount;

		public static Results Work()
		{
			var results = new Results();

#if MOCK
			using ( var fs = File.OpenRead( "data.1.1.dat" ) )
			{
				var f = new BinaryFormatter();
				var o = f.Deserialize( fs );
				var effs = ( SortedList<int, List<double>> ) o;
				results.Repeat.Add( 1, new Cache( effs ) );
			}
			using ( var fs = File.OpenRead( "data.3.3.dat" ) )
			{
				var f = new BinaryFormatter();
				var o = f.Deserialize( fs );
				var effs = ( SortedList<int, List<double>> ) o;
				results.Repeat.Add( 3, new Cache( effs ) );
			}
			using ( var fs = File.OpenRead( "data.10.2.dat" ) )
			{
				var f = new BinaryFormatter();
				var o = f.Deserialize( fs );
				var effs = ( SortedList<int, List<double>> ) o;
				results.Repeat.Add( 10, new Cache( effs ) );
			}
			using ( var fs = File.OpenRead( "data.100.1.dat" ) )
			{
				var f = new BinaryFormatter();
				var o = f.Deserialize( fs );
				var effs = ( SortedList<int, List<double>> ) o;
				results.Repeat.Add( 100, new Cache( effs ) );
			}

			using ( var fs = File.OpenRead( "steps.24.10.1.dat" ) )
			{
				var f = new BinaryFormatter();
				var o = f.Deserialize( fs );
				var steps = ( Steps ) o;
				results.RepeatDetails.Add( 1, steps );
			}
#else

			var effs = new SortedList<int, List<double>>();

#if LINEAR

			var data = new SortedList<int, byte[]>();

			for ( int i = 0 ; i < 9 ; i++ )
			{
				Trace.WriteLine( "\nTEST: " + i );

				foreach ( var cache in new int[] { 1, 2, 3, 4, 5, 6, 9, 12, 15, 18, 21, 24, 25, 26, 27, 28, 29, 30, 36, 42 } )
				{
					if ( !effs.ContainsKey( cache ) ) effs[ cache ] = new List<double>();
					if ( !data.ContainsKey( cache ) ) data[ cache ] = new byte[ cache * 1024 * 1024 ];

					effs[ cache ].Add( Work( data[ cache ], 1 ) );
					effs[ cache ].Sort();
				}
			}

			using ( var fs = File.OpenWrite( "data.3.3.dat" ) )
			{
				var f = new BinaryFormatter();
				f.Serialize( fs, effs );
			}

			Trace.WriteLine( "\nNOSTRIDE:" );

			foreach ( var eff in effs ) Trace.WriteLine( String.Format( "Cache: {0,2:0}MB => efficiency: {1}", eff.Key,
				String.Join( " ", eff.Value.Select( e => e.ToString( "N2" ) ).ToArray() ) ) );

			results.Repeat.Add( REPEAT, new Cache( effs ) );
#endif
#if STRIDE
			var data = new byte[ 24 * 1024 * 1024 ];
			for ( int i = 0 ; i < data.Length ; i++ ) data[ i ] = 1;

			var steps = new Steps();

			for ( int i = 0 ; i < 9 ; i++ )
			{
				Trace.WriteLine( "\nTEST: " + i );

				for ( int step = 0 ; step < 21 ; step++ )
				{
					if ( !steps.ContainsKey( step ) ) steps[ step ] = new Details();

					steps[ step ].Add( Work( data, step ) );
				}
			}

			using ( var fs = File.OpenWrite( "steps.24.10.2.dat" ) )
			{
				var f = new BinaryFormatter();
				f.Serialize( fs, steps );
			}

			Trace.WriteLine( "\nSTRIDE:" );

			foreach ( var step in steps ) Trace.WriteLine( String.Format( "Stride: 2^{0:00} = {1,9:N0} => efficiency: {2}", step.Key, 1 << step.Key,
				String.Join( " ", step.Value.Select( s => s.Eff.ToString( "N2" ) ).ToArray() ) ) );

			results.RepeatDetails.Add( REPEAT, steps );

#endif

			Trace.WriteLine( null );

#endif

			return results;
		}

		public static Detail Work( byte[] data, int step )
		{
			Trace.WriteLine( "\nSuperlinear " +
#if DEBUG
 "DEBUG "
#else
 "RELEASE "
#endif
 +
#if UNSAFE
 "UNSAFE "
#else
 "SAFE "
#endif
 + "CORES=" + CORES + " "
 + String.Format( "CACHE={0,2:N0}MB ", data.Length / ( 1024 * 1024 ) ) +
#if LINEAR
 "LINEAR "
#endif
#if STRIDE
 String.Format( "STEP=2^{0:N0}={1:N0} ", step, 1 << step )
#endif
 + "REPEAT=" + REPEAT
 );

			var seq = WorkSequential( data, step );
			var par = WorkParallel( data, step );

			double efficiency = seq.TotalSeconds / par.TotalSeconds;

			Trace.WriteLine( String.Format( "Efficiency = {0:N}", efficiency ) );

			Trace.WriteLine( null );

			return new Detail( seq, par );
		}

		static void InitData( byte[] counters )
		{
			Array.Clear( counters, 0, counters.Length );
		}

		static unsafe TimeSpan WorkSequential( byte[] counters, int step )
		{
			InitData( counters );

			fixed ( byte* pCounters = counters )
			{
				var ts = DoWork( 0, counters.Length, counters, pCounters, step );

				Trace.WriteLine( "WorkSequential: " + ts );

				return ts;
			}
		}

		static readonly Thread[] _Threads = new Thread[ CORES ];
		static readonly ManualResetEvent _StartEvent = new ManualResetEvent( false );
		static readonly ManualResetEvent _FinishEvent = new ManualResetEvent( false );
		static byte[] _Counters = null;
		static byte* _pCounters = null;
		static int _Stride = 0;
		static TimeSpan _ts = TimeSpan.Zero;
		static readonly object _tsKey = new object();
		static int _Countdown = 0;
		static readonly object _countdownKey = new object();
		static int _Step = 0;

		static Worker()
		{
			for ( int i = 0 ; i < _Threads.Length ; i++ )
			{
				int n = i;
				_Threads[ i ] = new Thread( ThreadMain );
				_Threads[ i ].Priority = ThreadPriority.Highest;
				_Threads[ i ].IsBackground = true;
				_Threads[ i ].Start( n );
			}
		}

		static unsafe void ThreadMain( object o )
		{
			int n = ( int ) o;

			for ( ; ; )
			{
				_StartEvent.WaitOne();

				var ts = DoWork( _Stride * n, _Stride, _Counters, _pCounters, _Step );

				lock ( _tsKey ) _ts += ts;

				lock ( _countdownKey ) { _Countdown--; Monitor.Pulse( _countdownKey ); }

				_FinishEvent.WaitOne();
			}
		}

		static unsafe TimeSpan WorkParallel( byte[] counters, int step )
		{
			InitData( counters );
			_Counters = counters;

			fixed ( byte* pCounters = _Counters )
			{
				_Stride = _Counters.Length / _Threads.Length;
				_pCounters = pCounters;
				_ts = TimeSpan.Zero;
				_Countdown = _Threads.Length;
				_Step = step;

				_FinishEvent.Reset();
				_StartEvent.Set();

				lock ( _countdownKey ) while ( _Countdown > 0 ) Monitor.Wait( _countdownKey );

				_StartEvent.Reset();
				_FinishEvent.Set();

				Trace.WriteLine( "WorkParallel  : " + _ts );

				return _ts;
			}
		}

		static unsafe TimeSpan DoWork( int from, int len, byte[] counters, byte* pCounters, int step )
		{
#if UNSAFE
			byte* c = pCounters;
#else
			byte[] c = counters;
#endif
			var sw = Stopwatch.StartNew();

			var start = sw.Elapsed;

			for ( int n = 0 ; n < REPEAT ; n++ )
				for ( long i = 0 ; i < len ; i++ )
				{
#if STRIDE
					long prod = i << step;
					long mod = prod % len;
					long div = prod / len;
					long index = from + mod + div;
#else
					long index = from + i;
#endif

					Debug.Assert( index >= from && index < from + len );

					c[ index ]++;
				}

			var finish = sw.Elapsed;

			return ( finish - start );
		}
	}
}