root/lang/c/mpio/trunk/mp/iothreads_impl.pre.h @ 19040

//
// mp::iothreads
//
// Copyright (C) 2008 FURUHASHI Sadayuki
//
//    Licensed under the Apache License, Version 2.0 (the "License");
//    you may not use this file except in compliance with the License.
//    You may obtain a copy of the License at
//
//        http://www.apache.org/licenses/LICENSE-2.0
//
//    Unless required by applicable law or agreed to in writing, software
//    distributed under the License is distributed on an "AS IS" BASIS,
//    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
//    See the License for the specific language governing permissions and
//    limitations under the License.
//

#ifndef MP_IOTHREADS_IMPL_H__
#define MP_IOTHREADS_IMPL_H__


namespace mp {
namespace iothreads {


scoped_ptr<manager> manager::s_instance;  // FIXME

void manager::initialize()
{
        s_instance.reset(new manager());
}

void manager::destroy()
{
        s_instance.reset(NULL);
}


manager::manager() :
        m_zone(m_source),
        m_end_flag(0)
{ }

manager::~manager()
{ }


template <typename ThreadIMPL>
inline ThreadIMPL* manager::add_thread() { return instance().add_thread_impl<ThreadIMPL>(); }

MP_ARGS_BEGIN
template <typename ThreadIMPL, MP_ARGS_TEMPLATE>
inline ThreadIMPL* manager::add_thread(MP_ARGS_PARAMS) { return instance().add_thread_impl<ThreadIMPL>(MP_ARGS_FUNC); }
MP_ARGS_END


template <typename ThreadIMPL>
ThreadIMPL* manager::add_thread_impl()
{
        ThreadIMPL* impl = m_zone.allocate<ThreadIMPL>();
        m_zone.allocate< pthread_thread<ThreadIMPL> >(impl);
        return impl;
}

MP_ARGS_BEGIN
template <typename ThreadIMPL, MP_ARGS_TEMPLATE>
ThreadIMPL* manager::add_thread_impl(MP_ARGS_PARAMS)
{
        ThreadIMPL* impl = m_zone.allocate<ThreadIMPL, MP_ARGS_TYPES>(MP_ARGS_FUNC);
        m_zone.allocate< pthread_thread<ThreadIMPL> >(impl);
        return impl;
}
MP_ARGS_END


inline void manager::submit(callback_message* msg)
{
        instance().submit_impl(msg);
}

void manager::submit_impl(callback_message* msg)
{
        try {
                m_messages.push_back(msg);
        } catch (...) {
                delete msg;
                throw;
        }
}

inline void manager::run()
{
        instance().run_impl();
}

void manager::run_impl()
{
        messages_t cache;
        while(!m_end_flag) {
                m_messages.swap(cache);
                for(messages_t::iterator it(cache.begin()), it_end(cache.end());
                                it != it_end;
                                ++it) {
                        callback_message* msg = *it;
                        try {
                                (*msg)();
                        } catch (...) {
                                // FIXME
                        }
                }
                clear_message(cache);
        }
}

void manager::clear_message(messages_t& messages)
{
        for(messages_t::iterator it(messages.begin()), it_end(messages.end());
                        it != it_end;
                        ++it) {
                delete *it;
        }
        messages.clear();
}


inline void manager::end()
{
        instance().m_end_flag = 1;
}

inline bool manager::is_end()
{
        return instance().m_end_flag == 1;
}


}  // namespace iothreads
}  // namespace mp

#endif /* mp/iothreads_impl.h */