view release on metacpan or  search on metacpan

lib/DBIO/Forked/Storage.pm  view on Meta::CPAN

      broker   => $self->_is_access_broker_connect_info(
        ref $info eq 'ARRAY' && @$info ? [ $info->[0] ] : []
      ) ? 1 : 0,
      elements => ref $info eq 'ARRAY' ? scalar @$info : 1,
    };
  }
  return $self->{connect_form};
}


sub async_option_names { qw( statement_timeout max_in_flight ) }

# The core checks the values of its own options; this backend checks the two it
# declares.
sub _assert_async_option {
  my ($self, $name, $value) = @_;
  $self->SUPER::_assert_async_option($name, $value);
  return unless $name eq 'statement_timeout' or $name eq 'max_in_flight';

  my $valid = !ref $value && defined $value && (
    $name eq 'max_in_flight'
      ? $value =~ /\A[1-9][0-9]*\z/
      : Scalar::Util::looks_like_number($value) && $value == $value
        && abs($value) != 9**9**9 && $value > 0
  );
  croak "DBIO::Forked: connect option '$name' takes "
      . ($name eq 'max_in_flight' ? 'a positive integer' : 'a number of seconds above 0')
    unless $valid;
  return;
}


sub transport_capabilities { qw( broker_connect_options ) }

# --- Schema / introspection ---

sub schema { $_[0]->{schema} }
sub debug  { $_[0]->{debug} }

# Model A forks one short-lived child per query, so there is no persistent
# connection pool to expose (unlike DBIO::Async). Override the inherited
# abstract pool() with an explicit "not applicable" rather than the misleading
# "subclass must override" croak.
sub pool {
  croak 'DBIO::Forked uses fork-per-query (Model A); there is no connection pool';
}


sub deploy_async {
  croak 'DBIO::Forked does not support deploy_async: the forked backend carries no '
      . 'DDL deploy (fork-per-query, Model A); deploy through the sync owner with '
      . '$schema->deploy';
}

# --- Async CRUD ---


sub select_async        { my $self = shift; $self->_run_forked('select',        @_) }
sub select_single_async { my $self = shift; $self->_run_forked('select_single', @_) }
sub insert_async        { my $self = shift; $self->_run_forked('insert',        @_) }
sub update_async        { my $self = shift; $self->_run_forked('update',        @_) }
sub delete_async        { my $self = shift; $self->_run_forked('delete',        @_) }

# txn_do_async is just _run_forked('txn_do', $body, @args): the child runs the
# inherited sync storage's txn_do($body, @args), so BEGIN/body/COMMIT all happen
# in the child on its freshly-reconnected connection. See the method POD for the
# return-value and sync-only limits.
sub txn_do_async        { my $self = shift; $self->_run_forked('txn_do',        @_) }

