Grammar Composition System

/src/agree-sys/70-parse-control.cs

//#define HEAP_PROCESS
//#define HEAP_PRIVATE_STATIC
//#define HEAP_PRIVATE
//#define GLOBAL_ALLOC
#define VIRTUAL_ALLOC

using System;
using System.IO;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using System.Collections.Concurrent;
using System.Runtime.InteropServices;

using miew.Concurrency;
using miew.Debugging;
using miew.Tokenization;
using miew.Enumerable;

#pragma warning disable 0649

using Va = miew.Memory.Virtual;

namespace agree
{
	///////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
	/// <summary>
	/// ParseControl : each instance of this class coordinates all aspects of parsing a single sentence:
	/// - morphology phase (note: but not tokenization)
	/// - coverage and chart dependencies
	/// - construct parse chart
	/// - main parse/packing phase
	/// - derivations/unpacking phase
	/// - statistics management
	/// - task management, concurrency, system options
	/// </summary>
	///////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
	public sealed partial class ParseControl : Allocator, ISysObj, IDisposable
	{
		public ParseControl(Grammar g)
			: base(g.tm)
		{
			this.config = g.config;
			this.parser_config = config.parser;
			this.g = g;
			this.tm = g.tm;
			this.lexicon = g.lex;
			this.stats = new Stats(this);

			tf = new TaskFactory(
						CancellationToken.None,
						TaskCreationOptions.AttachedToParent,
						TaskContinuationOptions.AttachedToParent,
						TaskScheduler.Default);
		}

		public ParseControl(Grammar g, TokenSet ts_input)
			: this(g)
		{
			this.ts_input = ts_input;
		}

		/// <summary>
		/// Invariants
		/// </summary>
		public readonly Config config;

		public readonly Config.Parser parser_config;

		public readonly Grammar g;

		public readonly Lexicon lexicon;

		/// <summary>
		/// Pertaining to this parse
		/// </summary>

		public TokenSet ts_input;

		public String InputText { get { return ts_input.Source.Text; } }

		public IEnumerable<IParseObj> chart_tokens;

		public LexicalAnalysis la;

		public ParseChart chart;

		/// <summary>
		/// Statistics for this parse
		/// </summary>

		readonly public Stats stats;

		readonly Stopwatch timer = new Stopwatch();

		/// <summary>
		/// Task management
		/// </summary>

		readonly TaskFactory tf;

		int c_tasks = 1;

		int cancel = 0;

		public bool f_parse_phase_complete = false;

		public bool AnyPacking { get { return parser_config.packing != Config.Parser.PackingOpts.None; } }

		///////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
		/// <summary>
		/// Guarantee a parent task context so that nested tasks will be attached as child tasks.
		/// </summary>
		///////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
		Task<ParseControl> StartRootTask(Func<Task> root_func)
		{
			timer.Start();

			if (parser_config.c_tasks_max == -1)
			{
				root_func();
				AfterChart();
				return Tasks.FromResult(this);
			}

			//return Task.Factory.StartNew(
			//    () => root_func().Wait(),
			//    CancellationToken.None,
			//    TaskCreationOptions.None,
			//    TaskScheduler.Default)
			return root_func()
			.ContinueWith<ParseControl>(t =>
				{
					AggregateException aex = t.Exception;
					if (aex != null)
					{
						aex = aex.Flatten();
						var rgex = aex.InnerExceptions.OfType<ParseException>().ToArray();
						if (rgex.Length == 1)
							throw rgex[0];
						throw aex;
					}
					return AfterChart();
				}, TaskContinuationOptions.ExecuteSynchronously);
		}

		ParseControl AfterChart()
		{
			stats.Parsing.Chart.SetElapsed(timer.ElapsedMilliseconds);
			f_parse_phase_complete = true;
			chart.ReleaseChart();	// doesn't seem to help
			chart.ClearSubscribers();

			if (parser_config.packing == Config.Parser.PackingOpts.None)
				stats.Parsing.Unpacking.c_derivations = stats.Parsing.Chart.c_root_edges;

			return this;
		}

