Refactor background psql TAP functions
authorDaniel Gustafsson <[email protected]>
Fri, 7 Apr 2023 20:14:20 +0000 (22:14 +0200)
committerDaniel Gustafsson <[email protected]>
Fri, 7 Apr 2023 20:14:20 +0000 (22:14 +0200)
This breaks out the background and interactive psql functionality into a
new class, PostgreSQL::Test::BackgroundPsql.  Sessions are still initiated
via PostgreSQL::Test::Cluster, but once started they can be manipulated by
the new helper functions which intend to make querying easier.  A sample
session for a command which can be expected to finish at a later time can
be seen below.

  my $session = $node->background_psql('postgres');
  $bsession->query_until(qr/start/, q(
    \echo start
CREATE INDEX CONCURRENTLY idx ON t(a);
  ));
  $bsession->quit;

Patch by Andres Freund with some additional hacking by me.

Author: Andres Freund <[email protected]>
Reviewed-by: Andrew Dunstan <[email protected]>
Discussion: https://postgr.es/m/20230130194350[email protected]

contrib/amcheck/t/003_cic_2pc.pl
src/bin/psql/t/010_tab_completion.pl
src/test/perl/PostgreSQL/Test/BackgroundPsql.pm [new file with mode: 0644]
src/test/perl/PostgreSQL/Test/Cluster.pm
src/test/recovery/t/010_logical_decoding_timelines.pl
src/test/recovery/t/031_recovery_conflict.pl
src/test/subscription/t/015_stream.pl

index eabe6fcf9645530ebd54d5b9d4275f7f8b8aa903..5323ed11ae94523a5aca929d347dcd3463e4f031 100644 (file)
@@ -36,63 +36,46 @@ $node->safe_psql('postgres', q(CREATE TABLE tbl(i int)));
 # statements.
 #
 
-my $main_in    = '';
-my $main_out   = '';
-my $main_timer = IPC::Run::timeout($PostgreSQL::Test::Utils::timeout_default);
-
-my $main_h =
-  $node->background_psql('postgres', \$main_in, \$main_out,
-   $main_timer, on_error_stop => 1);
-$main_in .= q(
+my $main_h = $node->background_psql('postgres');
+
+$main_h->query_safe(q(
 BEGIN;
 INSERT INTO tbl VALUES(0);
-\echo syncpoint1
-);
-pump $main_h until $main_out =~ /syncpoint1/ || $main_timer->is_expired;
-
-my $cic_in    = '';
-my $cic_out   = '';
-my $cic_timer = IPC::Run::timeout($PostgreSQL::Test::Utils::timeout_default);
-my $cic_h =
-  $node->background_psql('postgres', \$cic_in, \$cic_out,
-   $cic_timer, on_error_stop => 1);
-$cic_in .= q(
+));
+
+my $cic_h = $node->background_psql('postgres');
+
+$cic_h->query_until(qr/start/, q(
 \echo start
 CREATE INDEX CONCURRENTLY idx ON tbl(i);
-);
-pump $cic_h until $cic_out =~ /start/ || $cic_timer->is_expired;
+));
 
-$main_in .= q(
+$main_h->query_safe(q(
 PREPARE TRANSACTION 'a';
-);
+));
 
-$main_in .= q(
+$main_h->query_safe(q(
 BEGIN;
 INSERT INTO tbl VALUES(0);
-\echo syncpoint2
-);
-pump $main_h until $main_out =~ /syncpoint2/ || $main_timer->is_expired;
+));
 
 $node->safe_psql('postgres', q(COMMIT PREPARED 'a';));
 
-$main_in .= q(
+$main_h->query_safe(q(
 PREPARE TRANSACTION 'b';
 BEGIN;
 INSERT INTO tbl VALUES(0);
-\echo syncpoint3
-);
-pump $main_h until $main_out =~ /syncpoint3/ || $main_timer->is_expired;
+));
 
 $node->safe_psql('postgres', q(COMMIT PREPARED 'b';));
 
-$main_in .= q(
+$main_h->query_safe(q(
 PREPARE TRANSACTION 'c';
 COMMIT PREPARED 'c';
-);
-$main_h->pump_nb;
+));
 
-$main_h->finish;
-$cic_h->finish;
+$main_h->quit;
+$cic_h->quit;
 
 $result = $node->psql('postgres', q(SELECT bt_index_check('idx',true)));
 is($result, '0', 'bt_index_check after overlapping 2PC');
@@ -113,22 +96,15 @@ PREPARE TRANSACTION 'persists_forever';
 ));
 $node->restart;
 
