view release on metacpan or  search on metacpan

lib/Acme/Parataxis.pm  view on Meta::CPAN

        affix $l, 'get_job_coro_id',                                       [Int],                          Int;
        affix $l, 'free_job_slot',                                         [Int],                          Void;
        affix $l, 'get_thread_pool_size',                                  [],                             Int;
        affix $l, 'get_max_thread_pool_size',                              [],                             Int;
        affix $l, 'get_fd_setsize',                                        [],                             Int;
        affix $l, 'set_max_threads',                                       [Int],                          Void;
        affix $l, 'get_max_fibers',                                        [],                             Int;
        affix $l, 'set_max_fibers',                                        [Int],                          Void;
        affix $l, [ 'set_preempt_threshold' => '_set_preempt_threshold' ], [LongLong],                     Void;
        affix $l, [ 'maybe_yield' => '_maybe_yield' ],                     [],                             Pointer [SV];
        affix $l, 'get_preempt_count',                                     [],                             LongLong;

        # Capture the main interpreter context
        init_system();
        if ( $^O eq 'MSWin32' ) {
            my $perl_dll = $Config{libperl};
            $perl_dll =~ s/^lib//;
            $perl_dll =~ s/\.a$//;
            $perl_dll .= '.' . $Config{so};
            my $p = Affix::load_library($perl_dll);
            affix $p, 'win32_get_osfhandle', [Int], LongLong;
        }
    }

    BEGIN {
        my $lib_name = ( $^O eq 'MSWin32' ? '' : 'lib' ) . 'parataxis.' . $Config{so};
        my @paths;
        push @paths, File::Spec->catfile( dirname(__FILE__), $lib_name );
        push @paths, File::Spec->catfile( dirname(__FILE__), '..',   'arch', 'auto',      'Acme', 'Parataxis', $lib_name );
        push @paths, File::Spec->catfile( dirname(__FILE__), '..',   '..',   'arch',      'auto', 'Acme', 'Parataxis', $lib_name );
        push @paths, File::Spec->catfile( dirname(__FILE__), 'auto', 'Acme', 'Parataxis', $lib_name );
        for my $inc (@INC) {
            next if ref $inc;
            push @paths, File::Spec->catfile( $inc, 'auto', 'Acme', 'Parataxis', $lib_name );
        }
        for my $path (@paths) {
            if ( -e $path ) {
                $lib = Affix::load_library($path);
                last if $lib;
            }
        }
        die 'Could not find or load ' . $lib_name unless $lib;
        _bind_functions($lib);
    }

    # API aliases and wrappers
    sub fiber : prototype(&) ($code) { spawn( __PACKAGE__, $code ) }
    sub async : prototype(&) ($code) { return run($code) }

    # Index at which user arguments start in @_: 0 when it is a plain call, 1 when $_[0] is a self (package name or
    # blessed object). This mirrors the shift/unshift invocant dance used elsewhere, but reads $_[0] instead of
    # shifting, so calling it never reifies the caller's pad @_. A reified @_ left behind in a parked frame trips
    # pp_entersub's assert(!AvREAL(av))/assert(AvFILLp(av) == -1) when a different fiber later enters the same sub at
    # the same depth.
    sub _arg_offset {
        my $self = $_[0];
        return ( defined $self && ( ( ref $self || $self ) eq __PACKAGE__ || ( builtin::blessed($self) && $self->isa(__PACKAGE__) ) ) ) ? 1 : 0;
    }

    # Aggregate a list of futures into one that resolves when every input has: the result is an arrayref of each
    # input's result, in input order, or the stored error of the first input to fail (reject-fast, Promise.all style).
    # Inputs are not cancelled on a rejection; they are independent and keep running, and their later results are
    # discarded. Zero futures resolve immediately with [] (matching Promise.all([])). An already-resolved input
    # fires its on_ready callback inline, so a wait_all of ready inputs resolves inline (no fiber needed) and any
    # mixed ready/pending input resolves the moment the last straggler lands. Inputs must be Acme::Parataxis::Future
    # objects else croaks. Also callable as Acme::Parataxis->wait_all(...).
    sub wait_all {
        my $o       = _arg_offset( $_[0] );
        my @futures = @_[ $o .. $#_ ];
        @_ = ();
        croak 'wait_all() requires Acme::Parataxis::Future inputs'
            if grep { !( defined $_ && ref $_ && $_->isa('Acme::Parataxis::Future') ) } @futures;
        require Acme::Parataxis::Future;
        my $all = Acme::Parataxis::Future->new;
        if ( @futures == 0 ) {
            $all->set_result( [] );
            return $all;
        }
        my @results;
        my $pending = scalar @futures;
        for my $i ( 0 .. $#futures ) {
            my $idx = $i;
            my $fut = $futures[$i];
            $fut->on_ready(
                sub ($f) {
                    return if $all->is_ready;
                    if ( defined $f->error ) {
                        $all->set_error( $f->error );    # reject-fast: the first failure wins, copied wholesale
                        return;
                    }
                    $results[$idx] = $f->result;
                    $all->set_result( \@results ) if --$pending == 0;
                }
            );
        }
        return $all;
    }

    # Aggregate a list of futures into one that settles with a wholesale copy of the first input to settle - success
    # or failure (Promise.race). The losers are untouched and keep running; their eventual results are dropped by the
    # is_ready guard. An already-ready input settles the aggregate inline. At least one input is required (an empty
    # race would never settle) and each input must be an Acme::Parataxis::Future else croaks. Also callable as
    # Acme::Parataxis->wait_any(...).
    sub wait_any {
        my $o       = _arg_offset( $_[0] );
        my @futures = @_[ $o .. $#_ ];
        @_ = ();
        croak 'wait_any() requires at least one Acme::Parataxis::Future input' unless @futures;
        croak 'wait_any() requires Acme::Parataxis::Future inputs'
            if grep { !( defined $_ && ref $_ && $_->isa('Acme::Parataxis::Future') ) } @futures;
        require Acme::Parataxis::Future;
        my $any = Acme::Parataxis::Future->new;
        for my $f (@futures) {
            $f->on_ready(
                sub ($f) {
                    return if $any->is_ready;
                    if   ( defined $f->error ) { $any->set_error( $f->error ) }     # wholesale copy of the failure
                    else                       { $any->set_result( $f->result ) }
                }
            );
        }