Changeset 1561

Timestamp:

02/01/2007 12:17:22 AM (19 months ago)

Author:

jan

Message:

big threading cleanup, glib-2.0 + gthread-2.0 is now a requirement

* configure.in, src/CMakeLists.txt

• removed FAM support
• added checks for glib-2.0

* src/Makefile.am

• added mod_magnet to the build

* src/stat_cache.c, src/response.c

• removed FAM support
• added threaded stat()
• dropped splaytree in favour of GHashTable

* src/server.c, src/network_gthread_aio.c, src/network.c,

src/network_backends.h

• added new, gthread based async write()

* src/chunk.c, src/chunk.h

• added vars for async-write to the chunk-struct

* src/network_linux_aio.c, src/network_posix_aio.c,

src/configfile.c

• made number of threads configurable through server.max-write-threads

Location:

trunk

Files:

• configure.in (modified) (3 diffs)
• src/CMakeLists.txt (modified) (6 diffs)
• src/Makefile.am (modified) (4 diffs)
• src/base.h (modified) (5 diffs)
• src/chunk.c (modified) (2 diffs)
• src/chunk.h (modified) (1 diff)
• src/configfile.c (modified) (2 diffs)
• src/network.c (modified) (3 diffs)
• src/network_backends.h (modified) (1 diff)
• src/network_gthread_aio.c (copied) (copied from trunk/src/network_posix_aio.c) (3 diffs)
• src/network_linux_aio.c (modified) (2 diffs)
• src/network_posix_aio.c (modified) (2 diffs)
• src/response.c (modified) (2 diffs)
• src/server.c (modified) (14 diffs)
• src/stat_cache.c (modified) (18 diffs)

trunk/configure.in

@@ -300 +300 @@
 fi
 
