path: root/dozentenmodulserver/src/main/java/org/openslx/bwlp/sat/maintenance/ImageValidCheck.java

package org.openslx.bwlp.sat.maintenance;

import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.security.NoSuchAlgorithmException;
import java.sql.SQLException;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

import org.apache.log4j.Logger;
import org.openslx.bwlp.sat.database.mappers.DbImage;
import org.openslx.bwlp.sat.database.mappers.DbImageBlock;
import org.openslx.bwlp.sat.database.models.ImageVersionMeta;
import org.openslx.bwlp.sat.database.models.LocalImageVersion;
import org.openslx.bwlp.sat.util.FileSystem;
import org.openslx.bwlp.thrift.iface.TNotFoundException;
import org.openslx.filetransfer.util.ChunkStatus;
import org.openslx.filetransfer.util.FileChunk;
import org.openslx.filetransfer.util.HashChecker;
import org.openslx.filetransfer.util.HashChecker.HashCheckCallback;
import org.openslx.filetransfer.util.HashChecker.HashResult;
import org.openslx.filetransfer.util.StandaloneFileChunk;
import org.openslx.util.ThriftUtil;
import org.openslx.util.TimeoutHashMap;
import org.openslx.util.Util;

public class ImageValidCheck implements Runnable {

	public enum CheckResult {
		NULL_POINTER_EXCEPTION,
		ALREADY_IN_PROGRESS,
		TOO_MANY_QUEUED_JOBS,
		SUBMITTED,
		REJECTED_BY_SCHEDULER,
		WORKING,
		DONE,
		FILE_NOT_FOUND,
		FILE_ACCESS_ERROR,
		OTHER_ERROR,
	}

	private static final Logger LOGGER = Logger.getLogger(ImageValidCheck.class);
	
	private static final int MAX_CONCURRENT_CHECKS = 1;

	private static Queue<ImageValidCheck> queue = new LinkedList<>();
	private static Map<String, ImageValidCheck> inProgress = new HashMap<>();
	private static TimeoutHashMap<String, ImageValidCheck> done = new TimeoutHashMap<>(
			TimeUnit.MINUTES.toMillis(60));

	private final String versionId;
	private final boolean integrity;

	// TODO: Set appropriately in various places; make it possible to query from RPC
	private CheckResult result = CheckResult.DONE;

	// Hash checking

	private static final HashChecker hashChecker;

	static {
		long maxMem = Runtime.getRuntime().maxMemory() / (1024 * 1024);
		int hashQueueLen;
		if (maxMem < 1200) {
			hashQueueLen = 1;
		} else {
			hashQueueLen = 2;
		}
		HashChecker hc;
		try {
			hc = new HashChecker("SHA-1", hashQueueLen);
		} catch (NoSuchAlgorithmException e) {
			hc = null;
		}
		hashChecker = hc;
	}

	// End hash checking

	public static CheckResult check(String versionId, boolean integrity) {
		if (versionId == null)
			return CheckResult.NULL_POINTER_EXCEPTION;
		synchronized (inProgress) {
			if (inProgress.containsKey(versionId))
				return CheckResult.ALREADY_IN_PROGRESS;
			if (inProgress.size() >= MAX_CONCURRENT_CHECKS) {
				if (queue.size() > 1000) {
					return CheckResult.TOO_MANY_QUEUED_JOBS;
				}
				queue.add(new ImageValidCheck(versionId, integrity));
				return CheckResult.SUBMITTED;
			}
			ImageValidCheck check = new ImageValidCheck(versionId, integrity);
			if (Maintenance.trySubmit(check)) {
				inProgress.put(versionId, check);
				return CheckResult.SUBMITTED;
			}
		}
		return CheckResult.REJECTED_BY_SCHEDULER;
	}

	public static void checkForWork() {
		synchronized (inProgress) {
			while (inProgress.size() < MAX_CONCURRENT_CHECKS && !queue.isEmpty()) {
				ImageValidCheck check = queue.poll();
				if (check == null)
					break;
				if (inProgress.containsKey(check.versionId))
					continue; // Already checking this version, try next in queue
				if (Maintenance.trySubmit(check)) {
					inProgress.put(check.versionId, check);
				} else {
					if (!queue.offer(check)) {
						LOGGER.warn("Dropped queued check for image version " + check.versionId);
					}
					// Scheduler didn't accept job - don't try remaining queue
					break;
				}
			}
		}
	}

	private ImageValidCheck(String versionId, boolean integrity) {
		this.versionId = versionId;
		this.integrity = integrity;
	}

