gistfile1.pl

sub _task {
    my $self = shift;
    my $coro_exit = 0;
    my $wakeme = $Coro::current;
    $self->{empty_count} = 0;
    $self->{fetch_count} = 0;
    my $num = $self->parallel_request_num;

    # create fetcher coro
    $self->create_request($num);

    # call on idle
    $Coro::idle = sub {
        warn "idle $coro_exit";
        warn map { $_->desc } Coro::State::list;
        if ($self->stop_flag) {
            warn "nothing todo. idle";
            $wakeme->ready;
        }
        return;
    };

    # timeout watcher
    my $w = AnyEvent->timer(
        after    => 1,
        interval => 1,
        cb       => sub {
            log_it("check timeout");
            my $now = Time::HiRes::time;
            my @lwp_coro = grep { $_->desc eq "LWP" } Coro::State::list;
            for my $coro (@lwp_coro) {
                if ( $coro->{timeout_at} && $now > $coro->{timeout_at} ) {
                    $coro->cancel("timeout");
                }
            }
            if (@lwp_coro < ($self->parallel_request_num / 2) ) {
                my $add = $self->parallel_request_num - @lwp_coro;
                $self->create_request($add);
            }
            if ( @lwp_coro == 0 && $self->stop_flag) {
                log_it("nothing todo.");
                $coro_exit = 1;
                $wakeme->ready;
            }
        }
    );
    my @s = map {
        AnyEvent->signal(signal => $_, cb => sub {
            $StopFlag = 1;
        });
    } qw(TERM INT QUIT);

    # main loop
    async {
        $Coro::current->desc("main EV loop");
        while (!$coro_exit) {
            # warn sprintf "loop_count: %s", EV::loop_count;
            EV::loop(EV::LOOP_ONESHOT);
        }
    };

    schedule;
}