# --- Fork-per-query seam (Model A) ---
#
# The single point where Model A is realized. One short-lived fork per query:
#
#   1. pipe() a read/write pair, then fork().
#   2. Child: run the inherited primary sync storage's ORDINARY sync CRUD via
#      L</_forked_child_run> ($schema->storage->$op, shaped to the async row
#      contract -- see that method). The sync storage handles the fork trap
#      itself -- DBIO::Storage::DBI's _verify_pid sets InactiveDestroy on PID
#      change and _get_dbh reconnects fresh before the op -- so we never touch a
#      DBI handle, set no InactiveDestroy, and replay no connect_info. Crucially
#      we call the SYNC method (e.g. select), never select_async: sync does not
#      route to the async backend (core ADR 0030), so there is no re-fork. The
#      result rows (or the error) are Storable-frozen onto the write end; then
#      _exit (NOT exit -- no DESTROY/END in the child, which would tear down
#      resources shared with the parent).
#   3. Parent: close the write end and return a DBIO::Forked::Future bound to
#      the read fd and the child pid (is_ready peeks to EOF, get blocks + thaws
#      + reaps).
#
# EOF-framed: one frozen blob per child; the child closes its write end and the
# parent reads to EOF. No length prefix, no streaming.
sub _run_forked {
  my ($self, $op, @args) = @_;

  my $max = $self->async_option('max_in_flight');
  croak 'DBIO::Forked: max_in_flight reached, the call was refused before forking'
    if defined $max && $self->_in_flight_count >= $max;

  pipe(my $rh, my $wh) or croak "DBIO::Forked: pipe failed: $!";

  my $pid = fork();
  croak "DBIO::Forked: fork failed: $!" unless defined $pid;

  if (!$pid) {
    # --- Child ---
    close $rh;
    my @res = eval { $self->_forked_child_run($op, @args) };
    my $payload = $@ ? { error => "$@" } : { rows => \@res };
    my $blob = eval { Storable::freeze($payload) };
    if (!defined $blob) {
      # The result itself would not freeze (e.g. a txn_do body returned a live
      # Row object or a closure). Send a helpful error instead of a corrupt blob.
      $blob = Storable::freeze({ error => $self->_serialization_error($op, "$@") });
    }
    print {$wh} $blob;
    close $wh;
    POSIX::_exit(0);
  }

  # --- Parent ---
  close $wh;

lib/DBIO/Forked/Storage.pm  view on Meta::CPAN

#                      first_async, which feed the rows back through the ordinary
#                      collapse/inflate path). A storage whose select() already
#                      returns materialized rows (e.g. a fake) is passed through
#                      unchanged -- only a single blessed cursor is drained.
#   * select_single -- sync select_single() returns a flat list of column values;
#                      wrap it into a single raw-row arrayref (or empty), which is
#                      what select_single_async must resolve to (consumed by
#                      single_async / count_async, each reading one row).
#   * insert        -- sync insert() already returns the returned-columns HASHREF
#                      that Row::insert_async folds back into the object; pass it
#                      through (ADR 0031 contract).
#   * update/delete -- sync update()/delete() return ($rv, $sth, @bind) in list
#                      context and the affected-row count in scalar context. The
#                      Future resolves with the count alone (the async contract,
#                      DBIO::Storage::Async::update_async), and the statement
#                      handle would not freeze: ask for the scalar.
#   * txn_do -- pass the sync return through unchanged.
sub _forked_child_run {
  my ($self, $op, @args) = @_;
  my $storage = $self->{schema}->storage;

  if ($op eq 'select') {
    my @r = $storage->select(@args);
    return ( @r == 1 && Scalar::Util::blessed($r[0]) && $r[0]->can('all') )
      ? $r[0]->all   # real cursor -> raw rows
      : @r;          # already-materialized rows (fake storage)
  }
  elsif ($op eq 'select_single') {
    my @row = $storage->select_single(@args);
    return @row ? [ @row ] : ();
  }

  elsif ($op eq 'update' or $op eq 'delete') {
    my $count = $storage->$op(@args);
    return $count;
  }

  return $storage->$op(@args);
}

# Build a helpful error for a result that will not Storable-freeze. For txn_do
# the unserializable value is whatever the user's body returned, so the hint is
# return-value specific; for CRUD it is the driver's own rows.
sub _serialization_error {
  my ($self, $op, $err) = @_;
  return $op eq 'txn_do'
    ? 'DBIO::Forked: txn_do_async body must return Storable-serializable data '
      . '(scalars or plain array/hash refs), not live Row/ResultSet objects or '
      . "code refs: $err"
    : "DBIO::Forked: cannot serialize $op result over the pipe "
      . "(it must be plain serializable data): $err";
}

# --- Sync wrappers ---
# The sync entry points block on the forked result: run the query in a child
# and wait for it via ->get.

sub select        { my $self = shift; $self->select_async(@_)->get        }
sub select_single { my $self = shift; $self->select_single_async(@_)->get }
sub insert        { my $self = shift; $self->insert_async(@_)->get        }
sub update        { my $self = shift; $self->update_async(@_)->get        }
sub delete        { my $self = shift; $self->delete_async(@_)->get        }
sub txn_do        { my $self = shift; $self->txn_do_async(@_)->get        }

1;

__END__

=pod

=encoding UTF-8

=head1 NAME

DBIO::Forked::Storage - Fork-per-query async storage - make any sync DBIO driver async

=head1 VERSION

version 0.900004

=head1 DESCRIPTION

Fork-based async storage backend subclassing core L<DBIO::Storage::Async>. It
makes B<any> sync DBIO driver async without an async-capable client and without
an event loop, via C<fork()>-per-query (Model A, see L<DBIO::Forked>): the
child runs the driver's B<ordinary> sync CRUD through the inherited sync storage,
which reconnects itself in the child (see C<docs/adr/0002>; the backend
holds no connect info, see L</connect_info>) -- no SQL is
re-implemented here --
serializes the result rows back over a pipe with L<Storable>, and exits; the
parent returns a L<DBIO::Forked::Future> bound to the pipe read fd.

=head1 METHODS

=head2 new

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

Construct the fork-based async backend for C<$schema>. The schema reference is
weakened (the schema owns the storage, not the other way round). Connect info
is supplied separately via L</connect_info>.

=head2 future_class

Returns C<'DBIO::Forked::Future'> -- the loop-free, pipe-backed Future this
backend hands out.

=head2 connect_info

  $storage->connect_info([ $dsn, $user, $pass, \%attrs, \%dbio_opts ]);
  $storage->connect_info([ $broker ]);
  $storage->connect_info([ $broker, \%options ]);

Take the connect info the core resolver hands over (the sync storage's
C<< _connect_info >>) and keep its B<form> only: whether it is an
L<DBIO::AccessBroker> invocation, and how many elements it has. The backend
holds no access data -- no DSN, user, password, attribute, broker object or
options hash. Returns the form, a hash reference

  { broker => 0|1, elements => $n }