STREAM-PROGRAMMIERUNG IN HASKELL (Daniel van den Eijkel 9/2003) GHC 5.02 WinXP


Eine einfache Möglichkeit der Streamprogrammierung in Haskell besteht darin, 
Listen als Streams zu verwenden. Da Haskell-Ausdrücke verzögert ausgewertet
werden, können Listen so definiert werden, dass sie kein Ende besitzen, sondern
theoretisch unendlich viele Elemente enthalten können. Die Verwendung solcher Listen 
als Streams ist ein bekanntes Thema.

Der Nachteil von Listen als Streams besteht darin, dass diese Streams keine externen
Quellen wie z.B. Dateien besitzen können, da der Zugriff auf solche Quellen IO-Aktionen
erfordert, die nicht in den Typ der Liste integriert werden können. Generell schließt die 
Verwendung von Listen in diesem Kontext sämtliche IO-Aktionen und Seiteneffekte aus, will 
man keine schmutzigen Tricks (z.B. "unsafePerformIO" oder auch "fixIO") anwenden. Deshalb 
sollten Streams nicht als Listen, sondern als IO-Aktionen konzipiert werden. Dies wird im 
Folgenden beschrieben.

> module Stream 
>  ( M (..)     -- newtype M a = M { unM :: IO (S a, IO ()) }, Functor, Monad, MonadPlus M
>  , unitM      -- :: a -> M a
>  , joinM      -- :: M (M a) -> M a
>  , zeroM      -- :: M a
>  , appendM    -- :: M a -> M a -> M a
>  , zipM       -- :: (a -> b -> c) -> M a -> M b -> M c
>  , zipM'      -- :: (a,b) -> (a -> b -> c) -> M a -> M b -> M c
>  , sToM       -- :: S a -> M a
>  , repeatM    -- :: a -> M a
>  , takeM      -- :: Int -> M a -> M a
>  , takeM'     -- :: Int -> a -> M a -> M a
>  , dropM      -- :: Int -> M a -> M a
>  , listToM    -- :: [a] -> M a
>  , delayM     -- :: Int -> a -> M a -> M a
>  , runM       -- :: M a -> (a -> IO ()) -> IO ()
>  , outputM    -- :: FilePath -> M Double -> IO ()
>  , inputM     -- :: FilePath -> M Double
>  , cloneS     -- :: S a -> IO (S a, S a)
>  , branchM    -- :: (M a -> M a -> M b) -> M a -> M b
>  ) 
>  where
>
> import IOExts
> import Foreign
> import CString
> import Monad


1. Stream S _______________________________________________________________________________________


> newtype S a = S { unS :: IO (Maybe a) }

Ein Stream S vom Typ a liefert potentiell unendlich viele Elemente vom Typ a. Um das aktuelle 
bzw nächste Element aus dem Stream auszulesen, wird die IO-Aktion, die den Stream repräsentiert, 
ausgeführt. Sie liefert einen Wert von Typ 'Maybe a', der entweder einen Wert vom Typ a enthält 
- das ist das aktuelle Element - oder 'Nothing' ist, was bedeutet, dass der Stream am Ende ist und
keine Elemente mehr enthält. Die Ausführung der Aktion liefert aber nicht nur das nächste Element, 
falls eines existiert, sondern sie verändert zugleich den Zustand des Streams, so dass bei der nächsten 
Ausführung der Aktion das folgende Element ausgelesen wird. Enthält ein Stream n Elemente, so kann die 
Aktion n-mal ausgeführt werden, wobei sie die Elemente der Reihe nach ausliest, und bei jedem weiteren
Aufruf der Aktion wird sie 'Nothing' liefern. Es kann aber auch sein, dass der Stream niemals 'Nothing'
liefert, d.h. der Stream ist unendlich lang.
Anmerkung: Der Stream S wird als newtype deklariert und getagt: Er besitzt einen Konstruktor S und einen
Dekonstruktor unS. Diese werden bei der Compilierung des Programms entfernt und stellen lediglich so
etwas wie syntaktischen Zucker dar. Konkret wird dadurch aber eine Instanziierung von Klassen für den 
Stream ermöglicht. 

