add: use SO_REUSEPORT on platform supporting it

Author: mkleczek

CHANGELOG.md

@@ -13,6 +13,7 @@ All notable changes to this project will be documented in this file. From versio
 - Optimize requests with `Prefer: count=exact` that do not use ranges or `db-max-rows` by @laurenceisla in #3957
   + Removed unnecessary double count when building the `Content-Range`.
 - Add config `client_error_verbosity` to customize error verbosity by @taimoorzaeem in #4088, #3980, #3824
+- Use SO_REUSEPORT on platforms supporting it by @mkleczek in #4703 #4694
 
 ### Changed
 

src/PostgREST/Admin.hs

@@ -44,8 +44,8 @@ admin appState socketREST req respond  = do
       respond $ Wai.responseLBS (if isMainAppReachable then HTTP.status200 else HTTP.status500) [] mempty
     ["ready"] ->
       let
-        status | not isMainAppReachable = HTTP.status500
-               | isPending              = HTTP.status503
+        status | isPending              = HTTP.status503
+               | not isMainAppReachable = HTTP.status500
                | isLoaded               = HTTP.status200
                | otherwise              = HTTP.status500
       in

src/PostgREST/App.hs

@@ -63,12 +63,12 @@ import PostgREST.Version              (docsVersion, prettyVersion)
 
 import qualified Data.ByteString.Char8  as BS
 import qualified Data.List              as L
-import           Data.Streaming.Network (bindPortTCP,
-                                         bindRandomPortTCP)
+import           Data.Streaming.Network (HostPreference,
+                                         bindPortGenEx)
 import qualified Data.Text              as T
 import qualified Network.HTTP.Types     as HTTP
 import qualified Network.Socket         as NS
-import           PostgREST.Unix         (createAndBindDomainSocket)
+import           PostgREST.Unix         (createAndBindDomainSocketNoListen)
 import           Protolude              hiding (Handler)
 
 type Handler = ExceptT Error
@@ -77,16 +77,22 @@ run :: AppState -> IO ()
 run appState = do
   conf@AppConfig{..} <- AppState.getConfig appState
 
-  AppState.schemaCacheLoader appState -- Loads the initial SchemaCache
-  (mainSocket, adminSocket) <- initSockets conf
+  mainSocket <- initServerSocket conf
+  adminSocket <- initAdminServerSocket conf
   let closeSockets = do
         whenJust adminSocket NS.close
         NS.close mainSocket
   Unix.installSignalHandlers closeSockets (AppState.schemaCacheLoader appState) (AppState.readInDbConfig False appState)
 
+  Admin.runAdmin appState adminSocket mainSocket (serverSettings conf)
+
   Listener.runListener appState
 
-  Admin.runAdmin appState adminSocket mainSocket (serverSettings conf)
+  -- Kick off and wait for the initial SchemaCache load before listening on
+  -- the main API socket.
+  AppState.schemaCacheLoader appState
+  AppState.waitForSchemaCacheLoaded appState
+  void $ listenSocket mainSocket
 
   let app = postgrest configLogLevel appState (AppState.schemaCacheLoader appState)
 
@@ -244,39 +250,34 @@ addRetryHint delay response = do
 isServiceUnavailable :: Wai.Response -> Bool
 isServiceUnavailable response = Wai.responseStatus response == HTTP.status503
 
