Fwd: [Haskell] ANNOUNCE: Process library (for dataflow-oriented programming?)
This is a forwarded message From: Bulat Ziganshin <bulatz@HotPOP.com> To: haskell@haskell.org Date: Thursday, December 08, 2005, 1:36:05 AM Subject: [Haskell] ANNOUNCE: Process library (for dataflow-oriented programming?) ===8<==============Original message text=============== Hello haskell, Joel's program (discussed in cafe), which now uses MVars instead of Channels to send data between threads, may be a good example of dataflow-driven program: it consists of many hundreds of threads and when one thread sends data to another through MVar, this thread in most cases goes to sleep until receiving thread will process previous value of this MVar. so, threads are waked up and asleep according to passing values between them, and the whole program executes in order defined by these data dependencies, as opposite to the order of program statements one year ago i developed small library, which can be helpful if you want to use such style of programming. its ideas are modelled after Unix pipes, which are widely used to assemble complex data processing "engines" from simple "details". really this library is very thin layer over direct using of forkOS, channels and MVars; nevertheless, is is very convenient and beatiful you can download library as http://freearc.narod.ru/Process.tar.gz this page also contains sources of my program where you can find examples of using library in real toy :) below is a guide to library usage to create pipe, which contains 3 processes - "producer", "transformer" and "consumer": runP ( producer |> transformer |> consumer ) each process in pipe runned in separate Haskell thread. process is represented by ordinary Haskell function which gets an additional parameter - handle, which can be used to receive data from previous process in pipe (using receiveP) and send data to the next process (using sendP). for example, abovementioned processes can be implemented as: producer handle = mapM_ (sendP handle) [1..10] transformer handle = replicateM_ 10 $ do x <- receiveP handle sendP handle (x*2) consumer handle = replicateM_ 10 $ do x <- receiveP handle print x if first process in pipe tries to use receiveP or last process in pipe tries to use sendP, then run-time exception is generated. number of processes in pipe can be arbitrary. because each process is just ordinary Haskell function, you can add additional parameters to processes when constructing pipes: runP ( producer |> multiple 2 |> multiple 3 |> consumer ) multiple n handle = replicateM_ 10 $ do x <- receiveP handle sendP handle (x*n) moreover, you can construct pipe or part of it as ordinary data value, which then can be runned by runP: let pipe = case multipliers of [x] -> multiple x [x,y] -> multiple x |> multiple y [x,y,z] -> multiple x |> multiple y |> multiple z _ -> \handle -> fail "Zero or too much multipliers" runP ( producer |> pipe |> consumer ) there is also "back channel", which can be used to "return" data to previous process in the pipe, its operations is send_backP and receive_backP. it can be used to return acknowledgments, synchronize processes or to return resources back. brief example of its usage: producer: sendP pipe (buf,len) consumer: ; ; (buf,len) <- receiveP pipe ; hPutBuf file buf len ; send_backP pipe () receive_backP pipe ; --now we know that buf is free ; (i organized lines to show execution order) if processes joined in pipe with "|>" then channel between them uses MVar, so at any moment it may contain no more than 1 element. if channel between two processes is created with "|>>>" then Chan is used, which can contain arbitrary number of data items. be careful with such channels, because they can grow to unlimited size. "|>" and "|>>>" can be arbitrarily combined in one pipe: runP ( producer |>>> multiple 2 |> multiple 3 |>>> consumer ) back channel (used by send_backP and receive_backP) are always multi-element (uses Chan) runP returns when all processes in pipe are finished. if any process in pipe generates uncaught exception, then all processes in pipe are killed and this exception is re-raised in thread called runP pipe or single process can also be runned in background using runAsyncP: handle <- runAsyncP (multiple 2) handle returned here can be used to interact with first and last processes in pipe, in contrast to runP: handle <- runAsyncP (multiple 2) sendP handle 1 res <- receiveP handle of course, pipe runned asynchronously is not required to perform input, output, or both: handle <- runAsyncP ( producer |> transformer ) handle <- runAsyncP ( transformer |> consumer ) handle <- runAsyncP ( producer |> transformer |> consumer ) currently channels to "open ends" of background pipe are always one-element (uses MVars). types of channels inside pipe are determined, as usual, by using the "|>" or "|>>>" operator you can wait for finishing of asynchronous pipe with "joinP handle" you can "replace" process, which is runned in pipe, with another process or pipe by using runFuncP: transformer handle = runFuncP ( multiple 2 |> multiple 3 ) (receiveP handle) (send_backP handle) (sendP handle) (receive_backP handle) actually, runFuncP just executes its pipe using 4 supported functions for interaction with first and last pipe processes. to make this obvious i will say how runP can be implemented in terms of runFuncP: runP p = runFuncP p (error "First process in runP tried to receive") (error "First process in runP tried to send_back") (error "Last process in runP tried to send") (error "Last process in runP tried to receive_back") runFuncP returns when all processes in its pipe are finished. using it, you can create unlimited number of scenarios: running several runFuncP sequentially to deal with different parts of your data, process part of data itself and part with runFuncP, "redirect" your input to process runned by runFuncP but consume output from runFuncP in other way: consumer handle = runFuncP ( multiple 4 ) (receiveP handle) undefined print -- output action undefined runFuncP can also be used in other cases when we need to run single process or whole pipe "imitating" its interaction with "external world" by some functions: runFuncP transformer readLn -- input action undefined print -- output action undefined it will also be interesting to have some functions which can just insert "filtering" process or pipe on input or output side of current process: transformer channel = do channel <- insertInputFilterP (multiple 2) channel channel <- insertOutputFilterP (multiple 4) channel x <- receiveP channel sendP channel (x+1) ... but i don't need it and therefore this currently is not implemented the library also don't contains routines to check that input is ready or output can be done (because i don't need it and it is a bad programming style), nor routines to check for EOF (i prefer to encode this explicitly in structure of data sent across channels; you also can wrap your data in Maybe type and use Nothing to encode EOF) -- Best regards, Bulat mailto:bulatz@HotPOP.com _______________________________________________ Haskell mailing list Haskell@haskell.org http://www.haskell.org/mailman/listinfo/haskell ===8<===========End of original message text=========== -- Best regards, Bulat mailto:bulatz@HotPOP.com
Bulat, How is your library licensed? How can a process maintain internal state? How would I use your library to code a socket reader/writer that writes received events to the socket and propagates back anything that is received? The producer/consumer in front of this network client would be another process that analyzes the events sent back to it and produces events based on the analysis. How would I use it to launch a few network clients that seat there and process events until they decided to quit? The whole program needs to stay up until the last network client has exited. The pipeline to me looks like this: <-> Bot <-> Socket client ... Server / Bot launcher --- <-> Bot <-> Socket client ... Server \ <-> Bot <-> Socket client ... Server Where bot launcher starts a predefined # of bots and collects results sent back by each one. I think your library looks a bit like Yampa in that your processes are somewhat like signal functions. Thanks, Joel -- http://wagerlabs.com/
Hello Joel, Tuesday, December 13, 2005, 3:04:11 PM, you wrote: JR> How is your library licensed? is costs many megabucks because it's very complex proprietary design where some functions reach whole 12 lines! :) of course, you can do what you want with this library. may be the better way is to write your won, stealing one ot two ideas from mine you can find another interesting works in: http://www-i2.informatik.rwth-aachen.de/~stolz/Haskell/CA.hs http://www-i2.informatik.rwth-aachen.de/Research/distributedHaskell/pbdhs-20... (this one seems to be especially interesting for you, providing "ports" - i think, in Erlang style) http://www-i2.informatik.rwth-aachen.de/Research/distributedHaskell/network.... http://quux.org/devel/missingh/missingh_0.12.0.tar.gz (see Logging, Network, Threads directories) JR> How can a process maintain internal state? process in my lib is just an ordinary Haskell function and therefore this is done as in any other Haskell functions :) my examples are easified - to not bother with EOF i just organized in each process a loop which sends and/or receives just 10 messages. in my real program data sent between process are defined by structures like this: data Message = FileStarted String | FileData String | DataEnd so typical communication scenario is: sendP h (FileStart "1") ; sendP h (FileData "abc") ; sendP h (FileData "def") ; sendP h (FileData "ghi") sendP h (FileStart "2") ; sendP h (FileData "qwer") sendP h (FileStart "3") ; sendP h (FileData "123") sendP h DataEnd and each function realizing process finishes only when this process is done. sender process organizes cycle which reads files and sends their data to the channel. receiver process organizes cycle until `DataEnd` is received in one phrase, it's just the same organization as in your own program :))) JR> How would I use your library to code a socket reader/writer that JR> writes received events to the socket and propagates back anything JR> that is received? JR> The producer/consumer in front of this network client would be JR> another process that analyzes the events sent back to it and produces JR> events based on the analysis. i don't understand your questions JR> How would I use it to launch a few network clients that seat there JR> and process events until they decided to quit? The whole program JR> needs to stay up until the last network client has exited. JR> The pipeline to me looks like this: JR> <-> Bot <-> Socket client ... Server JR> / JR> Bot launcher --- <-> Bot <-> Socket client ... Server JR> \ JR> <-> Bot <-> Socket client ... Server JR> Where bot launcher starts a predefined # of bots and collects results JR> sent back by each one. my lib is not appropriate for yor task, because it is oriented to easify creation of processes which have only one input. but your main thread must receive data from all bots, and bot must receive data from two sources. the decision depends on the strategy of mixing these inputs - will it be fair FIFO or more advanced schema? if it's a FIFO then something like this (i'm skipped only exceptions processing and creating socket-reader process inside of each bot - writing to socket must be performed by bot itself): {-# OPTIONS_GHC -cpp #-} import Control.Concurrent import Control.Exception import Control.Monad import Data.Array import Data.Char import Data.Either import Data.HashTable import Data.IORef import Data.List import Data.Maybe import Data.Word import Debug.Trace import Foreign.C.String import Foreign.Marshal.Alloc import Foreign.Marshal.Array import Foreign.Marshal.Pool import Foreign.Marshal.Utils import Foreign.Ptr import Text.Regex import System.IO.Unsafe ----------------------------------------------------------------- --- Bot launcher implementation --------------------------------- ----------------------------------------------------------------- main = do (sendToMain, receiveFromBots) <- createChannel bots <- foreach [1..10] $ createProcess . bot sendToMain mapM_ (`sendToProcess` "Wake up, Neo!") bots while receiveFromBots (/="I want to stop the Matrix!") print -- .... mapM_ killProcess bots -- or, if you are more humane - "mapM_ waitProcessDie bots" :) ----------------------------------------------------------------- --- Bot implementation ------------------------------------------ ----------------------------------------------------------------- bot sendToMain n receiveMessagesForMe = do forever $ do x <- receiveMessagesForMe case x of "Wake up, Neo!" -> sendToMain$ show n++": I'm not sleeping!" "Are you wanna coffee?" -> sendToMain$ show n++": Yes, it is!" yield sendToMain "I want to stop the Matrix!" ----------------------------------------------------------------- --- Process implementation details ------------------------------ ----------------------------------------------------------------- -- |Abstract type for all of our processes data Process a = Process { pid :: ThreadId , sender :: a -> IO () -- action which must be used to send sata to this process , finished :: MVar () -- filled when this process is finisjed } -- Operations on processes -- |Create process and return structure Process to communicate with it -- The `process` function given are called with a parameter, which -- represent an action which must be used by process to receive it's -- data createProcess process = do finished <- newEmptyMVar (sendToMe, receiveMessagesForMe) <- createChannel pid <- forkIO $ do process receiveMessagesForMe `finally` putMVar finished () return Process { pid = pid , sender = sendToMe , finished = finished } killProcess = killThread.pid waitProcessDie = readMVar.finished sendToProcess = sender ----------------------------------------------------------------- --- Channel implementation details ------------------------------ ----------------------------------------------------------------- #if 1 -- Channels implemented via MVar createChannel = do c <- newEmptyMVar return (putMVar c, takeMVar c) #else -- Channels implemented via Chan createChannel = do c <- newChan return (writeChan c, readChan c) #endif ----------------------------------------------------------------- --- Random utility functions ------------------------------------ ----------------------------------------------------------------- while inp cond out = do x <- inp if (cond x) then do out x while inp cond out else return x forever action = do action forever action foreach = flip mapM JR> I think your library looks a bit like Yampa in that your processes JR> are somewhat like signal functions. yes, it is close things - data-driven execution and signal processing -- Best regards, Bulat mailto:bulatz@HotPOP.com
participants (2)
-
Bulat Ziganshin -
Joel Reymont