MikkelSchubert / adapterremoval / #45

Committed 20 Sep 2024 06:49PM UTC coverage: 26.244% (-49.2%) from 75.443%

Build # #45

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

attempt to fix coveralls run

Run Details

2458 of 9366 relevant lines covered (26.24%)

4362.23 hits per line

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

0.0

/src/fastq_io.cpp

/*************************************************************************\
 * AdapterRemoval - cleaning next-generation sequencing reads            *
 *                                                                       *
 * Copyright (C) 2015 by Mikkel Schubert - mikkelsch@gmail.com           *
 *                                                                       *
 * This program is free software: you can redistribute it and/or modify  *
 * it under the terms of the GNU General Public License as published by  *
 * the Free Software Foundation, either version 3 of the License, or     *
 * (at your option) any later version.                                   *
 *                                                                       *
 * This program is distributed in the hope that it will be useful,       *
 * but WITHOUT ANY WARRANTY; without even the implied warranty of        *
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the         *
 * GNU General Public License for more details.                          *
 *                                                                       *
 * You should have received a copy of the GNU General Public License     *
 * along with this program.  If not, see <http://www.gnu.org/licenses/>. *
\*************************************************************************/

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

MikkelSchubert / adapterremoval / #45

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