Changeset 19043

Timestamp:

09/09/08 20:46:02 (5 years ago)

Author:

frsyuki

Message:

lang/c/mpio: implemented mp::iothreads::{manager,writer,listener,timer}

Location:

lang/c/mpio/trunk

Files:

• mp/iothreads.pre.h (modified) (5 diffs)
• mp/iothreads/connect.h (modified) (1 diff)
• mp/iothreads/listen.h (modified) (2 diffs)
• mp/iothreads/reader.h (modified) (6 diffs)
• mp/iothreads/timer.h (modified) (2 diffs)
• mp/iothreads/writer.h (modified) (3 diffs)
• mp/iothreads_impl.pre.h (modified) (6 diffs)
• mp/multiplex.pre.h (modified) (1 diff)
• mp/multiplex_impl.pre.h (modified) (3 diffs)
• mp/sparse_array_impl.pre.h (modified) (1 diff)
• mp/utility.pre.h (modified) (2 diffs)
• src/iothreads.cc (added)
• src/iothreads_listen.cc (added)
• src/iothreads_timer.cc (added)
• src/iothreads_writer.cc (added)

lang/c/mpio/trunk/mp/iothreads.pre.h

@@ -25 +25 @@
 #include "mp/zone.h"
 #include "mp/callback_message.h"
-#include "mp/utility.h"
+#include <memory>
 
 
@@ -62 +62 @@
         static void submit(callback_message* msg);
 
+        static void join();
+
         static void run();
         static void end();
         static bool is_end();
-
-        // FIXME join
 
 private:
@@ -78 +78 @@
         void submit_impl(callback_message* msg);
         void run_impl();
+        void join_impl();
 
 private:
         blocking_vector<callback_message*> m_messages;
-        source<> m_source;
-        zone<source<> > m_zone;
-
-        typedef std::vector<callback_message*> messages_t;
+        zone<> m_zone;
 
 private:
         volatile sig_atomic_t m_end_flag;
+        typedef std::vector<pthread_base*> threads_t;
+        threads_t m_threads;
 
 private:
+        typedef std::vector<callback_message*> messages_t;
         void clear_message(messages_t& messages);
 
 private:
-        static scoped_ptr<manager> s_instance;
+        static std::auto_ptr<manager> s_instance;
         static manager& instance() { return *s_instance; }
 
@@ -101 +102 @@
 };
 
-
-inline void initialize()
-        { manager::initialize(); }
-
-inline void destroy()
-        { manager::destroy(); }
 
 inline bool is_end()
@@ -147 +142 @@
 #include "mp/iothreads/writer.h"
 #include "mp/iothreads/listen.h"
+//#include "mp/iothreads/connect.h"
 #include "mp/iothreads/timer.h"
 

lang/c/mpio/trunk/mp/iothreads/connect.h

@@ -19 +19 @@
 #ifndef MP_IOTHREADS_LISTEN_H__
 #define MP_IOTHREADS_LISTEN_H__
+
+#include <memory>
 
 #include <unistd.h>

lang/c/mpio/trunk/mp/iothreads/listen.h

@@ -20 +20 @@
 #define MP_IOTHREADS_LISTEN_H__
 
