/*******************************************************************************
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
*      https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*******************************************************************************/
package net.adoptopenjdk.casa.workload_caching;

import java.io.PrintStream;
import java.util.concurrent.ConcurrentLinkedQueue;

import net.adoptopenjdk.casa.util.Utilities;
import net.adoptopenjdk.casa.verbose_gc_parser.VerboseUtil;




/**
 * Encapsulates a buffer which is flushed to the given 
 * print stream at the prescribed delay. The buffer is a raw 
 * record of the number of completed transactions at a number of different 
 * points in time. The print thread assembles them into a 
 * coherent log and fills in the gaps.  
 * 
 *  
 */
public class ThroughputLoggerPrintThread extends Thread
{
	private ConcurrentLinkedQueue<LogRequest> requestQueue; 
	
	// Delay between printing the queue. 
	private double printDelay;
	
	// Stream to print queue out to. 
	private PrintStream destination;
	
	// Two sample windows are used to store all samples over two different time periods. 
	private SampleWindow sampleWindow1;
	private SampleWindow sampleWindow2;
		
	// The last transaction count sample taken, possibly older than lastElapsedTime  
	private long lastCount;
		
	private long lineNumber; 
	
	// The time the last line was written to the file. 
	private double lastTime;
	
	private double nextTime; 
		
	private double absoluteStartTime;
	private double startTime;   
	
	private double granularity;
	private double tolerance;  

	/**
	 * LogRequest objects are generated by the worker threads and 
	 * placed in a queue for later processing. They represent a count 
	 * of completed events at any given moment.
	 * 
	 * They are created in ThroughputLoger 
	 * 
	 * Their creation is initiated in CacheSimulation by a WorkerThread 
	 * 
	 *  
	 *
	 */
	public class LogRequest
	{		
		private double time;
		private long count; 
				
		public LogRequest(double time, long count) 
		{			 
			this.time = time; 
			this.count = count;  							
		}
		
		public double getTime()
		{
			return time; 
		}
		
		public long getCount()
		{
			return count; 
		}			
	}	
	
	/**
	 * Creates a new logger thread that flushes 
	 * the buffer after each printDelay. 
	 * 
	 * @param printDelay
	 * @param destination
	 */
	public ThroughputLoggerPrintThread(CacheSimulations simulations, PrintStream destination, double granularity, double tolernace, double sampleWindow1Time, double sampleWindow2Time)
	{		
		this.lastCount = 0; 
		this.lineNumber = 0;
		this.destination = destination;
		this.granularity = granularity; 
		this.tolerance = tolernace; 
		
		// Print roughly 1000 lines at a time.  
		this.printDelay = granularity * 100; 		
		
		// startTime will be changed when the first entry is formatted.  
		this.lastTime = this.startTime = Utilities.getTime();
		
		// absoluteStartTime will be changed when the first entry is formatted 
		this.absoluteStartTime = Utilities.getAbsoluteTime(); 
		
		// The time of the next entry will be the current (start) time plus the granularity
		this.nextTime = startTime + granularity; 
			
		// Init the request queue which will hold any logging requests issued.
		this.requestQueue = new ConcurrentLinkedQueue<LogRequest>(); 
				
		// Initialize the two sampling windows. 
		this.sampleWindow1 = new SampleWindow(sampleWindow1Time, granularity, Event.PARSE_ERROR);
		this.sampleWindow2 = new SampleWindow(sampleWindow2Time, granularity, Event.PARSE_ERROR);
		
		setDaemon(true); 
	}
	
