diff --git a/Etage.cabal b/Etage.cabal new file mode 100644 --- /dev/null +++ b/Etage.cabal @@ -0,0 +1,51 @@ +Name: Etage +Version: 0.1 +Synopsis: A general data-flow framework +Description: A general data-flow framework featuring nondeterminism and neurological pseudo-terminology. It can be used for + example for data-flow computations or event propagation networks. It tries hard to aide type checking and to + allow proper initialization and cleanup so that interfaces to input and output devices (data or events producers or + consumers) can be made (so that created models\/systems\/networks can be used directly in real world applications, for + example robots). + . + Its main goal is to model complex neural networks with more biological realism. Namely that impulses do + take time to travel and neuron responses are also not instantaneous. And of course that neural systems are in + its base nondeterministic and that some level of determinism is build upon that. All this of course makes reasoning + about such networks even harder (impossible?). + . + This framework is in fact just a simple abstraction of Haskell threads and data passing between them through + channels with threads' initialization and cleanup wrapped into a Haskell type class. + . + Feel free to contribute or suggest additional features or (example) programs or to create interfaces to other modules. +License: LGPL-3 +License-file: LICENSE +Author: Mitar Milutinovic +Maintainer: mitar.haskell@tnode.com +Copyright: (c) 2010 Mitar Milutinovic +Category: Control +Build-type: Simple +Cabal-version: >= 1.8 +Stability: experimental +Homepage: http://mitar.tnode.com + +Library + Exposed-modules: Control.Etage, + Control.Etage.Dump, + Control.Etage.Sequence, + Control.Etage.Worker, + Control.Etage.Timeout, + Control.Etage.Function, + Control.Etage.Fail + Build-depends: base >= 4.3 && < 5, + mtl >= 1.1 && < 3, + random > 1.0 && < 2, + unix >= 2.4 && < 3, + time >= 1.1 && < 2, + operational >= 0.2 && < 1, + ghc >= 7.0.2 + Other-modules: Control.Etage.Internals, + Control.Etage.Externals, + Control.Etage.Propagate, + Control.Etage.Incubator, + Control.Etage.Chan + HS-source-dirs: lib + GHC-options: -Wall diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,165 @@ + GNU LESSER GENERAL PUBLIC LICENSE + Version 3, 29 June 2007 + + Copyright (C) 2007 Free Software Foundation, Inc. + Everyone is permitted to copy and distribute verbatim copies + of this license document, but changing it is not allowed. + + + This version of the GNU Lesser General Public License incorporates +the terms and conditions of version 3 of the GNU General Public +License, supplemented by the additional permissions listed below. + + 0. Additional Definitions. + + As used herein, "this License" refers to version 3 of the GNU Lesser +General Public License, and the "GNU GPL" refers to version 3 of the GNU +General Public License. + + "The Library" refers to a covered work governed by this License, +other than an Application or a Combined Work as defined below. + + An "Application" is any work that makes use of an interface provided +by the Library, but which is not otherwise based on the Library. +Defining a subclass of a class defined by the Library is deemed a mode +of using an interface provided by the Library. + + A "Combined Work" is a work produced by combining or linking an +Application with the Library. The particular version of the Library +with which the Combined Work was made is also called the "Linked +Version". + + The "Minimal Corresponding Source" for a Combined Work means the +Corresponding Source for the Combined Work, excluding any source code +for portions of the Combined Work that, considered in isolation, are +based on the Application, and not on the Linked Version. + + The "Corresponding Application Code" for a Combined Work means the +object code and/or source code for the Application, including any data +and utility programs needed for reproducing the Combined Work from the +Application, but excluding the System Libraries of the Combined Work. + + 1. Exception to Section 3 of the GNU GPL. + + You may convey a covered work under sections 3 and 4 of this License +without being bound by section 3 of the GNU GPL. + + 2. Conveying Modified Versions. + + If you modify a copy of the Library, and, in your modifications, a +facility refers to a function or data to be supplied by an Application +that uses the facility (other than as an argument passed when the +facility is invoked), then you may convey a copy of the modified +version: + + a) under this License, provided that you make a good faith effort to + ensure that, in the event an Application does not supply the + function or data, the facility still operates, and performs + whatever part of its purpose remains meaningful, or + + b) under the GNU GPL, with none of the additional permissions of + this License applicable to that copy. + + 3. Object Code Incorporating Material from Library Header Files. + + The object code form of an Application may incorporate material from +a header file that is part of the Library. You may convey such object +code under terms of your choice, provided that, if the incorporated +material is not limited to numerical parameters, data structure +layouts and accessors, or small macros, inline functions and templates +(ten or fewer lines in length), you do both of the following: + + a) Give prominent notice with each copy of the object code that the + Library is used in it and that the Library and its use are + covered by this License. + + b) Accompany the object code with a copy of the GNU GPL and this license + document. + + 4. Combined Works. + + You may convey a Combined Work under terms of your choice that, +taken together, effectively do not restrict modification of the +portions of the Library contained in the Combined Work and reverse +engineering for debugging such modifications, if you also do each of +the following: + + a) Give prominent notice with each copy of the Combined Work that + the Library is used in it and that the Library and its use are + covered by this License. + + b) Accompany the Combined Work with a copy of the GNU GPL and this license + document. + + c) For a Combined Work that displays copyright notices during + execution, include the copyright notice for the Library among + these notices, as well as a reference directing the user to the + copies of the GNU GPL and this license document. + + d) Do one of the following: + + 0) Convey the Minimal Corresponding Source under the terms of this + License, and the Corresponding Application Code in a form + suitable for, and under terms that permit, the user to + recombine or relink the Application with a modified version of + the Linked Version to produce a modified Combined Work, in the + manner specified by section 6 of the GNU GPL for conveying + Corresponding Source. + + 1) Use a suitable shared library mechanism for linking with the + Library. A suitable mechanism is one that (a) uses at run time + a copy of the Library already present on the user's computer + system, and (b) will operate properly with a modified version + of the Library that is interface-compatible with the Linked + Version. + + e) Provide Installation Information, but only if you would otherwise + be required to provide such information under section 6 of the + GNU GPL, and only to the extent that such information is + necessary to install and execute a modified version of the + Combined Work produced by recombining or relinking the + Application with a modified version of the Linked Version. (If + you use option 4d0, the Installation Information must accompany + the Minimal Corresponding Source and Corresponding Application + Code. If you use option 4d1, you must provide the Installation + Information in the manner specified by section 6 of the GNU GPL + for conveying Corresponding Source.) + + 5. Combined Libraries. + + You may place library facilities that are a work based on the +Library side by side in a single library together with other library +facilities that are not Applications and are not covered by this +License, and convey such a combined library under terms of your +choice, if you do both of the following: + + a) Accompany the combined library with a copy of the same work based + on the Library, uncombined with any other library facilities, + conveyed under the terms of this License. + + b) Give prominent notice with the combined library that part of it + is a work based on the Library, and explaining where to find the + accompanying uncombined form of the same work. + + 6. Revised Versions of the GNU Lesser General Public License. + + The Free Software Foundation may publish revised and/or new versions +of the GNU Lesser General Public License from time to time. Such new +versions will be similar in spirit to the present version, but may +differ in detail to address new problems or concerns. + + Each version is given a distinguishing version number. If the +Library as you received it specifies that a certain numbered version +of the GNU Lesser General Public License "or any later version" +applies to it, you have the option of following the terms and +conditions either of that published version or of any later version +published by the Free Software Foundation. If the Library as you +received it does not specify a version number of the GNU Lesser +General Public License, you may choose any version of the GNU Lesser +General Public License ever published by the Free Software Foundation. + + If the Library as you received it specifies that a proxy can decide +whether future versions of the GNU Lesser General Public License shall +apply, that proxy's public statement of acceptance of any version is +permanent authorization for you to choose that version for the +Library. diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/lib/Control/Etage.hs b/lib/Control/Etage.hs new file mode 100644 --- /dev/null +++ b/lib/Control/Etage.hs @@ -0,0 +1,13 @@ +{-| +This data-flow framework consists mainly of 'Neuron's which are data processing units in data-flow network, receiving and sending +'Impulse's over bidirectional 'Nerve's 'attach'ed to each other. 'Neuron's and 'Nerve's are best 'grow'n in 'Incubation' monad, which takes care of +proper 'grow'ing and 'dissolve'-ing of 'Neuron's. It comes with some example 'Neuron's but you should probably define your own. +-} + +module Control.Etage ( + module Control.Etage.Incubator, + module Control.Etage.Externals +) where + +import Control.Etage.Externals +import Control.Etage.Incubator diff --git a/lib/Control/Etage/Chan.hs b/lib/Control/Etage/Chan.hs new file mode 100644 --- /dev/null +++ b/lib/Control/Etage/Chan.hs @@ -0,0 +1,180 @@ +{-# LANGUAGE DeriveDataTypeable #-} + +----------------------------------------------------------------------------- +-- | +-- Module : Control.Concurrent.Chan +-- Copyright : (c) The University of Glasgow 2001 +-- License : BSD-style (see the file libraries/base/LICENSE) +-- +-- Maintainer : libraries@haskell.org +-- Stability : experimental +-- Portability : non-portable (concurrency) +-- +-- Unbounded channels. +-- +----------------------------------------------------------------------------- + +-- Changes: Eq derived on Chan. + +module Control.Etage.Chan + ( + -- * The 'Chan' type + Chan, -- abstract + + -- * Operations + newChan, -- :: IO (Chan a) + writeChan, -- :: Chan a -> a -> IO () + readChan, -- :: Chan a -> IO a + tryReadChan, -- :: Chan a -> IO (Maybe a) + dupChan, -- :: Chan a -> IO (Chan a) + unGetChan, -- :: Chan a -> a -> IO () + isEmptyChan, -- :: Chan a -> IO Bool + + -- * Stream interface + getChanContents, -- :: Chan a -> IO [a] + writeList2Chan, -- :: Chan a -> [a] -> IO () + ) where + +import Prelude + +import System.IO.Unsafe ( unsafeInterleaveIO ) +import Control.Concurrent.MVar +import Data.Typeable + +import Control.Exception.Base + +-- A channel is represented by two @MVar@s keeping track of the two ends +-- of the channel contents,i.e., the read- and write ends. Empty @MVar@s +-- are used to handle consumers trying to read from an empty channel. + +-- |'Chan' is an abstract type representing an unbounded FIFO channel. +data Chan a + = Chan (MVar (Stream a)) + (MVar (Stream a)) + deriving (Eq, Typeable) + +type Stream a = MVar (ChItem a) + +data ChItem a = ChItem a (Stream a) + +-- See the Concurrent Haskell paper for a diagram explaining the +-- how the different channel operations proceed. + +-- @newChan@ sets up the read and write end of a channel by initialising +-- these two @MVar@s with an empty @MVar@. + +-- |Build and returns a new instance of 'Chan'. +newChan :: IO (Chan a) +newChan = do + hole <- newEmptyMVar + readVar <- newMVar hole + writeVar <- newMVar hole + return (Chan readVar writeVar) + +-- To put an element on a channel, a new hole at the write end is created. +-- What was previously the empty @MVar@ at the back of the channel is then +-- filled in with a new stream element holding the entered value and the +-- new hole. + +-- |Write a value to a 'Chan'. +writeChan :: Chan a -> a -> IO () +writeChan (Chan _ writeVar) val = do + new_hole <- newEmptyMVar + modifyMVar_ writeVar $ \old_hole -> do + putMVar old_hole (ChItem val new_hole) + return new_hole + +-- |Read the next value from the 'Chan'. +readChan :: Chan a -> IO a +readChan (Chan readVar _) = do + modifyMVar readVar $ \read_end -> do + (ChItem val new_read_end) <- readMVar read_end + -- Use readMVar here, not takeMVar, + -- else dupChan doesn't work + return (new_read_end, val) + +{-| + A semi-non-blocking version of 'readMVar'. The 'tryReadMVar' function returns immediately, with 'Nothing' if the 'MVar' was empty, or + @'Just' a@ if the 'MVar' was full with contents @a@, after it put the value back (it can block at this stage). +-} +tryReadMVar :: MVar a -> IO (Maybe a) +tryReadMVar m = + mask_ $ do + a <- tryTakeMVar m + case a of + Nothing -> return Nothing + Just a' -> do + putMVar m a' + return $ Just a' + +{-| + A semi-non-blocking version of 'modifyMVar'. The 'tryModifyMVar' function returns immediately, with 'Nothing' if the 'MVar' was empty, or + behave as 'modifyMVar' otherwise. This means that it can still block while putting the original (on exception) or new value (otherwise) back. +-} +{-# INLINE tryModifyMVar #-} +tryModifyMVar :: MVar a -> (a -> IO (a, Maybe b)) -> IO (Maybe b) +tryModifyMVar m io = + mask $ \restore -> do + a <- tryTakeMVar m + case a of + Nothing -> return Nothing + Just a' -> do + (a'', b) <- restore (io a') `onException` putMVar m a' + putMVar m a'' + return b + +-- |A non-blocking version of 'readChan'. The 'tryReadChan' function returns immediately, with 'Nothing' if the 'Chan' was empty or would +-- block, or @'Just' a@ with the next value from the 'Chan', otherwise. +tryReadChan :: Chan a -> IO (Maybe a) +tryReadChan (Chan readVar _) = do + tryModifyMVar readVar $ \read_end -> do + item <- tryReadMVar read_end + -- Use tryReadMVar here, not tryTakeMVar, + -- else dupChan doesn't work + case item of + Nothing -> return (read_end, Nothing) + Just (ChItem val new_read_end) -> return (new_read_end, Just val) + +-- |Duplicate a 'Chan': the duplicate channel begins empty, but data written to +-- either channel from then on will be available from both. Hence this creates +-- a kind of broadcast channel, where data written by anyone is seen by +-- everyone else. +dupChan :: Chan a -> IO (Chan a) +dupChan (Chan _ writeVar) = do + hole <- readMVar writeVar + newReadVar <- newMVar hole + return (Chan newReadVar writeVar) + +-- |Put a data item back onto a channel, where it will be the next item read. +unGetChan :: Chan a -> a -> IO () +unGetChan (Chan readVar _) val = do + new_read_end <- newEmptyMVar + modifyMVar_ readVar $ \read_end -> do + putMVar new_read_end (ChItem val read_end) + return new_read_end +{-# DEPRECATED unGetChan "if you need this operation, use Control.Concurrent.STM.TChan instead. See http://hackage.haskell.org/trac/ghc/ticket/4154 for details" #-} + +-- |Returns 'True' if the supplied 'Chan' is empty. +isEmptyChan :: Chan a -> IO Bool +isEmptyChan (Chan readVar writeVar) = do + withMVar readVar $ \r -> do + w <- readMVar writeVar + let eq = r == w + eq `seq` return eq +{-# DEPRECATED isEmptyChan "if you need this operation, use Control.Concurrent.STM.TChan instead. See http://hackage.haskell.org/trac/ghc/ticket/4154 for details" #-} + +-- Operators for interfacing with functional streams. + +-- |Return a lazy list representing the contents of the supplied +-- 'Chan', much like 'System.IO.hGetContents'. +getChanContents :: Chan a -> IO [a] +getChanContents ch + = unsafeInterleaveIO (do + x <- readChan ch + xs <- getChanContents ch + return (x:xs) + ) + +-- |Write an entire list of items to a 'Chan'. +writeList2Chan :: Chan a -> [a] -> IO () +writeList2Chan ch ls = sequence_ (map (writeChan ch) ls) diff --git a/lib/Control/Etage/Dump.hs b/lib/Control/Etage/Dump.hs new file mode 100644 --- /dev/null +++ b/lib/Control/Etage/Dump.hs @@ -0,0 +1,88 @@ +{-# LANGUAGE TypeFamilies, MultiParamTypeClasses, GADTs, FlexibleInstances, ScopedTypeVariables, TypeSynonymInstances, StandaloneDeriving, DeriveDataTypeable, NamedFieldPuns #-} + +{-| +This module defines a 'Neuron' which dumps all 'Impulse's it receives. You 'grow' it in 'Incubation' by using something like: + +> nerveDump <- growNeuron (\o -> o { showInsteadOfDump = True }) :: NerveOnlyFor DumpNeuron + +It is an example of a 'Neuron' which can recieve any 'Impulse' type. +-} + +module Control.Etage.Dump ( + DumpNeuron, + DumpFromImpulse, + DumpForImpulse, + DumpOptions, + NeuronFromImpulse, + NeuronForImpulse(..), + NeuronOptions(..) +) where + +import Control.Monad +import Data.Typeable +import System.IO + +import Control.Etage + +data DumpNeuron = DumpNeuron DumpOptions deriving (Typeable) + +-- | 'Impulse's from 'DumpNeuron'. This 'Neuron' does not define any 'Impulse's it would send. +type DumpFromImpulse = NeuronFromImpulse DumpNeuron +-- | 'Impulse's for 'DumpNeuron'. This 'Neuron' can recieve any 'Impulse' type. +type DumpForImpulse = NeuronForImpulse DumpNeuron +{-| +Options for 'DumpNeuron'. Those options are defined: + +[@handle :: 'Handle'@] 'Handle' to which it dumps. Default is 'stdout'. + +[@showInsteadOfDump :: 'Bool'@] Should it use 'show' when dumping 'Impulse's? By default it dumps 'impulseTime' and +'impulseValue' values. +-} +type DumpOptions = NeuronOptions DumpNeuron + +-- | Impulse instance for 'DumpNeuron'. +instance Impulse DumpFromImpulse where + impulseTime _ = undefined + impulseValue _ = undefined + +-- | Impulse instance for 'DumpNeuron'. +instance Impulse DumpForImpulse where + impulseTime (DumpForImpulse i) = impulseTime i + impulseValue (DumpForImpulse i) = impulseValue i + +deriving instance Show DumpFromImpulse + +instance Show DumpForImpulse where + show (DumpForImpulse i) = show i + +instance Eq DumpForImpulse where + (==) = impulseEq + +instance Ord DumpForImpulse where + compare = impulseCompare + +-- | A 'Neuron' which dumps all 'Impulse's it receives. +instance Neuron DumpNeuron where + data NeuronFromImpulse DumpNeuron + data NeuronForImpulse DumpNeuron where + DumpForImpulse :: Impulse i => i -> DumpForImpulse + data NeuronOptions DumpNeuron = DumpOptions { + handle :: Handle, + showInsteadOfDump :: Bool + } deriving (Eq, Show) + + mkDefaultOptions = return DumpOptions { + handle = stdout, + showInsteadOfDump = False + } + + grow options = return $ DumpNeuron options + + live nerve (DumpNeuron DumpOptions { handle, showInsteadOfDump }) = forever $ do + i <- getForNeuron nerve -- we want all not just newest + if showInsteadOfDump + then hPutStrLn handle $ show i + else hPutStrLn handle $ show (impulseTime i) ++ ": " ++ show (impulseValue i) + +instance Impulse i => ImpulseTranslator i DumpForImpulse where + translate i = [DumpForImpulse i] diff --git a/lib/Control/Etage/Externals.hs b/lib/Control/Etage/Externals.hs new file mode 100644 --- /dev/null +++ b/lib/Control/Etage/Externals.hs @@ -0,0 +1,430 @@ +{-# LANGUAGE TypeFamilies, MultiParamTypeClasses, GADTs, FlexibleInstances, FlexibleContexts, ScopedTypeVariables, TypeSynonymInstances, StandaloneDeriving, DeriveDataTypeable, NamedFieldPuns #-} + +module Control.Etage.Externals ( + -- * 'Neuron's and 'Impulse's + -- | Using only built-in 'Neuron's is not much fun. Main idea of this data-flow framework is to ease development of your own + -- 'Neuron's (data processing units). + Neuron(..), + attach', + detach, + detachAndWait, + detachMany, + detachManyAndWait, + LiveNeuron, + DissolveException, + dissolving, + DissolvingException, + + mkNeuronMapOnRandomCapability, + NeuronMapCapability(..), + + defaultOptions, + + Impulse(..), + ImpulseTime, + ImpulseValue, + ImpulseTranslator(..), + + translateAndSend, + + Nerve, + AxonConductive, + AxonNonConductive, + + -- * Sending and receiving outside the 'Neuron' + -- | Those functions are used outside the 'Neuron' when interacting with it. + sendForNeuron, + getFromNeuron, + maybeGetFromNeuron, + slurpFromNeuron, + waitAndSlurpFromNeuron, + getContentsFromNeuron, + sendListForNeuron, + + -- * Sending and receiving inside the 'Neuron' + -- | Those functions are used inside the 'Neuron' when implementing it. + sendFromNeuron, + getForNeuron, + maybeGetForNeuron, + slurpForNeuron, + waitAndSlurpForNeuron, + getNewestForNeuron, + getContentsForNeuron, + sendListFromNeuron, + + -- * Helper functions + prepareEnvironment, + getCurrentImpulseTime, + impulseEq, + impulseCompare +) where + +import Prelude hiding (catch) + +import Control.Concurrent hiding (Chan, writeChan, readChan, isEmptyChan, getChanContents) +import Data.Data +import Data.Function +import Data.List +import Control.Exception +import Data.Time.Clock.POSIX +import GHC.Conc (forkOnIO, numCapabilities) +import System.IO +import System.Posix.Signals +import System.Random + +import Control.Etage.Chan +import Control.Etage.Internals + +{-| +Sends an 'Impulse' from a 'Neuron'. 'Nerve' does not need to be conductive, 'Impulse' will be silently dropped in this case. +-} +sendFromNeuron :: Nerve from fromConductivity for forConductivity -> from -> IO () +sendFromNeuron (Nerve (Axon chan) _) i = writeChan chan i +sendFromNeuron (Nerve NoAxon _) _ = return () -- we allow sending but ignore so that same Neuron defintion can be used on all kinds of Nerves + +{-| +Gets an 'Impulse' from a 'Neuron'. It blocks until an 'Impulse' is available. 'Nerve' has to be conductive. +-} +getFromNeuron :: Nerve from AxonConductive for forConductivity -> IO from +getFromNeuron (Nerve (Axon chan) _) = readChan chan + +{-| +Similar to 'getFromNeuron' just that it does not block if 'Impulse' is not available. +-} +maybeGetFromNeuron :: Nerve from AxonConductive for forConductivity -> IO (Maybe from) +maybeGetFromNeuron (Nerve (Axon chan) _) = tryReadChan chan + +{-| +Gets all immediately available 'Impulse's from a 'Neuron'. There could be no 'Impulse's available and thus the result is an empty +list. Oldest 'Impulse' is the last in the list. 'Nerve' has to be conductive. +-} +slurpFromNeuron :: Nerve from AxonConductive for forConductivity -> IO [from] +slurpFromNeuron (Nerve (Axon chan) _) = slurpChan chan + +{-| +Similar to 'slurpFromNeuron' but it waits for at least one 'Impulse'. +-} +waitAndSlurpFromNeuron :: Nerve from AxonConductive for forConductivity -> IO [from] +waitAndSlurpFromNeuron nerve = do + oldest <- getFromNeuron nerve + others <- slurpFromNeuron nerve + return $ others ++ [oldest] + +{-| +Returns a lazy list of 'Impulse's from a 'Neuron'. 'Nerve' has to be conductive. +-} +getContentsFromNeuron :: Nerve from AxonConductive for forConductivity -> IO [from] +getContentsFromNeuron (Nerve (Axon chan) _) = getChanContents chan + +{-| +Sends all 'Impulses' from a given list to a 'Neuron'. 'Nerve' does not need to be conductive, 'Impulse's will be silently +dropped in this case. +-} +sendListFromNeuron :: Nerve from fromConductivity for forConductivity -> [from] -> IO () +sendListFromNeuron nerve = mapM_ (sendFromNeuron nerve) + +{-| +Sends an 'Impulse' to a 'Neuron'. 'Nerve' has to be conductive. +-} +sendForNeuron :: Nerve from fromConductivity for AxonConductive -> for -> IO () +sendForNeuron (Nerve _ (Axon chan)) = writeChan chan + +{-| +Gets an 'Impulse' for a 'Neuron'. It blocks until an 'Impulse' is available. 'Nerve' does not need to be conductive, +it will block indefinitely (until an exception) in this case. +-} +getForNeuron :: Nerve from fromConductivity for forConductivity -> IO for +getForNeuron (Nerve _ (Axon chan)) = readChan chan +getForNeuron (Nerve _ NoAxon) = waitForException + +{-| +Similar to 'getForNeuron' just that it does not block if 'Impulse' is not available. 'Nerve' does not need to be conductive, +it will always return 'Nothing' in this case. +-} +maybeGetForNeuron :: Nerve from fromConductivity for forConductivity -> IO (Maybe for) +maybeGetForNeuron (Nerve _ (Axon chan)) = tryReadChan chan +maybeGetForNeuron (Nerve _ NoAxon) = return Nothing -- we allow getting but return Nothing so that same Neuron defintion can be used on all kinds of Nerves + +{-| +Gets all immediately available 'Impulse's for a 'Neuron'. There could be no 'Impulse's available and thus the result is an empty +list. Oldest 'Impulse' is the last in the list. 'Nerve' does not need to be conductive, it will always return an empty list +in this case. +-} +slurpForNeuron :: Nerve from fromConductivity for forConductivity -> IO [for] +slurpForNeuron (Nerve _ (Axon chan)) = slurpChan chan +slurpForNeuron (Nerve _ NoAxon) = return [] -- we allow getting but return [] so that same Neuron defintion can be used on all kinds of Nerves + +{-| +Similar to 'slurpForNeuron' but it waits for at least one 'Impulse'. 'Nerve' does not need to be conductive, +it will block indefinitely (until an exception) in this case. +-} +waitAndSlurpForNeuron :: Nerve from fromConductivity for forConductivity -> IO [for] +waitAndSlurpForNeuron nerve = do + oldest <- getForNeuron nerve + others <- slurpForNeuron nerve + return $ others ++ [oldest] + +{-| +Similar to 'waitAndSlurpForNeuron' but it will return only the newest 'Impulse' for every 'NeuronForImpulse' data type constructor. +This is the same as @head \<$\> waitAndSlurpForNeuron@ iff 'NeuronForImpulse' has only one constructor defined. Otherwise it can +return multiple 'Impulse's, for each constructor one. +-} +getNewestForNeuron :: Data for => Nerve from fromConductivity for forConductivity -> IO [for] +getNewestForNeuron nerve = do + impulses <- waitAndSlurpForNeuron nerve + return $ nubBy ((==) `on` toConstr) impulses + +{-| +Returns a lazy list of 'Impulse's for a 'Neuron'. 'Nerve' does not need to be conductive, +it will block indefinitely (until an exception) in this case. +-} +getContentsForNeuron :: Nerve from fromConductivity for forConductivity -> IO [for] +getContentsForNeuron (Nerve _ (Axon chan)) = getChanContents chan +getContentsForNeuron (Nerve _ NoAxon) = waitForException + +{-| +Sends all 'Impulses' from a given list to a 'Neuron'. 'Nerve' has to be conductive. +-} +sendListForNeuron :: Nerve from fromConductivity for AxonConductive -> [for] -> IO () +sendListForNeuron nerve = mapM_ (sendForNeuron nerve) + +-- First-in (oldest) element in the channel is last in the list +slurpChan :: Chan a -> IO [a] +slurpChan chan = slurpChan' [] + where slurpChan' cs = do + mc <- tryReadChan chan + case mc of + Nothing -> return cs + Just c -> slurpChan' (c:cs) + +{-| +'Neuron's can be mapped to capabilities (OS threads) in different ways. The best is to let Haskell decide the best capability +(and also move 'Neuron's among them as necessary) by using 'NeuronFreelyMapOnCapability' value, but sometimes because of an external +(FFI) library limitations you have to map 'Neuron' to a fixed capability, you can use 'NeuronMapOnCapability' for that. + +Sometimes it is not important to which capability you map a 'Neuron', just that few 'Neuron's are mapped to the same. You can +use 'mkNeuronMapOnRandomCapability' to create such 'NeuronMapCapability' value. +-} +data NeuronMapCapability = + NeuronMapOnCapability Int -- ^ Map a 'Neuron' to fixed capability. + | NeuronFreelyMapOnCapability -- ^ Let Haskell decide on which capability is best to map a 'Neuron' at a given time. + deriving (Eq, Ord, Read, Show) + +{-| +Creates a 'NeuronMapOnCapability' value with a chosen capability picked by random. Useful when you have to map few 'Neuron's to the +same capability (because of an eternal (FFI) library limitations) but it does not matter to which one. So you create this value +and pass it as an option to all those 'Neuron's, making sure that they will return it with their 'getNeuronMapCapability' method. +For example, sometimes you have to assure that both your 'Neuron' and "Control.Etage.Worker" 'Neuron' are running on the same +capability so that you can correctly offload lengthly IO actions to it. This makes both 'Neuron's in fact still running in one +thread (which is often a limitation of external libraries), Haskell taking care of interleaving 'Neuron's IO actions. +-} +mkNeuronMapOnRandomCapability :: IO NeuronMapCapability +mkNeuronMapOnRandomCapability = do + c <- randomRIO (1, numCapabilities) + return $ NeuronMapOnCapability c + +divideNeuron :: Neuron n => NeuronOptions n -> IO () -> IO NeuronId +divideNeuron options a = fork a + where fork = case getNeuronMapCapability options of + NeuronFreelyMapOnCapability -> forkIO + NeuronMapOnCapability c -> forkOnIO c + +deriving instance Typeable1 NeuronFromImpulse +deriving instance Typeable1 NeuronForImpulse +deriving instance Typeable1 NeuronOptions + +-- | A type class which defines common methods and data types of 'Neuron's. +class (Typeable n, Impulse (NeuronFromImpulse n), Impulse (NeuronForImpulse n)) => Neuron n where + -- | A data type for 'Impulses' send from a 'Neuron'. 'Neuron' does not really need to use them. + data NeuronFromImpulse n + -- | A data type for 'Impulses' send for a 'Neuron'. 'Neuron' does not really need to use them. + data NeuronForImpulse n + -- | A data type for options. 'Neuron' does not really need to use them. + data NeuronOptions n + + -- | Method which returns default values for options. By default returns 'undefined'. + mkDefaultOptions :: IO (NeuronOptions n) + + -- | Method which returns how should 'Neuron' be mapped on capabilities (OS threads). By default returns + -- 'NeuronFreelyMapOnCapability'. + getNeuronMapCapability :: NeuronOptions n -> NeuronMapCapability + + -- | The first phase in a life-cycle of a 'Neuron' is to 'grow'. In this phase everything should be prepared and initialized. + -- It returns a 'Neuron' value which is then passed to next phases. If you want to use 'NeuronOptions' also in those phases + -- you should store them in the 'Neuron' value. By default returns 'undefined'. + grow :: NeuronOptions n -> IO n + -- | After 'grow'ing 'Neuron' 'live's. This is a phase in which it should read 'Impulse's from its 'Nerve' and send them back, + -- as defined by its logic/purpose. Some 'Neuron's only read, some only send, some do both or none. + -- + -- Most 'Neuron's do never finish this phase on its own (only by exception), but if your 'Neuron' does, consider using 'dissolving' + -- at the end which initiates dissolving also elsewhere in the network (or in the parent 'Neuron', if it has one). Examples + -- of such 'Neuron's are "Control.Etage.Timeout" and "Control.Etage.Sequence" (once a given sequence ends). + -- + -- By default it blocks indefinitely (until an exception). + live :: Nerve (NeuronFromImpulse n) fromConductivity (NeuronForImpulse n) forConductivity -> n -> IO () + -- | In this phase everything should be cleaned up and deinitialized. If you have 'grow'n child 'Neuron's you should take care + -- here to 'dissolve' them too. You can use 'detachAndWait' for that (or 'detachManyAndWait' if you have more of them). + -- By default it does nothing. + dissolve :: n -> IO () + + -- | This method should take care of 'grow'ing a 'Neuron' with a given 'Nerve' 'attach'ed to it. It takes a function which + -- changes default options and returns a 'LiveNeuron' value which can be used for 'detach'ing (and thus 'dissolve'-ing) the 'Neuron'. + -- It should create a thread for a 'Neuron' to 'live' in and it should assure proper cleanup and 'dissolve'-ing. + -- + -- By default it calls 'attach'' to do all that. + attach :: (NeuronOptions n -> NeuronOptions n) -> Nerve (NeuronFromImpulse n) fromConductivity (NeuronForImpulse n) forConductivity -> IO LiveNeuron + + mkDefaultOptions = return undefined + + getNeuronMapCapability _ = NeuronFreelyMapOnCapability + + grow _ = return undefined + dissolve _ = return () + live _ _ = waitForException + + attach = attach' + +{-| +Default implementation for 'attach' method. It takes a function which changes default options and returns a 'LiveNeuron' value +which can be used for 'detach'ing (and thus 'dissolve'-ing) the 'Neuron'. + +It changes default options according to a given function, creates thread for a 'Neuron' to live in based on 'getNeuronMapCapability', +'grow's a 'Neuron', runs 'live' and prepares everything for cleanup with 'dissolve', whether because 'live' finished or because of an +exception. In the later case it rethrows an exception in the parent 'Neuron' (or in 'Incubation'). It also signals the 'Neuron' +has 'dissolve'd for 'detachAndWait' and 'detachManyAndWait'. +-} +attach' :: Neuron n => (NeuronOptions n -> NeuronOptions n) -> Nerve (NeuronFromImpulse n) fromConductivity (NeuronForImpulse n) forConductivity -> IO LiveNeuron +attach' optionsSetter nerve = mask $ \restore -> do + currentThread <- myThreadId + dissolved <- newEmptySampleVar + defOptions <- mkDefaultOptions + let options = optionsSetter defOptions + nid <- divideNeuron options $ + bracket (grow options) dissolve (restore . live nerve) `catches` [ + Handler (\(_ :: DissolveException) -> return ()), -- we ignore DissolveException + Handler (\(e :: SomeException) -> uninterruptible $ throwTo currentThread e) + ] `finally` uninterruptible (writeSampleVar dissolved ()) + return $ LiveNeuron dissolved nid + +{-| +An exception which initiates 'dissolve'-ing of a 'Neuron'. Should be thrown inside the 'Neuron' with passing its 'Neuron' value as +argument (as passed to 'live' method). For throwing outside the 'Neuron' use 'DissolveException' (or simply 'detach' and others). +-} +data DissolvingException = DissolvingException String deriving (Show, Typeable) + +instance Exception DissolvingException + +{-| +Initiates 'dissolve'-ing of a 'Neuron' by throwing a 'DissolvingException'. To be used inside a 'Neuron' to maybe prematurely +finish its life but more importantly to initiate 'dissolve'-ing in the parent 'Neuron' (or in 'Incubation'). As an argument +it is accustomed to pass a 'Neuron' value as passed to 'live' method. +-} +dissolving :: Show n => n -> IO a +dissolving n = throwIO $ DissolvingException (show n) + +{-| +An exception which initiates 'dissolve'-ing of a 'Neuron'. Should be thrown outside the 'Neuron' to the 'Neuron'. For +throwing inside the 'Neuron' use 'DissolvingException' (or simply 'dissolving'). +-} +data DissolveException = DissolveException deriving (Show, Typeable) + +instance Exception DissolveException + +{-| +Initiates 'dissolve'-ing of a 'Neuron' by throwing a 'DissolveException'. To be used outside of a 'Neuron'. +-} +detach :: LiveNeuron -> IO () +detach (LiveNeuron _ neuronId) = mask_ . uninterruptible $ throwTo neuronId DissolveException + +{-| +Similar to 'detachAndWait' but it also waits 'Neuron' to finish 'dissolve'-ing. +-} +detachAndWait :: LiveNeuron -> IO () +detachAndWait n = detachManyAndWait [n] + +{-| +Similar to 'detach' but for many 'Neuron's at the same time. It initiates 'dissolve'-ing in the list order. +-} +detachMany :: [LiveNeuron] -> IO () +detachMany = mask_ . mapM_ detach + +{-| +Similar to 'detachAndWait' but for many 'Neuron's at the same time. It first initiates 'dissolve'-ing in the list order and then +wait for all 'Neuron's to finish 'dissolve'-ing. +-} +detachManyAndWait :: [LiveNeuron] -> IO () +detachManyAndWait neurons = mask_ $ do + detachMany neurons + mapM_ (\(LiveNeuron d _) -> uninterruptible $ readSampleVar d) neurons + +-- Some IO operations are interruptible, better than to make them uninterruptible (which can cause deadlocks) we simply retry interrupted operation +-- For this to really work all interruptible operations should be wrapped like this (so it is not good idea to use IO operations in such code sections) +uninterruptible :: IO a -> IO a +uninterruptible a = mask_ $ a `catch` (\(_ :: SomeException) -> uninterruptible a) + +-- | This type class defines a method for translating between 'Impulse' types. +class (Impulse i, Impulse j) => ImpulseTranslator i j where + -- | 'translate' gets an 'Impulse' of one type and returns a list of 'Impulses' of another type. + -- + -- 'Impulse's should be translated meaningfully, translating values as possible and filling others with reasonable defaults. + -- Timestamp should be just copied (translation should be seen as an instantaneous operation as it is a byproduct of type + -- constraints and chosen description format of 'Impulse's and not something found otherwise in a network. + -- Time spend in translation should be seen as a part of time spend in sending of an 'Impulse' along a 'Nerve'. + -- + -- One 'Impulse' can be translated into multiple other 'Impulse's as sometimes some 'Impulse's are higher level than other. + -- (Translating multiple 'Impulse's into one 'Impulse' requires keeping a state and should be done in a 'Neuron'.) The order is + -- important as first 'Impulse's in the list are send first along a 'Nerve'. + translate :: i -> [j] + +{-| +Function which can be used as an argument to 'growNeuron' or 'attach' which leaves default options as they are. + +In fact it is just an 'id'entity function. +-} +defaultOptions :: Neuron n => NeuronOptions n -> NeuronOptions n +defaultOptions = id + +{-| +Helper function which does some common initialization. Currently it sets 'stderr' buffering to 'LineBuffering' so that when +multiple 'Neuron's print to 'stderr' output is not mixed. It also installs handlers for 'keyboardSignal' and 'softwareTermination' +signals so that cleanup in 'Incubation' works as expected. +-} +prepareEnvironment :: IO () +prepareEnvironment = do + hSetBuffering stderr LineBuffering + + mainThreadId <- myThreadId + + -- TODO: User interrupt sometimes hangs dissolving (does it still in GHC 7.0?) + _ <- installHandler keyboardSignal (Catch (throwTo mainThreadId UserInterrupt)) Nothing -- sigINT + _ <- installHandler softwareTermination (Catch (throwTo mainThreadId UserInterrupt)) Nothing -- sigTERM + + return () + +{-| +Translates (if necessary 'ImpulseTranslator' exists) an 'Impulse' and sends translation to 'Neuron'. +-} +translateAndSend :: ImpulseTranslator i for => Nerve from fromConductivity for AxonConductive -> i -> IO () +translateAndSend nerve i = mapM_ (sendForNeuron nerve) $ translate i + +{-| +Returns current time. Useful when creating new 'Impulse's. +-} +getCurrentImpulseTime :: IO ImpulseTime +getCurrentImpulseTime = getPOSIXTime + +{-| +This function defines equality between 'Impulse's as equality of 'impulseTime' and 'impulseValue' values. Useful for 'Neuron's which +operate on all types of 'Impulse's and want 'Eq' defined on their 'Impulse's. Examples of such 'Neuron's are "Control.Etage.Dump" +and "Control.Etage.Function". +-} +impulseEq :: (Impulse i, Impulse j) => i -> j -> Bool +impulseEq a b = impulseTime a == impulseTime b && impulseValue a == impulseValue b + +{-| +This function defines ordering between 'Impulse's as ordering first by 'impulseTime' values and then by 'impulseValue' values. +Useful for 'Neuron's which operate on all types of 'Impulse's and want 'Ord' defined on their 'Impulse's. Examples of such +'Neuron's are "Control.Etage.Dump" and "Control.Etage.Function". +-} +impulseCompare :: (Impulse i, Impulse j) => i -> j -> Ordering +impulseCompare a b = (impulseTime a, impulseValue a) `compare` (impulseTime b, impulseValue b) diff --git a/lib/Control/Etage/Fail.hs b/lib/Control/Etage/Fail.hs new file mode 100644 --- /dev/null +++ b/lib/Control/Etage/Fail.hs @@ -0,0 +1,57 @@ +{-# LANGUAGE TypeFamilies, MultiParamTypeClasses, GADTs, FlexibleInstances, ScopedTypeVariables, DeriveDataTypeable, TypeSynonymInstances, StandaloneDeriving, NamedFieldPuns #-} + +{-| +This module defines a simple 'Neuron' which just fails (throws a 'DissolvingException') in 'grow'ing phase. It can be used to test +error recovery and cleanup in 'grow'ing phase or early stages of 'live'ing phase in other 'Neuron's by using something like: + +> _ <- growNeuron defaultOptions :: NerveNone FailNeuron + +somewhere among (or after) 'growNeuron' calls for other 'Neuron's in 'Incubation'. +-} + +module Control.Etage.Fail ( + FailNeuron, + FailFromImpulse, + FailForImpulse, + FailOptions, + NeuronFromImpulse, + NeuronForImpulse, + NeuronOptions +) where + +import Data.Typeable + +import Control.Etage + +data FailNeuron deriving (Typeable) + +instance Show FailNeuron where + show = show . typeOf + +-- | 'Impulse's from 'FailNeuron'. This 'Neuron' does not define any 'Impulse's it would send. +type FailFromImpulse = NeuronFromImpulse FailNeuron +-- | 'Impulse's for 'FailNeuron'. This 'Neuron' does not define any 'Impulse's it would receive. +type FailForImpulse = NeuronForImpulse FailNeuron +-- | Options for 'FailNeuron'. This 'Neuron' does not define any options. +type FailOptions = NeuronOptions FailNeuron + +-- | Impulse instance for 'FailNeuron'. +instance Impulse FailFromImpulse where + impulseTime _ = undefined + impulseValue _ = undefined + +-- | Impulse instance for 'FailNeuron'. +instance Impulse FailForImpulse where + impulseTime _ = undefined + impulseValue _ = undefined + +deriving instance Show FailFromImpulse +deriving instance Show FailForImpulse + +-- | A simple 'Neuron' which just fails in 'grow'ing phase. +instance Neuron FailNeuron where + data NeuronFromImpulse FailNeuron + data NeuronForImpulse FailNeuron + data NeuronOptions FailNeuron + + grow _ = dissolving (undefined :: FailNeuron) diff --git a/lib/Control/Etage/Function.hs b/lib/Control/Etage/Function.hs new file mode 100644 --- /dev/null +++ b/lib/Control/Etage/Function.hs @@ -0,0 +1,100 @@ +{-# LANGUAGE TypeFamilies, MultiParamTypeClasses, GADTs, FlexibleInstances, ScopedTypeVariables, DeriveDataTypeable, TypeSynonymInstances, NamedFieldPuns, BangPatterns #-} + +{-| +This module defines a 'Neuron' which sends results of applying a given function to recieved 'Impulse's. You 'grow' it in +'Incubation' by using something like: + +> nerveFunction <- growNeuron (\o -> o { function = negate . sum }) :: NerveBoth FunctionNeuron + +It is an example of a 'Neuron' which can operate on any 'Impulse' type by using 'impulseValue' type class method. +-} + +module Control.Etage.Function ( + FunctionNeuron, + FunctionFromImpulse, + FunctionForImpulse, + FunctionOptions, + NeuronFromImpulse, + NeuronForImpulse, + NeuronOptions(..) +) where + +import Control.Applicative +import Control.Monad +import Data.Time.Clock +import Data.Typeable + +import Control.Etage + +defaultFunction :: [Rational] -> Rational +defaultFunction = sum + +data FunctionNeuron = FunctionNeuron FunctionOptions deriving (Typeable) + +instance Show FunctionNeuron where + show = show . typeOf + +{-| +'Impulse's from 'FunctionNeuron'. This 'Impulse' constructor is defined: + +[@Value { impulseTimestamp :: 'ImpulseTime', value :: 'Rational', evaluationTime :: 'NominalDiffTime' }@] +@impulseTimestamp@ is time when the result was evaluated, @value@ contains the evaluated result, @evaluationTime@ is how long the +evaluation took. +-} +type FunctionFromImpulse = NeuronFromImpulse FunctionNeuron +-- | 'Impulse's for 'FunctionNeuron'. This 'Neuron' can recieve any 'Impulse' type. +type FunctionForImpulse = NeuronForImpulse FunctionNeuron +{-| +Options for 'FunctionNeuron'. This option is defined: + +[@function :: \['Rational'\] -> 'Rational'@] The function to apply to recieved 'Impulse's. Default is 'sum'. +-} +type FunctionOptions = NeuronOptions FunctionNeuron + +-- | Impulse instance for 'FunctionNeuron'. +instance Impulse FunctionFromImpulse where + impulseTime Value { impulseTimestamp } = impulseTimestamp + impulseValue Value { value } = [value] + +-- | Impulse instance for 'FunctionNeuron'. +instance Impulse FunctionForImpulse where + impulseTime (FunctionForImpulse i) = impulseTime i + impulseValue (FunctionForImpulse i) = impulseValue i + +instance Show FunctionForImpulse where + show (FunctionForImpulse i) = show i + +instance Eq FunctionForImpulse where + (==) = impulseEq + +instance Ord FunctionForImpulse where + compare = impulseCompare + +-- | A 'Neuron' which sends results of a given function for recieved 'Impulse's. +instance Neuron FunctionNeuron where + data NeuronFromImpulse FunctionNeuron = Value { + impulseTimestamp :: ImpulseTime, -- time is first so that ordering is first by time + value :: Rational, + evaluationTime :: NominalDiffTime + } deriving (Eq, Ord, Read, Show) + data NeuronForImpulse FunctionNeuron where + FunctionForImpulse :: Impulse i => i -> FunctionForImpulse + data NeuronOptions FunctionNeuron = FunctionOptions { + function :: [Rational] -> Rational + } + + mkDefaultOptions = return FunctionOptions { + function = defaultFunction + } + + grow options = return $ FunctionNeuron options + + live nerve (FunctionNeuron FunctionOptions { function }) = forever $ do + i <- head <$> waitAndSlurpForNeuron nerve -- just newest + time1 <- getCurrentImpulseTime + let !r = function . impulseValue $ i + time2 <- getCurrentImpulseTime + sendFromNeuron nerve Value { impulseTimestamp = time2, value = r, evaluationTime = time2 - time1 } + +instance Impulse i => ImpulseTranslator i FunctionForImpulse where + translate i = [FunctionForImpulse i] diff --git a/lib/Control/Etage/Incubator.hs b/lib/Control/Etage/Incubator.hs new file mode 100644 --- /dev/null +++ b/lib/Control/Etage/Incubator.hs @@ -0,0 +1,246 @@ +{-# LANGUAGE GADTs, FlexibleInstances, FlexibleContexts, GeneralizedNewtypeDeriving, ScopedTypeVariables #-} + +module Control.Etage.Incubator ( + -- * Incubation + -- | 'Incubation' is a 'Monad' helping 'grow'ing a network of 'Neuron's and 'Nerve's while taking care of all the details and + -- cleanup. It is the recommended and preferred way for 'grow'ing your networks. + -- + -- A basic example of using 'Incubation' and of this data-flow framework would be a program where one 'Neuron' is generating + -- 'Impulse's with random values ("Control.Etage.Sequence") and another 'Neuron' printing them out ("Control.Etage.Dump"): + -- + -- > main = do + -- > prepareEnvironment + -- > + -- > incubate $ do + -- > nerveRandom <- growNeuron defaultOptions :: NerveOnlyFrom (SequenceNeuron Int) + -- > nerveDump <- growNeuron defaultOptions :: NerveOnlyFor DumpNeuron + -- > + -- > nerveRandom `attachTo` [Translatable nerveDump] + incubate, + growNeuron, + attachTo, + NerveBoth, + NerveNone, + NerveOnlyFrom, + NerveOnlyFor, + Incubation, + Translatable(..), + -- * Internals + -- | Be careful when using those functions as you have to assure your network is well-behaved: + -- + -- * You should assure that for all 'Nerve's you defined as conductive from 'Neuron's and 'attach'ed them to 'Neuron's you + -- really receive sent impulses, otherwise there will be a memory leak. You should probably just define those nerves + -- as 'NerveOnlyFor' or 'NerveNone'. + -- + -- * If you 'attach' multiple 'Neuron's to the same 'Nerve' you should probably take care of branching 'Nerve's correctly. For + -- example, if multiple 'Neuron's are receiving from the same 'Nerve' you should first branch 'Nerve' with 'branchNerveFor', + -- otherwise 'Neuron's will not receive all 'Impulse's as some other 'Neuron' will receive it first (but this can be also + -- intentional). + -- On the other hand, if you are receiving from the same 'Neuron' at multiple parts of the network you should branch + -- 'Nerve' with 'branchNerveFrom' for each such part (or not, if intentional). This also holds for 'propagate': if you are using + -- it multiple times with the same 'Nerve' as @from@ argument you should first branch it with 'branchNerveFrom'. (But it is + -- probably easier to just use it once and list all @for@ 'Nerve's together.) + -- + -- * And of course in a case of an exception or in general when your are doing cleanup you should assure that 'detach' + -- (or 'detachAndWait') is called for each 'LiveNeuron' (or 'detachMany' or 'detachManyAndWait'). + -- + -- They are exposed so that you can decouple 'grow'ing and 'dissolve'-ing handling and that you can attach 'Nerve's + -- in some special ways. If you do not need that use 'Incubation'. + -- + -- For example, your 'Neuron' can 'grow' and use another 'Neuron' (in this example "Control.Etage.Worker") like this: + -- + -- > data YourNeuron = YourNeuron ... LiveNeuron (Nerve WorkerFromImpulse AxonNonConductive WorkerForImpulse AxonConductive) deriving (Typeable) + -- > + -- > grow options = do + -- > ... + -- > nerve <- growNerve + -- > bracketOnError (attach defaultOptions nerve) detachAndWait $ \neuron -> do + -- > ... + -- > return $ YourNeuron ... neuron nerve + -- > + -- > dissolve (YourNeuron ... neuron _) = do + -- > detachAndWait neuron + -- > ... + -- + -- We use 'bracketOnError' there to be sure that 'Neuron' is properly 'dissolve'd even if there is an exception later on in + -- 'grow'ing the parent 'Neuron'. And we use 'detachAndWait' so that we give time for child 'Neuron' to 'dissolve' properly. + -- Which 'Neuron' you want is in this case inferred from the type of the 'Nerve' you defined. + growNerve, + propagate, + branchNerveFor, + branchNerveFrom, + branchNerveBoth +) where + +import Control.Applicative +import Control.Exception +import Control.Monad +import Control.Monad.Operational +import Control.Monad.Trans +import Data.List +import Data.Typeable +import System.IO + +import Control.Etage.Chan +import Control.Etage.Propagate +import Control.Etage.Internals +import Control.Etage.Externals + +data IncubationOperation a where + NeuronOperation :: (Neuron n, GrowAxon (Axon (NeuronFromImpulse n) fromConductivity), GrowAxon (Axon (NeuronForImpulse n) forConductivity)) => (NeuronOptions n -> NeuronOptions n) -> IncubationOperation (Nerve (NeuronFromImpulse n) fromConductivity (NeuronForImpulse n) forConductivity) + AttachOperation :: forall from for forConductivity. (Typeable from, Typeable for, Typeable forConductivity) => Nerve from AxonConductive for forConductivity -> [Translatable from] -> IncubationOperation () + +type Incubation' a = ProgramT IncubationOperation IO a +{-| +An 'Incubation' monad type. It makes sure network is 'grow'n properly and that everything is cleaned up as necessary. +-} +newtype Incubation a = Incubation (Incubation' a) deriving (Monad, MonadIO, Applicative, Functor) + +-- TODO: Check if all chans have been attached with type checking (type nats)? (If this checking even shows as useful. And correct.) +{-| +Runs an 'Incubation', 'grow'ing 'Neuron's and 'attach'ing 'Nerve's and after that waiting for them to finish and cleanup. +It rethrows any exception which might have been thrown. +-} +incubate :: Incubation () -> IO () +incubate (Incubation program) = mask $ \restore -> do + (neurons, chans, attached) <- restore $ interpret [] [] [] program + flip finally (detachManyAndWait neurons) $ do + let na = nub chans \\ nub attached + typ = unlines . map (\(ChanBox c) -> ' ' : show (neuronTypeOf c)) $ na + unless (null na) $ hPutStrLn stderr $ "Warning: It seems not all created nerves were attached. This causes a memory leak as send impulses are not received. You should probably just define those nerves as NerveOnlyFor or NerveNone. Dangling nerves for neurons:\n" ++ typ + restore waitForException + +interpret :: [LiveNeuron] -> [ChanBox] -> [ChanBox] -> Incubation' () -> IO ([LiveNeuron], [ChanBox], [ChanBox]) +interpret neurons chans attached = viewT >=> eval neurons chans attached + where eval :: [LiveNeuron] -> [ChanBox] -> [ChanBox] -> ProgramViewT IncubationOperation IO () -> IO ([LiveNeuron], [ChanBox], [ChanBox]) + eval ns cs ats (Return _) = return (ns, cs, ats) + eval ns cs ats (NeuronOperation optionsSetter :>>= is) = do + nerve <- liftIO growNerve + let c = getFromChan nerve + bracketOnError (attach optionsSetter nerve) detachAndWait $ \n -> interpret (n:ns) (c ++ cs) ats . is $ nerve + eval ns cs ats (AttachOperation from for :>>= is) = do + let c = head . getFromChan $ from -- we know there exists from chan as type checking assures that (from is conductive) + (from', ats') <- if c `notElem` ats + then return (from, c:ats) + else do + branchFrom <- branchNerveFrom from -- we have to branch from chan as it is attached multiple times + return (branchFrom, ats) -- we store only original nerves in attached list + propagate from' for + interpret ns cs ats' . is $ () + +{-| +Grows a 'Neuron', taking a function which changes default options and returning a 'Nerve' 'attach'ed to the 'Neuron'. + +Internally it combines 'growNerve' and 'attach'. +-} +growNeuron :: (Neuron n, GrowAxon (Axon (NeuronFromImpulse n) fromConductivity), GrowAxon (Axon (NeuronForImpulse n) forConductivity)) => (NeuronOptions n -> NeuronOptions n) -> Incubation (Nerve (NeuronFromImpulse n) fromConductivity (NeuronForImpulse n) forConductivity) +growNeuron os = Incubation $ singleton (NeuronOperation os) + +{-| +Attaches a 'Nerve' to other 'Nerve's so that 'Impulse's send from the 'Neuron' over the first 'Nerve' are received by 'Neuron's +of other 'Nerve's. 'Impulse's are 'propagate'd only in this direction, not in the other. If you want also the other direction use +'attachTo' again for that direction. + +Internally it uses 'propagate'. +-} +attachTo :: forall from for forConductivity. (Typeable from, Typeable for, Typeable forConductivity) => Nerve from AxonConductive for forConductivity -> [Translatable from] -> Incubation () +attachTo n ts = Incubation $ singleton (AttachOperation n ts) + +class GrowAxon a where + growAxon :: IO a + +instance Impulse i => GrowAxon (Axon i AxonConductive) where + growAxon = Axon <$> newChan + +instance GrowAxon (Axon i AxonNonConductive) where + growAxon = return NoAxon + +{-| +Grows an unattached 'Nerve'. By specifying type of the 'Nerve' you can specify conductivity of both directions (which is then +type checked for consistency around the program) and thus specify which 'Impulse's you are interested in (and thus limit possible +memory leak). With type of 'Impulse's this 'Nerve' is capable of conducting you can also specify which 'Neuron' you are interested +in 'grow'ing on the one end of the 'Nerve'. + +For example, you could grow a 'Nerve' for "Control.Etage.Sequence" 'Neuron' and 'Neuron' itself like this: + +> nerve <- growNerve :: IO (Nerve (SequenceFromImpulse Int) AxonConductive (SequenceForImpulse Int) AxonNonConductive) +> neuron <- attach defaultOptions nerve + +and for example print all 'Impulse's as they are coming in: + +> print =<< getContentsFromNeuron nerve + +Check 'growNeuron' for a more high-level function (of 'Incubation') which both 'grow's a 'Neuron' and corresponding 'Nerve' taking +care of all the details. Use this function only if you need decoupled 'grow'ing. +-} +growNerve :: (Impulse from, Impulse for, GrowAxon (Axon from fromConductivity), GrowAxon (Axon for forConductivity)) => IO (Nerve from fromConductivity for forConductivity) +growNerve = do + from <- growAxon + for <- growAxon + return $ Nerve from for + +{-| +Type which helps you define a type of the result of 'growNeuron'. It takes type of the 'Neuron' you want to 'grow' as an argument +and specifies a 'Nerve' which is conductive in both directions. +-} +type NerveBoth n = Incubation (Nerve (NeuronFromImpulse n) AxonConductive (NeuronForImpulse n) AxonConductive) +{-| +Type which helps you define a type of the result of 'growNeuron'. It takes type of the 'Neuron' you want to 'grow' as an argument +and specifies a 'Nerve' which is not conductive in any directions. +-} +type NerveNone n = Incubation (Nerve (NeuronFromImpulse n) AxonNonConductive (NeuronForImpulse n) AxonNonConductive) +{-| +Type which helps you define a type of the result of 'growNeuron'. It takes type of the 'Neuron' you want to 'grow' as an argument +and specifies a 'Nerve' which is conductive only in the direction from the 'Neuron'. +-} +type NerveOnlyFrom n = Incubation (Nerve (NeuronFromImpulse n) AxonConductive (NeuronForImpulse n) AxonNonConductive) +{-| +Type which helps you define a type of the result of 'growNeuron'. It takes type of the 'Neuron' you want to 'grow' as an argument +and specifies a 'Nerve' which is conductive only in the direction to the 'Neuron'. +-} +type NerveOnlyFor n = Incubation (Nerve (NeuronFromImpulse n) AxonNonConductive (NeuronForImpulse n) AxonConductive) + +class (Typeable a, Eq a) => ChanClass a where + neuronTypeOf :: a -> TypeRep + +instance Impulse i => ChanClass (Chan i) where + neuronTypeOf = head . typeRepArgs . head . typeRepArgs . typeOf -- we assume here that impulses are just NeuronFromImpulse or NeuronForImpulse + +data ChanBox where + ChanBox :: ChanClass a => a -> ChanBox + +instance Eq ChanBox where + ChanBox a == ChanBox b = typeOf a == typeOf b && cast a == Just b -- tests both typeOf and cast to be sure (cast could be defined to succeed for different types?) + +getFromChan :: Nerve from fromConductivity for forConductivity -> [ChanBox] +getFromChan (Nerve (Axon c) _) = [ChanBox c] +getFromChan (Nerve NoAxon _) = [] + +{-| +Branches 'Nerve' on the 'Neuron' side. This allows multiple 'Neuron's to be attached to it and still receive all 'Impulse's +(otherwise just the first 'Neuron' which would read from a 'Nerve' would receive a given 'Impulse'). +Only new 'Impulse's from a moment of branching on are conducted over new the branch, old 'Impulse's are not reconducted. +Branching can be applied multiple times. +-} +branchNerveFor :: Nerve from fromConductivity for AxonConductive -> IO (Nerve from fromConductivity for AxonConductive) +branchNerveFor (Nerve from (Axon c)) = do + c' <- dupChan c + return $ Nerve from (Axon c') + +{-| +Branches 'Nerve' on the other (non-'Neuron') side. This allows using the same 'Nerve' at multiple parts of the network (program) +and still receive all 'Impulse's from 'Neuron' at all parts of the network (otherwise just the first read from a 'Nerve' would +receive a given 'Impulse'). +Only new 'Impulse's from a moment of branching on are conducted over the new branch, old 'Impulse's are not reconducted. +Branching can be applied multiple times. +-} +branchNerveFrom :: Nerve from AxonConductive for forConductivity -> IO (Nerve from AxonConductive for forConductivity) +branchNerveFrom (Nerve (Axon c) for) = do + c' <- dupChan c + return $ Nerve (Axon c') for + +{-| +Branches 'Nerve' on both sides. Same as both 'branchNerveFor' and 'branchNerveFrom'. +-} +branchNerveBoth :: Nerve from AxonConductive for AxonConductive -> IO (Nerve from AxonConductive for AxonConductive) +branchNerveBoth = branchNerveFrom >=> branchNerveFor diff --git a/lib/Control/Etage/Internals.hs b/lib/Control/Etage/Internals.hs new file mode 100644 --- /dev/null +++ b/lib/Control/Etage/Internals.hs @@ -0,0 +1,118 @@ +{-# LANGUAGE TypeFamilies, MultiParamTypeClasses, GADTs, FlexibleInstances, FlexibleContexts, ScopedTypeVariables, TypeSynonymInstances, StandaloneDeriving, DeriveDataTypeable, EmptyDataDecls, NamedFieldPuns #-} +{-# OPTIONS_GHC -fno-warn-orphans #-} + +module Control.Etage.Internals ( + Axon(..), + Nerve(..), + Impulse(..), + LiveNeuron(..), + ImpulseValue, + ImpulseTime, + AxonConductive, + AxonNonConductive, + NeuronDissolved, + NeuronId, + waitForException +) where + +import Control.Concurrent hiding (Chan) +import Data.Time.Clock.POSIX +import Data.Typeable +import Numeric +import Text.ParserCombinators.ReadP + +import Control.Etage.Chan + +-- TODO: Find better general representation for values (something analog to what a hologram is, so that it can be gradually simplified and gradually reconstructed). Could be an Incubation program itself? +-- | Type of a general representation of 'Impulse' values (data payload). Currently it is just a list of 'Rational' values. +type ImpulseValue = [Rational] + +-- | Type of 'Impulse' timestamp. You can use 'getCurrentImpulseTime' for timestamp representing current time. +type ImpulseTime = POSIXTime + +instance Read ImpulseTime where + readsPrec _ r = do + (time, sec) <- readFloat r + ('s', rest) <- readP_to_S (char 's') sec + return (time, rest) + +{-| +Type class with common methods for impulses send over 'Nerve's and processed in 'Neuron's so that it is possible to define +'Neuron's which operate on any 'Impulse' type. An example of such 'Neuron' is "Control.Etage.Function". +-} +class (Show i, Typeable i) => Impulse i where + -- | This method should return a timestamp when the 'Impulse' was created/finalized what should be the moment just before it is send over + -- the 'Nerve'. So the moment it formed into its final form and started leaving the 'Neuron'. + -- + -- Be careful that Haskell is a lazy language so such code: + -- + -- > let v = lengthlyComputation + -- > time <- getCurrentImpulseTime + -- > sendFromNeuron nerve Value { impulseTimestamp = time, value = v } + -- + -- will evaluate @v@ after 'getCurrentImpulseTime' call. You should make @v@ strict (using @BangPatterns@) like: + -- + -- > let !v = lengthlyComputation + impulseTime :: i -> ImpulseTime + -- | This method should return all values (data payload) the 'Impulse' defines. Currently order and format is not yet finalized so + -- it is just a list of 'Rational' values in some order (for now it probably should be the order in which the values are defined + -- in the 'Impulse' constructor). + -- + -- It is meant to allow general 'Neurons' which can work on any 'Impulse' type to be developed. For example 'Neuron's which + -- implement some machine learning or data mining algorithms. It is on purpose that values are cleared of any semantic + -- meaning so algorithms have better chance not to get in touch with some unintended domain specific knowledge. + impulseValue :: i -> ImpulseValue + +{-| +Is axon (one direction of a 'Nerve') conductive? Yes, it is. + +This is type checked and enforced. If you define axon as conductive you have to make make sure that 'Impulse's send along it are +really read somewhere, otherwise a memory leak will occur. +-} +data AxonConductive deriving (Typeable) +{-| +Is axon (one direction of a 'Nerve') conductive? No, it is not. + +This is type checked and enforced. It is useful to specify nonconductive axons when you are not interested in 'Impulse's from a +particular axon (direction), making sure there will not be a memory leak because 'Impulse's would pile up. +-} +data AxonNonConductive deriving (Typeable) + +data Axon impulse conductivity where + Axon :: Impulse i => Chan i -> Axon i AxonConductive + NoAxon :: Axon i AxonNonConductive + +{-| +Type representing a 'Nerve' between 'Neuron's. It is bi-directional (from and to a 'Neuron', each direction being one axon) and you +can specify type of 'Impulse's traveling along the axon and its conductivity (with AxonConductive or +AxonNonConductive). + +You mostly do not need to specify this type manually if you are using 'growNeuron' and one of 'NerveBoth', 'NerveNone', +'NerveOnlyFrom' and 'NerveOnlyFor' types. +-} +data Nerve from fromConductivity for forConductivity where + Nerve :: (Impulse from, Impulse for) => Axon from fromConductivity -> Axon for forConductivity -> Nerve from fromConductivity for forConductivity + +deriving instance Typeable4 Nerve + +instance (Typeable forConductivity, Typeable fromConductivity, Typeable from, Typeable for) => Show (Nerve from fromConductivity for forConductivity) where + show = show . typeOf + +type NeuronDissolved = SampleVar () +type NeuronId = ThreadId + +deriving instance Typeable1 SampleVar + +instance Show NeuronDissolved where + show = show . typeOf + +{-| +Type representing a 'live' 'Neuron'. +-} +data LiveNeuron = LiveNeuron NeuronDissolved NeuronId deriving (Eq, Typeable) + +instance Show LiveNeuron where + show = show . typeOf + +waitForException :: IO a +waitForException = newEmptyMVar >>= takeMVar diff --git a/lib/Control/Etage/Propagate.hs b/lib/Control/Etage/Propagate.hs new file mode 100644 --- /dev/null +++ b/lib/Control/Etage/Propagate.hs @@ -0,0 +1,79 @@ +{-# LANGUAGE TypeFamilies, GADTs, ScopedTypeVariables, TypeSynonymInstances, StandaloneDeriving, DeriveDataTypeable, NamedFieldPuns #-} + +module Control.Etage.Propagate ( + propagate, + Translatable(..) +) where + +import Control.Monad +import Data.Typeable + +import Control.Etage.Internals +import Control.Etage.Externals + +-- TODO: Implement delay in propagation (constant delay, random from some distribution) + +data (Typeable from, Typeable for, Typeable forConductivity) => PropagateNeuron from for forConductivity = PropagateNeuron (PropagateOptions from for forConductivity) deriving (Typeable) + +type PropagateFromImpulse from for forConductivity = NeuronFromImpulse (PropagateNeuron from for forConductivity) +type PropagateForImpulse from for forConductivity = NeuronForImpulse (PropagateNeuron from for forConductivity) +type PropagateOptions from for forConductivity = NeuronOptions (PropagateNeuron from for forConductivity) + +{-| +Impulse instance for internal 'Neuron' which implements 'propagate'. +-} +instance (Typeable from, Typeable for, Typeable forConductivity) => Impulse (PropagateFromImpulse from for forConductivity) where + impulseTime _ = undefined + impulseValue _ = undefined + +{-| +Impulse instance for internal 'Neuron' which implements 'propagate'. +-} +instance (Typeable from, Typeable for, Typeable forConductivity) => Impulse (PropagateForImpulse from for forConductivity) where + impulseTime _ = undefined + impulseValue _ = undefined + +deriving instance Show (PropagateFromImpulse from for forConductivity) +deriving instance Show (PropagateForImpulse from for forConductivity) + +{-| +An internal 'Neuron' which implements 'propagate'. +-} +instance (Typeable from, Typeable for, Typeable forConductivity) => Neuron (PropagateNeuron from for forConductivity) where + data NeuronFromImpulse (PropagateNeuron from for forConductivity) + data NeuronForImpulse (PropagateNeuron from for forConductivity) + data NeuronOptions (PropagateNeuron from for forConductivity) = PropagateOptions { + from :: Nerve from AxonConductive for forConductivity, + for ::[Translatable from] + } + + mkDefaultOptions = return PropagateOptions { + from = undefined, + for = undefined + } + + grow options = return $ PropagateNeuron options + + live _ (PropagateNeuron PropagateOptions { from, for }) = forever $ do + i <- getFromNeuron from + mapM_ (\(Translatable n) -> translateAndSend n i) for + +{-| +It 'grow's an internal 'Neuron' which propagates 'Impulse's from a given 'Nerve' to other 'Nerve's, 'translate'-ing as necessary. + +Check 'attachTo' for a more high-level function (of 'Incubation') taking care of all the details (like branching 'Nerve's as necessary). +Use this function only if you are dealing with 'grow'ing and 'attach'ing of 'Nerve's directly. +-} +propagate :: forall from for forConductivity. (Typeable from, Typeable for, Typeable forConductivity) => Nerve from AxonConductive for forConductivity -> [Translatable from] -> IO () +propagate from for = do + -- we do not manage this neuron, it will be cleaned by RTS at program exit + -- TODO: What if this is not the only thing the program is doing? Should we cleanup this threads at the end of Incubation, too? + _ <- attach (\o -> o { from, for } :: NeuronOptions (PropagateNeuron from for forConductivity)) undefined + return () + +{-| +An existentially quantified types encompassing all 'Nerve's which can be 'translate'd from the same 'Impulse' type. Used in 'attachTo' +(and 'propagate') to list all 'Nerve's to which you want a given 'Nerve' to 'attach' to (and 'Impulse's to 'propagate'). +-} +data Translatable i where + Translatable :: ImpulseTranslator i for => Nerve from fromConductivity for AxonConductive -> Translatable i diff --git a/lib/Control/Etage/Sequence.hs b/lib/Control/Etage/Sequence.hs new file mode 100644 --- /dev/null +++ b/lib/Control/Etage/Sequence.hs @@ -0,0 +1,103 @@ +{-# LANGUAGE TypeFamilies, MultiParamTypeClasses, GADTs, FlexibleInstances, ScopedTypeVariables, DeriveDataTypeable, TypeSynonymInstances, StandaloneDeriving, NamedFieldPuns #-} + +{-| +This module defines a 'Neuron' which generates values based on a given sequence at a given interval. +If it reaches the end of a sequence it initiates 'dissolving'. +You 'grow' default version of it, which gives you an infinite source of random 'Int's at random interval of maximum length of 1 +second, in 'Incubation' by using something like: + +> nerveRandom <- growNeuron defaultOptions :: NerveOnlyFrom (SequenceNeuron Int) + +or for an infinite source of ones with same random interval: + +> nerveOnes <- growNeuron (\o -> o { valueSource = repeat 1 }) :: NerveOnlyFrom (SequenceNeuron Int) + +It is an example of a 'Neuron' with a parametrized type. +-} + +module Control.Etage.Sequence ( + SequenceNeuron, + SequenceFromImpulse, + SequenceForImpulse, + SequenceOptions, + NeuronFromImpulse(..), + NeuronForImpulse, + NeuronOptions(..) +) where + +import Control.Concurrent +import Control.Monad +import Data.Typeable +import System.Random + +import Control.Etage + +defaultMaxInterval :: Int +defaultMaxInterval = 1000000 -- microseconds, 1 second + +data (Real r, Random r, Show r, Typeable r) => SequenceNeuron r = SequenceNeuron (SequenceOptions r) deriving (Typeable) + +instance Typeable r => Show (SequenceNeuron r) where + show = show . typeOf + +{-| +'Impulse's from 'SequenceNeuron'. This 'Impulse' constructor is defined: + +[@Value { impulseTimestamp :: 'ImpulseTime', value :: 'Rational' }@] +@impulseTimestamp@ is time when the value was send, @value@ contains the value. +-} +type SequenceFromImpulse r = NeuronFromImpulse (SequenceNeuron r) +-- | 'Impulse's for 'SequenceNeuron'. This 'Neuron' does not define any 'Impulse's it would receive. +type SequenceForImpulse r = NeuronForImpulse (SequenceNeuron r) +{-| +Options for 'SequenceNeuron'. This options are defined: + +[@valueSource :: \[r\]@] The list of values to send. If the end of the list is reached, 'Neuron' initiates 'dissolving'. Default +is an infinite list of values of type @r@ generated by the 'StdGen' random generator. + +[@intervalSource :: \['Int'\]@] The list of intervals between values. It is defined as a delay in microseconds before the next value +is send. If the end of the list is reached, 'Neuron' initiates 'dissolving'. Default is a list of random delays with maximum +length of 1 second generated by the 'StdGen' random generator. +-} +type SequenceOptions r = NeuronOptions (SequenceNeuron r) + +-- | Impulse instance for 'SequenceNeuron'. +instance (Real r, Random r, Show r, Typeable r) => Impulse (SequenceFromImpulse r) where + impulseTime Value { impulseTimestamp } = impulseTimestamp + impulseValue Value { value } = [toRational value] + +-- | Impulse instance for 'SequenceNeuron'. +instance (Real r, Random r, Show r, Typeable r) => Impulse (SequenceForImpulse r) where + impulseTime _ = undefined + impulseValue _ = undefined + +deriving instance Show (SequenceForImpulse r) + +-- | A 'Neuron' which generates values based on a given sequence at a given interval. +instance (Real r, Random r, Show r, Typeable r) => Neuron (SequenceNeuron r) where + data NeuronFromImpulse (SequenceNeuron r) = Value { + impulseTimestamp :: ImpulseTime, -- time is first so that ordering is first by time + value :: r + } deriving (Eq, Ord, Read, Show) + data NeuronForImpulse (SequenceNeuron r) + data NeuronOptions (SequenceNeuron r) = SequenceOptions { + valueSource :: [r], + intervalSource :: [Int] -- microseconds + } deriving (Eq, Ord, Read, Show) + + mkDefaultOptions = do + generator <- newStdGen + generator' <- newStdGen + return SequenceOptions { + valueSource = randoms generator, + intervalSource = randomRs (0, defaultMaxInterval) generator' + } + + grow options = return $ SequenceNeuron options + + live nerve n@(SequenceNeuron SequenceOptions { valueSource, intervalSource }) = do + forM_ (zip valueSource intervalSource) $ \(v, i) -> do + threadDelay i + time <- getCurrentImpulseTime + sendFromNeuron nerve $ Value time v + dissolving n diff --git a/lib/Control/Etage/Timeout.hs b/lib/Control/Etage/Timeout.hs new file mode 100644 --- /dev/null +++ b/lib/Control/Etage/Timeout.hs @@ -0,0 +1,76 @@ +{-# LANGUAGE TypeFamilies, MultiParamTypeClasses, GADTs, FlexibleInstances, ScopedTypeVariables, DeriveDataTypeable, TypeSynonymInstances, StandaloneDeriving, NamedFieldPuns #-} + +{-| +This module defines a simple 'Neuron' which initiates 'dissolving' after a given delay. It can be used to limit execution time of +the network. You 'grow' it in 'Incubation' by using something like: + +> _ <- growNeuron defaultOptions :: NerveNone TimeoutNeuron + +somewhere among (best at the end) 'growNeuron' calls for other 'Neuron's in 'Incubation'. +-} + +module Control.Etage.Timeout ( + TimeoutNeuron, + TimeoutFromImpulse, + TimeoutForImpulse, + TimeoutOptions, + NeuronFromImpulse, + NeuronForImpulse, + NeuronOptions(..) +) where + +import Control.Concurrent +import Control.Monad +import Data.Typeable + +import Control.Etage + +defaultTimeout :: Int +defaultTimeout = 60000000 -- microseconds, 60 seconds + +data TimeoutNeuron = TimeoutNeuron TimeoutOptions deriving (Typeable) + +instance Show TimeoutNeuron where + show = show . typeOf + +-- | 'Impulse's from 'TimeoutNeuron'. This 'Neuron' does not define any 'Impulse's it would send. +type TimeoutFromImpulse = NeuronFromImpulse TimeoutNeuron +-- | 'Impulse's for 'TimeoutNeuron'. This 'Neuron' does not define any 'Impulse's it would receive. +type TimeoutForImpulse = NeuronForImpulse TimeoutNeuron +{-| +Options for 'TimeoutNeuron'. This option is defined: + +[@timeout :: 'Int'@] The length of the delay in microseconds before initiating 'dissolving'. Default is 60 seconds. +-} +type TimeoutOptions = NeuronOptions TimeoutNeuron + +-- | Impulse instance for 'TimeoutNeuron'. +instance Impulse TimeoutFromImpulse where + impulseTime _ = undefined + impulseValue _ = undefined + +-- | Impulse instance for 'TimeoutNeuron'. +instance Impulse TimeoutForImpulse where + impulseTime _ = undefined + impulseValue _ = undefined + +deriving instance Show TimeoutFromImpulse +deriving instance Show TimeoutForImpulse + +-- | A simple 'Neuron' which initiates 'dissolving' after a given delay. +instance Neuron TimeoutNeuron where + data NeuronFromImpulse TimeoutNeuron + data NeuronForImpulse TimeoutNeuron + data NeuronOptions TimeoutNeuron = TimeoutOptions { + timeout :: Int -- microseconds + } deriving (Eq, Ord, Read, Show) + + mkDefaultOptions = return TimeoutOptions { + timeout = defaultTimeout + } + + grow options = return $ TimeoutNeuron options + + live _ n@(TimeoutNeuron TimeoutOptions { timeout }) = do + threadDelay timeout + dissolving n diff --git a/lib/Control/Etage/Worker.hs b/lib/Control/Etage/Worker.hs new file mode 100644 --- /dev/null +++ b/lib/Control/Etage/Worker.hs @@ -0,0 +1,88 @@ +{-# LANGUAGE TypeFamilies, MultiParamTypeClasses, GADTs, FlexibleInstances, ScopedTypeVariables, TypeSynonymInstances, StandaloneDeriving, DeriveDataTypeable, EmptyDataDecls, NamedFieldPuns #-} + +{-| +This module defines a worker 'Neuron' which evaluates 'IO' actions it receives. It is useful to offload lengthly 'IO' actions +into another thread. In the case of too many queued 'IO' actions they are silently dropped and only newest ones are evaluated. +You 'grow' it in 'Incubation' by using something like: + +> nerveWorker <- growNeuron defaultOptions :: NerveOnlyFor WorkerNeuron +-} + +module Control.Etage.Worker ( + WorkerNeuron, + WorkerFromImpulse, + WorkerForImpulse, + WorkerOptions, + NeuronFromImpulse, + NeuronForImpulse(..), + NeuronOptions(..), + WorkType +) where + +import Control.Applicative +import Control.Monad +import Data.Typeable + +import Control.Etage + +-- | Type of work this worker 'Neuron' evaluates. +type WorkType = IO () + +instance Show WorkType where + show = show . typeOf + +-- TODO: We could maybe send results back? + +data WorkerNeuron deriving (Typeable) + +-- | 'Impulse's from 'WorkerNeuron'. This 'Neuron' does not define any 'Impulse's it would send. +type WorkerFromImpulse = NeuronFromImpulse WorkerNeuron +{-| +'Impulse's for 'WorkerNeuron'. This 'Impulse' constructor is defined: + +[@Work { impulseTimestamp :: ImpulseTime, work :: WorkType }@] +@impulseTimestamp@ is time when the action was enqueued for evaluation in the 'WorkerNeuron', @work@ is enqueued action. +-} +type WorkerForImpulse = NeuronForImpulse WorkerNeuron +{-| +Options for 'WorkerNeuron'. This option is defined: + +[@mapOnCapability :: 'NeuronMapCapability'@] How to map the 'Neuron' on capabilities (OS threads). With this option you can fix +multiple 'Neuron's on the same capability (for example, by generating one value with 'mkNeuronMapOnRandomCapability' and using +it for all those 'Neuron's) which is sometimes necessary when dealing with external (FFI) libraries. Default value +is 'NeuronFreelyMapOnCapability'. +-} +type WorkerOptions = NeuronOptions WorkerNeuron + +-- | Impulse instance for 'WorkerNeuron'. +instance Impulse WorkerFromImpulse where + impulseTime _ = undefined + impulseValue _ = undefined + +-- | Impulse instance for 'WorkerNeuron'. +instance Impulse WorkerForImpulse where + impulseTime Work { impulseTimestamp } = impulseTimestamp + impulseValue _ = [] + +deriving instance Show WorkerFromImpulse + +-- | A worker 'Neuron' which evaluates 'IO' actions it receives. +instance Neuron WorkerNeuron where + data NeuronFromImpulse WorkerNeuron + data NeuronForImpulse WorkerNeuron = Work { + impulseTimestamp :: ImpulseTime, + work :: WorkType + } deriving (Show) + data NeuronOptions WorkerNeuron = WorkerOptions { + mapOnCapability :: NeuronMapCapability + } deriving (Eq, Ord, Read, Show) + + mkDefaultOptions = return WorkerOptions { + mapOnCapability = NeuronFreelyMapOnCapability + } + + getNeuronMapCapability WorkerOptions { mapOnCapability } = mapOnCapability + + live nerve _ = forever $ do + Work { work } <- head <$> waitAndSlurpForNeuron nerve -- just newest + work