Re: [libpqxx-general] multi-thread application with pqxx

Fei Liu Tue, 17 Jul 2007 12:02:51 -0700

Fei Liu wrote:

Jeroen T. Vermeulen wrote:

On Thu, July 12, 2007 23:03, Fei Liu wrote:


Hello Liu,

My implementation is now using a 120 pool size (server is configured to
allow 128 concurrently), my bottle neck now is the database. My
connection pool is quickly exhausted in my test. What's a good strategy
to improve the performance? I am thinking of queueing the database
requests and only execute them when a certain number (say 50 sql
statements) is reached?

I think 120 is probably still too many.  One of the points of having a

connection pool is that you can create fewer connections and reuse them.Many web applications get by on <20, as far as I know. Don't create this

many unless your testing shows that (1) you need it and (2) it really
helps.

What you can do is make sure your pooled connections are released
regularly, so hopefully you don't need so many.  You could allocate them
on a per-transaction basis: lock pool, grab connection, unlock pool, open
transaction, do work, commit, destroy transaction, lock pool, release
connection, unlock pool.  "Grabbing" or "releasing" a connection could be
as simple as taking it out of a list or setting an "I own it" flag
somewhere.

In my case the postgres server simply stops responding after a while and
allocated connection is not released from client...But CPU load is never
really very high...Maybe there is something else wrong.

That's beginning to sound more like a postgres issue; people at
[EMAIL PROTECTED] may know more about it.


Jeroen

Thanks for the comments, I'll do some digging and hopefully havesomething to report back.

Fei

It appears my multi-thread application (100 connections every 5 seconds)is stalled when working with postgresql database server. I have limitednumber of connections in my connection pool to postgresql to 20. At thebegining, connection is allocated and released from connection pool aspostgres serves data request. The pool can recover from exhaustion. Butvery quickly (after about 400 client requests), it seems postgres serverstops serving and connection to postgres server is not released any moreresulting a resource exhausting for clients.

I am using a slightly model to protect thread generation. I have anotherthread pool that limits number of concurrent threads to about 30. I haveenclosed my wrapper interface to libpqxx. Maybe I did something wrongwith my wrapper? My client code uses strictly the transactor interface.What's the implementation strategy of pqxx::transactor<> ?

Fei

