Changeset 368

Timestamp:

06/06/07 23:49:50 (2 years ago)

Author:

blackhedd

Message:

initial epoll support

Files:

• version_0/ext/em.cpp (modified) (12 diffs)

version_0/ext/em.cpp

@@ -43 +43 @@
-        NextHeartbeatTime (0),
-        LoopBreakerReader (-1),
-
+
+
+
-{
-        // Default time-slice is just smaller than one hundred mills.
@@ -84 +86 @@
-        close (LoopBreakerReader);
-        close (LoopBreakerWriter);
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-
-
@@ -166 +187 @@
-
-
+/****************************************
+(STATIC) EventMachine_t::SetRlimitNofile
+****************************************/
+
+int EventMachine_t::SetRlimitNofile (int nofiles)
+{
+        struct rlimit rlim;
+        if (nofiles >= 0) {
+                rlim.rlim_cur = nofiles;
+                if (nofiles > rlim.rlim_max)
+                        rlim.rlim_max = nofiles;
+                setrlimit (RLIMIT_NOFILE, &rlim);
+                // ignore the error return, for now at least.
+        }
+        getrlimit (RLIMIT_NOFILE, &rlim);
+        return rlim.rlim_cur;
+}
+
+
-/*********************************
-EventMachine_t::SignalLoopBreaker
@@ -238 +278 @@
-        #ifdef OS_WIN32
-        HookControlC (true);
+        #endif
+
+        #ifdef HAVE_EPOLL
+        if (bEpoll) {
+                epfd = epoll_create (MaxEpollDescriptors);
+                if (epfd == -1) {
+                        char buf[200];
+                        snprintf (buf, sizeof(buf)-1, "unable to create epoll descriptor: %s", strerror(errno));
+                        throw std::runtime_error (buf);
+                }
+                int cloexec = fcntl (epfd, F_GETFD, 0);
+                assert (cloexec >= 0);
+                cloexec |= FD_CLOEXEC;
+                fcntl (epfd, F_SETFD, cloexec);
+        }
-        #endif
-
@@ -257 +312 @@
-
-
-
-/************************
-EventMachine_t::_RunOnce
@@ -263 +317 @@
-
-bool EventMachine_t::_RunOnce()
+{
+        if (bEpoll)
+                return _RunEpollOnce();
+        else
+                return _RunSelectOnce();
+}
+
+
+
+/*****************************
+EventMachine_t::_RunEpollOnce
+*****************************/
+
+bool EventMachine_t::_RunEpollOnce()
+{
+        #ifdef HAVE_EPOLL
+        assert (epfd != -1);
+        struct epoll_event ev [MaxEpollDescriptors];
+        int s = epoll_wait (epfd, ev, MaxEpollDescriptors, 10);
+        if (s > 0) {
+                for (int i=0; i < s; i++) {
+                        EventableDescriptor *ed = (EventableDescriptor*) ev[i].data.ptr;
+                        assert (ed);
+
+                        if (ev[i].events & (EPOLLERR | EPOLLHUP))
+                                ed->ScheduleClose (false);
+                        if (ev[i].events & EPOLLIN)
+                                ed->Read();
+                        if (ev[i].events & EPOLLOUT) {
+                                ed->Write();
+                                int e = epoll_ctl (epfd, EPOLL_CTL_MOD, ed->GetSocket(), ed->GetEpollEvent());
+                        }
+                }
+        }
+        else if (s < 0) {
+                // epoll_wait can fail on error in a handful of ways.
+                // If this happens, then wait for a little while to avoid busy-looping.
+                // If the error was EINTR, we probably caught SIGCHLD or something,
+                // so keep the wait short.
+                timeval tv = {0, ((errno == EINTR) ? 5 : 50) * 1000};
+                EmSelect (0, NULL, NULL, NULL, &tv);
+        }
+
+        { // cleanup dying sockets
+                // vector::pop_back works in constant time.
+                int i, j;
+                int nSockets = Descriptors.size();
+                for (i=0, j=0; i < nSockets; i++) {
+                        EventableDescriptor *ed = Descriptors[i];
+                        assert (ed);
+                        if (ed->ShouldDelete()) {
+                                if (bEpoll) { // well, this will always be true. Otherwise we wouldn't be here.
+                                        assert (epfd != -1);
+                                        int e = epoll_ctl (epfd, EPOLL_CTL_DEL, ed->GetSocket(), ed->GetEpollEvent());
+                                        // ENOENT is not an error because the socket may be already closed when we get here.
+                                        if (e) {
+                                                char buf [200];
+                                                snprintf (buf, sizeof(buf)-1, "unable to delete epoll event: %s", strerror(errno));
+                                                throw std::runtime_error (buf);
+                                        }
+                                }
+
+                                delete ed;
+                        }
+                        else
+                                Descriptors [j++] = ed;
+                }
+                while ((size_t)j < Descriptors.size())
+                        Descriptors.pop_back();
+
+        }
+
+        // TODO, heartbeats.
+
+        timeval tv = {0,0};
+        EmSelect (0, NULL, NULL, NULL, &tv);
+
+        return true;
+        #else
+        throw std::runtime_error ("epoll is not implemented on this platform");
+        #endif
+}
+
+
+/*********************************
+EventMachine_t::_ModifyEpollEvent
+*********************************/
+
+void EventMachine_t::_ModifyEpollEvent (EventableDescriptor *ed)
+{
+        #ifdef HAVE_EPOLL
+        if (bEpoll) {
+                assert (epfd != -1);
+                assert (ed);
+                int e = epoll_ctl (epfd, EPOLL_CTL_MOD, ed->GetSocket(), ed->GetEpollEvent());
+                if (e) {
+                        char buf [200];
+                        snprintf (buf, sizeof(buf)-1, "unable to modify epoll event: %s", strerror(errno));
+                        throw std::runtime_error (buf);
+                }
+        }
+        #endif
+}
+
+
+/******************************
+EventMachine_t::_RunSelectOnce
+******************************/
+
+bool EventMachine_t::_RunSelectOnce()
-{
-        // Crank the event machine once.
@@ -582 +746 @@
-                 * (To wit, the ConnectionCompleted event gets sent to the client.)
-                 */
-
+this, 
-                if (!cd)
-                        throw std::runtime_error ("no connection allocated");
@@ -600 +764 @@
-                        // Put the connection on the stack and wait for it to complete
-                        // or time out.
-
+this, 
-                        if (!cd)
-                                throw std::runtime_error ("no connection allocated");
@@ -620 +784 @@
-                         * for people to know that a failure occurred.
-                         */
-
+this, 
-                        if (!cd)
-                                throw std::runtime_error ("no connection allocated");
@@ -647 +811 @@
-                // Put the connection on the stack and wait for it to complete
-                // or time out.
-
+this, 
-                if (!cd)
-                        throw std::runtime_error ("no connection allocated");
@@ -724 +888 @@
-        // we still set the "pending" flag, so some needed initializations take
-        // place.
-
+this, 
-        if (!cd)
-                throw std::runtime_error ("no connection allocated");
@@ -928 +1092 @@
-                if (ed == NULL)
-                        throw std::runtime_error ("adding bad descriptor");
+
+                #if HAVE_EPOLL
+                if (bEpoll) {
+                        assert (epfd != -1);
+                        int e = epoll_ctl (epfd, EPOLL_CTL_ADD, ed->GetSocket(), ed->GetEpollEvent());
+                        if (e) {
+                                char buf [200];
+                                snprintf (buf, sizeof(buf)-1, "unable to add new descriptor: %s", strerror(errno));
+                                throw std::runtime_error (buf);
+                        }
+                }
+                #endif
+
-                Descriptors.push_back (ed);
-        }