		public Task<ParseControl> Parse(TokenSet ts_input)
		{
			this.ts_input = ts_input;
			Task<ParseControl> tpc = StartRootTask(() =>
			{
				/// continue from the morphology job ... to the post morphology job
				/// this returns immediately, but the root task won't be signaled until all child tasks complete.
				return SynchronousContinuation(MorphologyJob(ts_input), PostMorphologyJob);
			});

			/// non-parse activities go here; they will not delay the delivery of the parse chart
			if (parser_config.f_autotune)
				tpc.ContinueWith(t => g.tm.Autotune());
			return tpc;
		}

		public bool TryEnterMultithread()
		{
			int ctm = parser_config.c_tasks_max;
			if (ctm < int.MaxValue)
			{
				if (!config.system.MultiThreading || c_tasks >= ctm)
					return false;
				Interlocked.Increment(ref c_tasks);
			}
			return true;
		}

		public void LeaveMultithread()
		{
			if (parser_config.c_tasks_max < int.MaxValue)
				Interlocked.Decrement(ref c_tasks);
		}

		public
#if NODE_INDEX
		new 
#endif
 void Dispose()
		{
			stats.Totals.ms_time = timer.ElapsedMilliseconds;
#if NODE_INDEX
			base.Dispose();
#endif
			stats.Disconnect();
		}

#if false
		public Task<Y> SynchronousContinuation<X, Y>(Task<X> t_prev, Func<X, Y> f)
		{
			if (cfg.c_tasks_max == -1)
				return Tasks.FromResult(f(t_prev.Result));

			return t_prev.ContinueWith<Y>(t =>
				{
					return f(t.Result);
				}, TaskContinuationOptions.ExecuteSynchronously);
		}
		public Task<X> SynchronousContinuation<X>(Task t_prev, Func<X> f)
		{
			if (cfg.c_tasks_max == -1)
				return Tasks.FromResult(f());

			return t_prev.ContinueWith<X>(t =>
			{
				return f();
			}, TaskContinuationOptions.ExecuteSynchronously);
		}
#endif

		public Task SynchronousContinuation(Task t_prev, Action a)
		{
			if (parser_config.c_tasks_max == -1)
			{
				a();
				return Tasks.CompletedTask;
			}
			else
				return t_prev.ContinueWith(t => a(), TaskContinuationOptions.ExecuteSynchronously);
		}

		static void _nop(Task[] tsk) { }

		public Task WhenAll(Action[] rga)
		{
			if (parser_config.c_tasks_max == int.MaxValue)
			{
				Task[] rgt = new Task[rga.Length];
				for (int i = 0; i < rga.Length; i++)
					rgt[i] = tf.StartNew(rga[i]);
				rga = null;
				return tf.ContinueWhenAll(rgt, _nop);
			}

			for (int i = 0; i < rga.Length; i++)
				rga[i]();
			return Tasks.CompletedTask;
		}

		[DebuggerHidden()]
		internal Task StartAttachedChildTask(Action a)
		{
			if (parser_config.c_tasks_max == int.MaxValue)
				return tf.StartNew(a);

			if (c_tasks >= parser_config.c_tasks_max)
			{
				a();
				return Tasks.CompletedTask;
			}
			Interlocked.Increment(ref c_tasks);
			return tf.StartNew(() =>
				{
					a();
					Interlocked.Decrement(ref c_tasks);
				});
		}

		[DebuggerHidden()]
		internal Task<X> StartAttachedChildTask<X>(Func<X> f)
		{
			if (parser_config.c_tasks_max == int.MaxValue)
				return tf.StartNew(f);

			if (c_tasks >= parser_config.c_tasks_max)
			{
				return Tasks.FromResult(f());
			}
			Interlocked.Increment(ref c_tasks);
			return tf.StartNew<X>(() =>
				{
					X x = f();
					Interlocked.Decrement(ref c_tasks);
					return x;
				});
		}

		[DebuggerHidden()]
		internal Task StartAttachedChildTask<X>(Action<X> ax, X x)
		{
			if (parser_config.c_tasks_max == int.MaxValue)
				return tf.StartNew(() => ax(x));

			if (c_tasks >= parser_config.c_tasks_max)
			{
				ax(x);
				return Tasks.CompletedTask;
			}
			Interlocked.Increment(ref c_tasks);
			return tf.StartNew(() =>
				{
					ax(x);
					Interlocked.Decrement(ref c_tasks);
				});
		}