unix 3 [ ] STREAM CONNECTED 169365531976/postgres: pos /tmp/.s.PGSQL.5583unix 3 [ ] STREAM CONNECTED 169365431740/ns_ge_classifunix 3 [ ] STREAM CONNECTED 169365331975/postgres: pos /tmp/.s.PGSQL.5583unix 3 [ ] STREAM CONNECTED 169365231740/ns_ge_classifunix 3 [ ] STREAM CONNECTED 169365131974/postgres: pos /tmp/.s.PGSQL.5583unix 3 [ ] STREAM CONNECTED 169365031740/ns_ge_classifunix 3 [ ] STREAM CONNECTED 169364931973/postgres: pos /tmp/.s.PGSQL.5583unix 3 [ ] STREAM CONNECTED 169364831740/ns_ge_classifunix 3 [ ] STREAM CONNECTED 169364731972/postgres: pos /tmp/.s.PGSQL.5583unix 3 [ ] STREAM CONNECTED 169364631740/ns_ge_classifunix 3 [ ] STREAM CONNECTED 169364531971/postgres: pos /tmp/.s.PGSQL.5583unix 3 [ ] STREAM CONNECTED 169364431740/ns_ge_classifunix 3 [ ] STREAM CONNECTED 169364131969/postgres: pos /tmp/.s.PGSQL.5583unix 3 [ ] STREAM CONNECTED 169364031740/ns_ge_classifunix 3 [ ] STREAM CONNECTED 169363931968/postgres: pos /tmp/.s.PGSQL.5583unix 3 [ ] STREAM CONNECTED 169363831740/ns_ge_classifunix 3 [ ] STREAM CONNECTED 169363731967/postgres: pos /tmp/.s.PGSQL.5583unix 3 [ ] STREAM CONNECTED 169363631740/ns_ge_classifunix 3 [ ] STREAM CONNECTED 169358531941/postgres: pos /tmp/.s.PGSQL.5583unix 3 [ ] STREAM CONNECTED 169358431740/ns_ge_classifunix 3 [ ] STREAM CONNECTED 169358331940/postgres: pos /tmp/.s.PGSQL.5583unix 3 [ ] STREAM CONNECTED 169358231740/ns_ge_classifunix 3 [ ] STREAM CONNECTED 169358131939/postgres: pos /tmp/.s.PGSQL.5583unix 3 [ ] STREAM CONNECTED 169358031740/ns_ge_classifunix 3 [ ] STREAM CONNECTED 169357931938/postgres: pos /tmp/.s.PGSQL.5583unix 3 [ ] STREAM CONNECTED 169357831740/ns_ge_classifunix 3 [ ] STREAM CONNECTED 169357731937/postgres: pos /tmp/.s.PGSQL.5583unix 3 [ ] STREAM CONNECTED 169357631740/ns_ge_classifunix 3 [ ] STREAM CONNECTED 169357531936/postgres: pos /tmp/.s.PGSQL.5583unix 3 [ ] STREAM CONNECTED 169357431740/ns_ge_classifunix 3 [ ] STREAM CONNECTED 169357331935/postgres: pos /tmp/.s.PGSQL.5583unix 3 [ ] STREAM CONNECTED 169357231740/ns_ge_classifunix 3 [ ] STREAM CONNECTED 169357131934/postgres: pos /tmp/.s.PGSQL.5583unix 3 [ ] STREAM CONNECTED 169357031740/ns_ge_classifunix 3 [ ] STREAM CONNECTED 169342731851/postgres: pos /tmp/.s.PGSQL.5583unix 3 [ ] STREAM CONNECTED 169342631740/ns_ge_classifunix 3 [ ] STREAM CONNECTED 169342531777/postgres: pos /tmp/.s.PGSQL.5583unix 3 [ ] STREAM CONNECTED 169342431740/ns_ge_classifunix 3 [ ] STREAM CONNECTED 169341931764/postgres: pos /tmp/.s.PGSQL.5583unix 3 [ ] STREAM CONNECTED 169341831740/ns_ge_classif


After about 10 minutes, they are still there, they are never released...

#ifndef NS_GE_PGDB
#define NS_GE_PGDB

// $Id$
/**
 * @file
 * @brief provides APIs to facilitate policy and database interaction
 */

// $Log$

#include <cassert>
#include <iostream>
#include <sstream>
#include <string>
#include <vector>

#include <pqxx/pqxx>
#include <pqxx/connection>
#include <pqxx/transaction>
#include <pqxx/transactor>

#include "resource_pool.h"

namespace utils{
    namespace database{
        std::string sql_escape(const std::string & str);
        std::string sql_escape(const std::vector<std::string> & values); 

        struct uuid{
            static std::string get(){
                static unsigned long long id = 0;
                std::ostringstream os;
                os << (id++);
                return os.str();
            }
        };

        class pgdb_insert : public pqxx::transactor<> {
            std::string table;
            std::vector<std::string> values;

        public:
            pgdb_insert(const std::string & table, const std::vector<std::string> & values) : 
                pqxx::transactor<>("pgdb_insert"+uuid::get()), table(table), values(values) {}

            void operator()(argument_type &T) {
                T.exec("INSERT INTO " + table + " VALUES " + sql_escape(values));
            }
        };

        class pgdb_delete : public pqxx::transactor<> {
            std::string table;
            std::string where;
        public:
            pgdb_delete(const std::string & table, const std::string & where) :
                pqxx::transactor<>("pgdb_delete"+uuid::get()), table(table), where(where) {}

            void operator()(argument_type & T) {
                T.exec("DELETE FROM " + table + " " + where);
            }
        };

        // execute arbitrary single sql statement
        class pgdb_statement : public pqxx::transactor<> {
            std::string statement;
        public:
            pgdb_statement(const std::string & sql_statement) : 
                pqxx::transactor<>("pgdb_statement"+uuid::get()), statement(sql_statement) {}
            void operator()(argument_type & T) {
                T.exec(statement);
            }
        };

