metaproxy_1::ThreadPoolSocketObserver Class Reference

#include <thread_pool_observer.hpp>

Inheritance diagram for metaproxy_1::ThreadPoolSocketObserver:

Collaboration diagram for metaproxy_1::ThreadPoolSocketObserver:

Classes
class	Rep
class	Worker

Public Member Functions
	ThreadPoolSocketObserver (yazpp_1::ISocketObservable *obs, unsigned min_threads, unsigned max_threads, unsigned stack_size)
virtual	~ThreadPoolSocketObserver ()
void	put (IThreadPoolMsg *m)
void	cleanup (IThreadPoolMsg *m, void *info)
IThreadPoolMsg *	get ()
void	run (void *p)
void	get_thread_info (int &tbusy, int &total)

Private Member Functions
void	add_worker (void)
void	socketNotify (int event)

Private Attributes
boost::scoped_ptr< Rep >	m_p

Detailed Description

Definition at line 36 of file thread_pool_observer.hpp.

Constructor & Destructor Documentation

ThreadPoolSocketObserver()

ThreadPoolSocketObserver::ThreadPoolSocketObserver	(	yazpp_1::ISocketObservable *	obs,
		unsigned	min_threads,
		unsigned	max_threads,
		unsigned	stack_size
	)

Definition at line 102 of file thread_pool_observer.cpp.

    : m_p(new Rep(obs))
{
    obs->addObserver(m_p->m_pipe.read_fd(), this);
    obs->maskObserver(this, SOCKET_OBSERVE_READ);
 
    m_p->m_stop_flag = false;
    m_p->m_no_threads = 0;
    m_p->m_min_threads = min_threads;
    m_p->m_max_threads = max_threads;
    m_p->m_waiting_threads = 0;
    unsigned i;
#if BOOST_VERSION >= 105000
    if (stack_size > 0)
        m_p->attrs.set_stack_size(stack_size);
#else
    if (stack_size)
        yaz_log(YLOG_WARN, "stack_size has no effect (Requires Boost 1.50)");
#endif
    for (i = 0; i < min_threads; i++)
        add_worker();
}

References add_worker(), and m_p.

Here is the call graph for this function:

~ThreadPoolSocketObserver()

ThreadPoolSocketObserver::~ThreadPoolSocketObserver ( )

virtual

Definition at line 128 of file thread_pool_observer.cpp.

{
    {
        boost::mutex::scoped_lock input_lock(m_p->m_mutex_input_data);
        m_p->m_stop_flag = true;
        m_p->m_cond_input_data.notify_all();
    }
    m_p->m_thrds.join_all();
    m_p->m_socketObservable->deleteObserver(this);
}

References m_p.

Member Function Documentation

add_worker()

void ThreadPoolSocketObserver::add_worker ( void  )

private

Definition at line 139 of file thread_pool_observer.cpp.

{
    Worker w(this);
#if BOOST_VERSION >= 105000
    boost::thread *x = new boost::thread(m_p->attrs, w);
#else
    boost::thread *x = new boost::thread(w);
#endif
    m_p->m_no_threads++;
    m_p->m_thrds.add_thread(x);
}

References m_p.

Referenced by put(), and ThreadPoolSocketObserver().

cleanup()

void ThreadPoolSocketObserver::cleanup	(	IThreadPoolMsg *	m,
		void *	info
	)

Definition at line 247 of file thread_pool_observer.cpp.

{
    boost::mutex::scoped_lock input_lock(m_p->m_mutex_input_data);
 
    std::deque<IThreadPoolMsg *>::iterator it = m_p->m_input.begin();
    while (it != m_p->m_input.end())
    {
        if ((*it)->cleanup(info))
        {
            delete *it;
            it = m_p->m_input.erase(it);
        }
        else
            it++;
    }
}

References m_p.

get()

IThreadPoolMsg * metaproxy_1::ThreadPoolSocketObserver::get

(

)

get_thread_info()

void ThreadPoolSocketObserver::get_thread_info	(	int &	tbusy,
		int &	total
	)

Definition at line 201 of file thread_pool_observer.cpp.

{
    tbusy = m_p->m_no_threads - m_p->m_waiting_threads;
    total = m_p->m_no_threads;
}

References m_p.

put()

void ThreadPoolSocketObserver::put

(

IThreadPoolMsg *

m

)

Definition at line 264 of file thread_pool_observer.cpp.

