<commit_before>/*******************************************************************************
 * thrill/api/read_lines.hpp
 *
 * Part of Project Thrill.
 *
 * Copyright (C) 2015 <NAME> <<EMAIL>>
 * Copyright (C) 2015 <NAME> <<EMAIL>>
 *
 * This file has no license. Only Chuck Norris can compile it.
 ******************************************************************************/

#pragma once
#ifndef THRILL_API_READ_LINES_HEADER
#define THRILL_API_READ_LINES_HEADER

#include <thrill/api/dia.hpp>
#include <thrill/api/source_node.hpp>
#include <thrill/common/defines.hpp>
#include <thrill/common/logger.hpp>
#include <thrill/common/math.hpp>
#include <thrill/common/string.hpp>
#include <thrill/core/file_io.hpp>
#include <thrill/net/buffer_builder.hpp>

#include <string>
#include <utility>
#include <vector>

namespace thrill {
namespace api {

//! \addtogroup api Interface
//! \{

/*!
 * A DIANode which performs a line-based Read operation. Read reads a file from
 * the file system and emits it as a DIA.
 */
class ReadLinesNode : public SourceNode<std::string>
{
public:
    using Super = SourceNode<std::string>;
    using Super::context_;

    using FileSizePair = std::pair<std::string, size_t>;

    static const bool debug = false;

    /*!
     * Constructor for a ReadLinesNode. Sets the Context
     * and file path.
     *
     * \param ctx Reference to Context, which holds references to data and
     * network.
     *
     * \param path Path of the input file(s)
     */
    ReadLinesNode(Context& ctx,
                  const std::string& path,
                  StatsNode* stats_node)
        : Super(ctx, { }, stats_node),
          path_(path)
    {
		LOG << "Opening read notes for " << path_;
		
        filesize_prefix_ = core::GlobFileSizePrefixSum(path_);

        for (auto file : filesize_prefix_) {
            if (core::IsCompressed(file.first)) {
                contains_compressed_file_ = true;
                break;
            }
        }
    }

    void PushData() final {
        if (contains_compressed_file_) {
            InputLineIteratorCompressed it = InputLineIteratorCompressed(
                filesize_prefix_, context_.my_rank(), context_.num_workers());

            // Hook Read
            while (it.HasNext()) {
                this->PushItem(it.Next());
            }
        }
        else {
            InputLineIteratorUncompressed it = InputLineIteratorUncompressed(
                filesize_prefix_, context_.my_rank(), context_.num_workers());

            // Hook Read
            while (it.HasNext()) {
                this->PushItem(it.Next());
            }
        }
    }

    void Dispose() final { }

    /*!
     * Produces an 'empty' function stack, which only contains the identity
     * emitter function.
     *
     * \return Empty function stack
     */
    auto ProduceStack() {
        return FunctionStack<std::string>();
    }

private:
    //! True, if at least one input file is compressed.
    bool contains_compressed_file_ = false;
    //! Path of the input file.
    std::string path_;

    std::vector<std::pair<std::string, size_t> > filesize_prefix_;

    // REVIEW(an): this is useless, you never use the inheritance.  But, you
    // actually SHOULD use it! for all member fields and methods that are in
    // common. But NOT for virtual functions. Remove the virtuals. Find out what
    // functions the methods below have in common and make them functions of the
    // superclass.
    class InputLineIterator
    {
    public:
        static const bool debug = false;
        const size_t read_size = 2 * 1024 * 1024;

        virtual ~InputLineIterator() { }
    };

    //! InputLineIterator gives you access to lines of a file
    class InputLineIteratorUncompressed : public InputLineIterator
    {
    public:
        using Base = InputLineIterator;

