Changeset c346b2

Timestamp:

10/05/08 13:49:56 (4 years ago)

Author:

Tomash Brechko <tomash.brechko@…>

Branches:

master

Children:

e61b4e

Parents:

46fd4a

git-author:

Tomash Brechko <tomash.brechko@…> (10/04/08 15:44:14)

git-committer:

Tomash Brechko <tomash.brechko@…> (10/05/08 13:49:56)

Message:

Use poll() instead of select(). Implement it where not available.

Possible select()'able descriptor shortage reported by
Vladimir Timofeev.

Files:

• Changes (modified) (2 diffs)
• MANIFEST (modified) (1 diff)
• src/Makefile.PL (modified) (2 diffs)
• src/client.c (modified) (13 diffs)
• src/connect.c (modified) (4 diffs)
• src/poll_select.c (added)
• src/poll_select.h (added)
• src/socket_posix.c (modified) (1 diff)
• src/socket_posix.h (modified) (1 diff)
• src/socket_win32.h (modified) (1 diff)

Changes

@@ -3 +3 @@
 0.13  ???
         - introduce Win32 support (based on the patch by Yasuhiro
-          Matsumoto---arigatou!).
+          Matsumoto---arigatou!), and use poll() instead of select()
+          (suggested by Vladimir Timofeev).
 
         Changes since 0.12:
@@ -10 +11 @@
         compiler, so I can't even test the build.  Win32 port is
         expected to be supported by community.
+
+        Use poll() system call instead of select().  The latter has
+        the limit on the file descriptor value.  I.e. even when the
+        number of memcached servers is low, if your application opens
+        lots of other files, then after some point socket() returns
+        fd value larger that select() can handle.  poll() doesn't have
+        this limitation.  On a side note, we don't have to use
+        advanced calls like epoll()/kqueue(), because number of
+        memcached servers is normally not very large (and single
+        request touches even a smaller subset).
 
 

MANIFEST

@@ -38 +38 @@
 src/socket_win32.c
 src/socket_win32.h
+src/poll_select.c
+src/poll_select.h

src/Makefile.PL

@@ -5 +5 @@
 
 
+my $includes = '/usr/include';
+my @define;
 my @c = ('parse_keyword.c', 'compute_crc32.c', <*.c>);
-my $exclude;
+my %exclude;
 if ($^O eq 'MSWin32') {
-    $exclude = 'socket_posix.c';
+    ++$exclude{'socket_posix.c'};
 } else {
-    $exclude = 'socket_win32.c';
+    ++$exclude{'socket_win32.c'};
+
+    if (-f "$includes/poll.h") {
+        push @define, '-DHAVE_POLL_H';
+        ++$exclude{'poll_select.c'};
+    } elsif (-f "$includes/sys/poll.h") {
+        push @define, '-DHAVE_SYS_POLL_H';
+        ++$exclude{'poll_select.c'};
+    }
 }
 
-my @object = grep { $_ ne $exclude } @c;
+my @object = grep { not exists $exclude{$_} } @c;
 map { s/\.c$/\$(OBJ_EXT)/ } @object;
 
@@ -24 +34 @@
     AUTHOR            => 'Tomash Brechko <tomash.brechko@gmail.com>',
     LIBS              => [''], # e.g., '-lm'
-    DEFINE            => '', # e.g., '-DHAVE_SOMETHING'
+    DEFINE            => "@define", # e.g., '-DHAVE_SOMETHING'
     INC               => '-I.', # e.g., '-I. -I/usr/include/other'
     OBJECT            => "@object",

src/client.c

@@ -105 +105 @@
   struct client *client;
   int fd;
+  struct pollfd *pollfd;
   enum socket_mode_e socket_mode;
   int noreply;
