<commit_before>/*
 * Copyright 2019 Graz University of Technology
 *
 * 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
 *
 *   http://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 org.tugraz.sysds.runtime.lineage;

import org.tugraz.sysds.api.DMLScript;
import org.tugraz.sysds.hops.OptimizerUtils;
import org.tugraz.sysds.hops.cost.CostEstimatorStaticRuntime;
import org.tugraz.sysds.lops.MMTSJ.MMTSJType;
import org.tugraz.sysds.runtime.DMLRuntimeException;
import org.tugraz.sysds.runtime.controlprogram.caching.MatrixObject;
import org.tugraz.sysds.runtime.controlprogram.context.ExecutionContext;
import org.tugraz.sysds.runtime.instructions.CPInstructionParser;
import org.tugraz.sysds.runtime.instructions.Instruction;
import org.tugraz.sysds.runtime.instructions.cp.CPInstruction.CPType;
import org.tugraz.sysds.runtime.instructions.cp.ComputationCPInstruction;
import org.tugraz.sysds.runtime.instructions.cp.MMTSJCPInstruction;
import org.tugraz.sysds.runtime.lineage.LineageCacheConfig.ReuseCacheType;
import org.tugraz.sysds.runtime.matrix.data.MatrixBlock;
import org.tugraz.sysds.runtime.util.LocalFileUtils;

import java.io.IOException;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;

public class LineageCache {
	private static final Map<LineageItem, Entry> _cache = new HashMap<>();
	private static final Map<LineageItem, SpilledItem> _spillList = new HashMap<>();
	private static final HashSet<LineageItem> _removelist = new HashSet<>();
	private static final long CACHELIMIT= (long)512*1024*1024; // 500MB
	private static String outdir = null;
	private static long _cachesize = 0;
	private static Entry _head = null;
	private static Entry _end = null;

	//--------------------- CACHE LOGIC METHODS ----------------------
	
	public static boolean reuse(Instruction inst, ExecutionContext ec) {
		if (ReuseCacheType.isNone())
			return false;
		
		boolean reuse = false;
		if (inst instanceof ComputationCPInstruction && LineageCache.isReusable(inst)) {
			LineageItem item = ((ComputationCPInstruction) inst).getLineageItems(ec)[0];
			
			synchronized( _cache ) {
				//try to reuse full or partial intermediates
				if (LineageCacheConfig.getCacheType().isFullReuse())
					reuse = fullReuse(item, (ComputationCPInstruction)inst, ec); 
				if (LineageCacheConfig.getCacheType().isPartialReuse())
					reuse |= LineageRewriteReuse.executeRewrites(inst, ec);

				//create a placeholder if no reuse to avoid redundancy
				//(e.g., concurrent threads that try to start the computation)
				if( ! reuse )
					putIntern(item, null, 0);
			}
		}
		
		return reuse;
	}
	
	//NOTE: safe to pin the object in memory as coming from CPInstruction
	
	public static void put(Instruction inst, ExecutionContext ec) {
		if (inst instanceof ComputationCPInstruction && isReusable(inst) ) {
			LineageItem item = ((LineageTraceable) inst).getLineageItems(ec)[0];
			MatrixObject mo = ec.getMatrixObject(((ComputationCPInstruction) inst).output);
			synchronized( _cache ) {
				putIntern(item, mo.acquireReadAndRelease(), getRecomputeEstimate(inst, ec));
			}
		}
	}
	
	public static void putValue(Instruction inst, ExecutionContext ec) {
		if (ReuseCacheType.isNone())
			return;
		if (inst instanceof ComputationCPInstruction && isReusable(inst) ) {
			LineageItem item = ((LineageTraceable) inst).getLineageItems(ec)[0];
			MatrixObject mo = ec.getMatrixObject(((ComputationCPInstruction) inst).output);
			MatrixBlock value = mo.acquireReadAndRelease();
			_cache.get(item).setValue(value, getRecomputeEstimate(inst, ec)); //outside sync to prevent deadlocks
			
			synchronized( _cache ) {
				if( !isBelowThreshold(value) ) 
					makeSpace(value);
				updateSize(value, true);
			}
		}
	}
	
	private static void putIntern(LineageItem key, MatrixBlock value, double compcost) {
		if (_cache.containsKey(key))
			//can come here if reuse_partial option is enabled
			return; 
			//throw new DMLRuntimeException("Redundant lineage caching detected: "+inst);
		
		// Create a new entry.
		Entry newItem = new Entry(key, value, compcost);
		
		// Make space by removing or spilling LRU entries.
		if( value != null ) {
			if( !isBelowThreshold(value) ) 
				makeSpace(value);
			updateSize(value, true);
		}
		
		// Place the entry at head position.
		setHead(newItem);
		
		_cache.put(key, newItem);
		if (DMLScript.STATISTICS)
			LineageCacheStatistics.incrementMemWrites();
	}
	
	protected static boolean probe(LineageItem key) {
		boolean p = (_cache.containsKey(key) || _spillList.containsKey(key));
		if (!p && DMLScript.STATISTICS && _removelist.contains(key))
			// The sought entry was in cache but removed later 
			LineageCacheStatistics.incrementDelHits();
		return p;
	}
	
	public static void resetCache() {
		_cache.clear();
		_spillList.clear();
		if (DMLScript.STATISTICS)
			_removelist.clear();
	}
	

	private static boolean fullReuse (LineageItem item, ComputationCPInstruction inst, ExecutionContext ec) {
		if (LineageCache.probe(item)) {
			MatrixBlock d = LineageCache.get(item);
			ec.setMatrixOutput(inst.output.getName(), d);
			return true;
		}
		return false;
	}
	
	protected static MatrixBlock get(LineageItem key) {
		// This method is called only when entry is present either in cache or in local FS.
		if (_cache.containsKey(key)) {
			// Read and put the entry at head.
			Entry e = _cache.get(key);
			delete(e);
			setHead(e);
			if (DMLScript.STATISTICS)
				LineageCacheStatistics.incrementMemHits();
			return e.getValue();
		}
		else
			return readFromLocalFS(key);
	}
	
	public static boolean isReusable (Instruction inst) {
		// TODO: Move this to the new class LineageCacheConfig and extend
		return inst.getOpcode().equalsIgnoreCase("tsmm")
				|| inst.getOpcode().equalsIgnoreCase("ba+*")
				|| inst.getOpcode().equalsIgnoreCase("*")
				|| inst.getOpcode().equalsIgnoreCase("rightIndex");
	}
	
	//---------------- CACHE SPACE MANAGEMENT METHODS -----------------
	
	private static boolean isBelowThreshold(MatrixBlock value) {
		//FIXME: we cannot throw such an exception here, instead need to gracefully ignore
		if (value.getInMemorySize() > CACHELIMIT)
			throw new RuntimeException("Single item larger than the size of lineage cache");
		return ((value.getInMemorySize() + _cachesize) <= CACHELIMIT);
	}
	
	private static void makeSpace(MatrixBlock value) {
		double valSize = value.getInMemorySize();<commit_msg>bugfix: only support matrix matrix elementwise multiplication for lineage cache.<commit_after>/*
 * Copyright 2019 Graz University of Technology
 *
 * 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
 *
 *   http://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 org.tugraz.sysds.runtime.lineage;

import org.tugraz.sysds.api.DMLScript;
import org.tugraz.sysds.hops.OptimizerUtils;
import org.tugraz.sysds.hops.cost.CostEstimatorStaticRuntime;
import org.tugraz.sysds.lops.MMTSJ.MMTSJType;
import org.tugraz.sysds.runtime.DMLRuntimeException;
import org.tugraz.sysds.runtime.controlprogram.caching.MatrixObject;
import org.tugraz.sysds.runtime.controlprogram.context.ExecutionContext;
import org.tugraz.sysds.runtime.instructions.CPInstructionParser;
import org.tugraz.sysds.runtime.instructions.Instruction;
import org.tugraz.sysds.runtime.instructions.cp.BinaryMatrixMatrixCPInstruction;
import org.tugraz.sysds.runtime.instructions.cp.CPInstruction.CPType;
import org.tugraz.sysds.runtime.instructions.cp.ComputationCPInstruction;
import org.tugraz.sysds.runtime.instructions.cp.MMTSJCPInstruction;
import org.tugraz.sysds.runtime.lineage.LineageCacheConfig.ReuseCacheType;
import org.tugraz.sysds.runtime.matrix.data.MatrixBlock;
import org.tugraz.sysds.runtime.util.LocalFileUtils;

import java.io.IOException;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;

public class LineageCache {
	private static final Map<LineageItem, Entry> _cache = new HashMap<>();
	private static final Map<LineageItem, SpilledItem> _spillList = new HashMap<>();
	private static final HashSet<LineageItem> _removelist = new HashSet<>();
	private static final long CACHELIMIT= (long)512*1024*1024; // 500MB
	private static String outdir = null;
	private static long _cachesize = 0;
	private static Entry _head = null;
	private static Entry _end = null;

	//--------------------- CACHE LOGIC METHODS ----------------------
	
	public static boolean reuse(Instruction inst, ExecutionContext ec) {
		if (ReuseCacheType.isNone())
			return false;
		
		boolean reuse = false;
		if (inst instanceof ComputationCPInstruction && LineageCache.isReusable(inst)) {
			LineageItem item = ((ComputationCPInstruction) inst).getLineageItems(ec)[0];
			
			synchronized( _cache ) {
				//try to reuse full or partial intermediates
				if (LineageCacheConfig.getCacheType().isFullReuse())
					reuse = fullReuse(item, (ComputationCPInstruction)inst, ec); 
				if (LineageCacheConfig.getCacheType().isPartialReuse())
					reuse |= LineageRewriteReuse.executeRewrites(inst, ec);

				//create a placeholder if no reuse to avoid redundancy
				//(e.g., concurrent threads that try to start the computation)
				if( ! reuse )
					putIntern(item, null, 0);
			}
		}
		
		return reuse;
	}
	
	//NOTE: safe to pin the object in memory as coming from CPInstruction
	
	public static void put(Instruction inst, ExecutionContext ec) {
		if (inst instanceof ComputationCPInstruction && isReusable(inst) ) {
			LineageItem item = ((LineageTraceable) inst).getLineageItems(ec)[0];
			MatrixObject mo = ec.getMatrixObject(((ComputationCPInstruction) inst).output);
			synchronized( _cache ) {
				putIntern(item, mo.acquireReadAndRelease(), getRecomputeEstimate(inst, ec));
			}
		}
	}
	
	public static void putValue(Instruction inst, ExecutionContext ec) {
		if (ReuseCacheType.isNone())
			return;
		if (inst instanceof ComputationCPInstruction && isReusable(inst) ) {
			LineageItem item = ((LineageTraceable) inst).getLineageItems(ec)[0];
			MatrixObject mo = ec.getMatrixObject(((ComputationCPInstruction) inst).output);
			MatrixBlock value = mo.acquireReadAndRelease();
			_cache.get(item).setValue(value, getRecomputeEstimate(inst, ec)); //outside sync to prevent deadlocks
			
			synchronized( _cache ) {
				if( !isBelowThreshold(value) ) 
					makeSpace(value);
				updateSize(value, true);
			}
		}
	}
	
	private static void putIntern(LineageItem key, MatrixBlock value, double compcost) {
		if (_cache.containsKey(key))
			//can come here if reuse_partial option is enabled
			return; 
			//throw new DMLRuntimeException("Redundant lineage caching detected: "+inst);
		
		// Create a new entry.
		Entry newItem = new Entry(key, value, compcost);
		
		// Make space by removing or spilling LRU entries.
		if( value != null ) {
			if( !isBelowThreshold(value) ) 
				makeSpace(value);
			updateSize(value, true);
		}
		
		// Place the entry at head position.
		setHead(newItem);
		
		_cache.put(key, newItem);
		if (DMLScript.STATISTICS)
			LineageCacheStatistics.incrementMemWrites();
	}
	
	protected static boolean probe(LineageItem key) {
		boolean p = (_cache.containsKey(key) || _spillList.containsKey(key));
		if (!p && DMLScript.STATISTICS && _removelist.contains(key))
			// The sought entry was in cache but removed later 
			LineageCacheStatistics.incrementDelHits();
		return p;
	}
	
	public static void resetCache() {
		_cache.clear();
		_spillList.clear();
		if (DMLScript.STATISTICS)
			_removelist.clear();
	}
	

	private static boolean fullReuse (LineageItem item, ComputationCPInstruction inst, ExecutionContext ec) {
		if (LineageCache.probe(item)) {
			MatrixBlock d = LineageCache.get(item);
			ec.setMatrixOutput(inst.output.getName(), d);
			return true;
		}
		return false;
	}
	
	protected static MatrixBlock get(LineageItem key) {
		// This method is called only when entry is present either in cache or in local FS.
		if (_cache.containsKey(key)) {
			// Read and put the entry at head.
			Entry e = _cache.get(key);
			delete(e);
			setHead(e);
			if (DMLScript.STATISTICS)
				LineageCacheStatistics.incrementMemHits();
			return e.getValue();
		}
		else
			return readFromLocalFS(key);
	}
	
	public static boolean isReusable (Instruction inst) {
		// TODO: Move this to the new class LineageCacheConfig and extend
		return inst.getOpcode().equalsIgnoreCase("tsmm")
				|| inst.getOpcode().equalsIgnoreCase("ba+*")
				|| (inst.getOpcode().equalsIgnoreCase("*") &&
					inst instanceof BinaryMatrixMatrixCPInstruction) //TODO support scalar
				|| inst.getOpcode().equalsIgnoreCase("rightIndex");
	}
	
	//---------------- CACHE SPACE MANAGEMENT METHODS -----------------
	
	private static boolean isBelowThreshold(MatrixBlock value) {
		//FIXME: we cannot throw such an exception here, instead need to gracefully ignore
		if (value.getInMemorySize() > CACHELIMIT)
			throw new RuntimeException("Single item larger than the size of lineage cache");
		return ((value.getInMemorySize() + _cachesize) <= CACHELIMIT);
	}
	
	private static void makeSpace(MatrixBlock value) {
		double valSize = value.getInMemorySize();
{'fraction_non_alphanumeric': 0.11236358414704194, 'fraction_numerical': 0.0038770821367030443, 'mean_word_length': 4.218808542525291, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 17, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '3139679', 'n_tokens_mistral': 4681, 'n_tokens_neox': 4187, 'n_words': 1138}