changing NodeConnection to Pool of NodeConnection and taking Cluster …

…Connection out of the Pool

src/Database/Redis/Cluster.hs

@@ -13,7 +13,7 @@ module Database.Redis.Cluster
   , HashSlot
   , Shard(..)
   , connect
-  , disconnect
+  , destroyNodeResources
   , requestPipelined
   , requestMasterNodes
   , nodes
@@ -26,6 +26,7 @@ import Data.Maybe(mapMaybe, fromMaybe)
 import Data.List(nub, sortBy, find)
 import Data.Map(fromListWith, assocs)
 import Data.Function(on)
+import Data.Pool(Pool, createPool, withResource, destroyAllResources)
 import Control.Exception(Exception, SomeException, throwIO, BlockedIndefinitelyOnMVar(..), catches, Handler(..), try)
 import Control.Concurrent.Async(race)
 import Control.Concurrent(threadDelay)
@@ -61,7 +62,7 @@ type IsReadOnly = Bool
 data Connection = Connection (HM.HashMap NodeID NodeConnection) (MVar Pipeline) (MVar ShardMap) CMD.InfoMap IsReadOnly
 
 -- | A connection to a single node in the cluster, similar to 'ProtocolPipelining.Connection'
-data NodeConnection = NodeConnection CC.ConnectionContext (IOR.IORef (Maybe B.ByteString)) NodeID
+data NodeConnection = NodeConnection (Pool CC.ConnectionContext) (IOR.IORef (Maybe B.ByteString)) NodeID
 
 instance Show NodeConnection where
     show (NodeConnection _ _ id1) = "nodeId: " <> show id1
@@ -124,8 +125,8 @@ instance Exception CrossSlotException
 data NoNodeException = NoNodeException  deriving (Show, Typeable)
 instance Exception NoNodeException
 
-connect :: (Host -> CC.PortID -> Maybe Int -> IO CC.ConnectionContext) -> [CMD.CommandInfo] -> MVar ShardMap -> Maybe Int -> Bool -> ([NodeConnection] -> IO ShardMap) -> IO Connection
-connect withAuth commandInfos shardMapVar timeoutOpt isReadOnly refreshShardMap = do
+connect :: (Host -> CC.PortID -> Maybe Int -> IO CC.ConnectionContext) -> [CMD.CommandInfo] -> MVar ShardMap -> Maybe Int -> Bool -> ([NodeConnection] -> IO ShardMap) -> Time.NominalDiffTime -> Int -> IO Connection
+connect withAuth commandInfos shardMapVar timeoutOpt isReadOnly refreshShardMap idleTime maxResources = do
         shardMap <- readMVar shardMapVar
         stateVar <- newMVar $ Pending []
         pipelineVar <- newMVar $ Pipeline stateVar
@@ -157,15 +158,15 @@ connect withAuth commandInfos shardMapVar timeoutOpt isReadOnly refreshShardMap
            ) (mempty, False) info
     connectNode :: Node -> IO (NodeID, NodeConnection)
     connectNode (Node n _ host port) = do
-        ctx <- withAuth host (CC.PortNumber $ toEnum port) timeoutOpt
+        ctx <- createPool (withAuth host (CC.PortNumber $ toEnum port) timeoutOpt) CC.disconnect 1 idleTime maxResources
         ref <- IOR.newIORef Nothing
         return (n, NodeConnection ctx ref n)
     refreshShardMapVar :: ShardMap -> IO ()
     refreshShardMapVar shardMap = hasLocked $ modifyMVar_ shardMapVar (const (pure shardMap))
 
-disconnect :: Connection -> IO ()
-disconnect (Connection nodeConnMap _ _ _ _ ) = mapM_ disconnectNode (HM.elems nodeConnMap) where
-    disconnectNode (NodeConnection nodeCtx _ _) = CC.disconnect nodeCtx
+destroyNodeResources :: Connection -> IO ()
+destroyNodeResources (Connection nodeConnMap _ _ _ _ ) = mapM_ disconnectNode (HM.elems nodeConnMap) where
+    disconnectNode (NodeConnection nodePool _ _) = destroyAllResources nodePool
 
 -- Add a request to the current pipeline for this connection. The pipeline will
 -- be executed implicitly as soon as any result returned from this function is
@@ -441,34 +442,33 @@ allMasterNodes (Connection nodeConns _ _ _ _) (ShardMap shardMap) =
     onlyMasterNodes = (\(Shard master _) -> master) <$> nub (IntMap.elems shardMap)
 
 requestNode :: NodeConnection -> [[B.ByteString]] -> IO [Reply]
-requestNode (NodeConnection ctx lastRecvRef _) requests = do
+requestNode (NodeConnection pool lastRecvRef nid) requests = do
     envTimeout <- round . (\x -> (x :: Time.NominalDiffTime) * 1000000) . realToFrac . fromMaybe (0.5 :: Double) . (>>= readMaybe) <$> lookupEnv "REDIS_REQUEST_NODE_TIMEOUT"
