--- live/BasicUsageEnvironment/BasicTaskScheduler.cpp +++ live/BasicUsageEnvironment/BasicTaskScheduler.cpp @@ -25,6 +25,22 @@ #include #include #endif +// Use poll() rather than select(), because an fd_set is a fixed-size bitmap indexed by descriptor +// number, and so cannot represent a socket >= FD_SETSIZE at all. Windows keeps select(), where an +// fd_set is a counted array of SOCKETs with no such limit; VxWorks and QNX4 keep it because neither +// was available to test on. Define NO_POLL to force select() anywhere else. +#if !defined(__WIN32__) && !defined(_WIN32) && !defined(_WIN32_WCE) && !defined(VXWORKS) && !defined(_QNX4) && !defined(NO_POLL) +#define BASIC_TASK_SCHEDULER_USE_POLL 1 +#include +// Almost every scheduler watches only a handful of sockets, so the array is normally on the stack. +#define BASIC_TASK_SCHEDULER_POLL_FDS_ON_STACK 64 +// ppoll() keeps the microsecond resolution that select() had, where poll() rounds the delay up to a +// whole millisecond. It is a GNU extension, and _GNU_SOURCE is what makes glibc declare it (C++ +// compilers predefine that on Linux). Everywhere else, poll() and its coarser timeout. +#if defined(__linux__) && defined(_GNU_SOURCE) +#define BASIC_TASK_SCHEDULER_USE_PPOLL 1 +#endif +#endif ////////// BasicTaskScheduler ////////// @@ -63,10 +79,67 @@ #define MILLION 1000000 #endif +#ifdef BASIC_TASK_SCHEDULER_USE_POLL +// Return the condition set that "poll()" reported for "sock", following what "select()" reported: a +// peer close or error appears in "readfds", an error also appears in "writefds", and "exceptfds" +// means out-of-band data only. POLLHUP is also taken as writable, which "select()" did not do: +// POSIX makes POLLHUP and POLLOUT mutually exclusive, so without it a hung-up socket registered only +// for SOCKET_WRITABLE would never get a callback, and so could never be deregistered. "hint" is the +// index this socket is expected at; the array is filled by walking "fHandlers" in the same order the +// caller walks it, so the guess is almost always right and the array isn't rescanned. +static int pollResultFor(int sock, int conditionSet, + struct pollfd const* pollFds, unsigned numPollFds, unsigned& hint) { + if (numPollFds == 0) return 0; + for (unsigned n = 0; n < numPollFds; ++n) { + unsigned const i = (hint + n)%numPollFds; + if (pollFds[i].fd != sock) continue; + hint = i + 1; + short const revents = pollFds[i].revents; + int result = 0; + if ((revents&(POLLIN|POLLHUP|POLLERR)) != 0 && (conditionSet&SOCKET_READABLE) != 0) result |= SOCKET_READABLE; + if ((revents&(POLLOUT|POLLERR|POLLHUP)) != 0 && (conditionSet&SOCKET_WRITABLE) != 0) result |= SOCKET_WRITABLE; + if ((revents&POLLPRI) != 0 && (conditionSet&SOCKET_EXCEPTION) != 0) result |= SOCKET_EXCEPTION; + return result; + } + return 0; +} +#endif + void BasicTaskScheduler::SingleStep(unsigned maxDelayTime) { +#ifdef BASIC_TASK_SCHEDULER_USE_POLL + // Build the "poll()" array from "fHandlers", which already records what each socket waits for: + unsigned numHandlers = 0; + { + HandlerIterator countIter(*fHandlers); + while (countIter.next() != NULL) ++numHandlers; + } + struct pollfd pollFdsBuf[BASIC_TASK_SCHEDULER_POLL_FDS_ON_STACK]; + struct pollfd* pollFdsHeap = NULL; + struct pollfd* pollFds = pollFdsBuf; + if (numHandlers > BASIC_TASK_SCHEDULER_POLL_FDS_ON_STACK) { + pollFdsHeap = new struct pollfd[numHandlers]; + pollFds = pollFdsHeap; + } + unsigned numPollFds = 0; + { + HandlerIterator fillIter(*fHandlers); + HandlerDescriptor* h; + while ((h = fillIter.next()) != NULL && numPollFds < numHandlers) { + pollFds[numPollFds].fd = h->socketNum; + pollFds[numPollFds].events = 0; + pollFds[numPollFds].revents = 0; + if (h->conditionSet&SOCKET_READABLE) pollFds[numPollFds].events |= POLLIN; + if (h->conditionSet&SOCKET_WRITABLE) pollFds[numPollFds].events |= POLLOUT; + if (h->conditionSet&SOCKET_EXCEPTION) pollFds[numPollFds].events |= POLLPRI; + ++numPollFds; + } + } + unsigned pollHint = 0; +#else fd_set readSet = fReadSet; // make a copy for this select() call fd_set writeSet = fWriteSet; // ditto fd_set exceptionSet = fExceptionSet; // ditto +#endif DelayInterval const& timeToDelay = fDelayQueue.timeToNextAlarm(); struct timeval tv_timeToDelay; @@ -87,7 +160,28 @@ tv_timeToDelay.tv_usec = maxDelayTime%MILLION; } +#ifdef BASIC_TASK_SCHEDULER_USE_POLL +#ifdef BASIC_TASK_SCHEDULER_USE_PPOLL + struct timespec ts; + ts.tv_sec = tv_timeToDelay.tv_sec; + ts.tv_nsec = tv_timeToDelay.tv_usec*1000; + int selectResult = ppoll(pollFds, numPollFds, &ts, NULL); +#else + int const timeoutMillis + = (int)(tv_timeToDelay.tv_sec*1000 + (tv_timeToDelay.tv_usec + 999)/1000); // rounded up + int selectResult = poll(pollFds, numPollFds, timeoutMillis); +#endif + // A descriptor closed without being deregistered is reported as POLLNVAL, and "poll()" still + // succeeds. "select()" instead failed the whole call with EBADF, so do the same here; otherwise + // nothing consumes POLLNVAL and the event loop spins. + if (selectResult > 0) { + for (unsigned i = 0; i < numPollFds; ++i) { + if (pollFds[i].revents&POLLNVAL) { selectResult = -1; errno = EBADF; break; } + } + } +#else int selectResult = select(fMaxNumSockets, &readSet, &writeSet, &exceptionSet, &tv_timeToDelay); +#endif if (selectResult < 0) { #if defined(__WIN32__) || defined(_WIN32) int err = WSAGetLastError(); @@ -106,6 +200,22 @@ #endif // Unexpected error - treat this as fatal: #if !defined(_WIN32_WCE) +#ifdef BASIC_TASK_SCHEDULER_USE_POLL + perror("BasicTaskScheduler::SingleStep(): poll() fails"); + // Because this failure is often "Bad file descriptor" - which is caused by an invalid socket number (i.e., a socket number + // that had already been closed) being used in "poll()" - we print out the sockets that were being used in "poll()", + // to assist in debugging: + fprintf(stderr, "socket numbers used in the poll() call:"); + for (unsigned i = 0; i < numPollFds; ++i) { + fprintf(stderr, " %d(", pollFds[i].fd); + if (pollFds[i].events&POLLIN) fprintf(stderr, "r"); + if (pollFds[i].events&POLLOUT) fprintf(stderr, "w"); + if (pollFds[i].events&POLLPRI) fprintf(stderr, "e"); + if (pollFds[i].revents&POLLNVAL) fprintf(stderr, "n"); // this is the invalid one + fprintf(stderr, ")"); + } + fprintf(stderr, "\n"); +#else perror("BasicTaskScheduler::SingleStep(): select() fails"); // Because this failure is often "Bad file descriptor" - which is caused by an invalid socket number (i.e., a socket number // that had already been closed) being used in "select()" - we print out the sockets that were being used in "select()", @@ -122,6 +232,7 @@ } fprintf(stderr, "\n"); #endif +#endif internalError(); } } @@ -143,9 +254,13 @@ while ((handler = iter.next()) != NULL) { int sock = handler->socketNum; // alias int resultConditionSet = 0; +#ifdef BASIC_TASK_SCHEDULER_USE_POLL + resultConditionSet = pollResultFor(sock, handler->conditionSet, pollFds, numPollFds, pollHint); +#else if (FD_ISSET(sock, &readSet) && FD_ISSET(sock, &fReadSet)/*sanity check*/) resultConditionSet |= SOCKET_READABLE; if (FD_ISSET(sock, &writeSet) && FD_ISSET(sock, &fWriteSet)/*sanity check*/) resultConditionSet |= SOCKET_WRITABLE; if (FD_ISSET(sock, &exceptionSet) && FD_ISSET(sock, &fExceptionSet)/*sanity check*/) resultConditionSet |= SOCKET_EXCEPTION; +#endif if ((resultConditionSet&handler->conditionSet) != 0 && handler->handlerProc != NULL) { fLastHandledSocketNum = sock; // Note: we set "fLastHandledSocketNum" before calling the handler, @@ -158,12 +273,19 @@ // We didn't call a handler, but we didn't get to check all of them, // so try again from the beginning: iter.reset(); +#ifdef BASIC_TASK_SCHEDULER_USE_POLL + pollHint = 0; // this loop restarts from the beginning of "fHandlers" +#endif while ((handler = iter.next()) != NULL) { int sock = handler->socketNum; // alias int resultConditionSet = 0; +#ifdef BASIC_TASK_SCHEDULER_USE_POLL + resultConditionSet = pollResultFor(sock, handler->conditionSet, pollFds, numPollFds, pollHint); +#else if (FD_ISSET(sock, &readSet) && FD_ISSET(sock, &fReadSet)/*sanity check*/) resultConditionSet |= SOCKET_READABLE; if (FD_ISSET(sock, &writeSet) && FD_ISSET(sock, &fWriteSet)/*sanity check*/) resultConditionSet |= SOCKET_WRITABLE; if (FD_ISSET(sock, &exceptionSet) && FD_ISSET(sock, &fExceptionSet)/*sanity check*/) resultConditionSet |= SOCKET_EXCEPTION; +#endif if ((resultConditionSet&handler->conditionSet) != 0 && handler->handlerProc != NULL) { fLastHandledSocketNum = sock; // Note: we set "fLastHandledSocketNum" before calling the handler, @@ -207,17 +329,23 @@ // Also handle any delayed event that may have come due. fDelayQueue.handleAlarm(); + +#ifdef BASIC_TASK_SCHEDULER_USE_POLL + delete[] pollFdsHeap; +#endif } void BasicTaskScheduler ::setBackgroundHandling(int socketNum, int conditionSet, BackgroundHandlerProc* handlerProc, void* clientData) { if (socketNum < 0) return; +#ifndef BASIC_TASK_SCHEDULER_USE_POLL #if !defined(__WIN32__) && !defined(_WIN32) && defined(FD_SETSIZE) if (socketNum >= (int)(FD_SETSIZE)) return; #endif FD_CLR((unsigned)socketNum, &fReadSet); FD_CLR((unsigned)socketNum, &fWriteSet); FD_CLR((unsigned)socketNum, &fExceptionSet); +#endif if (conditionSet == 0) { fHandlers->clearHandler(socketNum); if (socketNum+1 == fMaxNumSockets) { @@ -228,20 +356,24 @@ if (socketNum+1 > fMaxNumSockets) { fMaxNumSockets = socketNum+1; } +#ifndef BASIC_TASK_SCHEDULER_USE_POLL if (conditionSet&SOCKET_READABLE) FD_SET((unsigned)socketNum, &fReadSet); if (conditionSet&SOCKET_WRITABLE) FD_SET((unsigned)socketNum, &fWriteSet); if (conditionSet&SOCKET_EXCEPTION) FD_SET((unsigned)socketNum, &fExceptionSet); +#endif } } void BasicTaskScheduler::moveSocketHandling(int oldSocketNum, int newSocketNum) { if (oldSocketNum < 0 || newSocketNum < 0) return; // sanity check +#ifndef BASIC_TASK_SCHEDULER_USE_POLL #if !defined(__WIN32__) && !defined(_WIN32) && defined(FD_SETSIZE) if (oldSocketNum >= (int)(FD_SETSIZE) || newSocketNum >= (int)(FD_SETSIZE)) return; // sanity check #endif if (FD_ISSET(oldSocketNum, &fReadSet)) {FD_CLR((unsigned)oldSocketNum, &fReadSet); FD_SET((unsigned)newSocketNum, &fReadSet);} if (FD_ISSET(oldSocketNum, &fWriteSet)) {FD_CLR((unsigned)oldSocketNum, &fWriteSet); FD_SET((unsigned)newSocketNum, &fWriteSet);} if (FD_ISSET(oldSocketNum, &fExceptionSet)) {FD_CLR((unsigned)oldSocketNum, &fExceptionSet); FD_SET((unsigned)newSocketNum, &fExceptionSet);} +#endif fHandlers->moveHandler(oldSocketNum, newSocketNum); if (oldSocketNum+1 == fMaxNumSockets) {