	@Override
	public void run() {
		try {
			if (!FileSystem.waitForStorage()) {
				LOGGER.warn("Will not check " + versionId + ": Storage not online");
				return;
			}
			LocalImageVersion imageVersion;
			try {
				imageVersion = DbImage.getLocalImageData(versionId);
			} catch (SQLException e) {
				return;
			} catch (TNotFoundException e) {
				LOGGER.warn("Cannot check validity of image version - not found: " + versionId);
				return;
			}
			boolean valid = checkValid(imageVersion);
			if (valid && integrity) {
				try {
					valid = checkBlockHashes(imageVersion);
				} catch (IOException e) {
					result = CheckResult.FILE_ACCESS_ERROR;
					valid = false;
				} catch (Exception e) {
					result = CheckResult.OTHER_ERROR;
				}
			}
			if (imageVersion.isValid == valid)
				return; // nothing changed
			// Update
			try {
				DbImage.markValid(valid, false, imageVersion);
			} catch (SQLException e) {
			}
		} finally {
			synchronized (inProgress) {
				inProgress.remove(this.versionId);
			}
			checkForWork();
		}
	}

	/**
	 * Do a complete hash check of the given image file.
	 */
	private boolean checkBlockHashes(final LocalImageVersion imageVersion) throws IOException,
			InterruptedException {
		ImageVersionMeta versionDetails;
		try {
			versionDetails = DbImage.getVersionDetails(versionId);
		} catch (TNotFoundException e) {
			LOGGER.warn("Cannot check hash of image version - not found: " + versionId);
			return false;
		} catch (SQLException e) {
			return false;
		}
		// TODO
		if (versionDetails.sha1sums == null || versionDetails.sha1sums.isEmpty()) {
			LOGGER.info("Image does not have block hashes -- assuming ok");
			return true;
		}
		int numChecked = 0;
		final Semaphore sem = new Semaphore(0);
		final AtomicBoolean fileOk = new AtomicBoolean(true);
		File path = FileSystem.composeAbsoluteImagePath(imageVersion);
		try (RandomAccessFile raf = new RandomAccessFile(path, "r")) {
			long startOffset = 0;
			for (ByteBuffer hash : versionDetails.sha1sums) {
				if (hash == null) {
					startOffset += FileChunk.CHUNK_SIZE;
					continue;
				}
				long endOffset = startOffset + FileChunk.CHUNK_SIZE;
				if (endOffset > imageVersion.fileSize) {
					endOffset = imageVersion.fileSize;
				}
				StandaloneFileChunk chunk = new StandaloneFileChunk(startOffset, endOffset,
						ThriftUtil.unwrapByteBuffer(hash));
				byte[] buffer = new byte[(int) (endOffset - startOffset)];
				raf.seek(startOffset);
				raf.readFully(buffer);
				hashChecker.queue(chunk, buffer, new HashCheckCallback() {
					@Override
					public void hashCheckDone(HashResult result, byte[] data, FileChunk chunk) {
						if (result == HashResult.FAILURE) {
							// Hashing failed, cannot tell whether OK or not :(
						} else {
							if (result == HashResult.INVALID) {
								fileOk.set(false);
								((StandaloneFileChunk) chunk).overrideStatus(ChunkStatus.MISSING);
							} else {
								// >:(
								((StandaloneFileChunk) chunk).overrideStatus(ChunkStatus.COMPLETE);
							}
							try {
								// We don't know what the state was in DB before, so just fire updates
								DbImageBlock.asyncUpdate(imageVersion.imageVersionId, chunk);
							} catch (InterruptedException e) {
								Thread.currentThread().interrupt();
							}
						}
						sem.release();
					}
				}, true);
				numChecked += 1;
				startOffset += FileChunk.CHUNK_SIZE;
			}
		} catch (InterruptedException e) {
			Thread.currentThread().interrupt();
			throw e;
		}
		// Wait until the last callback fired
		sem.acquire(numChecked);
		return fileOk.get();
	}

	/**
	 * "Inexpensive" validity checks. File exists, readable, size ok, etc.
	 */
	private boolean checkValid(LocalImageVersion imageVersion) {
		if (imageVersion == null)
			return false;
		if (imageVersion.expireTime < Util.unixTime()) {
			LOGGER.info(versionId + ": expired");
			return false;
		}
		if (imageVersion.filePath == null || imageVersion.filePath.isEmpty()) {
			LOGGER.info(versionId + ": DB does not contain a path");
			return false;
		}
		File path = FileSystem.composeAbsoluteImagePath(imageVersion);
		if (path == null) {
			LOGGER.info(versionId + ": path from DB is not valid");
			return false;
		}
		if (!path.exists()) {
			LOGGER.info(versionId + ": File does not exist (" + path.getAbsolutePath() + ")");
			return false;
		}
		if (!path.canRead()) {
			LOGGER.info(versionId + ": File exists but not readable (" + path.getAbsolutePath() + ")");
			return false;
		}
		if (path.length() != imageVersion.fileSize) {
			LOGGER.info(versionId + ": File exists but has wrong size (expected: " + imageVersion.fileSize
					+ ", found: " + path.length() + ")");
			return false;
		}
		return true;
	}

}