phaniarnab commented on a change in pull request #1522:
URL: https://github.com/apache/systemds/pull/1522#discussion_r800676408
##########
File path:
src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederatedWorkerHandler.java
##########
@@ -246,74 +246,43 @@ private FederatedResponse readData(String filename,
DataType dataType,
// early throwing of exception to avoid infinitely
waiting threads for data
throw new FederatedWorkerHandlerException("Could not
recognize datatype");
- CacheableData<?> cd = _frc.get(filename);
- if(cd == null) {
+ ExecutionContext ec = ecm.get(tid);
+ LineageItem linItem = new LineageItem(filename);
+
+ CacheableData<?> cd = null;
+
+ if(!LineageCache.reuseRead(Long.toString(id), dataType,
linItem, ec)) {
+ if(ReuseCacheType.isNone() || dataType !=
DataType.MATRIX)
+ // using the read cache if lineage reuse is
disabled or the data is not a matrix
+ cd = _frc.get(filename);
try {
- switch(dataType) {
- case MATRIX:
- cd = new
MatrixObject(Types.ValueType.FP64, filename);
- break;
- case FRAME:
- cd = new FrameObject(filename);
- break;
- default:
- throw new
FederatedWorkerHandlerException("Could not recognize datatype");
- }
+ if(cd == null) { // data is neither in lineage
cache nor in read cache
+ long t0 = !ReuseCacheType.isNone() ?
System.nanoTime() : 0;
- FileFormat fmt = null;
- boolean header = false;
- String delim = null;
- FileSystem fs = null;
- MetaDataAll mtd;
-
- try {
- final String mtdName =
DataExpression.getMTDFileName(filename);
- Path path = new Path(mtdName);
- fs =
IOUtilFunctions.getFileSystem(mtdName);
- try(BufferedReader br = new
BufferedReader(new InputStreamReader(fs.open(path)))) {
- mtd = new MetaDataAll(br);
- if(!mtd.mtdExists())
- throw new
FederatedWorkerHandlerException("Could not parse metadata file");
- mc.setRows(mtd.getDim1());
- mc.setCols(mtd.getDim2());
- mc.setNonZeros(mtd.getNnz());
- header = mtd.getHasHeader();
- cd =
mtd.parseAndSetPrivacyConstraint(cd);
- fmt = mtd.getFileFormat();
- delim = mtd.getDelim();
- }
- }
- catch(DMLPrivacyException |
FederatedWorkerHandlerException ex) {
- throw ex;
- }
- catch(Exception ex) {
- String msg = "Exception of type " +
ex.getClass() + " thrown when processing READ request";
- LOG.error(msg, ex);
- throw new DMLRuntimeException(msg);
- }
- finally {
- IOUtilFunctions.closeSilently(fs);
- }
+ cd = readDataNoReuse(filename,
dataType, mc); // actual read of the data
+
+ long t1 = !ReuseCacheType.isNone() ?
System.nanoTime() : 0;
- // put meta data object in symbol table, read
on first operation
- cd.setMetaData(new MetaDataFormat(mc, fmt));
- if(fmt == FileFormat.CSV)
- cd.setFileFormatProperties(new
FileFormatPropertiesCSV(header, delim,
-
DataExpression.DEFAULT_DELIM_SPARSE));
- cd.enableCleanup(false); // guard against
deletion
+ if(!ReuseCacheType.isNone() && dataType
== DataType.MATRIX)
+ // put the object into the
lineage cache
+ LineageCache.putReadObject(cd,
linItem, ec, t1 - t0);
Review comment:
Note: As we lazily read the actual data, this `computetime `only records
the metadata read. A small computetime wrongly dictates the cache eviction
logic to remove this entry early in case the cache is full. The `putReadObject`
will trigger the actually read and place the MatrixBlock in the lineage cache.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
