/*
 * Copyright (c) 2014, 2015, Dynatrace and/or its affiliates. All rights reserved.
 * 
 * This file is part of the Lock Contention Tracing Subsystem for the HotSpot
 * Virtual Machine, which is developed at Christian Doppler Laboratory on
 * Monitoring and Evolution of Very-Large-Scale Software Systems. Please
 * contact us at <http://mevss.jku.at/> if you need additional information
 * or have any questions.
 *
 * This code is free software; you can redistribute it and/or modify it
 * under the terms of the GNU General Public License version 2 only, as
 * published by the Free Software Foundation.
 *
 * This code is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
 * version 2 for more details (a copy is included in the LICENSE file that
 * accompanied this code).
 *
 * You should have received a copy of the GNU General Public License version
 * 2 along with this work. If not, see <http://www.gnu.org/licenses/>.
 *
 */ 
package sun.evtracing.processing.statistics;

import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;

import sun.evtracing.parser.GlobalSequenceOrderedEvent;
import sun.evtracing.parser.ParkReturnCode;
import sun.evtracing.parser.metadata.JavaStack;
import sun.evtracing.parser.metadata.JavaThread;
import sun.evtracing.processing.Sanity;
import sun.evtracing.processing.statistics.aggregator.ContentionProcessor;
import sun.evtracing.processing.statistics.metadata.Contention;
import sun.evtracing.processing.statistics.metadata.Group;
import sun.evtracing.processing.statistics.metadata.JavaObjectStack;
import sun.evtracing.processing.statistics.metadata.ParkEventTuple;

public abstract class ParkBlockerAbstractQueuedSynchronizerAnalyser implements ParkBlockerAnalyser {
	// Defines how to handle time between unpark and parkEnd
	public enum BlameShiftingMode {
		ACCURATE,	// shift to unknown
		PREVIOUS,	// shift to previous lock holder
		NEXT,		// shift to next lock holder
	}

	@FunctionalInterface
	public interface PrevNextConsumer<T> {
		public void accept(T prev, T cur, T next);
	}

	private static final BlameShiftingMode blameShiftingMode = BlameShiftingMode.valueOf(System.getProperty("sun.evtracing.jucBlameShifitingMode", "NEXT"));
	private final ContentionProcessor cproc;
	private final Set<ParkEventTuple> parker = new HashSet<>();
	private long lastEnd;
	private long submittedDuration;

	public ParkBlockerAbstractQueuedSynchronizerAnalyser(ContentionProcessor cproc) {
		this.cproc = cproc;
	}

	@Override
	public void begin(ParkEventTuple tuple) {
		parker.add(tuple);
	}

	@Override
	public void end(ParkEventTuple tuple) {
		parker.remove(tuple);

		for (ParkEventTuple t : parker) {
			t.addOverlapping(tuple);
		}
	}

	@Override
	public void finish(ParkEventTuple tuple) {
		try {
			List<ParkEventTuple> overlapping = tuple.overlapping().stream()
				.filter(t -> !t.isInterrupted() && t.knownOwner())
				.sorted(new ParkEventTupleUnparkComparator())
				.collect(Collectors.toList());

			overlapping.add(tuple);

			lastEnd = tuple.begin().timestamp();
			submittedDuration = 0;
			iterateWithPrevNext(overlapping, (prev, cur, next) -> processOverlappingContention(tuple, prev, cur, next));

			assert tuple.end().timestamp() - tuple.begin().timestamp() == submittedDuration;
		} catch (AssertionError | RuntimeException e) {
			ParkEventTupleJsonSerializer.serialize("traceviewer_timestamp.json", tuple, false);
			ParkEventTupleJsonSerializer.serialize("traceviewer_sequence.json", tuple, true);
			throw e;
		} finally {
			lastEnd = 0;
			submittedDuration = 0;
		}
	}

	private <T> void iterateWithPrevNext(Iterable<T> iterable, PrevNextConsumer<T> consumer) {
		T prev;
		T cur = null;
		T next = null;
		for (T t : iterable) {
			prev = cur;
			cur = next;
			next = t;
			if (cur != null) {
				consumer.accept(prev, cur, next);
			}
		}
		if (next != null) {
			consumer.accept(cur, next, null);
		}
	}

