michalrus
commented
6 years ago The extracted code (with ClassyPrelude as implicit Prelude):
import qualified Control.Exception as Unsafe
import qualified Data.Aeson as J
import GHC.Conc (myThreadId)
import qualified Network.Wai as W
import qualified Network.Wai.Handler.WebSockets as W
import qualified Network.WebSockets as W
import qualified Servant.Server.Internal.ServantErr as W (responseServantErr)
data KillWaiThread =
KillWaiThread
deriving (Show)
instance Exception KillWaiThread
-- |The whole point of this module is to have application-level pings for greater control in application. I.e. to solve <https://github.com/jaspervdj/websockets/issues/159>.
pingingWsApp :: (W.Connection -> IO ()) -> W.Application
pingingWsApp action __unused1 __unused2 = do
onTcpPong <- newTVarIO (pure ())
W.websocketsOr
W.defaultConnectionOptions
{ W.connectionOnPong = join (readTVarIO onTcpPong)
}
(wsApp onTcpPong)
backup
__unused1
__unused2
where
backup _ respond' = respond' . W.responseServantErr $ toServantErr NotAWebSocket
-----------------------------------------------
wsApp onTcpPong pendingConn =
ignoreKillWaiThread . ignoreClosedTCP $ do
conn <- W.acceptRequest pendingConn
waiThreadId <- myThreadId
pingTimeoutKiller :: TVar (Maybe (Async ())) <- newTVarIO Nothing
flip finally (traverse_ cancel =<< readTVarIO pingTimeoutKiller) $ do
atomically . writeTVar onTcpPong $ do
newKiller <-
async $ do
threadDelay' (2 * connectionTimeout)
sendDisconnect conn ("Ping timeout." :: Text)
Unsafe.throwTo waiThreadId KillWaiThread -- this is ugly, but what else can we do w/ the API provided by 'Network.WebSockets'? / Unsafe.throw, because we don’t want it wrapped — we’ll catch it later in 'ignoreKillWaiThread'.
previousKiller <- atomically $ swapTVar pingTimeoutKiller (Just newKiller)
cancel `traverse_` previousKiller
W.forkPingThread conn (round connectionTimeout)
action conn
-----------------------------------------------
ignoreClosedTCP = handle (\(_ :: W.ConnectionException) -> pure ()) -- don’t log them, no point
-----------------------------------------------
ignoreKillWaiThread = Unsafe.handle (\(_ :: KillWaiThread) -> pure ()) -- don’t log them, no point
-----------------------------------------------
sendDisconnect conn why = do
sendJson conn . J.object $ ["tag" J..= ("Error" :: Text), "description" J..= why]
W.sendClose conn ("{}" :: ByteString)
thomasjm
jaspervdj
nbouscal
domenkozar
I have a working solution with the current API, but it feels sooooo hacky. :smiley_cat:
connectionOnPongdoesn’t get itsConnection(why?), we’re bindingonPong <- newTVarIO (pure ()).connectionOnPong, we’re passingjoin (readTVarIO onPong).Connection, we’re putting an action into thisonPong, which:Async, which:threadDelays(2 * tcpTimeout).throwTo waiThread PingTimeout(this is awful).Async.finally, that cancels thatAsync, as well, so that after clean disconnection we don’t run the code meant for ping timeout.This way, on average after
1.5 * tcpTimeout(*), a dropped connection is detected and killed, and the whole system can learn about it, run appropriate clean up code etc.() Even if we were scheduling this killer
Asyncright after sending a Ping (which seems more in line with the naming (ping timeout), it’d still happen after `1.5 tcpTimeout, because on average, the Ping message would be sent0.5 * tcpTimeout` after the connection got dropped.When we don’t do that, dropping the connection on
iptablesdoes absolutely nothing, it just stays there as open, even after Warp’ssetTimeouttimeout passes! I don’t understand why, though.How to do this properly? Because my way surely doesn’t feel appropriate (although it works well).