我并不认为Haskell正在处理消息传递之类的事情,但是这些结果(我正在查看比其他任何事情好4-5倍的Linux)让我对此提出质疑.一些讨论让我进入了Cloud Haskell,它实际上似乎是一个有用的编写良好的分布式系统相关库的集合(对于这个和静态指针的工作,对Well-Typed和Tweag I/O的严重赞誉).
我对Actors和消息传递的体验源于我对Akka的使用.我正在尝试理解Cloud Haskell能够做什么来确定我是否希望将其用于可扩展性,效率和可靠性非常重要的潜在项目.所以:
我意识到这个问题有点宽泛,但我认为如果有人对Cloud Haskell有一些经验,他们可能会很容易地解决所有这些问题 - 这个问题对于那些搜索一般信息的人来说会更有用关于Cloud Haskell与Akka的未来.
问题
你好!我在Cloud Haskell中编写了一个简单的Server-Worker程序.问题是,当我尝试创建时ManagedProcess,在服务器发送步骤之后,我的示例即使在使用时也会永久挂起callTimeout(在100毫秒后应该会中断).代码很简单,但我发现它没有任何问题.
我也在邮件列表上发布了这个问题,但就我所知的SO社区而言,我在这里可以更快地得到答案.如果我从邮件列表中得到答案,我也会在这里发帖.
源代码
的Worker.hs:
{-# LANGUAGE DeriveDataTypeable #-}
{-# LANGUAGE ExistentialQuantification #-}
{-# LANGUAGE DeriveGeneric #-}
{-# LANGUAGE TemplateHaskell #-}
module Main where
import Network.Transport (EndPointAddress(EndPointAddress))
import Control.Distributed.Process hiding (call)
import Control.Distributed.Process.Platform hiding (__remoteTable)
import Control.Distributed.Process.Platform.Async
import Control.Distributed.Process.Platform.ManagedProcess
import Control.Distributed.Process.Platform.Time
import Control.Distributed.Process.Platform.Timer (sleep)
import Control.Distributed.Process.Closure (mkClosure, remotable)
import Network.Transport.TCP (createTransport, defaultTCPParameters)
import Control.Distributed.Process.Node hiding (call)
import Control.Concurrent (threadDelay)
import GHC.Generics (Generic)
import Data.Binary (Binary)
import Data.Typeable (Typeable)
import Data.ByteString.Char8 (pack)
import System.Environment …我一直在玩Cloud Haskell.我在hackage文档中注意到有一种应用程序界面.但特别是我正在尝试closurePure使用以下签名查找或编写函数:
closurePure :: (Typeable a, Binary a) => a -> Closure a
这基本上是纯粹的限制版本.
虽然Closure数据类型本身是抽象的,但以下closure提供了:
closure :: Static (ByteString -> a) -> ByteString -> Closure a
所以我可以做到这一点:
closurePure :: (Typeable a, Binary a) => a -> Closure a
closurePure x = closure ??? (encode x)
问题是在什么地方放置???.
我的第一次尝试如下:
myDecode :: (Typeable a, Binary a) => Static (ByteString -> a)
myDecode = staticPtr (static decode)
在教程(http://haskell-distributed.github.io)之后,这段代码 - 在调用devGHCi之后 - 让一个CPU核心旋转到100%并无限期地保持在该水平.这似乎是由于呼吁而发生的closeTransport.
但是,逐行评估GHCi中的代码并没有显示出这种效果.那么差异来自何处以及如何才能正确清理？
import Network.Transport (closeTransport)
import Network.Transport.TCP (createTransport, defaultTCPParameters)
import Control.Distributed.Process.Node (newLocalNode, closeLocalNode, initRemoteTable)
dev :: IO ()
dev = do
Right t <- createTransport "localhost" "9000" defaultTCPParameters
n <- newLocalNode t initRemoteTable
-- do the work
closeLocalNode n -- <- doesn't affect the observed behaviour
closeTransport t
我正在阅读distributed-process相关软件包的源代码.在distributed-static我们的定义StaticLabel:
data StaticLabel =
StaticLabel String
| StaticApply !StaticLabel !StaticLabel
| StaticPtr SDynamic
deriving (Eq, Ord, Typeable, Show)
然后将其定义Static为newtype包装器
StaticLabel,附加了一个用于类型安全的幻像变量:
newtype Static a = Static StaticLabel
deriving (Eq, Ord, Typeable, Show)
我没有任何问题StaticApply,它只是将两个静态值混为一谈.然而,StaticLabel并StaticPtr似乎以实现不同的方式相同的目标.
如果我们继续使用,StaticLabel我们只需要/转换一个String标签,然后可以用来查找以下Dynamic值RemoteTable:
newtype RemoteTable = RemoteTable (Map String Dynamic)
在哪里Dynamic(定义rank1dynamic):
data Dynamic = Dynamic TypeRep GHC.Any
作为几乎是一样SDynamic包含在StaticPtr:
data …我使用network-transport-tcp-0.5.1进行creatipn TCP传输.
resultCT <- createTransport host port defaultTCPParameters
但是现在我使用的是network-transport-tcp-0.6.0软件包并因为新类型而出错:
createTransportExposeInternals
:: N.HostName
-> N.ServiceName
-> (N.ServiceName -> (N.HostName, N.ServiceName))
-> TCPParameters
-> IO (Either IOException (Transport, TransportInternals))
如何使用(N.ServiceName - >(N.HostName,N.ServiceName))的默认值？也许有一些例子？谢谢!
我正在使用Cloud Haskell进行消息处理.我也使用一般的monad变换器堆栈(底部有IO)进行状态,配置等的一般跟踪.
我遇到了我必须unsafePerformIO在Process monad中使用的情况.我在描述下面的情况.请记住,这是一个非常人为的例子,可以简化并呈现问题的关键
data AppConfig
data AppState
type App = ReaderT AppConfig (StateT AppState IO)
runApp :: App a -> Int -> IO (a, AppState)
runApp k maxDepth =
let config = AppConfig maxDepth
state = AppState 0
in runStateT (runReaderT k config) state
msgHandler :: App ()
msgHandler = -- some message handling logic here --
runServer :: Process ()
runServer = do
let run handler = return $ unsafePerformIO $ runApp handler
(_,_) <- receiveWait …