Concurrent DB connection pool in Haskell

Question

I am a Java programmer who learns Haskell.
I work on a small web-app that uses Happstack and talks to a database via HDBC.

I've written select and exec functions and I use them like this:

module Main where

import Control.Exception (throw)

import Database.HDBC
import Database.HDBC.Sqlite3 -- just for this example, I use MySQL in production

main = do
    exec "CREATE TABLE IF NOT EXISTS users (name VARCHAR(80) NOT NULL)" []

    exec "INSERT INTO users VALUES ('John')" []
    exec "INSERT INTO users VALUES ('Rick')" []

    rows <- select "SELECT name FROM users" []

    let toS x = (fromSql x)::String
    let names = map (toS . head) rows

    print names

Very simple as you see. There is query, params and result.
Connection creation and commit/rollback stuff is hidden inside select and exec.
This is good, I don't want to care about it in my "logic" code.

exec :: String -> [SqlValue] -> IO Integer
exec query params = withDb $ \c -> run c query params

select :: String -> [SqlValue] -> IO [[SqlValue]]
select query params = withDb $ \c -> quickQuery' c query params

withDb :: (Connection -> IO a) -> IO a
withDb f = do
    conn <- handleSqlError $ connectSqlite3 "users.db"
    catchSql
        (do r <- f conn
            commit conn
            disconnect conn
            return r)
        (\e@(SqlError _ _ m) -> do
            rollback conn
            disconnect conn
            throw e)

Bad points:

• a new connection is always created for every call - this kills performance on heavy load
• DB url "users.db" is hardcoded - I can't reuse these functions across other projects w/o editing

QUESTION 1: how to introduce a pool of connections with some defined (min, max) number of concurrent connections, so the connections will be reused between select/exec calls?

QUESTION 2: How to make "users.db" string configurable? (How to move it to client code?)

It should be a transparent feature: user code should not require explicit connection handling/release.

Caia

I don't have a full answer for you, but your problem is that you abstracted away the connection incorrectly. You probably want to put it in a Reader-like structure, so that it can be passed to each query. – Flavory

Hmm, SQL operations are all stuck in the IO monad, so maybe ReaderT IO? Sounds reasonable. – Audubon

Answer 1

QUESTION 2: I've never used HDBC, but I'd probably write something like this.

trySql :: Connection -> (Connection -> IO a) -> IO a
trySql conn f = handleSql catcher $ do
    r <- f conn
    commit conn
    return r
  where catcher e = rollback conn >> throw e

Open the Connection somewhere outside of the function, and don't disconnect it within the function.

QUESTION 1: Hmm, a connection pool doesn't seem that hard to implement...

import Control.Concurrent
import Control.Exception

data Pool a =
    Pool { poolMin :: Int, poolMax :: Int, poolUsed :: Int, poolFree :: [a] }

newConnPool low high newConn delConn = do
    cs <- handleSqlError . sequence . replicate low newConn
    mPool <- newMVar $ Pool low high 0 cs
    return (mPool, newConn, delConn)

delConnPool (mPool, newConn, delConn) = do
    pool <- takeMVar mPool
    if length (poolFree pool) /= poolUsed pool
      then putMVar mPool pool >> fail "pool in use"
      else mapM_ delConn $ poolFree pool

takeConn (mPool, newConn, delConn) = modifyMVar mPool $ \pool ->
    case poolFree pool of
        conn:cs ->
            return (pool { poolUsed = poolUsed pool + 1, poolFree = cs }, conn)
        _ | poolUsed pool < poolMax pool -> do
            conn <- handleSqlError newConn
            return (pool { poolUsed = poolUsed pool + 1 }, conn)
        _ -> fail "pool is exhausted"

putConn (mPool, newConn, delConn) conn = modifyMVar_ mPool $ \pool ->
    let used = poolUsed pool in
    if used > poolMin conn
      then handleSqlError (delConn conn) >> return (pool { poolUsed = used - 1 })
      else return $ pool { poolUsed = used - 1, poolFree = conn : poolFree pool }

withConn connPool = bracket (takeConn connPool) (putConn conPool)

You probably shouldn't take this verbatim as I haven't even compile-tested it (and fail there is pretty unfriendly), but the idea is to do something like

connPool <- newConnPool 0 50 (connectSqlite3 "user.db") disconnect

and pass connPool around as needed.

Audubon

Cool! Is it thread safe? Is it ok to create a single "connPool" and use it in all Happstack handlers? – Caia

It should be thread-safe, all the work is done within modifyMVar (which is takeMVar + putMVar), which effectively sequences all the take/put operations. But you really should audit this code yourself, to see if it suits your needs. – Audubon

Before using the pool test how your database driver copes with disconnects. I tried to use this Pool implementation with hdbc-odbc driver against MS SQL Server. It works fine. But then i stop sql server, try the application, which gives me the error obviously, then start sql server back, and try application again. It still gives an error. Unfortunately disconnects on the network happen. So make sure you deal with faulty connections and spawn new ones. – Amadeus

Answer 2

The resource-pool package provides a high-performance resource pool which can be used for database connection pooling. For example:

import Data.Pool (createPool, withResource)

main = do
    pool <- createPool newConn delConn 1 10 5
    withResource pool $ \conn -> doSomething conn

Creates a database connection pool with 1 sub-pool and up to 5 connections. Each connection is allowed to be idle for 10 seconds before being destroyed.

Trilbi

I've just used (and I am loving) Data.Conduit.Pool (pool-conduit package). Its a wrapper around Data.Pool (used by yesod and others) hackage.haskell.org/package/pool-conduit-0.1.1 – Arabelle

Answer 3

I modified the code above, now it's able to compile at least.

module ConnPool ( newConnPool, withConn, delConnPool ) where

import Control.Concurrent
import Control.Exception
import Control.Monad (replicateM)
import Database.HDBC

data Pool a =
    Pool { poolMin :: Int, poolMax :: Int, poolUsed :: Int, poolFree :: [a] }

newConnPool :: Int -> Int -> IO a -> (a -> IO ()) -> IO (MVar (Pool a), IO a, (a -> IO ()))
newConnPool low high newConn delConn = do
--    cs <- handleSqlError . sequence . replicate low newConn
    cs <- replicateM low newConn 
    mPool <- newMVar $ Pool low high 0 cs 
    return (mPool, newConn, delConn)

delConnPool (mPool, newConn, delConn) = do
    pool <- takeMVar mPool
    if length (poolFree pool) /= poolUsed pool
      then putMVar mPool pool >> fail "pool in use"
      else mapM_ delConn $ poolFree pool

takeConn (mPool, newConn, delConn) = modifyMVar mPool $ \pool ->
    case poolFree pool of
        conn:cs ->
            return (pool { poolUsed = poolUsed pool + 1, poolFree = cs }, conn)
        _ | poolUsed pool < poolMax pool -> do
            conn <- handleSqlError newConn
            return (pool { poolUsed = poolUsed pool + 1 }, conn)
        _ -> fail "pool is exhausted"

putConn :: (MVar (Pool a), IO a, (a -> IO b)) -> a -> IO ()
putConn (mPool, newConn, delConn) conn = modifyMVar_ mPool $ \pool ->
    let used = poolUsed pool in
    if used > poolMin pool
    then handleSqlError (delConn conn) >> return (pool { poolUsed = used - 1 })
    else return $ pool { poolUsed = used - 1, poolFree = conn : (poolFree pool) }

withConn connPool = bracket (takeConn connPool) (putConn connPool)

Muzhik