view release on metacpan or  search on metacpan

examples/bulk-insert/app.pl  view on Meta::CPAN

    my $rate = $elapsed > 0 ? $state->{confirmed} / $elapsed : 0;
    log_line(
        $start,
        sprintf(
            'done inserted=%d batches=%d elapsed=%.2fs rate=%.0f/s keys_expire_in=%ds',
            $state->{confirmed},
            $state->{batches},
            $elapsed,
            $rate,
            $opts->{ttl},
        ),
    );

    return;
}

sub redis_args {
    my (%extra) = @_;
    return (
        host            => $ENV{REDIS_HOST} // 'localhost',
        port            => $ENV{REDIS_PORT} // 6379,
        connect_timeout => 2,
        %extra,
    );
}

async sub connect_redis {
    my ($role, %extra) = @_;
    my $redis = Async::Redis->new(redis_args(%extra));

    eval {
        await $redis->connect;
        1;
    } or do {
        my $error = $@ || 'unknown error';
        chomp $error;
        die "could not connect $role Redis client: $error\n";
    };

    return $redis;
}

async sub insert_keys {
    my ($redis, $state, $opts, $prefix) = @_;
    my $payload = 'x' x $opts->{payload_bytes};

    for (my $first = 1; $first <= $opts->{count}; $first += $opts->{batch}) {
        my $last = $first + $opts->{batch} - 1;
        $last = $opts->{count} if $last > $opts->{count};

        my @futures;
        for my $i ($first .. $last) {
            push @futures, $redis->set("$prefix$i", $payload, ex => $opts->{ttl});
        }

        $state->{issued} += scalar @futures;
        await Future->needs_all(@futures);
        $state->{confirmed} += scalar @futures;
        $state->{batches}++;

        # Give timer-based tasks a scheduling point between large batches.
        await Future::IO->sleep(0);
    }

    return;
}

async sub heartbeat {
    my ($redis, $state, $opts, $start) = @_;

    while (!$state->{done}) {
        await Future::IO->sleep($opts->{heartbeat});

        my $ping_start = time();
        await $redis->ping;
        my $ping_ms = (time() - $ping_start) * 1000;

        my $elapsed = time() - $start;
        my $rate = $elapsed > 0 ? $state->{confirmed} / $elapsed : 0;
        log_line(
            $start,
            sprintf(
                'heartbeat issued=%d confirmed=%d batches=%d ping_ms=%.1f rate=%.0f/s',
                $state->{issued},
                $state->{confirmed},
                $state->{batches},
                $ping_ms,
                $rate,
            ),
        );
    }

    return;
}

sub log_line {
    my ($start, $message) = @_;
    printf "[%5.2fs] %s\n", time() - $start, $message;
    return;
}