-dnl Check for gamin
-AC_MSG_CHECKING(for FAM)
-AC_ARG_WITH(fam, AC_HELP_STRING([--with-fam],[fam/gamin for reducing number of stat() calls]),
-[WITH_FAM=$withval],[WITH_FAM=no])
-AC_MSG_RESULT([$WITH_FAM])
-
-if test "$WITH_FAM" != "no"; then
-  AC_CHECK_LIB(fam, FAMOpen2, [
-    AC_CHECK_HEADERS([fam.h],[
-      FAM_LIBS=-lfam
-      AC_DEFINE([HAVE_LIBFAM], [1], [libfam])
-      AC_DEFINE([HAVE_FAM_H], [1], [fam.h])
-    ])
-  ])
-  if test "x$FAM_LIBS" = x; then
-    PKG_CHECK_MODULES(FAM, gamin >= 0.1.0, [
-      AC_DEFINE([HAVE_LIBFAM], [1], [libfam])
-      AC_DEFINE([HAVE_FAM_H], [1], [fam.h])
-    ])
-  fi
-  OLD_LIBS=$LIBS
-  LIBS=$FAM_LIBS
-  AC_CHECK_FUNCS([FAMNoExists])
-  LIBS=$OLD_LIBS
-fi
-
 AC_MSG_CHECKING(for properties in mod_webdav)
 AC_ARG_WITH(webdav-props, AC_HELP_STRING([--with-webdav-props],[properties in mod_webdav]),
@@ -407 +381 @@
  AC_SUBST(LUA_LIBS)
 fi
+
+PKG_CHECK_MODULES(GTHREAD, gthread-2.0, [
+  AC_DEFINE([HAVE_GTHREAD], [1], [libglib])
+  AC_DEFINE([HAVE_GLIB_H], [1], [glib.h])
+])
 
 dnl Check for libaio
@@ -665 +644 @@
 fi
 
-features="stat-cache-fam"
-if test ! "x$FAM_LIBS" = x; then
-        enable_feature="$enable_feature $features"
-else
-        disable_feature="$disable_feature $features"
-fi
-
 features="webdav-properties"
 if test "x$XML_LIBS" \!= x -a "x$SQLITE_LIBS" \!= x; then

trunk/src/CMakeLists.txt

@@ -19 +19 @@
 OPTION(WITH_BZIP "with bzip2-support for mod_compress [default: off]")
 OPTION(WITH_ZLIB "with deflate-support for mod_compress [default: on]" ON)
-OPTION(WITH_FAM "with FAM-support for the stat-cache [default: off]")
 OPTION(WITH_LDAP "with LDAP-support for the mod_auth [default: off]")
 OPTION(WITH_LIBAIO "with libaio for the linux [default: off]")
@@ -56 +55 @@
 CHECK_INCLUDE_FILES(time.h HAVE_TIME_H)
 
+INCLUDE_DIRECTORIES(/opt/gnome/include/glib-2.0 /opt/gnome/lib64/glib-2.0/include/)
+
 IF(WITH_XATTR)
   CHECK_INCLUDE_FILES(attr/attributes.h HAVE_ATTR_ATTRIBUTES_H)
@@ -85 +86 @@
   CHECK_LIBRARY_EXISTS(bz2 BZ2_bzCompressInit "" HAVE_LIBBZ2)
 ENDIF(WITH_BZIP)
-IF(WITH_FAM)
-  CHECK_INCLUDE_FILES(fam.h HAVE_FAM_H)
-ENDIF(WITH_FAM)
+
 CHECK_INCLUDE_FILES(getopt.h HAVE_GETOPT_H)
 CHECK_INCLUDE_FILES(inttypes.h HAVE_INTTYPES_H)
@@ -231 +230 @@
       network_linux_aio.c
       network_posix_aio.c
+      network_gthread_aio.c
       http_resp.c
       http_resp_parser.c
@@ -370 +370 @@
 ENDIF(CMAKE_SYSTEM MATCHES "Linux")
 
+TARGET_LINK_LIBRARIES(lighttpd glib-2.0 gthread-2.0)
+SET_TARGET_PROPERTIES(lighttpd PROPERTIES LINK_FLAGS "-L/opt/gnome/lib64/ -pthread -lrt")
 SET_TARGET_PROPERTIES(lighttpd PROPERTIES CMAKE_INSTALL_PREFIX ${CMAKE_INSTALL_PREFIX})
 
@@ -383 +385 @@
   TARGET_LINK_LIBRARIES(lighttpd pcre)
 ENDIF(HAVE_PCRE_H)
-
-IF(HAVE_FAM_H)
-  TARGET_LINK_LIBRARIES(lighttpd fam)
-ENDIF(HAVE_FAM_H)
 
 IF(HAVE_LIBSSL AND HAVE_LIBCRYPTO)

trunk/src/Makefile.am

@@ -1 @@
-AM_CFLAGS = $(FAM_CFLAGS)
+AM_CFLAGS = $(GTHREAD_CFLAGS)
 
 noinst_PROGRAMS=proc_open lemon # simple-fcgi 
@@ -71 +71 @@
       network_linux_aio.c \
       network_posix_aio.c \
+      network_gthread_aio.c \
       splaytree.c \
       http_resp.c http_resp_parser.c \
@@ -124 +125 @@
 mod_webdav_la_LIBADD = $(common_libadd) $(XML_LIBS) $(SQLITE_LIBS) $(UUID_LIB)
 
-#lib_LTLIBRARIES += mod_cml.la
-#mod_cml_la_SOURCES = mod_cml.c mod_cml_lua.c mod_cml_funcs.c
-#mod_cml_la_CFLAGS = $(AM_CFLAGS) $(LUA_CFLAGS)
-#mod_cml_la_LDFLAGS = -module -export-dynamic -avoid-version -no-undefined
-#mod_cml_la_LIBADD = $(MEMCACHE_LIB) $(common_libadd) $(LUA_LIBS) -lm
+lib_LTLIBRARIES += mod_magnet.la
+mod_magnet_la_SOURCES = mod_magnet.c mod_magnet_cache.c
+mod_magnet_la_CFLAGS = $(AM_CFLAGS) $(LUA_CFLAGS)
+mod_magnet_la_LDFLAGS = -module -export-dynamic -avoid-version -no-undefined
+mod_magnet_la_LIBADD = $(MEMCACHE_LIB) $(common_libadd) $(LUA_LIBS) -lm
 
 lib_LTLIBRARIES += mod_trigger_b4_dl.la
@@ -320 +321 @@
       http_parser.h \
       ajp13.h \
-      mod_proxy_core_protocol.h
+      mod_proxy_core_protocol.h \
+      mod_magnet_cache.h
 
 DEFS= @DEFS@ -DLIBRARY_DIR="\"$(libdir)\""
 
 lighttpd_SOURCES = $(src)
-lighttpd_LDADD = $(PCRE_LIB) $(DL_LIB) $(SENDFILE_LIB) $(ATTR_LIB) $(common_libadd) $(SSL_LIB) $(FAM_LIBS) $(AIO_LIB) $(POSIX_AIO_LIB)
+lighttpd_LDADD = $(PCRE_LIB) $(DL_LIB) $(SENDFILE_LIB) $(ATTR_LIB) $(common_libadd) $(SSL_LIB) $(FAM_LIBS) $(AIO_LIB) $(POSIX_AIO_LIB) $(GTHREAD_LIBS)
 lighttpd_LDFLAGS = -export-dynamic
 lighttpd_CCPFLAGS = $(FAM_CFLAGS)

trunk/src/base.h

@@ -49 +49 @@
 #endif
 
+#include <glib.h>
+
 #ifndef O_BINARY
 # define O_BINARY 0
@@ -192 +194 @@
         time_t stat_ts;
 
+        enum {
+                STAT_CACHE_ENTRY_UNSET, 
+                STAT_CACHE_ENTRY_ASYNC_STAT, 
+                STAT_CACHE_ENTRY_STAT_FINISHED
+        } state;
+
 #ifdef HAVE_LSTAT
         char is_symlink;
 #endif
 
-#if defined(HAVE_FAM_H) || defined(HAVE_SYS_INOTIFY_H)
-        int    dir_version;
+#if defined(HAVE_SYS_INOTIFY_H)
+        int    dir_version; /* when this entry was created the dir had this version */
         int    dir_ndx;
 #endif
@@ -205 +213 @@
 
 typedef struct {
-        splay_tree *files; /* the nodes of the tree are stat_cache_entries */
-
-        buffer *dir_name; /* for building the dirname from the filename */
-        buffer *hash_key;
-
-#if defined(HAVE_FAM_H) || defined(HAVE_SYS_INOTIFY_H)
-        splay_tree *dirs; /* the nodes of the tree are fam_dir_entry */
-
-        iosocket *sock;
-#endif
-#if defined(HAVE_FAM_H)
-        FAMConnection *fam;
+        GHashTable *files; /* a HashTable of stat_cache_entries for the files */
+        GHashTable *dirs;  /* a HashTable of stat_cache_entries for the dirs */
+
+        buffer *dir_name;  /* for building the dirname from the filename */
+        buffer *hash_key;  /* tmp-buf for building the hash-key */
+
+#if defined(HAVE_SYS_INOTIFY_H)
+        iosocket *sock;    /* socket to the inotify fd (this should be in a backend struct */
 #endif
 } stat_cache;
@@ -481 +485 @@
         buffer *errorlog_file;
         unsigned short errorlog_use_syslog;
+
+        unsigned short max_stat_threads;
+        unsigned short max_write_threads;
 } server_config;
 
