Changeset 18476 for lang

Timestamp:

08/30/08 15:15:48 (3 months ago)

Author:

frsyuki

Message:

lang/c/mpio: added mp::iothreads::reader and mp::iothreads::writer

Location:

lang/c/mpio/trunk/mp

Files:

• iothreads.pre.h (modified) (3 diffs)
• iothreads/connect.h (modified) (3 diffs)
• iothreads/listen.h (modified) (1 diff)
• iothreads/reader.h (added)
• iothreads/timer.h (modified) (1 diff)
• iothreads/writer.h (added)
• iothreads_impl.pre.h (modified) (1 diff)
• multiplex.pre.h (modified) (1 diff)
• multiplex/connect.h (modified) (3 diffs)
• multiplex_impl.pre.h (modified) (6 diffs)

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

@@ -54 +54 @@
 public:
         template <typename ThreadIMPL>
-        static void add_thread();
+        static ThreadIMPL* add_thread();
 MP_ARGS_BEGIN
         template <typename ThreadIMPL, MP_ARGS_TEMPLATE>
-        static void add_thread(MP_ARGS_PARAMS);
+        static ThreadIMPL* add_thread(MP_ARGS_PARAMS);
 MP_ARGS_END
 
@@ -70 +70 @@
 private:
         template <typename ThreadIMPL>
-        void add_thread_impl();
+        ThreadIMPL* add_thread_impl();
 MP_ARGS_BEGIN
         template <typename ThreadIMPL, MP_ARGS_TEMPLATE>
-        void add_thread_impl(MP_ARGS_PARAMS);
+        ThreadIMPL* add_thread_impl(MP_ARGS_PARAMS);
 MP_ARGS_END
 
@@ -145 +145 @@
 
 #include "mp/iothreads_impl.h"
+#include "mp/iothreads/writer.h"
 #include "mp/iothreads/listen.h"
 #include "mp/iothreads/timer.h"
 
-
-/*
-namespace mp {
-
-
-template <typename ThreadTag = main_thread_tag>
-class iothreads {
-public:
-
-        template <typename IMPL>
-        class handler {
-        public:
-                handler(int fd);
-                virtual ~handler();
-        protected:
-                template <typename T>  // fixme
-                void submit(void (*callback)(T* obj, void* user), T* obj, void* user);
-        public:
-                //virtual int read_event() = 0;
-        };
-
-        template <typename IMPL>
-        class stream_handler : public handler {
-        public
-                stream_handler(int fd);  // allocate buffer
-                virtual ~stream_handler();
-        public:
-                int read_event();
-        public:
-                // throw exception to close fd
-                //virtual void receive_data(stream& s) = 0;
-                //virtual void unbind() = 0;
-        };
-
-        template <typename IMPL>
-        class timer : public handler {
-        public:
-                timer(int fd);
-                virtual ~timer();
-        public:
-                static start(double interval);
-        public:
-                //virtual void triger() = 0;
-        };
-
-
-        struct thread {
-                virtual ~thread() {}
-                virtual void run() = 0;
-        };
-
-        class io_thread : public thread {
-        public:
-                io_thread();
-                ~io_thread();
-        public:
-                void run();
-        private:
-                struct cb_t {
-                public:
-                        cb_t();
-                        ~cb_t();
-                public:
-                        short event();
-                        void event(short ev);
-                        handler& handler();
-                        // fixme buffer accessor
-                private:
-                        short m_event;
-                        handler& m_handler;
-                        char* m_wbuffer;  // fixme
-                };
-        private:
-                void close_connection(cb_t& cb);
-                void rswitch(cb_t& cb);
-                void wswitch(cb_t& cb);
-        public:
-                void add(handler* handler);
-                bool have(int fd);
-                void send_data(int fd, char* data, size_t len);  // fixme
-                void send_message(int fd, char* msg, size_t len);  // fixme
-        private:
-                // fixme
-                void msg_add();
-                void msg_send_data();
-                void msg_send_message();
-        };
-
-
-public:
-        static void initialize(unsigned int num_threads);
-        static void destroy();
-        ~buffer();
-
-        template <typename Handler>
-        void add(int fd);
-MP_ARGS_BEGIN
-        template <typename Handler, MP_ARGS_TEMPLATE>
-        inline int add(int fd, MP_ARGS_PARAMS);
-MP_ARGS_END
-
-public:
-        void remove(handler* ctx);
-
-public:
-        void send_data(int fd, const char* data, size_t len);
-        void send_message(int fd, const char* data, size_t len);  // fixme
-
-public:
-        template <typename Thread>
-        void custom_thread();
-MP_ARGS_BEGIN
-        template <typename Thread, MP_ARGS_TEMPLATE>
-        inline int custom_thread(MP_ARGS_PARAMS);
-MP_ARGS_END
-
-public:
-        void run();
-        void end();
-        bool is_end() const;
-
-public:
-        void submit(void* data);  // fixme
-
-private:
-        iothreads(unsigned int num_threads);
-
-private:
-        unsigned int m_num_threads;
-
-private:
-        iothreads(const iothreads&);
-        iothreads();
-};
-
-
-
-}  // namespace mp
-*/
-
-
 #endif /* mp/iothreads.h */
 

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

