diff --git a/modules/src/main/java/org/archive/modules/CoreAttributeConstants.java b/modules/src/main/java/org/archive/modules/CoreAttributeConstants.java index 4d9d4530..f0579b61 100644 --- a/modules/src/main/java/org/archive/modules/CoreAttributeConstants.java +++ b/modules/src/main/java/org/archive/modules/CoreAttributeConstants.java @@ -124,5 +124,6 @@ public interface CoreAttributeConstants { // arbitrary additions to WARC response record headers public static final String A_WARC_RESPONSE_HEADERS = "warc-response-headers"; - + + public static final String A_WARC_STATS = "warc-stats"; } diff --git a/modules/src/main/java/org/archive/modules/writer/WARCWriterProcessor.java b/modules/src/main/java/org/archive/modules/writer/WARCWriterProcessor.java index 84aaba37..12e4c637 100644 --- a/modules/src/main/java/org/archive/modules/writer/WARCWriterProcessor.java +++ b/modules/src/main/java/org/archive/modules/writer/WARCWriterProcessor.java @@ -37,6 +37,7 @@ import static org.archive.modules.CoreAttributeConstants.A_FTP_CONTROL_CONVERSAT import static org.archive.modules.CoreAttributeConstants.A_FTP_FETCH_STATUS; import static org.archive.modules.CoreAttributeConstants.A_SOURCE_TAG; import static org.archive.modules.CoreAttributeConstants.A_WARC_RESPONSE_HEADERS; +import static org.archive.modules.CoreAttributeConstants.A_WARC_STATS; import static org.archive.modules.CoreAttributeConstants.HEADER_TRUNC; import static org.archive.modules.CoreAttributeConstants.LENGTH_TRUNC; import static org.archive.modules.CoreAttributeConstants.TIMER_TRUNC; @@ -104,6 +105,9 @@ public class WARCWriterProcessor extends WriterPoolProcessor implements WARCWrit Logger.getLogger(WARCWriterProcessor.class.getName()); private ConcurrentMap> stats = new ConcurrentHashMap>(); + public ConcurrentMap> getStats() { + return stats; + } private AtomicLong urlsWritten = new AtomicLong(); @@ -204,28 +208,28 @@ public class WARCWriterProcessor extends WriterPoolProcessor implements WARCWrit final CrawlURI curi) throws IOException { WARCWriter writer = (WARCWriter) getPool().borrowFile(); - + + // Reset writer temp stats so they reflect only this set of records. + writer.resetTmpStats(); + writer.resetTmpRecordLog(); + long position = writer.getPosition(); try { - // See if we need to open a new file because we've exceeded maxBytes. - // Call to checkFileSize will open new file if we're at maximum for - // current file. + // Roll over to new warc file if we've exceeded maxBytes. writer.checkSize(); if (writer.getPosition() != position) { - // We just closed the file because it was larger than maxBytes. - // Add to the totalBytesWritten the size of the first record - // in the file, if any. + // We rolled over to a new warc and wrote a warcinfo record. + // Tally stats and reset temp stats, to avoid including warcinfo + // record in stats for current url. setTotalBytesWritten(getTotalBytesWritten() + (writer.getPosition() - position)); + addStats(writer.getTmpStats()); + writer.resetTmpStats(); + writer.resetTmpRecordLog(); + position = writer.getPosition(); } - - // Reset writer temp stats so they reflect only this set of records. - // They'll be added to totals below, in finally block, after records - // have been written. - writer.resetTmpStats(); - writer.resetTmpRecordLog(); - + // Write a request, response, and metadata all in the one // 'transaction'. final URI baseid = getRecordID(); @@ -256,9 +260,19 @@ public class WARCWriterProcessor extends WriterPoolProcessor implements WARCWrit getPool().returnFile(writer); } } + // XXX this looks wrong, check should happen *before* writing the + // record, the way checkBytesWritten() currently works return checkBytesWritten(); } + protected Map> copyStats(Map> orig) { + Map> copy = new HashMap>(orig.size()); + for (String k: orig.keySet()) { + copy.put(k, new HashMap(orig.get(k))); + } + return copy; + } + protected void updateMetadataAfterWrite(final CrawlURI curi, WARCWriter writer, long startPosition) { if (WARCWriter.getStat(writer.getTmpStats(), WARCWriter.TOTALS, WARCWriter.NUM_RECORDS) > 0l) { @@ -275,6 +289,8 @@ public class WARCWriterProcessor extends WriterPoolProcessor implements WARCWrit curi.addExtraInfo("warcFilename", writer.getFilenameWithoutOccupiedSuffix()); curi.addExtraInfo("warcFileOffset", startPosition); + curi.getData().put(A_WARC_STATS, copyStats(writer.getTmpStats())); + // history for uri-based dedupe Map[] history = curi.getFetchHistory(); if (history != null && history[0] != null) {