<commit_before>
=head1 NAME

VRPipe::DataSource - define pipeline input from arbitrary sources

=head1 SYNOPSIS

*** more documentation to come

=head1 DESCRIPTION

A DataSource defines the inputs you want to supply to one or more
L<VRPipe::PipelineSetup>s. There are a number of different types of DataSource.
Each type understands a different kind of source. For example the 'fofn' type
understands text files that list file paths. The 'vrtrack' type understands the
contents of a VRTrack database.

Each type has one or more methods, and each method can have options. For
example, you could use the 'fofn' type with the 'all' method and no options,
and give it a 'source' of a file of filenames. That DataSource would then
define that each file in your fofn should be an independent input to any
PipelineSetup using this DataSource.

For example, consider a L<VRPipe::Pipeline> with 2 steps. The first step takes
a txt file as input, copies it, and appends a line of text to the copy, then
outputs the new text file. The second step takes the output of the first, makes
a copy of it, and removes the first line before outputting this altered
version.

Now let's say we make a DataSource of type 'fofn', method 'all', source
'/abs/that/to/my.fofn', where the contents of /abs/that/to/my.fofn are:
    
    /abs/path/to/file.1
    /abs/path/to/file.2

/abs/path/to/file.1 is a text file 10 lines long, and /abs/path/to/file.2 is a
text file 5 lines long.

This DataSource will be comprised of 2 L<VRPipe::DataElement>s, one for
/abs/path/to/file.1, and the other for /abs/path/to/file.2

Finally, we make a PipelineSetup that uses the previously mentioned Pipeline
and DataSource. What will happen is this:

In a certain working directory, /abs/path/to/file.1 is input into the first
step. At the same time, in a different working directory, /abs/path/to/file.2
is input into the first step. In parallel the code of the first step us run on
the inputs, and in the first working directory a file 11 lines long is made,
and in the second working directory a file of 6 lines long is made.

Soon after the 11 line long file is made, it is passed to the second step.
Independently (but probably around the same time in this example), soon after
the 6 line long file is made, it is passed to the second step. In a third
working directory the 11 line long file is processed to generate a 10 line long
file, and (probably in parallel) in a fourth working directory the 6 line long
file is processed to generate a 5 line long file.

At this point the pipeline is complete, and you can find the 2 output files in
the third and fourth working directories and see which input file they
correspond to by using the B<vrpipe-output> script.

*** more documentation to come

=head1 AUTHOR

<NAME> <<EMAIL>>.

=head1 COPYRIGHT AND LICENSE

Copyright (c) 2011-2012 Genome Research Limited.

This file is part of VRPipe.

VRPipe 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 L<http://www.gnu.org/licenses/>.

=cut

use VRPipe::Base;

class VRPipe::DataSource extends VRPipe::Persistent {
    use DateTime;
    use VRPipe::DataSourceFactory;
    
    has 'type' => (
        is     => 'rw',
        isa    => Varchar [64],
        traits => ['VRPipe::Persistent::Attributes'],
        is_key => 1
    );
    
    has 'method' => (
        is     => 'rw',
        isa    => Varchar [64],
        traits => ['VRPipe::Persistent::Attributes'],
        is_key => 1
    );
    
    has 'source' => (
        is     => 'rw',
        isa    => Text,
        coerce => 1,
        traits => ['VRPipe::Persistent::Attributes'],
        is_key => 1
    );
    
    has 'options' => (
        is                   => 'rw',
        isa                  => 'HashRef',
        traits               => ['VRPipe::Persistent::Attributes'],
        default              => sub { {} },
        allow_key_to_default => 1,
        is_key               => 1
    );
    
    has '_lock' => (
        is          => 'rw',
        isa         => Datetime,
        traits      => ['VRPipe::Persistent::Attributes'],
        is_nullable => 1
    );
    
    has '_changed_marker' => (
        is          => 'rw',
        isa         => Varchar [255],
        traits      => ['VRPipe::Persistent::Attributes'],
        is_nullable => 1
    );
    
    has '_source_instance' => (
        is      => 'rw',
        isa     => 'Defined',
        lazy    => 1,
        builder => '_build_source',
        handles => [qw(description source_description method_description)]
    );
    
    method _build_source {
        my $changed_marker = $self->_changed_marker;
        return VRPipe::DataSourceFactory->create(
            $self->type,
            {
                method  => $self->method,
                source  => $self->source,
                options => $self->options,
                $changed_marker ? ('_changed_marker' => $changed_marker) : (),
                '_datasource_id' => $self->id
            }
        );
    }
    
    __PACKAGE__->make_persistent(has_many => [elements => 'VRPipe::DataElement']);
    
    around elements {
        my (undef, undef, $status_messages) = @_;
        # we don't return $self->$orig because that returns all associated
        # elements; we need to first create all elements, and then only return
        # elements that are still "current" (haven't been deleted in the source)
        $self->_prepare_elements_and_states($status_messages);
        return VRPipe::DataElement->search_paged({ datasource => $self->id, withdrawn => 0 });
    }
    
