smtpthread/SMTPConnectionPool.cc

// DOCUMENT:  SMTPConnectionPool-Class
// VERSION:   $Revision: 1.5 $
// DATE:      $Date: 2007-10-26 23:09:19 $
// AUTHOR:    M.Beranek
// COPYRIGHT: M.Beranek

#include <iostream>
#include <string>
#include <unistd.h>
#include <vector>
#include <math.h>
#include "SMTPConnectionPool.h"

using namespace std;

namespace SmtpThread
{
// --------------------------------------------------------------------------
    /**
     * Default-Constructor:
     */
    SMTPConnectionPool::SMTPConnectionPool()
    {
        init();
    }
    
    
// --------------------------------------------------------------------------
    /**
     * Destructor. Clean up all connections.
     */
    SMTPConnectionPool::~SMTPConnectionPool()
    {
        if( _optimzePoolSize )
        {
            cout << "last pool-size: " << pool.size() << endl;
            cout << "avg. number of parallel connections: " << _avgNumOfTries << " +-" << _rms << endl;;
        }
        close();
    }
    
    
// --------------------------------------------------------------------------
    /**
     * Initializes an SMTP-Connection-Pool with a default
     * pool-size of 30 connections.
     */
    void SMTPConnectionPool::init()
    {
        
        _minPoolSize = 1;
        _maxPoolSize = 100;
        _optimzePoolSize = true;
        
        _statsTotalMails    = 0;
        _statsTotalErrors   = 0;
        _statsMailsAfterOpt = 0;
        _avgNumOfTries      = 0;
        _rms                = 0;
        
        _myHostname = "localhost";
        _smtpHost   = "127.0.0.1";
        _smtpPort   = 25;
        
        setPoolSize( 1 );
        
    }
    
    
// --------------------------------------------------------------------------
    /**
     * Closes all opened connections. Mails, wich are submitted
     * to theses connections, are sent before the connection is
     * actually closed.
     */
    void SMTPConnectionPool::close()
    {
        // for all pooled connections:
        for( unsigned long idx = 0; idx < pool.size(); idx++ )
        {
            // stop threading:
            pool[ idx ]->Stop();
        }
    }
    
    
// --------------------------------------------------------------------------
    /**
     * Accepts a message and delegates it to one of the opened
     * connections in the connection-pool.
     * @param string from Sender-Email
     * @param string to Receiver-Email
     * @param string msg Email-Body, containing headers like subject, from, to etc.
     */
    int SMTPConnectionPool::sendMail( string& from, string& to, string& msg )
    {
        // success or not:
        bool found = false;
        
        // the connection, which is currentlx checked:
        unsigned long conNumber = 0;
        
        // conter for the number pool-runs:
        int numCheck = 0;
        
        // maximum number of pool-runs to find a ready connection:
        int maxTries = 100;
        
        // the currently checked connection:
        ThreadedSMTPConnection *smtp;
        
        // count messages:
        _statsTotalMails ++;
        _statsMailsAfterOpt ++;
        
        // try finding a free connection:
        while( ! found )
        {
            //cout << "trying connection " << conNumber << endl;
            
            // get connection from pool, increment connetion-counter:
            smtp = pool[ conNumber++ ];
            
            // if connection is ready:
            if( smtp->isReady() )
            {
                
                // set while-condition:
                found = true;
                
                // delegate message to a connection:
                smtp->sendMail( from, to, msg );
                
                // cout << "connection: " << conNumber -1 << endl;
                
                // If we should optimze the pool-size:
                if( _optimzePoolSize )
                {
                    
                    // Calculate RMS (Root-Mean-Square):
                    double d = _avgNumOfTries - 1.0 * ( numCheck +1 ) * ( conNumber );
                    double dSquare = d * d;
                    double curremtRMS = sqrt( dSquare );
                    
                    // Average number of tries / Average RMS:
                    if( (_statsTotalMails - _statsTotalErrors) > 0 )
                    {
                        // _avgNumOfTries, weight: number of all successfull mails till now (i.e without current mail).
                        // current number of tries: (i.e. pool-runs * conNumber, weight: 1.
                        // pool-runs: numCheck +1 (numCheck starts with 0)
                        // normalized by all successfull sent mails.
                        _avgNumOfTries = (
                                ( 1.0 * (
                                ( _statsTotalMails - _statsTotalErrors -1 ) * _avgNumOfTries
                                +
                                ( numCheck +1 ) * ( conNumber ) )
                                )
                                /
                                (
                                1.0 * ( _statsTotalMails - _statsTotalErrors )
                                )
                                );
                        
                        // _rms, weight: number of all successfull mails till now (i.e without current mail).
                        // currentRMS, weight: 1.
                        _rms = (
                                ( 1.0 * (
                                ( _statsTotalMails - _statsTotalErrors -1 ) * _rms
                                +
                                curremtRMS )
                                )
                                /
                                (
                                1.0 * ( _statsTotalMails - _statsTotalErrors )
                                )
                                );
                        
                        // cout << "avg. RMS: " << _rms << endl;
                    }
                    
                    
                    // only if we have more mails sent than our current pool-size:
                    //if( _statsTotalMails > pool.size() )
                    if( _statsMailsAfterOpt > pool.size() )
                    {
                        // for now: opt. poolsize = avg-num-of-tries + 2 * RMS.
                        unsigned long optPoolSize = (unsigned long) ceil( _avgNumOfTries + 2.0 * _rms );
                        
                        // if pool-size differs:
                        if( ( pool.size() < optPoolSize ) || ( pool.size() > optPoolSize +1 ) )
                        {
                            // if new pool-size within our limits:
                            if( ( optPoolSize < _maxPoolSize ) && ( optPoolSize > _minPoolSize ) )
                            {
                                // set the poolsize:
                                setPoolSize( optPoolSize );
                                
                                // reset counter:
                                _statsMailsAfterOpt = 0;
                            }
                        }
                    }
                }
                
                // return success:
                return 0;
            }
            
            // after the last connection is checked
            // (i.e. after a complete pool-run):
            if( conNumber >= pool.size() )
            {;
             
             // start again:
             conNumber = 0;
             
             // increment the number of pool-runs:
             numCheck++;
             
             // wait a few micro-seconds:
             // 1s    = 1000000
             // 0.5s  =  500000
             // 0.25s =  250000
             usleep( 250000 );
             
             // if maximum of pool-runs reached:
             if( numCheck > maxTries )
             {
                 
                 // count errors:
                 _statsTotalErrors ++;
                 
                 // fail:
                 return -1;
                 
             }
            }
        }
        // just to avoid compiler-warning:
        return -1;
    }
    
    
// --------------------------------------------------------------------------
    /**
     * Sets the pool-size.
     *
     * If pool-size is enlarged, the missing connection are set up automatically.
     * If pool-size is decreased, the opend connections will be closed and
     * destroyed. If mail is unsent in a connection, these mails are sent before
     * closing.
     *
     * @param unsigned long poolSize number of connections in pool
     */
    void SMTPConnectionPool::setPoolSize( unsigned long poolSize )
    {
        // Debug-Ouptut:
        cout << "adjusting pool-size to " << poolSize << endl;
        
        // current pool-size:
        unsigned long pSize = pool.size();
        
        // enlarge pool-size:
        if( poolSize > pSize )
        {
            
            // create new connections until pool-size is reached:
            while( pool.size() < poolSize )
            {
                // initialize connections:
                ThreadedSMTPConnection* con = new ThreadedSMTPConnection();
                
                // set mailhost:
                con->setSMTPHost( _smtpHost );
                con->setSMTPPort( _smtpPort );
                con->setMyHostname( _myHostname );
                
                // init connection:
                // con->openConnection();
                
                // start threading:
                con->Start( (void*) 0 );
                
                // push into vector:
                pool.push_back( con );
            }
        }
        
        // decrease pool-size:
        if( poolSize < pSize )
        {
            
            // destroy connections until pool-size is reached:
            while( pool.size() > poolSize )
            {
                
                // The destructor of the "SMTPConnection"-object
                // gets called automatically by "pop_back", so there
                // is no need to call "delete" on the object.
                
                // remove last connection from vector:
                pool.pop_back();
                
            }
        }
    }
    
    
// --------------------------------------------------------------------------
    /**
     * Returns the current size of the connection-pool.
     *
     * @return unsigned long pool-size
     */
    unsigned long SMTPConnectionPool::getPoolSize()
    {
        // return pool-size:
        return pool.size();
    }
    
    
// --------------------------------------------------------------------------
    void SMTPConnectionPool::setSMTPHost( string host )
    {
        _smtpHost = host;
        
        ThreadedSMTPConnection* smtp;
        for( unsigned long idx = 0; idx < pool.size(); idx++  )
        {
            smtp = pool[ idx ];
            smtp->setSMTPHost( host );
        }
    }
    
    
// --------------------------------------------------------------------------
    void SMTPConnectionPool::setSMTPPort( int port )
    {
        _smtpPort = port;
        
        ThreadedSMTPConnection* smtp;
        for( unsigned long idx = 0; idx < pool.size(); idx++  )
        {
            smtp = pool[ idx ];
            smtp->setSMTPPort( port );
        }
    }
    
    
// --------------------------------------------------------------------------
    void SMTPConnectionPool::setMyHostname( string myname )
    {
        _myHostname = myname;
        
        ThreadedSMTPConnection* smtp;
        for( unsigned long idx = 0; idx < pool.size(); idx++  )
        {
            smtp = pool[ idx ];
            smtp->setMyHostname( myname );
        }
    }
    
    
// --------------------------------------------------------------------------
    void SMTPConnectionPool::setOptimizePool( bool opt )
    {
        _optimzePoolSize = opt;
    }
    
    
// --------------------------------------------------------------------------
    void SMTPConnectionPool::setMinPoolSize( long minimum )
    {
        _minPoolSize = minimum;
        
        // Adjust Poolsize:
        if( pool.size() < _minPoolSize )
        {
            setPoolSize( _minPoolSize );
        }
    }
    
    
// --------------------------------------------------------------------------
    void SMTPConnectionPool::setMaxPoolSize( long maximum )
    {
        _maxPoolSize = maximum;
        
        // Adjust Poolsize:
        if( pool.size() > _maxPoolSize )
        {
            setPoolSize( _maxPoolSize );
        }
    }
    
}

---