support a P2PConnection that uses TMVars rather than Handles

This will allow having an internal thread speaking P2P protocol,
which will be needed to support proxying to external special remotes.

No serialization is done on the internal P2P protocol of course.

When a ByteString is being exchanged, it may or may not be exactly
the length indicated by DATA. While that has to be carefully managed
for the serialized P2P protocol, here it would require buffering the
whole lazy bytestring in memory to check its length when sending,
so it's better to do length checks on the receiving side.

diff --git a/Command/EnableTor.hs b/Command/EnableTor.hs
index 1900d5349b1dd754778830e91185fdcb644b7bf1..f80c4c06fd4d4c5baea8d415a75db84e25e9bc0f 100644 (file)
--- a/Command/EnableTor.hs
+++ b/Command/EnableTor.hs
@@ -135,8 +135,8 @@ checkHiddenService = bracket setup cleanup go
                let conn = P2PConnection
                        { connRepo = Nothing
                        , connCheckAuth = const False
-                       , connIhdl = h
-                       , connOhdl = h
+                       , connIhdl = P2PHandle h
+                       , connOhdl = P2PHandle h
                        , connIdent = ConnIdent Nothing
                        }
                runst <- mkRunState Client

diff --git a/Command/P2PStdIO.hs b/Command/P2PStdIO.hs
index dc3e081d812207e5b0193ff81cef04e7a7fce9db..ee1e7ef2bda4b08d994fb74c47d89694cee4cb48 100644 (file)
--- a/Command/P2PStdIO.hs
+++ b/Command/P2PStdIO.hs
@@ -67,7 +67,7 @@ performProxy clientuuid servermode r = do
                p2pErrHandler
   where
        withclientversion clientside (Just (clientmaxversion, othermsg)) = do
-               remoteside <- proxySshRemoteSide clientmaxversion mempty r
+               remoteside <- proxyRemoteSide clientmaxversion mempty r
                protocolversion <- either (const (min P2P.maxProtocolVersion clientmaxversion)) id
                        <$> runRemoteSide remoteside 
                                (P2P.net P2P.getProtocolVersion)