> instance Functor S where
>  fmap f (S ioma) = S $ fmap (fmap f) ioma

Streams sind als Funktoren instantiiert. Die Bedeutung dürfte klar sein.

> instance Monad S where
>  return a = S $ return $ Just a
>  S ioma >>= f = S $ fmap (maybe (return Nothing) (unS.f)) ioma >>= id

Streams sind ebenfalls als Monaden instantiiert. 
return erzeugt aus einem beliebigen Element vom Typ a einen unendlichen Stream, der bei jedem
Aufruf dieses Element liefert. 
(>>=) bindet einen Stream a an eine Funktion f :: (a -> Stream b) und führt diese aus, nachdem a aus dem
Stream gelesen wurde. Liefert der Stream Nothing, dann ist das Ergebnis von (stream >>= f) ebenfalls
Nothing. 

> zipS :: (a->b->c) -> S a -> S b -> S c
> zipS f s1 s2 = s1 >>= \a -> s2 >>= \b -> return (f a b)

Die Funktion zipS hat eine analoge Bedeutung zur zip-Funktion für Listen. Sobald einer
der beiden Streams am Ende ist, liefert zipS Nothing.

> zipS' :: (a,b) -> (a->b->c) -> S a -> S b -> S c
> zipS' (c1,c2) f (S s1) (S s2) = S $ 
>  s1 >>= \a ->
>  case a of
>   Nothing -> s2 >>= \b ->
>              case b of 
>               Nothing -> return Nothing
>               Just v2 -> return (Just (f c1 v2))
>   Just v1 -> s2 >>= \b ->
>              case b of 
>               Nothing -> return (Just (f v1 c2))
>               Just v2 -> return (Just (f v1 v2))

zipS' liefert im Gegensatz zu zipS erst dann Nothing, wenn beide Streams am Ende sind. Ist hingegen nur
ein Stream am Ende, wird statt der Elemente aus diesem Stream eine Konstante benutzt. 

> ioToS :: IO a -> S a
> ioToS ioa = S $ ioa >>= return . Just

ioToS wandelt eine IO-Aktion vom Typ IO a in einen Stream um. Dazu muss der Resultatyp a in den 
'Maybe a'-Typ des Streams verwandelt werden. Der resultierende Stream ist unendlich lang.
Um eine IO-Aktion vom Typ IO (Maybe a) in einen Stream zu verwandeln, muss diese lediglich mit dem
Konstruktor S getagt werden.


2. MakeStream M ___________________________________________________________________________________


Ein Stream S kann beispielsweise eine Datei repräsentieren. Bei jedem Aufruf des Streams wird das nächste
Zeichen aus der Datei gelesen und der Dateizeiger rückt ein Element weiter. Nachdem er das Ende der Datei
erreicht hat, liefert ein Aufruf nur noch Nothing. Um aber eine Datei als Stream verwenden zu können, muss
sie zunächst geöffnet werden. Zudem sollte es die Möglichkeit geben, die Datei zu schliessen, wenn der Stream
Nothing liefert oder wenn er nicht mehr benötigt wird. 

Neben der Stream-Aktion, die die Elemente liefert, werden also zwei weitere Aktionen benötigt, um den
Stream öffnen und wieder schliessen zu können. 

> newtype M a = M { unM :: IO (S a, IO ()) }

Der Datentyp M steht für MakeStream und stellt einen Streamkonstruktor dar (Mir ist bislang kein besseres Wort 
eingefallen). Sobald die Aktion, die in M eingekapselt ist, ausgeführt wird, wird der Stream geöffnet. Sie 
liefert als Resultat einerseits den geöffneten Stream, andererseits eine Aktion, die diesen Stream wieder 
schliesst. 

