fixed & improved file I/O error handling

Author: muellan

src/classification.cpp

@@ -608,21 +608,17 @@ void map_queries_to_targets_default(
         results.mapout << buf.out.str();
     };
 
-    //runs in case of notification events
-    const auto showStatus = [&] (const std::string& msg, float progress) {
+    //runs if something needs to be appended to the output
+    const auto appendToOutput = [&] (const std::string& msg) {
         if(opt.output.mapViewMode != map_view_mode::none) {
             results.mapout << opt.output.format.comment << msg << '\n';
         }
-        if(progress >= 0.0f) {
-            show_progress_indicator(results.status, progress);
-        }
     };
 
-
     //run (parallel) database queries according to processing options
     query_database(infiles, db, opt.process,
                    makeBatchBuffer, processQuery, finalizeBatch,
-                   showStatus);
+                   appendToOutput);
 
     //target -> hits list?
     if(tgtMatches.empty()) return;

src/cmdline_utility.cpp

@@ -30,6 +30,7 @@ namespace mc {
 //-------------------------------------------------------------------
 void show_progress_indicator(std::ostream& os, float done, int totalLength)
 {
+    if(done < 0.f) done = 0.f;
     auto m = int((totalLength - 7) * done);
     os << "\r[";
     for(int j = 0; j < m; ++j) os << '=';

src/mode_query.cpp

@@ -154,11 +154,9 @@ void process_input_files(const vector<string>& infiles,
         return;
     }
     else {
-        bool anyreadable = false;
-        for(const auto& f : infiles) {
-            if(file_readable(f)) { anyreadable = true; break; }
-        }
-        if(!anyreadable) {
+        bool noneReadable = std::none_of(infiles.begin(), infiles.end(),
+                           [](const auto& f) { return file_readable(f); });
+        if(noneReadable) {
             throw std::runtime_error{
                         "None of the query sequence files could be opened"};
         }

src/query_options.cpp

@@ -102,6 +102,21 @@ get_query_processing_options(const args_parser& args,
     auto batchSize = args.get<int>("batch-size", defaults.batchSize);
     if(batchSize >= 1) opt.batchSize = batchSize;
 
+    auto numSeq = args.get<int>("per-thread-sequential-queries",
+                                defaults.perThreadSequentialQueries);
+
+    if(numSeq >= 1) {
+        opt.perThreadSequentialQueries = numSeq;
+    }
+    else {
+        //adapt to number of threads
+        if(numThreads < 64) {
+            opt.perThreadSequentialQueries = 4;
+        } else {
+            opt.perThreadSequentialQueries = 8;
+        }
+    }
+
     opt.queryLimit = args.get<std::int_least64_t>({"query-limit"},
                                           defaults.queryLimit);
 

src/query_options.h

@@ -107,6 +107,8 @@ struct query_processing_options
 
     std::size_t batchSize = 4096 * std::thread::hardware_concurrency();
 
+    int perThreadSequentialQueries = 4;
+
     //limits number of reads per sequence source (file)
     std::int_least64_t queryLimit = std::numeric_limits<std::int_least64_t>::max();
 };

src/querying.h

@@ -23,11 +23,13 @@
 #define MC_QUERYING_H_
 
 #include <vector>
+#include <iostream>
 
 #include "config.h"
 #include "sequence_io.h"
 #include "cmdline_utility.h"
 #include "query_options.h"
+#include "cmdline_utility.h"
 
 
 namespace mc {
@@ -75,15 +77,15 @@ struct sequence_query
  *****************************************************************************/
 template<
     class BufferSource, class BufferUpdate, class BufferSink,
-    class StatusCallback
+    class LogCallback
 >
 void query_batched(
     // parallel_queue& queue,
     const std::string& filename1, const std::string& filename2,
     const database& db, const query_processing_options& opt,
     query_id& startId,
     BufferSource&& getBuffer, BufferUpdate&& update, BufferSink&& finalize,
-    StatusCallback&& showStatus)
+    LogCallback&& log)
 {
     std::mutex exceptionMtx;
     std::mutex finalizeMtx;
@@ -99,22 +101,23 @@ void query_batched(
     // assign work to threads
     for(int i = 0; i < opt.numThreads; ++i) {
         threads.emplace_back([&] {
-            int threadId = i;
+            const int threadId = i;
             try {
                 sequence_pair_reader reader{filename1, filename2};
 
                 auto buffer = getBuffer();
                 match_locations matches;
-                bool empty = true;
+                bool bufferEmpty = true;
+
+                const auto readSequentially = opt.perThreadSequentialQueries;
 
-                const int readAtOnce = 4; // 16 threads
-                // const int readAtOnce = 4; // 32 threads
-                // const int readAtOnce = 8; // 64 threads
                 std::vector<sequence_pair_reader::sequence_pair> sequences;
-                sequences.reserve(readAtOnce);
+                sequences.reserve(readSequentially);
 
-                while(reader.has_next()) {
-                    for(std::size_t i = 0; i < opt.batchSize && queryLimit.fetch_sub(readAtOnce) > 0; ++i) {
+                while(reader.has_next() && queryLimit > 0) {
+                    for(std::size_t i = 0;
+                        i < opt.batchSize && queryLimit.fetch_sub(readSequentially) > 0; ++i)
+                    {
                         query_id myQid;
                         sequence_pair_reader::stream_positions myPos;
 
@@ -129,25 +132,20 @@ void query_batched(
                         reader.index_offset(myQid);
 
                         // get work id and skip to this read
-                        auto wid = workId.fetch_add(readAtOnce);
+                        auto wid = workId.fetch_add(readSequentially);
                         if(myQid != wid) {
                             reader.skip(wid-myQid);
                         }
                         if(!reader.has_next()) break;
                         // auto seq = reader.next();
-                        for(int i = 0; i < readAtOnce; ++i) {
-                            // auto seq = reader.next();
-                            // if(!seq.first.header.empty())
-                            //     sequences.emplace_back(seq);
+                        for(int i = 0; i < readSequentially; ++i) {
                             sequences.emplace_back(reader.next());
                         }
 
                         // update most recent position and query id
                         myQid = reader.index();
                         myPos = reader.tell();
                         while(myQid > qid) {// reading qid unsafe
-                            // std::unique_lock<std::mutex> lock(tipMtx, std::try_to_lock);
-                            // if(lock.owns_lock()) {
                             if(tipMtx.try_lock()) {
                                 if(myQid > qid) {// reading qid safe
                                     qid = myQid;
@@ -159,7 +157,7 @@ void query_batched(
 
                         for(auto& seq : sequences) {
                             if(!seq.first.header.empty()) {
-                                empty = false;
+                                bufferEmpty = false;
                                 matches.clear();
 
                                 db.accumulate_matches(seq.first.data, matches);
@@ -177,21 +175,21 @@ void query_batched(
                         }
                         sequences.clear();
                     }
-                    if(!empty) {
+                    if(!bufferEmpty) {
                         std::lock_guard<std::mutex> lock(finalizeMtx);
                         finalize(std::move(buffer));
                     }
                 }
             }
             catch(file_access_error& e) {
-                if(threadId == 0) {
+                if(threadId == 1) {
                     std::lock_guard<std::mutex> lock(exceptionMtx);
-                    showStatus(std::string("FAIL: ") + e.what(), -1);
+                    log(std::string("FAIL: ") + e.what());
                 }
             }
             catch(std::exception& e) {
                 std::lock_guard<std::mutex> lock(exceptionMtx);
-                showStatus(std::string("FAIL: ") + e.what(), -1);
+                log(std::string("FAIL: ") + e.what());
             }
         }); //emplace
     }
@@ -220,14 +218,14 @@ void query_batched(
  *****************************************************************************/
 template<
     class BufferSource, class BufferUpdate, class BufferSink,
-    class StatusCallback
+    class InfoCallback, class ProgressHandler, class LogHandler
 >
 void query_database(
     const std::vector<std::string>& infilenames,
     const database& db,
     const query_processing_options& opt,
     BufferSource&& bufsrc, BufferUpdate&& bufupdate, BufferSink&& bufsink,
-    StatusCallback&& showStatus)
+    InfoCallback&& showInfo, ProgressHandler&& showProgress, LogHandler&& log)
 {
     // parallel_queue queue(opt.numThreads);
     std::vector<std::thread> threads(opt.numThreads);
@@ -243,22 +241,56 @@ void query_database(
         const auto& fname2 = (opt.pairing == pairing_mode::none)
                              ? nofile : infilenames[i+stride];
 
-        const auto progress = infilenames.size() > 1 ? i / float(infilenames.size()) : -1;
         if(opt.pairing == pairing_mode::files) {
-            showStatus(fname1 + " + " + fname2, progress);
+            showInfo(fname1 + " + " + fname2);
         } else {
-            showStatus(fname1, progress);
+            showInfo(fname1);
         }
+        showProgress(infilenames.size() > 1 ? i/float(infilenames.size()) : -1);
 
         query_batched(fname1, fname2, db, opt, readIdOffset,
                       std::forward<BufferSource>(bufsrc),
                       std::forward<BufferUpdate>(bufupdate),
                       std::forward<BufferSink>(bufsink),
-                      std::forward<StatusCallback>(showStatus));
+                      std::forward<LogHandler>(log));
     }
 }
 
 
+
+/*************************************************************************//**
+ *
+ * @brief queries database
+ *
+ * @tparam BufferSource  returns a per-batch buffer object
+ *
+ * @tparam BufferUpdate  takes database matches of one query and a buffer;
+ *                       must be thread-safe (only const operations on DB!)
+ *
+ * @tparam BufferSink    recieves buffer after batch is finished
+ *
+ *****************************************************************************/
+template<
+    class BufferSource, class BufferUpdate, class BufferSink, class InfoCallback
+>
+void query_database(
+    const std::vector<std::string>& infilenames,
+    const database& db,
+    const query_processing_options& opt,
+    BufferSource&& bufsrc, BufferUpdate&& bufupdate, BufferSink&& bufsink,
+    InfoCallback&& showInfo)
+{
+    query_database(infilenames, db, opt,
+                   std::forward<BufferSource>(bufsrc),
+                   std::forward<BufferUpdate>(bufupdate),
+                   std::forward<BufferSink>(bufsink),
+                   std::forward<InfoCallback>(showInfo),
+                   [] (float p) { show_progress_indicator(std::cerr, p); },
+                   [] (const std::string& s) {std::cerr << s << '\n'; }
+                   );
+}
+
+
 } // namespace mc
 
 

src/sequence_io.cpp

@@ -101,7 +101,7 @@ fasta_reader::fasta_reader(const string& filename):
     if(!filename.empty()) {
         file_.open(filename);
 
-        if(file_.rdstate() & std::ifstream::failbit) {
+        if(!file_.good()) {
             invalidate();
             throw file_access_error{"can't open file " + filename};
         }
@@ -226,7 +226,7 @@ fastq_reader::fastq_reader(const string& filename):
     if(!filename.empty()) {
         file_.open(filename);
 
-        if(file_.rdstate() & std::ifstream::failbit) {
+        if(!file_.good()) {
             invalidate();
             throw file_access_error{"can't open file " + filename};
         }
@@ -326,7 +326,7 @@ sequence_header_reader::sequence_header_reader(const string& filename):
 {
     file_.open(filename);
 
-    if(file_.rdstate() & std::ifstream::failbit) {
+    if(!file_.good()) {
         invalidate();
         throw file_access_error{"can't open file " + filename};
     }
@@ -527,13 +527,13 @@ make_sequence_reader(const string& filename)
        filename.find(".fnq")   == (n-4) ||
        filename.find(".fastq") == (n-6) )
     {
-        return std::unique_ptr<sequence_reader>{new fastq_reader{filename}};
+        return std::make_unique<fastq_reader>(filename);
     }
     else if(filename.find(".fa")    == (n-3) ||
             filename.find(".fna")   == (n-4) ||
             filename.find(".fasta") == (n-6) )
     {
-        return std::unique_ptr<sequence_reader>{new fasta_reader{filename}};
+        return std::make_unique<fasta_reader>(filename);
     }
 
     //try to determine file type content
@@ -543,10 +543,10 @@ make_sequence_reader(const string& filename)
         getline(is,line);
         if(!line.empty()) {
             if(line[0] == '>') {
-                return std::unique_ptr<sequence_reader>{new fasta_reader{filename}};
+                return std::make_unique<fasta_reader>(filename);
             }
             else if(line[0] == '@') {
-                return std::unique_ptr<sequence_reader>{new fastq_reader{filename}};
+                return std::make_unique<fastq_reader>(filename);
             }
         }
         throw file_read_error{"file format not recognized"};