LargeField.hs

-- | Coverage for receiving a single large field: one row with one binary
-- @bytea@ cell of 64 MiB, generated by the server, fetched through
-- 'Pqi.execParams' and read back with 'Pqi.getvalue'.
--
-- The divergence: @pqi-native@'s 'Pqi.Native.Transport.fillTo' grew its read
-- buffer by appending every received chunk to a strict @ByteString@
-- (@modifyIORef' buffer (<> chunk)@), copying the whole accumulated frame on
-- each receive, and asked @recv@ for a buffer as large as the rest of the
-- frame, allocating that much again on each receive. Receiving a frame thus
-- allocated quadratically in its size: over a Unix-domain socket, ~32 GiB
-- for a 16 MiB field and ~512 GiB for a 64 MiB one, where @libpq@ (through the reference adapter)
-- allocates about one copy of the value on the Haskell heap. The value
-- itself came back intact, only slowly.
--
-- Where it surfaced: @pqi-native@ issue #13.
--
-- Why this observation: the divergence is in cost, not in any returned
-- value, and cost is not directly comparable across adapters. Wall-clock
-- time depends on the machine and the container link, so the observation is
-- the calling thread's Haskell allocation over the query and the read of the
-- value ('getAllocationCounter', which counts per thread and needs no RTS
-- flags), reduced to whether it stays within 'allocationFactor' times the
-- field size. Any receive path that copies the data a bounded number of
-- times passes. How far the quadratic one overshoots depends on how the
-- transport chunks the stream: over a Unix-domain socket the issue measured
-- ~8000 times the field size, and over this suite's Docker-forwarded TCP
-- link, whose receives are far larger, ~36 times, which still clears the
-- bound ninefold. The returned length and contents are observed as well, so
-- that a fix cannot pass by truncating the value.
module Pqi.Conformance.Operation.ExecParams.LargeField
  ( spec,
  )
where

import Control.Exception (evaluate)
import qualified Data.ByteString as ByteString
import qualified Data.ByteString.Char8 as ByteString.Char8
import qualified Pqi
import Pqi.Conformance.Harness
import Pqi.Conformance.Prelude
import System.Mem (getAllocationCounter)
import Test.Hspec

-- | 64 MiB: large enough for quadratic growth to dwarf any linear constant
-- even over the large receives of a forwarded TCP link (at 16 MiB the
-- unfixed candidate allocated only ~14 times the field size there), small
-- enough to keep the reference fast.
fieldSize :: Int
fieldSize = 64 * 1024 * 1024

-- | How many times over the field size the calling thread may allocate while
-- receiving and reading it. The reference allocates about once (measured
-- 1.0 times); the bound leaves room for a receive path that copies a few
-- times more.
allocationFactor :: Int
allocationFactor = 4

spec :: Pqi.Adapter -> SpecWith ByteString
spec adapter =
  describe "execParams" do
    describe "large field" do
      it "receives a 64 MiB binary bytea with allocation linear in its size" \conninfo ->
        differential adapter conninfo scenario

-- | Fetch the field, then report the result status, the value's length,
-- whether every byte is the expected one, and whether the allocation stayed
-- within the bound. The allocation is taken across 'Pqi.execParams' and
-- 'Pqi.getvalue' only. The length is forced inside that span, since a lazy
-- adapter could otherwise defer the work past it.
scenario :: Pqi.Connection -> IO (Maybe (Pqi.ExecStatus, Maybe (Int, Bool), Bool))
scenario connection = do
  before <- getAllocationCounter
  result <-
    Pqi.execParams
      connection
      "select decode(repeat('ab', $1 :: int4), 'hex')"
      [Just (23, ByteString.Char8.pack (show fieldSize), Pqi.Text)]
      Pqi.Binary
  for result \r -> do
    status <- Pqi.resultStatus r
    value <- Pqi.getvalue r 0 0
    for_ value \v -> evaluate (ByteString.length v)
    after <- getAllocationCounter
    -- The counter counts down as the thread allocates.
    let allocated = fromIntegral (before - after) :: Int
        withinBound = allocated <= allocationFactor * fieldSize
    pure
      ( status,
        fmap (\v -> (ByteString.length v, ByteString.all (== 0xab) v)) value,
        withinBound
      )