view release on metacpan or  search on metacpan

lib/Async/Simple/Pool.pm  view on Meta::CPAN

    $self->log( 'BUILD: task class', $task_class );

    Class::Load::load_class( $task_class );

    my @bad_task_param_names = grep !$task_class->can($_), keys %{ $self->task_params // {} };

    if ( scalar @bad_task_param_names ) {
        $self->log( 'BUILD: bad_task_param_names', \@bad_task_param_names );
        die 'Unknown params found: (' . join( ', ', @bad_task_param_names ) . ' )';
    };

    if ( scalar keys %{ $self->data } ) {
        $self->log( 'BUILD', '$self->process called' );
        $self->process;
    }
};


=head2 process

Main dispatcher of child tasks

    - writes data to tasks

    - checks for results


We don't care about all internal fails, dying or hang ons of your tasks.

If your task can do something bad, please write workaround for this case inside your "sub".

Will be called inside new() in case you pass data there.

=cut

sub process {
    my ( $self, $new_data ) = @_;

    if ( $new_data ) {
        $self->log( 'PROCESS: new data received', $new_data )  if $self->logger;

        my ( $data, $keys ) = _conv_data_to_internal( $self->data, $new_data );

        $self->log( 'PROCESS: new data parsed', $data )  if $self->logger;

        $self->data( $data );
        push @{ $self->queue_keys }, @$keys;
        push @{ $self->all_keys   }, @$keys;
    };

    my $break_on_busy = $self->break_on eq 'busy';
    my $break_on_run  = $self->break_on eq 'run';

    while( 1 ) {
        $self->log( 'PROCESS', 'internal cycle unless exit condition' )  if $self->logger;

        $self->read_tasks()  if grep $_->has_id, @{ $self->tasks };
        $self->write_tasks();

        if ( $break_on_busy ) {
            $self->log( 'PROCESS', 'internal cycle exit: all threads are busy' )  if $self->logger;
            last;
        }

        # Has not started data
        next if scalar @{ $self->queue_keys };

        if ( $break_on_run ) {
            $self->log( 'PROCESS', 'internal cycle exit: all tasks are started' )  if $self->logger;
            last;
        }

        # Has unprocessed data
        next if grep $_->has_id, @{ $self->tasks };

        $self->log( 'PROCESS', 'internal cycle exit: all tasks done' )  if $self->logger;
        last;
    };

    $self->log( 'PROCESS: finished', $self->results )  if $self->logger;

    return $self->results;
};


=head2 results

Internal.
Returns all results that already gathered
by default returns hash, where keys equal to indexes of source data list
and values are the results for data at these indexes.

=cut

sub results {
    my ( $self ) = @_;

    my $data = $self->data;

    my $is_list = $self->result_type =~ /list/;
    my $is_full = $self->result_type =~ /full/;

    my $results = $is_list ? [] : {};

    for ( @{ $self->all_keys } ) {
        my $result     = $data->{$_}->{result};
        my $has_result = exists $data->{$_}->{result};

        next  if !$is_full && !$has_result;

        $is_list
            ? ( push @$results, $result  )
            : ( $results->{$_} = $result );

        if ( $self->flush_data && $has_result ) {
            delete $data->{$_};
        };
    };

    $self->all_keys( [ keys %$data ] )  if $self->flush_data;