-my $reindex_in  = '';
-my $reindex_out = '';
-my $reindex_timer =
-  IPC::Run::timeout($PostgreSQL::Test::Utils::timeout_default);
-my $reindex_h =
-  $node->background_psql('postgres', \$reindex_in, \$reindex_out,
-   $reindex_timer, on_error_stop => 1);
-$reindex_in .= q(
+my $reindex_h = $node->background_psql('postgres');
+$reindex_h->query_until(qr/start/, q(
 \echo start
 DROP INDEX CONCURRENTLY idx;
 CREATE INDEX CONCURRENTLY idx ON tbl(i);
-);
-pump $reindex_h until $reindex_out =~ /start/ || $reindex_timer->is_expired;
+));
 
 $node->safe_psql('postgres', "COMMIT PREPARED 'spans_restart'");
-$reindex_h->finish;
+$reindex_h->quit;
 $result = $node->psql('postgres', q(SELECT bt_index_check('idx',true)));
 is($result, '0', 'bt_index_check after 2PC and restart');
 
index 55a88f98122f8c364fe9469d3a830ec8a1c34d9f..576b81958e0741997001bc1dfed0e7e56645be1b 100644 (file)
@@ -7,7 +7,6 @@ use warnings;
 use PostgreSQL::Test::Cluster;
 use PostgreSQL::Test::Utils;
 use Test::More;
-use IPC::Run qw(pump finish timer);
 use Data::Dumper;
 
 # Do nothing unless Makefile has told us that the build is --with-readline.
@@ -92,14 +91,7 @@ print $FH "other stuff\n";
 close $FH;
 
 # fire up an interactive psql session
-my $in  = '';
-my $out = '';
-
-my $timer = timer($PostgreSQL::Test::Utils::timeout_default);
-
-my $h = $node->interactive_psql('postgres', \$in, \$out, $timer);
-
-like($out, qr/psql/, "print startup banner");
+my $h = $node->interactive_psql('postgres');
 
 # Simple test case: type something and see if psql responds as expected
 sub check_completion
@@ -109,15 +101,12 @@ sub check_completion
    # report test failures from caller location
    local $Test::Builder::Level = $Test::Builder::Level + 1;
 
-   # reset output collector
-   $out = "";
    # restart per-command timer
-   $timer->start($PostgreSQL::Test::Utils::timeout_default);
-   # send the data to be sent
-   $in .= $send;
-   # wait ...
-   pump $h until ($out =~ $pattern || $timer->is_expired);
-   my $okay = ($out =~ $pattern && !$timer->is_expired);
+   $h->{timeout}->start($PostgreSQL::Test::Utils::timeout_default);
+
+   # send the data to be sent and wait for its result
+   my $out = $h->query_until($pattern, $send);
+   my $okay = ($out =~ $pattern && !$h->{timeout}->is_expired);
    ok($okay, $annotation);
    # for debugging, log actual output if it didn't match
    local $Data::Dumper::Terse = 1;
