diff --git a/bin/plox-read.hs b/bin/plox-read.hs
--- a/bin/plox-read.hs
+++ b/bin/plox-read.hs
@@ -15,7 +15,7 @@
     case args of
         [tag] -> do
             hSetBuffering stdout LineBuffering
-            MP.withSource tag service
+            MP.withSourceWait tag service
         _ -> usage
 
 service :: MP.Source -> IO ()
diff --git a/lib/System/Miniplex/Sekrit.hs b/lib/System/Miniplex/Sekrit.hs
--- a/lib/System/Miniplex/Sekrit.hs
+++ b/lib/System/Miniplex/Sekrit.hs
@@ -5,12 +5,23 @@
     , bytesFromInt
     , intFromBytes
     , closeOnExec
+    , enoents
+    , eexists
+    , mode644
 ) where
 
+import Control.Exception
+import Control.Monad
+
 import Foreign.C.Error
+
 import Network.Socket
+
 import Data.Bits
+
 import System.Directory
+import System.IO.Error
+import System.Posix.Files
 import System.Posix.IO
 import System.Posix.Types
 
@@ -18,25 +29,25 @@
 good s = not (null s) && head s /= '.' && all (`elem` chs) s
     where
     chs =
-        ['a' .. 'z'] ++ ['A' .. 'Z'] ++ ['0' .. '9'] ++ ",$=-#%.+_^&'@~()[]{}"
+        ['a' .. 'z'] ++ ['A' .. 'Z'] ++ ['0' .. '9'] ++ ",=-#%.+_^&'@~()[]{}"
 
-pathFromTag :: String -> String -> IO String
+pathFromTag :: String -> String -> IO (String, String, String)
 pathFromTag context what
     | not $ good what =
         ioError $ errnoToIOError context eADDRNOTAVAIL Nothing (Just what)
     | otherwise = do
         dir <- getAppUserDataDirectory "miniplex"
         createDirectoryIfMissing False dir
-        return $ dir ++ "/" ++ what
+        let d x =  dir ++ "/" ++ x
+        return (d what, d $ "L$$", d $ "R$" ++ what)
 
 reallySend :: Socket -> String -> IO ()
-reallySend sock str = step (length str) str
+reallySend sock = step
     where
-    step n msg
-        | n <= 0 = return ()
-        | otherwise = do
-            w <- send sock msg
-            step (n - w) (drop w msg)
+    step "" = return ()
+    step msg = do
+        w <- send sock msg
+        step (drop w msg)
 
 reallyRecv :: Socket -> Int -> IO String
 reallyRecv sock = step
@@ -67,3 +78,18 @@
         (`dupTo` Fd (fdSocket s))
     return ()
 -}
+
+enoents :: Exception -> Maybe Exception
+enoents e = do
+    io <- ioErrors e
+    guard $ isDoesNotExistError io
+    return e
+
+eexists :: Exception -> Maybe Exception
+eexists e = do
+    io <- ioErrors e
+    guard $ isAlreadyExistsError io
+    return e
+
+mode644 :: FileMode
+mode644 = ownerReadMode .|. ownerWriteMode .|. groupReadMode .|. otherReadMode
diff --git a/lib/System/Miniplex/Sink.hs b/lib/System/Miniplex/Sink.hs
--- a/lib/System/Miniplex/Sink.hs
+++ b/lib/System/Miniplex/Sink.hs
@@ -26,6 +26,8 @@
 import Control.Monad
 import Control.Monad.Fix
 import System.Directory
+import System.IO.Lock
+import System.Posix.IO
 import qualified Network as N
 import Network.Socket
 
@@ -93,19 +95,38 @@
 -- in your @~\/.miniplex\/@.)
 create :: String -> IO Sink
 create what = do
-    n <- pathFromTag "System.Miniplex.Sink.create" what
-    s <- N.listenOn (N.UnixSocket n)
+    (com, lck, ret) <- pathFromTag "System.Miniplex.Sink.create" what
+    bracketOnError (N.listenOn (N.UnixSocket com)) sClose $ \s -> do
     closeOnExec s
+    bracket
+        (openFd lck WriteOnly (Just mode644) defaultFileFlags)
+        (\fd -> closeFd fd)
+        $ \lf -> do
+    bracket (setLockAll lf LockWrite) unLock $ \_ -> do
+    brush ret
     vs <- atomically $ newTVar []
     z <- atomically $ newTVar Nothing
     t <- forkClose s (barghest s vs z)
     return $ Sink
