view release on metacpan or  search on metacpan

t/07-transport-paths.t  view on Meta::CPAN

#      select_single_async resolve to: a cursor is drained, a single row is
#      one arrayref, no row is nothing;
#   3. a child that dies before it writes: a failed Future saying so, reaped;
#   4. a Future dropped without get: DESTROY ends and reaps the child, also when the
#      child is blocked writing a result nobody will read;
#   5. a result larger than the pipe buffer: complete through get, and
#      complete through repeated is_ready (drained over several calls);
#   6. *_async inside a txn_do_async body: there is no guard, it forks again
#      from within the child;
#   7. pipeline, listen, deploy_async: what the caller is told.
#
# WHAT RUNS FOR REAL: fork + pipe + Storable, DBIO::Forked::Storage and its
# Future; in section 7 also core's DBIO::Schema and DBIO::Storage::DBI as the
# owner, never connected.
# THE SUBSTITUTIONS: no DBI and no database. Sections 1-6 run a mock sync
# storage whose ops answer fixed data or echo their arguments together with
# the pid they ran in -- so a result also says WHERE it was computed.

use DBIO::Schema;
use DBIO::Storage::DBI;
use DBIO::Storage::Async;
use DBIO::Forked::Storage;
use DBIO::Forked::Future;

alarm 60;

my $BIG_BYTES = 200 * 1024;   # three times a 64 KiB pipe buffer, and then some

{
  # What a real sync select() returns: a cursor, not rows. Blessed, answers all.
  package t::Mock::Cursor;
  sub new { my ($class, @rows) = @_; bless { rows => [ @rows ] }, $class }
  sub all { @{ $_[0]{rows} } }

  package t::Mock::Storage;
  sub new { bless {}, shift }

  # 200 KiB whose every four bytes differ: a lost, repeated or misplaced chunk
  # changes the string, which a run of one character would not show.
  sub big { pack 'N*', 0 .. $BIG_BYTES / 4 - 1 }

  sub select {
    my ($self, $what) = @_;
    return t::Mock::Cursor->new([ 1, 'Alice' ], [ 2, 'Bob' ]) if $what eq 'cursor';
    return t::Mock::Cursor->new                               if $what eq 'empty-cursor';
    return ([ 1, 'Alice' ])                                   if $what eq 'one-row';
    return ([ 1, $self->big ])                                if $what eq 'big';
    return ([ $$ ])                                           if $what eq 'whoami';
    if ($what eq 'slow') { Time::HiRes::sleep(0.3); return ([ 'late' ]) }
    kill KILL => $$                                           if $what eq 'die-unanswered';
    return ([ 1, 'Alice' ], [ 2, 'Bob' ]);
  }

  # Sync select_single returns the column values as a flat list.
  sub select_single {
    my ($self, $what) = @_;
    return $what eq 'miss' ? () : (7, 'Grace');
  }

  sub insert { my $self = shift; return { op => 'insert', pid => $$, args => [ @_ ] } }
  sub update { my $self = shift; return { op => 'update', pid => $$, args => [ @_ ] } }
  sub delete { my $self = shift; return { op => 'delete', pid => $$, args => [ @_ ] } }

  sub txn_do {
    my ($self, $code, @args) = @_;
    return { op => 'txn_do', pid => $$, body => [ $code->(@args) ] };
  }

  package t::Mock::Schema;
  sub new { my ($class, $storage) = @_; bless { storage => $storage }, $class }
  sub storage { $_[0]->{storage} }
}

my $schema  = t::Mock::Schema->new(t::Mock::Storage->new);
my $storage = DBIO::Forked::Storage->new($schema);

# Nothing left to wait for: no child of this process, running or dead. The
# stricter reading of "no zombies" -- a 0 here would be a child still alive.
sub no_children_ok {
  my ($name) = @_;
  my $found = waitpid(-1, WNOHANG);
  is $found, -1, $name;
}

# Computed in another process: a pid, and not ours.
sub in_child_ok {
  my ($pid, $name) = @_;
  ok defined $pid && $pid =~ /^\d+\z/ && $pid != $$, $name;
}

# ---------------------------------------------------------------------------
# 1. The six blocking wrappers.
# ---------------------------------------------------------------------------
{
  is_deeply [ $storage->select('rows') ], [ [ 1, 'Alice' ], [ 2, 'Bob' ] ],
    'select: blocks and returns the rows';

  # The wrappers answer in the shapes of the *_async methods they block on, not
  # in those of the sync storage: rows where that returns a cursor, one row
  # arrayref where that returns the column values as a list.
  is_deeply [ $storage->select_single('hit') ], [ [ 7, 'Grace' ] ],
    'select_single: blocks and returns the one row';

  for my $op (qw( insert update delete )) {
    my $got = $storage->$op('artist', { name => 'Carol' }, { id => 3 });
    is $got->{op}, $op, "$op: blocks and returns what the sync $op answered";
    is_deeply $got->{args}, [ 'artist', { name => 'Carol' }, { id => 3 } ],
      "$op: with the arguments as given";
    in_child_ok($got->{pid}, "$op: computed in a child, not in this process");
  }

  my $got = $storage->txn_do(sub { return [ reverse @_ ] }, 'x', 'y');
  is $got->{op}, 'txn_do', 'txn_do: blocks and returns what the sync txn_do answered';
  is_deeply $got->{body}, [ [ 'y', 'x' ] ], 'txn_do: the body ran with its arguments';
  in_child_ok($got->{pid}, 'txn_do: computed in a child, not in this process');

  no_children_ok('the wrappers reaped every child they forked');
}

# ---------------------------------------------------------------------------
# 2. delete_async, select_single_async, and the two reshaped reads.