Changeset 21142 for lang/c/mpio/trunk/src/iothreads_reader.cc

Timestamp:

10/11/08 23:06:14 (3 months ago)

Author:

frsyuki

Message:

lang/c/mpio: added mp::iothreads::close()

Files:

• lang/c/mpio/trunk/src/iothreads_reader.cc (modified) (9 diffs)

lang/c/mpio/trunk/src/iothreads_reader.cc

@@ -36 +36 @@
 public:
         void add(handler* h);
+        void close(int fd);
         void send_message(int fd, message_t* msg);
 
@@ -46 +47 @@
 private:
         typedef enum {
-                HANDLER,
+                ADD_HANDLER,
+                CLOSE_HANDLER,
                 MESSAGE,
         } notify_type_t;
@@ -53 +55 @@
                 notify_type_t type;
                 union {
-                        handler* h;
-                        struct {
+                        handler* h;             // ADD_HANDLER
+                        int fd;                 // CLOSE_HANDLER
+                        struct {                // MESSAGE
                                 message_t* func;
                                 int fd;
@@ -65 +68 @@
 
 private:
+        void add_handler(handler* h);
+        void close_handler(handler* h);
+
+private:
         typedef event<handler*> ev_t;
         ev_t m_ev;
@@ -105 +112 @@
 }
 
+void reader::impl::close(int fd)
+{
+        worker_of(fd).close(fd);
+}
+
 void reader::impl::send_message(int fd, message_t* msg)
 {
@@ -126 +138 @@
 inline void reader::impl::worker::add(handler* h)
 {
-        notify_entry e = { HANDLER };
+        notify_entry e = { ADD_HANDLER };
         e.as.h = h;
+        m_notify.send(e);
+}
+
+inline void reader::impl::worker::close(int fd)
+{
+        notify_entry e = { CLOSE_HANDLER };
+        e.as.fd = fd;
         m_notify.send(e);
 }
@@ -179 +198 @@
                 h->read_event();
         } catch (...) {
-                int fd = h->fd();
-                m_ev.remove(fd, EV_READ);
-                // FIXME closeはロジックスレッドで
-                // fdに対してイベント到着
-                // 直後にclose
-                // 直後にaccept
-                // イベントに対して返信
-                // 期待とは異なるfdに返信される
-                //::close(fd);
-                try {
-                        iothreads::submit(close_delete, fd, h);
-                } catch (...) {
-                        ::close(fd);
-                        delete h;
-                }
+                close_handler(h);
         }
 }
@@ -199 +204 @@
 inline void reader::impl::worker::notify_impl(notify_entry& e)
 {
-        if(e.type == HANDLER) {
+        if(e.type == ADD_HANDLER) {
                 handler* h = e.as.h;
-                try {
-                        m_ev.add(h->fd(), EV_READ, h);
-                } catch (...) {
-                        ::close(h->fd());
-                        delete h;
-                }
-
-        } else {
+                add_handler(h);
+
+        } else if(e.type == CLOSE_HANDLER) {
+                int fd = e.as.fd;
+                if(!m_ev.test(fd)) { return; }
+                close_handler(m_ev.data(fd));
+
+        } else {  // e.type == MESSAGE
                 int fd = e.as.m.fd;
                 if(!m_ev.test(fd)) { return; }
                 message_t* func = e.as.m.func;
                 try {
-                        // FIXME check m_ev.test(fd);
                         (*func)(*m_ev.data(fd));
                 } catch (...) {
@@ -223 +227 @@
 }
 
+inline void reader::impl::worker::add_handler(handler* h)
+{
+        try {
+                m_ev.add(h->fd(), EV_READ, h);
+        } catch (...) {
+                ::close(h->fd());
+                delete h;
+        }
+}
+
+inline void reader::impl::worker::close_handler(handler* h)
+{
+        m_ev.remove(h->fd(), EV_READ);
+        //::close(fd);  // FIXME closeはロジックスレッドで
+        // 1. fdに対してイベント到着
+        // 2. 直後にclose
+        // 3. 直後にaccept, closeしたfdと同じ番号になり得る
+        // 4. イベントに対して返信するためにsend_data
+        // 5. 期待とは異なるfdに返信されてしまう
+        try {
+                iothreads::submit(close_delete, h->fd(), h);
+        } catch (...) {
+                ::close(h->fd());
+                delete h;
+        }
+}
+
 
 }  // namespace iothreads