-#include <unistd.h>
-#include <sys/types.h>
-#include <sys/socket.h>
-#include <errno.h>
-
 namespace mp {
 namespace iothreads {
@@ -33 +28 @@
         class listen_thread {
         public:
-                listen_thread(int lsock, void (*callback)(void*, int fd), void* obj) :
-                        m_lsock(lsock), m_callback(callback), m_obj(obj) {}
-                void operator() ()
-                {
-                        while(!iothreads::is_end()) {
-                                int sock = ::accept(m_lsock, NULL, NULL);
-                                if(sock < 0) {
-                                        if(errno == EINTR || errno == EAGAIN) { continue; }
-                                        // FIXME
-                                        throw std::runtime_error("socket broken");
-                                } else if(sock == 0) {
-                                        // FIXME
-                                        throw std::runtime_error("socket broken");
-                                }
-                                iothreads::submit(m_callback, m_obj, sock);
-                        }
-                }
+                listen_thread(int lsock,
+                                void (*callback)(void*, int fd), void* obj);
+                void operator() ();
         private:
                 int m_lsock;

lang/c/mpio/trunk/mp/iothreads/reader.h

@@ -37 +37 @@
                         { return m_fd; }
                 void send_data(const char* buf, size_t len)
-                        { return reader::send_data(m_fd, buf, len); }
+                        { return writer::send_data(m_fd, buf, len); }
         private:
                 int m_fd;
@@ -72 +72 @@
 MP_ARGS_END
 
-        handler& data_impl(int fd);
-        void send_data_impl(int fd, const char* buf, size_t len);
         //void send_message_impl(int fd, callback_message* msg);
 
@@ -90 +88 @@
                 class cb_t {
                 public:
-                        cb_t(handler* h, short ev);
+                        cb_t(handler* h);
                         ~cb_t();
                 public:
-                        short event() const;
-                        void event(short ev);
                         handler& get();
                 private:
                         handler* m_handler;
-                        short m_event;
                 };
 
@@ -105 +100 @@
                 ev_t m_ev;
 
-                fdnotify<> m_notify;  // FIXME
+                fdnotify<handler*> m_notify;
 
         private:
@@ -118 +113 @@
         private:
                 // inter-thread impl
-                void msg_add(handler* h);
-                void msg_send_message(int fd, callback_message* msg);
+                void add_impl(handler* h);
+                void send_message_impl(int fd, callback_message* msg);
         };
 
@@ -155 +150 @@
 
 
-short reader::worker::cb_t::event() const
-{
-        return m_event;
-}
-
-void reader::worker::cb_t::event(short ev)
-{
-        m_event = ev;
-}
+template <typename Handler>
+void reader::add(int fd)
+{
+        instance().add_impl<Handler>(fd);
+}
+
+MP_ARGS_BEGIN
+template <typename Handler, MP_ARGS_TEMPLATE>
+void reader::add(int fd, MP_ARGS_PARAMS)
+{
+        instance().add_impl<Handler>(fd, MP_ARGS_FUNC);
+}
+MP_ARGS_END
+
+
+template <typename Handler>
+void reader::add_impl(int fd)
+{
+        handler* h = m_zone.allocate<Handler>(fd);
+        worker_of(fd).add(h);
+        h->connected();
+}
+
+MP_ARGS_BEGIN
+template <typename Handler, MP_ARGS_TEMPLATE>
+void reader::add_impl(int fd, MP_ARGS_PARAMS)
+{
+        handler* h = m_zone.allocate<Handler>(fd, MP_ARGS_FUNC);
+        worker_of(fd).add(h);
+        h->connected();
+}
+MP_ARGS_END
+
+void reader::worker::add(handler* h)
+{
+        m_notify.send(h);
+}
+
+handler& reader::worker::cb_t::cb_t(handler* h) :
+                m_handler(h) {}
+
+handler& reader::worker::cb_t::~cb_t() {}
 
 handler& reader::worker::cb_t::get()
 {
         return *m_handler;
+}
+
+
+void reader::worker::operator() ()
+{
+        cb_t* pcb;
+        int fd;
+        short event;
+        int ret;
+        while(!m_end_flag) {
+                while( m_ev.next(&fd, &event, &pcb) ) {
+                        if(fd == m_notify.getfd()) {
+                                handler* h;
+                                while(m_notify.try_receive(&h)) {
+                                        add_impl(h);
+                                }
+                        } else {
+                                try_read(*pcb);
+                        }
+                }
+                if( (ret = m_ev.wait()) < 0 ) { return ret; }
+        }
+        return 0;
+}
+
+void reader::worker::add_impl(handler* h)
+{
+        m_ev.add(h->fd(), EV_READ, h);
+}
+
+void reader::worker::try_read(cb_t& cb)
+{
+        try {
+                cb.get().read_event();
+        } catch (...) {
+                // FIXME log
+                close_connection(cb);
+        }
+}
+
+void reader::worker::close_connection(cb_t& cb)
+{
+        int fd = cb.get().fd();
+        m_ev.remove(fd, EV_READ);
+        ::close(fd);
+        iothreads::submit(
+                        &object_callback<void (void)>::mem_fun<reader, &reader::remove>,
+                        &cb.get());
+        // FIXME delete handler
 }
 

lang/c/mpio/trunk/mp/iothreads/timer.h

@@ -20 +20 @@
 #define MP_IOTHREADS_TIMER_H__
 
-#include <unistd.h>
-
 namespace mp {
 namespace iothreads {
@@ -30 +28 @@
         class timer_thread {
         public:
-                timer_thread(double interval_sec, void (*callback)(void*), void* obj) :
-                        m_interval_usec(interval_sec * 1000 * 1000),
-                        m_callback(callback),
-                        m_obj(obj) {}
-                void operator() ()
-                {
-                        while(!iothreads::is_end()) {
-                                usleep(m_interval_usec);
-                                iothreads::submit(m_callback, m_obj);
-                        }
-                }
+                timer_thread(double interval_sec,
+                                void (*callback)(void*), void* obj);
+                void operator() ();
         private:
                 useconds_t m_interval_usec;

lang/c/mpio/trunk/mp/iothreads/writer.h

@@ -20 +20 @@
 #define MP_IOTHREADS_WRITER_H__
 
-#include "mp/event.h"
-#include "mp/fdnotify.h"
-#include <sys/types.h>
-#include <sys/uio.h>
-#include <unistd.h>
-#include <errno.h>
+#include <memory>
 
 namespace mp {
@@ -35 +30 @@
         static void initialize(unsigned int num_threads);
         static void destroy();
-        ~writer();
 
 public:
-        typedef void (*finalize_t)(void* user, int fd, const void* buf, size_t buflen);
+        typedef void (*finalize_t)(void* user, const void* buf, size_t buflen);
 
         static void send_data(int fd, const void* buf, size_t buflen);
@@ -48 +42 @@
                         finalize_t finalize, void* user, bool thread_safe = false);
 
-private:
-
-        void send_data_impl(int fd, const void* buf, size_t buflen);
-        void send_data_impl(int fd, const void* buf, size_t buflen, finalize_t finalize,
-                        void* user, bool thread_safe = false);
-
-        void send_data_thread_impl(int fd, const void* buf, size_t buflen);
-        void send_data_thread_impl(int fd, const void* buf, size_t buflen,
-                        finalize_t finalize, void* user, bool thread_safe = false);
+public:
+        static void finalize_free(void* user, const void* buf, size_t buflen)
+                { free((void*)buf); }
 
 private:
-        class worker {
-        public:
-                worker();
-                ~worker();
-
-        public:
-                void add(int fd, char* ebit,
-                                const void* buf, size_t buflen, size_t off,
-                                finalize_t finalize, void* user,
-                                bool thread_safe);
-
-        public:
-                void operator() ();
-
-        private:
-                struct notify_entry {
-                        int fd;
-                        char* ebit;
-                        const void* buf;
-                        size_t buflen;
-                        size_t off;
-                        finalize_t finalize;
-                        void* user;
-                        bool thread_safe;
-                };
-
-                fdnotify<notify_entry> m_notify;
-
-                void add_impl(notify_entry& e);
-
-        private:
-                class request {
-                public:
-                        request(char* ebit, const void* buf, size_t buflen,
-                                        finalize_t finalize, void* user,
-                                        bool thread_safe);
-                        ~request();
-
-                public:
-                        void finalize(int fd);
-
-                private:
-                        char* m_ebit;
-                        const void* m_buf;
-                        size_t m_buflen;
-                        finalize_t m_finalize;
-                        void* m_user;
-                        bool m_thread_safe;
-                };
-
-                class context {
-                public:
-                        context();
-                        ~context();
-
-                public:
-                        bool empty();
-                        void add(char* ebit,
-                                        const void* buf, size_t buflen, size_t off,
-                                        finalize_t finalize, void* user,
-                                        bool thread_safe);
-
-                public:
-                        iovec* vec();
-                        int veclen();
-                        void consumed(int fd, int num);
-
-                private:
-                        typedef std::vector<iovec> bufvec_t;
-                        bufvec_t m_bufvec;
-
-                        typedef std::vector<request> reqvec_t;
-                        reqvec_t m_reqvec;
-                };
-
-                typedef event<context> ev_t;
-                ev_t m_ev;
-        };
-
-
-        class empty_checker {
-        public:
-                empty_checker();
-                ~empty_checker();
-        public:
-                bool empty();
-                char* use();
-        private:
-                class chunk_t {
-                public:
-                        chunk_t();
-                public:
-                        bool empty();
-                        char* use();
-                private:
-                        char m_bit[sizeof(uint64_t)];
-                        unsigned short m_used;
-                };
-                typedef std::vector<chunk_t> bits_t;
-                bits_t m_bits;
-        };
-
-        typedef sparse_array<empty_checker> empty_check_t;
-        empty_check_t m_empty_check;
-
-        std::vector<worker*> m_workers;
-
-private:
-        worker& worker_of(int fd);
-        void add_data_with_copy(int fd, const void* buf, size_t buflen);
-        void add_data_with_finalize(int fd, const void* buf, size_t buflen, size_t off,
-                        finalize_t finalize, void* user, bool thread_safe);
-
-private:
-        writer(unsigned int num_threads);
-
-        static scoped_ptr<writer> s_instance;
-        static writer& instance();
+        class impl;
+        static std::auto_ptr<impl> s_instance;
+        static impl& instance();
 
 private:
         writer();
+        ~writer();
         writer(const writer&);
-
-private:
-        static void finalize_free(void* user, int fd, const void* buf, size_t buflen)
-                { free((void*)buf); }
 };
 
 
-void writer::initialize(unsigned int num_threads)
+typedef writer::finalize_t finalize_t;
+
+inline void send_data(int fd, const void* buf, size_t buflen)
 {
-        s_instance.reset(new writer(num_threads));
+        writer::send_data(fd, buf, buflen);
 }
 
-void writer::destroy()
+inline void send_data(int fd, const void* buf, size_t buflen, finalize_t finalize,
+                void* user, bool thread_safe = false)
 {
-        s_instance.reset(NULL);
+        writer::send_data(fd, buf, buflen, finalize, user, thread_safe);
 }
-
-writer& writer::instance()
-{
-        return *s_instance;
-}
-
-
-writer::writer(unsigned int num_threads)
-{
-        m_workers.reserve(num_threads);
-        for(unsigned int i=0; i < num_threads; ++i) {
-                worker* w = iothreads::manager::add_thread<worker>();
-                m_workers.push_back(w);
-        }
-}
-
-writer::~writer() {}
-
-
-void writer::send_data(int fd, const void* buf, size_t buflen)
-{
-        instance().send_data_impl(fd, buf, buflen);
-}
-
-void writer::send_data_impl(int fd, const void* buf, size_t buflen)
-{
-        if(!m_empty_check.test(fd) || m_empty_check.data(fd).empty()) {
-                ssize_t wl = ::write(fd, buf, buflen);
-                if(wl < 0) {
-                        if(errno == EAGAIN || errno == EINTR) {
-                                add_data_with_copy(fd, buf, buflen);
-                        } else { return; }
-                } else if(wl == 0) { return; }
-                if(static_cast<size_t>(wl) < buflen) {
-                        add_data_with_copy(fd, (const void*)(((const char*)buf)+wl), buflen-wl);
-                } else {
-                        return;
-                }
-        } else {
-                add_data_with_copy(fd, buf, buflen);
-        }
-}
-
-void writer::add_data_with_copy(int fd, const void* buf, size_t buflen)
-{
-        if(!m_empty_check.test(fd)) { m_empty_check.set(fd); }
-        void* nbuf = malloc(buflen);
-        if(nbuf == NULL) { throw std::bad_alloc(); }
-        memcpy(nbuf, buf, buflen);
-        worker_of(fd).add(fd, m_empty_check.data(fd).use(),
-                        buf, buflen, 0,
-                        &writer::finalize_free, NULL,
-                        true);
-}
-
-void writer::send_data(int fd, const void* buf, size_t buflen, finalize_t finalize, void* user, bool thread_safe)
-{
-        instance().send_data_impl(fd, buf, buflen,
-                        finalize, user, thread_safe);
-}
-
-void writer::send_data_impl(int fd, const void* buf, size_t buflen, finalize_t finalize, void* user, bool thread_safe)
-{
-        if(!m_empty_check.test(fd) || m_empty_check.data(fd).empty()) {
-                ssize_t wl = ::write(fd, buf, buflen);
-                if(wl < 0) {
-                        if(errno == EAGAIN || errno == EINTR) {
-                                add_data_with_finalize(fd, buf, buflen, 0, finalize, user, thread_safe);
-                        } else { return; }
-                } else if(wl == 0) { return; }
-                if(static_cast<size_t>(wl) < buflen) {
-                        add_data_with_finalize(fd, buf, buflen, wl, finalize, user, thread_safe);
-                } else {
-                        (*finalize)(user, fd, buf, buflen);
-                }
-        } else {
-                add_data_with_finalize(fd, buf, buflen, 0, finalize, user, thread_safe);
-        }
-}
-
-void writer::add_data_with_finalize(int fd, const void* buf, size_t buflen, size_t off,
-                finalize_t finalize, void* user, bool thread_safe)
-{
-        if(!m_empty_check.test(fd)) { m_empty_check.set(fd); }
-        worker_of(fd).add(fd, m_empty_check.data(fd).use(),
-                        buf, buflen, off,
-                        finalize, user,
-                        thread_safe);
-}
-
-writer::worker& writer::worker_of(int fd)
-{
-        return *m_workers[fd % m_workers.size()];
-}
-
-
-void writer::send_data_thread(int fd, const void* buf, size_t buflen)
-{
-        instance().send_data_thread_impl(fd, buf, buflen);
-}
-
-void writer::send_data_thread_impl(int fd, const void* buf, size_t buflen)
-{
-        add_data_with_copy(fd, buf, buflen);
-}
-
-void writer::send_data_thread(int fd, const void* buf, size_t buflen,
-                finalize_t finalize, void* user, bool thread_safe)
-{
-        instance().send_data_thread_impl(fd, buf, buflen,
-                        finalize, user, thread_safe);
-}
-
-void writer::send_data_thread_impl(int fd, const void* buf, size_t buflen,
-                finalize_t finalize, void* user, bool thread_safe)
-{
-        add_data_with_finalize(fd, buf, buflen, 0, finalize, user, thread_safe);
-}
-
-
-writer::empty_checker::empty_checker()
-{
-        m_bits.resize(1);
-}
-
-writer::empty_checker::~empty_checker() { }
-
-
-bool writer::empty_checker::empty()
-{
-        for(bits_t::iterator it(m_bits.begin()), it_end(m_bits.end());
-                        it != it_end;
-                        ++it) {
-                if(!it->empty()) {
-                        return false;
-                }
-        }
-        return true;
-}
-
-char* writer::empty_checker::use()
-{
-        char* b = m_bits.back().use();
-        if(b) { return b; }
-        m_bits.push_back(chunk_t());
-        return m_bits.back().use();
-}
-
-writer::empty_checker::chunk_t::chunk_t() :
-        m_used(0)
-{
-        memset(m_bit, 0, sizeof(m_bit));
-}
-
-bool writer::empty_checker::chunk_t::empty()
-{
-        static const uint64_t ZERO = 0ULL;
-        return (memcmp(m_bit, (char*)&ZERO, sizeof(m_bit)) == 0);
-}
-
-char* writer::empty_checker::chunk_t::use()
-{
-        if(m_used >= sizeof(m_bit)) { return NULL; }
-        char* b = &m_bit[m_used++];
-        *b = 1;
-        return b;
-}
-
-
-writer::worker::request::request(char* ebit, const void* buf, size_t buflen,
-                finalize_t finalize, void* user,
-                bool thread_safe) :
-        m_ebit(ebit), m_buf(buf), m_buflen(buflen),
-        m_finalize(finalize), m_user(user),
-        m_thread_safe(thread_safe)
-{ }
-
-writer::worker::request::~request() { }
-
-void writer::worker::request::finalize(int fd)
-{
-        *m_ebit = 0;
-        if(m_finalize) {
-                if(m_thread_safe) {
-                        (*m_finalize)(m_user, fd, m_buf, m_buflen);
-                } else {
-                        iothreads::submit(m_finalize, m_user, fd, m_buf, m_buflen);
-                }
-        }
-}
-
-
-writer::worker::context::context() { }
-
-writer::worker::context::~context()
-{
-        if(!m_reqvec.empty()) {
-                for(reqvec_t::iterator it(m_reqvec.begin()), it_end(m_reqvec.end());
-                                it != it_end;
-                                ++it) {
-                        // FIXME
-                        //it->finalize();
-                }
-        }
-}
-
-
-inline void writer::worker::context::add(char* ebit,
-                const void* buf, size_t buflen, size_t off,
-                finalize_t finalize, void* user,
-                bool thread_safe)
-{
-        struct iovec vec = { ((char*)buf)+off, buflen-off };
-        m_bufvec.push_back(vec);
-        try {
-                m_reqvec.push_back(request(ebit, buf, buflen, finalize, user, thread_safe));
-        } catch (...) {
-                m_bufvec.pop_back();
-                // FIXME
-        }
-}
-
-
-iovec* writer::worker::context::vec()
-{
-        return &m_bufvec[0];
-}
-
-int writer::worker::context::veclen()
-{
-        return m_bufvec.size();
-}
-
-bool writer::worker::context::empty()
-{
-        return m_bufvec.empty();
-}
-
-inline void writer::worker::context::consumed(int fd, int num)
-{
-        for(int i=0; i < num; ++i) {
-                m_reqvec[i].finalize(fd);
-        }
-        m_bufvec.erase(m_bufvec.begin(), m_bufvec.begin()+num);
-        m_reqvec.erase(m_reqvec.begin(), m_reqvec.begin()+num);
-}
-
-
-writer::worker::worker()
-{
-        m_ev.add(m_notify.getfd(), EV_READ);
-}
-
-writer::worker::~worker() {}
-
-
-void writer::worker::add(int fd, char* ebit,
-                const void* buf, size_t buflen, size_t off,
-                finalize_t finalize, void* user,
-                bool thread_safe)
-{
-        notify_entry e = {
-                fd, ebit,
-                buf, buflen, off,
-                finalize, user, thread_safe
-        };
-        m_notify.send(e);
-}
-
-inline void writer::worker::add_impl(notify_entry& e)
-{
-        if(!m_ev.test(e.fd)) {
-                m_ev.add(e.fd, EV_WRITE);
-        }
-        m_ev.data(e.fd).add(e.ebit, e.buf, e.buflen, e.off,
-                        e.finalize, e.user, e.thread_safe);
-}
-
-void writer::worker::operator() ()
-{
-        int fd;
-        short event;
-        context* pctx;
-        while(!iothreads::is_end()) {
-                while(m_ev.next(&fd, &event, &pctx)) {
-                        if(fd == m_notify.getfd()) {
-                                notify_entry e;
-                                while(m_notify.try_receive(&e)) {
-                                        add_impl(e);
-                                }
-                        } else {
-                                context& ctx(m_ev.data(fd));
-                                ssize_t wl = ::writev(fd, ctx.vec(), ctx.veclen());
-                                if(wl < 0) {
-                                        if(errno == EAGAIN || errno == EINTR) {
-                                                continue;
-                                        } else {
-                                                m_ev.remove(fd, EV_WRITE);  // FIXME
-                                                continue;
-                                        }
-                                } else if(wl == 0){
-                                        m_ev.remove(fd, EV_WRITE);  // FIXME
-                                        continue;
-                                }
-                                int i;
-                                for(i=0; i < ctx.veclen(); ++i) {
-                                        if(static_cast<size_t>(wl) >= ctx.vec()[i].iov_len) {
-                                                wl -= ctx.vec()[i].iov_len;
-                                        } else {
-                                                ctx.vec()[i].iov_base = (void*)(((char*)ctx.vec()[i].iov_base) + wl);
-                                                ctx.vec()[i].iov_len -= wl;
-                                                break;
-                                        }
-                                }
-                                if(i > 0) {
-                                        ctx.consumed(fd, i);
-                                        if(ctx.empty()) {
-                                                m_ev.remove(fd, EV_WRITE);
-                                        }
-                                }
-                                //sleep(1);
-                                //throw std::bad_alloc();
-                        }
-                }
-                m_ev.wait();  // FIXME
-        }
-}
-
-
-// FIXME
-scoped_ptr<writer> writer::s_instance;
 
 

lang/c/mpio/trunk/mp/iothreads_impl.pre.h

@@ -25 +25 @@
 
 
-scoped_ptr<manager> manager::s_instance;  // FIXME
-
-void manager::initialize()
-{
-        s_instance.reset(new manager());
+template <typename ThreadIMPL>
+ThreadIMPL* manager::add_thread() {
+        return instance().add_thread_impl<ThreadIMPL>();
 }
-
-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); }
+ThreadIMPL* manager::add_thread(MP_ARGS_PARAMS) {
+        return instance().add_thread_impl<ThreadIMPL>(MP_ARGS_FUNC);
+}
 MP_ARGS_END
 
@@ -60 +42 @@
 {
         ThreadIMPL* impl = m_zone.allocate<ThreadIMPL>();
-        m_zone.allocate< pthread_thread<ThreadIMPL> >(impl);
+        pthread_base* th = m_zone.allocate< pthread_thread<ThreadIMPL> >(impl);
+        m_threads.push_back(th);
         return impl;
 }
@@ -69 +52 @@
 {
         ThreadIMPL* impl = m_zone.allocate<ThreadIMPL, MP_ARGS_TYPES>(MP_ARGS_FUNC);
-        m_zone.allocate< pthread_thread<ThreadIMPL> >(impl);
+        pthread_base* th = m_zone.allocate< pthread_thread<ThreadIMPL> >(impl);
+        m_threads.push_back(th);
         return impl;
 }
@@ -75 +59 @@
 
 
-inline void manager::submit(callback_message* msg)
-{
-        instance().submit_impl(msg);
-}
-
-void manager::submit_impl(callback_message* msg)
+inline void manager::submit_impl(callback_message* msg)
 {
         try {
@@ -90 +69 @@
 }
 
+
+inline void manager::submit(callback_message* msg)
+{
+        instance().submit_impl(msg);
+}
+
 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()
@@ -136 +91 @@
 
 
+inline void manager::join()
+{
+        instance().join_impl();
+}
+
+inline void manager::join_impl()
+{
+        for(threads_t::iterator it(m_threads.begin()), it_end(m_threads.end());
+                        it != it_end;
+                        ++it) {
+                (*it)->join();
+        }
+}
+
+
 }  // namespace iothreads
 }  // namespace mp

lang/c/mpio/trunk/mp/multiplex.pre.h

@@ -93 +93 @@
         typedef event<cb_t> ev_t;
         ev_t m_ev;
-        source<> m_source;
-        zone<source<> > m_zone;
+        zone<> m_zone;
 
 private:

lang/c/mpio/trunk/mp/multiplex_impl.pre.h

@@ -139 +139 @@
 void manager::add_impl(int fd)
 {
+        // FIXME sparse_array
         handler* h = m_zone.allocate<Handler>(fd);
         m_ev.add(fd, EV_READ, h, EV_READ);
@@ -148 +149 @@
 void manager::add_impl(int fd, MP_ARGS_PARAMS)
 {
+        // FIXME sparse_array
         handler* h = m_zone.allocate<Handler>(fd, MP_ARGS_FUNC);
         m_ev.add(fd, EV_READ, h, EV_READ);
@@ -275 +277 @@
 {
         int fd = cb.get().fd();
+        // FIXME remove handler form sparse_array
         m_ev.remove(fd, cb.event());
         ::close(fd);

lang/c/mpio/trunk/mp/sparse_array_impl.pre.h

@@ -126 +126 @@
 void* sparse_array<T>::set_impl(size_type index)
 {
-        while( base_array.size() * EXTEND_ARRAY_SIZE <= index ) {
+        while( base_array.size() <= index / EXTEND_ARRAY_SIZE ) {
                 extend_array();
         }

lang/c/mpio/trunk/mp/utility.pre.h

@@ -20 +20 @@
 #define MP_UTILITY_H__
 
+#include <unistd.h>
+#include <fcntl.h>
 #include <memory>
 
@@ -31 +33 @@
 
 typedef thread_tag<0> main_thread_tag;
+
+
+inline void set_nonblock(int fd)
+{
+        // FIXME error
+        fcntl(fd, F_SETFL, O_NONBLOCK);
+}