-        { name = n
+        { name = com
         , sock = s
         , ctrl = vs
         , acct = t
         , zmsg = z
         }
+
+brush :: String -> IO ()
+brush file = do
+    flip finally (del file) $ do
+    r <- tryJust enoents $ openFd file WriteOnly Nothing defaultFileFlags{ nonBlock = True }
+    case r of
+        Left _ -> return ()
+        Right fd -> do
+            flip finally (closeFd fd) $ do
+            fdWrite fd "\0"
+            return ()
+    where
+    del x = handleJust ioErrors (const $ return ()) $ removeFile x
 
 -- | @'write' si msg@ asynchronously writes @msg@ to @si@, where it will be
 -- received by all currently connected readers.
diff --git a/lib/System/Miniplex/Source.hs b/lib/System/Miniplex/Source.hs
--- a/lib/System/Miniplex/Source.hs
+++ b/lib/System/Miniplex/Source.hs
@@ -4,25 +4,32 @@
 module System.Miniplex.Source (
     Source,
     attach,
+    attachWait,
     read,
     getMsgs,
     detach,
-    withSource
+    withSource,
+    withSourceWait
 ) where
 
 import System.Miniplex.Sekrit
 
-import Prelude hiding (read)
+import Prelude hiding (read, catch)
 
+import Control.Concurrent
 import Control.Exception
 import Control.Monad
+import Control.Monad.Fix
 import Control.Monad.Reader ()
 
 import Data.Typeable
 
 import Network.Socket
 
+import System.IO.Lock
 import System.IO.Unsafe
+import System.Posix.Files
+import System.Posix.IO
 
 newtype Source = Source {
     sock :: Socket
@@ -33,16 +40,38 @@
 -- exception is thrown.
 attach :: String -> IO Source
 attach what = do
-    n <- pathFromTag "System.Miniplex.Source.attach" what
-    bracketOnError
-        (socket AF_UNIX Stream 0)
-        sClose
-        $ \s -> do
-            closeOnExec s
-            connect s (SockAddrUnix n)
-            shutdown s ShutdownSend
-            return $ Source {sock = s}
+    (n, _, _) <- pathFromTag "System.Miniplex.Source.attach" what
+    bracketOnError (socket AF_UNIX Stream 0) sClose $ \s -> do
+        closeOnExec s
+        connect s (SockAddrUnix n)
+        shutdown s ShutdownSend
+        return $ Source s
 
+-- | Similar to 'attach', but if the specified sink doesn't exist, 'attachWait'
+-- blocks until it becomes available.
+attachWait :: String -> IO Source
+attachWait what = do
+    (com, lck, ret) <- pathFromTag "System.Miniplex.Source.attachWait" what
+    bracketOnError (socket AF_UNIX Stream 0) sClose $ \s -> do
+        closeOnExec s
+        block . fix $ \retry -> do
+            ld <- do
+                bracket (openFd lck ReadOnly (Just mode644) defaultFileFlags) closeFd $ \lf -> do
+                setLockAll lf LockRead
+            x <- tryJust ioErrors (unblock $ connect s (SockAddrUnix com)) `catch` \e -> do
+                unLock ld
+                throwIO e
+            case x of
+                Right () -> unLock ld
+                Left _ -> do
+                    wf <- flip finally (unLock ld) $ unblock $ do
+                        handleJust eexists (const $ return ()) $ createNamedPipe ret mode644
+                        openFd ret ReadOnly Nothing defaultFileFlags{ nonBlock = True }
+                    unblock (threadWaitRead wf) `finally` closeFd wf
+                    retry
+        shutdown s ShutdownSend
+        return $ Source s
+
 -- | Synchronously reads a message from a source (i.e. it blocks if there is
 -- currently no message available).
 read :: Source -> IO String
@@ -70,4 +99,10 @@
 withSource :: String -> (Source -> IO a) -> IO a
 withSource tag f = block $ do
     so <- attach tag
+    unblock (f so) `finally` detach so
+
+-- | Similar to 'withSource', but calls 'attachWait' instead of 'attach'.
+withSourceWait :: String -> (Source -> IO a) -> IO a
+withSourceWait tag f = block $ do
+    so <- unblock $ attachWait tag
     unblock (f so) `finally` detach so
diff --git a/miniplex.cabal b/miniplex.cabal
--- a/miniplex.cabal
+++ b/miniplex.cabal
@@ -1,5 +1,5 @@
 name: miniplex
-version: 0.2.2
+version: 0.3.3
 cabal-version: >= 1.2
 license: LGPL
 license-file: LICENSE
@@ -22,17 +22,17 @@
   hs-source-dirs: lib
   if flag(small_base)
     build-depends: directory
-  build-depends: base, network, stm, unix, mtl
-  ghc-options: -Wall
+  build-depends: base, network, stm, unix, mtl, colock >=0.2.2
+  ghc-options: -O2 -Wall
 
 executable plox-read
   hs-source-dirs: bin, lib
   main-is: plox-read.hs
   other-modules: System.Miniplex
-  ghc-options: -Wall
+  ghc-options: -O2 -Wall
 
 executable plox-write
   hs-source-dirs: bin, lib
   main-is: plox-write.hs
   other-modules: System.Miniplex
-  ghc-options: -Wall
+  ghc-options: -O2 -Wall