CODEWITHSUNDEEP
multithreadinghaskellzeromq
iamnat
iamnat

Reputation: 4166

Unclosed ZeroMQ sockets in threads

If a process thread opens an inproc ZMQ socket, and then dies because of some unhandled exception, what happens if the socket is not closed? How bad is this sort of a practice?

To be more specific, I've implemented a very simple message broker very similar to http://zguide.zeromq.org/page:all#Multithreading-with-MQ in Haskell.

The worker thread opens a new socket, and waits in an infinite loop for processing messages. The socket is not closed anywhere in the worker thread.

Now, if there is an unhandled exception in the worker thread, and the thread dies, how bad is to just restart the thread without caring?

I'm pasting the worker code from the Haskell example:

worker :: ZMQ z ()
worker = do
    receiver <- socket Rep
    connect receiver "inproc://workers"
    forever $ do
        receive receiver >>= liftIO . printf "Received request:%s\n" . unpack    
        -- Suppose there is some exception here
        liftIO $ threadDelay (1 * 1000 * 1000)  
        send receiver [] "World"

Upvotes: 0

Views: 216

Answers (1)

iamnat
iamnat

Reputation: 4166

So it seems that if you don't close the inproc socket, the restarted thread can't accept messages very well. I'm not sure I understand this behaviour, but I can confirm that this modified example from the ZMQ haskell guide works:

import System.ZMQ3.Monadic
import Prelude hiding (catch)
import Control.Monad.CatchIO

worker :: ZMQ z ()
worker = do
            liftIO $ putStrLn "Starting the worker thread..."
            receiver <- socket Rep 
            connect receiver "inproc://workers"
            catch
                (forever $ do
                    liftIO $ putStrLn "Waiting for an inproc message" 
                    request <- receiveMulti receiver -- request :: ByteString.Char8
                    liftIO $ putStrLn "I'm doing something that may throw an error"
                    -- error "IO Error has happened"
                )   
                (\(e :: IOError) -> do
                    liftIO $ putStrLn $ "Caught error: " ++ (show e)
                    close receiver -- Commenting this out will result in the restarted worker thread being unable to accept new messages                                                                                               
                )

Upvotes: 1

Related Questions