view release on metacpan or  search on metacpan

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

        my $fileno = fileno($fh);
        die 'Not a valid filehandle' unless defined $fileno;
        my $handle = $^O eq 'MSWin32' ? win32_get_osfhandle($fileno) : $fileno;
        _submit_job( 3, $handle, $timeout );
        return yield('WAITING');
    }

    sub maybe_yield {
        my $invocant = shift;
        if ( !defined $invocant ||
            ( ( ref $invocant || $invocant ) ne __PACKAGE__ && !( builtin::blessed($invocant) && $invocant->isa(__PACKAGE__) ) ) ) {
            unshift @_, $invocant if defined $invocant;
        }
        my $result = Acme::Parataxis::_maybe_yield();
        return unless defined $result;
        return wantarray ? @$result : $result->[-1];
    }
    sub tid            { get_os_thread_id_export() }
    sub current_fid    { get_current_parataxis_id() }
    sub root           { state $root //= Acme::Parataxis::Root->new() }
    sub max_threads () { Acme::Parataxis::get_max_thread_pool_size() }

    # Scheduler internals
    sub _scheduler_enqueue_by_id ($fid) {
        return if $SCHEDULER_QUEUED{$fid};
        if ( my $fiber = Acme::Parataxis->by_id($fid) ) {
            _enqueue($fiber);
        }
    }

    sub poll_io {
        my @ready;
        while (1) {
            my $job_idx = check_for_completion();
            last if $job_idx == -1;
            my $fid = get_job_coro_id($job_idx);
            my $res = get_job_result($job_idx);
            push @ready, [ $fid, $res ];
            free_job_slot($job_idx);
        }
        return @ready;
    }

    sub _handle_run ( $fiber, $status ) {
        if ( $status == 1 ) {
            Acme::Parataxis::_mark_done($fiber);
            my $err = $fiber->[F_ERROR];
            die $err if defined $err;
            return 1;
        }
        if ( $status == 0 ) {
            $fiber->[F_PRIORITY] //= 0;
            _enqueue($fiber);
        }
        return $status;
    }

    sub run ($code) {
        if ($IS_RUNNING) {

            # Nested run/async inside a shared global scheduler. Queue a fresh fiber for the block and park the current
            # fiber until it completes.
            my $fiber = __PACKAGE__->new( code => $code );
            _enqueue($fiber);
            return $fiber->await;
        }
        @SCHEDULER_QUEUE  = ();
        %SCHEDULER_QUEUED = ();
        $IS_RUNNING       = 1;
        my $main_fiber = __PACKAGE__->new( code => $code );
        _enqueue($main_fiber);
        while ($IS_RUNNING) {
            my @ready;
            if ( get_outstanding_jobs() ) {
                my $out = [];
                drain_jobs($out);
                @ready = @$out;
            }
            for my $ready (@ready) {
                my ( $fid, $res ) = @$ready;
                my $fiber = __PACKAGE__->by_id($fid);
                next unless $fiber;
                my $yield_val = $fiber->call($res);
                if ( defined $fiber && !$fiber->is_done ) {
                    _enqueue($fiber) unless defined $yield_val && $yield_val eq 'WAITING';
                }
            }
            if (@SCHEDULER_QUEUE) {
                my @work = @SCHEDULER_QUEUE;
                @SCHEDULER_QUEUE  = ();
                %SCHEDULER_QUEUED = ();
                for my $current (@work) {
                    next unless $current;
                    _handle_run( $current, run_fiber_checked( $current->fid, undef ) );
                }
            }
            my $active_count = get_live_fiber_count();
            if ( $IS_RUNNING && !@SCHEDULER_QUEUE && !@ready ) {
                if ( get_outstanding_jobs() ) {
                    usleep(1000);    # Wait for background jobs to finish
                }
                else {
                    die 'FATAL: deadlock detected...' if $active_count > 0;
                    $IS_RUNNING = 0                   if defined $main_fiber && $main_fiber->is_done;
                }
            }
        }
        return $main_fiber->[F_RESULT];
    }
    sub stop () { $IS_RUNNING = 0 }

    sub new ( $class, %args ) {
        my $self = bless [ $args{code}, 0, undef, undef, undef, 0, [], undef, undef, 0 ], $class;
        my $fid  = Acme::Parataxis::create_fiber( $args{code}, $self );
        croak 'could not allocate a fiber: the fiber table is full (destroy some fibers first)' if $fid < 0;
        $self->[F_FID] = $fid;
        return $self;
    }
    sub fid   ($self) { $self->[F_FID] }
    sub code  ($self) { $self->[F_CODE] }
    sub error ($self) { $self->[F_ERROR] }