mirror of
https://github.com/urbit/shrub.git
synced 2024-12-21 01:41:37 +03:00
king: packet tests, scry tqueue
This commit is contained in:
pilfer-pandex
parent
45ad5da9e9
commit
def318abd9
@ -235,12 +235,11 @@ ames env who isFake scry enqueueEv stderr = (initialEvents, runAmes)
|
||||
scryVersion \v -> do
|
||||
v0 <- readTVarIO versSlot
|
||||
atomically $ writeTVar versSlot (Just v)
|
||||
putStrLn "wow"
|
||||
if (v0 == Just v)
|
||||
then logInfo $ displayShow ("ames: proto version unchanged at", v)
|
||||
else stderr ("ames: protocol version now " <> tshow v)
|
||||
|
||||
threadDelay (1_000_000) -- 10m
|
||||
threadDelay (10 * 60 * 1_000_000) -- 10m
|
||||
|
||||
queuePacketsThread :: HasLogFunc e
|
||||
=> TVar Word
|
||||
@ -335,7 +334,6 @@ ames env who isFake scry enqueueEv stderr = (initialEvents, runAmes)
|
||||
wen <- io Time.now
|
||||
let nkt = MkKnot $ tshow $ Time.MkDate wen
|
||||
let pax = Path $ "ax" : MkKnot (tshow who) : "" : nkt : p
|
||||
putStrLn ("scrying for " <> tshow pax)
|
||||
let kon = runRIO env . \case
|
||||
Just (_, fromNoun @n -> Just v) -> k (Just v)
|
||||
Just (_, n) -> do
|
||||
|
||||
@ -22,6 +22,7 @@ data Packet = Packet
|
||||
, pktOrigin :: Maybe AmesDest
|
||||
, pktContent :: Bytes
|
||||
}
|
||||
deriving Eq
|
||||
|
||||
instance Show Packet where
|
||||
show Packet {..}
|
||||
|
||||
@ -277,7 +277,6 @@ pier (serf, log) vSlog startedSig = do
|
||||
-- TODO Instead of using a TMVar, pull directly from the IO driver
|
||||
-- event sources.
|
||||
computeQ :: TMVar RunReq <- newEmptyTMVarIO
|
||||
|
||||
persistQ :: TQueue (Fact, FX) <- newTQueueIO
|
||||
executeQ :: TQueue FX <- newTQueueIO
|
||||
saveSig :: TMVar () <- newEmptyTMVarIO
|
||||
@ -305,7 +304,7 @@ pier (serf, log) vSlog startedSig = do
|
||||
atomically $ Term.trace muxed txt
|
||||
logOther "serf" (display $ T.strip txt)
|
||||
|
||||
scrySig <- newEmptyTMVarIO
|
||||
scryQ <- newTQueueIO
|
||||
onKill <- view onKillPierSigL
|
||||
|
||||
-- Our call above to set the logging function which echos errors from the
|
||||
@ -315,7 +314,7 @@ pier (serf, log) vSlog startedSig = do
|
||||
let compute = putTMVar computeQ
|
||||
let execute = writeTQueue executeQ
|
||||
let persist = writeTQueue persistQ
|
||||
let scry = \w b g k -> putTMVar scrySig (w, b, g, k)
|
||||
let scry = \w b g k -> writeTQueue scryQ (w, b, g, k)
|
||||
let sigint = Serf.sendSIGINT serf
|
||||
|
||||
(bootEvents, startDrivers) <- do
|
||||
@ -328,7 +327,7 @@ pier (serf, log) vSlog startedSig = do
|
||||
let computeConfig = ComputeConfig { ccOnWork = takeTMVar computeQ
|
||||
, ccOnKill = onKill
|
||||
, ccOnSave = takeTMVar saveSig
|
||||
, ccOnScry = takeTMVar scrySig
|
||||
, ccOnScry = readTQueue scryQ
|
||||
, ccPutResult = persist
|
||||
, ccShowSpinner = Term.spin muxed
|
||||
, ccHideSpinner = Term.stopSpin muxed
|
||||
|
||||
@ -12,12 +12,14 @@ import Urbit.EventLog.LMDB
|
||||
import Urbit.King.Config
|
||||
import Urbit.Noun
|
||||
import Urbit.Noun.Time
|
||||
import Urbit.Prelude
|
||||
import Urbit.Prelude hiding (elements)
|
||||
import Urbit.Vere.Ames
|
||||
import Urbit.Vere.Ames.Packet
|
||||
import Urbit.Vere.Pier.Types
|
||||
import Urbit.Vere.Ports
|
||||
|
||||
import Control.Concurrent (runInBoundThread)
|
||||
import Data.Serialize (decode, encode)
|
||||
import Data.LargeWord (LargeKey(..))
|
||||
import GHC.Natural (Natural)
|
||||
import Network.Socket (tupleToHostAddress)
|
||||
@ -26,155 +28,14 @@ import Urbit.King.App (HasKingId(..))
|
||||
import qualified Urbit.EventLog.LMDB as Log
|
||||
import qualified Urbit.Noun.Time as Time
|
||||
|
||||
|
||||
--------------------------------------------------------------------------------
|
||||
|
||||
type HasAmes e =
|
||||
( HasLogFunc e
|
||||
, HasNetworkConfig e
|
||||
, HasKingId e
|
||||
, HasPortControlApi e)
|
||||
|
||||
-- Utils -----------------------------------------------------------------------
|
||||
|
||||
pid :: KingId
|
||||
pid = KingId 0
|
||||
|
||||
turfEf :: NewtEf
|
||||
turfEf = NewtEfTurf (0, ()) []
|
||||
|
||||
sendEf :: Galaxy -> Wen -> Bytes -> NewtEf
|
||||
sendEf g w bs = NewtEfSend (0, ()) (EachYes g) bs
|
||||
|
||||
data NetworkTestApp = NetworkTestApp
|
||||
{ _ntaLogFunc :: !LogFunc
|
||||
, _ntaNetworkConfig :: !NetworkConfig
|
||||
, _ntaPortControlApi :: !PortControlApi
|
||||
, _ntaKingId :: !Word16
|
||||
}
|
||||
|
||||
makeLenses ''NetworkTestApp
|
||||
|
||||
instance HasLogFunc NetworkTestApp where
|
||||
logFuncL = ntaLogFunc
|
||||
|
||||
instance HasNetworkConfig NetworkTestApp where
|
||||
networkConfigL = ntaNetworkConfig
|
||||
|
||||
instance HasKingId NetworkTestApp where
|
||||
kingIdL = ntaKingId
|
||||
|
||||
instance HasPortControlApi NetworkTestApp where
|
||||
portControlApiL = ntaPortControlApi
|
||||
|
||||
runNetworkApp :: RIO NetworkTestApp a -> IO a
|
||||
runNetworkApp =
|
||||
runRIO NetworkTestApp
|
||||
{ _ntaLogFunc = mkLogFunc (\_ _ _ _ -> pure ())
|
||||
, _ntaKingId = 34
|
||||
, _ntaPortControlApi = buildInactivePorts
|
||||
, _ntaNetworkConfig = NetworkConfig { _ncNetMode = NMNormal
|
||||
, _ncAmesPort = Nothing
|
||||
, _ncNoAmes = False
|
||||
, _ncNoHttp = False
|
||||
, _ncNoHttps = False
|
||||
, _ncHttpPort = Nothing
|
||||
, _ncHttpsPort = Nothing
|
||||
, _ncLocalPort = Nothing
|
||||
}
|
||||
}
|
||||
|
||||
runGala
|
||||
:: forall e
|
||||
. HasAmes e
|
||||
=> Word8
|
||||
-> RAcquire e (TQueue EvErr, NewtEf -> IO ())
|
||||
runGala point = do
|
||||
env <- ask
|
||||
que <- newTQueueIO
|
||||
cry <- newTQueueIO
|
||||
flip mkRAcquire cancel $ async $ forever $ do
|
||||
act <- atomically $ readTQueue cry
|
||||
putStrLn "taking action"
|
||||
io act
|
||||
let enqueue = \p -> writeTQueue que p $> Intake
|
||||
let (_, runAmes) =
|
||||
ames env (fromIntegral point) True (scry cry) enqueue noStderr
|
||||
cb <- runAmes
|
||||
io (cb turfEf)
|
||||
pure (que, cb)
|
||||
where
|
||||
noStderr _ = pure ()
|
||||
scry :: TQueue (IO ()) -> Time.Wen -> Gang -> Path -> (Maybe (Term, Noun) -> IO ()) -> STM ()
|
||||
scry q _ _ (Path p) cb = writeTQueue q $ case unKnot <$> p of
|
||||
["ax",_,"",_,"protocol","version"] -> (putStrLn "yes" >>) $ cb $ Just
|
||||
( error "ames test: should not depend on scry term"
|
||||
, A 0
|
||||
) :: IO ()
|
||||
["ax",_,"",_,"peers",ship,"forward-lane"] -> cb $ Just
|
||||
( error "ames test: should not depend on scry term"
|
||||
, toNoun [fromIntegral $ hash ship :: Word32]
|
||||
)
|
||||
pax -> error ("ames test: fell scry " <> show pax)
|
||||
|
||||
|
||||
waitForPacket :: TQueue EvErr -> Bytes -> IO Bool
|
||||
waitForPacket q val = go
|
||||
where
|
||||
go = atomically (readTQueue q) >>= \case
|
||||
EvErr (EvBlip (BlipEvNewt (NewtEvBorn (_, ()) ()))) _ -> go
|
||||
EvErr (EvBlip (BlipEvAmes (AmesEvHear () _ bs))) _ -> pure (bs == val)
|
||||
_ -> pure False
|
||||
|
||||
runRAcquire :: RAcquire e a -> RIO e a
|
||||
runRAcquire acq = rwith acq pure
|
||||
|
||||
sendThread :: (NewtEf -> IO ()) -> (Galaxy, Bytes) -> RAcquire e ()
|
||||
sendThread cb (to, val) = void $ mkRAcquire start cancel
|
||||
where
|
||||
start = async $ forever $ do threadDelay 1_000
|
||||
wen <- io $ now
|
||||
io $ cb (sendEf to wen val)
|
||||
threadDelay 10_000
|
||||
|
||||
zodSelfMsg :: Property
|
||||
zodSelfMsg = forAll arbitrary (ioProperty . runNetworkApp . runTest)
|
||||
where
|
||||
runTest :: (HasAmes e) => Bytes -> RIO e Bool
|
||||
runTest val = runRAcquire $ do
|
||||
env <- ask
|
||||
(zodQ, zod) <- runGala 0
|
||||
() <- sendThread zod (0, val)
|
||||
liftIO (waitForPacket zodQ val)
|
||||
|
||||
twoTalk :: Property
|
||||
twoTalk = forAll arbitrary (ioProperty . runNetworkApp . runTest)
|
||||
where
|
||||
runTest :: (HasAmes e) => (Word8, Word8, Bytes) -> RIO e Bool
|
||||
runTest (aliceShip, bobShip, val) =
|
||||
if aliceShip == bobShip
|
||||
then pure True
|
||||
else go aliceShip bobShip val
|
||||
|
||||
go :: (HasAmes e) => Word8 -> Word8 -> Bytes -> RIO e Bool
|
||||
go aliceShip bobShip val = runRAcquire $ do
|
||||
(aliceQ, alice) <- runGala aliceShip
|
||||
(bobQ, bob) <- runGala bobShip
|
||||
sendThread alice (Patp bobShip, val)
|
||||
sendThread bob (Patp aliceShip, val)
|
||||
liftIO (waitForPacket aliceQ val >> waitForPacket bobQ val)
|
||||
packetSplitMorphism :: Packet -> Bool
|
||||
packetSplitMorphism p = (decode . encode) p == Right p
|
||||
|
||||
tests :: TestTree
|
||||
tests =
|
||||
testGroup "Ames"
|
||||
[ localOption (QuickCheckTests 10) $
|
||||
testProperty "Zod can send a message to itself" $
|
||||
zodSelfMsg
|
||||
|
||||
-- TODO Why doesn't this work in CI?
|
||||
-- , localOption (QuickCheckTests 10) $
|
||||
-- testProperty "Two galaxies can talk" $
|
||||
-- twoTalk
|
||||
[ testProperty "Packet coding looks good" $
|
||||
packetSplitMorphism
|
||||
]
|
||||
|
||||
|
||||
@ -230,7 +91,26 @@ instance Arbitrary AmesAddress where
|
||||
arbitrary = AAIpv4 <$> arb <*> arb
|
||||
|
||||
instance Arbitrary Ship where
|
||||
arbitrary = Ship <$> arb
|
||||
arbitrary = Ship <$> elements
|
||||
[ 0
|
||||
, 42
|
||||
, 256
|
||||
, 24_530
|
||||
, 2_071_856_128
|
||||
, 2_824_325_100
|
||||
, 430_648_908_188_615_680
|
||||
, 2^60 + 1337
|
||||
]
|
||||
|
||||
instance Arbitrary LogIdentity where
|
||||
arbitrary = LogIdentity <$> arb <*> arb <*> arb
|
||||
|
||||
instance Arbitrary Packet where
|
||||
arbitrary = do
|
||||
pktVersion <- suchThat arb (< 8)
|
||||
pktEncrypted <- arb
|
||||
pktSndr <- arb
|
||||
pktRcvr <- arb
|
||||
pktOrigin <- arb
|
||||
pktContent <- arb
|
||||
pure Packet {..}
|
||||
|
||||
Loading…
Reference in New Issue
Block a user