@@ -594 +601 @@
 
 #ifdef HAVE_LIBAIO_H
-#define LINUX_IO_MAX_IOCBS 128
         io_context_t linux_io_ctx;
 
-        struct iocb linux_io_iocbs[LINUX_IO_MAX_IOCBS];
+        struct iocb *linux_io_iocbs;
 
 #endif
 #ifdef HAVE_AIO_H
-#define POSIX_AIO_MAX_IOCBS 128
-        struct aiocb posix_aio_iocbs[POSIX_AIO_MAX_IOCBS];
-        struct aiocb * posix_aio_iocbs_watch[POSIX_AIO_MAX_IOCBS];
-
-        void *posix_aio_data[POSIX_AIO_MAX_IOCBS];
-#endif
-
+        struct aiocb *posix_aio_iocbs;
+        struct aiocb ** posix_aio_iocbs_watch;
+
+        void **posix_aio_data;
+#endif
+
+        GAsyncQueue *stat_queue; /* send a stat_job into this queue and joblist_queue will get a wakeup when the stat is finished */
+        GAsyncQueue *joblist_queue;
+        GAsyncQueue *aio_write_queue;
 } server;
 

trunk/src/chunk.c

@@ -45 +45 @@
         c->file.mmap.start = MAP_FAILED;
         c->next = NULL;
