Make deduplication opt-in and prepare 2.2.0beta3

configure.ac

@@ -8,7 +8,7 @@
 # and http://www.bioinf.uni-freiburg.de/~mmann/HowTo/automake.html
 
 AC_PREREQ([2.69])
-AC_INIT([BULK_EXTRACTOR],[2.2.0beta2],[bugs@digitalcorpora.org])
+AC_INIT([BULK_EXTRACTOR],[2.2.0beta3],[bugs@digitalcorpora.org])
 AC_CONFIG_MACRO_DIR(m4)
 
 AC_MSG_NOTICE([at start CPPFLAGS are $CPPFLAGS])

doc/RELEASE_NOTES.md

@@ -14,7 +14,7 @@ linked from the [historical source map](#historical-source-map).
 ## 2.2.0 (draft)
 
 **Status:** Unreleased. The source version is currently
-`2.2.0beta2`; the corresponding Git tag will be `v2.2.0beta2`.
+`2.2.0beta3`; the corresponding Git tag will be `v2.2.0beta3`.
 
 ### Executive summary
 
@@ -94,6 +94,12 @@ through the project's normal pull-request and CI process.
 
 ### Reliability and correctness
 
+- Buffer deduplication is now disabled by default so concurrent runs
+  preserve every forensic path for equal content. `--deduplicate` restores the
+  previous space-saving behavior, whose first-seen path can depend on worker
+  scheduling. Every scan creates `duplicates.txt`; its deterministic rows list
+  the canonical first path, each additional path, and the SHA-1 content hash
+  ([#682](https://github.com/simsong/bulk_extractor/issues/682)).
 - Restart records the start of each page so a resumed run deliberately skips
   pages that were in progress at the crash, avoiding repeated data-dependent
   crashes; the behavior is covered by a controlled-crash regression test

doc/latex_manuals/BECurrentGuide.tex

@@ -299,8 +299,19 @@ \subsection{Pages, scanners, and recursion}
 scanner can decode a validated region and return the derived buffer to the
 framework for another pass. This permits, for example, the email scanner to
 find an address in text extracted from a PDF or decompressed from a GZIP
-stream. Depth limits, duplicate detection, buffer characteristics, and scanner
-flags prevent or limit inappropriate recursion.
+stream. Depth limits, buffer characteristics, and scanner flags prevent or
+limit inappropriate recursion. By default, equal recursive buffers are all
+scanned so that their distinct forensic paths are preserved. The optional
+\code{--deduplicate} switch suppresses repeated buffer content to reduce work
+and output size, but concurrent runs may retain different paths for that equal
+content.
+
+Every scan creates \code{duplicates.txt}. Its data rows have the form
+\code{path1<TAB>path2<TAB>hash}, where \code{path1} is the lowest forensic
+path for the content, \code{path2} is each additional path, and \code{hash} is
+the SHA-1 content hash. Hashes and paths are written in sorted order, so the
+report itself does not depend on worker scheduling. The file is empty when the
+scan finds no duplicate buffers.
 
 \begin{figure}[ht]
 \centering
@@ -843,6 +854,7 @@ \section{Command-line reference}
 \code{-G BYTES} & Set the primary page size. \\
 \code{-g BYTES} & Set the page margin. \\
 \code{-M N} & Set maximum recursive depth. \\
+\code{--deduplicate} & Bypass buffers whose content was already processed; disabled by default. \\
 \code{-s FRACTION[:PASSES]} & Randomly sample the input. \\
 \code{--max\_minute\_wait N} & Bound processing and shutdown wait time. \\
 \code{--max\_bad\_alloc\_errors N} & Set tolerated allocation failures. \\

doc/scanner_api.md

@@ -115,8 +115,12 @@ callback returns. feature_recorder::write() and write_buf() are safe for
 concurrent scanner calls; protect any other shared scanner state with an
 appropriate mutex or atomics.
 
-The framework can bypass a scanner for a small buffer, a duplicate buffer, an
-ngram buffer, excessive depth, or scanner flags. A scanner must not assume it
+The framework can bypass a scanner for a small buffer, an ngram buffer,
+excessive depth, or scanner flags. With `--deduplicate`, it can also bypass a
+buffer whose content hash was seen earlier. Deduplication is disabled by
+default because concurrent scans can otherwise choose different forensic paths
+for equal content. Duplicate detection still runs and produces
+`duplicates.txt`; only the bypass is optional. A scanner must not assume it
 will see every input buffer.
 
 ## Feature recorders

man/bulk_extractor.1

@@ -159,6 +159,11 @@ Set the page margin size.
 .B -M, --max_depth \fIN\fR
 Set the maximum recursive-processing depth.
 .TP
+.B --deduplicate
+Bypass buffers whose content hash was seen earlier.  This can reduce processing
+and output size, but concurrent runs may retain different forensic paths for
+equal content.  Disabled by default.
+.TP
 .B -s, --sampling \fIFRACTION\fR[:\fIPASSES\fR]
 Set random-sampling parameters.
 .TP
@@ -230,6 +235,11 @@ is not supplied.
 .I outdir/report.xml
 DFXML report describing the scan.
 .TP
+.I outdir/duplicates.txt
+Duplicate-buffer report.  Each data row contains the canonical first forensic
+path, an additional path, and the SHA-1 content hash, separated by tabs.  The
+file is created even when no duplicate buffers are found.
+.TP
 .I outdir/bulk_extractor.log
 Default diagnostic log file.
 .SH SEE ALSO

src/be20_api/scanner_config.h

@@ -114,6 +114,7 @@ struct scanner_config {
     std::string hash_algorithm {"sha1"};          // which hash algorithm are using; default to SHA1
 
     bool allow_recurse { true };         // can be turned off for testing
+    bool deduplicate { false };          // bypass duplicate sbufs when enabled
 
     inline static const std::string NO_INPUT = "<NO-INPUT>"; // 'filename' indicator that the FRS has no input file
     inline static const std::string NO_OUTDIR = "<NO-OUTDIR>"; // 'dirname' indicator that the FRS produces no file output

src/be20_api/scanner_params.cpp

@@ -31,7 +31,8 @@ feature_recorder& scanner_params::named_feature_recorder(const std::string featu
 bool scanner_params::check_previously_processed(const sbuf_t &s) const
 {
     assert(ss!=nullptr);
-    return ss->previously_processed_count(s)==0;
+    const bool seen_before = ss->previously_processed_count(s) > 0;
+    return sc.deduplicate && seen_before;
 }
 
 void scanner_params::recurse(const sbuf_t* new_sbuf) const {

src/be20_api/scanner_set.cpp

@@ -79,6 +79,11 @@ scanner_set::scanner_set(scanner_config& sc_, const feature_recorder_set::flags_
     fs.banner_filename                     = sc.banner_file;
     pool.debug                             = getenv_debug("DEBUG_THREAD_POOL");
 
+    feature_recorder_def::flags_t duplicate_flags;
+    duplicate_flags.no_stoplist = true;
+    duplicate_flags.no_alertlist = true;
+    fs.create_feature_recorder(feature_recorder_def(DUPLICATES_RECORDER_NAME, duplicate_flags));
+
     const char *dsi = std::getenv("DEBUG_SCANNERS_IGNORE");
     if (dsi!=nullptr) debug_flags.debug_scanners_ignore=dsi;
 
@@ -1008,14 +1013,12 @@ void scanner_set::process_sbuf(const sbuf_t* sbufp)
 
     update_maximum<unsigned int>(max_depth_seen, sbuf.depth());
 
-    /* Determine if we have seen this buffer before.
-     * Some scanners are okay with duplicate processing, but the default is that they are not.
-     * Nevertheless, we will add dupliate bytes to the hash of the disk image.
-     */
-    sbufp->seen_before = previously_processed_count(sbuf) > 0; // abstraction violation
-    if (sbufp->seen_before) {
+    /* Always record duplicate paths. Only mark or bypass duplicate content when requested. */
+    const bool seen_before = previously_processed_count(sbuf) > 0;
+    sbufp->seen_before = sc.deduplicate && seen_before; // abstraction violation
+    if (seen_before) {
         dup_bytes_encountered += sbuf.bufsize;
-        if (!scan_seen_before_enabled) {
+        if (sc.deduplicate && !scan_seen_before_enabled) {
             duplicate_sbufs_bypassed++;
             if (debug_flags.debug_benchmark && writer) {
                 writer->xmlout("debug:bypass", "",
@@ -1142,6 +1145,7 @@ void scanner_set::shutdown()
     scanner_params sp(sc, this, nullptr, scanner_params::PHASE_SHUTDOWN, nullptr);
     for (const auto &it : enabled_scanners) { (*it)(sp); }
 
+    write_duplicate_report();
     fs.feature_recorders_shutdown();
 
     /* Tell every feature recorder to flush all of its histograms if they haven't been generated.
@@ -1176,6 +1180,23 @@ std::string scanner_set::hash(const sbuf_t& sbuf) const
 }
 
 uint64_t scanner_set::previously_processed_count(const sbuf_t& sbuf) {
-    std::string hash = sbuf.hash();
-    return previously_processed_counter[ hash ]++;
+    const std::string hash = sbuf.hash();
+    const std::lock_guard<std::mutex> lock(Mpreviously_processed);
+    auto &paths = previously_processed[hash];
+    const uint64_t count = paths.size();
+    paths.insert(sbuf.pos0);
+    return count;
+}
+
+void scanner_set::write_duplicate_report() {
+    feature_recorder &recorder = fs.named_feature_recorder(DUPLICATES_RECORDER_NAME);
+    const std::lock_guard<std::mutex> lock(Mpreviously_processed);
+    for (const auto &[hash, paths] : previously_processed) {
+        if (paths.size() < 2) continue;
+        const auto first = paths.begin();
+        auto path = first;
+        for (++path; path != paths.end(); ++path) {
+            recorder.write(*first, path->str(), hash);
+        }
+    }
 }

src/be20_api/scanner_set.h

@@ -16,7 +16,6 @@
 #endif
 
 #include "utils.h"
-#include "atomic_map.h"
 #include "sbuf.h"
 #include "scanner_config.h"
 #include "scanner_params.h"
@@ -119,7 +118,9 @@ class scanner_set {
     void *cpu_benchmark();
     static void launch_cpu_benchmark_thread(void *arg);
     class feature_recorder_set fs;      // the feature recorders
-    atomic_map<std::string, std::atomic<uint64_t>> previously_processed_counter {};
+    std::map<std::string, std::multiset<pos0_t>> previously_processed {};
+    mutable std::mutex Mpreviously_processed {};
+    void write_duplicate_report();
     std::map<std::thread::id, std::string> thread_status {}; // the status of each thread::id
     mutable std::mutex Mthread_status {};                       // mutex for thread_status
 
@@ -203,6 +204,7 @@ class scanner_set {
     static const inline std::string SBUFS_CREATED_STR {"sbufs_created"};
     static const inline std::string SBUFS_REMAINING_STR {"sbufs_remaining"};
     static const inline std::string MAX_OFFSET {"max_offset"};
+    static const inline std::string DUPLICATES_RECORDER_NAME {"duplicates"};
 
     bool get_threading() const   { return threading;};
     int get_worker_count() const { return threading ? pool.get_worker_count()  : 1; };

src/be20_api/test_be20_api.cpp

@@ -1404,11 +1404,43 @@ TEST_CASE("previously_processed", "[scanner_set]") {
     REQUIRE(ss.previously_processed_count(slg) == 2);
 }
 
+TEST_CASE("scanner duplicate check skips only repeated content", "[scanner_set]") {
+    scanner_config sc;
+    sc.outdir = NamedTemporaryDirectory();
+    sc.deduplicate = true;
+    scanner_set ss(sc, feature_recorder_set::flags_t(), nullptr);
+    sbuf_t slg("Simson");
+    scanner_params sp(sc, &ss, nullptr, scanner_params::PHASE_SCAN, &slg);
+    REQUIRE_FALSE(sp.check_previously_processed(slg));
+    REQUIRE(sp.check_previously_processed(slg));
+}
+
+TEST_CASE("scanner duplicate check records content when bypass is disabled", "[scanner_set]") {
+    scanner_config sc;
+    sc.outdir = NamedTemporaryDirectory();
+    scanner_set ss(sc, feature_recorder_set::flags_t(), nullptr);
+    sbuf_t slg("Simson");
+    scanner_params sp(sc, &ss, nullptr, scanner_params::PHASE_SCAN, &slg);
+    REQUIRE_FALSE(sp.check_previously_processed(slg));
+    REQUIRE_FALSE(sp.check_previously_processed(slg));
+    REQUIRE(ss.previously_processed_count(slg) == 2);
+}
+
+TEST_CASE("duplicates report is always created", "[scanner_set]") {
+    scanner_config sc;
+    sc.outdir = NamedTemporaryDirectory();
+    scanner_set ss(sc, feature_recorder_set::flags_t(), nullptr);
+    const auto report = sc.outdir / "duplicates.txt";
+    REQUIRE(std::filesystem::exists(report));
+    REQUIRE(std::filesystem::file_size(report) == 0);
+}
+
 TEST_CASE("duplicate sbufs bypass scanner fan-out unless requested", "[scanner_set]") {
     scanner_config sc;
-    sc.outdir = get_tempdir();
+    sc.outdir = NamedTemporaryDirectory();
+    sc.deduplicate = true;
     sc.enable_all_scanners();
-    const auto dfxml_file = get_tempdir() / "duplicate_bypass.xml";
+    const auto dfxml_file = sc.outdir / "duplicate_bypass.xml";
     dfxml_writer writer(dfxml_file, false);
     duplicate_bypass_scans = 0;
     scanner_set ss(sc, feature_recorder_set::flags_t(), &writer);
@@ -1430,7 +1462,8 @@ TEST_CASE("duplicate sbufs bypass scanner fan-out unless requested", "[scanner_s
 
 TEST_CASE("duplicate sbufs reach scanners that opt in", "[scanner_set]") {
     scanner_config sc;
-    sc.outdir = get_tempdir();
+    sc.outdir = NamedTemporaryDirectory();
+    sc.deduplicate = true;
     sc.enable_all_scanners();
     duplicate_bypass_scans = 0;
     duplicate_opt_in_scans = 0;
@@ -1447,6 +1480,29 @@ TEST_CASE("duplicate sbufs reach scanners that opt in", "[scanner_set]") {
     ss.shutdown();
 }
 
+TEST_CASE("duplicate sbuf bypass is disabled by default", "[scanner_set]") {
+    scanner_config sc;
+    sc.outdir = NamedTemporaryDirectory();
+    sc.enable_all_scanners();
+    duplicate_bypass_scans = 0;
+    sbuf_t duplicate("duplicate");
+    const auto hash = duplicate.hash();
+    scanner_set ss(sc, feature_recorder_set::flags_t(), nullptr);
+    ss.add_scanner(scan_duplicate_bypass_test);
+    ss.apply_scanner_commands();
+    ss.phase_scan();
+    ss.schedule_sbuf(sbuf_t::sbuf_malloc(pos0_t("zeta"), "duplicate"));
+    ss.schedule_sbuf(sbuf_t::sbuf_malloc(pos0_t("alpha"), "duplicate"));
+    ss.schedule_sbuf(sbuf_t::sbuf_malloc(pos0_t("middle"), "duplicate"));
+    REQUIRE(duplicate_bypass_scans == 3);
+    REQUIRE(ss.get_dup_bytes_encountered() == 18);
+    REQUIRE(ss.get_duplicate_sbufs_bypassed() == 0);
+    ss.shutdown();
+    const auto lines = getLines(sc.outdir / "duplicates.txt");
+    REQUIRE(std::find(lines.begin(), lines.end(), "alpha-0\tmiddle-0\t" + hash) != lines.end());
+    REQUIRE(std::find(lines.begin(), lines.end(), "alpha-0\tzeta-0\t" + hash) != lines.end());
+}
+
 
 
 TEST_CASE("enable/disable", "[scanner_set]") {
@@ -1471,10 +1527,10 @@ TEST_CASE("enable/disable", "[scanner_set]") {
         REQUIRE(ss.is_scanner_enabled(SHA1_TEST) == true);
         REQUIRE(ss.is_find_scanner_enabled() == false); // only sha1 scanner is enabled
 
-        /* Make sure that the scanner set has a two feature recorders:
-         * the alert recorder and the sha1_bufs recorder
+        /* Make sure that the scanner set has three feature recorders:
+         * the duplicates recorder, alert recorder, and sha1_bufs recorder
          */
-        REQUIRE(ss.feature_recorder_count() == 2);
+        REQUIRE(ss.feature_recorder_count() == 3);
 
         /* Make sure it has a single histogram */
         REQUIRE(ss.histogram_count() == 1);

src/bulk_extractor.cpp

@@ -226,9 +226,10 @@ int bulk_extractor_main( std::ostream &cout, std::ostream &cerr, int argc,char *
     ("image_name", image_name_help.c_str(), cxxopts::value<std::string>())
         ("A,offset_add", "Offset added (in bytes) to feature locations", cxxopts::value<int64_t>()->default_value("0"))
         ("b,banner_file", "Path of file whose contents are prepended to top of all feature files",cxxopts::value<std::string>())
-	("C,context_window", "Size of context window reported in bytes",
+        ("C,context_window", "Size of context window reported in bytes",
          cxxopts::value<int>()->default_value(std::to_string(sc.context_window_default)))
         ("d,debug", "enable debug-level diagnostic logging")
+        ("deduplicate", "bypass buffers whose content was already processed")
         ("E,enable_exclusive", "disable all scanners except the one specified. Same as -x all -E scanner.", cxxopts::value<std::string>())
         ("e,enable",   "enable a scanner (can be repeated)", cxxopts::value<std::vector<std::string>>())
         ("x,disable",  "disable a scanner (can be repeated)", cxxopts::value<std::vector<std::string>>())
@@ -300,6 +301,7 @@ int bulk_extractor_main( std::ostream &cout, std::ostream &cerr, int argc,char *
 
     sc.offset_add  = result["offset_add"].as<int64_t>();
     sc.context_window_default = result["context_window"].as<int>();
+    sc.deduplicate = result.count("deduplicate") != 0;
     const int max_minute_wait = result["max_minute_wait"].as<int>();
     if (max_minute_wait <= 0) {
         throw std::runtime_error("--max_minute_wait must be positive");

src/test_be3.cpp

@@ -209,6 +209,58 @@ TEST_CASE("report DFXML validates against the bundled schema", "[end-to-end]")
     REQUIRE(std::system(command.c_str()) == 0);
 }
 
+TEST_CASE("deduplication is opt-in and reports duplicate paths", "[end-to-end]")
+{
+    const auto root = NamedTemporaryDirectory();
+    const auto input = root / "duplicate-gzip.raw";
+    const auto gzip_path = test_dir() / "test_hello.gz";
+    std::ifstream gzip_file(gzip_path, std::ios::binary);
+    const std::string gzip((std::istreambuf_iterator<char>(gzip_file)),
+                           std::istreambuf_iterator<char>());
+    std::ofstream input_file(input, std::ios::binary);
+    input_file.write(gzip.data(), gzip.size());
+    input_file.write(gzip.data(), gzip.size());
+    input_file.close();
+
+    const auto run = [&](const std::filesystem::path &outdir, bool deduplicate) {
+        const std::string input_string = input.string();
+        const std::string outdir_string = outdir.string();
+        const char *default_argv[] = {
+            "bulk_extractor", "-0q", "-J", "-x", "all", "-e", "email", "-e", "gzip",
+            "-o", outdir_string.c_str(), input_string.c_str(), nullptr
+        };
+        const char *deduplicate_argv[] = {
+            "bulk_extractor", "-0q", "-J", "--deduplicate", "-x", "all",
+            "-e", "email", "-e", "gzip", "-o", outdir_string.c_str(),
+            input_string.c_str(), nullptr
+        };
+        std::stringstream output;
+        REQUIRE(run_be(output, deduplicate ? deduplicate_argv : default_argv) == 0);
+    };
+
+    const auto default_outdir = root / "default";
+    const auto deduplicate_outdir = root / "deduplicate";
+    run(default_outdir, false);
+    run(deduplicate_outdir, true);
+
+    const auto email_count = [](const std::filesystem::path &outdir) {
+        const auto lines = getLines(outdir / "email.txt");
+        return std::count_if(lines.begin(), lines.end(), [](const auto &line) {
+            return line.find("\thello@world.com\t") != std::string::npos;
+        });
+    };
+    REQUIRE(email_count(default_outdir) == 2);
+    REQUIRE(email_count(deduplicate_outdir) == 1);
+
+    const sbuf_t decoded("hello@world.com\n");
+    const std::string duplicate = "0-GZIP-0\t" + std::to_string(gzip.size())
+        + "-GZIP-0\t" + decoded.hash();
+    for (const auto &outdir : {default_outdir, deduplicate_outdir}) {
+        const auto lines = getLines(outdir / "duplicates.txt");
+        REQUIRE(std::find(lines.begin(), lines.end(), duplicate) != lines.end());
+    }
+}
+
 TEST_CASE("Windows raw-device paths are recognized narrowly", "[image_process]")
 {
     REQUIRE(process_raw::is_windows_raw_device_path(R"(\\.\PhysicalDrive0)"));