-    eresp <- race requestNodeImpl (threadDelay envTimeout)
+    eresp <- race (withResource pool requestNodeImpl) (threadDelay envTimeout)
     case eresp of
-      Left e -> return e
-      Right _ -> putStrLn "timeout happened" *> throwIO NoNodeException
-
+        Left e -> return e
+        Right _ -> putStrLn ("timeout happened" ++ show nid) *> throwIO NoNodeException
     where
-    requestNodeImpl :: IO [Reply]
-    requestNodeImpl = do
-        mapM_ (sendNode . renderRequest) requests
+    requestNodeImpl :: CC.ConnectionContext -> IO [Reply]
+    requestNodeImpl ctx = do
+        mapM_ (sendNode ctx . renderRequest) requests
         _ <- CC.flush ctx
-        replicateM (length requests) recvNode
-    sendNode :: B.ByteString -> IO ()
-    sendNode = CC.send ctx
-    recvNode :: IO Reply
-    recvNode = do
+        replicateM (length requests) $ recvNode ctx
+    sendNode :: CC.ConnectionContext -> B.ByteString -> IO ()
+    sendNode = CC.send
+    recvNode :: CC.ConnectionContext -> IO Reply
+    recvNode ctx = do
         maybeLastRecv <- IOR.readIORef lastRecvRef
         scanResult <- case maybeLastRecv of
             Just lastRecv -> Scanner.scanWith (CC.recv ctx) reply lastRecv
             Nothing -> Scanner.scanWith (CC.recv ctx) reply B.empty
 
         case scanResult of
