Aynı türden bir işleme boru, 2 IO kaynakları

SORU

26 Mayıs 2013, Pazar

Aynı türden bir işleme boru, 2 IO kaynakları

GHC Haskell uygulama kullanarak stm, ağ-kanal ve boru benim, otomatik olarak runTCPServer kullanarak çatallı olan her yuva için bir tel var. İpliklerini yayın TChan kullanarak diğer teller ile iletişim kurabilirsiniz.

Kanal kurmak istiyorum bunu nasıl sergiliyor"": . zincir

enter image description here

Yani, biz burada iki kaynağı (her bağımlı için yardımcı olan kanallar) üretmek Packet nesne encoder kabul edecek ve dönüşmesi ByteString sonra göndermek yuva. İki giriş (performans kaygısı) verimli füzyon ile zorluk büyük bir miktar vardı.

Eğer birisi doğru yönde bana gelin eğer seviniriz.

Beni kaba bir girişim yapmadan bu soru post olacağından, daha önce burada çalıştım vereceğim;

/A TMChan bir Kaynak (closeable kanal) üreten bir işlev cherrypicked yazdım;

-- | Takes a generic type of STM chan and, given read and close functionality,
--   returns a conduit 'Source' which consumes the elements of the channel.
chanSource 
    :: (MonadIO m, MonadSTM m)
    => a                    -- ^ The channel
    -> (a -> STM (Maybe b)) -- ^ The read function
    -> (a -> STM ())        -- ^ The close/finalizer function
    -> Source m b
chanSource ch readCh closeCh = ConduitM pull
    where close     = liftSTM $ closeCh ch
          pull      = PipeM $ liftSTM $ readCh ch >>= translate
          translate = return . maybe (Done ()) (HaveOutput pull close)

Aynı şekilde, bir lavabo içine bir Chan dönüştürme fonksiyonu;

-- | Takes a stream and, given write and close functionality, returns a sink
--   which wil consume elements and broadcast them into the channel 
chanSink
    :: (MonadIO m, MonadSTM m)
    => a                 -- ^ The channel
    -> (a -> b -> STM()) -- ^ The write function
    -> (a -> STM())      -- ^ The close/finalizer function
    -> Sink b m ()
chanSink ch writeCh closeCh = ConduitM sink
    where close  = const . liftSTM $ closeCh ch
          sink   = NeedInput push close
          write  = liftSTM . writeCh ch
          push x = PipeM $ write x >> return sink

Sonra mergeSources basit; çatal 2 iş parçacığı (ben gerçekten yapmak istemiyor, ama ne halt) koyabilirsiniz kendi yeni öğeler bir liste ben de üreten bir kaynak;

-- | Merges a list of 'Source' objects, sinking them into a 'TMChan' and returns
--   a source which consumes the elements of the channel.
mergeSources
    :: (MonadIO m, MonadBaseControl IO m, MonadSTM m)
    => [Source (ResourceT m) a]             -- ^ The list of sources
    -> ResourceT m (Source (ResourceT m) a)
mergeSources sx = liftSTM newTMChan >>= liftA2 (>>) (fsrc sx) retn
    where push c s = s $$ chanSink c writeTMChan closeTMChan
          fsrc x c = mapM_ (\s -> resourceForkIO $ push c s) x
          retn c   = return $ chanSource c readTMChan closeTMChan

Bu işlevler typecheck yapımında başarılı iken, bu fonksiyonların kullanımı typecheck almak başarısız oldu;

-- | Helper which represents a conduit chain for each client connection
serverApp :: Application SessionIO
serverApp appdata = do
    use ssBroadcast >>= liftIO . atomically . dupTMChan >>= assign ssBroadcast
    -- appSource appdata $$ decoder $= protocol =$= encoder =$ appSink appdata
    mergsrc $$ protocol $= encoder =$ appSink appdata
    where chansrc = chanSource (use ssBroadcast) readTMChan closeTMChan
          mergsrc = mergeSources [appSource appdata $= decoder, chansrc]

-- | Structure which holds mutable information for clients
data SessionState = SessionState
    { _ssBroadcast     :: TMChan Packet -- ^ Outbound packet broadcast channel
    }

makeLenses ''SessionState

-- | A transformer encompassing both SessionReader and SessionState
type Session m = ReaderT SessionReader (StateT SessionState m)

-- | Macro providing Session applied to an IO monad
type SessionIO = Session IO

Nasıl olsa orada kusurlu olmak birçok Ara listeleri ve dönüşüm gibi bu yöntem görüyorum. Bu performans için iyi değil. Yol arayan.

PS. Kimden ne anlıyorum, bu bir yinelenen; Fusing conduits with multiple inputs , içinde bulunduğum durum her iki kaynak üretmek aynı tip ve umurumda değil hangi kaynak Packet nesne üretildiği sürece, ben değilim bekleyen bir süre başka nesneler hazır tüketilen.

PPS. Ben kullanımı (ve bu nedenle bilgi gereksinimi) örnek kodu Lens özür dilemek için.

CEVAP

5 Temmuz 2013, Cuma

Eğer herhangi bir yardım var mı bilmiyorum, ama Iain önerisi uygulamaya çalıştım ve kanallar kadar yakında durur mergeSources' bir türevi yaptı:

mergeSources' :: (MonadIO m, MonadBaseControl IO m)
              => [Source (ResourceT m) a] -- ^ The sources to merge.
              -> Int -- ^ The bound of the intermediate channel.
              -> ResourceT m (Source (ResourceT m) a)
mergeSources' sx bound = do
    c <- liftSTM $ newTBMChan bound
    mapM_ (\s -> resourceForkIO $
                    s $$ chanSink c writeTBMChan closeTBMChan) sx
    return $ sourceTBMChan c

(Bu basit toplama here kullanılabilir).

mergeSources (inanmayarak onları almak olabilir bir şey anlamadım) sürümü için bazı yorumlar:

...TBMChan yerine ...TMChan kullanarak tehlikeli görünüyor. Eğer yazarlar okuyucu daha hızlı, yığın darbe olacak. Diyagramı bakarak bu kolay eğer TCP eş veri yeterince hızlı okumuyor diye bir şey olabilir gibi görünüyor. Kesinlikle ...TBMChan belki büyük ama sınırlı ile bağlı kullanmak istiyorum.
MonadSTM m kısıtlama gerekmez. Tüm STM şeyler IO ile içine sarılmış
```
liftSTM = liftIO . atomically
```
Belki de bu serverApp 27 *kullanırken size biraz yardımcı olacaktır.
Sadece kozmetik bir sorun buldum
```
liftSTM newTMChan >>= liftA2 (>>) (fsrc sx) retn
```
çok zor (->) r monad liftA2 kullanımı yüzünden okumak. Derim
```
do
    c <- liftSTM newTMChan
    fsrc sx c
    retn c
```
daha uzun, ama okuması çok daha kolay olurdu.