MikkelSchubert / adapterremoval / #73

Committed 22 Mar 2025 10:19PM UTC coverage: 27.088% (-0.002%) from 27.09%

Build # #73

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

updates to formating and licensing headers (#95)

* use SPDX headers for licenses

This reduces verbosity and works around an issue with clang-format where
some formatting would not be applied due to the \***\ headers.

* set AllowAllArgumentsOnNextLine and InsertBraces

This results in more consistent formatting using clang-format

Run Details

18 of 61 new or added lines in 12 files covered. (29.51%)

343 existing lines in 3 files now uncovered.

2601 of 9602 relevant lines covered (27.09%)

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

MikkelSchubert / adapterremoval / #73

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