<commit_before>﻿using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Diagnostics.Contracts;
using System.Reflection;
using System.Threading;
using log4net;
using Metrolib;
using Tailviewer.BusinessLogic;
using Tailviewer.BusinessLogic.LogFiles;

namespace Tailviewer.Core.LogFiles
{
	/// <summary>
	///     Responsible for merging consecutive lines into multi-line log entries,
	///     if they belong together.
	/// </summary>
	/// <remarks>
	///     Two lines are defined to belong together if the first line contains a log
	///     level and the next one does not.
	/// </remarks>
	public sealed class MultiLineLogFile
		: AbstractLogFile
			, ILogFileListener
	{
		private static readonly ILog Log = LogManager.GetLogger(MethodBase.GetCurrentMethod().DeclaringType);

		private const int BatchSize = 10000;
		private readonly LogLine[] _buffer;
		private readonly List<LogEntryInfo> _indices;
		private readonly TimeSpan _maximumWaitTime;
		private readonly ConcurrentQueue<LogFileSection> _pendingModifications;
		private readonly ILogFile _source;
		private LogEntryInfo _currentLogEntry;
		private LogLineIndex _currentSourceIndex;
		private ErrorFlags _error;
		private Size _fileSize;

		private LogFileSection _fullSourceSection;
		private DateTime _lastModified;
		private int _maxCharactersPerLine;
		private DateTime? _startTimestamp;
		private LevelFlags _currentLogEntryLevel;

		private DateTime _created;
		//private readonly List<LogFileSection> _allModifications;

		public MultiLineLogFile(ITaskScheduler taskScheduler, ILogFile source, TimeSpan maximumWaitTime)
			: base(taskScheduler)
		{
			if (source == null)
				throw new ArgumentNullException(nameof(source));

			_maximumWaitTime = maximumWaitTime;
			_pendingModifications = new ConcurrentQueue<LogFileSection>();
			//_allModifications = new List<LogFileSection>();
			_indices = new List<LogEntryInfo>();
			_currentLogEntry = new LogEntryInfo(-1, 0);

			_buffer = new LogLine[BatchSize];

			_source = source;
			_source.AddListener(this, maximumWaitTime, BatchSize);
			StartTask();
		}

		/// <inheritdoc />
		public override int MaxCharactersPerLine => _maxCharactersPerLine;

		/// <inheritdoc />
		public override ErrorFlags Error => _error;

		/// <inheritdoc />
		public override DateTime? StartTimestamp => _startTimestamp;

		/// <inheritdoc />
		public override DateTime LastModified => _lastModified;

		/// <inheritdoc />
		public override DateTime Created => _created;

		/// <inheritdoc />
		public override Size Size => _fileSize;

		/// <inheritdoc />
		public override int Count => (int) _currentSourceIndex;

		/// <inheritdoc />
		public void OnLogFileModified(ILogFile logFile, LogFileSection section)
		{
			_pendingModifications.Enqueue(section);
			ResetEndOfSourceReached();
		}

		/// <inheritdoc />
		protected override void DisposeAdditional()
		{
			_source.RemoveListener(this);
		}

		/// <inheritdoc />
		public override int OriginalCount => _source.OriginalCount;

		/// <inheritdoc />
		public override void GetSection(LogFileSection section, LogLine[] dest)
		{
			_source.GetSection(section, dest);
			lock (_indices)
			{
				for (var i = 0; i < section.Count; ++i)
					dest[i] = PatchNoLock(dest[i]);
			}
		}

		/// <inheritdoc />
		public override LogLine GetLine(int index)
		{
			var actualLine = _source.GetLine(index);
			LogLine line;

			lock (_indices)
			{
				line = PatchNoLock(actualLine);
			}

			return line;
		}

		/// <inheritdoc />
		public override double Progress => 1;

		private LogLine PatchNoLock(LogLine line)
		{
			var info = _indices[line.LineIndex];

			LevelFlags level;
			DateTime? timestamp;
			if (line.LineIndex != info.FirstLineIndex)
			{
				// This line belongs to the previous line and together they form
				// (part of) a log entry. Even though only a single line mentions
				// the log level, all lines are given the same log level.
				var firstLine = _source.GetLine((int) info.FirstLineIndex);
				level = firstLine.Level;
				timestamp = firstLine.Timestamp;
			}
			else
			{
				level = line.Level;
				timestamp = line.Timestamp;
			}

			return new LogLine(line.LineIndex, info.EntryIndex,
				line.Message,
				level,
				timestamp);
		}

		/// <inheritdoc />
		protected override TimeSpan RunOnce(CancellationToken token)
		{
			var lastCount = _fullSourceSection.Count;
			bool performedWork = false;

			LogFileSection section;
			while (_pendingModifications.TryDequeue(out section) && !token.IsCancellationRequested)
			{
				if (section.IsReset)
				{
					Clear();
				}
				else if (section.IsInvalidate)
				{
					Invalidate(section);
				}
				else
				{
					_fullSourceSection = LogFileSection.MinimumBoundingLine(_fullSourceSection, section);
				}
				//_allModifications.Add(section);
				performedWork = true;
			}

			if (!_fullSourceSection.IsEndOfSection(_currentSourceIndex))
			{
				var remaining = Math.Min(_fullSourceSection.Count - _currentSourceIndex, BatchSize);
				_source.GetSection(new LogFileSection(_currentSourceIndex, remaining), _buffer);
				LogLineIndex? resetIndex = null;

				lock (_indices)
				{
					for (var i = 0; i < remaining; ++i)
					{
						var line = _buffer[i];
						if (_currentLogEntry.EntryIndex.IsInvalid ||
						    line.Level != LevelFlags.None ||
						    _currentLogEntryLevel == LevelFlags.None)
						{
							_currentLogEntry = _currentLogEntry.NextEntry(line.LineIndex);
							_currentLogEntryLevel = line.Level;
						}
						else if (_currentLogEntry.FirstLineIndex < lastCount && resetIndex == null)
						{
							var index = _currentLogEntry.FirstLineIndex;
							resetIndex = index;

							_currentLogEntryLevel = _source.GetLine((int) index).Level;
						}
						_indices.Add(_currentLogEntry);
					}
				}

				if (resetIndex != null)
				{
					var resetCount = lastCount - resetIndex.Value;
					if (resetCount > 0)
						Listeners.Invalidate((int) resetIndex.Value, resetCount);
				}

				_currentSourceIndex += remaining;
			}

			_maxCharactersPerLine = _source.MaxCharactersPerLine;
			_error = _source.Error;<commit_msg>MultiLineLogFile shouldn't allocate huge buffers upon construction<commit_after>﻿using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Diagnostics.Contracts;
using System.Reflection;
using System.Threading;
using log4net;
using Metrolib;
using Tailviewer.BusinessLogic;
using Tailviewer.BusinessLogic.LogFiles;

namespace Tailviewer.Core.LogFiles
{
	/// <summary>
	///     Responsible for merging consecutive lines into multi-line log entries,
	///     if they belong together.
	/// </summary>
	/// <remarks>
	///     Two lines are defined to belong together if the first line contains a log
	///     level and the next one does not.
	/// </remarks>
	public sealed class MultiLineLogFile
		: AbstractLogFile
			, ILogFileListener
	{
		private static readonly ILog Log = LogManager.GetLogger(MethodBase.GetCurrentMethod().DeclaringType);

		private const int MaximumBatchSize = 10000;

		private readonly List<LogEntryInfo> _indices;
		private readonly TimeSpan _maximumWaitTime;
		private readonly ConcurrentQueue<LogFileSection> _pendingModifications;
		private readonly ILogFile _source;
		private LogEntryInfo _currentLogEntry;
		private LogLineIndex _currentSourceIndex;
		private ErrorFlags _error;
		private Size _fileSize;

		private LogFileSection _fullSourceSection;
		private DateTime _lastModified;
		private int _maxCharactersPerLine;
		private DateTime? _startTimestamp;
		private LevelFlags _currentLogEntryLevel;

		private DateTime _created;
		//private readonly List<LogFileSection> _allModifications;

		public MultiLineLogFile(ITaskScheduler taskScheduler, ILogFile source, TimeSpan maximumWaitTime)
			: base(taskScheduler)
		{
			if (source == null)
				throw new ArgumentNullException(nameof(source));

			_maximumWaitTime = maximumWaitTime;
			_pendingModifications = new ConcurrentQueue<LogFileSection>();
			//_allModifications = new List<LogFileSection>();
			_indices = new List<LogEntryInfo>();
			_currentLogEntry = new LogEntryInfo(-1, 0);
			
			_source = source;
			_source.AddListener(this, maximumWaitTime, MaximumBatchSize);
			StartTask();
		}

		/// <inheritdoc />
		public override int MaxCharactersPerLine => _maxCharactersPerLine;

		/// <inheritdoc />
		public override ErrorFlags Error => _error;

		/// <inheritdoc />
		public override DateTime? StartTimestamp => _startTimestamp;

		/// <inheritdoc />
		public override DateTime LastModified => _lastModified;

		/// <inheritdoc />
		public override DateTime Created => _created;

		/// <inheritdoc />
		public override Size Size => _fileSize;

		/// <inheritdoc />
		public override int Count => (int) _currentSourceIndex;

		/// <inheritdoc />
		public void OnLogFileModified(ILogFile logFile, LogFileSection section)
		{
			_pendingModifications.Enqueue(section);
			ResetEndOfSourceReached();
		}

		/// <inheritdoc />
		protected override void DisposeAdditional()
		{
			_source.RemoveListener(this);
		}

		/// <inheritdoc />
		public override int OriginalCount => _source.OriginalCount;

		/// <inheritdoc />
		public override void GetSection(LogFileSection section, LogLine[] dest)
		{
			_source.GetSection(section, dest);
			lock (_indices)
			{
				for (var i = 0; i < section.Count; ++i)
					dest[i] = PatchNoLock(dest[i]);
			}
		}

		/// <inheritdoc />
		public override LogLine GetLine(int index)
		{
			var actualLine = _source.GetLine(index);
			LogLine line;

			lock (_indices)
			{
				line = PatchNoLock(actualLine);
			}

			return line;
		}

		/// <inheritdoc />
		public override double Progress => 1;

		private LogLine PatchNoLock(LogLine line)
		{
			var info = _indices[line.LineIndex];

			LevelFlags level;
			DateTime? timestamp;
			if (line.LineIndex != info.FirstLineIndex)
			{
				// This line belongs to the previous line and together they form
				// (part of) a log entry. Even though only a single line mentions
				// the log level, all lines are given the same log level.
				var firstLine = _source.GetLine((int) info.FirstLineIndex);
				level = firstLine.Level;
				timestamp = firstLine.Timestamp;
			}
			else
			{
				level = line.Level;
				timestamp = line.Timestamp;
			}

			return new LogLine(line.LineIndex, info.EntryIndex,
				line.Message,
				level,
				timestamp);
		}

		/// <inheritdoc />
		protected override TimeSpan RunOnce(CancellationToken token)
		{
			var lastCount = _fullSourceSection.Count;
			bool performedWork = false;

			LogFileSection section;
			while (_pendingModifications.TryDequeue(out section) && !token.IsCancellationRequested)
			{
				if (section.IsReset)
				{
					Clear();
				}
				else if (section.IsInvalidate)
				{
					Invalidate(section);
				}
				else
				{
					_fullSourceSection = LogFileSection.MinimumBoundingLine(_fullSourceSection, section);
				}
				//_allModifications.Add(section);
				performedWork = true;
			}

			if (!_fullSourceSection.IsEndOfSection(_currentSourceIndex))
			{
				var remaining = Math.Min(_fullSourceSection.Count - _currentSourceIndex, MaximumBatchSize);
				var buffer = new LogLine[remaining];
				_source.GetSection(new LogFileSection(_currentSourceIndex, remaining), buffer);
				LogLineIndex? resetIndex = null;

				lock (_indices)
				{
					for (var i = 0; i < remaining; ++i)
					{
						var line = buffer[i];
						if (_currentLogEntry.EntryIndex.IsInvalid ||
						    line.Level != LevelFlags.None ||
						    _currentLogEntryLevel == LevelFlags.None)
						{
							_currentLogEntry = _currentLogEntry.NextEntry(line.LineIndex);
							_currentLogEntryLevel = line.Level;
						}
						else if (_currentLogEntry.FirstLineIndex < lastCount && resetIndex == null)
						{
							var index = _currentLogEntry.FirstLineIndex;
							resetIndex = index;

							_currentLogEntryLevel = _source.GetLine((int) index).Level;
						}
						_indices.Add(_currentLogEntry);
					}
				}

				if (resetIndex != null)
				{
					var resetCount = lastCount - resetIndex.Value;
					if (resetCount > 0)
						Listeners.Invalidate((int) resetIndex.Value, resetCount);
				}

				_currentSourceIndex += remaining;
			}

			_maxCharactersPerLine = _source.MaxCharactersPerLine;
			_error = _source.Error;
{'fraction_non_alphanumeric': 0.09704950012192148, 'fraction_numerical': 0.001950743721043648, 'mean_word_length': 3.467683369644154, 'pattern_counts': {'":': 0, '<': 57, '<?xml version=': 0, '>': 71, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3215483', 'n_tokens_mistral': 4422, 'n_tokens_neox': 3554, 'n_words': 896}