Multiplexing and pool rewrite

Author: roji

.github/workflows/build.yml

@@ -41,6 +41,9 @@ jobs:
           sudo apt-get install -qq postgresql-${{ matrix.pg_major }} postgresql-${{ matrix.pg_major }}-postgis-${{ env.postgis_version }}
           sudo -u postgres psql -c "CREATE USER npgsql_tests SUPERUSER PASSWORD 'npgsql_tests'"
           sudo -u postgres psql -c "CREATE DATABASE npgsql_tests OWNER npgsql_tests"
+          sudo -u postgres psql -c "CREATE EXTENSION postgis" npgsql_tests
+          sudo -u postgres psql -c "CREATE EXTENSION hstore" npgsql_tests
+          sudo -u postgres psql -c "CREATE EXTENSION citext" npgsql_tests
 
           export PGDATA=/etc/postgresql/${{ matrix.pg_major }}/main
           sudo sed -i 's/#ssl = off/ssl = on/' $PGDATA/postgresql.conf
@@ -91,6 +94,9 @@ jobs:
           # Configure test account
           pgsql/bin/psql -U postgres -c "CREATE ROLE npgsql_tests SUPERUSER LOGIN PASSWORD 'npgsql_tests'"
           pgsql/bin/psql -U postgres -c "CREATE DATABASE npgsql_tests OWNER npgsql_tests"
+          pgsql/bin/psql -U postgres -c "CREATE EXTENSION postgis" npgsql_tests
+          pgsql/bin/psql -U postgres -c "CREATE EXTENSION hstore" npgsql_tests
+          pgsql/bin/psql -U postgres -c "CREATE EXTENSION citext" npgsql_tests
 
           # Disable trust authentication, requiring MD5 passwords - some tests must fail if a password isn't provided.
           echo "host all all all md5" > pgsql/PGDATA/pg_hba.conf

src/Npgsql/ConnectorPool.Multiplexing.cs

@@ -5,7 +5,7 @@
     <PackageTags>npgsql postgresql postgres ado ado.net database sql</PackageTags>
     <!-- At this point we target netcoreapp3.0 to avoid taking a dependency on System.Text.Json, which is
          necessary for all other TFMs. -->
-    <TargetFrameworks>netcoreapp5.0</TargetFrameworks>
+    <TargetFrameworks>netcoreapp3.0</TargetFrameworks>
     <TargetFrameworks Condition="'$(DeveloperBuild)' != 'True'">net461;netstandard2.0;netstandard2.1;netcoreapp3.0</TargetFrameworks>
     <AllowUnsafeBlocks>true</AllowUnsafeBlocks>
   </PropertyGroup>
@@ -24,5 +24,6 @@
   </ItemGroup>
   <ItemGroup Condition=" '$(TargetFramework)' == 'net461' OR '$(TargetFramework)' == 'netstandard2.0' OR '$(TargetFramework)' == 'netstandard2.1' ">
     <PackageReference Include="System.Text.Json" Version="4.6.0" />
+    <PackageReference Include="System.Threading.Channels" Version="4.7.0" />
   </ItemGroup>
 </Project>

src/Npgsql/ConnectorPool.cs

@@ -59,8 +59,9 @@ internal NpgsqlBinaryExporter(NpgsqlConnector connector, string copyToCommand)
                 copyOutResponse = (CopyOutResponseMessage)msg;
                 if (!copyOutResponse.IsBinary)
                 {
-                    _connector.Break();
-                    throw new ArgumentException("copyToCommand triggered a text transfer, only binary is allowed", nameof(copyToCommand));
+                    throw _connector.Break(
+                        new ArgumentException("copyToCommand triggered a text transfer, only binary is allowed",
+                            nameof(copyToCommand)));
                 }
                 break;
             case BackendMessageCode.CompletedResponse:
@@ -281,9 +282,9 @@ async ValueTask<T> DoRead<T>(NpgsqlTypeHandler handler, bool async)
                 _column++;
                 return result;
             }
-            catch
+            catch (Exception e)
             {
-                _connector.Break();
+                _connector.Break(e);
                 Cleanup();
                 throw;
             }
@@ -400,6 +401,7 @@ void Cleanup()
             if (connector != null)
             {
                 connector.CurrentCopyOperation = null;
+                _connector.Connection?.EndBindingScope(ConnectorBindingScope.Copy);
                 _connector = null;
             }
 

src/Npgsql/Npgsql.csproj

@@ -63,8 +63,9 @@ internal NpgsqlBinaryImporter(NpgsqlConnector connector, string copyFromCommand)
                     copyInResponse = (CopyInResponseMessage)msg;
                     if (!copyInResponse.IsBinary)
                     {
-                        _connector.Break();
-                        throw new ArgumentException("copyFromCommand triggered a text transfer, only binary is allowed", nameof(copyFromCommand));
+                        throw _connector.Break(
+                            new ArgumentException("copyFromCommand triggered a text transfer, only binary is allowed",
+                                nameof(copyFromCommand)));
                     }
                     break;
                 case BackendMessageCode.CompletedResponse:
@@ -448,8 +449,8 @@ async Task Cancel(bool async)
             {
                 var msg = await _connector.ReadMessage(async);
                 // The CopyFail should immediately trigger an exception from the read above.
-                _connector.Break();
-                throw new NpgsqlException("Expected ErrorResponse when cancelling COPY but got: " + msg.Code);
+                throw _connector.Break(
+                    new NpgsqlException("Expected ErrorResponse when cancelling COPY but got: " + msg.Code));
             }
             catch (PostgresException e)
             {
@@ -508,6 +509,7 @@ void Cleanup()
             if (connector != null)
             {
                 connector.CurrentCopyOperation = null;
+                _connector.Connection?.EndBindingScope(ConnectorBindingScope.Copy);
                 _connector = null;
             }