diff --git a/commons/src/main/java/org/archive/io/WriterPoolMember.java b/commons/src/main/java/org/archive/io/WriterPoolMember.java index 02c85f18..4d5c18a6 100644 --- a/commons/src/main/java/org/archive/io/WriterPoolMember.java +++ b/commons/src/main/java/org/archive/io/WriterPoolMember.java @@ -406,21 +406,33 @@ public abstract class WriterPoolMember implements ArchiveFileConstants { return compressed; } - protected void write(final byte [] b) throws IOException { - this.out.write(b); + /** + * @return number of bytes written, which is always {@code b.length} + */ + protected int write(final byte[] b) throws IOException { + this.out.write(b); + return b.length; } - - protected void flush() throws IOException { - this.out.flush(); - } - protected void write(byte[] b, int off, int len) throws IOException { - this.out.write(b, off, len); - } + protected void flush() throws IOException { + this.out.flush(); + } - protected void write(int b) throws IOException { - this.out.write(b); - } + /** + * @return number of bytes written, which is always {@code len} + */ + protected int write(byte[] b, int off, int len) throws IOException { + this.out.write(b, off, len); + return len; + } + + /** + * @return number of bytes written, which is always {@code 1} + */ + protected int write(int b) throws IOException { + this.out.write(b); + return 1; + } /** * Copy bytes from the provided InputStream to the target file/stream being @@ -433,9 +445,10 @@ public abstract class WriterPoolMember implements ArchiveFileConstants { * @param enforceLength * whether to throw an exception if too many/too few bytes are * available from stream + * @return number of bytes written (normally equal to {@code enforceLength}) * @throws IOException */ - protected void copyFrom(final InputStream is, final long recordLength, + protected long copyFrom(final InputStream is, final long recordLength, boolean enforceLength) throws IOException { int read = scratchbuffer.length; long tot = 0; @@ -452,6 +465,8 @@ public abstract class WriterPoolMember implements ArchiveFileConstants { throw new IOException("Read " + tot + " but expected " + recordLength); } + + return tot; } public void close() throws IOException { diff --git a/commons/src/main/java/org/archive/io/warc/WARCWriter.java b/commons/src/main/java/org/archive/io/warc/WARCWriter.java index 6d29d9a7..b5075d4b 100644 --- a/commons/src/main/java/org/archive/io/warc/WARCWriter.java +++ b/commons/src/main/java/org/archive/io/warc/WARCWriter.java @@ -27,6 +27,7 @@ import java.io.InputStream; import java.io.OutputStream; import java.net.URI; import java.net.URISyntaxException; +import java.util.HashMap; import java.util.Iterator; import java.util.List; import java.util.Map; @@ -56,6 +57,12 @@ import org.archive.util.anvl.Element; */ public class WARCWriter extends WriterPoolMember implements WARCConstants { + public static final String TOTALS = "totals"; + public static final String SIZE_ON_DISK = "sizeOnDisk"; + public static final String TOTAL_BYTES = "totalBytes"; + public static final String CONTENT_BYTES = "contentBytes"; + public static final String NUM_RECORDS = "numRecords"; + private static final Logger logger = Logger.getLogger(WARCWriter.class.getName()); @@ -77,6 +84,7 @@ implements WARCConstants { */ private final List fileMetadata; + private Map> stats; /** * Shutdown Constructor @@ -226,51 +234,114 @@ implements WARCConstants { } protected void writeRecord(final String type, final String url, - final String create14DigitDate, final String mimetype, - final URI recordId, ANVLRecord xtraHeaders, + final String create14DigitDate, final String mimetype, + final URI recordId, ANVLRecord xtraHeaders, final InputStream contentStream, final long contentLength, boolean enforceLength) throws IOException { - if (!TYPES_LIST.contains(type)) { - throw new IllegalArgumentException("Unknown record type: " + type); - } - if (contentLength == 0 && + if (!TYPES_LIST.contains(type)) { + throw new IllegalArgumentException("Unknown record type: " + type); + } + if (contentLength == 0 && (xtraHeaders == null || xtraHeaders.size() <= 0)) { - throw new IllegalArgumentException("Cannot write record " + - "of content-length zero and base headers only."); - } + throw new IllegalArgumentException("Cannot write record " + + "of content-length zero and base headers only."); + } - String header; - try { - header = createRecordHeader(type, url, - create14DigitDate, mimetype, recordId, xtraHeaders, - contentLength); + String header; + try { + header = createRecordHeader(type, url, + create14DigitDate, mimetype, recordId, xtraHeaders, + contentLength); - } catch (IllegalArgumentException e) { - logger.log(Level.SEVERE,"could not write record type: " + type - + "for URL: " + url, e); - return; - } - - try { + } catch (IllegalArgumentException e) { + logger.log(Level.SEVERE,"could not write record type: " + type + + "for URL: " + url, e); + return; + } + + long contentBytes = 0; + long totalBytes = 0; + long startPosition; + + try { + checkSize(); // may start a new output file + startPosition = getPosition(); preWriteRecordTasks(); + // TODO: Revisit endcoding of header. - write(header.getBytes(WARC_HEADER_ENCODING)); - + totalBytes += write(header.getBytes(WARC_HEADER_ENCODING)); + if (contentStream != null && contentLength > 0) { // Write out the header/body separator. - write(CRLF_BYTES); // TODO: should this be written even for zero-length? - copyFrom(contentStream, contentLength, enforceLength); + totalBytes += write(CRLF_BYTES); // TODO: should this be written even for zero-length? + contentBytes += copyFrom(contentStream, contentLength, enforceLength); + totalBytes += contentBytes; } - + // Write out the two blank lines at end of all records. - write(CRLF_BYTES); - write(CRLF_BYTES); + totalBytes += write(CRLF_BYTES); + totalBytes += write(CRLF_BYTES); } finally { postWriteRecordTasks(); } + + // TODO: should this be in the finally block? + tally(type, contentBytes, totalBytes, getPosition() - startPosition); } + // if compression is enabled, sizeOnDisk means compressed bytes; if not, it + // should be the same as totalBytes (right?) + protected void tally(String recordType, long contentBytes, long totalBytes, long sizeOnDisk) { + if (stats == null) { + stats = new HashMap>(); + } + + // add to stats for this record type + Map substats = stats.get(recordType); + if (substats == null) { + substats = new HashMap(); + stats.put(recordType, substats); + } + subtally(substats, contentBytes, totalBytes, sizeOnDisk); + + // add to totals + substats = stats.get(TOTALS); + if (substats == null) { + substats = new HashMap(); + stats.put(TOTALS, substats); + } + subtally(substats, contentBytes, totalBytes, sizeOnDisk); + } + + protected void subtally(Map substats, long contentBytes, + long totalBytes, long sizeOnDisk) { + + if (substats.get(NUM_RECORDS) == null) { + substats.put(NUM_RECORDS, 1l); + } else { + substats.put(NUM_RECORDS, substats.get(CONTENT_BYTES) + 1l); + } + + if (substats.get(CONTENT_BYTES) == null) { + substats.put(CONTENT_BYTES, contentBytes); + } else { + substats.put(CONTENT_BYTES, substats.get(CONTENT_BYTES) + contentBytes); + } + + if (substats.get(TOTAL_BYTES) == null) { + substats.put(TOTAL_BYTES, totalBytes); + } else { + substats.put(TOTAL_BYTES, substats.get(TOTAL_BYTES) + totalBytes); + } + + if (substats.get(SIZE_ON_DISK) == null) { + substats.put(SIZE_ON_DISK, sizeOnDisk); + } else { + substats.put(SIZE_ON_DISK, substats.get(SIZE_ON_DISK) + sizeOnDisk); + } + } + protected URI generateRecordId(final Map qualifiers) throws IOException { URI rid = null; @@ -453,4 +524,26 @@ implements WARCConstants { } return result; } + + public void resetStats() { + if (stats != null) { + for (Map substats : stats.values()) { + for (Map.Entry entry : substats.entrySet()) { + entry.setValue(0l); + } + } + } + } + + public Map> getStats() { + return stats; + } + + public static long getStat(Map> statz, String key, String subkey) { + if (statz != null && statz.get(key) != null && statz.get(key).get(subkey) != null) { + return statz.get(key).get(subkey); + } else { + return 0; + } + } } \ No newline at end of file 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 de0b04c0..fdaacbe9 100644 --- a/modules/src/main/java/org/archive/modules/writer/WARCWriterProcessor.java +++ b/modules/src/main/java/org/archive/modules/writer/WARCWriterProcessor.java @@ -94,6 +94,9 @@ public class WARCWriterProcessor extends WriterPoolProcessor { private static final long serialVersionUID = 6182850087635847443L; private static final Logger logger = Logger.getLogger(WARCWriterProcessor.class.getName()); + + private HashMap> stats; + private int urlsWritten; public long getDefaultMaxFileSize() { return 1000000000L; // 1 SI giga-byte (10^9 bytes), per WARC appendix A @@ -200,21 +203,10 @@ public class WARCWriterProcessor extends WriterPoolProcessor { final CrawlURI curi) throws IOException { WriterPoolMember writer = getPool().borrowFile(); - long position = writer.getPosition(); - // See if we need to open a new file because we've exceeed maxBytes. - // Call to checkFileSize will open new file if we're at maximum for - // current file. - 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. - setTotalBytesWritten(getTotalBytesWritten() + - (writer.getPosition() - position)); - position = writer.getPosition(); - } WARCWriter w = (WARCWriter)writer; + w.resetStats(); + try { // Write a request, response, and metadata all in the one // 'transaction'. @@ -240,14 +232,37 @@ public class WARCWriterProcessor extends WriterPoolProcessor { throw e; } finally { if (writer != null) { - setTotalBytesWritten(getTotalBytesWritten() + - (writer.getPosition() - position)); + if (WARCWriter.getStat(w.getStats(), WARCWriter.TOTALS, WARCWriter.NUM_RECORDS) > 0) { + addStats(w.getStats()); + urlsWritten++; + } + logger.fine("wrote " + WARCWriter.getStat(w.getStats(), WARCWriter.TOTALS, WARCWriter.SIZE_ON_DISK) + " bytes to " + w.getFile().getName() + " for " + curi); + setTotalBytesWritten(getTotalBytesWritten() + WARCWriter.getStat(w.getStats(), WARCWriter.TOTALS, WARCWriter.SIZE_ON_DISK)); getPool().returnFile(writer); } } return checkBytesWritten(); } - + + protected void addStats(Map> statz) { + if (stats == null) { + stats = new HashMap>(); + } + + for (String key: statz.keySet()) { + if (stats.get(key) == null) { + stats.put(key, new HashMap()); + } + for (String subkey: statz.get(key).keySet()) { + if (stats.get(key).get(subkey) == null) { + stats.get(key).put(subkey, statz.get(key).get(subkey)); + } else { + stats.get(key).put(subkey, stats.get(key).get(subkey) + statz.get(key).get(subkey)); + } + } + } + } + private void writeDnsRecords(final CrawlURI curi, WARCWriter w, final URI baseid, final String timestamp) throws IOException { ANVLRecord headers = null; @@ -648,4 +663,30 @@ public class WARCWriterProcessor extends WriterPoolProcessor { record.addLabelValue(label, value); } } + + @Override + public String report() { + logger.info("final stats: " + stats); + + StringBuilder buf = new StringBuilder(); + buf.append("Processor: " + getClass().getName() + "\n"); + buf.append(" Function: Writes WARCs\n"); + buf.append(" Total CrawlURIs: " + urlsWritten + "\n"); + buf.append(" Revisit records: " + WARCWriter.getStat(stats, WARCWriter.REVISIT, WARCWriter.NUM_RECORDS) + "\n"); + + long bytes = WARCWriter.getStat(stats, WARCWriter.RESPONSE, WARCWriter.CONTENT_BYTES) + + WARCWriter.getStat(stats, WARCWriter.RESOURCE, WARCWriter.CONTENT_BYTES); + buf.append(" Crawled content bytes (including http headers): " + + bytes + " (" + ArchiveUtils.formatBytesForDisplay(bytes) + ")\n"); + + bytes = WARCWriter.getStat(stats, WARCWriter.TOTALS, WARCWriter.TOTAL_BYTES); + buf.append(" Total uncompressed bytes (including all warc records): " + + bytes + " (" + ArchiveUtils.formatBytesForDisplay(bytes) + ")\n"); + + buf.append(" Total size on disk ("+ (getCompress() ? "compressed" : "uncompressed") + "): " + + getTotalBytesWritten() + " (" + ArchiveUtils.formatBytesForDisplay(getTotalBytesWritten()) + ")\n"); + + return buf.toString(); + } + } \ No newline at end of file