        //! Creates an instance of iterator that reads file line based
        InputLineIteratorUncompressed(
            std::vector<FileSizePair> files,
            size_t my_id,
            size_t num_workers)
            : files_(files),
              my_id_(my_id),
              num_workers_(num_workers) {

            input_size_ = files[NumFiles()].second;
            // Go to start of 'local part'.
            auto my_start_and_end = common::CalculateLocalRange(input_size_, num_workers_, my_id_);

            size_t my_start = std::get<0>(my_start_and_end);
            my_end_ = std::get<1>(my_start_and_end);

            while (files_[current_file_ + 1].second <= my_start) {
                current_file_++;
            }
            if (my_start < my_end_) {
                LOG << "Opening file " << current_file_;
                c_file_ = OpenFile(files_[current_file_].first);
            }
            else {
                LOG << "my_start : " << my_start << " my_end_: " << my_end_;
                return;
            }

            // find offset in current file:
            // offset = start - sum of previous file sizes
            offset_ = lseek(c_file_, my_start - files_[current_file_].second, SEEK_CUR);
            buffer_.Reserve(Base::read_size);
            ssize_t buffer_size = read(c_file_, buffer_.data(), Base::read_size);
            buffer_.set_size(buffer_size);
			current_ = buffer_.begin();

            if (offset_ != 0) {
                bool found_n = false;

                // find next newline, discard all previous data as previous worker already covers it
                while (!found_n) {
                    for (auto it = current_; it != buffer_.end(); it++) {
                        if (THRILL_UNLIKELY(*it == '\n')) {
                            current_ = it + 1;
                            found_n = true;
                            break;<commit_msg>apply further code cleanup from compressed to uncompressed<commit_after>/*******************************************************************************
 * thrill/api/read_lines.hpp
 *
 * Part of Project Thrill.
 *
 * Copyright (C) 2015 <NAME> <<EMAIL>>
 * Copyright (C) 2015 <NAME> <<EMAIL>.net>
 *
 * This file has no license. Only Chuck Norris can compile it.
 ******************************************************************************/

#pragma once
#ifndef THRILL_API_READ_LINES_HEADER
#define THRILL_API_READ_LINES_HEADER

#include <thrill/api/dia.hpp>
#include <thrill/api/source_node.hpp>
#include <thrill/common/defines.hpp>
#include <thrill/common/logger.hpp>
#include <thrill/common/math.hpp>
#include <thrill/common/string.hpp>
#include <thrill/core/file_io.hpp>
#include <thrill/net/buffer_builder.hpp>

#include <string>
#include <utility>
#include <vector>

namespace thrill {
namespace api {

//! \addtogroup api Interface
//! \{

/*!
 * A DIANode which performs a line-based Read operation. Read reads a file from
 * the file system and emits it as a DIA.
 */
class ReadLinesNode : public SourceNode<std::string>
{
public:
    using Super = SourceNode<std::string>;
    using Super::context_;

    using FileSizePair = std::pair<std::string, size_t>;

    static const bool debug = false;

    /*!
     * Constructor for a ReadLinesNode. Sets the Context
     * and file path.
     *
     * \param ctx Reference to Context, which holds references to data and
     * network.
     *
     * \param path Path of the input file(s)
     */
    ReadLinesNode(Context& ctx,
                  const std::string& path,
                  StatsNode* stats_node)
        : Super(ctx, { }, stats_node),
          path_(path)
    {
		LOG << "Opening read notes for " << path_;
		
        filesize_prefix_ = core::GlobFileSizePrefixSum(path_);

        for (auto file : filesize_prefix_) {
            if (core::IsCompressed(file.first)) {
                contains_compressed_file_ = true;
                break;
            }
        }
    }

    void PushData() final {
        if (contains_compressed_file_) {
            InputLineIteratorCompressed it = InputLineIteratorCompressed(
                filesize_prefix_, context_.my_rank(), context_.num_workers());

            // Hook Read
            while (it.HasNext()) {
                this->PushItem(it.Next());
            }
        }
        else {
            InputLineIteratorUncompressed it = InputLineIteratorUncompressed(
                filesize_prefix_, context_.my_rank(), context_.num_workers());

            // Hook Read
            while (it.HasNext()) {
                this->PushItem(it.Next());
            }
        }
    }

    void Dispose() final { }

    /*!
     * Produces an 'empty' function stack, which only contains the identity
     * emitter function.
     *
     * \return Empty function stack
     */
    auto ProduceStack() {
        return FunctionStack<std::string>();
    }

private:
    //! True, if at least one input file is compressed.
    bool contains_compressed_file_ = false;
    //! Path of the input file.
    std::string path_;

    std::vector<std::pair<std::string, size_t> > filesize_prefix_;

    // REVIEW(an): this is useless, you never use the inheritance.  But, you
    // actually SHOULD use it! for all member fields and methods that are in
    // common. But NOT for virtual functions. Remove the virtuals. Find out what
    // functions the methods below have in common and make them functions of the
    // superclass.
    class InputLineIterator
    {
    public:
        static const bool debug = false;
        const size_t read_size = 2 * 1024 * 1024;

        virtual ~InputLineIterator() { }
    };

    //! InputLineIterator gives you access to lines of a file
    class InputLineIteratorUncompressed : public InputLineIterator
    {
    public:
        using Base = InputLineIterator;

        //! Creates an instance of iterator that reads file line based
        InputLineIteratorUncompressed(
            std::vector<FileSizePair> files,
            size_t my_id,
            size_t num_workers)
            : files_(files),
              my_id_(my_id),
              num_workers_(num_workers) {

            input_size_ = files[NumFiles()].second;
            // Go to start of 'local part'.
            size_t my_start;
            std::tie(my_start, my_end_) = common::CalculateLocalRange(input_size_, num_workers_, my_id_);

            while (files_[current_file_ + 1].second <= my_start) {
                current_file_++;
            }
            if (my_start < my_end_) {
                LOG << "Opening file " << current_file_;
                c_file_ = OpenFile(files_[current_file_].first);
            }
            else {
                LOG << "my_start : " << my_start << " my_end_: " << my_end_;
                return;
            }

            // find offset in current file:
            // offset = start - sum of previous file sizes
            offset_ = lseek(c_file_, my_start - files_[current_file_].second, SEEK_CUR);
            buffer_.Reserve(Base::read_size);
            ssize_t buffer_size = read(c_file_, buffer_.data(), Base::read_size);
            buffer_.set_size(buffer_size);
			current_ = buffer_.begin();

            if (offset_ != 0) {
                bool found_n = false;

                // find next newline, discard all previous data as previous worker already covers it
                while (!found_n) {
                    for (auto it = current_; it != buffer_.end(); it++) {
                        if (THRILL_UNLIKELY(*it == '\n')) {
                            current_ = it + 1;
                            found_n = true;
                            break;
{'fraction_non_alphanumeric': 0.1263988895636332, 'fraction_numerical': 0.0036436193285330092, 'mean_word_length': 2.098091910776673, 'pattern_counts': {'":': 0, '<': 89, '<?xml version=': 0, '>': 57, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3712321', 'n_tokens_mistral': 3326, 'n_tokens_neox': 3124, 'n_words': 879}