		//internal void ParallelLoop<T>(IEnumerable<T> ie, Action<T> a)
		//{
		//    if (parser_config.c_tasks_max == int.MaxValue)
		//        Parallel.ForEach(ie, a);
		//    else
		//        foreach (T t in ie)
		//            a(t);
		//}

		public bool IsParseCancelled
		{
			get
			{
				if (cancel > 0)
					return true;
				if (timer.Elapsed.TotalSeconds <= parser_config.ItemTimeoutSeconds)
					return false;
				if (Interlocked.CompareExchange(ref cancel, 1, 0) == 0)
				{
					timer.Stop();
					double ms = timer.ElapsedMilliseconds / 1000.0;
					throw new ParseTimeoutException("({0:0.00} sec.)", ms);
				}
				return true;
			}
		}

		public string SysObjName
		{
			get { return String.Format("pc-{0:X}", this.GetHashCode()); }
		}

		public string SysObjDescription
		{
			get { return String.Format("{0}", ts_input.Source.Text); }
		}

		public miew.ReadOnly.IReadOnlyDictionary<string, ISysObj> SysObjChildren
		{
			get { return chart.SysObjChildren; }
		}

		public ISysObj SysObjParent
		{
			get { return g; }
		}
	};

	///////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
	/// <summary>
	/// 
	/// </summary>
	///////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
	public unsafe class Allocator
#if NODE_INDEX
		: IDisposable
#endif
	{
		[DllImport("kernel32.dll", SetLastError = true)]
		static extern IntPtr HeapCreate(uint flOptions, UIntPtr dwInitialSize, UIntPtr dwMaximumSize);

		[DllImport("kernel32.dll", SetLastError = false)]
		static extern IntPtr HeapAlloc(IntPtr hHeap, uint dwFlags, UIntPtr dwBytes);

		[DllImport("kernel32.dll", SetLastError = true)]
		static extern bool HeapDestroy(IntPtr hHeap);

		[DllImport("kernel32.dll", SetLastError = true)]
		static extern bool HeapFree(IntPtr hHeap, uint dwFlags, IntPtr lpMem);

		[DllImport("kernel32.dll")]
		static extern IntPtr GlobalAlloc(uint uFlags, UIntPtr dwBytes);

		[DllImport("kernel32.dll")]
		static extern IntPtr GlobalFree(IntPtr hMem);

		[DllImport("kernel32.dll")]
		static extern UInt32 GetLargePageMinimum();

#if HEAP_PRIVATE_STATIC
		static Allocator()
		{
		    heap = HeapCreate(0, new UIntPtr(1024 * 1024 * 20), UIntPtr.Zero);
		}
#endif

		///////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
		/// <summary>
		/// 
		/// </summary>
		///////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
		public bool UnifyCheck(Tfs tfs0, Tfs tfs1)
		{
			Config.Parser.UnifierType ut = tm.config.parser.checker;
			if (ut == Config.Parser.UnifierType.n_way)
				return NWayCheck.Check(tfs0, tfs1);
			else if (ut == Config.Parser.UnifierType.qd)
				return new UnificationQd(tm).Check(tfs0, tfs1);
			else
				return Unification.Check(tfs0, tfs1);
		}

		public Allocator(TypeMgr tm)
		{
			this.tm = tm;
#if NODE_INDEX
#if HEAP_PRIVATE
			uint flOptions = 0;
			if (!tm.config.system.MultiThreading)
				flOptions |= 1;
			heap = HeapCreate(flOptions, new UIntPtr(1024 * 1024 * 1), UIntPtr.Zero);
#elif VIRTUAL_ALLOC
			heap = Va.VirtualAlloc(IntPtr.Zero, new UIntPtr(CB_VA_HEAP), Va.AllocationType.RESERVE, Va.MemoryProtection.NOACCESS);

			if (heap.ToInt64() == 0)
				throw new Exception();
#else
			this.m_sl = new SpinLock(false);
			this.hglobals = new List<IntPtr>();
#endif
#endif
		}

		public TypeMgr tm;

#if NODE_INDEX
#if HEAP_PRIVATE_STATIC
		static
#endif
#if HEAP_PRIVATE_STATIC || HEAP_PRIVATE || VIRTUAL_ALLOC
		IntPtr heap;
#endif
#if HEAP_PROCESS || HEAP_PRIVATE_STATIC || GLOBAL_ALLOC
		List<IntPtr> hglobals;
		SpinLock m_sl;
#endif

#if VIRTUAL_ALLOC
		const long CB_VA_COMMIT_CHUNK = 0x8000;
		const long CB_VA_HEAP = (long)2 * (long)1024 * (long)1024 * (long)1024;
		long o_alloc = 0;
		long o_commit = 0;
#endif

		int cb_alloc = 0;
#endif

		public void* Alloc(int cb)
		{
#if NODE_INDEX
			if (cb <= 0)
				throw new Exception();
#if HEAP_PROCESS
			IntPtr ip = Marshal.AllocHGlobal(cb);
#elif GLOBAL_ALLOC
			IntPtr ip = GlobalAlloc(0, new UIntPtr((uint)cb));
#elif VIRTUAL_ALLOC
			long o_end = Interlocked.Add(ref o_alloc, cb);
			if (o_end >= CB_VA_HEAP)
				throw new Exception();

			long _tmp = o_commit;
			if (o_end >= _tmp)
			{
				long q = (((o_end - _tmp) + CB_VA_COMMIT_CHUNK) & ~(CB_VA_COMMIT_CHUNK - 1));
				Va.VirtualAlloc(new IntPtr(heap.ToInt64() + _tmp), new UIntPtr((uint)q), Va.AllocationType.COMMIT, Va.MemoryProtection.READWRITE);
				Interlocked.CompareExchange(ref o_commit, _tmp + q, _tmp);
			}
			IntPtr ip = new IntPtr(heap.ToInt64() + o_end - cb);
#else
			IntPtr ip = HeapAlloc(heap, 0, new UIntPtr((uint)cb));
#endif

#if HEAP_PROCESS || HEAP_PRIVATE_STATIC || GLOBAL_ALLOC
			bool entered = false;
			m_sl.Enter(ref entered);
			cb_alloc += cb;
			hglobals.Add(ip);
			m_sl.Exit(false);
#else
			cb_alloc += cb;
#endif
			return ip.ToPointer();
#else
			return null;
#endif
		}

#if NODE_INDEX
		public void Dispose()
		{
#if DEBUG
			if (this.GetType() == typeof(Allocator))
				throw new Exception();
#endif
#if HEAP_PRIVATE || VIRTUAL_ALLOC
			IntPtr _tmp = heap;
			if (_tmp != IntPtr.Zero && Interlocked.CompareExchange(ref heap, IntPtr.Zero, _tmp) == _tmp)
			{
#if VIRTUAL_ALLOC
				if (!Va.VirtualFree(_tmp, UIntPtr.Zero, 0x8000))
					throw new Exception();
#else
				HeapDestroy(_tmp);
#endif
				//Task.Factory.StartNew(() => HeapDestroy(_tmp));
			}
#else
			List<IntPtr> _tmp = hglobals;
			if (_tmp != null && Interlocked.CompareExchange(ref hglobals, null, _tmp) == _tmp)
			{
				Task.Factory.StartNew(() =>
				{
					for (int i = 0; i < _tmp.Count; i++)
#if HEAP_PROCESS
						Marshal.FreeHGlobal(_tmp[i]);
#elif GLOBAL_ALLOC
						GlobalFree(_tmp[i]);
#else
						HeapFree(heap, 0, _tmp[i]);
#endif
				});
			}
#endif
		}
#endif
	};

	///////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
	/// <summary>
	/// 
	/// </summary>
	///////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
	public class ParseException : Exception
	{
		public ParseException(String fmt, params Object[] args)
			: base(fmt == null ? String.Empty : String.Format(fmt, args))
		{
		}
		public ParseException()
			: this(null)
		{
		}
	};

	public class ParseTimeoutException : ParseException
	{
		public ParseTimeoutException()
		{
		}
		public ParseTimeoutException(String fmt, params Object[] args)
			: base(String.Format(fmt, args))
		{
		}
	};

	public class LexicalAnalysisException : ParseException
	{
		public LexicalAnalysisException(String fmt, params Object[] args)
			: base(fmt == null ? String.Empty : String.Format(fmt, args))
		{
		}
	};

	public class LexicalCoverageException : ParseException
	{
		public LexicalCoverageException(String fmt, params Object[] args)
			: base(fmt == null ? String.Empty : String.Format(fmt, args))
		{
		}
	};
}