        // execute a sequence of statements
        class pgdb_statements : public pqxx::transactor<pqxx::transaction<pqxx::serializable> > {
            std::vector<std::string> statements;
        public:
            pgdb_statements(const std::vector<std::string> & sql_statements) : 
                pqxx::transactor<pqxx::transaction<pqxx::serializable> >("pgdb_statements"+uuid::get()), 
                statements(sql_statements) {}
            void operator()(argument_type & T) {
                std::vector<std::string>::const_iterator it = 
                    statements.begin();
                for(; it != statements.end(); ++it)
                    T.exec(*it);
            }
        };

        class pqdb{
        private:
            utils::resource_pool<pqxx::connection> pool;
        public:
            pqdb(): pool("port=5583 dbname=netilla user=postgres password=") { }
            template <typename pgdb_op>
                void perform(const pgdb_op & op);
            // We want to avoid unnecessary data copying and provide thread safety. 
            // The caller
            // is responsible to pass in pre-allocated memory to store result.
            //
            // The single argument version relies on compiler optimization 
            // to optimize away the return copying.
            void query(const std::string & sql, pqxx::result & r);
            pqxx::result query(const std::string & sql);

            //void exec(const std::string & sql);

            unsigned int size(){    
                return pool.size();
            }
        };

        template <typename pgdb_op>
        void pqdb::perform(const pgdb_op & op){
            unsigned int id = 0;
            try{
                pqxx::connection & c = pool.alloc(id);
                c.perform(op);
                id = pool.release(id);
            }
            catch(pqxx::integrity_constraint_violation & e){
                // ignore duplicate key violation
            }
            catch (const pqxx::sql_error &e)
            {
                // If we're interested in the text of a failed query, we can write separate
                // exception handling code for this type of exception
                std::cerr << "SQL error: " << e.what() << std::endl
                     << "Query was: '" << e.query() << "'" << std::endl;
            }
            catch (const std::exception &e)
            {
                // All exceptions thrown by libpqxx are derived from std::exception
                std::cerr << "Exception: " << e.what() << std::endl;
            }
            catch(...){
                // This is really unexpected (see above)
                std::cerr << "Unhandled exception" << std::endl;
                if(id)
                    pool.release(id);
            }
        }
    }
}

#endif

// $Id$
/**
 * @file
 * @brief implementation of pg db APIs
 */
// $Log$

#include "ns_ge_pqdb.h"
#include <iostream>
#include <algorithm>
#include <vector>
#include <string>

namespace utils{
    namespace database{
        std::string sql_escape(const std::string & str){
            std::string r = "'";
            r.reserve(str.size()*2);
            std::string::const_iterator it = str.begin();
            for(; it != str.end(); ++it){
                if(*it == '\'') r += "\\\'";
                else r.push_back(*it);
            }
            r.push_back('\'');
            return r;
        }
        std::string sql_escape(const std::vector<std::string> & values){
            std::vector<std::string>::const_iterator it = 
                values.begin();
            std::string r = "(";
            for(; it != values.end(); ++it){
                //std::cout << *it << std::endl;
                if((it+1) != values.end())
                    r += sql_escape(*it) + ", ";
                else
                    r += sql_escape(*it);
            }
            r += ");";
            //std::cout << r << std::endl;
            return r;
        }
        void pqdb::query(const std::string & sql, pqxx::result & r){
            unsigned int id;
            pqxx::connection & c = pool.alloc(id);
            pqxx::work T(c, "query");
            r = T.exec(sql);
            pool.release(id);
        }
        pqxx::result pqdb::query(const std::string & sql){
            unsigned int id;
            pqxx::connection & c = pool.alloc(id);
            pqxx::work T(c, "query");
            return T.exec(sql);
            pool.release(id);
        }

//        void pqdb::exec(const std::string & sql){
//            unsigned int id;
//            pqxx::connection & c = pool.alloc(id);
//            pqxx::work T(c, "insert"+uuid::get());
//            T.exec(sql);
//            pool.release(id);
//        }
    }
}

_______________________________________________
Libpqxx-general mailing list
[email protected]
http://gborg.postgresql.org/mailman/listinfo/libpqxx-general

Re: [libpqxx-general] multi-thread application with pqxx

Reply via email to