MikkelSchubert / adapterremoval / #80

Committed 06 Apr 2025 08:21PM UTC coverage: 27.836% (+0.1%) from 27.703%

Build # #80

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

improvements to utility functions (#104)

* move is_ascii_* helperfunction to strutils
* update join_text to support any containers
* use std::device_random to generate RNG seeds
* add function for getting an underlying enum value
* addition of / improvements to / tweaks to related tests

Run Details

20 of 30 new or added lines in 9 files covered. (66.67%)

1 existing line in 1 file now uncovered.

2719 of 9768 relevant lines covered (27.84%)

4127.62 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/src/fastq_io.cpp

// SPDX-License-Identifier: GPL-3.0-or-later
// SPDX-FileCopyrightText: 2015 Mikkel Schubert <mikkelsch@gmail.com>
#include "fastq_io.hpp"      // declarations
#include "debug.hpp"         // for AR_REQUIRE, AR_REQUIRE_SINGLE_THREAD
#include "errors.hpp"        // for io_error, gzip_error, fastq_error
#include "fastq.hpp"         // for fastq
#include "fastq_enc.hpp"     // for MATE_SEPARATOR
#include "output.hpp"        // for output_file
#include "simd.hpp"          // for size_t
#include "statistics.hpp"    // for fastq_statistics, fastq_stats_ptr, stat...
#include "strutils.hpp"      // for shell_escape, string_vec, ends_with
#include "userconfig.hpp"    // for userconfig
#include "utilities.hpp"     // for prng_seed
#include <algorithm>         // for max, min
#include <cerrno>            // for errno
#include <cstring>           // for size_t, memcpy
#include <isa-l/crc.h>       // for crc32_gzip_refl
#include <isa-l/igzip_lib.h> // for isal_zstream, isal_deflate_init, isal_d...
#include <libdeflate.h>      // for libdeflate_alloc_compressor, libdeflate...
#include <memory>            // for unique_ptr, make_unique, __shared_ptr_a...
#include <sstream>           // for basic_ostream, basic_ostringstream, ope...
#include <utility>           // for move, swap

namespace adapterremoval {

// Default bgzip header, as described in the SAM spec. v1.6 section 4.1.
// Includes 2 trailing placeholder bytes for total block size (BSIZE)
constexpr std::string_view BGZF_HEADER = {
  "\37\213\10\4\0\0\0\0\0\377\6\0\102\103\2\0\0\0",
  18
};

// Eof of file marker for bgzip files; see SAM spec. v1.6 section 4.1.2
constexpr std::string_view BGZF_EOF = {
  "\37\213\10\4\0\0\0\0\0\377\6\0\102\103\2\0\33\0\3\0\0\0\0\0\0\0\0\0",
  28,
};

//! Roughly how much extra space is needed for headers, CRC32, and ISIZE
constexpr size_t BGZF_META = BGZF_HEADER.size() + 4 + 4;

////////////////////////////////////////////////////////////////////////////////
// Helper function for isa-l

namespace {

//! The compression level used for block/stream compression with isa-l
constexpr size_t ISAL_COMPRESSION_LEVEL = 1;
//! The default buffer size for compression at level ISAL_COMPRESSION_LEVEL
constexpr size_t ISAL_BUFFER_SIZE = ISAL_DEF_LVL1_SMALL;

/**
 * ISA-l streaming is enabled only at compression level 1, since little
 * difference was observed between levels 1 to 3. However, it still offers a
 * faster compression (with a lower ratio) than libdeflate level 1.
 */
bool
is_isal_streaming_enabled(const output_file file, unsigned compression_level)
{
  switch (file.format) {
    case output_format::fastq:
    case output_format::sam:
    case output_format::bam:
    case output_format::ubam:
      return false;
    case output_format::fastq_gzip:
    case output_format::sam_gzip:
      return compression_level == ISAL_COMPRESSION_LEVEL;
    default:
      AR_FAIL("invalid output format");
  }
}

} // namespace

///////////////////////////////////////////////////////////////////////////////
// Implementations for 'read_fastq'

enum class read_fastq::file_type
{
  read_1,
  read_2,
  interleaved
};

namespace {

bool
read_record(joined_line_readers& reader, fastq_vec& chunk)
{
  // Line numbers change as we attempt to read the record, and potentially
  // points to the next record in the case of invalid qualities/nucleotides
  const auto line_number = reader.linenumber();

  try {
    chunk.emplace_back();
    auto& record = chunk.back();

    if (record.read_unsafe(reader)) {
      return true;
    } else {
      chunk.pop_back();
      return false;
    }
  } catch (const fastq_error& error) {
    std::ostringstream stream;
    stream << "Error reading FASTQ record from '" << reader.filename()
           << "' at line " << line_number << "; aborting:\n"
           << indent_lines(error.what());

    throw fastq_error(stream.str());
  }
}

const string_vec&
select_filenames(const userconfig& config, const read_fastq::file_type mode)
{
  switch (mode) {
    case read_fastq::file_type::read_1:
    case read_fastq::file_type::interleaved:
      return config.input_files_1;
    case read_fastq::file_type::read_2:
      return config.input_files_2;
    default:
      AR_FAIL("invalid read_fastq::file_type value");
  }
}

} // namespace

read_fastq::read_fastq(const userconfig& config,
                       const size_t next_step,
                       const read_fastq::file_type mode)
  : analytical_step(processing_order::ordered, "read_fastq")
  , m_reader(select_filenames(config, mode))
  , m_next_step(next_step)
  , m_mode(mode)
  , m_head(config.head)
  , m_mate_separator(config.mate_separator)
  , m_mate_separator_identified(config.mate_separator)
{
}

void
read_fastq::add_steps(scheduler& sch,
                      const userconfig& config,
                      size_t next_step)
{
  if (config.interleaved_input) {
    sch.add<read_fastq>(config, next_step, read_fastq::file_type::interleaved);
  } else if (config.paired_ended_mode) {
    next_step =
      sch.add<read_fastq>(config, next_step, read_fastq::file_type::read_2);
    sch.add<read_fastq>(config, next_step, read_fastq::file_type::read_1);
  } else {
    sch.add<read_fastq>(config, next_step, read_fastq::file_type::read_1);
  }
}

chunk_vec
read_fastq::process(chunk_ptr chunk)
{
  AR_REQUIRE_SINGLE_THREAD(m_lock);

  if (m_mode == file_type::read_1 || m_mode == file_type::interleaved) {
    // The scheduler only terminates when the first step stops returning chunks
    if (m_eof) {
      return {};
    }

    chunk = std::make_unique<analytical_chunk>();
  } else {
    AR_REQUIRE(!m_eof && chunk);
  }

  auto& reads_1 = chunk->reads_1;
  auto& reads_2 = chunk->reads_2;

  if (m_mode == file_type::read_1 || m_mode == file_type::interleaved) {
    if (m_mode == file_type::read_1) {
      read_single_end(reads_1);
    } else {
      read_interleaved(reads_1, reads_2);
    }
  } else if (m_mode == file_type::read_2) {
    read_single_end(reads_2);
  } else {
    AR_FAIL("invalid file_type value");
  }

  if (m_mode != file_type::read_1) {
    if (reads_1.size() != reads_2.size()) {
      throw fastq_error("Found unequal number of mate 1 and mate 2 reads; "
                        "input files may be truncated. Please fix before "
                        "continuing.");
    } else if (!m_mate_separator_identified) {
      AR_REQUIRE(reads_1.size() == reads_2.size());
      // Mate separators are identified using the first block, in order to
      // reduce the need for locking in the post-processing step

      // Attempt to determine the mate separator character
      m_mate_separator = fastq::guess_mate_separator(reads_1, reads_2);
      m_mate_separator_identified = true;
    }
  }

  // Head must be checked after the first loop, to produce at least one chunk
  m_eof |= !m_head;
  chunk->eof = m_eof;
  chunk->mate_separator = m_mate_separator;
  chunk->first = m_first;
  m_first = false;

  chunk_vec chunks;
  chunks.emplace_back(m_next_step, std::move(chunk));
  return chunks;
}

void
read_fastq::read_single_end(fastq_vec& reads)
{
  for (; reads.size() < INPUT_READS && m_head && !m_eof; m_head--) {
    m_eof = !read_record(m_reader, reads);
  }
}

void
read_fastq::read_interleaved(fastq_vec& reads_1, fastq_vec& reads_2)
{
  for (; reads_1.size() < INPUT_READS && m_head && !m_eof; m_head--) {
    m_eof = !read_record(m_reader, reads_1);
    read_record(m_reader, reads_2);
  }
}

void
read_fastq::finalize()
{
  AR_REQUIRE_SINGLE_THREAD(m_lock);
  AR_REQUIRE(m_eof);
}

///////////////////////////////////////////////////////////////////////////////
// Implementations for 'post_process_fastq'

post_process_fastq::post_process_fastq(const userconfig& config,
                                       size_t next_step,
                                       statistics& stats)
  : analytical_step(processing_order::unordered, "post_process_fastq")
  , m_statistics_1(stats.input_1)
  , m_statistics_2(stats.input_2)
  , m_next_step(next_step)
  , m_encoding(config.io_encoding)
  , m_timer(config.log_progress)
{
  AR_REQUIRE(m_statistics_1 && m_statistics_2);

  for (size_t i = 0; i < config.max_threads; ++i) {
    m_stats.emplace_back(config.report_sample_rate, prng_seed());
  }
}

chunk_vec
post_process_fastq::process(chunk_ptr chunk)
{
  AR_REQUIRE(chunk);
  auto& reads_1 = chunk->reads_1;
  auto& reads_2 = chunk->reads_2;

  auto stats = m_stats.acquire();

  AR_REQUIRE((reads_1.size() == reads_2.size()) || reads_2.empty());
  if (reads_2.empty()) {
    for (auto& read_1 : reads_1) {
      read_1.post_process(m_encoding);
      stats->stats_1.process(read_1);
    }
  } else {
    auto it_1 = reads_1.begin();
    auto it_2 = reads_2.begin();
    for (; it_1 != reads_1.end(); ++it_1, ++it_2) {
      fastq::normalize_paired_reads(*it_1, *it_2, chunk->mate_separator);

      it_1->post_process(m_encoding);
      stats->stats_1.process(*it_1);

      it_2->post_process(m_encoding);
      stats->stats_2.process(*it_2);
    }

    // fastq::normalize_paired_reads replaces the mate separator if present
    if (chunk->mate_separator) {
      chunk->mate_separator = MATE_SEPARATOR;
    }
  }

  m_stats.release(stats);

  {
    std::unique_lock<std::mutex> lock(m_timer_lock);
    m_timer.increment(reads_1.size() + reads_2.size());
  }

  chunk_vec chunks;
  chunks.emplace_back(m_next_step, std::move(chunk));

  return chunks;
}

void
post_process_fastq::finalize()
{
  AR_REQUIRE_SINGLE_THREAD(m_timer_lock);

  while (auto it = m_stats.try_acquire()) {
    *m_statistics_1 += it->stats_1;
    *m_statistics_2 += it->stats_2;
  }

  m_timer.finalize();
}

///////////////////////////////////////////////////////////////////////////////
// Implementations for 'split_fastq'

split_fastq::split_fastq(const userconfig& config,
                         const output_file& file,
                         size_t next_step)
  : analytical_step(processing_order::ordered, "split_fastq")
  , m_next_step(next_step)
  , m_isal_stream(is_isal_streaming_enabled(file, config.compression_level))
{
}

void
split_fastq::finalize()
{
  AR_REQUIRE_SINGLE_THREAD(m_lock);

  AR_REQUIRE(m_eof);
  AR_REQUIRE(m_buffer.capacity() == 0);
}

chunk_vec
split_fastq::process(const chunk_ptr chunk)
{
  AR_REQUIRE(chunk);
  AR_REQUIRE_SINGLE_THREAD(m_lock);
  AR_REQUIRE(!m_eof);
  m_eof = chunk->eof;

  chunk_vec chunks;
  for (const auto& src : chunk->buffers) {
    for (size_t src_offset = 0; src_offset < src.size();) {
      const auto n =
        std::min(src.size() - src_offset, BGZF_BLOCK_SIZE - m_buffer.size());
      m_buffer.append(src.data() + src_offset, n);

      src_offset += n;

      if (m_buffer.size() == BGZF_BLOCK_SIZE) {
        auto block = std::make_unique<analytical_chunk>();

        if (m_isal_stream) {
          m_isal_crc32 =
            crc32_gzip_refl(m_isal_crc32, m_buffer.data(), m_buffer.size());
        }

        block->uncompressed_size = m_buffer.size();
        block->buffers.emplace_back(std::move(m_buffer));

        chunks.emplace_back(m_next_step, std::move(block));

        m_buffer = buffer();
        m_buffer.reserve(BGZF_BLOCK_SIZE);
      }
    }
  }

  if (m_eof) {
    auto block = std::make_unique<analytical_chunk>();
    block->eof = true;

    if (m_isal_stream) {
      m_isal_crc32 =
        crc32_gzip_refl(m_isal_crc32, m_buffer.data(), m_buffer.size());
    }

    block->crc32 = m_isal_crc32;
    block->uncompressed_size = m_buffer.size();
    block->buffers.emplace_back(std::move(m_buffer));

    chunks.emplace_back(m_next_step, std::move(block));
  }

  return chunks;
}

///////////////////////////////////////////////////////////////////////////////
// Implementations for 'gzip_split_fastq'

namespace {

size_t
isal_deflate_block(buffer& input_buffer,
                   buffer& output_buffer,
                   const size_t output_offset,
                   const bool eof)
{
  isal_zstream stream{};
  isal_deflate_stateless_init(&stream);

  stream.flush = FULL_FLUSH;
  stream.end_of_stream = eof;

  stream.level = ISAL_COMPRESSION_LEVEL;
  stream.level_buf_size = ISAL_BUFFER_SIZE;
  buffer level_buffer{ stream.level_buf_size };
  stream.level_buf = level_buffer.data();

  stream.avail_in = input_buffer.size();
  stream.next_in = input_buffer.data();
  stream.next_out = output_buffer.data() + output_offset;
  stream.avail_out = output_buffer.size() - output_offset;

  switch (isal_deflate_stateless(&stream)) {
    case COMP_OK:
      break;
    case INVALID_FLUSH:
      throw gzip_error("isal_deflate_stateless: invalid flush");
    case ISAL_INVALID_LEVEL:
      throw gzip_error("isal_deflate_stateless: invalid level");
    case ISAL_INVALID_LEVEL_BUF:
      throw gzip_error("isal_deflate_stateless: invalid buffer size");
    default:
      throw gzip_error("isal_deflate_stateless: unexpected error");
  }

  // The easily compressible input should fit in a single output block
  AR_REQUIRE(stream.avail_in == 0);

  return stream.total_out;
}

} // namespace

gzip_split_fastq::gzip_split_fastq(const userconfig& config,
                                   const output_file& file,
                                   size_t next_step)
  : analytical_step(processing_order::unordered, "gzip_split_fastq")
  , m_config(config)
  , m_isal_stream(is_isal_streaming_enabled(file, config.compression_level))
  , m_format(file.format)
  , m_next_step(next_step)
{
}

chunk_vec
gzip_split_fastq::process(chunk_ptr chunk)
{
  AR_REQUIRE(chunk);
  AR_REQUIRE(chunk->buffers.size() == 1);

  buffer& input_buffer = chunk->buffers.front();
  buffer output_buffer;

  if (m_isal_stream) {
    output_buffer.resize(input_buffer.size());
    const auto output_size =
      isal_deflate_block(input_buffer, output_buffer, 0, chunk->eof);
    output_buffer.resize(output_size);
  } else {
    if (m_format == output_format::ubam || m_config.compression_level == 0) {
      output_buffer.reserve(input_buffer.size() + BGZF_META);
      output_buffer.append(BGZF_HEADER);
      output_buffer.append_u8(1); // BFINAL=1, BTYPE=00; see RFC1951
      output_buffer.append_u16(input_buffer.size());
      output_buffer.append_u16(~input_buffer.size());
      output_buffer.append(input_buffer);
    } else if (m_config.compression_level == ISAL_COMPRESSION_LEVEL) {
      output_buffer.reserve(input_buffer.size());
      output_buffer.append(BGZF_HEADER);
      output_buffer.resize(output_buffer.capacity());

      const auto output_size = isal_deflate_block(input_buffer,
                                                  output_buffer,
                                                  BGZF_HEADER.size(),
                                                  true);

      // Resize the buffer to the actually used size
      output_buffer.resize(output_size + BGZF_HEADER.size());
    } else {
      // Libdeflate compression levels 1 to 12 are mapped onto 2 to 13
      AR_REQUIRE(m_config.compression_level >= 2 &&
                 m_config.compression_level <= 13);
      auto* compressor =
        libdeflate_alloc_compressor(m_config.compression_level - 1);
      const auto output_bound =
        libdeflate_deflate_compress_bound(compressor, input_buffer.size());

      output_buffer.reserve(output_bound + BGZF_META);
      output_buffer.append(BGZF_HEADER);
      output_buffer.resize(output_buffer.capacity());

      const auto output_size =
        libdeflate_deflate_compress(compressor,
                                    input_buffer.data(),
                                    input_buffer.size(),
                                    output_buffer.data() + BGZF_HEADER.size(),
                                    output_buffer.size() - BGZF_HEADER.size());
      libdeflate_free_compressor(compressor);
      // The easily compressible input should fit in a single output block
      AR_REQUIRE(output_size);

      // Resize the buffer to the actually used size
      output_buffer.resize(output_size + BGZF_HEADER.size());
    }

    const auto input_crc32 =
      libdeflate_crc32(0, input_buffer.data(), input_buffer.size());
    output_buffer.append_u32(input_crc32);         // checksum of data
    output_buffer.append_u32(input_buffer.size()); // size of data

    AR_REQUIRE(output_buffer.size() <= BGZF_MAX_BLOCK_SIZE);
    // Write the final block size; -1 to fit 65536 in 16 bit
    output_buffer.put_u16(16, output_buffer.size() - 1);

    if (chunk->eof) {
      output_buffer.append(BGZF_EOF);
    }
  }

  // Enable reuse of the analytical_chunks
  std::swap(input_buffer, output_buffer);

  chunk_vec chunks;
  chunks.emplace_back(m_next_step, std::move(chunk));

  return chunks;
}

///////////////////////////////////////////////////////////////////////////////
// Implementations for 'write_fastq'

write_fastq::write_fastq(const userconfig& config, const output_file& file)
  // Allow disk IO and writing to STDOUT at the same time
  : analytical_step(processing_order::ordered_io, "write_fastq")
  , m_output(file.name)
  , m_isal_stream(is_isal_streaming_enabled(file, config.compression_level))
{
  if (m_isal_stream) {
    buffer level_buf(ISAL_BUFFER_SIZE);
    buffer output_buf(OUTPUT_BLOCK_SIZE);

    struct isal_zstream stream = {};
    struct isal_gzip_header header = {};

    isal_gzip_header_init(&header);
    isal_deflate_init(&stream);
    stream.avail_in = 0;
    stream.flush = NO_FLUSH;
    stream.level = ISAL_COMPRESSION_LEVEL;
    stream.level_buf = level_buf.data();
    stream.level_buf_size = level_buf.size();
    stream.gzip_flag = IGZIP_GZIP_NO_HDR;
    stream.next_out = output_buf.data();
    stream.avail_out = output_buf.size();

    const auto ret = isal_write_gzip_header(&stream, &header);
    AR_REQUIRE(ret == 0, "buffer was not large enough for gzip header");

    output_buf.resize(stream.total_out);

    m_output.write(output_buf);
  }
}

chunk_vec
write_fastq::process(chunk_ptr chunk)
{
  AR_REQUIRE(chunk);
  AR_REQUIRE_SINGLE_THREAD(m_lock);
  AR_REQUIRE(!m_eof);

  try {
    m_eof = chunk->eof;
    m_uncompressed_bytes += chunk->uncompressed_size;

    const auto mode = (m_eof && !m_isal_stream) ? flush::on : flush::off;

    m_output.write(chunk->buffers, mode);

    if (m_eof && m_isal_stream) {
      buffer trailer;
      trailer.append_u32(chunk->crc32);
      trailer.append_u32(m_uncompressed_bytes);

      m_output.write(trailer, flush::on);
    }
  } catch (const std::ios_base::failure&) {
    std::ostringstream msg;
    msg << "Error writing to FASTQ file " << shell_escape(m_output.filename());

    throw io_error(msg.str(), errno);
  }

  return {};
}

void
write_fastq::finalize()
{
  AR_REQUIRE_SINGLE_THREAD(m_lock);
  AR_REQUIRE(m_eof);

  // Close file to trigger any exceptions due to badbit / failbit
  try {
    m_output.close();
  } catch (const std::ios_base::failure&) {
    std::ostringstream msg;
    msg << "Error closing FASTQ file " << shell_escape(m_output.filename());

    throw io_error(msg.str(), errno);
  }
}

} // namespace adapterremoval

1	// SPDX-License-Identifier: GPL-3.0-or-later
2	// SPDX-FileCopyrightText: 2015 Mikkel Schubert <mikkelsch@gmail.com>
3	#include "fastq_io.hpp" // declarations
4	#include "debug.hpp" // for AR_REQUIRE, AR_REQUIRE_SINGLE_THREAD
5	#include "errors.hpp" // for io_error, gzip_error, fastq_error
6	#include "fastq.hpp" // for fastq
7	#include "fastq_enc.hpp" // for MATE_SEPARATOR
8	#include "output.hpp" // for output_file
9	#include "simd.hpp" // for size_t
10	#include "statistics.hpp" // for fastq_statistics, fastq_stats_ptr, stat...
11	#include "strutils.hpp" // for shell_escape, string_vec, ends_with
12	#include "userconfig.hpp" // for userconfig
13	#include "utilities.hpp" // for prng_seed
14	#include <algorithm> // for max, min
15	#include <cerrno> // for errno
16	#include <cstring> // for size_t, memcpy
17	#include <isa-l/crc.h> // for crc32_gzip_refl
18	#include <isa-l/igzip_lib.h> // for isal_zstream, isal_deflate_init, isal_d...
19	#include <libdeflate.h> // for libdeflate_alloc_compressor, libdeflate...
20	#include <memory> // for unique_ptr, make_unique, __shared_ptr_a...
21	#include <sstream> // for basic_ostream, basic_ostringstream, ope...
22	#include <utility> // for move, swap
23
24	namespace adapterremoval {
25
26	// Default bgzip header, as described in the SAM spec. v1.6 section 4.1.
27	// Includes 2 trailing placeholder bytes for total block size (BSIZE)
28	constexpr std::string_view BGZF_HEADER = {
29	"\37\213\10\4\0\0\0\0\0\377\6\0\102\103\2\0\0\0",
30	18
31	};
32
33	// Eof of file marker for bgzip files; see SAM spec. v1.6 section 4.1.2
34	constexpr std::string_view BGZF_EOF = {
35	"\37\213\10\4\0\0\0\0\0\377\6\0\102\103\2\0\33\0\3\0\0\0\0\0\0\0\0\0",
36	28,
37	};
38
39	//! Roughly how much extra space is needed for headers, CRC32, and ISIZE
40	constexpr size_t BGZF_META = BGZF_HEADER.size() + 4 + 4;
41
42	////////////////////////////////////////////////////////////////////////////////
43	// Helper function for isa-l
44
45	namespace {
46
47	//! The compression level used for block/stream compression with isa-l
48	constexpr size_t ISAL_COMPRESSION_LEVEL = 1;
49	//! The default buffer size for compression at level ISAL_COMPRESSION_LEVEL
50	constexpr size_t ISAL_BUFFER_SIZE = ISAL_DEF_LVL1_SMALL;
51
52	/**
53	* ISA-l streaming is enabled only at compression level 1, since little
54	* difference was observed between levels 1 to 3. However, it still offers a
55	* faster compression (with a lower ratio) than libdeflate level 1.
56	*/
57	bool
58	is_isal_streaming_enabled(const output_file file, unsigned compression_level)	×
59	{
60	switch (file.format) {	×
61	case output_format::fastq:
62	case output_format::sam:
63	case output_format::bam:
64	case output_format::ubam:
65	return false;
66	case output_format::fastq_gzip:	×
67	case output_format::sam_gzip:	×
68	return compression_level == ISAL_COMPRESSION_LEVEL;	×
69	default:	×
70	AR_FAIL("invalid output format");	×
71	}
72	}
73
74	} // namespace
75
76	///////////////////////////////////////////////////////////////////////////////
77	// Implementations for 'read_fastq'
78
79	enum class read_fastq::file_type
80	{
81	read_1,
82	read_2,
83	interleaved
84	};
85
86	namespace {
87
88	bool
89	read_record(joined_line_readers& reader, fastq_vec& chunk)	×
90	{
91	// Line numbers change as we attempt to read the record, and potentially
92	// points to the next record in the case of invalid qualities/nucleotides
93	const auto line_number = reader.linenumber();	×
94
95	try {	×
96	chunk.emplace_back();	×
97	auto& record = chunk.back();	×
98
99	if (record.read_unsafe(reader)) {	×
100	return true;
101	} else {
102	chunk.pop_back();	×
103	return false;	×
104	}
105	} catch (const fastq_error& error) {	×
106	std::ostringstream stream;	×
107	stream << "Error reading FASTQ record from '" << reader.filename()	×
108	<< "' at line " << line_number << "; aborting:\n"	×
109	<< indent_lines(error.what());	×
110
111	throw fastq_error(stream.str());	×
112	}	×
113	}
114
115	const string_vec&
116	select_filenames(const userconfig& config, const read_fastq::file_type mode)	×
117	{
118	switch (mode) {	×
119	case read_fastq::file_type::read_1:	×
120	case read_fastq::file_type::interleaved:	×
121	return config.input_files_1;	×
122	case read_fastq::file_type::read_2:	×
123	return config.input_files_2;	×
124	default:	×
125	AR_FAIL("invalid read_fastq::file_type value");	×
126	}
127	}
128
129	} // namespace
130
131	read_fastq::read_fastq(const userconfig& config,	×
132	const size_t next_step,
133	const read_fastq::file_type mode)	×
134	: analytical_step(processing_order::ordered, "read_fastq")
135	, m_reader(select_filenames(config, mode))	×
136	, m_next_step(next_step)	×
137	, m_mode(mode)	×
138	, m_head(config.head)	×
139	, m_mate_separator(config.mate_separator)	×
140	, m_mate_separator_identified(config.mate_separator)	×
141	{
142	}
143
144	void
145	read_fastq::add_steps(scheduler& sch,	×
146	const userconfig& config,
147	size_t next_step)
148	{
149	if (config.interleaved_input) {	×
150	sch.add<read_fastq>(config, next_step, read_fastq::file_type::interleaved);	×
151	} else if (config.paired_ended_mode) {	×
152	next_step =	×
153	sch.add<read_fastq>(config, next_step, read_fastq::file_type::read_2);	×
154	sch.add<read_fastq>(config, next_step, read_fastq::file_type::read_1);	×
155	} else {
156	sch.add<read_fastq>(config, next_step, read_fastq::file_type::read_1);	×
157	}
158	}
159
160	chunk_vec
161	read_fastq::process(chunk_ptr chunk)	×
162	{
163	AR_REQUIRE_SINGLE_THREAD(m_lock);	×
164
165	if (m_mode == file_type::read_1 \|\| m_mode == file_type::interleaved) {	×
166	// The scheduler only terminates when the first step stops returning chunks
167	if (m_eof) {	×
168	return {};	×
169	}
170
171	chunk = std::make_unique<analytical_chunk>();	×
172	} else {
173	AR_REQUIRE(!m_eof && chunk);	×
174	}
175
176	auto& reads_1 = chunk->reads_1;	×
177	auto& reads_2 = chunk->reads_2;	×
178
179	if (m_mode == file_type::read_1 \|\| m_mode == file_type::interleaved) {	×
180	if (m_mode == file_type::read_1) {	×
181	read_single_end(reads_1);	×
182	} else {
183	read_interleaved(reads_1, reads_2);	×
184	}
185	} else if (m_mode == file_type::read_2) {	×
186	read_single_end(reads_2);	×
187	} else {
188	AR_FAIL("invalid file_type value");	×
189	}
190
191	if (m_mode != file_type::read_1) {	×
192	if (reads_1.size() != reads_2.size()) {	×
193	throw fastq_error("Found unequal number of mate 1 and mate 2 reads; "	×
194	"input files may be truncated. Please fix before "
195	"continuing.");	×
196	} else if (!m_mate_separator_identified) {	×
197	AR_REQUIRE(reads_1.size() == reads_2.size());	×
198	// Mate separators are identified using the first block, in order to
199	// reduce the need for locking in the post-processing step
200
201	// Attempt to determine the mate separator character
202	m_mate_separator = fastq::guess_mate_separator(reads_1, reads_2);	×
203	m_mate_separator_identified = true;	×
204	}
205	}
206
207	// Head must be checked after the first loop, to produce at least one chunk
208	m_eof \|= !m_head;	×
209	chunk->eof = m_eof;	×
210	chunk->mate_separator = m_mate_separator;	×
211	chunk->first = m_first;	×
212	m_first = false;	×
213
214	chunk_vec chunks;	×
215	chunks.emplace_back(m_next_step, std::move(chunk));	×
216	return chunks;	×
217	}
218
219	void
220	read_fastq::read_single_end(fastq_vec& reads)	×
221	{
222	for (; reads.size() < INPUT_READS && m_head && !m_eof; m_head--) {	×
223	m_eof = !read_record(m_reader, reads);	×
224	}
225	}
226
227	void
228	read_fastq::read_interleaved(fastq_vec& reads_1, fastq_vec& reads_2)	×
229	{
230	for (; reads_1.size() < INPUT_READS && m_head && !m_eof; m_head--) {	×
231	m_eof = !read_record(m_reader, reads_1);	×
232	read_record(m_reader, reads_2);	×
233	}
234	}
235
236	void
237	read_fastq::finalize()	×
238	{
239	AR_REQUIRE_SINGLE_THREAD(m_lock);	×
240	AR_REQUIRE(m_eof);	×
241	}
242
243	///////////////////////////////////////////////////////////////////////////////
244	// Implementations for 'post_process_fastq'
245
246	post_process_fastq::post_process_fastq(const userconfig& config,	×
247	size_t next_step,
248	statistics& stats)	×
249	: analytical_step(processing_order::unordered, "post_process_fastq")
250	, m_statistics_1(stats.input_1)	×
251	, m_statistics_2(stats.input_2)	×
252	, m_next_step(next_step)	×
253	, m_encoding(config.io_encoding)	×
254	, m_timer(config.log_progress)	×
255	{
256	AR_REQUIRE(m_statistics_1 && m_statistics_2);	×
257
258	for (size_t i = 0; i < config.max_threads; ++i) {	×
NEW 259	m_stats.emplace_back(config.report_sample_rate, prng_seed());	×
260	}
261	}
262
263	chunk_vec
264	post_process_fastq::process(chunk_ptr chunk)	×
265	{
266	AR_REQUIRE(chunk);	×
267	auto& reads_1 = chunk->reads_1;	×
268	auto& reads_2 = chunk->reads_2;	×
269
270	auto stats = m_stats.acquire();	×
271
272	AR_REQUIRE((reads_1.size() == reads_2.size()) \|\| reads_2.empty());	×
273	if (reads_2.empty()) {	×
274	for (auto& read_1 : reads_1) {	×
275	read_1.post_process(m_encoding);	×
276	stats->stats_1.process(read_1);	×
277	}
278	} else {
279	auto it_1 = reads_1.begin();	×
280	auto it_2 = reads_2.begin();	×
281	for (; it_1 != reads_1.end(); ++it_1, ++it_2) {	×
282	fastq::normalize_paired_reads(it_1, it_2, chunk->mate_separator);	×
283
284	it_1->post_process(m_encoding);	×
285	stats->stats_1.process(*it_1);	×
286
287	it_2->post_process(m_encoding);	×
288	stats->stats_2.process(*it_2);	×
289	}
290
291	// fastq::normalize_paired_reads replaces the mate separator if present
292	if (chunk->mate_separator) {	×
293	chunk->mate_separator = MATE_SEPARATOR;	×
294	}
295	}
296
297	m_stats.release(stats);	×
298
299	{	×
300	std::unique_lock<std::mutex> lock(m_timer_lock);	×
301	m_timer.increment(reads_1.size() + reads_2.size());	×
302	}
303
304	chunk_vec chunks;	×
305	chunks.emplace_back(m_next_step, std::move(chunk));	×
306
307	return chunks;	×
308	}
309
310	void
311	post_process_fastq::finalize()	×
312	{
313	AR_REQUIRE_SINGLE_THREAD(m_timer_lock);	×
314
315	while (auto it = m_stats.try_acquire()) {	×
316	*m_statistics_1 += it->stats_1;	×
317	*m_statistics_2 += it->stats_2;	×
318	}
319
320	m_timer.finalize();	×
321	}
322
323	///////////////////////////////////////////////////////////////////////////////
324	// Implementations for 'split_fastq'
325
326	split_fastq::split_fastq(const userconfig& config,	×
327	const output_file& file,
328	size_t next_step)	×
329	: analytical_step(processing_order::ordered, "split_fastq")
330	, m_next_step(next_step)	×
331	, m_isal_stream(is_isal_streaming_enabled(file, config.compression_level))	×
332	{
333	}
334
335	void
336	split_fastq::finalize()	×
337	{
338	AR_REQUIRE_SINGLE_THREAD(m_lock);	×
339
340	AR_REQUIRE(m_eof);	×
341	AR_REQUIRE(m_buffer.capacity() == 0);	×
342	}
343
344	chunk_vec
345	split_fastq::process(const chunk_ptr chunk)	×
346	{
347	AR_REQUIRE(chunk);	×
348	AR_REQUIRE_SINGLE_THREAD(m_lock);	×
349	AR_REQUIRE(!m_eof);	×
350	m_eof = chunk->eof;	×
351
352	chunk_vec chunks;	×
353	for (const auto& src : chunk->buffers) {	×
354	for (size_t src_offset = 0; src_offset < src.size();) {	×
355	const auto n =	×
356	std::min(src.size() - src_offset, BGZF_BLOCK_SIZE - m_buffer.size());	×
357	m_buffer.append(src.data() + src_offset, n);	×
358
359	src_offset += n;	×
360
361	if (m_buffer.size() == BGZF_BLOCK_SIZE) {	×
362	auto block = std::make_unique<analytical_chunk>();	×
363
364	if (m_isal_stream) {	×
365	m_isal_crc32 =	×
366	crc32_gzip_refl(m_isal_crc32, m_buffer.data(), m_buffer.size());	×
367	}
368
369	block->uncompressed_size = m_buffer.size();	×
370	block->buffers.emplace_back(std::move(m_buffer));	×
371
372	chunks.emplace_back(m_next_step, std::move(block));	×
373
374	m_buffer = buffer();	×
375	m_buffer.reserve(BGZF_BLOCK_SIZE);	×
376	}
377	}
378	}
379
380	if (m_eof) {	×
381	auto block = std::make_unique<analytical_chunk>();	×
382	block->eof = true;	×
383
384	if (m_isal_stream) {	×
385	m_isal_crc32 =	×
386	crc32_gzip_refl(m_isal_crc32, m_buffer.data(), m_buffer.size());	×
387	}
388
389	block->crc32 = m_isal_crc32;	×
390	block->uncompressed_size = m_buffer.size();	×
391	block->buffers.emplace_back(std::move(m_buffer));	×
392
393	chunks.emplace_back(m_next_step, std::move(block));	×
394	}
395
396	return chunks;	×
397	}
398
399	///////////////////////////////////////////////////////////////////////////////
400	// Implementations for 'gzip_split_fastq'
401
402	namespace {
403
404	size_t
405	isal_deflate_block(buffer& input_buffer,	×
406	buffer& output_buffer,
407	const size_t output_offset,
408	const bool eof)
409	{
410	isal_zstream stream{};	×
411	isal_deflate_stateless_init(&stream);	×
412
413	stream.flush = FULL_FLUSH;	×
414	stream.end_of_stream = eof;	×
415
416	stream.level = ISAL_COMPRESSION_LEVEL;	×
417	stream.level_buf_size = ISAL_BUFFER_SIZE;	×
418	buffer level_buffer{ stream.level_buf_size };	×
419	stream.level_buf = level_buffer.data();	×
420
421	stream.avail_in = input_buffer.size();	×
422	stream.next_in = input_buffer.data();	×
423	stream.next_out = output_buffer.data() + output_offset;	×
424	stream.avail_out = output_buffer.size() - output_offset;	×
425
426	switch (isal_deflate_stateless(&stream)) {	×
427	case COMP_OK:	×
428	break;	×
429	case INVALID_FLUSH:	×
430	throw gzip_error("isal_deflate_stateless: invalid flush");	×
431	case ISAL_INVALID_LEVEL:	×
432	throw gzip_error("isal_deflate_stateless: invalid level");	×
433	case ISAL_INVALID_LEVEL_BUF:	×
434	throw gzip_error("isal_deflate_stateless: invalid buffer size");	×
435	default:	×
436	throw gzip_error("isal_deflate_stateless: unexpected error");	×
437	}
438
439	// The easily compressible input should fit in a single output block
440	AR_REQUIRE(stream.avail_in == 0);	×
441
442	return stream.total_out;	×
443	}
444
445	} // namespace
446
447	gzip_split_fastq::gzip_split_fastq(const userconfig& config,	×
448	const output_file& file,
449	size_t next_step)	×
450	: analytical_step(processing_order::unordered, "gzip_split_fastq")
451	, m_config(config)	×
452	, m_isal_stream(is_isal_streaming_enabled(file, config.compression_level))	×
453	, m_format(file.format)	×
454	, m_next_step(next_step)	×
455	{
456	}
457
458	chunk_vec
459	gzip_split_fastq::process(chunk_ptr chunk)	×
460	{
461	AR_REQUIRE(chunk);	×
462	AR_REQUIRE(chunk->buffers.size() == 1);	×
463
464	buffer& input_buffer = chunk->buffers.front();	×
465	buffer output_buffer;	×
466
467	if (m_isal_stream) {	×
468	output_buffer.resize(input_buffer.size());	×
469	const auto output_size =	×
470	isal_deflate_block(input_buffer, output_buffer, 0, chunk->eof);	×
471	output_buffer.resize(output_size);	×
472	} else {
473	if (m_format == output_format::ubam \|\| m_config.compression_level == 0) {	×
474	output_buffer.reserve(input_buffer.size() + BGZF_META);	×
475	output_buffer.append(BGZF_HEADER);	×
476	output_buffer.append_u8(1); // BFINAL=1, BTYPE=00; see RFC1951	×
477	output_buffer.append_u16(input_buffer.size());	×
478	output_buffer.append_u16(~input_buffer.size());	×
479	output_buffer.append(input_buffer);	×
480	} else if (m_config.compression_level == ISAL_COMPRESSION_LEVEL) {	×
481	output_buffer.reserve(input_buffer.size());	×
482	output_buffer.append(BGZF_HEADER);	×
483	output_buffer.resize(output_buffer.capacity());	×
484
485	const auto output_size = isal_deflate_block(input_buffer,	×
486	output_buffer,
487	BGZF_HEADER.size(),
488	true);
489
490	// Resize the buffer to the actually used size
491	output_buffer.resize(output_size + BGZF_HEADER.size());	×
492	} else {
493	// Libdeflate compression levels 1 to 12 are mapped onto 2 to 13
494	AR_REQUIRE(m_config.compression_level >= 2 &&	×
495	m_config.compression_level <= 13);
496	auto* compressor =	×
497	libdeflate_alloc_compressor(m_config.compression_level - 1);	×
498	const auto output_bound =	×
499	libdeflate_deflate_compress_bound(compressor, input_buffer.size());	×
500
501	output_buffer.reserve(output_bound + BGZF_META);	×
502	output_buffer.append(BGZF_HEADER);	×
503	output_buffer.resize(output_buffer.capacity());	×
504
505	const auto output_size =	×
506	libdeflate_deflate_compress(compressor,	×
507	input_buffer.data(),	×
508	input_buffer.size(),	×
509	output_buffer.data() + BGZF_HEADER.size(),	×
510	output_buffer.size() - BGZF_HEADER.size());	×
511	libdeflate_free_compressor(compressor);	×
512	// The easily compressible input should fit in a single output block
513	AR_REQUIRE(output_size);	×
514
515	// Resize the buffer to the actually used size
516	output_buffer.resize(output_size + BGZF_HEADER.size());	×
517	}
518
519	const auto input_crc32 =	×
520	libdeflate_crc32(0, input_buffer.data(), input_buffer.size());	×
521	output_buffer.append_u32(input_crc32); // checksum of data	×
522	output_buffer.append_u32(input_buffer.size()); // size of data	×
523
524	AR_REQUIRE(output_buffer.size() <= BGZF_MAX_BLOCK_SIZE);	×
525	// Write the final block size; -1 to fit 65536 in 16 bit
526	output_buffer.put_u16(16, output_buffer.size() - 1);	×
527
528	if (chunk->eof) {	×
529	output_buffer.append(BGZF_EOF);	×
530	}
531	}
532
533	// Enable reuse of the analytical_chunks
534	std::swap(input_buffer, output_buffer);	×
535
536	chunk_vec chunks;	×
537	chunks.emplace_back(m_next_step, std::move(chunk));	×
538
539	return chunks;	×
540	}
541
542	///////////////////////////////////////////////////////////////////////////////
543	// Implementations for 'write_fastq'
544
545	write_fastq::write_fastq(const userconfig& config, const output_file& file)	×
546	// Allow disk IO and writing to STDOUT at the same time
547	: analytical_step(processing_order::ordered_io, "write_fastq")
548	, m_output(file.name)	×
549	, m_isal_stream(is_isal_streaming_enabled(file, config.compression_level))	×
550	{
551	if (m_isal_stream) {	×
552	buffer level_buf(ISAL_BUFFER_SIZE);	×
553	buffer output_buf(OUTPUT_BLOCK_SIZE);	×
554
555	struct isal_zstream stream = {};	×
556	struct isal_gzip_header header = {};	×
557
558	isal_gzip_header_init(&header);	×
559	isal_deflate_init(&stream);	×
560	stream.avail_in = 0;	×
561	stream.flush = NO_FLUSH;	×
562	stream.level = ISAL_COMPRESSION_LEVEL;	×
563	stream.level_buf = level_buf.data();	×
564	stream.level_buf_size = level_buf.size();	×
565	stream.gzip_flag = IGZIP_GZIP_NO_HDR;	×
566	stream.next_out = output_buf.data();	×
567	stream.avail_out = output_buf.size();	×
568
569	const auto ret = isal_write_gzip_header(&stream, &header);	×
570	AR_REQUIRE(ret == 0, "buffer was not large enough for gzip header");	×
571
572	output_buf.resize(stream.total_out);	×
573
574	m_output.write(output_buf);	×
575	}
576	}
577
578	chunk_vec
579	write_fastq::process(chunk_ptr chunk)	×
580	{
581	AR_REQUIRE(chunk);	×
582	AR_REQUIRE_SINGLE_THREAD(m_lock);	×
583	AR_REQUIRE(!m_eof);	×
584
585	try {	×
586	m_eof = chunk->eof;	×
587	m_uncompressed_bytes += chunk->uncompressed_size;	×
588
589	const auto mode = (m_eof && !m_isal_stream) ? flush::on : flush::off;	×
590
591	m_output.write(chunk->buffers, mode);	×
592
593	if (m_eof && m_isal_stream) {	×
594	buffer trailer;	×
595	trailer.append_u32(chunk->crc32);	×
596	trailer.append_u32(m_uncompressed_bytes);	×
597
598	m_output.write(trailer, flush::on);	×
599	}
600	} catch (const std::ios_base::failure&) {	×
601	std::ostringstream msg;	×
602	msg << "Error writing to FASTQ file " << shell_escape(m_output.filename());	×
603
604	throw io_error(msg.str(), errno);	×
605	}	×
606
607	return {};	×
608	}
609
610	void
611	write_fastq::finalize()	×
612	{
613	AR_REQUIRE_SINGLE_THREAD(m_lock);	×
614	AR_REQUIRE(m_eof);	×
615
616	// Close file to trigger any exceptions due to badbit / failbit
617	try {	×
618	m_output.close();	×
619	} catch (const std::ios_base::failure&) {	×
620	std::ostringstream msg;	×
621	msg << "Error closing FASTQ file " << shell_escape(m_output.filename());	×
622
623	throw io_error(msg.str(), errno);	×
624	}	×
625	}
626
627	} // namespace adapterremoval

MikkelSchubert / adapterremoval / #80

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous