Added a sharing mode to the VCFV4 class

This determines how the class reuses records, i.e. either manually or automatically. In manual mode, records must be explicitly returned to the class for reuse, otherwise the record's shared_ptr will destroy it when the use counter reaches 0. In automatic mode, the VCFV4 class keeps a copy of the record's shared_ptr and automatically reuses it when the use counter reaches 1.

Author: alancleary

libtiledbvcf/src/vcf/vcf_v4.cc

@@ -25,13 +25,13 @@
  */
 
 #include "vcf/vcf_v4.h"
-#include "utils/logger_public.h"
 
 namespace tiledb {
 namespace vcf {
 
-VCFV4::VCFV4()
-    : open_(false)
+VCFV4::VCFV4(SharingMode mode)
+    : mode_(mode)
+    , open_(false)
     , inited_(false)
     , max_record_buffer_size_(10000)
     , hdr_(nullptr)
@@ -156,7 +156,13 @@ void VCFV4::pop_record() {
 }
 
 void VCFV4::return_record(SafeSharedBCFRec& record) {
-  record_queue_pool_.emplace(std::move(record));
+  if (mode_ == SharingMode::MANUAL) {
+    record_queue_pool_.emplace(std::move(record));
+  } else {
+    LOG_ERROR(
+        "VCFV4::return_record records cannot be manually returned for mode {}",
+        modeToString(mode_));
+  }
 }
 
 std::string VCFV4::contig_name(bcf1_t* const r) const {
@@ -285,6 +291,21 @@ bool VCFV4::seek(const std::string& contig_name, uint32_t pos) {
   return !record_queue_.empty();
 }
 
+void VCFV4::create_record(SafeBCFRec& tmp_r) {
+  SafeSharedBCFRec r(bcf_dup(tmp_r.get()), bcf_destroy);
+  bcf_unpack(r.get(), BCF_UN_ALL);
+  record_queue_.emplace(std::move(r));
+}
+
+void VCFV4::reuse_record(SafeBCFRec& tmp_r) {
+  SafeSharedBCFRec r = record_queue_pool_.front();
+  record_queue_pool_.pop();
+  // Use `bcf_copy` to destroy (free) the stale data to prevent memory leaks
+  bcf_copy(r.get(), tmp_r.get());
+  bcf_unpack(r.get(), BCF_UN_ALL);
+  record_queue_.emplace(std::move(r));
+}
+
 void VCFV4::read_records() {
   if (!record_queue_.empty())
     std::queue<SafeSharedBCFRec>().swap(record_queue_);
@@ -301,19 +322,21 @@ void VCFV4::read_records() {
       break;
     }
 
-    if (!record_queue_pool_.empty()) {
-      // Pop a stale record for re-use. Note that `bcf_copy`
-      // destroys (frees) the stale data to prevent a memory
-      // leak.
-      SafeSharedBCFRec r = record_queue_pool_.front();
-      record_queue_pool_.pop();
-      bcf_copy(r.get(), tmp_r.get());
-      bcf_unpack(r.get(), BCF_UN_ALL);
-      record_queue_.emplace(std::move(r));
+    if (!record_queue_pool_.empty() && mode_ == SharingMode::MANUAL) {
+      reuse_record(tmp_r);
+    } else if (!record_queue_pool_.empty() && mode_ == SharingMode::AUTOMATIC) {
+      // Check if the record at the front of the pool is stale, i.e. the pool
+      // has the only copy
+      SafeSharedBCFRec& r = record_queue_pool_.front();
+      if (r.use_count() == 1) {
+        reuse_record(tmp_r);
+      } else {
+        // Resuse a stale record
+        create_record(tmp_r);
+      }
+      record_queue_pool_.push(record_queue_.front());
     } else {
-      SafeSharedBCFRec r(bcf_dup(tmp_r.get()), bcf_destroy);
-      bcf_unpack(r.get(), BCF_UN_ALL);
-      record_queue_.emplace(std::move(r));
+      create_record(tmp_r);
     }
     record_buffer_size += sizeof(bcf1_t) + record_queue_.front()->shared.m +
                           record_queue_.front()->indiv.m;
@@ -328,6 +351,7 @@ void VCFV4::read_records() {
 }
 
 void VCFV4::swap(VCFV4& other) {
+  std::swap(mode_, other.mode_);
   std::swap(open_, other.open_);
   std::swap(path_, other.path_);
   std::swap(index_path_, other.index_path_);

libtiledbvcf/src/vcf/vcf_v4.h

@@ -41,6 +41,7 @@
 #include <unordered_map>
 #include <vector>
 
+#include "utils/logger_public.h"
 #include "utils/utils.h"
 #include "vcf/htslib_value.h"
 #include "vcf/region.h"
@@ -54,7 +55,25 @@ namespace vcf {
  */
 class VCFV4 {
  public:
-  VCFV4();
+  enum SharingMode {
+    MANUAL,
+    AUTOMATIC,
+  };
+
+  /**
+   * Constructor that determines how `SafeSharedBCFRec` records are managed. In
+   * MANUAL mode, records must be returned using `return_record()` to prevent
+   * them from being automatically destroyed when the `SafeSharedBCFRec` counter
+   * reaches 0. In AUTOMATIC mode, `VCFV4` keeps a copy of each
+   * `SafeSharedBCFRec` and releases it for reuse when the counter reachers 1.
+   *
+   * Note that AUTOMATIC mode uses a lazy algorithm for releasing records, so
+   * it is only appropriate when records can be released in roughly the same
+   * order that they were popped.
+   *
+   * @param mode The mode to use for managing SafeSharedBCFRec pointers
+   */
+  VCFV4(SharingMode mode = SharingMode::MANUAL);
   ~VCFV4();
 
   VCFV4(VCFV4&& other) = delete;
@@ -135,6 +154,15 @@ class VCFV4 {
   void set_max_record_buff_size(uint64_t max_record_buffer_size);
 
  private:
+  inline const std::string modeToString(SharingMode mode) {
+    switch (mode) {
+      case MANUAL:
+        return "MANUAL";
+      case AUTOMATIC:
+        return "AUTOMATIC";
+    }
+    LOG_ERROR("VCFV4::modeToString {} is not a valid SharingMode", mode);
+  }
   /** BCF/VCF iterator wrapper. */
   class Iter {
    public:
@@ -166,6 +194,9 @@ class VCFV4 {
     kstring_t tmps_ = {0, 0, nullptr};
   };
 
+  /** The mode used to manage the SafeSharedBCFRec pointers. */
+  SharingMode mode_;
+
   /** True if the file is open. */
   bool open_;
 
@@ -199,6 +230,21 @@ class VCFV4 {
   /** The HTS index handle, if the index format is HTS. */
   hts_idx_t* index_hts_;
 
+  /**
+   * Creates a new record and adds it to the `record_queue_`.
+   *
+   * @param tmp_r The record to wrap
+   */
+  void create_record(SafeBCFRec& tmp_r);
+
+  /**
+   * Pops the first record from `record_queue_pool_` and reuses it to add a
+   * record to `record_queue_`.
+   *
+   * @param tmp_r The record to wrap
+   */
+  void reuse_record(SafeBCFRec& tmp_r);
+
   /** Reads records into the record buffer using `iter_`. */
   void read_records();