	/**
	 * Appends a log entry to the given StringBuilder for the 
	 * given time and count. This has the side effect of advancing 
	 * both SampleWindows and incrementing the lineNumber. 
	 * 
	 * Before the first nonzero count, calls with zero counts are 
	 * ignored and will not be formatted into log entries. startTime and 
	 * absoluteStartTime will be set to the time of the first formatted 
	 * entry. 
	 * 
	 * @param time
	 * @param count
	 * @param builder
	 */
	private void formatLogEntry(double time, long count, StringBuilder builder)
	{
		/*
		 * If we haven't logged anything yet and the 
		 * count is still zero, hold off and shift the 
		 * log start time ahead.  
		 */
		if (lineNumber == 0 && count == 0)
		{
			/*
			 * Advance the absolute start time by the time between 
			 * its initialization (startTime) and this entry, effectively 
			 * moving it to the time of this entry. 
			 */
			absoluteStartTime += time - startTime;
			
			// Advance startTime to the time of this entry. 
			startTime = time; 
			
			return; 
		}

		// Average since the beginning of time. 
		double average = count / (time - startTime);
		
		// Compute differences in count and time since last formatted entry. 
		double deltaT = time > lastTime ? time - lastTime : 0; 
		double deltaC = count > lastCount ? count - lastCount : 0; 
		
		double momentaryAverage = deltaC / deltaT;			
		
		// Update the sample windows
		sampleWindow1.addSample(momentaryAverage);
		sampleWindow2.addSample(momentaryAverage);
			
		// Average each window 
		final double window1Average = sampleWindow1.getAverage();				
		final double window2Average = sampleWindow2.getAverage();					
				
		// Format the line based on the given time and count  		
		builder.append(VerboseUtil.formatTimestamp(nextTime - startTime, Utilities.roundDouble(absoluteStartTime, 3)));
		builder.append("\t"); 
		builder.append(String.format("%12.8g", Utilities.roundDouble(momentaryAverage, 6))); 
		builder.append("\t"); 
		builder.append(String.format("%12.8g", Utilities.roundDouble(window1Average, 6)));
		builder.append("\t"); 
		builder.append(String.format("%12.8g", Utilities.roundDouble(window2Average, 6)));
		builder.append("\t"); 
		builder.append(String.format("%12.8g", Utilities.roundDouble(average, 6)));
		builder.append("\n");
		
		// Advance the line number as we've just added a line 
		lineNumber++; 
	}
	
	/**
	 * Queues the log request to be printed at a later date. 
	 * 
	 * @param request
	 */
	public void add(LogRequest request)
	{
		requestQueue.offer(request);
	}
	
	/**
	 * Causes the thread to start monitoring the buffer
	 */
	public void start()
	{
		synchronized(this)
		{
			super.start();		
		}
	}
	
	/**
	 * Shuts down the logger thread, ensuring the 
	 * buffer has been flushed. 
	 * 
	 */
	public void shutdown() throws InterruptedException
	{	
		synchronized(this)
		{
			interrupt(); 
			join(); 
			
			requestQueue = null; 
		}
	}
	
	/**
	 * Interrupts the logger thread without 
	 * joining, not necessarily flushing the buffer. 
	 */
	public void kill()
	{
		synchronized(this)
		{
			interrupt();
		}
	}
	
	/**
	 * Flushes the contents of the buffer to the 
	 * print stream immediately.  
	 */
	public boolean flushBufferToLogFile() 
	{	
		if (requestQueue.isEmpty())
			return false; 
		
		StringBuilder builder = new StringBuilder(); 
		
		while (!requestQueue.isEmpty())
		{
			LogRequest request = requestQueue.poll();			
			
			double time = request.getTime();
			long count = request.getCount(); 
			
			if (count <= lastCount)
				continue;
			
			if (time <= lastTime)
				continue;			
						
			/*
			 * Catch up for missing requests. The most charitable interpretation 
			 * is that the VM was in STW (GC) and that the throughput was therefore 
			 * 0. Therefore the count will remain unchanged.  
			 */
			while (time > nextTime + tolerance)
			{	
				// Log an artificial entry with an unchanged count (delta = 0)  
				formatLogEntry(nextTime, lastCount, builder);
								
				// Advance the lastTime ahead to the artificial time we just logged for. 
				lastTime = nextTime;
				
				// Advance the expected nextTime by one x granularity 
				nextTime += granularity;	
			}
						
			// Print new request. 
			if (time <= nextTime + tolerance && time > nextTime - tolerance)
			{		
				// The time is within tolerance so print a new entry based on the count and time 
				formatLogEntry(time, count, builder);
				
				// Set the last time to the actual time of this count. 
				lastTime = time;
				
				// Set the last count to the actual time of this count. 
				lastCount = count;
				
				// Set the next expected time to the current time plus one interval. 
				nextTime += granularity;
			}			
		}	
	
		// Flush the builder to the destination stream. 
		destination.print(builder);
		
		return (builder.length() > 0); 		
	}
	
	public void run()
	{
		try 
		{
			// Flush the buffer every printDelay seconds. 
			while (!Thread.interrupted())
			{
				flushBufferToLogFile();  
				
				Utilities.idle(printDelay);				
			}
							
			while (flushBufferToLogFile());  
		}
		catch (InterruptedException e)
		{
			// While there's still stuff getting flushed, keep flushing. 
			while(flushBufferToLogFile()); 
		}
		catch (Throwable e)
		{
			return; 
		}
	}
}