X-Git-Url: http://git.megacz.com/?a=blobdiff_plain;f=GHC%2FConc.lhs;h=0b6bc91385edf4ff19050ab5209cffe64d478acf;hb=30464c0cb915c2ae900909568fa8677bba341e45;hp=d30513dc9fe7c7e62431f1b5c1e1d6911512253a;hpb=98c3a9174b16c68a9b2a8f196418425c152b632c;p=haskell-directory.git

diff --git a/GHC/Conc.lhs b/GHC/Conc.lhs
index d30513d..0b6bc91 100644
--- a/GHC/Conc.lhs
+++ b/GHC/Conc.lhs
@@ -1,5 +1,5 @@
 \begin{code}
-{-# OPTIONS -fno-implicit-prelude #-}
+{-# OPTIONS_GHC -fno-implicit-prelude #-}
 -----------------------------------------------------------------------------
 -- |
 -- Module      :  GHC.Conc
@@ -14,11 +14,19 @@
 -- 
 -----------------------------------------------------------------------------
 
-#include "config.h"
+-- No: #hide, because bits of this module are exposed by the stm package.
+-- However, we don't want this module to be the home location for the
+-- bits it exports, we'd rather have Control.Concurrent and the other
+-- higher level modules be the home.  Hence:
+
+-- #not-home
 module GHC.Conc
 	( ThreadId(..)
 
-	-- Forking and suchlike
+	-- * Forking and suchlike
+	, forkIO	-- :: IO a -> IO ThreadId
+	, forkOnIO	-- :: Int -> IO a -> IO ThreadId
+	, childHandler  -- :: Exception -> IO ()
 	, myThreadId 	-- :: IO ThreadId
 	, killThread	-- :: ThreadId -> IO ()
 	, throwTo       -- :: ThreadId -> Exception -> IO ()
@@ -27,12 +35,13 @@ module GHC.Conc
 	, yield         -- :: IO ()
 	, labelThread	-- :: ThreadId -> String -> IO ()
 
-	-- Waiting
+	-- * Waiting
 	, threadDelay	  	-- :: Int -> IO ()
+	, registerDelay		-- :: Int -> IO (TVar Bool)
 	, threadWaitRead	-- :: Int -> IO ()
 	, threadWaitWrite	-- :: Int -> IO ()
 
-	-- MVars
+	-- * MVars
 	, MVar		-- abstract
 	, newMVar 	-- :: a -> IO (MVar a)
 	, newEmptyMVar  -- :: IO (MVar a)
@@ -43,7 +52,21 @@ module GHC.Conc
 	, isEmptyMVar	-- :: MVar a -> IO Bool
 	, addMVarFinalizer -- :: MVar a -> IO () -> IO ()
 
-#ifdef mingw32_TARGET_OS
+   	-- * TVars
+	, STM           -- abstract
+	, atomically    -- :: STM a -> IO a
+	, retry         -- :: STM a
+	, orElse        -- :: STM a -> STM a -> STM a
+        , catchSTM      -- :: STM a -> (Exception -> STM a) -> STM a
+	, TVar          -- abstract
+	, newTVar 	-- :: a -> STM (TVar a)
+	, newTVarIO 	-- :: a -> STM (TVar a)
+	, readTVar	-- :: TVar a -> STM a
+	, writeTVar	-- :: a -> TVar a -> STM ()
+	, unsafeIOToSTM	-- :: IO a -> STM a
+
+   	-- * Miscellaneous
+#ifdef mingw32_HOST_OS
 	, asyncRead     -- :: Int -> Int -> Int -> Ptr a -> IO (Int, Int)
 	, asyncWrite    -- :: Int -> Int -> Int -> Ptr a -> IO (Int, Int)
 	, asyncDoProc   -- :: FunPtr (Ptr a -> IO Int) -> Ptr a -> IO Int
@@ -51,18 +74,34 @@ module GHC.Conc
 	, asyncReadBA   -- :: Int -> Int -> Int -> Int -> MutableByteArray# RealWorld -> IO (Int, Int)
 	, asyncWriteBA  -- :: Int -> Int -> Int -> Int -> MutableByteArray# RealWorld -> IO (Int, Int)
 #endif
+
+#ifndef mingw32_HOST_OS
+	, ensureIOManagerIsRunning
+#endif
         ) where
 
+import System.Posix.Types
+import System.Posix.Internals
+import Foreign
+import Foreign.C
+
+#ifndef __HADDOCK__
+import {-# SOURCE #-} GHC.TopHandler ( reportError, reportStackOverflow )
+#endif
+
 import Data.Maybe
 
 import GHC.Base
-import GHC.IOBase	( IO(..), MVar(..), ioException, IOException(..), IOErrorType(..) )
-import GHC.Num		( fromInteger, negate )
-import GHC.Real		( fromIntegral )
+import GHC.IOBase
+import GHC.Num		( Num(..) )
+import GHC.Real		( fromIntegral, quot )
 import GHC.Base		( Int(..) )
-import GHC.Exception    ( Exception(..), AsyncException(..) )
+import GHC.Exception    ( catchException, Exception(..), AsyncException(..) )
 import GHC.Pack		( packCString# )
 import GHC.Ptr          ( Ptr(..), plusPtr, FunPtr(..) )
+import GHC.STRef
+import GHC.Show		( Show(..), showString )
+import Data.Typeable
 
 infixr 0 `par`, `pseq`
 \end{code}
@@ -74,7 +113,7 @@ infixr 0 `par`, `pseq`
 %************************************************************************
 
 \begin{code}
-data ThreadId = ThreadId ThreadId#
+data ThreadId = ThreadId ThreadId# deriving( Typeable )
 -- ToDo: data ThreadId = ThreadId (Weak ThreadId#)
 -- But since ThreadId# is unlifted, the Weak type must use open
 -- type variables.
@@ -96,7 +135,69 @@ This misfeature will hopefully be corrected at a later date.
 it defines 'ThreadId' as a synonym for ().
 -}
 
---forkIO has now been hoisted out into the Concurrent library.
+instance Show ThreadId where
+   showsPrec d t = 
+   	showString "ThreadId " . 
+        showsPrec d (getThreadId (id2TSO t))
+
+foreign import ccall unsafe "rts_getThreadId" getThreadId :: ThreadId# -> Int
+
+id2TSO :: ThreadId -> ThreadId#
+id2TSO (ThreadId t) = t
+
+foreign import ccall unsafe "cmp_thread" cmp_thread :: ThreadId# -> ThreadId# -> CInt
+-- Returns -1, 0, 1
+
+cmpThread :: ThreadId -> ThreadId -> Ordering
+cmpThread t1 t2 = 
+   case cmp_thread (id2TSO t1) (id2TSO t2) of
+      -1 -> LT
+      0  -> EQ
+      _  -> GT -- must be 1
+
+instance Eq ThreadId where
+   t1 == t2 = 
+      case t1 `cmpThread` t2 of
+         EQ -> True
+         _  -> False
+
+instance Ord ThreadId where
+   compare = cmpThread
+
+{- |
+This sparks off a new thread to run the 'IO' computation passed as the
+first argument, and returns the 'ThreadId' of the newly created
+thread.
+
+The new thread will be a lightweight thread; if you want to use a foreign
+library that uses thread-local storage, use 'forkOS' instead.
+-}
+forkIO :: IO () -> IO ThreadId
+forkIO action = IO $ \ s -> 
+   case (fork# action_plus s) of (# s1, id #) -> (# s1, ThreadId id #)
+ where
+  action_plus = catchException action childHandler
+
+forkOnIO :: Int -> IO () -> IO ThreadId
+forkOnIO (I# cpu) action = IO $ \ s -> 
+   case (forkOn# cpu action_plus s) of (# s1, id #) -> (# s1, ThreadId id #)
+ where
+  action_plus = catchException action childHandler
+
+childHandler :: Exception -> IO ()
+childHandler err = catchException (real_handler err) childHandler
+
+real_handler :: Exception -> IO ()
+real_handler ex =
+  case ex of
+	-- ignore thread GC and killThread exceptions:
+	BlockedOnDeadMVar            -> return ()
+	BlockedIndefinitely          -> return ()
+	AsyncException ThreadKilled  -> return ()
+
+	-- report all others:
+	AsyncException StackOverflow -> reportStackOverflow
+	other       -> reportError other
 
 {- | 'killThread' terminates the given thread (GHC only).
 Any work already done by the thread isn\'t
@@ -118,7 +219,13 @@ target thread.  The calling thread can thus be certain that the target
 thread has received the exception.  This is a useful property to know
 when dealing with race conditions: eg. if there are two threads that
 can kill each other, it is guaranteed that only one of the threads
-will get to kill the other. -}
+will get to kill the other.
+
+If the target thread is currently making a foreign call, then the
+exception will not be raised (and hence 'throwTo' will not return)
+until the call has completed.  This is the case regardless of whether
+the call is inside a 'block' or not.
+ -}
 throwTo :: ThreadId -> Exception -> IO ()
 throwTo (ThreadId id) ex = IO $ \ s ->
    case (killThread# id ex s) of s1 -> (# s1, () #)
@@ -173,6 +280,110 @@ par :: a -> b -> b
 par  x y = case (par# x) of { _ -> lazy y }
 \end{code}
 
+
+%************************************************************************
+%*									*
+\subsection[stm]{Transactional heap operations}
+%*									*
+%************************************************************************
+
+TVars are shared memory locations which support atomic memory
+transactions.
+
+\begin{code}
+-- |A monad supporting atomic memory transactions.
+newtype STM a = STM (State# RealWorld -> (# State# RealWorld, a #)) deriving( Typeable )
+
+unSTM :: STM a -> (State# RealWorld -> (# State# RealWorld, a #))
+unSTM (STM a) = a
+
+instance  Functor STM where
+   fmap f x = x >>= (return . f)
+
+instance  Monad STM  where
+    {-# INLINE return #-}
+    {-# INLINE (>>)   #-}
+    {-# INLINE (>>=)  #-}
+    m >> k      = thenSTM m k
+    return x	= returnSTM x
+    m >>= k     = bindSTM m k
+
+bindSTM :: STM a -> (a -> STM b) -> STM b
+bindSTM (STM m) k = STM ( \s ->
+  case m s of 
+    (# new_s, a #) -> unSTM (k a) new_s
+  )
+
+thenSTM :: STM a -> STM b -> STM b
+thenSTM (STM m) k = STM ( \s ->
+  case m s of 
+    (# new_s, a #) -> unSTM k new_s
+  )
+
+returnSTM :: a -> STM a
+returnSTM x = STM (\s -> (# s, x #))
+
+-- | Unsafely performs IO in the STM monad.
+unsafeIOToSTM :: IO a -> STM a
+unsafeIOToSTM (IO m) = STM m
+
+-- |Perform a series of STM actions atomically.
+atomically :: STM a -> IO a
+atomically (STM m) = IO (\s -> (atomically# m) s )
+
+-- |Retry execution of the current memory transaction because it has seen
+-- values in TVars which mean that it should not continue (e.g. the TVars
+-- represent a shared buffer that is now empty).  The implementation may
+-- block the thread until one of the TVars that it has read from has been
+-- udpated. (GHC only)
+retry :: STM a
+retry = STM $ \s# -> retry# s#
+
+-- |Compose two alternative STM actions (GHC only).  If the first action
+-- completes without retrying then it forms the result of the orElse.
+-- Otherwise, if the first action retries, then the second action is
+-- tried in its place.  If both actions retry then the orElse as a
+-- whole retries.
+orElse :: STM a -> STM a -> STM a
+orElse (STM m) e = STM $ \s -> catchRetry# m (unSTM e) s
+
+-- |Exception handling within STM actions.
+catchSTM :: STM a -> (Exception -> STM a) -> STM a
+catchSTM (STM m) k = STM $ \s -> catchSTM# m (\ex -> unSTM (k ex)) s
+
+-- |Shared memory locations that support atomic memory transactions.
+data TVar a = TVar (TVar# RealWorld a) deriving( Typeable )
+
+instance Eq (TVar a) where
+	(TVar tvar1#) == (TVar tvar2#) = sameTVar# tvar1# tvar2#
+
+-- |Create a new TVar holding a value supplied
+newTVar :: a -> STM (TVar a)
+newTVar val = STM $ \s1# ->
+    case newTVar# val s1# of
+	 (# s2#, tvar# #) -> (# s2#, TVar tvar# #)
+
+-- |@IO@ version of 'newTVar'.  This is useful for creating top-level
+-- 'TVar's using 'System.IO.Unsafe.unsafePerformIO', because using
+-- 'atomically' inside 'System.IO.Unsafe.unsafePerformIO' isn't
+-- possible.
+newTVarIO :: a -> IO (TVar a)
+newTVarIO val = IO $ \s1# ->
+    case newTVar# val s1# of
+	 (# s2#, tvar# #) -> (# s2#, TVar tvar# #)
+
+-- |Return the current value stored in a TVar
+readTVar :: TVar a -> STM a
+readTVar (TVar tvar#) = STM $ \s# -> readTVar# tvar# s#
+
+-- |Write the supplied value into a TVar
+writeTVar :: TVar a -> a -> STM ()
+writeTVar (TVar tvar#) val = STM $ \s1# ->
+    case writeTVar# tvar# val s1# of
+    	 s2# -> (# s2#, () #)
+  
+\end{code}
+
 %************************************************************************
 %*									*
 \subsection[mvars]{M-Structures}
@@ -207,16 +418,34 @@ newMVar value =
 -- empty, 'takeMVar' will wait until it is full.  After a 'takeMVar', 
 -- the 'MVar' is left empty.
 -- 
--- If several threads are competing to take the same 'MVar', one is chosen
--- to continue at random when the 'MVar' becomes full.
+-- There are two further important properties of 'takeMVar':
+--
+--   * 'takeMVar' is single-wakeup.  That is, if there are multiple
+--     threads blocked in 'takeMVar', and the 'MVar' becomes full,
+--     only one thread will be woken up.  The runtime guarantees that
+--     the woken thread completes its 'takeMVar' operation.
+--
+--   * When multiple threads are blocked on an 'MVar', they are
+--     woken up in FIFO order.  This is useful for providing
+--     fairness properties of abstractions built using 'MVar's.
+--
 takeMVar :: MVar a -> IO a
 takeMVar (MVar mvar#) = IO $ \ s# -> takeMVar# mvar# s#
 
 -- |Put a value into an 'MVar'.  If the 'MVar' is currently full,
 -- 'putMVar' will wait until it becomes empty.
 --
--- If several threads are competing to fill the same 'MVar', one is
--- chosen to continue at random with the 'MVar' becomes empty.
+-- There are two further important properties of 'putMVar':
+--
+--   * 'putMVar' is single-wakeup.  That is, if there are multiple
+--     threads blocked in 'putMVar', and the 'MVar' becomes empty,
+--     only one thread will be woken up.  The runtime guarantees that
+--     the woken thread completes its 'putMVar' operation.
+--
+--   * When multiple threads are blocked on an 'MVar', they are
+--     woken up in FIFO order.  This is useful for providing
+--     fairness properties of abstractions built using 'MVar's.
+--
 putMVar  :: MVar a -> a -> IO ()
 putMVar (MVar mvar#) x = IO $ \ s# ->
     case putMVar# mvar# x s# of
@@ -266,52 +495,21 @@ addMVarFinalizer (MVar m) finalizer =
 %*									*
 %************************************************************************
 
-@threadWaitRead@ delays rescheduling of a thread until input on the
-specified file descriptor is available for reading (just like select).
-@threadWaitWrite@ is similar, but for writing on a file descriptor.
-
 \begin{code}
--- | Suspends the current thread for a given number of microseconds
--- (GHC only).
---
--- Note that the resolution used by the Haskell runtime system's
--- internal timer is 1\/50 second, and 'threadDelay' will round down
--- its argument to the nearest multiple of this resolution.  In
--- particular, to get the smallest non-zero delay, pass 20000 as the
--- parameter to 'threadDelay'.
---
--- There is no guarantee that the thread will be rescheduled promptly
--- when the delay has expired, but the thread will never continue to
--- run /earlier/ than specified.
---
-threadDelay :: Int -> IO ()
-
--- | Block the current thread until data is available to read on the
--- given file descriptor (GHC only).
-threadWaitRead :: Int -> IO ()
-
--- | Block the current thread until data can be written to the
--- given file descriptor (GHC only).
-threadWaitWrite :: Int -> IO ()
-
-threadDelay     (I# ms) = IO $ \s -> case delay# ms s     of s -> (# s, () #)
-threadWaitRead  (I# fd) = IO $ \s -> case waitRead# fd s  of s -> (# s, () #)
-threadWaitWrite (I# fd) = IO $ \s -> case waitWrite# fd s of s -> (# s, () #)
-
-#ifdef mingw32_TARGET_OS
+#ifdef mingw32_HOST_OS
 
 -- Note: threadDelay, threadWaitRead and threadWaitWrite aren't really functional
 -- on Win32, but left in there because lib code (still) uses them (the manner
 -- in which they're used doesn't cause problems on a Win32 platform though.)
 
 asyncRead :: Int -> Int -> Int -> Ptr a -> IO (Int, Int)
-asyncRead  (I# fd) (I# isSock) (I# len) (Ptr buf) = 
-  IO $ \s -> case asyncRead# fd isSock len buf s  of 
+asyncRead  (I# fd) (I# isSock) (I# len) (Ptr buf) =
+  IO $ \s -> case asyncRead# fd isSock len buf s of 
   	       (# s, len#, err# #) -> (# s, (I# len#, I# err#) #)
 
 asyncWrite :: Int -> Int -> Int -> Ptr a -> IO (Int, Int)
-asyncWrite  (I# fd) (I# isSock) (I# len) (Ptr buf) = 
-  IO $ \s -> case asyncWrite# fd isSock len buf s  of 
+asyncWrite  (I# fd) (I# isSock) (I# len) (Ptr buf) =
+  IO $ \s -> case asyncWrite# fd isSock len buf s of 
   	       (# s, len#, err# #) -> (# s, (I# len#, I# err#) #)
 
 asyncDoProc :: FunPtr (Ptr a -> IO Int) -> Ptr a -> IO Int
@@ -334,4 +532,392 @@ asyncWriteBA fd isSock len off bufB =
   asyncWrite fd isSock len ((Ptr (byteArrayContents# (unsafeCoerce# bufB))) `plusPtr` off)
 
 #endif
+
+-- -----------------------------------------------------------------------------
+-- Thread IO API
+
+-- | Block the current thread until data is available to read on the
+-- given file descriptor (GHC only).
+threadWaitRead :: Fd -> IO ()
+threadWaitRead fd
+#ifndef mingw32_HOST_OS
+  | threaded  = waitForReadEvent fd
+#endif
+  | otherwise = IO $ \s -> 
+	case fromIntegral fd of { I# fd# ->
+	case waitRead# fd# s of { s -> (# s, () #)
+	}}
+
+-- | Block the current thread until data can be written to the
+-- given file descriptor (GHC only).
+threadWaitWrite :: Fd -> IO ()
+threadWaitWrite fd
+#ifndef mingw32_HOST_OS
+  | threaded  = waitForWriteEvent fd
+#endif
+  | otherwise = IO $ \s -> 
+	case fromIntegral fd of { I# fd# ->
+	case waitWrite# fd# s of { s -> (# s, () #)
+	}}
+
+-- | Suspends the current thread for a given number of microseconds
+-- (GHC only).
+--
+-- Note that the resolution used by the Haskell runtime system's
+-- internal timer is 1\/50 second, and 'threadDelay' will round its
+-- argument up to the nearest multiple of this resolution.
+--
+-- There is no guarantee that the thread will be rescheduled promptly
+-- when the delay has expired, but the thread will never continue to
+-- run /earlier/ than specified.
+--
+threadDelay :: Int -> IO ()
+threadDelay time
+#ifndef mingw32_HOST_OS
+  | threaded  = waitForDelayEvent time
+#else
+  | threaded  = c_Sleep (fromIntegral (time `quot` 1000))
+#endif
+  | otherwise = IO $ \s -> 
+	case fromIntegral time of { I# time# ->
+	case delay# time# s of { s -> (# s, () #)
+	}}
+
+registerDelay :: Int -> IO (TVar Bool)
+registerDelay usecs 
+#ifndef mingw32_HOST_OS
+  | threaded = waitForDelayEventSTM usecs
+  | otherwise = error "registerDelay: requires -threaded"
+#else
+  = error "registerDelay: not currently supported on Windows"
+#endif
+
+-- On Windows, we just make a safe call to 'Sleep' to implement threadDelay.
+#ifdef mingw32_HOST_OS
+foreign import stdcall safe "Sleep" c_Sleep :: CInt -> IO ()
+#endif
+
+foreign import ccall unsafe "rtsSupportsBoundThreads" threaded :: Bool
+
+-- ----------------------------------------------------------------------------
+-- Threaded RTS implementation of threadWaitRead, threadWaitWrite, threadDelay
+
+-- In the threaded RTS, we employ a single IO Manager thread to wait
+-- for all outstanding IO requests (threadWaitRead,threadWaitWrite)
+-- and delays (threadDelay).  
+--
+-- We can do this because in the threaded RTS the IO Manager can make
+-- a non-blocking call to select(), so we don't have to do select() in
+-- the scheduler as we have to in the non-threaded RTS.  We get performance
+-- benefits from doing it this way, because we only have to restart the select()
+-- when a new request arrives, rather than doing one select() each time
+-- around the scheduler loop.  Furthermore, the scheduler can be simplified
+-- by not having to check for completed IO requests.
+
+-- Issues, possible problems:
+--
+--	- we might want bound threads to just do the blocking
+--	  operation rather than communicating with the IO manager
+--	  thread.  This would prevent simgle-threaded programs which do
+--	  IO from requiring multiple OS threads.  However, it would also
+--	  prevent bound threads waiting on IO from being killed or sent
+--	  exceptions.
+--
+--	- Apprently exec() doesn't work on Linux in a multithreaded program.
+--	  I couldn't repeat this.
+--
+-- 	- How do we handle signal delivery in the multithreaded RTS?
+--
+--	- forkProcess will kill the IO manager thread.  Let's just
+--	  hope we don't need to do any blocking IO between fork & exec.
+
+#ifndef mingw32_HOST_OS
+
+data IOReq
+  = Read   {-# UNPACK #-} !Fd {-# UNPACK #-} !(MVar ())
+  | Write  {-# UNPACK #-} !Fd {-# UNPACK #-} !(MVar ())
+
+data DelayReq
+  = Delay    {-# UNPACK #-} !Int {-# UNPACK #-} !(MVar ())
+  | DelaySTM {-# UNPACK #-} !Int {-# UNPACK #-} !(TVar Bool)
+
+pendingEvents :: IORef [IOReq]
+pendingDelays :: IORef [DelayReq]
+	-- could use a strict list or array here
+{-# NOINLINE pendingEvents #-}
+{-# NOINLINE pendingDelays #-}
+(pendingEvents,pendingDelays) = unsafePerformIO $ do
+  startIOManagerThread
+  reqs <- newIORef []
+  dels <- newIORef []
+  return (reqs, dels)
+	-- the first time we schedule an IO request, the service thread
+	-- will be created (cool, huh?)
+
+ensureIOManagerIsRunning :: IO ()
+ensureIOManagerIsRunning 
+  | threaded  = seq pendingEvents $ return ()
+  | otherwise = return ()
+
+startIOManagerThread :: IO ()
+startIOManagerThread = do
+        allocaArray 2 $ \fds -> do
+	throwErrnoIfMinus1 "startIOManagerThread" (c_pipe fds)
+	rd_end <- peekElemOff fds 0
+	wr_end <- peekElemOff fds 1
+	writeIORef stick (fromIntegral wr_end)
+	c_setIOManagerPipe wr_end
+	forkIO $ do
+	    allocaBytes sizeofFdSet   $ \readfds -> do
+	    allocaBytes sizeofFdSet   $ \writefds -> do 
+	    allocaBytes sizeofTimeVal $ \timeval -> do
+	    service_loop (fromIntegral rd_end) readfds writefds timeval [] []
+	return ()
+
+service_loop
+   :: Fd		-- listen to this for wakeup calls
+   -> Ptr CFdSet
+   -> Ptr CFdSet
+   -> Ptr CTimeVal
+   -> [IOReq]
+   -> [DelayReq]
+   -> IO ()
+service_loop wakeup readfds writefds ptimeval old_reqs old_delays = do
+
+  -- pick up new IO requests
+  new_reqs <- atomicModifyIORef pendingEvents (\a -> ([],a))
+  let reqs = new_reqs ++ old_reqs
+
+  -- pick up new delay requests
+  new_delays <- atomicModifyIORef pendingDelays (\a -> ([],a))
+  let  delays = foldr insertDelay old_delays new_delays
+
+  -- build the FDSets for select()
+  fdZero readfds
+  fdZero writefds
+  fdSet wakeup readfds
+  maxfd <- buildFdSets 0 readfds writefds reqs
+
+  -- perform the select()
+  let do_select delays = do
+	  -- check the current time and wake up any thread in
+	  -- threadDelay whose timeout has expired.  Also find the
+	  -- timeout value for the select() call.
+	  now <- getTicksOfDay
+	  (delays', timeout) <- getDelay now ptimeval delays
+
+	  res <- c_select ((max wakeup maxfd)+1) readfds writefds 
+			nullPtr timeout
+	  if (res == -1)
+	     then do
+		err <- getErrno
+		case err of
+		  _ | err == eINTR ->  do_select delays'
+			-- EINTR: just redo the select()
+		  _ | err == eBADF ->  return (True, delays)
+			-- EBADF: one of the file descriptors is closed or bad,
+			-- we don't know which one, so wake everyone up.
+		  _ | otherwise    ->  throwErrno "select"
+			-- otherwise (ENOMEM or EINVAL) something has gone
+			-- wrong; report the error.
+	     else
+		return (False,delays')
+
+  (wakeup_all,delays') <- do_select delays
+
+  exit <-
+    if wakeup_all then return False
+      else do
+        b <- fdIsSet wakeup readfds
+        if b == 0 
+          then return False
+          else alloca $ \p -> do 
+	         c_read (fromIntegral wakeup) p 1; return ()
+		 s <- peek p		
+		 case s of
+		  _ | s == io_MANAGER_WAKEUP -> return False
+		  _ | s == io_MANAGER_DIE    -> return True
+		  _ -> do handler_tbl <- peek handlers
+		          sp <- peekElemOff handler_tbl (fromIntegral s)
+		          forkIO (do io <- deRefStablePtr sp; io)
+		          return False
+
+  if exit then return () else do
+
+  takeMVar prodding
+  putMVar prodding False
+
+  reqs' <- if wakeup_all then do wakeupAll reqs; return []
+			 else completeRequests reqs readfds writefds []
+
+  service_loop wakeup readfds writefds ptimeval reqs' delays'
+
+stick :: IORef Fd
+{-# NOINLINE stick #-}
+stick = unsafePerformIO (newIORef 0)
+
+io_MANAGER_WAKEUP = 0xff :: CChar
+io_MANAGER_DIE    = 0xfe :: CChar
+
+prodding :: MVar Bool
+{-# NOINLINE prodding #-}
+prodding = unsafePerformIO (newMVar False)
+
+prodServiceThread :: IO ()
+prodServiceThread = do
+  b <- takeMVar prodding
+  if (not b) 
+    then do fd <- readIORef stick
+	    with io_MANAGER_WAKEUP $ \pbuf -> do 
+		c_write (fromIntegral fd) pbuf 1; return ()
+    else return ()
+  putMVar prodding True
+
+foreign import ccall "&signal_handlers" handlers :: Ptr (Ptr (StablePtr (IO ())))
+
+foreign import ccall "setIOManagerPipe"
+  c_setIOManagerPipe :: CInt -> IO ()
+
+-- -----------------------------------------------------------------------------
+-- IO requests
+
+buildFdSets maxfd readfds writefds [] = return maxfd
+buildFdSets maxfd readfds writefds (Read fd m : reqs)
+  | fd >= fD_SETSIZE =  error "buildFdSets: file descriptor out of range"
+  | otherwise        =  do
+	fdSet fd readfds
+        buildFdSets (max maxfd fd) readfds writefds reqs
+buildFdSets maxfd readfds writefds (Write fd m : reqs)
+  | fd >= fD_SETSIZE =  error "buildFdSets: file descriptor out of range"
+  | otherwise        =  do
+	fdSet fd writefds
+	buildFdSets (max maxfd fd) readfds writefds reqs
+
+completeRequests [] _ _ reqs' = return reqs'
+completeRequests (Read fd m : reqs) readfds writefds reqs' = do
+  b <- fdIsSet fd readfds
+  if b /= 0
+    then do putMVar m (); completeRequests reqs readfds writefds reqs'
+    else completeRequests reqs readfds writefds (Read fd m : reqs')
+completeRequests (Write fd m : reqs) readfds writefds reqs' = do
+  b <- fdIsSet fd writefds
+  if b /= 0
+    then do putMVar m (); completeRequests reqs readfds writefds reqs'
+    else completeRequests reqs readfds writefds (Write fd m : reqs')
+
+wakeupAll [] = return ()
+wakeupAll (Read  fd m : reqs) = do putMVar m (); wakeupAll reqs
+wakeupAll (Write fd m : reqs) = do putMVar m (); wakeupAll reqs
+
+waitForReadEvent :: Fd -> IO ()
+waitForReadEvent fd = do
+  m <- newEmptyMVar
+  atomicModifyIORef pendingEvents (\xs -> (Read fd m : xs, ()))
+  prodServiceThread
+  takeMVar m
+
+waitForWriteEvent :: Fd -> IO ()
+waitForWriteEvent fd = do
+  m <- newEmptyMVar
+  atomicModifyIORef pendingEvents (\xs -> (Write fd m : xs, ()))
+  prodServiceThread
+  takeMVar m
+
+-- XXX: move into GHC.IOBase from Data.IORef?
+atomicModifyIORef :: IORef a -> (a -> (a,b)) -> IO b
+atomicModifyIORef (IORef (STRef r#)) f = IO $ \s -> atomicModifyMutVar# r# f s
+
+-- -----------------------------------------------------------------------------
+-- Delays
+
+waitForDelayEvent :: Int -> IO ()
+waitForDelayEvent usecs = do
+  m <- newEmptyMVar
+  now <- getTicksOfDay
+  let target = now + usecs `quot` tick_usecs
+  atomicModifyIORef pendingDelays (\xs -> (Delay target m : xs, ()))
+  prodServiceThread
+  takeMVar m
+
+-- Delays for use in STM
+waitForDelayEventSTM :: Int -> IO (TVar Bool)
+waitForDelayEventSTM usecs = do
+   t <- atomically $ newTVar False
+   now <- getTicksOfDay
+   let target = now + usecs `quot` tick_usecs
+   atomicModifyIORef pendingDelays (\xs -> (DelaySTM target t : xs, ()))
+   prodServiceThread
+   return t  
+    
+-- Walk the queue of pending delays, waking up any that have passed
+-- and return the smallest delay to wait for.  The queue of pending
+-- delays is kept ordered.
+getDelay :: Ticks -> Ptr CTimeVal -> [DelayReq] -> IO ([DelayReq], Ptr CTimeVal)
+getDelay now ptimeval [] = return ([],nullPtr)
+getDelay now ptimeval all@(d : rest) 
+  = case d of
+     Delay time m | now >= time -> do
+	putMVar m ()
+	getDelay now ptimeval rest
+     DelaySTM time t | now >= time -> do
+	atomically $ writeTVar t True
+	getDelay now ptimeval rest
+     _otherwise -> do
+	setTimevalTicks ptimeval (delayTime d - now)
+    	return (all,ptimeval)
+
+insertDelay :: DelayReq -> [DelayReq] -> [DelayReq]
+insertDelay d [] = [d]
+insertDelay d1 ds@(d2 : rest)
+  | delayTime d1 <= delayTime d2 = d1 : ds
+  | otherwise                    = d2 : insertDelay d1 rest
+
+delayTime (Delay t _) = t
+delayTime (DelaySTM t _) = t
+
+type Ticks = Int
+tick_freq  = 50 :: Ticks  -- accuracy of threadDelay (ticks per sec)
+tick_usecs = 1000000 `quot` tick_freq :: Int
+
+newtype CTimeVal = CTimeVal ()
+
+foreign import ccall unsafe "sizeofTimeVal"
+  sizeofTimeVal :: Int
+
+foreign import ccall unsafe "getTicksOfDay" 
+  getTicksOfDay :: IO Ticks
+
+foreign import ccall unsafe "setTimevalTicks" 
+  setTimevalTicks :: Ptr CTimeVal -> Ticks -> IO ()
+
+-- ----------------------------------------------------------------------------
+-- select() interface
+
+-- ToDo: move to System.Posix.Internals?
+
+newtype CFdSet = CFdSet ()
+
+foreign import ccall safe "select"
+  c_select :: Fd -> Ptr CFdSet -> Ptr CFdSet -> Ptr CFdSet -> Ptr CTimeVal
+           -> IO CInt
+
+foreign import ccall unsafe "hsFD_SETSIZE"
+  fD_SETSIZE :: Fd
+
+foreign import ccall unsafe "hsFD_CLR"
+  fdClr :: Fd -> Ptr CFdSet -> IO ()
+
+foreign import ccall unsafe "hsFD_ISSET"
+  fdIsSet :: Fd -> Ptr CFdSet -> IO CInt
+
+foreign import ccall unsafe "hsFD_SET"
+  fdSet :: Fd -> Ptr CFdSet -> IO ()
+
+foreign import ccall unsafe "hsFD_ZERO"
+  fdZero :: Ptr CFdSet -> IO ()
+
+foreign import ccall unsafe "sizeof_fd_set"
+  sizeofFdSet :: Int
+
+#endif
 \end{code}