+        
+        c->async.written = -1;
 
         return c;
@@ -86 +88 @@
         c->file.copy.length = 0;
         c->file.copy.offset = 0;
+
+        c->async.written = -1;
+        c->async.ret_val = 0;
 }
 

trunk/src/chunk.h

@@ -37 +37 @@
                           - file-chunk: file.length
                         */
+
+        struct {
+                off_t written;
+                int ret_val;
+        } async;
 
         struct chunk *next;

trunk/src/configfile.c

@@ -92 +92 @@
                 { "debug.log-condition-cache-handling", NULL, T_CONFIG_BOOLEAN, T_CONFIG_SCOPE_SERVER },    /* 46 */
                 { "server.use-noatime",          NULL, T_CONFIG_BOOLEAN, T_CONFIG_SCOPE_CONNECTION }, /* 47 */
+                { "server.max-stat-threads",     NULL, T_CONFIG_SHORT, T_CONFIG_SCOPE_CONNECTION },   /* 48 */
+                { "server.max-write-threads",    NULL, T_CONFIG_SHORT, T_CONFIG_SCOPE_CONNECTION },   /* 49 */
 
                 { "server.host",                 "use server.bind instead", T_CONFIG_DEPRECATED, T_CONFIG_SCOPE_UNSET },
@@ -135 +137 @@
         cv[12].destination = &(srv->srvconf.max_request_size);
         cv[47].destination = &(srv->srvconf.use_noatime);
+        cv[48].destination = &(srv->srvconf.max_stat_threads);
+        cv[49].destination = &(srv->srvconf.max_write_threads);
 
         srv->config_storage = calloc(1, srv->config_context->used * sizeof(specific_config *));

trunk/src/network.c

@@ -446 +446 @@
         NETWORK_BACKEND_LINUX_AIO_SENDFILE,
         NETWORK_BACKEND_POSIX_AIO,
+        NETWORK_BACKEND_GTHREAD_AIO,
 
         NETWORK_BACKEND_FREEBSD_SENDFILE,
@@ -471 +472 @@
                 { NETWORK_BACKEND_LINUX_AIO_SENDFILE,   "linux-aio-sendfile" },
 #endif
+#if defined USE_FREEBSD_SENDFILE
+                { NETWORK_BACKEND_FREEBSD_SENDFILE,     "freebsd-sendfile" },
+#endif
+#if defined USE_SOLARIS_SENDFILEV
+                { NETWORK_BACKEND_SOLARIS_SENDFILEV,    "solaris-sendfilev" },
+#endif
 #if defined USE_POSIX_AIO
                 { NETWORK_BACKEND_POSIX_AIO,            "posix-aio" },
 #endif
-
-#if defined USE_FREEBSD_SENDFILE
-                { NETWORK_BACKEND_FREEBSD_SENDFILE,     "freebsd-sendfile" },
-#endif
-#if defined USE_SOLARIS_SENDFILEV
-                { NETWORK_BACKEND_SOLARIS_SENDFILEV,    "solaris-sendfilev" },
-#endif
+                { NETWORK_BACKEND_GTHREAD_AIO,          "gthread-aio" },
 #if defined USE_WRITEV
                 { NETWORK_BACKEND_WRITEV,               "writev" },
@@ -581 +582 @@
                 break;
 #endif
-
+        case NETWORK_BACKEND_GTHREAD_AIO:
+                SET_NETWORK_BACKEND(read, gthreadaio);
+                break;
 #ifdef USE_FREEBSD_SENDFILE
         case NETWORK_BACKEND_FREEBSD_SENDFILE:

trunk/src/network_backends.h

@@ -83 +83 @@
 NETWORK_BACKEND_WRITE(linuxaiosendfile);
 NETWORK_BACKEND_WRITE(posixaio);
+NETWORK_BACKEND_WRITE(gthreadaio);
 NETWORK_BACKEND_WRITE(freebsdsendfile);
 NETWORK_BACKEND_WRITE(solarissendfilev);

trunk/src/network_gthread_aio.c

@@ -37 +37 @@
 #include "sys-files.h"
 
