Changeset 18476 for lang/c/mpio/trunk/mp/multiplex/connect.h

Timestamp:

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

Author:

frsyuki

Message:

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

Files:

• lang/c/mpio/trunk/mp/multiplex/connect.h (modified) (3 diffs)

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