	private void processOverlappingContention(ParkEventTuple tuple, ParkEventTuple prev, ParkEventTuple cur, ParkEventTuple next) {
		if (Sanity.isEnabled()) {
			assertContentionTransition(tuple, prev, cur, next);
		}

		boolean submitKnown;
		if (next != null) {
			submitKnown = cur.knownOwner() && tuple.begin().sequenceNumber() < cur.unpark().sequenceNumber();
		} else {
			assert cur == tuple;
			submitKnown = cur.knownOwner() && cur.causedContention();
			if (!submitKnown && cur.end().parkReturnCode() == ParkReturnCode.TimedOut) {
				// TODO: Special handling necessary for pushing remaining duration
			}
		}
		if (submitKnown) {
			JavaThread contendingThread = cur.end().thread().metadata();
			JavaThread ownerThread;
			JavaStack ownerSite;
			if (isShared(cur)) {
				ownerThread = sharedThread();
				ownerSite = sharedSite();
			} else {
				ownerThread = cur.unpark().thread().metadata();
				assert contendingThread.threadId() != ownerThread.threadId();
				ownerSite = cur.unpark().stack().metadata();
			}
			long unparkTime;
			if (blameShiftingMode == BlameShiftingMode.PREVIOUS) {
				unparkTime = cur.end().timestamp();
			} else {
				unparkTime = cur.unpark().timestamp();
			}
			long duration = unparkTime - lastEnd;

			submitContention(tuple, ownerThread, ownerSite, lastEnd, duration);

			lastEnd = unparkTime;
		}

		if (next == null || (blameShiftingMode != BlameShiftingMode.NEXT && isSequentialUnpark(cur, next))) {
			long endTime = cur.end().timestamp();
			long duration = endTime - lastEnd;

			if (duration != 0) {
				submitContention(tuple, JavaThread.UNKNOWN, JavaStack.UNKNOWN, lastEnd, duration);

				lastEnd = endTime;
			}
		}
	}

	abstract boolean isShared(ParkEventTuple cur);
	abstract JavaThread sharedThread();
	abstract JavaStack sharedSite();

	private boolean isSequentialUnpark(ParkEventTuple cur, ParkEventTuple next) {
		return cur.end().sequenceNumber() < next.end().unparkSequenceNumber();
	}

	private void assertContentionTransition(ParkEventTuple tuple, ParkEventTuple prev, ParkEventTuple cur, ParkEventTuple next) {
		if (next != null) {
			// cur.begin().sequenceNumber() < tuple.begin().sequenceNumber() is not always fulfilled
			// as a thread may have to park multiple times because the lock was sneaked by other threads
			assert tuple.begin().sequenceNumber() < cur.end().sequenceNumber();
			assert cur.end().sequenceNumber() < tuple.end().sequenceNumber();
		}

		if (prev != null) {
			boolean sequentialUnpark = cur.knownOwner() && prev.end().sequenceNumber() < cur.unpark().sequenceNumber();
			GlobalSequenceOrderedEvent prevLastEvent;
			if (sequentialUnpark) {
				prevLastEvent = prev.end();
			} else {
				prevLastEvent = prev.unpark();
			}
			// prev.begin().sequenceNumber() < cur.begin().sequenceNumber() is not always fulfilled
			// as the begin sequence number does not correspond to the enqueuing order
			assert prevLastEvent.sequenceNumber() < cur.end().sequenceNumber();
			if (next != null || sequentialUnpark) {
				assert prevLastEvent.sequenceNumber() < cur.unpark().sequenceNumber();
			}
			if (!cur.knownOwner()) {
				assert next == null;
				assert cur == tuple;
				// ignore
			} else if (!sequentialUnpark) {
				// unpark out of queue order -> last owner sneaked the lock
				// concurrent unpark -> multiple unpark/end overlap
				if (prev.end().sequenceNumber() < cur.end().sequenceNumber()) {
					if (!prev.causedContention() || !cur.causedContention()) {
						// unharmful concurrent unpark
					} else {
						Sanity.warn("inaccurate blame-shifting during concurrent unpark");
					}
				} else if (next != null || prev.unpark().sequenceNumber() < cur.unpark().sequenceNumber()) {
					//assert !prev.causedContention() || !cur.causedContention();
				}
			} else if (prev.begin().thread().identifier() == cur.unpark().thread().identifier()) {
				// unpark according to queue order -> last owner followed queue order
				// cur.begin().sequenceNumber() < prevLastEvent.sequenceNumber() is not always fulfilled
				// as the begin sequence number does not correspond to the enqueuing order
			} else if (prev.begin().thread().identifier() == cur.begin().thread().identifier()) {
				// unpark out of queue order -> last owner sneaked the lock
				// thread already parked again
				assert prevLastEvent.sequenceNumber() < cur.begin().sequenceNumber();
			} else if (!prev.causedContention()) {
				// unpark out of queue order -> last owner sneaked the lock
				// previous wakeup was spurious
			} else {
				// unpark out of queue order -> last owner sneaked the lock
				// special case:
				// * T1 owns lock, T2, T4 is next on the wait queue
				// * T1 unparks T2
				// * T2 returns from park call
				// * some thread interrupts T2
				// * T2 does not acquire lock and cancels its queue entry
				// * T3 sneaks the lock
				// * T3 unparks T4 as T2 is already canceled
				// * T2 tries to unpark a successor
			}
		}
	}

	private void submitContention(ParkEventTuple tuple, JavaThread ownerThread, JavaStack ownerSite, long startTime, long duration) {
		JavaThread contendingThread = tuple.end().thread().metadata();
		JavaStack contendingSite = tuple.begin().stack().metadata();
		JavaObjectStack blocker = tuple.blocker();

		cproc.submit(new Contention(Group.JavaUtilConcurrent, contendingThread, ownerThread, contendingSite, ownerSite, blocker, startTime, duration));
		submittedDuration += duration;
	}
}