diff --git a/P2P/IO.hs b/P2P/IO.hs
index ccbdf1d6ba5ba6590ff0a8393abb18c26a6f551b..643aafc4ec6b37efbf2bf165b93d6570cd10eecf 100644 (file)
--- a/P2P/IO.hs
+++ b/P2P/IO.hs
@@ -11,6 +11,7 @@ module P2P.IO
        ( RunProto
        , RunState(..)
        , mkRunState
+       , P2PHandle(..)
        , P2PConnection(..)
        , ConnIdent(..)
        , ClosableConnection(..)
@@ -74,11 +75,15 @@ mkRunState mk = do
        tvar <- newTVarIO defaultProtocolVersion
        return (mk tvar)
 
+data P2PHandle
+       = P2PHandle Handle
+       | P2PHandleTMVar (TMVar (Either L.ByteString Message))
+
 data P2PConnection = P2PConnection
        { connRepo :: Maybe Repo
        , connCheckAuth :: (AuthToken -> Bool)
-       , connIhdl :: Handle
-       , connOhdl :: Handle
+       , connIhdl :: P2PHandle
+       , connOhdl :: P2PHandle
        , connIdent :: ConnIdent
        }
 
@@ -94,8 +99,8 @@ stdioP2PConnection :: Maybe Git.Repo -> P2PConnection
 stdioP2PConnection g = P2PConnection
        { connRepo = g
        , connCheckAuth = const False
-       , connIhdl = stdin
-       , connOhdl = stdout
+       , connIhdl = P2PHandle stdin
+       , connOhdl = P2PHandle stdout
        , connIdent = ConnIdent Nothing
        }
 
@@ -106,15 +111,18 @@ connectPeer g (TorAnnex onionaddress onionport) = do
        return $ P2PConnection
                { connRepo = g
                , connCheckAuth = const False
-               , connIhdl = h
-               , connOhdl = h
+               , connIhdl = P2PHandle h
+               , connOhdl = P2PHandle h
                , connIdent = ConnIdent Nothing
                }
 
 closeConnection :: P2PConnection -> IO ()
 closeConnection conn = do
-       hClose (connIhdl conn)
-       hClose (connOhdl conn)
+       closehandle (connIhdl conn)
+       closehandle (connOhdl conn)
+  where
+       closehandle (P2PHandle h) = hClose h
+       closehandle (P2PHandleTMVar _) = return ()
 
 -- Serves the protocol on a unix socket.
 --
@@ -164,6 +172,11 @@ runNetProto runst conn = go
        go (Free (Local _)) = return $ Left $
                ProtoFailureMessage "unexpected annex operation attempted"
 
+data P2PTMVarException = P2PTMVarException String
+       deriving (Show)
+
+instance Exception P2PTMVarException
+
 -- Interpreter of the Net part of Proto.
 --
 -- An interpreter of Proto has to be provided, to handle the rest of Proto
@@ -171,40 +184,68 @@ runNetProto runst conn = go
 runNet :: (MonadIO m, MonadMask m) => RunState -> P2PConnection -> RunProto m -> NetF (Proto a) -> m (Either ProtoFailure a)
 runNet runst conn runner f = case f of
        SendMessage m next -> do
-               v <- liftIO $ tryNonAsync $ do
-                       let l = unwords (formatMessage m)
+               v <- liftIO $ do
                        debugMessage conn "P2P >" m
-                       hPutStrLn (connOhdl conn) l
-                       hFlush (connOhdl conn)
+                       case connOhdl conn of
+                               P2PHandle h -> tryNonAsync $ do
+                                       hPutStrLn h $ unwords (formatMessage m)
+                                       hFlush h
+                               P2PHandleTMVar mv ->
+                                       ifM (atomically (tryPutTMVar mv (Right m)))
+                                               ( return $ Right ()
+                                               , return $ Left $ toException $
+                                                       P2PTMVarException "TMVar left full"
+                                               )
                case v of
                        Left e -> return $ Left $ ProtoFailureException e
                        Right () -> runner next
-       ReceiveMessage next -> do
-               v <- liftIO $ tryIOError $ getProtocolLine (connIhdl conn)
-               case v of
-                       Left e -> return $ Left $ ProtoFailureIOError e
-                       Right Nothing -> return $ Left $
-                               ProtoFailureMessage "protocol error"
-                       Right (Just l) -> case parseMessage l of
-                               Just m -> do
-                                       liftIO $ debugMessage conn "P2P <" m
-                                       runner (next (Just m))
-                               Nothing -> runner (next Nothing)
-       SendBytes len b p next -> do
-               v <- liftIO $ tryNonAsync $ do
-                       ok <- sendExactly len b (connOhdl conn) p
-                       hFlush (connOhdl conn)
-                       return ok
-               case v of
-                       Right True -> runner next
-                       Right False -> return $ Left $
-                               ProtoFailureMessage "short data write"
-                       Left e -> return $ Left $ ProtoFailureException e
-       ReceiveBytes len p next -> do
-               v <- liftIO $ tryNonAsync $ receiveExactly len (connIhdl conn) p
-               case v of
-                       Left e -> return $ Left $ ProtoFailureException e
-                       Right b -> runner (next b)
+       ReceiveMessage next ->
+               let protoerr = return $ Left $
+                       ProtoFailureMessage "protocol error"
+                   gotmessage m = do
+                       liftIO $ debugMessage conn "P2P <" m
+                       runner (next (Just m))
+               in case connIhdl conn of
+                       P2PHandle h -> do
+                               v <- liftIO $ tryIOError $ getProtocolLine h
+                               case v of
+                                       Left e -> return $ Left $ ProtoFailureIOError e
+                                       Right Nothing -> protoerr
+                                       Right (Just l) -> case parseMessage l of
+                                               Just m -> gotmessage m
+                                               Nothing -> runner (next Nothing)
+                       P2PHandleTMVar mv -> 
+                               liftIO (atomically (takeTMVar mv)) >>= \case
+                                       Right m -> gotmessage m
+                                       Left _b -> protoerr
+       SendBytes len b p next ->
+               case connOhdl conn of
+                       P2PHandle h -> do
+                               v <- liftIO $ tryNonAsync $ do
+                                       ok <- sendExactly len b h p
+                                       hFlush h
+                                       return ok
+                               case v of
+                                       Right True -> runner next
+                                       Right False -> return $ Left $
+                                               ProtoFailureMessage "short data write"
+                                       Left e -> return $ Left $ ProtoFailureException e
+                       P2PHandleTMVar mv -> do
+                               liftIO $ atomically $ putTMVar mv (Left b)
+                               runner next
+       ReceiveBytes len p next ->
+               case connIhdl conn of
+                       P2PHandle h -> do
+                               v <- liftIO $ tryNonAsync $ receiveExactly len h p
+                               case v of
+                                       Right b -> runner (next b)
+                                       Left e -> return $ Left $
+                                               ProtoFailureException e
+                       P2PHandleTMVar mv -> 
+                               liftIO (atomically (takeTMVar mv)) >>= \case
+                                       Left b -> runner (next b)
+                                       Right _m -> return $ Left $
+                                               ProtoFailureMessage "protocol error"
        CheckAuthToken _u t next -> do
                let authed = connCheckAuth conn t
                runner (next authed)

diff --git a/Remote/Helper/Ssh.hs b/Remote/Helper/Ssh.hs
index bd7630e72a1d3f168649706f64ba71a9613ba501..1244d4719c359ea7484dd4d2bb69236ae4d3c712 100644 (file)
--- a/Remote/Helper/Ssh.hs
+++ b/Remote/Helper/Ssh.hs
@@ -264,8 +264,8 @@ openP2PShellConnection' r maxprotoversion bypass = do
                let conn = P2P.P2PConnection
                        { P2P.connRepo = Nothing
                        , P2P.connCheckAuth = const False
-                       , P2P.connIhdl = to
-                       , P2P.connOhdl = from
+                       , P2P.connIhdl = P2P.P2PHandle to
+                       , P2P.connOhdl = P2P.P2PHandle from
                        , P2P.connIdent = P2P.ConnIdent $
                                Just $ "git-annex-shell connection " ++ show pidnum
                        }

diff --git a/RemoteDaemon/Transport/Tor.hs b/RemoteDaemon/Transport/Tor.hs
index 0446d07d6aa91e0cb036d0bd3bc3099aaa28d012..515e3d333b9840187cb09fe96b2fb7f0a837dd3b 100644 (file)
--- a/RemoteDaemon/Transport/Tor.hs
+++ b/RemoteDaemon/Transport/Tor.hs
@@ -113,8 +113,8 @@ serveClient th@(TransportHandle _ _ rd) u r q = bracket setup cleanup start
                        let conn = P2PConnection
                                { connRepo = Just r
                                , connCheckAuth = (`isAllowedAuthToken` allowed)
-                               , connIhdl = h
-                               , connOhdl = h
+                               , connIhdl = P2PHandle h
+                               , connOhdl = P2PHandle h
                                , connIdent = ConnIdent $ Just "tor remotedaemon"
                                }
                        -- not really Client, but we don't know their uuid yet