    method incomplete_element_states (VRPipe::PipelineSetup $setup, Bool :$include_withdrawn = 0, Bool :$prepare = 1, Bool :$only_not_started = 0) {
        if ($prepare) {
            $self->_prepare_elements_and_states;
        }
        
        my $pipeline  = $setup->pipeline;
        my $num_steps = $pipeline->step_members;
        
        return VRPipe::DataElementState->search_paged({ pipelinesetup => $setup->id, completed_steps => $only_not_started ? 0 : { '<', $num_steps }, $include_withdrawn ? () : ('dataelement.withdrawn' => 0) }, { prefetch => 'dataelement' });
    }
    
    # for critical columns that get updated by multiple processes, ensure we
    # always get the latest value from db
    around _lock (Datetime $datetime?) {
        if ($datetime) {
            return $self->$orig($datetime);
        }
        $self->reselect_values_from_db;
        return $self->$orig;
    }
    
    around _changed_marker (Str $marker?) {
        if ($marker) {
            return $self->$orig($marker);
        }
        $self->reselect_values_from_db;
        return $self->$orig;
    }
    
    method _prepare_elements_and_states (Bool $status_messages = 0) {
        my $source = $self->_source_instance || return;
        
        my @setup_ids = VRPipe::PipelineSetup->get_column_values('id', { datasource => $self->id });
        
        if ($source->_has_changed) {
            # we must not go through and update the dataelements more than
            # once simultaneously, and we must not return elements in a
            # partially updated state, so we lock/block at this point
            my $block    = 0;
            my $continue = 1;
            do {
                my $transaction = sub {
                    $self->lock_row($self);
                    my $lock_time = $self->_lock;
                    
                    # check that the process that got the lock is still running,
                    # otherwise ignore the lock
                    if ($lock_time) {
                        my $elapsed = time() - $lock_time->epoch;
                        if ($elapsed > 60) {
                            undef $lock_time;
                        }
                    }
                    
                    # if some other process has a recent lock, we will block
                    if ($lock_time) {
                        $block = 1;
                        warn "another process is updating dataelements, please wait...\n" if $status_messages;
                        sleep(2);
                        return;<commit_msg>turn status_messages on in _prepare_elements_and_states() when versboe<commit_after>
=head1 NAME

VRPipe::DataSource - define pipeline input from arbitrary sources

=head1 SYNOPSIS

*** more documentation to come

=head1 DESCRIPTION

A DataSource defines the inputs you want to supply to one or more
L<VRPipe::PipelineSetup>s. There are a number of different types of DataSource.
Each type understands a different kind of source. For example the 'fofn' type
understands text files that list file paths. The 'vrtrack' type understands the
contents of a VRTrack database.

Each type has one or more methods, and each method can have options. For
example, you could use the 'fofn' type with the 'all' method and no options,
and give it a 'source' of a file of filenames. That DataSource would then
define that each file in your fofn should be an independent input to any
PipelineSetup using this DataSource.

For example, consider a L<VRPipe::Pipeline> with 2 steps. The first step takes
a txt file as input, copies it, and appends a line of text to the copy, then
outputs the new text file. The second step takes the output of the first, makes
a copy of it, and removes the first line before outputting this altered
version.

Now let's say we make a DataSource of type 'fofn', method 'all', source
'/abs/that/to/my.fofn', where the contents of /abs/that/to/my.fofn are:
    
    /abs/path/to/file.1
    /abs/path/to/file.2

/abs/path/to/file.1 is a text file 10 lines long, and /abs/path/to/file.2 is a
text file 5 lines long.

This DataSource will be comprised of 2 L<VRPipe::DataElement>s, one for
/abs/path/to/file.1, and the other for /abs/path/to/file.2

Finally, we make a PipelineSetup that uses the previously mentioned Pipeline
and DataSource. What will happen is this:

In a certain working directory, /abs/path/to/file.1 is input into the first
step. At the same time, in a different working directory, /abs/path/to/file.2
is input into the first step. In parallel the code of the first step us run on
the inputs, and in the first working directory a file 11 lines long is made,
and in the second working directory a file of 6 lines long is made.

Soon after the 11 line long file is made, it is passed to the second step.
Independently (but probably around the same time in this example), soon after
the 6 line long file is made, it is passed to the second step. In a third
working directory the 11 line long file is processed to generate a 10 line long
file, and (probably in parallel) in a fourth working directory the 6 line long
file is processed to generate a 5 line long file.

At this point the pipeline is complete, and you can find the 2 output files in
the third and fourth working directories and see which input file they
correspond to by using the B<vrpipe-output> script.

*** more documentation to come

=head1 AUTHOR

<NAME> <<EMAIL>>.

=head1 COPYRIGHT AND LICENSE

Copyright (c) 2011-2012 Genome Research Limited.

This file is part of VRPipe.

VRPipe 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 L<http://www.gnu.org/licenses/>.

=cut

use VRPipe::Base;

class VRPipe::DataSource extends VRPipe::Persistent {
    use DateTime;
    use VRPipe::DataSourceFactory;
    
    has 'type' => (
        is     => 'rw',
        isa    => Varchar [64],
        traits => ['VRPipe::Persistent::Attributes'],
        is_key => 1
    );
    
    has 'method' => (
        is     => 'rw',
        isa    => Varchar [64],
        traits => ['VRPipe::Persistent::Attributes'],
        is_key => 1
    );
    
    has 'source' => (
        is     => 'rw',
        isa    => Text,
        coerce => 1,
        traits => ['VRPipe::Persistent::Attributes'],
        is_key => 1
    );
    
    has 'options' => (
        is                   => 'rw',
        isa                  => 'HashRef',
        traits               => ['VRPipe::Persistent::Attributes'],
        default              => sub { {} },
        allow_key_to_default => 1,
        is_key               => 1
    );
    
    has '_lock' => (
        is          => 'rw',
        isa         => Datetime,
        traits      => ['VRPipe::Persistent::Attributes'],
        is_nullable => 1
    );
    
    has '_changed_marker' => (
        is          => 'rw',
        isa         => Varchar [255],
        traits      => ['VRPipe::Persistent::Attributes'],
        is_nullable => 1
    );
    
    has '_source_instance' => (
        is      => 'rw',
        isa     => 'Defined',
        lazy    => 1,
        builder => '_build_source',
        handles => [qw(description source_description method_description)]
    );
    
    method _build_source {
        my $changed_marker = $self->_changed_marker;
        return VRPipe::DataSourceFactory->create(
            $self->type,
            {
                method  => $self->method,
                source  => $self->source,
                options => $self->options,
                $changed_marker ? ('_changed_marker' => $changed_marker) : (),
                '_datasource_id' => $self->id
            }
        );
    }
    
    __PACKAGE__->make_persistent(has_many => [elements => 'VRPipe::DataElement']);
    
    around elements {
        my (undef, undef, $status_messages) = @_;
        # we don't return $self->$orig because that returns all associated
        # elements; we need to first create all elements, and then only return
        # elements that are still "current" (haven't been deleted in the source)
        $self->_prepare_elements_and_states($status_messages);
        return VRPipe::DataElement->search_paged({ datasource => $self->id, withdrawn => 0 });
    }
    
    method incomplete_element_states (VRPipe::PipelineSetup $setup, Bool :$include_withdrawn = 0, Bool :$prepare = 1, Bool :$only_not_started = 0) {
        if ($prepare) {
            $self->_prepare_elements_and_states;
        }
        
        my $pipeline  = $setup->pipeline;
        my $num_steps = $pipeline->step_members;
        
        return VRPipe::DataElementState->search_paged({ pipelinesetup => $setup->id, completed_steps => $only_not_started ? 0 : { '<', $num_steps }, $include_withdrawn ? () : ('dataelement.withdrawn' => 0) }, { prefetch => 'dataelement' });
    }
    
    # for critical columns that get updated by multiple processes, ensure we
    # always get the latest value from db
    around _lock (Datetime $datetime?) {
        if ($datetime) {
            return $self->$orig($datetime);
        }
        $self->reselect_values_from_db;
        return $self->$orig;
    }
    
    around _changed_marker (Str $marker?) {
        if ($marker) {
            return $self->$orig($marker);
        }
        $self->reselect_values_from_db;
        return $self->$orig;
    }
    
    method _prepare_elements_and_states (Bool $status_messages?) {
        my $source = $self->_source_instance || return;
        unless (defined $status_messages) {
            $status_messages = $self->verbose > 0 ? 1 : 0;
        }
        
        my @setup_ids = VRPipe::PipelineSetup->get_column_values('id', { datasource => $self->id });
        
        if ($source->_has_changed) {
            # we must not go through and update the dataelements more than
            # once simultaneously, and we must not return elements in a
            # partially updated state, so we lock/block at this point
            my $block    = 0;
            my $continue = 1;
            do {
                my $transaction = sub {
                    $self->lock_row($self);
                    my $lock_time = $self->_lock;
                    
                    # check that the process that got the lock is still running,
                    # otherwise ignore the lock
                    if ($lock_time) {
                        my $elapsed = time() - $lock_time->epoch;
                        if ($elapsed > 60) {
                            undef $lock_time;
                        }
                    }
                    
                    # if some other process has a recent lock, we will block
                    if ($lock_time) {
                        $block = 1;
                        warn "another process is updating dataelements, please wait...\n" if $status_messages;
                        sleep(2);
                        return;
{'fraction_non_alphanumeric': 0.09277819139797262, 'fraction_numerical': 0.008017868392417388, 'mean_word_length': 2.191154970760234, 'pattern_counts': {'":': 0, '<': 21, '<?xml version=': 0, '>': 189, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '95302', 'n_tokens_mistral': 4929, 'n_tokens_neox': 4703, 'n_words': 1852}