-          Scanner.Fail{}       -> CC.errConnClosed
-          Scanner.More{}    -> error "Hedis: parseWith returned Partial"
-          Scanner.Done rest' r -> do
-            IOR.writeIORef lastRecvRef (Just rest')
-            return r
+            Scanner.Fail{}       -> CC.errConnClosed
+            Scanner.More{}    -> error "Hedis: parseWith returned Partial"
+            Scanner.Done rest' r -> do
+                IOR.writeIORef lastRecvRef (Just rest')
+                return r
 
 {-# INLINE nodes #-}
 nodes :: ShardMap -> [Node]

src/Database/Redis/Connection.hs

@@ -2,7 +2,6 @@
 {-# LANGUAGE RecordWildCards #-}
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE ScopedTypeVariables #-}
-
 module Database.Redis.Connection where
 
 import Control.Exception
@@ -29,6 +28,7 @@ import qualified Database.Redis.Cluster as Cluster
 import qualified Database.Redis.ConnectionContext as CC
 import           Control.Concurrent (threadDelay)
 import           Control.Concurrent.Async (race)
+import qualified Database.Redis.Types as T
 --import qualified Database.Redis.Cluster.Pipeline as ClusterPipeline
 
 import Database.Redis.Commands
@@ -50,7 +50,7 @@ import Database.Redis.Commands
 --  'connect' function to create one.
 data Connection
     = NonClusteredConnection (Pool PP.Connection)
-    | ClusteredConnection (MVar ShardMap) (Pool Cluster.Connection)
+    | ClusteredConnection (MVar ShardMap) Cluster.Connection
 
 -- |Information for connnecting to a Redis server.
 --
@@ -169,7 +169,7 @@ checkedConnect connInfo = do
 -- |Destroy all idle resources in the pool.
 disconnect :: Connection -> IO ()
 disconnect (NonClusteredConnection pool) = destroyAllResources pool
-disconnect (ClusteredConnection _ pool) = destroyAllResources pool
+disconnect (ClusteredConnection _ conn) = Cluster.destroyNodeResources conn
 
 -- | Memory bracket around 'connect' and 'disconnect'.
 withConnect :: (Catch.MonadMask m, MonadIO m) => ConnectInfo -> (Connection -> m c) -> m c
@@ -187,8 +187,8 @@ withCheckedConnect connInfo = bracket (checkedConnect connInfo) disconnect
 runRedis :: Connection -> Redis a -> IO a
 runRedis (NonClusteredConnection pool) redis =
   withResource pool $ \conn -> runRedisInternal conn redis
-runRedis (ClusteredConnection _ pool) redis =
-    withResource pool $ \conn -> runRedisClusteredInternal conn (refreshShardMap conn) redis
+runRedis (ClusteredConnection _ conn) redis =
+    runRedisClusteredInternal conn (refreshShardMap conn) redis
 
 newtype ClusterConnectError = ClusterConnectError Reply
     deriving (Eq, Show, Typeable)
@@ -220,9 +220,10 @@ connectCluster bootstrapConnInfo = do
         Right infos -> do
             let
                 isConnectionReadOnly = connectReadOnly bootstrapConnInfo
-                clusterConnection = Cluster.connect withAuth infos shardMapVar timeoutOptUs isConnectionReadOnly refreshShardMapWithNodeConn
-            pool <- createPool (clusterConnect isConnectionReadOnly clusterConnection) Cluster.disconnect 1 (connectMaxIdleTime bootstrapConnInfo) (connectMaxConnections bootstrapConnInfo)
-            return $ ClusteredConnection shardMapVar pool
+                clusterConnection = Cluster.connect withAuth infos shardMapVar timeoutOptUs isConnectionReadOnly refreshShardMapWithNodeConn (connectMaxIdleTime bootstrapConnInfo) (connectMaxConnections bootstrapConnInfo)
+            -- pool <- createPool (clusterConnect isConnectionReadOnly clusterConnection) Cluster.disconnect 3 (connectMaxIdleTime bootstrapConnInfo) (connectMaxConnections bootstrapConnInfo)
+            connection <- clusterConnect isConnectionReadOnly clusterConnection
+            return $ ClusteredConnection shardMapVar connection
     where
       withAuth host port timeout = do
         conn <- PP.connect host port timeout
@@ -245,13 +246,24 @@ connectCluster bootstrapConnInfo = do
       clusterConnect :: Bool -> IO Cluster.Connection -> IO Cluster.Connection
       clusterConnect readOnlyConnection connection = do
           clusterConn@(Cluster.Connection nodeMap _ _ _ _) <- connection
-          nodesConns <-  sequence $ ( PP.fromCtx . (\(Cluster.NodeConnection ctx _ _) -> ctx ) . snd) <$> (HM.toList nodeMap)
+          nodesConns <-  sequence $ (ctxToConn . snd) <$> (HM.toList nodeMap)
           when readOnlyConnection $
-                  mapM_ (\conn -> do
-                          PP.beginReceiving conn
-                          runRedisInternal conn readOnly
+                  mapM_ (\maybeConn -> case maybeConn of
+                            Just conn -> do
+                                PP.beginReceiving conn
+                                runRedisInternal conn readOnly
+                            Nothing -> return $ Right (T.Status "Connection does not exist")
                       ) nodesConns
           return clusterConn
+          where
+          ctxToConn :: Cluster.NodeConnection -> IO (Maybe PP.Connection)
+          ctxToConn (Cluster.NodeConnection pool _ nid) = do
+            maybeConn <- try $ withResource pool PP.fromCtx
+            case maybeConn of
+                Right ppConn -> return $ Just ppConn
+                Left (_ :: SomeException) -> do
+                    putStrLn ("SomeException Occured in NodeID " ++ show nid)
+                    return Nothing
 
 shardMapFromClusterSlotsResponse :: ClusterSlotsResponse -> IO ShardMap
 shardMapFromClusterSlotsResponse ClusterSlotsResponse{..} = ShardMap <$> foldr mkShardMap (pure IntMap.empty)  clusterSlotsResponseEntries where
@@ -276,19 +288,20 @@ refreshShardMap (Cluster.Connection nodeConns _ _ _ _) =
 
 refreshShardMapWithNodeConn :: [Cluster.NodeConnection] -> IO ShardMap
 refreshShardMapWithNodeConn [] = throwIO $ ClusterConnectError (Error "Couldn't refresh shardMap due to connection error")
-refreshShardMapWithNodeConn ((Cluster.NodeConnection ctx _ _) : xs) = do
-    pipelineConn <- PP.fromCtx ctx
-    raceResult <- race (threadDelay (10^(3 :: Int))) (try $ refreshShardMapWithConn pipelineConn True) -- racing with delay of 1 ms 
-    case raceResult of
-        Left () -> do
-            print $ "TimeoutForConnection " <> show ctx 
-            refreshShardMapWithNodeConn xs
-        Right eiShardMapResp -> 
-            case eiShardMapResp of
-                Right shardMap -> pure shardMap 
-                Left (err :: SomeException) -> do
-                    print $ "ShardMapRefreshError-" <> show err 
-                    refreshShardMapWithNodeConn xs
+refreshShardMapWithNodeConn ((Cluster.NodeConnection pool _ _) : xs) =
+    withResource pool $ \ctx -> do
+        pipelineConn <- PP.fromCtx ctx
+        raceResult <- race (threadDelay (10^(3 :: Int))) (try $ refreshShardMapWithConn pipelineConn True) -- racing with delay of 1 ms 
+        case raceResult of
+            Left () -> do
+                print $ "TimeoutForConnection " <> show ctx 
+                refreshShardMapWithNodeConn xs
+            Right eiShardMapResp -> 
+                case eiShardMapResp of
+                    Right shardMap -> pure shardMap 
+                    Left (err :: SomeException) -> do
+                        print $ "ShardMapRefreshError-" <> show err 
+                        refreshShardMapWithNodeConn xs
 
 refreshShardMapWithConn :: PP.Connection -> Bool -> IO ShardMap
 refreshShardMapWithConn pipelineConn _ = do