> instance Functor M where
>  fmap f (M iom) = M $ fmap (\(s,c) -> (fmap f s,c)) iom

Streamkonstruktoren sind als Funktoren instantiiert. Die Anwendung von "fmap f" auf einen Streamkonstruktor 
entspricht der Anwendung von "fmap f" auf den Stream, den der Streamkonstruktor erzeugt.

> instance Monad M where
>  return = unitM
>  m >>= f = joinM $ fmap f m

> unitM :: a -> M a
> unitM a = M $ do
>  r <- newIORef True
>  let next = readIORef r >>= \b ->
>             writeIORef r False >>
>             if b then return (Just a) else return Nothing
>  return (S next, return ())

> joinM :: M (M a) -> M a
> joinM (M iotop) = M $ do
>  (S nextm, close) <- iotop
>  miocur <- nextm
>  let setup :: Maybe (M a) -> IO (Maybe (S a, IO ()))
>      setup m = case m of
>                 Nothing    -> return Nothing
>                 Just (M f) -> f >>= return . Just
>  mscur <- setup miocur
>  rmcur <- newIORef mscur
>  let next = readIORef rmcur >>= \mscur ->
>             case mscur of
>              Nothing      -> return Nothing
>              Just (S g,c) -> g >>= \mv -> 
>                              case mv of
>                               Nothing -> c >> nextm >>= setup >>= writeIORef rmcur >> next
>                               jv      -> return jv
>      close' = readIORef rmcur >>= maybe (return ()) (\ (_,c) -> c) >> close
>  return $ (S next, close')

Streamkonstruktoren sind als Monaden instantiiert. 
return erzeugt einen Streamkonstruktor, der bei Ausführung einen Stream mit genau einem Element erzeugt.
(>>=) führt einen Streamkonstruktor m aus, wodurch dieser einen Stream öffnet. Die Funktion f :: (a -> M b)
wird durch fmap auf jeden der Werte von m angewendet, wodurch ein Stream von Streamkonstruktoren S (M b) 
entsteht. Dieser Stream von Streamkonstruktoren wird durch die Funktion joinM in einen einfachen Stream S b
umgewandelt, indem jeder Streamkonstruktor aus S (M b) zuerst ausgeführt wird, dann wird der dadurch 
erzeugte Stream bis an sein Ende ausgelesen, er wird geschlossen, und dann wird das Ganze für den nächsten 
Streamkonstruktor wiederholt, bis S (M b) keine Elemente mehr liefert. 
Dies ist analog zur Monadeninstantiierung der Liste, joinM entspricht der concat-Funktion.

> instance MonadPlus M where
>  mzero = zeroM
>  mplus = appendM

> zeroM :: M a
> zeroM = M $ return (S $ return Nothing, return ())

> appendM :: M a -> M a -> M a
> appendM (M ioa) (M iob) = M $ do
>  f <- ioa
>  rb <- newIORef False
>  rf <- newIORef f
>  let next = readIORef rf >>= \(S s, c) ->
>             s >>= \mv ->
>             case mv of
>              Nothing -> readIORef rb >>= \b -> 
>                         if b 
>                          then return Nothing
>                          else c >> writeIORef rb True >>
>                               iob >>= writeIORef rf >>
>                               next
>              x       -> return x
>      close = readIORef rf >>= \(_,c) -> c
>  return $ (S next, close)

Streamkonstruktoren sind ebenfalls als MonadPlus instantiiert. 
mzero ist ein Nullelement. Es handelt sich hier um einen Streamkonstruktor, der einen Stream erzeugt, der
keine Elemente liefert. Dieses Element ist gefählich, denn in Verbindung mit joinM kann es zu einem
"hängenbleibenden" Programm führen. Dieses Problem gibt es aber auch bei Listen, z.B 
 let k = [] ++ k in (null k)   oder   (concat k)
mplus bzw appendM hängt zwei Streamkonstruktoren aneinander, so dass der resultierende Streamkonstruktor zuerst 
den einen Stream erzeugt, ihn ausliest, schliesst und dann den anderen erzeugt, ausliest und schliesst.


> zipM :: (a -> b -> c) -> M a -> M b -> M c
> zipM f (M io1) (M io2) = M $ do 
>  (s1,c1) <- io1 
>  (s2,c2) <- io2
>  return (zipS f s1 s2, c1 >> c2)


> zipM' :: (a,b) -> (a -> b -> c) -> M a -> M b -> M c
> zipM' c f (M io1) (M io2) = M $ do 
>  (s1,c1) <- io1 
>  (s2,c2) <- io2
>  return (zipS' c f s1 s2, c1 >> c2)

zipM und zipM' verwenden zipS bzw zipS', um zwei Streams zu "zippen". 

> sToM :: S a -> M a
> sToM s = M $ return (s, return ())

sToM wandelt einen Stream in einen Streamkonstruktor um. Dies geht nur bei Streams, die nicht in besonderer
Weise geöffnet und geschlossen werden müssen.

> repeatM :: a -> M a
> repeatM a = sToM (return a)

repeatM a ist ein Streamkonstruktor, der einen unendlichen Stream erzeugt. Alle Elemente dieses Streams
stimmen mit dem Argument a überein. Der Ausdruck "joinM $ repeatM m" loopt einen beliebeigen 
Streamkonstruktor m, d.h er erzeugt bei Ausführung einen unendlichen Stream, der aus einer unendlichfachen
Wiederholung des von m erzeugten Streams besteht, sofern dieser endlich ist. Falls dieser unendlich ist, so 
entspricht der Ausdruck dem Ausdruck "m". 

> takeM :: Int -> M a -> M a
> takeM n (M ioa) = M $ do
>  (S s,c) <- ioa
>  rc      <- newIORef 0
>  let next = readIORef rc >>= \c ->
>             if c >= n
>              then return Nothing
>              else writeIORef rc (c+1) >> s
>  return (S next, c)

> takeM' :: Int -> a -> M a -> M a
> takeM' n a (M ioa) = M $ do
>  (S s,c) <- ioa
>  rc      <- newIORef 0
>  let next = readIORef rc >>= \c ->
>             if c >= n
>              then return Nothing
>              else writeIORef rc (c+1) >>
>                   s >>= \mv -> 
>                   return $ maybe (Just a) Just mv
>  return (S next, c)

takeM n m schneidet den durch m erzeugten Stream nach n Elementen ab. Falls m weniger als n Elemente
liefert, endet der resultierende Stream schon nach weniger als n Elementen.
takeM' n a m hingegen ist ein Streamkonstrukter, der einen Stream mit garantierter Länge n erzeugt. 
Falls der Stream von m kürzer ist, wird der Wert a zum "Auffüllen" genommen. 

> dropM :: Int -> M a -> M a
> dropM n (M ioa) = M $ do
>  (S s,c) <- ioa
>  let loop 0 = return ()
>      loop k = s >> loop (k-1)
>  loop n
>  return (S s, c)

dropM n m erzeugt bei Ausführung den Stream von m, liest n Elemente, und liefert erst dann den 
geöffneten Stream. Das bedeutet, dass die ersten n Elemente des Streams von m abgeschnitten werden, 
und das eigentliche Auslesen der Werte erst danach beginnt. 

> listToM :: [a] -> M a
> listToM xs = M $ do
>  r <- newIORef xs
>  let next = readIORef r >>= \xs ->
>             if null xs
>              then return Nothing 
>              else writeIORef r (tail xs) >> 
>                   return (Just (head xs))
>  return (S next, return ())

listToM wandelt eine Liste in einen Streamkonstruktor um. 

> delayM :: Int -> a -> M a -> M a
> delayM n a (M ioa) = M $ do
>  arr     <- newIOArray (0,n-1) a
>  rp      <- newIORef 0
>  (S s,c) <- ioa
>  let next = s >>= \mv ->
>             case mv of
>              Nothing -> return Nothing
>              Just v  -> readIORef rp >>= \pos ->
>                         readIOArray arr pos >>= \v' ->
>                         writeIOArray arr pos v >>
>                         let pos' = mod (1+pos) n in
>                         writeIORef rp pos' >>
>                         return (Just v')
>  return (S next, c)

delayM n a verzögert einen Stream um n Elemente, wobei die ersten n Aufrufe den Wert a liefern.

3. Ausführung der Streams _________________________________________________________________________


> runM :: M a -> (a -> IO ()) -> IO ()
> runM (M iom) cont = do
>  (S s, c) <- iom
>  let loop = s >>= maybe c (\a -> cont a >> loop)
>  loop

Um einen Streamkonstruktor auszuführen, muss die Funktion runM verwendet werden. Sie erwartet neben
dem auszuführenden Streamkonstruktor eine Continuation, die sie mit jedem aus dem Stream empfangenen
Wert nacheinander ausführt. 
runM führt den Streamkonstruktor aus, wodurch der entsprechende Stream geöffnet wird. Dann wird eine
Schleife aufgerufen, die den jeweils nächsten Wert aus dem Stream liesst und ihn an die Continuation
gibt. Sobald diese terminiert, wird die Schleife wieder aufgerufen. Wenn der Stream keine Werte mehr
liefert, wird die Schleife verlassen und der Stream geschlossen.
Eine Continuation könnte "print" sein oder eine Aktion, die Daten an eine Datei anhängt. 

> outputM :: FilePath -> M Double -> IO ()
> outputM filepath m = do
>  (write, close) <- openOutput filepath
>  runM m write
>  close

outputM öffnet eine Datei, wodurch es eine Continuation "write" bekommt. Zusammen mit dem auszuführenden 
Streamkonstruktor wird diese an runM gegeben. Nachdem runM terminiert, wird die Datei, in der nun 
die Werte aus dem Stream gespeichert sind, wieder geschlossen. 

> inputM :: FilePath -> M Double
> inputM name = M $ do
>  (s,c) <- openInput name
>  return (S s, c)

inputM verwandelt einen Dateinamen in einen Streamkonstruktor. Bei dessen Ausführung wird die 
entsprechende Datei als Stream geöffnet.


4. Weiterführendes ________________________________________________________________________________


> cloneS :: S a -> IO (S a, S a)
> cloneS (S ioma) = do
>  r <- newIORef Nothing
>  let f1 = ioma >>= \ma -> writeIORef r ma >> return ma
>      f2 = readIORef r
>  return (S f1, S f2)

cloneS ist eine Funktion, die eine unechte Kopie eines Streams erzeugt, die mit dem Stream synchronisiert
ist. Die Kopie des Streams liefert bei Aufruf den selben Wert, den der Original-Stream als letztes 
geliefert hat. Die Verwendung dieser Funktion ist gefährlich und sollte nur systematisch stattfinden:
Wird die Kopie mehrmals aufgerufen mit der Intention, aufeinanderfolgende Werte auszulesen, muss zwischen
den Aufrufen jeweils der Original-Stream aufgerufen werden. 

> branchM :: (M a -> M a -> M b) -> M a -> M b
> branchM f (M iom) = M $ do
>  (s, c) <- iom
>  (s,s') <- cloneS s
>  (r,c') <- unM $ f (sToM s) (sToM s')
>  return (r, c' >> c)

branchM verwendet cloneS, um eine Kopie eines Streams zu erzeugen. Diese wird gemeinsam mit dem Original
an eine Funktion geliefert, die unter Verwendung beider Streams einen neuen Streamkonstruktor liefert.
Zumindest garantiert branchM, dass die Kopie des Streams nur in einem begrenzten Bereich sichtbar ist, 
innerhalb dem der Original-Stream geöffnet ist. 

Geplant sind folgende Erweiterungen:

- Ein Datentyp, der die Streamkonstruktoren kapselt und besonders das Problem der Synchronisierung von 
  Kopie und Originalstream hinter einem Typsystem versteckt.
- Manche Streamkonstruktoren sind asynchron, d.h. die Streams, die sie erzeugen, lesen nicht genau so
  viele Elemente wie sie liefern. (Analog zu "Listenfressen" oder blindem Buffer). Beispiele sind
  dropS und joinS. Dies muss berücksichtigt werden bei der Synchronisation von Kopien.
- Es soll neben dem beschriebenen Stream-Typ noch einen anderen Stream-Typ geben, der nicht einzelne
  Werte liefert, sondern ganze Datenblöcke von Buffer zu Buffer verschiebt. Dieser Anspruch bringt aber
  einige Probleme mit sich, die hier nicht diskutiert werden sollen. 
  ( Stream S entspricht einer C-Funktion "read ()", die ein byte z.B. aus einer Datei liest, aber einen
    Int-Wert zurückgibt. Dies ist notwendig, um mit dem Wert -1 das Ende der Datei anzeigen zu können.
    In Haskell wird hier dafür der Maybe-Typ verwendet.
    Ein Stream, der mit Buffern arbeitet, müsste analog zur Funktion "read (char * arr)" funktionieren,
    deren Seiteneffekt im Füllen von arr besteht und dessen Resultatwert die Anzahl der bytes anzeigt.
    Hier würde der Wert 0 das Ende der Datei anzeigen. Das Problem bei Buffern ist aber unter anderem, 
    dass eine fmap f mit f :: (a->b) auf jeden Fall auf zwei Buffer operieren muss. Schnelle 
    Operationen, die den Buffer "in place" modifizieren, sind nicht möglich. Ausserdem sollten die Buffer 
    kompatibel mit C-Funktionen sein, was entweder eine Beschränkung auf primitive Datentypen bedeutet
    oder aber mindestens zwei Arten von Buffers (für Storable (primitive Typen) und allgemeine Typen) 
    erfordert.)


5. Low-Level Dateioperationen _____________________________________________________________________


Die Implementierung der Dateioperationen liegt hier zum Teil in C vor. Es besteht jedoch 
kein zwingender Grund dafür. 
Anmerkung: Die openInput-Funktion ist dafür ausgelegt, wav-Dateien zu lesen. Diese besitzen
einen Header der Länge 44 bytes, der übersprungen wird. 
Die Ausgabe-Dateien besitzen jedoch keinen Header. Dieser muss entweder nachträglich
hinzugefügt werden, oder man benögt ein Programm, das Dateien ohne Header interpretieren kann - im
Falle von wav-Dateien beispielsweise "Cool Edit" von Syntrillium. Das Programm "AddWaveHeader" macht 
aus einer beliebigen Datei eine wave-Datei (16bit 44100khz mono) ACHTUNG DESTRUKTIV !!! 

> data BufferedInputFile_
> type BufferedInputFile = Ptr BufferedInputFile_

> foreign import openBufferedInputFile    :: CString -> Int -> IO BufferedInputFile
> foreign import closeBufferedInputFile   :: BufferedInputFile -> IO ()
> foreign import fillBufferedInputFile    :: BufferedInputFile -> Ptr Double -> IO Int
> foreign import newDArray                :: Int -> IO (Ptr Double)
> foreign import freeDArray               :: Ptr Double -> IO ()

> openInput :: FilePath -> IO (IO (Maybe Double), IO ())
> openInput name = do
>  cname <- newCString name
>  bif   <- openBufferedInputFile cname 1
>  buf   <- newDArray 1024
>  mx    <- fillBufferedInputFile bif buf
>  rmx   <- newIORef mx
>  rpos  <- newIORef 0
>  let next = readIORef rmx  >>= \mx ->
>             readIORef rpos >>= \pos ->
>             if pos >= mx
>              then if mx < 1024
>                    then return Nothing
>                    else fillBufferedInputFile bif buf >>= 
>                         writeIORef rmx >>
>                         writeIORef rpos 0 >>
>                         next
>              else writeIORef rpos (pos+1) >>
>                   peekElemOff buf pos >>= \d ->
>                   return (Just d)
>      close = freeDArray buf >> closeBufferedInputFile bif
>  return (next, close)


> data FileDestination 
> foreign import openFileDestination       :: CString -> Int -> IO (Ptr FileDestination)
> foreign import closeFileDestination      :: Ptr FileDestination -> IO ()
> foreign import outputNextFileDestination :: Ptr FileDestination -> Double -> IO ()
> openOutput :: FilePath -> IO (Double -> IO (), IO ())
> openOutput filepath = do
>  filename <- newCString filepath
>  fos      <- openFileDestination filename 1000
>  return (outputNextFileDestination fos, closeFileDestination fos)


Anhang: C-Datei "cbase.c" _________________________________________________________________________


compilieren: ghc -c cbase.c 
             ghc -c Stream.lhs -package lang -fglasgow-exts
             ghc main.hs cbase.o Stream.o -package lang -fglasgow-exts

 GHC 5.02, Windows XP
////////////////////////////////////////////////

 #include <stdio.h>
 #include <io.h>
 #include <fcntl.h>
 #include <stdlib.h>
 #include <math.h>

// BufferedInputFile 

typedef struct
{ int file;
  char * buffer;
} BufferedInputFile;

BufferedInputFile * openBufferedInputFile (char * filename, int hasHeader)
{ char * bu;
  BufferedInputFile * fi;
  fi = malloc (sizeof (BufferedInputFile));
  fi->file = open (filename, O_RDONLY | O_BINARY);
  fi->buffer = malloc (2048);
  if (hasHeader != 0)
  { bu = malloc (44);
    read (fi->file, bu, 44);  
    free (bu);
  }
  return fi;
}

void closeBufferedInputFile (BufferedInputFile  * fi)
{ //printf ("call bif close");
  close (fi->file);
  free (fi->buffer);
  free (fi);
  //printf ("bif close done");
}

int fillBufferedInputFile (BufferedInputFile * fi, double * buf)
{ 
  int n = read (fi->file, fi->buffer, 2048);
  int x, y, i;
  char c1, c2;
  int l = n / 2;
  for (i=0; i<l; i++)
  { int k = i * 2;
    c1 = fi->buffer [k];
    c2 = fi->buffer [k+1];
    x = (int) c1;
    y = (int) c2;
    if (x < 0)
      x += 256;
    x += 256 * y;
    buf [i] = (double) x;
  }
  return l;
}

// FileDestination 

typedef struct
{ int file;
  int pos;
  int len;
  char * buffer;
} FileDestination;

FileDestination * openFileDestination (char * filename, int bufsize)
{ FileDestination * fd = malloc (sizeof (FileDestination));
  fd->file = open (filename, O_WRONLY | O_CREAT | O_TRUNC | O_BINARY , 0777);
  fd->pos = 0;
  fd->len = bufsize;
  fd->buffer = malloc (bufsize);
  return fd;
}

void closeFileDestination (FileDestination * fd)
{ write (fd->file, fd->buffer, fd->pos);
  close (fd->file);
  free (fd->buffer);
  free (fd);
}

void outputNextFileDestination (FileDestination * fd, double d)
{ int v = (int) d;
  fd->buffer [fd->pos++] = (char) (v % 256);
  fd->buffer [fd->pos++] = (char) ((v >> 8) % 256);
  if (fd->pos >= fd->len)
  { write (fd->file, fd->buffer, fd->pos);
    fd->pos = 0;
  }
} 

// DArray

double * newDArray (int n)
{ double * arr = malloc (n * 8);
  int i;
  for (i=0; i<n; i++)
    arr [i] = 0.0;
  return arr;
}

void freeDArray (double * b)
{ free (b);
}