-NETWORK_BACKEND_WRITE(posixaio) {
+typedef struct {
+        chunk *c;
+
+        void *con;
+
+        int sock_fd;
+} write_job;
+
+write_job *write_job_init() {
+        write_job *wj = calloc(1, sizeof(*wj));
+
+        return wj;
+}
+
+void write_job_free(write_job *wj) {
+        if (!wj) return;
+
+        free(wj);
+}
+
+gpointer aio_write_thread(gpointer _srv) {
+        server *srv = (server *)_srv;
+        write_job *wj = NULL;
+
+        /* take the stat-job-queue */
+        GAsyncQueue * inq = g_async_queue_ref(srv->aio_write_queue);
+        GAsyncQueue * outq = g_async_queue_ref(srv->joblist_queue);
+
+        /* */
+        while ((wj = g_async_queue_pop(inq))) {
+                /* let's see what we have to stat */
+                ssize_t r;
+                off_t offset;
+                size_t toSend;
+                const off_t max_toSend = 2 * 256 * 1024; /** should be larger than the send buffer */
+                chunk *c = wj->c;
+                int mmap_fd;
+                        
+                offset = c->file.start + c->offset;
+
+                toSend = c->file.length - c->offset > max_toSend ?
+                        max_toSend : c->file.length - c->offset;
+
+                c->file.copy.offset = 0;
+                c->file.copy.length = toSend;
+
+                /* open a file in /dev/shm to write to */
+                if (c->file.mmap.start == MAP_FAILED) {
+                        if (-1 == (mmap_fd = open("/dev/zero", O_RDWR))) {
+                                if (errno != EMFILE) {
+                                        TRACE("open(/dev/zero) returned: %d (%s), open fds: %d",
+                                                errno, strerror(errno), srv->cur_fds);
+                                        c->async.ret_val = NETWORK_STATUS_FATAL_ERROR;
+                                } else {
+                                        c->async.ret_val = NETWORK_STATUS_WAIT_FOR_FD;
+                                }
+                        } else {
+                                c->file.mmap.offset = 0;
+                                c->file.mmap.length = c->file.copy.length; /* align to page-size */
+        
+                                c->file.mmap.start = mmap(0, c->file.mmap.length,
+                                                PROT_READ | PROT_WRITE, MAP_SHARED, mmap_fd, 0);
+                                if (c->file.mmap.start == MAP_FAILED) {
+                                        c->async.ret_val = NETWORK_STATUS_FATAL_ERROR;
+                                }
+        
+                                close(mmap_fd);
+                                mmap_fd = -1;
+                        }
+                }
+                
+                if (c->file.mmap.start != MAP_FAILED) {
+                        lseek(c->file.fd, c->file.start + c->offset, SEEK_SET);
+
+                        if (-1 == (r = read(c->file.fd, c->file.mmap.start, c->file.copy.length))) {
+                                switch(errno) {
+                                default:
+                                        ERROR("reading file failed: %d (%s)", errno, strerror(errno));
+
+                                        c->async.ret_val = NETWORK_STATUS_FATAL_ERROR;
+                                }
+                        } else if (r == 0) {
+                                ERROR("read() returned 0 ... not good: %s", "");
+
+                                c->async.ret_val = NETWORK_STATUS_FATAL_ERROR;
+                        } else if (r != c->file.copy.length) {
+                                ERROR("read() returned %d instead of %d", r, c->file.copy.length);
+
+                                c->async.ret_val = NETWORK_STATUS_FATAL_ERROR;
+                        }
+                }
+
+                /* read async, write as usual */ 
+                g_async_queue_push(outq, wj->con);
+        
+                write_job_free(wj);
+        }
+        
+        g_async_queue_unref(srv->aio_write_queue);
+        g_async_queue_unref(srv->joblist_queue);
+
+        return NULL;
+
+}
+
+
+NETWORK_BACKEND_WRITE(gthreadaio) {
         chunk *c, *tc;
         size_t chunks_written = 0;
@@ -70 +176 @@
                         break;
                 case FILE_CHUNK: {
+                        stat_cache_entry *sce = NULL;
                         ssize_t r;
-                        int rounds = 8;
+
+                        /* we might be on our way back from the async request and have a status-code */
+                        if (c->async.ret_val != NETWORK_STATUS_UNSET) {
+                                ret = c->async.ret_val;
+
+                                c->async.ret_val = NETWORK_STATUS_UNSET;
+
+                                return ret;
+                        }
 
                         /* open file if not already opened */
                         if (-1 == c->file.fd) {
-                                if (-1 == (c->file.fd = open(c->file.name->ptr, O_RDONLY | /* O_DIRECT | */ (srv->srvconf.use_noatime ? O_NOATIME : 0)))) {
+                                if (-1 == (c->file.fd = open(c->file.name->ptr, O_RDONLY /* | O_DIRECT */ | (srv->srvconf.use_noatime ? O_NOATIME : 0)))) {
                                         ERROR("opening '%s' failed: %s", BUF_STR(c->file.name), strerror(errno));
 
@@ -88 +203 @@
                         }
 
-                        do {
-                                size_t toSend;
-                                const size_t max_toSend = 2 * 256 * 1024; /** should be larger than the send buffer */
-                                off_t offset;
-
-                                offset = c->file.start + c->offset;
-
-                                toSend = c->file.length - c->offset > max_toSend ?
-                                        max_toSend : c->file.length - c->offset;
-
-                                if (0 == c->file.copy.length) {
-                                        int async_error = 0;
-
-                                        size_t iocb_ndx;
-
-
-                                        c->file.copy.offset = 0;
-                                        c->file.copy.length = toSend;
-
-                                        if (c->file.copy.length == 0) {
-                                                async_error = 1;
-                                        }
-
-                                        /* if we reused the previous tmp-file we get overlaps
-                                         *
-                                         *    1 ... 3904 are ok
-                                         * 3905 ... 4096 are replaces by 8001 ... 8192
-                                         *
-                                         * somehow the second read writes into the mmap() before
-                                         * the sendfile is finished which is very strange.
-                                         *
-                                         * if someone finds the reason for this, feel free to remove
-                                         * this if again and number of reduce the syscalls a bit.
-                                         */
-                                        if (-1 != c->file.copy.fd) {
-                                                munmap(c->file.mmap.start, c->file.mmap.length);
-
-                                                close(c->file.copy.fd);
-                                                c->file.copy.fd = -1;
-                                        }
-
-                                        /* do we have a IOCB we can use ? */
-
-                                        for (iocb_ndx = 0; async_error == 0 && iocb_ndx < POSIX_AIO_MAX_IOCBS; iocb_ndx++) {
-                                                if (NULL == srv->posix_aio_data[iocb_ndx]) {
-                                                        break;
-                                                }
-                                        }
-
-                                        if (iocb_ndx == POSIX_AIO_MAX_IOCBS) {
-                                                async_error = 1;
-                                        }
-
-
-                                        /* get mmap()ed mem-block in /dev/shm
-                                         *
-                                         * in case we don't have a iocb available, we still need the mmap() for the blocking
-                                         * read()
-                                         *  */
-                                        if (-1 == c->file.copy.fd ) {
-                                                int mmap_fd = -1;
-
-                                                /* open a file in /dev/shm to write to */
-                                                if (-1 == (mmap_fd = open("/dev/zero", O_RDWR))) {
-                                                        async_error = 1;
-
-                                                        if (errno != EMFILE) {
-                                                                TRACE("open(/dev/zero) returned: %d (%s), open fds: %d, falling back to sync-io",
-                                                                        errno, strerror(errno), srv->cur_fds);
-                                                        } else {
-                                                                return NETWORK_STATUS_WAIT_FOR_FD;
-                                                        }
-                                                } else {
-                                                        c->file.mmap.offset = 0;
-                                                        c->file.mmap.length = c->file.copy.length; /* align to page-size */
-
-                                                        c->file.mmap.start = mmap(0, c->file.mmap.length,
-                                                                        PROT_READ | PROT_WRITE, MAP_SHARED, mmap_fd, 0);
-                                                        if (c->file.mmap.start == MAP_FAILED) {
-                                                                async_error = 1;
-                                                        }
-
-                                                        close(mmap_fd);
-                                                        mmap_fd = -1;
-                                                }
-                                        }
-
-
-                                        /* looks like we couldn't get a temp-file [disk-full] */
-                                        if (async_error == 0 && c->file.mmap.start != MAP_FAILED) {
-                                                struct aiocb *iocb = NULL;
-
-                                                assert(c->file.copy.length > 0);
-
-        &n