@@ -249 +250 @@
 struct client
 {
+  struct array pollfds;
   struct array servers;
 
@@ -350 +352 @@
     return NULL;
 
+  array_init(&c->pollfds);
   array_init(&c->servers);
   array_init(&c->index_list);
@@ -394 +397 @@
 
   array_destroy(&c->servers);
+  array_destroy(&c->pollfds);
   array_destroy(&c->index_list);
   array_destroy(&c->str_buf);
@@ -423 +427 @@
 client_set_connect_timeout(struct client *c, int to)
 {
-  c->connect_timeout = to;
+  c->connect_timeout = (to > 0 ? to : -1);
 }
 
@@ -430 +434 @@
 client_set_io_timeout(struct client *c, int to)
 {
-  c->io_timeout = to;
+  c->io_timeout = (to > 0 ? to : -1);
 }
 
@@ -477 +481 @@
 
   if (weight <= 0.0)
+    return MEMCACHED_FAILURE;
+
+  if (array_extend(c->pollfds, struct pollfd, 1, ARRAY_EXTEND_EXACT) == -1)
     return MEMCACHED_FAILURE;
 
@@ -492 +499 @@
     return MEMCACHED_FAILURE;
 
+  array_push(c->pollfds);
   array_push(c->servers);
 
@@ -1380 +1388 @@
 client_execute(struct client *c)
 {
-  struct timeval to, *pto = c->io_timeout > 0 ? &to : NULL;
-  fd_set write_set, read_set;
   int first_iter = 1;
 
@@ -1399 +1405 @@
     {
       struct server *s;
-      int max_fd, res;
-
-      max_fd = -1;
+      struct pollfd *pollfd_beg, *pollfd;
+      int nfds, res;
+
+      nfds = 0;
       for (array_each(c->servers, struct server, s))
         {
@@ -1420 +1427 @@
           else
             {
-              may_write = FD_ISSET(state->fd, &write_set);
-              may_read = FD_ISSET(state->fd, &read_set);
+              const short revents = state->pollfd->revents;
+
+              may_write = revents & (POLLOUT | POLLERR | POLLHUP);
+              may_read = revents & (POLLIN | POLLERR | POLLHUP);
             }
 
@@ -1446 +1455 @@
 
               if (is_active(state))
-                {
-                  if (max_fd < state->fd)
-                    max_fd = state->fd;
-                }
+                ++nfds;
             }
           else
             {
-              if (max_fd < state->fd)
-                max_fd = state->fd;
+              ++nfds;
             }
         }
 
-      if (max_fd == -1)
+      if (nfds == 0)
         break;
 
-      FD_ZERO(&write_set);
-      FD_ZERO(&read_set);
+      pollfd_beg = array_beg(c->pollfds, struct pollfd);
+      pollfd = pollfd_beg;
+
       for (array_each(c->servers, struct server, s))
         {
@@ -1469 +1475 @@
           if (is_active(state))
             {
+              pollfd->events = 0;
+
               if (state->iov_count > 0)
-                FD_SET(state->fd, &write_set);
+                pollfd->events |= POLLOUT;
               if (state->reply_count > 0 || state->nowait_count > 0)
-                FD_SET(state->fd, &read_set);
+                pollfd->events |= POLLIN;
+
+              if (pollfd->events != 0)
+                {
+                  pollfd->fd = state->fd;
+                  state->pollfd = pollfd;
+                  ++pollfd;
+                }
             }
         }
 
       do
-        {
-          /*
-            For maximum portability across systems that may or may not
-            modify the timeout argument we treat it as undefined after
-            the call, and reinitialize on every iteration.
-          */
-          if (pto)
-            {
-              pto->tv_sec = c->io_timeout / 1000;
-              pto->tv_usec = (c->io_timeout % 1000) * 1000;
-            }
-          res = select(max_fd + 1, &read_set, &write_set, NULL, pto);
-        }
+        res = poll(pollfd_beg, pollfd - pollfd_beg, c->io_timeout);
       while (res == -1 && errno == EINTR);
 

src/connect.c

@@ -36 +36 @@
                     int timeout)
 {
-  struct timeval to, *pto;
   struct addrinfo hint, *addr, *a;
   int fd = -1, res;
-
-  pto = timeout > 0 ? &to : NULL;
 
   memset(&hint, 0, sizeof(hint));
@@ -62 +59 @@
   for (a = addr; a != NULL; a = a->ai_next)
     {
-      fd_set write_set;
+      struct pollfd pollfd;
       int socket_error;
       socklen_t socket_error_len;
@@ -69 +66 @@
       if (fd == -1)
         break;
+
+      if (! can_poll_fd(fd))
+        {
+          close(fd);
+          fd = -1;
+          break;
+        }
 
       res = set_nonblock(fd);
@@ -88 +92 @@
         }
 
-      FD_ZERO(&write_set);
-      FD_SET(fd, &write_set);
+      pollfd.fd = fd;
+      pollfd.events = POLLOUT;
       do
-        {
-          if (pto)
-            {
-              pto->tv_sec = timeout / 1000;
-              pto->tv_usec = (timeout % 1000) * 1000;
-            }
-          res = select(fd + 1, NULL, &write_set, NULL, pto);
-        }
+        res = poll(&pollfd, 1, timeout);
       while (res == -1 && errno == EINTR);
       if (res <= 0)

src/socket_posix.c

@@ -58 +58 @@
     return -1;
 
+  if (! can_poll_fd(fd))
+    {
+      close(fd);
+
+      return -1;
+    }
+
   s_unix.sun_family = AF_UNIX;
   memcpy(s_unix.sun_path, path, path_len);

src/socket_posix.h

@@ -31 +31 @@
 #include <errno.h>
 
+#if defined(HAVE_POLL_H)
+
+#include <poll.h>
+
+#define can_poll_fd(fd)  1
+
+#elif defined(HAVE_SYS_POLL_H)
+
+#include <sys/poll.h>
+
+#define can_poll_fd(fd)  1
+
+#else  /* ! defined(HAVE_POLL_H) && ! defined(HAVE_SYS_POLL_H) */
+
+#include "poll_select.h"
+
+#define poll(fds, nfds, timeout)  poll_select(fds, nfds, timeout)
+
+#endif  /* ! defined(HAVE_POLL_H) && ! defined(HAVE_SYS_POLL_H) */
+
 
 extern

src/socket_win32.h

@@ -31 +31 @@
 #include <ws2tcpip.h>
 #include <sys/types.h>
+
+#if _WIN32_WINNT >= 0x0600
+
+#define poll(fds, nfds, timeout)  WSAPoll(fds, nfds, timeout)
+
+#define can_poll_fd(fd)  1
+
+#else  /* ! (_WIN32_WINNT >= 0x0600) */
+
+#include "poll_select.h"
+
+#define poll(fds, nfds, timeout)  poll_select(fds, nfds, timeout)
+
+#endif  /* ! (_WIN32_WINNT >= 0x0600) */