relax atomic consistency; after all, nothing could possibly go wrong

Author: mlin

loaders/common.hpp

@@ -133,48 +133,56 @@ template <class Item> class BackgroundProducer {
     virtual bool Produce(Item &) = 0;
 
   private:
-    vector<Item> ring_;
+    // ~constant:
     int R_;
     unique_ptr<thread> worker_;
+
+    // writable by either thread:
     atomic<bool> stop_;
+
+    // writable by background producer thread:
+    vector<Item> ring_;
+    atomic<long long> p_; // # produced
     string errmsg_;
-    atomic<long long> p_, // produced
-        c_;               // if c_>0 then item (c_-1)%R is currently being consumed
-    chrono::duration<double> p_blocked_ = chrono::duration<double>::zero(),
-                             c_blocked_ = chrono::duration<double>::zero();
     chrono::time_point<chrono::high_resolution_clock> t0_;
+    chrono::duration<double> p_blocked_ = chrono::duration<double>::zero();
 
-    bool empty() { return p_ == c_; }
-
-    bool full() { return p_ - max(c_.load(), 1LL) == R_ - 1; }
+    // writable by consumer thread:
+    atomic<long long> c_; // if c_>0 then item (c_-1)%R is currently being consumed
+    const Item *item_ = nullptr;
+    chrono::duration<double> c_blocked_ = chrono::duration<double>::zero();
 
     void background_thread() {
         t0_ = chrono::high_resolution_clock::now();
         do {
-            assert(p_ >= c_);
+            auto p = p_.load(memory_order_acquire);
             bool ok = false;
             try {
-                ok = Produce(ring_[p_ % R_]);
+                ok = Produce(ring_[p % R_]);
             } catch (exception &exn) {
                 errmsg_ = exn.what();
                 if (errmsg_.empty()) {
                     errmsg_ = "unknown error on producer thread";
                 }
-                stop_ = true;
+                stop_.store(true, memory_order_release);
             }
             if (ok) {
-                ++p_;
-                if (full()) {
+                ++p;
+                p_.store(p, memory_order_release);
+                if (p - max(c_.load(memory_order_acquire), 1LL) == R_ - 1) {
                     auto t_spin = chrono::high_resolution_clock::now();
                     do {
+                        // assumption -- producer will usually be faster than the consumer, and
+                        // ringsize_ provides buffer if that's occasionally not the case
                         this_thread::sleep_for(chrono::milliseconds(1));
-                    } while (!stop_ && full());
+                    } while (!stop_.load(memory_order_relaxed) &&
+                             (p - max(c_.load(memory_order_acquire), 1LL) == R_ - 1));
                     p_blocked_ += chrono::high_resolution_clock::now() - t_spin;
                 }
             } else {
-                stop_ = true;
+                stop_.store(true, memory_order_relaxed);
             }
-        } while (!stop_);
+        } while (!stop_.load(memory_order_relaxed));
     }
 
   public:
@@ -192,36 +200,36 @@ template <class Item> class BackgroundProducer {
     bool next() {
         if (!worker_) {
             while (ring_.size() < R_) {
-                ring_.push_back(Item());
+                ring_.emplace_back();
             }
             worker_.reset(new thread([this]() { this->background_thread(); }));
         }
-        if (empty()) {
-            auto t_start = chrono::high_resolution_clock::now();
-            while (!stop_ && empty()) {
+        auto c = c_.load(memory_order_acquire), p = p_.load(memory_order_acquire);
+        if (c == p) {
+            auto t_spin = chrono::high_resolution_clock::now();
+            while (c == p && !stop_.load(memory_order_relaxed)) {
                 this_thread::yield();
+                p = p_.load(memory_order_acquire);
             }
-            c_blocked_ += chrono::high_resolution_clock::now() - t_start;
+            c_blocked_ += chrono::high_resolution_clock::now() - t_spin;
         }
-        if (stop_ && empty()) {
+        if (c == p && stop_.load(memory_order_acquire)) {
             if (!errmsg_.empty()) {
                 throw runtime_error(errmsg_);
             }
             return false;
         }
-        assert(c_ < p_);
-        c_++;
+        assert(c < p);
+        item_ = &ring_[c % R_];
+        c_.store(c + 1, memory_order_release);
         return true;
     }
 
     // get current item for consumption; defined only after next() returned true
-    Item &item() {
-        assert(c_ && errmsg_.empty());
-        return ring_[(c_ - 1) % R_];
-    }
+    const Item &item() { return *item_; }
 
     void abort() {
-        stop_ = true;
+        stop_.store(true, memory_order_relaxed);
         if (worker_) {
             worker_->join();
         }

loaders/sam_into_sqlite.cc

@@ -112,26 +112,36 @@ struct SamItem {
     shared_ptr<kstring_t> line;
     vector<char *> fields;
     shared_ptr<bam1_t> rec;
+
+    string tags_json;
+    int rg_id = -1;
+
+    SamItem() {
+        line = kstringXX();
+        rec = shared_ptr<bam1_t>(bam_init1(), &bam_destroy1);
+    }
+
+    void Clear() {
+        line->l = 0;
+        fields.clear();
+        tags_json.clear();
+        rg_id = -1;
+    }
 };
 
 class SamReader : public BackgroundProducer<SamItem> {
   public:
-    SamReader(samFile *sam, bam_hdr_t *hdr, int ringsize)
-        : BackgroundProducer<SamItem>(ringsize), sam_(sam), hdr_(hdr) {}
+    SamReader(samFile *sam, bam_hdr_t *hdr, const map<string, int> &readgroups, int ringsize)
+        : BackgroundProducer<SamItem>(ringsize), sam_(sam), hdr_(hdr), readgroups_(readgroups) {}
 
   private:
     samFile *sam_;
     bam_hdr_t *hdr_;
+    const map<string, int> &readgroups_;
+    OStringStream tagsbuf_;
 
     bool Produce(SamItem &it) override {
-        if (!it.line) {
-            it.line = kstringXX();
-        }
-        it.line->l = 0;
-        it.fields.clear();
-        if (!it.rec) {
-            it.rec = shared_ptr<bam1_t>(bam_init1(), &bam_destroy1);
-        }
+        it.Clear();
         int ret = sam_read1(sam_, hdr_, it.rec.get());
         if (ret == -1) {
             return false;
@@ -148,6 +158,10 @@ class SamReader : public BackgroundProducer<SamItem> {
             throw runtime_error("Corrupt SAM record; fields.size() = " +
                                 to_string(it.fields.size()));
         }
+        // formulate tags JSON on here on background thread
+        tagsbuf_.Clear();
+        it.rg_id = write_tags_json(readgroups_, it.fields, tagsbuf_);
+        it.tags_json = tagsbuf_.Get();
         return true;
     }
 };
@@ -314,10 +328,9 @@ int main(int argc, char *argv[]) {
 
         // stream bam1_t records
         progress &&cerr << "inserting reads...";
-        SamReader reader(sam.get(), hdr.get(), 64);
-        OStringStream cigarstr, tagsbuf;
+        SamReader reader(sam.get(), hdr.get(), readgroups, 64);
         while (reader.next()) {
-            SamItem &it = reader.item();
+            const SamItem &it = reader.item();
             bam1_t *rec = it.rec.get();
             auto &sam_fields = it.fields;
 
@@ -347,10 +360,8 @@ int main(int argc, char *argv[]) {
             if (rec->core.isize)
                 insert_read.bind(9, rec->core.isize); // tlen
 
-            tagsbuf.Clear();
-            int rg_id = write_tags_json(readgroups, sam_fields, tagsbuf);
-            if (rg_id >= 0 && rg_id < readgroups.size()) {
-                insert_read.bind(10, rg_id); // rg_id
+            if (it.rg_id >= 0 && it.rg_id < readgroups.size()) {
+                insert_read.bind(10, it.rg_id); // rg_id
             }
 
             insert_read.exec();
@@ -370,7 +381,7 @@ int main(int argc, char *argv[]) {
 
             insert_tags.reset();
             insert_tags.bind(1, rowid);
-            insert_tags.bind(2, tagsbuf.Get()); // tags_json
+            insert_tags.bindNoCopy(2, it.tags_json.c_str()); // tags_json
             insert_tags.exec();
         }
         progress &&cerr << reader.log() << endl;