@@ -451,10 +440,7 @@ check_completion(
 clear_line();
 
 # send psql an explicit \q to shut it down, else pty won't close properly
-$timer->start($PostgreSQL::Test::Utils::timeout_default);
-$in .= "\\q\n";
-finish $h or die "psql returned $?";
-$timer->reset;
+$h->quit or die "psql returned $?";
 
 # done
 $node->stop;
diff --git a/src/test/perl/PostgreSQL/Test/BackgroundPsql.pm b/src/test/perl/PostgreSQL/Test/BackgroundPsql.pm
new file mode 100644 (file)
index 0000000..d767e28
--- /dev/null
@@ -0,0 +1,299 @@
+
+# Copyright (c) 2021-2023, PostgreSQL Global Development Group
+
+=pod
+
+=head1 NAME
+
+PostgreSQL::Test::BackgroundPsql - class for controlling background psql processes
+
+=head1 SYNOPSIS
+
+  use PostgreSQL::Test::Cluster;
+
+  my $node = PostgreSQL::Test::Cluster->new('mynode');
+
+  # Create a data directory with initdb
+  $node->init();
+
+  # Start the PostgreSQL server
+  $node->start();
+
+  # Create and start an interactive psql session
+  my $isession = $node->interactive_psql('postgres');
+  # Apply timeout per query rather than per session
+  $isession->set_query_timer_restart();
+  # Run a query and get the output as seen by psql
+  my $ret = $isession->query("SELECT 1");
+  # Run a backslash command and wait until the prompt returns
+  $isession->query_until(qr/postgres #/, "\\d foo\n");
+  # Close the session and exit psql
+  $isession->quit;
+
+  # Create and start a background psql session
+  my $bsession = $node->background_psql('postgres');
+
+  # Run a query which is guaranteed to not return in case it fails
+  $bsession->query_safe("SELECT 1");
+  # Initiate a command which can be expected to terminate at a later stage
+  $bsession->query_until(qr/start/, q(
+    \echo start
+   CREATE INDEX CONCURRENTLY idx ON t(a);
+  ));
+  # Close the session and exit psql
+  $bsession->quit;
+
+=head1 DESCRIPTION
+
+PostgreSQL::Test::BackgroundPsql contains functionality for controlling
+a background or interactive psql session operating on a PostgreSQL node
+initiated by PostgreSQL::Test::Cluster.
+
+=cut
+
+package PostgreSQL::Test::BackgroundPsql;
+
+use strict;
+use warnings;
+
+use Carp;
+use Config;
+use IPC::Run;
+use PostgreSQL::Test::Utils qw(pump_until);
+use Test::More;
+
+=pod
+
+=head1 METHODS
+
+=over
+
+=item PostgreSQL::Test::BackroundPsql->new(interactive, @params)
+
+Builds a new object of class C<PostgreSQL::Test::BackgroundPsql> for either
+an interactive or background session and starts it. If C<interactive> is
+true then a PTY will be attached. C<psql_params> should contain the full
+command to run psql with all desired parameters and a complete connection
+string. For C<interactive> sessions, IO::Pty is required.
+
+=cut
+
+sub new
+{
+   my $class = shift;
+   my ($interactive, $psql_params) = @_;
+   my $psql = {'stdin' => '', 'stdout' => '', 'stderr' => '', 'query_timer_restart' => undef};
+   my $run;
+
+   # This constructor should only be called from PostgreSQL::Test::Cluster
+    my ($package, $file, $line) = caller;
+    die "Forbidden caller of constructor: package: $package, file: $file:$line"
+     unless $package->isa('PostgreSQL::Test::Cluster');
+
+   $psql->{timeout} = IPC::Run::timeout($PostgreSQL::Test::Utils::timeout_default);
+
+   if ($interactive)
+   {
+       $run = IPC::Run::start $psql_params,
+         '<pty<', \$psql->{stdin}, '>pty>', \$psql->{stdout}, '2>', \$psql->{stderr},
+         $psql->{timeout};
+   }
+   else
+   {
+       $run = IPC::Run::start $psql_params,
+         '<', \$psql->{stdin}, '>', \$psql->{stdout}, '2>', \$psql->{stderr},
+         $psql->{timeout};
+   }
+
+   $psql->{run} = $run;
+
+   my $self = bless $psql, $class;
+
+   $self->_wait_connect();
+
+   return $self;
+}
+
+# Internal routine for awaiting psql starting up and being ready to consume
+# input.
+sub _wait_connect
+{
+   my ($self) = @_;
+
+   # Request some output, and pump until we see it.  This means that psql
+   # connection failures are caught here, relieving callers of the need to
+   # handle those.  (Right now, we have no particularly good handling for
+   # errors anyway, but that might be added later.)
+   my $banner = "background_psql: ready";
+   $self->{stdin} .= "\\echo $banner\n";
+   $self->{run}->pump() until $self->{stdout} =~ /$banner/ || $self->{timeout}->is_expired;
+   $self->{stdout} = ''; # clear out banner
+
+   die "psql startup timed out" if $self->{timeout}->is_expired;
+}
+
+=pod
+
+=item $session->quit
+
+Close the session and clean up resources. Each test run must be closed with
+C<quit>.
+
+=cut
+
+sub quit
+{
+   my ($self) = @_;
+
+   $self->{stdin} .= "\\q\n";
+
+   return $self->{run}->finish;
+}
+
+=pod
+
+=item $session->reconnect_and_clear
+
+Terminate the current session and connect again.
+
+=cut
+
+sub reconnect_and_clear
+{
+   my ($self) = @_;
+
+   # If psql isn't dead already, tell it to quit as \q, when already dead,
+   # causes IPC::Run to unhelpfully error out with "ack Broken pipe:".
+   $self->{run}->pump_nb();
+   if ($self->{run}->pumpable())
+   {
+       $self->{stdin} .= "\\q\n";
+   }
+   $self->{run}->finish;
+
+   # restart
+   $self->{run}->run();
+   $self->{stdin}  = '';
+   $self->{stdout} = '';
+
+   $self->_wait_connect()
+}
+
+=pod
+
+=item $session->query()
+
+Executes a query in the current session and returns the output in scalar
+context and (output, error) in list context where error is 1 in case there
+was output generated on stderr when executing the query.
+
+=cut
+
+sub query
+{
+   my ($self, $query) = @_;
+   my $ret;
+   my $output;
+   local $Test::Builder::Level = $Test::Builder::Level + 1;
+
+   note "issuing query via background psql: $query";
+
+   $self->{timeout}->start() if (defined($self->{query_timer_restart}));
+
+   # Feed the query to psql's stdin, followed by \n (so psql processes the
+   # line), by a ; (so that psql issues the query, if it doesnt't include a ;
+   # itself), and a separator echoed with \echo, that we can wait on.
+   my $banner = "background_psql: QUERY_SEPARATOR";
+   $self->{stdin} .= "$query\n;\n\\echo $banner\n";
+
+   pump_until($self->{run}, $self->{timeout}, \$self->{stdout}, qr/$banner/);
+
+   die "psql query timed out" if $self->{timeout}->is_expired;
+   $output = $self->{stdout};
+
+   # remove banner again, our caller doesn't care
+   $output =~ s/\n$banner$//s;
+
+   # clear out output for the next query
+   $self->{stdout} = '';
+
+   $ret = $self->{stderr} eq "" ? 0 : 1;
+
+   return wantarray ? ( $output, $ret ) : $output;
+}
+
+=pod
+
+=item $session->query_safe()
+
+Wrapper around C<query> which errors out if the query failed to execute.
+Query failure is determined by it producing output on stderr.
+
+=cut
+
+sub query_safe
+{
+   my ($self, $query) = @_;
+
+   my $ret = $self->query($query);
+
+   if ($self->{stderr} ne "")
+   {
+       die "query failed: $self->{stderr}";
+   }
+
+   return $ret;
+}
+
+=pod
+
+=item $session->query_until(until, query)
+
+Issue C<query> and wait for C<until> appearing in the query output rather than
+waiting for query completion. C<query> needs to end with newline and semicolon
+(if applicable, interactive psql input may not require it) for psql to process
+the input.
+
+=cut
+
+sub query_until
+{
+   my ($self, $until, $query) = @_;
+   my $ret;
+   local $Test::Builder::Level = $Test::Builder::Level + 1;
+
+   $self->{timeout}->start() if (defined($self->{query_timer_restart}));
+   $self->{stdin} .= $query;
+
+   pump_until($self->{run}, $self->{timeout}, \$self->{stdout}, $until);
+
+   die "psql query timed out" if $self->{timeout}->is_expired;
+
+   $ret = $self->{stdout};
+
+   # clear out output for the next query
+   $self->{stdout} = '';
+
+   return $ret;
+}
+
+=pod
+
+=item $session->set_query_timer_restart()
+
+Configures the timer to be restarted before each query such that the defined
+timeout is valid per query rather than per test run.
+
+=back
+
+=cut
+
+sub set_query_timer_restart
+{
+   my $self = shift;
+
+   $self->{query_timer_restart} = shift if @_;
+   return $self->{query_timer_restart};
+}
+
+1;
index a3aef8b5e918fea66136f279ce351e7e613aac28..f3e36538d762c1226d978f745828eb153d908b80 100644 (file)
@@ -113,6 +113,7 @@ use PostgreSQL::Test::RecursiveCopy;
 use Socket;
 use Test::More;
 use PostgreSQL::Test::Utils ();
+use PostgreSQL::Test::BackgroundPsql ();
 use Time::HiRes qw(usleep);
 use Scalar::Util qw(blessed);
 
@@ -1966,18 +1967,12 @@ sub psql
 
 =pod
 
-=item $node->background_psql($dbname, \$stdin, \$stdout, $timer, %params) => harness
+=item $node->background_psql($dbname, %params) => PostgreSQL::Test::BackgroundPsql instance
 
-Invoke B<psql> on B<$dbname> and return an IPC::Run harness object, which the
-caller may use to send input to B<psql>.  The process's stdin is sourced from
-the $stdin scalar reference, and its stdout and stderr go to the $stdout
-scalar reference.  This allows the caller to act on other parts of the system
-while idling this backend.
+Invoke B<psql> on B<$dbname> and return a BackgroundPsql object.
 
-The specified timer object is attached to the harness, as well.  It's caller's
-responsibility to set the timeout length (usually
-$PostgreSQL::Test::Utils::timeout_default), and to restart the timer after
-each command if the timeout is per-command.
+A default timeout of $PostgreSQL::Test::Utils::timeout_default is set up,
+which can be modified later.
 
 psql is invoked in tuples-only unaligned mode with reading of B<.psqlrc>
 disabled.  That may be overridden by passing extra psql parameters.
@@ -1986,7 +1981,7 @@ Dies on failure to invoke psql, or if psql fails to connect.  Errors occurring
 later are the caller's problem.  psql runs with on_error_stop by default so
 that it will stop running sql and return 3 if passed SQL results in an error.
 
-Be sure to "finish" the harness when done with it.
+Be sure to "quit" the returned object when done with it.
 
 =over
 
@@ -2012,7 +2007,7 @@ If given, it must be an array reference containing additional parameters to B<ps
 
 sub background_psql
 {
-   my ($self, $dbname, $stdin, $stdout, $timer, %params) = @_;
+   my ($self, $dbname, %params) = @_;
 
    local %ENV = $self->_get_env();
 
@@ -2033,41 +2028,18 @@ sub background_psql
    push @psql_params, @{ $params{extra_params} }
      if defined $params{extra_params};
 
-   # Ensure there is no data waiting to be sent:
-   $$stdin = "" if ref($stdin);
-   # IPC::Run would otherwise append to existing contents:
-   $$stdout = "" if ref($stdout);
-
-   my $harness = IPC::Run::start \@psql_params,
-     '<', $stdin, '>', $stdout, $timer;
-
-   # Request some output, and pump until we see it.  This means that psql
-   # connection failures are caught here, relieving callers of the need to
-   # handle those.  (Right now, we have no particularly good handling for
-   # errors anyway, but that might be added later.)
-   my $banner = "background_psql: ready";
-   $$stdin = "\\echo $banner\n";
-   pump $harness until $$stdout =~ /$banner/ || $timer->is_expired;
-
-   die "psql startup timed out" if $timer->is_expired;
-
-   return $harness;
+   return PostgreSQL::Test::BackgroundPsql->new(0, \@psql_params);
 }
 
 =pod
 
-=item $node->interactive_psql($dbname, \$stdin, \$stdout, $timer, %params) => harness
+=item $node->interactive_psql($dbname, %params) => BackgroundPsql instance
 
-Invoke B<psql> on B<$dbname> and return an IPC::Run harness object,
-which the caller may use to send interactive input to B<psql>.
-The process's stdin is sourced from the $stdin scalar reference,
-and its stdout and stderr go to the $stdout scalar reference.
-ptys are used so that psql thinks it's being called interactively.
+Invoke B<psql> on B<$dbname> and return a BackgroundPsql object, which the
+caller may use to send interactive input to B<psql>.
 
-The specified timer object is attached to the harness, as well.  It's caller's
-responsibility to set the timeout length (usually
-$PostgreSQL::Test::Utils::timeout_default), and to restart the timer after
-each command if the timeout is per-command.
+A default timeout of $PostgreSQL::Test::Utils::timeout_default is set up,
+which can be modified later.
 
 psql is invoked in tuples-only unaligned mode with reading of B<.psqlrc>
 disabled.  That may be overridden by passing extra psql parameters.
@@ -2075,7 +2047,7 @@ disabled.  That may be overridden by passing extra psql parameters.
 Dies on failure to invoke psql, or if psql fails to connect.
 Errors occurring later are the caller's problem.
 
-Be sure to "finish" the harness when done with it.
+Be sure to "quit" the returned object when done with it.
 
 The only extra parameter currently accepted is
 
@@ -2093,7 +2065,7 @@ This requires IO::Pty in addition to IPC::Run.
 
 sub interactive_psql
 {
-   my ($self, $dbname, $stdin, $stdout, $timer, %params) = @_;
+   my ($self, $dbname, %params) = @_;
 
    local %ENV = $self->_get_env();
 
@@ -2104,26 +2076,7 @@ sub interactive_psql
    push @psql_params, @{ $params{extra_params} }
      if defined $params{extra_params};
 
-   # Ensure there is no data waiting to be sent:
-   $$stdin = "" if ref($stdin);
-   # IPC::Run would otherwise append to existing contents:
-   $$stdout = "" if ref($stdout);
-
-   my $harness = IPC::Run::start \@psql_params,
-     '<pty<', $stdin, '>pty>', $stdout, $timer;
-
-   # Pump until we see psql's help banner.  This ensures that callers
-   # won't write anything to the pty before it's ready, avoiding an
-   # implementation issue in IPC::Run.  Also, it means that psql
-   # connection failures are caught here, relieving callers of
-   # the need to handle those.  (Right now, we have no particularly
-   # good handling for errors anyway, but that might be added later.)
-   pump $harness
-     until $$stdout =~ /Type "help" for help/ || $timer->is_expired;
-
-   die "psql startup timed out" if $timer->is_expired;
-
-   return $harness;
+   return PostgreSQL::Test::BackgroundPsql->new(1, \@psql_params);
 }
 
 # Common sub of pgbench-invoking interfaces.  Makes any requested script files
index eb1a3b6ef8c3df3221309f8df630d1122d690ca8..993f654a9b0e33aea0710f4e8472cb19fe57f248 100644 (file)
@@ -28,7 +28,6 @@ use PostgreSQL::Test::Cluster;
 use PostgreSQL::Test::Utils;
 use Test::More;
 use File::Copy;
-use IPC::Run ();
 use Scalar::Util qw(blessed);
 
 my ($stdout, $stderr, $ret);
index 84375faccb79b1079d67a10446384a5cdcbfd36a..e29bc6c181c8d58af3f826dc66154f98541ede14 100644 (file)
@@ -67,14 +67,8 @@ $node_primary->wait_for_replay_catchup($node_standby);
 
 
 # a longrunning psql that we can use to trigger conflicts
-my $psql_timeout = IPC::Run::timer($PostgreSQL::Test::Utils::timeout_default);
-my %psql_standby = ('stdin' => '', 'stdout' => '');
-$psql_standby{run} =
-  $node_standby->background_psql($test_db, \$psql_standby{stdin},
-   \$psql_standby{stdout},
-   $psql_timeout);
-$psql_standby{stdout} = '';
-
+my $psql_standby = $node_standby->background_psql($test_db,
+   on_error_stop => 0);
 my $expected_conflicts = 0;
 
 
@@ -102,15 +96,14 @@ my $cursor1 = "test_recovery_conflict_cursor";
 
 # DECLARE and use a cursor on standby, causing buffer with the only block of
 # the relation to be pinned on the standby
-$psql_standby{stdin} .= qq[
-        BEGIN;
-        DECLARE $cursor1 CURSOR FOR SELECT b FROM $table1;
-        FETCH FORWARD FROM $cursor1;
-        ];
+my $res = $psql_standby->query_safe(qq[
+    BEGIN;
+    DECLARE $cursor1 CURSOR FOR SELECT b FROM $table1;
+    FETCH FORWARD FROM $cursor1;
+]);
 # FETCH FORWARD should have returned a 0 since all values of b in the table
 # are 0
-ok(pump_until_standby(qr/^0$/m),
-   "$sect: cursor with conflicting pin established");
+like($res, qr/^0$/m, "$sect: cursor with conflicting pin established");
 
 # to check the log starting now for recovery conflict messages
 my $log_location = -s $node_standby->logfile;
@@ -125,7 +118,7 @@ $node_primary->safe_psql($test_db, qq[VACUUM $table1;]);
 $node_primary->wait_for_replay_catchup($node_standby);
 
 check_conflict_log("User was holding shared buffer pin for too long");
-reconnect_and_clear();
+$psql_standby->reconnect_and_clear();
 check_conflict_stat("bufferpin");
 
 
@@ -138,15 +131,12 @@ $node_primary->safe_psql($test_db,
 $node_primary->wait_for_replay_catchup($node_standby);
 
 # DECLARE and FETCH from cursor on the standby
-$psql_standby{stdin} .= qq[
+$res = $psql_standby->query_safe(qq[
         BEGIN;
         DECLARE $cursor1 CURSOR FOR SELECT b FROM $table1;
         FETCH FORWARD FROM $cursor1;
-        ];
-ok( pump_until(
-       $psql_standby{run},     $psql_timeout,
-       \$psql_standby{stdout}, qr/^0$/m,),
-   "$sect: cursor with conflicting snapshot established");
+        ]);
+like($res, qr/^0$/m, "$sect: cursor with conflicting snapshot established");
 
 # Do some HOT updates
 $node_primary->safe_psql($test_db,
@@ -160,7 +150,7 @@ $node_primary->wait_for_replay_catchup($node_standby);
 
 check_conflict_log(
    "User query might have needed to see row versions that must be removed");
-reconnect_and_clear();
+$psql_standby->reconnect_and_clear();
 check_conflict_stat("snapshot");
 
 
@@ -169,12 +159,12 @@ $sect = "lock conflict";
 $expected_conflicts++;
 
 # acquire lock to conflict with
-$psql_standby{stdin} .= qq[
+$res = $psql_standby->query_safe(qq[
         BEGIN;
         LOCK TABLE $table1 IN ACCESS SHARE MODE;
         SELECT 1;
-        ];
-ok(pump_until_standby(qr/^1$/m), "$sect: conflicting lock acquired");
+        ]);
+like($res, qr/^1$/m, "$sect: conflicting lock acquired");
 
 # DROP TABLE containing block which standby has in a pinned buffer
 $node_primary->safe_psql($test_db, qq[DROP TABLE $table1;]);
@@ -182,7 +172,7 @@ $node_primary->safe_psql($test_db, qq[DROP TABLE $table1;]);
 $node_primary->wait_for_replay_catchup($node_standby);
 
 check_conflict_log("User was holding a relation lock for too long");
-reconnect_and_clear();
+$psql_standby->reconnect_and_clear();
 check_conflict_stat("lock");
 
 
@@ -193,14 +183,14 @@ $expected_conflicts++;
 # DECLARE a cursor for a query which, with sufficiently low work_mem, will
 # spill tuples into temp files in the temporary tablespace created during
 # setup.
-$psql_standby{stdin} .= qq[
+$res = $psql_standby->query_safe(qq[
         BEGIN;
         SET work_mem = '64kB';
         DECLARE $cursor1 CURSOR FOR
           SELECT count(*) FROM generate_series(1,6000);
         FETCH FORWARD FROM $cursor1;
-        ];
-ok(pump_until_standby(qr/^6000$/m),
+        ]);
+like($res, qr/^6000$/m,
    "$sect: cursor with conflicting temp file established");
 
 # Drop the tablespace currently containing spill files for the query on the
@@ -211,7 +201,7 @@ $node_primary->wait_for_replay_catchup($node_standby);
 
 check_conflict_log(
    "User was or might have been using tablespace that must be dropped");
-reconnect_and_clear();
+$psql_standby->reconnect_and_clear();
 check_conflict_stat("tablespace");
 
 
@@ -227,7 +217,7 @@ $node_standby->adjust_conf(
    'max_standby_streaming_delay',
    "${PostgreSQL::Test::Utils::timeout_default}s");
 $node_standby->restart();
-reconnect_and_clear();
+$psql_standby->reconnect_and_clear();
 
 # Generate a few dead rows, to later be cleaned up by vacuum. Then acquire a
 # lock on another relation in a prepared xact, so it's held continuously by
@@ -250,19 +240,15 @@ SELECT txid_current();
 
 $node_primary->wait_for_replay_catchup($node_standby);
 
-$psql_standby{stdin} .= qq[
+$res = $psql_standby->query_until(qr/^1$/m, qq[
     BEGIN;
     -- hold pin
     DECLARE $cursor1 CURSOR FOR SELECT a FROM $table1;
     FETCH FORWARD FROM $cursor1;
     -- wait for lock held by prepared transaction
    SELECT * FROM $table2;
-    ];
-ok( pump_until(
-       $psql_standby{run},     $psql_timeout,
-       \$psql_standby{stdout}, qr/^1$/m,),
-   "$sect: cursor holding conflicting pin, also waiting for lock, established"
-);
+    ]);
+ok( 1, "$sect: cursor holding conflicting pin, also waiting for lock, established");
 
 # just to make sure we're waiting for lock already
 ok( $node_standby->poll_query_until(
@@ -277,7 +263,7 @@ $node_primary->safe_psql($test_db, qq[VACUUM $table1;]);
 $node_primary->wait_for_replay_catchup($node_standby);
 
 check_conflict_log("User transaction caused buffer deadlock with recovery.");
-reconnect_and_clear();
+$psql_standby->reconnect_and_clear();
 check_conflict_stat("deadlock");
 
 # clean up for next tests
@@ -285,7 +271,7 @@ $node_primary->safe_psql($test_db, qq[ROLLBACK PREPARED 'lock';]);
 $node_standby->adjust_conf('postgresql.conf', 'max_standby_streaming_delay',
    '50ms');
 $node_standby->restart();
-reconnect_and_clear();
+$psql_standby->reconnect_and_clear();
 
 
 # Check that expected number of conflicts show in pg_stat_database. Needs to
@@ -309,8 +295,7 @@ check_conflict_log("User was connected to a database that must be dropped");
 
 # explicitly shut down psql instances gracefully - to avoid hangs or worse on
 # windows
-$psql_standby{stdin} .= "\\q\n";
-$psql_standby{run}->finish;
+$psql_standby->quit;
 
 $node_standby->stop();
 $node_primary->stop();
@@ -318,37 +303,6 @@ $node_primary->stop();
 
 done_testing();
 
-
-sub pump_until_standby
-{
-   my $match = shift;
-
-   return pump_until($psql_standby{run}, $psql_timeout,
-       \$psql_standby{stdout}, $match);
-}
-
-sub reconnect_and_clear
-{
-   # If psql isn't dead already, tell it to quit as \q, when already dead,
-   # causes IPC::Run to unhelpfully error out with "ack Broken pipe:".
-   $psql_standby{run}->pump_nb();
-   if ($psql_standby{run}->pumpable())
-   {
-       $psql_standby{stdin} .= "\\q\n";
-   }
-   $psql_standby{run}->finish;
-
-   # restart
-   $psql_standby{run}->run();
-   $psql_standby{stdin}  = '';
-   $psql_standby{stdout} = '';
-
-   # Run query to ensure connection has finished re-establishing
-   $psql_standby{stdin} .= qq[SELECT 1;\n];
-   die unless pump_until_standby(qr/^1$/m);
-   $psql_standby{stdout} = '';
-}
-
 sub check_conflict_log
 {
    my $message          = shift;
index 0e0f27f14dfa65ed62402165988129797e5af14e..88344bdbaa0c779b23b6471df4e1f9cb2b7414e0 100644 (file)
@@ -28,26 +28,20 @@ sub test_streaming
    my ($node_publisher, $node_subscriber, $appname, $is_parallel) = @_;
 
    # Interleave a pair of transactions, each exceeding the 64kB limit.
-   my $in  = '';
-   my $out = '';
-
    my $offset = 0;
 
-   my $timer = IPC::Run::timeout($PostgreSQL::Test::Utils::timeout_default);
-
-   my $h = $node_publisher->background_psql('postgres', \$in, \$out, $timer,
+   my $h = $node_publisher->background_psql('postgres',
        on_error_stop => 0);
 
    # Check the subscriber log from now on.
    $offset = -s $node_subscriber->logfile;
 
-   $in .= q{
+   $h->query_safe(q{
    BEGIN;
    INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(3, 5000) s(i);
    UPDATE test_tab SET b = md5(b) WHERE mod(a,2) = 0;
    DELETE FROM test_tab WHERE mod(a,3) = 0;
-   };
-   $h->pump_nb;
+   });
 
    $node_publisher->safe_psql(
        'postgres', q{
@@ -57,11 +51,9 @@ sub test_streaming
    COMMIT;
    });
 
-   $in .= q{
-   COMMIT;
-   \q
-   };
-   $h->finish;    # errors make the next test fail, so ignore them here
+   $h->query_safe('COMMIT');
+    # errors make the next test fail, so ignore them here
+   $h->quit;
 
    $node_publisher->wait_for_catchup($appname);
 
@@ -219,12 +211,7 @@ $node_subscriber->reload;
 $node_subscriber->safe_psql('postgres', q{SELECT 1});
 
 # Interleave a pair of transactions, each exceeding the 64kB limit.
-my $in  = '';
-my $out = '';
-
-my $timer = IPC::Run::timeout($PostgreSQL::Test::Utils::timeout_default);
-
-my $h = $node_publisher->background_psql('postgres', \$in, \$out, $timer,
+my $h = $node_publisher->background_psql('postgres',
    on_error_stop => 0);
 
 # Confirm if a deadlock between the leader apply worker and the parallel apply
@@ -232,11 +219,10 @@ my $h = $node_publisher->background_psql('postgres', \$in, \$out, $timer,
 
 my $offset = -s $node_subscriber->logfile;
 
-$in .= q{
+$h->query_safe(q{
 BEGIN;
 INSERT INTO test_tab_2 SELECT i FROM generate_series(1, 5000) s(i);
-};
-$h->pump_nb;
+});
 
 # Ensure that the parallel apply worker executes the insert command before the
 # leader worker.
@@ -246,11 +232,8 @@ $node_subscriber->wait_for_log(
 
 $node_publisher->safe_psql('postgres', "INSERT INTO test_tab_2 values(1)");
 
-$in .= q{
-COMMIT;
-\q
-};
-$h->finish;
+$h->query_safe('COMMIT');
+$h->quit;
 
 $node_subscriber->wait_for_log(qr/ERROR: ( [A-Z0-9]+:)? deadlock detected/,
    $offset);
@@ -277,11 +260,10 @@ $node_subscriber->safe_psql('postgres',
 # Check the subscriber log from now on.
 $offset = -s $node_subscriber->logfile;
 
-$in .= q{
+$h->query_safe(q{
 BEGIN;
 INSERT INTO test_tab_2 SELECT i FROM generate_series(1, 5000) s(i);
-};
-$h->pump_nb;
+});
 
 # Ensure that the first parallel apply worker executes the insert command
 # before the second one.
@@ -292,11 +274,8 @@ $node_subscriber->wait_for_log(
 $node_publisher->safe_psql('postgres',
    "INSERT INTO test_tab_2 SELECT i FROM generate_series(1, 5000) s(i)");
 
-$in .= q{
-COMMIT;
-\q
-};
-$h->finish;
+$h->query_safe('COMMIT');
+$h->quit;
 
 $node_subscriber->wait_for_log(qr/ERROR: ( [A-Z0-9]+:)? deadlock detected/,
    $offset);