@@ -35 +35 @@
 
 private:
-        class connect_thread : public pthread_thread<connect_thread> {
+        class connect_thread {
         public:
                 // FIXME
@@ -78 +78 @@
         {
                 for(unsigned int i=0; i < num_threads; ++i) {
-                        iothreads::add_thread<connect_thread>();
+                        iothreads::manager::add_thread<connect_thread>();
                         // FIXME
                 }
@@ -103 +103 @@
 #endif /* mp/iothreads/listen.h */
 
-

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

@@ -31 +31 @@
 struct listener {
 private:
-        class listen_thread : public pthread_thread<listen_thread> {
+        class listen_thread {
         public:
                 listen_thread(int lsock, void (*callback)(void*, int fd), void* obj) :
-                        pthread_thread<listen_thread>(this),
                         m_lsock(lsock), m_callback(callback), m_obj(obj) {}
                 void operator() ()

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

@@ -28 +28 @@
 struct timer {
 private:
-        class timer_thread : public pthread_thread<timer_thread> {
+        class timer_thread {
         public:
                 timer_thread(double interval_sec, void (*callback)(void*), void* obj) :
-                        pthread_thread<timer_thread>(this),
                         m_interval_usec(interval_sec * 1000 * 1000),
                         m_callback(callback),

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

@@ -48 +48 @@
 
 template <typename ThreadIMPL>
-inline void manager::add_thread() { instance().add_thread_impl<ThreadIMPL>(); }
+inline ThreadIMPL* manager::add_thread() { return instance().add_thread_impl<ThreadIMPL>(); }
 
 MP_ARGS_BEGIN
 template <typename ThreadIMPL, MP_ARGS_TEMPLATE>
-inline void manager::add_thread(MP_ARGS_PARAMS) { instance().add_thread_impl<ThreadIMPL>(MP_ARGS_FUNC); }
+inline ThreadIMPL* manager::add_thread(MP_ARGS_PARAMS) { return instance().add_thread_impl<ThreadIMPL>(MP_ARGS_FUNC); }
 MP_ARGS_END
 
 
 template <typename ThreadIMPL>
-void manager::add_thread_impl()
+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>
-void manager::add_thread_impl(MP_ARGS_PARAMS)
+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

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

@@ -36 +36 @@
         virtual void connected() {}
 public:
-        int fd() const { return m_fd; }
+        int fd() const
+                { return m_fd; }
+        void send_data(const char* buf, size_t len)
+                { return manager::send_data(m_fd, buf, len); }
 private:
         int m_fd;

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

@@ -23 +23 @@
 #include <sys/types.h>
 #include <sys/socket.h>
+#include <netinet/in.h>
+#include <arpa/inet.h>
 #include <errno.h>
 
@@ -31 +33 @@
 struct connector {
 private:
-        static void initialize(unsigned int num_threads);
-        static void destroy();
+        static void initialize(unsigned int num_threads)
+        {
+                s_instance.reset(new connector(num_threads));
+        }
+
+        static void destroy()
+        {
+                s_instance.reset(NULL);
+        }
 
 private:
+        class entry {
+        public:
+                entry(socklen_t socklen, void (*callback)(void*, sockaddr*, socklen_t, int), void* obj)
+                        m_socklen(socklen), m_callback(callback), m_obj(obj),
+                        m_fd(0) {}
+                ~entry();
+        public:
+                sockaddr* addr() { return (sockaddr*)(((char*)this)+sizeof(entry)); }
+                socklen_t addrlen() { return m_socklen; }
+                void setfd(int fd) { m_fd = fd; }
+        public:
+                void callback()
+                {
+                        (*m_callback)(m_obj, addr(), m_socklen, m_fd);
+                }
+        private:
+                socklen_t m_socklen;
+                void (*m_callback)(void*, sockaddr*, socklen_t, int);
+                void* m_obj;
+                int m_fd;
+        };
+
         class connect_thread : public pthread_thread<connect_thread> {
         public:
-                // FIXME
+                connect_thread(fdnotify<entry*>& return_notify) :
+                        m_return_notify(return_notify)
+                {
+                        m_ev.add(m_notify.getfd(), EV_READ);  // FIXME error
+                }
+        public:
+                void operator() ()
+                {
+                        entry* pe;
+                        while(!multiplex::is_end()) {
+                                while(m_notify.try_receive(&pe)) {
+                                        int sock = ::socket(AF_INET, SOCK_STREAM, 0);  // FIXME AF_INET6
+                                        if(sock < 0) { continue; }  // FIXME
+                                        if(::connect(sock, pe->addr(), pe->socklen()) < 0) {
+                                                ::close(sock);
+                                        } else {
+                                                pe->setfd(sock);
+                                        }
+                                        m_return_notify.send(pe);
+                                }
+                                m_ev.wait();  // FIXME  timeout / error
+                        }
+                }
+        public:
+                void connect(entry* e) { m_notify.send(e); }
+        private:
+                typedef event<> ev_t;
+                ev_t m_ev;
+                fdnotify<entry*> m_return_notify;
+                fdnotify<entry*> m_notify;
         };
 
+
+        typedef source<sizeof(entry)+sizeof(sockaddr_in), 16> source_t;
+
         class connect_handler : public handler {
+        public:
+                connect_handler(int fd, fdnotify<event*>& return_notify, souce_t& source) :
+                        handler(fd), m_return_notify(return_notify), m_source(source) {}
         pubilc:
-                // FIXME
+                void read_event()
+                {
+                        entry* pe;
+                        while(m_return_notify.try_receive(&pe)) {
+                                try {
+                                        pe->callback();
+                                } catch(...) {
+                                        pe->~entry();
+                                        m_source.free(pe);
+                                        throw;
+                                }
+                                pe->~entry();
+                                m_source.free(pe);
+                        }
+                }
+        private:
+                fdnotify<>& m_return_notify;
+                source_t& m_source;
         };
 
 private:
-        connector(unsigned int num_threads)
+        connector(unsigned int num_threads) :
+                m_num_threads(num_threads),
+                m_rr_seed(0)
         {
+                m_threads = (connect_thread**)m_source.malloc(sizeof(m_threads*)*num_threads);
                 for(unsigned int i=0; i < num_threads; ++i) {
-                        iothreads::add_thread<connect_thread>();
-                        // FIXME
+                        m_threads[i] = m_zone.allocate<connect_thread>(m_return_notify);
                 }
+                multiplex::add<connect_handler>(m_return_notify.getfd(), m_source);
         }
 
         ~connector() {}  // FIXME
 
-        // entry
-        // return_entry
+        void connect_impl(sockaddr* addr, socklen_t addrlen, void (*callback)(void*, sockaddr*, socklen_t, int), void* obj)
+        {
+                void* mem = m_source.malloc(sizeof(entry)+addrlen);
+                try {
+                        entry* e = new (mem) entry(addr, addrlen, callback, obj);
+                        try {
+                                m_threads[m_rr_seed % m_num_threads]->connect(e);
+                                ++m_rr_seed;
+                        } catch (...) {
+                                e->~entry();
+                        }
+                } catch(...) {
+                        m_source.free(mem);
+                }
+        }
 
-        ~connector() {}  // FIXME
+private:
+        source_t m_source;
+        zone<source> m_zone;
+        connect_thread** m_threads;
+        fdnotify<entry*> m_return_notify;
+        const unsigned int m_num_threads;
+        unsigned int m_rr_seed;
 
 private:
         static scoped_ptr<connector> s_instance;
         static connector& instance();
+
+public:
+        template <typename T>
+        static void connect(sockaddr* addr, socklen_t addrlen, void (*callback)(void*, sockaddr*, socklen_t, int), T* obj)
+        {
+                s_instance->connect_impl(addr, addrlen, callback,
+                                reinterpret_cast<void*>(obj));
+        }
 };
 
@@ -74 +187 @@
 
 
+// FIXME
+scoped_ptr<connector> connector::s_instance;
+
+
 }  // namespace multiplex
 }  // namespace mp

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

@@ -57 +57 @@
 manager::cb_t::cb_t(handler* h, short ev) :
         m_handler(h), m_event(ev),
-        m_wbuffer(NULL), m_allocated(0), m_used(0) {}
+        m_wbuffer((char*)malloc(2048)), m_allocated(2048), m_used(0)
+{
+        if(!m_wbuffer) { throw std::bad_alloc(); }
+}
 
 manager::cb_t::~cb_t()
@@ -98 +101 @@
                 m_used = 0;
         } else {
-                std::memmove(m_wbuffer, m_wbuffer+len, len);
+                std::memmove(m_wbuffer, m_wbuffer+len, m_used-len);
         }
 }
@@ -169 +172 @@
 void manager::send_data_impl(int fd, const char* buf, size_t len)
 {
-        m_ev.data(fd).append_buffer(buf, len);
+        cb_t& cb( m_ev.data(fd) );
+        cb.append_buffer(buf, len);
+        m_wswitch_ctx.push_back(&cb);
 }
 
@@ -196 +201 @@
         while(!m_end_flag) {
                 while( m_ev.next(&fd, &event, &pcb) ) {
-                        for(wswitch_ctx_t::iterator it(m_wswitch_ctx.begin()), it_end(m_wswitch_ctx.end());
-                                        it != it_end;
-                                        ++it) {
-                                if( try_write(*pcb) ) {
-                                        wswitch(*pcb);
-                                }
-                        }
-                        m_wswitch_ctx.clear();
                         if( event & EV_READ ) {
                                 try_read(*pcb);
@@ -211 +208 @@
                         }
                 }
+                for(wswitch_ctx_t::iterator it(m_wswitch_ctx.begin()), it_end(m_wswitch_ctx.end());
+                                it != it_end;
+                                ++it) {
+                        if( try_write(*pcb) ) {
+                                wswitch(*pcb);
+                        }
+                }
+                m_wswitch_ctx.clear();
                 if( (ret = m_ev.wait()) < 0 ) { return ret; }
         }
@@ -229 +234 @@
 bool manager::try_write(cb_t& cb)
 {
+        if(cb.length() == 0) { return false; }
         ssize_t len = ::write(cb.get().fd(), cb.wbuffer(), cb.length());
         if(len < 0) {