{
    boost::mutex::scoped_lock input_lock(m_p->m_mutex_input_data);
    if (m_p->m_waiting_threads == 0 &&
        m_p->m_no_threads < m_p->m_max_threads)
    {
        add_worker();
    }
    while (m_p->m_input.size() >= m_p->m_no_threads * queue_size_per_thread)
        m_p->m_cond_input_full.wait(input_lock);
    m_p->m_input.push_back(m);
    m_p->m_cond_input_data.notify_one();
}

References add_worker(), m_p, and metaproxy_1::queue_size_per_thread.

Here is the call graph for this function:

run()

void ThreadPoolSocketObserver::run

(

void *

p

)

Definition at line 207 of file thread_pool_observer.cpp.

{
    while(1)
    {
        IThreadPoolMsg *in = 0;
        {
            boost::mutex::scoped_lock input_lock(m_p->m_mutex_input_data);
            m_p->m_waiting_threads++;
            while (!m_p->m_stop_flag && m_p->m_input.size() == 0)
                m_p->m_cond_input_data.wait(input_lock);
            m_p->m_waiting_threads--;
            if (m_p->m_stop_flag)
                break;
 
            in = m_p->m_input.front();
            m_p->m_input.pop_front();
            m_p->m_cond_input_full.notify_all();
        }
        IThreadPoolMsg *out = in->handle();
        {
            boost::mutex::scoped_lock output_lock(m_p->m_mutex_output_data);
            m_p->m_output.push_back(out);
#ifdef WIN32
            send(m_p->m_pipe.write_fd(), "", 1, 0);
#else
            ssize_t r = write(m_p->m_pipe.write_fd(), "", 1);
            if (r != 1)
            {
                if (r == (ssize_t) (-1))
                    yaz_log(YLOG_WARN|YLOG_ERRNO,
                            "ThreadPoolSocketObserver::run. write fail");
                else
                    yaz_log(YLOG_WARN,
                            "ThreadPoolSocketObserver::run. write returned 0");
            }
#endif
        }
    }
}

References metaproxy_1::IThreadPoolMsg::handle(), and m_p.

Referenced by metaproxy_1::ThreadPoolSocketObserver::Worker::operator()().

Here is the call graph for this function:

socketNotify()

void ThreadPoolSocketObserver::socketNotify ( int  event )

private

Definition at line 151 of file thread_pool_observer.cpp.

{
    if (event & SOCKET_OBSERVE_READ)
    {
        char buf[2];
#ifdef WIN32
        recv(m_p->m_pipe.read_fd(), buf, 1, 0);
#else
        ssize_t r = read(m_p->m_pipe.read_fd(), buf, 1);
        if (r != 1)
        {
            if (r == (ssize_t) (-1))
                yaz_log(YLOG_WARN|YLOG_ERRNO,
                        "ThreadPoolSocketObserver::socketNotify. read fail");
            else
                yaz_log(YLOG_WARN,
                        "ThreadPoolSocketObserver::socketNotify. read returned 0");
        }
#endif
        while (1)
        {
            IThreadPoolMsg *out;
            size_t output_size = 0;
            {
                boost::mutex::scoped_lock output_lock(m_p->m_mutex_output_data);
                if (m_p->m_output.empty()) {
                    break;
                }
                out = m_p->m_output.front();
                m_p->m_output.pop_front();
                output_size = m_p->m_output.size();
            }
            if (out)
            {
                size_t input_size = 0;
                std::ostringstream os;
                {
                    boost::mutex::scoped_lock input_lock(m_p->m_mutex_input_data);
                    input_size = m_p->m_input.size();
                }
                os << "tbusy/total "
                    << m_p->m_no_threads - m_p->m_waiting_threads
                    << "/" << m_p->m_no_threads
                    << " queue in/out " << input_size << "/" <<  output_size;
                out->result(os.str().c_str());
            }
        }
    }
}

References m_p, and metaproxy_1::IThreadPoolMsg::result().

Here is the call graph for this function:

Member Data Documentation

m_p

boost::scoped_ptr<Rep> metaproxy_1::ThreadPoolSocketObserver::m_p

private

Definition at line 52 of file thread_pool_observer.hpp.

Referenced by add_worker(), cleanup(), get_thread_info(), put(), run(), socketNotify(), ThreadPoolSocketObserver(), and ~ThreadPoolSocketObserver().

---

The documentation for this class was generated from the following files:

• thread_pool_observer.hpp
• thread_pool_observer.cpp