-type AppSockets = (NS.Socket, Maybe NS.Socket)
-
-initSockets :: AppConfig -> IO AppSockets
-initSockets AppConfig{..} = do
-  let
-    cfg'usp = configServerUnixSocket
-    cfg'uspm = configServerUnixSocketMode
-    cfg'host = configServerHost
-    cfg'port = configServerPort
-    cfg'adminHost = configAdminServerHost
-    cfg'adminPort = configAdminServerPort
-
-  sock <- case cfg'usp of
-    -- I'm not using `streaming-commons`' bindPath function here because it's not defined for Windows,
-    -- but we need to have runtime error if we try to use it in Windows, not compile time error
-    Just path -> createAndBindDomainSocket path cfg'uspm
-    Nothing -> do
-      (_, sock) <-
-        if cfg'port /= 0
-          then do
-            sock <- bindPortTCP cfg'port (fromString $ T.unpack cfg'host)
-            pure (cfg'port, sock)
-          else do
-            -- explicitly bind to a random port, returning bound port number
-            (num, sock) <- bindRandomPortTCP (fromString $ T.unpack cfg'host)
-            pure (num, sock)
-      pure sock
-
-  adminSock <- case cfg'adminPort of
-    Just adminPort -> do
-      adminSock <- bindPortTCP adminPort (fromString $ T.unpack cfg'adminHost)
-      pure $ Just adminSock
-    Nothing -> pure Nothing
-
-  pure (sock, adminSock)
+initServerSocket :: AppConfig -> IO NS.Socket
+initServerSocket AppConfig{..} = case configServerUnixSocket of
+  -- I'm not using `streaming-commons`' bindPath function here because it's not defined for Windows,
+  -- but we need to have runtime error if we try to use it in Windows, not compile time error
+  Just path -> createAndBindDomainSocketNoListen path configServerUnixSocketMode
+  Nothing ->
+    bindPortTCPWithoutListen configServerPort (fromString $ T.unpack configServerHost)
+
+initAdminServerSocket :: AppConfig -> IO (Maybe NS.Socket)
+initAdminServerSocket AppConfig{..} =
+  traverse (`bindPortTCPWithReusePort` adminHost) configAdminServerPort
+  where
+    adminHost = fromString $ T.unpack configAdminServerHost
+
+bindPortTCPWithReusePort :: Int -> HostPreference -> IO NS.Socket
+bindPortTCPWithReusePort port hostPreference
+  = bindPortTCPWithoutListen port hostPreference >>= listenSocket
+
+bindPortTCPWithoutListen :: Int -> HostPreference -> IO NS.Socket
+bindPortTCPWithoutListen port hostPreference = do
+  -- Some unix variants can expose ReusePort but reject it at runtime.
+  -- Fall back to binding without ReusePort when that happens.
+  socketWithReusePort <- try (bindPortGenEx reusePortOpts NS.Stream port hostPreference) :: IO (Either SomeException NS.Socket)
+  either (const $ bindPortGenEx [] NS.Stream port hostPreference) pure socketWithReusePort
+  where
+    reusePortOpts = [(NS.ReusePort, 1)]
 
+listenSocket :: NS.Socket -> IO NS.Socket
+listenSocket sock = do
+  NS.listen sock (max 2048 NS.maxListenQueue)
+  pure sock

src/PostgREST/AppState.hs

@@ -22,6 +22,7 @@ module PostgREST.AppState
   , usePool
   , readInDbConfig
   , schemaCacheLoader
+  , waitForSchemaCacheLoaded
   , getObserver
   , isLoaded
   , isPending
@@ -76,7 +77,7 @@ data AppState = AppState
   -- | Schema cache
   , stateSchemaCache       :: IORef (Maybe SchemaCache)
   -- | The schema cache status
-  , stateSCacheStatus      :: IORef SchemaCacheStatus
+  , stateSCacheStatus      :: SchemaCacheStatus
   -- | State of the LISTEN channel
   , stateIsListenerOn      :: IORef Bool
   -- | starts the connection worker with a debounce
@@ -99,11 +100,11 @@ data AppState = AppState
   , stateMetrics           :: Metrics.MetricsState
   }
 
--- | Schema cache status
-data SchemaCacheStatus
-  = SCLoaded
-  | SCPending
-  deriving Eq
+-- | Schema cache status.
+-- Empty means pending and full means loaded.
+newtype SchemaCacheStatus = SchemaCacheStatus
+  { getSCStatusMVar :: MVar ()
+  }
 
 init :: AppConfig -> IO AppState
 init conf@AppConfig{configLogLevel, configDbPoolSize} = do
@@ -122,7 +123,7 @@ initWithPool pool conf loggerState metricsState observer = do
   appState <- AppState pool
     <$> newIORef minimumPgVersion -- assume we're in a supported version when starting, this will be corrected on a later step
     <*> newIORef Nothing
-    <*> newIORef SCPending
+    <*> newSchemaCacheStatus
     <*> newIORef False
     <*> pure (pure ())
     <*> newIORef conf
@@ -240,6 +241,9 @@ putSchemaCache appState = atomicWriteIORef (stateSchemaCache appState)
 schemaCacheLoader :: AppState -> IO ()
 schemaCacheLoader = debouncedSCacheLoader
 
+waitForSchemaCacheLoaded :: AppState -> IO ()
+waitForSchemaCacheLoaded = void . readMVar . getSCStatusMVar . stateSCacheStatus
+
 getNextDelay :: AppState -> IO Int
 getNextDelay = readIORef . stateNextDelay
 
@@ -277,18 +281,15 @@ putIsListenerOn = atomicWriteIORef . stateIsListenerOn
 
 isLoaded :: AppState -> IO Bool
 isLoaded x = do
-  scacheStatus <- readIORef $ stateSCacheStatus x
+  scacheLoaded <- isSchemaCacheLoaded x
   connEstablished <- isConnEstablished x
-  return $ scacheStatus == SCLoaded && connEstablished
+  return $ scacheLoaded && connEstablished
 
 isPending :: AppState -> IO Bool
 isPending x = do
-  scacheStatus <- readIORef $ stateSCacheStatus x
+  scacheLoaded <- isSchemaCacheLoaded x
   connEstablished <- isConnEstablished x
-  return $ scacheStatus == SCPending || not connEstablished
-
-putSCacheStatus :: AppState -> SchemaCacheStatus -> IO ()
-putSCacheStatus = atomicWriteIORef . stateSCacheStatus
+  return $ not scacheLoaded || not connEstablished
 
 getObserver :: AppState -> ObservationHandler
 getObserver = stateObserver
@@ -347,19 +348,19 @@ retryingSchemaCacheLoad appState@AppState{stateObserver=observer, stateMainThrea
         timeItT $ usePool appState (transaction SQL.ReadCommitted SQL.Read $ querySchemaCache conf)
       case result of
         Left e -> do
-          putSCacheStatus appState SCPending
+          markSchemaCachePending appState
           putSchemaCache appState Nothing
           observer $ SchemaCacheErrorObs configDbSchemas configDbExtraSearchPath e
           return Nothing
 
         Right sCache -> do
           -- IMPORTANT: While the pending schema cache state starts from running the above querySchemaCache, only at this stage we block API requests due to the usage of an
-          -- IORef on putSchemaCache. This is why SCacheStatus is put at SCPending here to signal the Admin server (using isPending) that we're on a recovery state.
-          putSCacheStatus appState SCPending
+          -- IORef on putSchemaCache. This is why schema cache status is marked as pending here to signal the Admin server (using isPending) that we're on a recovery state.
+          markSchemaCachePending appState
           putSchemaCache appState $ Just sCache
           observer $ SchemaCacheQueriedObs resultTime
           observer . uncurry SchemaCacheLoadedObs =<< timeItT (evaluate $ showSummary sCache)
-          putSCacheStatus appState SCLoaded
+          markSchemaCacheLoaded appState
           return $ Just sCache
 
     shouldRetry :: RetryStatus -> (Maybe PgVersion, Maybe SchemaCache) -> IO Bool
@@ -375,6 +376,18 @@ retryingSchemaCacheLoad appState@AppState{stateObserver=observer, stateMainThrea
 
     oneSecondInUs = 1000000 -- one second in microseconds
 
+newSchemaCacheStatus :: IO SchemaCacheStatus
+newSchemaCacheStatus = SchemaCacheStatus <$> newEmptyMVar
+
+markSchemaCachePending :: AppState -> IO ()
+markSchemaCachePending = void . tryTakeMVar . getSCStatusMVar . stateSCacheStatus
+
+markSchemaCacheLoaded :: AppState -> IO ()
+markSchemaCacheLoaded = void . (`tryPutMVar` ()) . getSCStatusMVar . stateSCacheStatus
+
+isSchemaCacheLoaded :: AppState -> IO Bool
+isSchemaCacheLoaded = fmap not . isEmptyMVar . getSCStatusMVar . stateSCacheStatus
+
 -- | Reads the in-db config and reads the config file again
 -- | We don't retry reading the in-db config after it fails immediately, because it could have user errors. We just report the error and continue.
 readInDbConfig :: Bool -> AppState -> IO ()

src/PostgREST/Unix.hs

@@ -2,7 +2,7 @@
 
 module PostgREST.Unix
   ( installSignalHandlers
-  , createAndBindDomainSocket
+  , createAndBindDomainSocketNoListen
   ) where
 
 #ifndef mingw32_HOST_OS
@@ -32,16 +32,16 @@ installSignalHandlers interrupt usr1 usr2 = do
 installSignalHandlers _ _ _ = pass
 #endif
 
--- | Create a unix domain socket and bind it to the given path.
+-- | Create a unix domain socket and bind it to the given path, without
+-- | listening yet.
 -- | The socket file will be deleted if it already exists.
-createAndBindDomainSocket :: String -> FileMode -> IO NS.Socket
-createAndBindDomainSocket path mode = do
+createAndBindDomainSocketNoListen :: String -> FileMode -> IO NS.Socket
+createAndBindDomainSocketNoListen path mode = do
   unless NS.isUnixDomainSocketAvailable $
     panic "Cannot run with unix socket on non-unix platforms. Consider deleting the `server-unix-socket` config entry in order to continue."
   deleteSocketFileIfExist path
   sock <- NS.socket NS.AF_UNIX NS.Stream NS.defaultProtocol
   NS.bind sock $ NS.SockAddrUnix path
-  NS.listen sock (max 2048 NS.maxListenQueue)
   setFileMode path mode
   return sock
   where

test/io/postgrest.py

@@ -88,7 +88,7 @@ def run(
     admin_port=None,
     host=None,
     wait_for_readiness=True,
-    wait_max_seconds=1,
+    wait_max_seconds=3,
     no_pool_connection_available=False,
     no_startup_stdout=True,
 ):
@@ -188,6 +188,7 @@ def wait_until_exit(postgrest, timeout=1):
 def wait_until_status_code(url, max_seconds, status_code):
     "Wait for the given HTTP endpoint to return a status code"
     session = requests_unixsocket.Session()
+    response = None
 
     for _ in range(max_seconds * 10):
         try: