Hi,
This is the traffic shaping patch reworked based on some ideas we
discussed on this list. It's working but it's not yet finished and may
need some code cleanup. Currently the fdwatch_msecs is statically set to
10ms, just for testing. On future it will dinamically change, as we
discussed.
Feel free to give sugestions and stuff.
Thanks.
Diego Giagio
Index: cherokee/virtual_server.c
===================================================================
--- cherokee/virtual_server.c (revision 64)
+++ cherokee/virtual_server.c (working copy)
@@ -87,6 +87,11 @@
ret = cherokee_dirs_table_new (&vsrv->userdir_dirs);
if (unlikely(ret < ret_ok)) return ret;
+ /* Bandwidth throttler
+ */
+ cherokee_throttler_new (&vsrv->throttler);
+ if (unlikely(ret < ret_ok)) return ret;
+
/* Return the object
*/
*vserver = vsrv;
@@ -138,6 +143,13 @@
cherokee_buffer_free (vserver->name);
cherokee_buffer_free (vserver->root);
+ /* Destroy the bandwidth throttler
+ */
+ if (vserver->throttler != NULL) {
+ cherokee_throttler_free (vserver->throttler);
+ vserver->throttler = NULL;
+ }
+
if (vserver->logger != NULL) {
cherokee_logger_free (vserver->logger);
vserver->logger = NULL;
Index: cherokee/virtual_server.h
===================================================================
--- cherokee/virtual_server.h (revision 64)
+++ cherokee/virtual_server.h (working copy)
@@ -43,6 +43,7 @@
#include "exts_table.h"
#include "dirs_table_entry.h"
#include "logger.h"
+#include "throttler.h"
typedef struct {
struct list_head list_entry;
@@ -71,6 +72,8 @@
#endif
} data;
+ cherokee_throttler_t *throttler; /* Bandwidth throttler */
+
char *server_cert;
char *server_key;
char *ca_cert;
@@ -90,10 +93,10 @@
} cherokee_virtual_server_t;
-#define VSERVER(v) ((cherokee_virtual_server_t *)(v))
-#define VSERVER_LOGGER(v) (LOGGER(VSERVER(v)->logger))
+#define VSERVER(v) ((cherokee_virtual_server_t *)(v))
+#define VSERVER_LOGGER(v) (LOGGER(VSERVER(v)->logger))
+#define VSERVER_THROTTLER(v) (THROTTLER(VSERVER(v)->throttler))
-
ret_t cherokee_virtual_server_new (cherokee_virtual_server_t **vserver);
ret_t cherokee_virtual_server_free (cherokee_virtual_server_t *vserver);
ret_t cherokee_virtual_server_clean (cherokee_virtual_server_t *vserver);
Index: cherokee/throttler.c
===================================================================
--- cherokee/throttler.c (revision 0)
+++ cherokee/throttler.c (revision 0)
@@ -0,0 +1,327 @@
+/* -*- Mode: C; tab-width: 8; indent-tabs-mode: t; c-basic-offset: 8 -*- */
+
+/* Cherokee
+ *
+ * Authors:
+ * Alvaro Lopez Ortega <[EMAIL PROTECTED]>
+ *
+ * Copyright (C) 2001, 2002, 2003, 2004, 2005 Alvaro Lopez Ortega
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of version 2 of the GNU General Public
+ * License as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, write to the Free Software
+ * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307
+ * USA
+ */
+
+#include "common-internal.h"
+#include "throttler.h"
+#include "connection.h"
+
+#include <time.h>
+#include <sys/time.h>
+
+#ifndef timeradd
+#define timeradd(a, b, result) \
+ do { \
+ (result)->tv_sec = (a)->tv_sec + (b)->tv_sec; \
+ (result)->tv_usec = (a)->tv_usec + (b)->tv_usec; \
+ if ((result)->tv_usec >= 1000000) \
+ {
\
+ ++(result)->tv_sec; \
+ (result)->tv_usec -= 1000000; \
+ }
\
+ } while (0)
+#endif
+
+#ifndef timersub
+#define timersub(tvp, uvp, vvp)
\
+ do { \
+ (vvp)->tv_sec = (tvp)->tv_sec - (uvp)->tv_sec; \
+ (vvp)->tv_usec = (tvp)->tv_usec - (uvp)->tv_usec; \
+ if ((vvp)->tv_usec < 0) { \
+ (vvp)->tv_sec--; \
+ (vvp)->tv_usec += 1000000; \
+ } \
+ } while (0)
+#endif
+
+#ifndef timerclear
+#define timerclear(tvp) (tvp)->tv_sec = (tvp)->tv_usec = 0
+#endif
+
+#ifndef timerisset
+#define timerisset(tvp) ((tvp)->tv_sec || (tvp)->tv_usec)
+#endif
+
+#ifndef CLOCKS_PER_SEC
+#define CLOCKS_PER_SEC CLK_TCK
+#endif
+
+#define ENTRY_MAX_AGE 1 /* 1 second */
+
+struct conn_entry_t {
+ struct list_head list;
+ cherokee_connection_t *conn;
+ time_t queue_time;
+};
+
+
+static ret_t
+conn_entry_new (struct conn_entry_t **entry, cherokee_connection_t *conn)
+{
+ CHEROKEE_NEW_TYPE(n, struct conn_entry_t);
+
+ n->conn = conn;
+ n->queue_time = 0;
+
+ INIT_LIST_HEAD(&n->list);
+
+ *entry = n;
+
+ return ret_ok;
+}
+
+static ret_t
+conn_entry_free (struct conn_entry_t *entry)
+{
+ free (entry);
+ return ret_ok;
+}
+
+ret_t
+cherokee_throttler_new (cherokee_throttler_t **throttler)
+{
+ CHEROKEE_NEW_STRUCT(n, throttler);
+
+ timerclear (&n->curr_time);
+ timerclear (&n->last_time);
+ timerclear (&n->next_time);
+
+ n->bytes = 0;
+ n->limit = 0;
+
+ /* Initialize the connection queue
+ */
+ INIT_LIST_HEAD(&n->conn_queue);
+
+ /* Initialize the mutex
+ */
+ CHEROKEE_MUTEX_INIT(&n->mutex, NULL);
+
+ *throttler = n;
+ return ret_ok;
+}
+
+ret_t
+cherokee_throttler_free (cherokee_throttler_t *throttler)
+{
+ struct conn_entry_t *entry;
+ struct list_head *pos;
+ struct list_head *pos_tmp;
+
+ /* Free remaining entries in the queue
+ */
+ list_for_each_safe (pos, pos_tmp, &throttler->conn_queue) {
+ entry = list_entry (pos, struct conn_entry_t, list);
+ list_del (&entry->list);
+ free (entry);
+ }
+
+ free (throttler);
+ return ret_ok;
+}
+
+static unsigned long inline
+tv2ms (struct timeval tv)
+{
+ unsigned long ms;
+ ms = tv.tv_sec * 1000;
+ ms += tv.tv_usec / 1000;
+
+ return ms == 0 ? 1 : ms;
+}
+
+static void inline
+ms2tv (unsigned long ms, struct timeval *tv)
+{
+ tv->tv_sec = (ms / 1000);
+ tv->tv_usec = (ms % 1000) * 1000;
+}
+
+static void
+conn_queue_put_entry (cherokee_throttler_t *throttler, struct conn_entry_t
*entry)
+{
+ time (&entry->queue_time);
+ list_add_tail (&entry->list, &throttler->conn_queue);
+}
+
+static void
+conn_queue_put (cherokee_throttler_t *throttler, cherokee_connection_t *conn)
+{
+ struct list_head *pos;
+ struct conn_entry_t *entry;
+ int do_queue = true;
+
+ /* Check if this connection is already queued
+ */
+ list_for_each (pos, &throttler->conn_queue) {
+ entry = list_entry (pos, struct conn_entry_t, list);
+ if (entry->conn == conn) {
+ /* Don't requeue
+ */
+ do_queue = false;
+ break;
+ }
+ }
+
+ if (do_queue) {
+ conn_entry_new (&entry, conn);
+ conn_queue_put_entry (throttler, entry);
+ }
+}
+
+static void
+conn_queue_del (cherokee_throttler_t *throttler, struct conn_entry_t *entry)
+{
+ /* Delete entry
+ */
+ list_del (&entry->list);
+}
+
+static struct conn_entry_t *
+conn_queue_peek (cherokee_throttler_t *throttler)
+{
+ struct conn_entry_t *entry;
+ struct list_head *pos;
+ time_t now;
+
+ time (&now);
+
+ peek_loop:
+ entry = NULL;
+
+ if (!list_empty (&throttler->conn_queue)) {
+ pos = throttler->conn_queue.next;
+ entry = list_entry (pos, struct conn_entry_t, list);
+
+ if ((now - entry->queue_time) > ENTRY_MAX_AGE) {
+ /* Remove old entry
+ */
+ conn_queue_del (throttler, entry);
+ conn_entry_free (entry);
+ goto peek_loop;
+ }
+ }
+
+ return entry;
+}
+
+ret_t
+cherokee_throttler_check (cherokee_throttler_t *throttler,
+ cherokee_connection_t *conn,
+ struct timeval *next)
+{
+ struct conn_entry_t *entry;
+ struct timeval tv;
+ unsigned long elapsed;
+ unsigned long curr_speed;
+
+ if (throttler == NULL || throttler->limit == 0)
+ return ret_ok;
+
+ CHEROKEE_MUTEX_LOCK (&throttler->mutex);
+
+ /* Retrieve current time
+ */
+ gettimeofday (&throttler->curr_time, NULL);
+
+ /* Calculate the diff between the current and last time
+ */
+ timersub (&throttler->curr_time, &throttler->last_time, &tv);
+
+ /* Calculate elapsed time in ms
+ */
+ elapsed = tv2ms (tv);
+ if (elapsed < 1000)
+ elapsed = 1000;
+
+ /* Calculate current speed in kb/s
+ */
+ curr_speed = throttler->bytes / ((elapsed * 1024) / 1000);
+
+ /* Check if the current speed has exceeded the limit
+ */
+ if (curr_speed > throttler->limit) {
+ conn_queue_put (throttler, conn);
+
+ if (!timerisset (&throttler->next_time)) {
+ unsigned long ms;
+ struct timeval tv_ms;
+
+ /* Calculate the next time
+ */
+ ms = (curr_speed * 1000) / throttler->limit;
+ ms2tv (ms, &tv_ms);
+ timeradd (&throttler->curr_time, &tv_ms,
&throttler->next_time);
+ }
+
+ *next = throttler->next_time;
+
+ CHEROKEE_MUTEX_UNLOCK (&throttler->mutex);
+ return ret_eagain;
+ }
+
+ /* Get the current entry from the queue
+ */
+ entry = conn_queue_peek (throttler);
+ if (entry != NULL) {
+ /* Check if it's the connection we should give bandwidth
+ */
+ if (entry->conn != conn) {
+ CHEROKEE_MUTEX_UNLOCK (&throttler->mutex);
+ return ret_eagain;
+ }
+
+ /* Remove this connection from the head
+ */
+ conn_queue_del (throttler, entry);
+
+ /* Requeue on tail
+ */
+ conn_queue_put_entry (throttler, entry);
+ }
+
+ /* Update time and byte counter
+ */
+ if (throttler->curr_time.tv_sec > throttler->last_time.tv_sec) {
+ throttler->last_time = throttler->curr_time;
+ throttler->bytes = 0;
+ }
+
+ timerclear (&throttler->next_time);
+
+ CHEROKEE_MUTEX_UNLOCK (&throttler->mutex);
+ return ret_ok;
+}
+
+ret_t
+cherokee_throttler_update (cherokee_throttler_t *throttler, size_t n)
+{
+ if (throttler == NULL || throttler->limit == 0)
+ return ret_ok;
+
+ CHEROKEE_MUTEX_LOCK (&throttler->mutex);
+ throttler->bytes += n;
+ CHEROKEE_MUTEX_UNLOCK (&throttler->mutex);
+
+ return ret_ok;
+}
Index: cherokee/throttler.h
===================================================================
--- cherokee/throttler.h (revision 0)
+++ cherokee/throttler.h (revision 0)
@@ -0,0 +1,67 @@
+/* -*- Mode: C; tab-width: 8; indent-tabs-mode: t; c-basic-offset: 8 -*- */
+
+/* Cherokee
+ *
+ * Authors:
+ * Alvaro Lopez Ortega <[EMAIL PROTECTED]>
+ *
+ * Copyright (C) 2001, 2002, 2003, 2004, 2005 Alvaro Lopez Ortega
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of version 2 of the GNU General Public
+ * License as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, write to the Free Software
+ * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307
+ * USA
+ */
+
+#if !defined (CHEROKEE_INSIDE_CHEROKEE_H) && !defined (CHEROKEE_COMPILATION)
+# error "Only <cherokee/cherokee.h> can be included directly, this file may
disappear or change contents."
+#endif
+
+#ifndef CHEROKEE_THROTTLER_H
+#define CHEROKEE_THROTTLER_H
+
+#include <cherokee/common.h>
+#include "connection.h"
+#include "list.h"
+
+#include <sys/time.h>
+
+CHEROKEE_BEGIN_DECLS
+
+typedef struct {
+ struct timeval curr_time;
+ struct timeval last_time;
+ struct timeval next_time;
+
+ unsigned long bytes;
+ unsigned long limit;
+
+ struct list_head conn_queue;
+
+#ifdef HAVE_PTHREAD
+ pthread_mutex_t mutex;
+#endif
+
+} cherokee_throttler_t;
+
+
+#define THROTTLER(t) ((cherokee_throttler_t *)(t))
+
+ret_t cherokee_throttler_new (cherokee_throttler_t **throttler);
+ret_t cherokee_throttler_free (cherokee_throttler_t *throttler);
+
+ret_t cherokee_throttler_check (cherokee_throttler_t *throttler,
cherokee_connection_t *conn, struct timeval *next);
+ret_t cherokee_throttler_update (cherokee_throttler_t *throttler, size_t n);
+
+CHEROKEE_END_DECLS
+
+#endif /* CHEROKEE_CONNECTION_H */
Index: cherokee/read_config_grammar.y
===================================================================
--- cherokee/read_config_grammar.y (revision 64)
+++ cherokee/read_config_grammar.y (working copy)
@@ -281,7 +281,7 @@
%token T_SERVER T_USERDIR T_PIDFILE T_LISTEN T_SERVER_TOKENS T_ENCODER T_ALLOW
T_IO_CACHE T_DIRECTORYINDEX
%token T_ICONS T_AUTH T_NAME T_METHOD T_PASSWDFILE T_SSL_CA_LIST_FILE T_FROM
T_SOCKET T_LOG_FLUSH_INTERVAL
%token T_INCLUDE T_PANIC_ACTION T_JUST_ABOUT T_LISTEN_QUEUE_SIZE T_SENDFILE
T_MINSIZE T_MAXSIZE T_MAX_FDS
-%token T_SHOW T_CHROOT T_ONLY_SECURE T_MAX_CONNECTION_REUSE T_REWRITE
T_POLL_METHOD T_EXTENSION T_IPV6 T_ENV
+%token T_SHOW T_CHROOT T_ONLY_SECURE T_MAX_CONNECTION_REUSE T_REWRITE
T_POLL_METHOD T_EXTENSION T_IPV6 T_ENV T_LIMIT
%token <number> T_NUMBER T_PORT T_PORT_TLS
%token <string> T_QSTRING T_FULLDIR T_ID T_HTTP_URL T_HTTPS_URL T_HOSTNAME
T_IP T_DOMAIN_NAME T_ADDRESS_PORT
@@ -348,6 +348,7 @@
| ssl_key_file
| ssl_ca_list_file
| userdir
+ | limit
;
directory_options :
@@ -470,6 +471,18 @@
SRV(server)->port_tls = $2;
};
+limit : T_LIMIT T_NUMBER
+{
+ cherokee_virtual_server_t *vsrv = auto_virtual_server;
+ long limit;
+
+ limit = $2;
+ if (limit < 0)
+ limit = 0;
+
+ VSERVER_THROTTLER(vsrv)->limit = limit;
+};
+
listen : T_LISTEN host_name
{
SRV(server)->listen_to = $2;
@@ -518,7 +531,6 @@
cherokee_buffer_add (vserver->root, root, root_len);
};
-
log : T_LOG T_ID
{
ret_t ret;
Index: cherokee/thread.c
===================================================================
--- cherokee/thread.c (revision 64)
+++ cherokee/thread.c (working copy)
@@ -153,14 +153,15 @@
INIT_LIST_HEAD((list_t*)&n->active_list);
INIT_LIST_HEAD((list_t*)&n->reuse_list);
INIT_LIST_HEAD((list_t*)&n->polling_list);
+ INIT_LIST_HEAD((list_t*)&n->sleeping_list);
if (fdpoll_type == cherokee_poll_UNSET)
cherokee_fdpoll_best_new (&n->fdpoll, system_fd_num, fd_num);
else
cherokee_fdpoll_new (&n->fdpoll, fdpoll_type, system_fd_num,
fd_num);
-
n->active_list_num = 0;
n->polling_list_num = 0;
+ n->sleeping_list_num = 0;
n->reuse_list_num = 0;
n->exit = false;
@@ -233,7 +234,7 @@
static void
-add_connection (cherokee_thread_t *thd, cherokee_connection_t *conn)
+add_connection_active (cherokee_thread_t *thd, cherokee_connection_t *conn)
{
list_add ((list_t *)conn, &thd->active_list);
thd->active_list_num++;
@@ -247,8 +248,15 @@
}
static void
-del_connection (cherokee_thread_t *thd, cherokee_connection_t *conn)
+add_connection_sleeping (cherokee_thread_t *thd, cherokee_connection_t *conn)
{
+ list_add ((list_t *)conn, &thd->sleeping_list);
+ thd->sleeping_list_num++;
+}
+
+static void
+del_connection_active (cherokee_thread_t *thd, cherokee_connection_t *conn)
+{
list_del ((list_t *)conn);
thd->active_list_num--;
}
@@ -260,7 +268,14 @@
thd->polling_list_num--;
}
+static void
+del_connection_sleeping (cherokee_thread_t *thd, cherokee_connection_t *conn)
+{
+ list_del ((list_t *)conn);
+ thd->sleeping_list_num--;
+}
+
static ret_t
connection_reuse_or_free (cherokee_thread_t *thread, cherokee_connection_t
*conn)
{
@@ -327,7 +342,7 @@
/* Remove from active connections list
*/
- del_connection (thread, conn);
+ del_connection_active (thread, conn);
/* Finally, purge connection
*/
@@ -369,7 +384,32 @@
conn->timeout = thread->bogo_now + srv->timeout;
}
+static ret_t
+process_sleeping_connections (cherokee_thread_t *thd)
+{
+ int re;
+ list_t *tmp, *i;
+ cherokee_connection_t *conn;
+ struct timeval now;
+ gettimeofday (&now, NULL);
+
+ list_for_each_safe (i, tmp, (list_t*)&thd->sleeping_list) {
+ conn = CONN(i);
+
+ if (conn->throttler_time.tv_sec <= now.tv_sec) {
+ if (conn->throttler_time.tv_sec < now.tv_sec ||
+ conn->throttler_time.tv_usec <= now.tv_usec) {
+ /* Move to the 'active' list
+ */
+ cherokee_thread_reactive_from_sleeping (thd,
conn);
+ }
+ }
+ }
+
+ return ret_ok;
+}
+
static ret_t
process_polling_connections (cherokee_thread_t *thd)
{
@@ -422,7 +462,6 @@
cherokee_connection_t *conn = NULL;
cherokee_server_t *srv = thd->server;
-
/* Process active connections
*/
list_for_each_safe (i, tmp, (list_t*)&thd->active_list) {
@@ -449,16 +488,33 @@
*/
process = ((conn->phase == phase_reading_header) &&
(!cherokee_buffer_is_empty (conn->incoming_header)));
-
+
/* Process the connection?
* 2.- Inspect the file descriptor
*/
if (process == false) {
int num;
+ /* First, check if this connection is being throttled
+ * and there's enough bandwidth to proceed
+ */
+ if (CONN_THROTTLER(conn) != NULL) {
+ struct timeval next;
+
+ ret = cherokee_throttler_check
(CONN_THROTTLER(conn), conn, &next);
+ if (ret != ret_ok) {
+ conn->throttler_time = next;
+ cherokee_thread_deactive_to_sleeping
(thd, conn);
+ continue;
+ }
+ }
+
+ /* Then, check if the fd is ready
+ */
num = cherokee_fdpoll_check (thd->fdpoll,
SOCKET_FD(conn->socket),
SOCKET_STATUS(conn->socket));
+
switch (num) {
case -1:
purge_closed_connection(thd, conn);
@@ -469,7 +525,7 @@
process = true;
}
-
+
/* Process the connection?
* Finial.-
*/
@@ -669,6 +725,10 @@
*/
conn->logger_ref = CONN_VSRV(conn)->logger;
+ /* Setup the bandwidth throttler
+ */
+ conn->throttler = CONN_VSRV(conn)->throttler;
+
/* Is it already an error response?
*/
if (http_type_300(conn->error_code) ||
@@ -864,7 +924,6 @@
conn->phase = phase_send_headers;
case phase_send_headers:
-
/* Send headers to the client
*/
ret = cherokee_connection_send_header (conn);
@@ -1041,7 +1100,7 @@
cherokee_sockaddr_t new_sa;
cherokee_connection_t *new_conn;
-
+
/* Return if there're no new connections
*/
if (cherokee_fdpoll_check (thd->fdpoll, srv_socket, 0) == 0) {
@@ -1482,6 +1541,10 @@
*/
CHEROKEE_MUTEX_LOCK (&thd->ownership);
+ /* Process sleeping connections
+ */
+ process_sleeping_connections (thd);
+
/* Process polling connections
*/
process_polling_connections (thd);
@@ -1550,7 +1613,7 @@
if (unlikely (ret < ret_ok)) return ret;
conn_set_mode (thd, conn, socket_reading);
- add_connection (thd, conn);
+ add_connection_active (thd, conn);
return ret_ok;
}
@@ -1597,6 +1660,18 @@
}
ret_t
+cherokee_thread_deactive_to_sleeping (cherokee_thread_t *thd,
cherokee_connection_t *conn)
+{
+ cherokee_socket_t *socket = CONN_SOCK(conn);
+
+ /* Remove the connection file descriptor
+ */
+ cherokee_fdpoll_del (thd->fdpoll, SOCKET_FD(socket));
+
+ return cherokee_thread_move_connection_to_sleeping (thd, conn);
+}
+
+ret_t
cherokee_thread_reactive_from_polling (cherokee_thread_t *thd,
cherokee_connection_t *conn)
{
cherokee_socket_t *socket = CONN_SOCK(conn);
@@ -1610,25 +1685,55 @@
*/
conn->extra_polling_fd = -1;
- return cherokee_thread_move_connection_to_active (thd, conn);
+ return cherokee_thread_move_connection_from_polling_to_active (thd,
conn);
}
+ret_t
+cherokee_thread_reactive_from_sleeping (cherokee_thread_t *thd,
cherokee_connection_t *conn)
+{
+ cherokee_socket_t *socket = CONN_SOCK(conn);
+ /* Set the connection file descriptor
+ */
+ cherokee_fdpoll_add (thd->fdpoll, socket->socket, socket->status);
+
+ return cherokee_thread_move_connection_from_sleeping_to_active (thd,
conn);
+}
+
+
ret_t
cherokee_thread_move_connection_to_polling (cherokee_thread_t *thd,
cherokee_connection_t *conn)
{
- del_connection (thd, conn);
+ del_connection_active (thd, conn);
add_connection_polling (thd, conn);
return ret_ok;
}
+ret_t
+cherokee_thread_move_connection_to_sleeping (cherokee_thread_t *thd,
cherokee_connection_t *conn)
+{
+ del_connection_active (thd, conn);
+ add_connection_sleeping (thd, conn);
+ return ret_ok;
+}
+
+
ret_t
-cherokee_thread_move_connection_to_active (cherokee_thread_t *thd,
cherokee_connection_t *conn)
+cherokee_thread_move_connection_from_polling_to_active (cherokee_thread_t
*thd, cherokee_connection_t *conn)
{
del_connection_polling (thd, conn);
- add_connection (thd, conn);
+ add_connection_active (thd, conn);
return ret_ok;
}
+
+ret_t
+cherokee_thread_move_connection_from_sleeping_to_active (cherokee_thread_t
*thd, cherokee_connection_t *conn)
+{
+ del_connection_sleeping (thd, conn);
+ add_connection_active (thd, conn);
+
+ return ret_ok;
+}
Index: cherokee/handler_file.c
===================================================================
--- cherokee/handler_file.c (revision 64)
+++ cherokee/handler_file.c (working copy)
@@ -397,24 +397,26 @@
}
#endif
- /* Maybe use sendfile
- */
+ if (CONN_THROTTLER(conn)->limit == 0) {
+ /* Maybe use sendfile
+ */
#ifdef HAVE_SENDFILE
- n->using_sendfile = ((conn->mmaped == NULL) &&
- (conn->encoder == NULL) &&
- (n->info->st_size >= srv->sendfile.min) &&
- (n->info->st_size < srv->sendfile.max) &&
- (conn->socket->is_tls == non_TLS));
+ n->using_sendfile = ((conn->mmaped == NULL) &&
+ (conn->encoder == NULL) &&
+ (n->info->st_size >= srv->sendfile.min) &&
+ (n->info->st_size < srv->sendfile.max) &&
+ (conn->socket->is_tls == non_TLS));
# ifdef HAVE_SENDFILE_BROKEN
- n->using_sendfile = false;
+ n->using_sendfile = false;
# endif
- if (n->using_sendfile) {
- cherokee_connection_set_cork(conn, 1);
+ if (n->using_sendfile) {
+ cherokee_connection_set_cork(conn, 1);
+ }
+#endif
}
-#endif
-
+
return ret_ok;
}
Index: cherokee/thread.h
===================================================================
--- cherokee/thread.h (revision 64)
+++ cherokee/thread.h (working copy)
@@ -73,6 +73,8 @@
list_t active_list;
int polling_list_num;
list_t polling_list;
+ int sleeping_list_num;
+ list_t sleeping_list;
list_t reuse_list;
int reuse_list_num;
@@ -90,7 +92,6 @@
#define THREAD_SRV(t) (SRV(THREAD(t)->server))
#define THREAD_IS_REAL(t) (THREAD(t)->real_thread)
-
#ifdef HAVE_PTHREAD
# define cherokee_thread_step(t,b) cherokee_thread_step_MULTI_THREAD(t,b)
ret_t cherokee_thread_step_MULTI_THREAD (cherokee_thread_t *thd,
cherokee_boolean_t dont_block);
@@ -112,7 +113,13 @@
ret_t cherokee_thread_deactive_to_polling (cherokee_thread_t *thd,
cherokee_connection_t *conn, int fd, int rw);
ret_t cherokee_thread_reactive_from_polling (cherokee_thread_t *thd,
cherokee_connection_t *conn);
ret_t cherokee_thread_move_connection_to_polling (cherokee_thread_t *thd,
cherokee_connection_t *conn);
-ret_t cherokee_thread_move_connection_to_active (cherokee_thread_t *thd,
cherokee_connection_t *conn);
+ret_t cherokee_thread_move_connection_from_polling_to_active
(cherokee_thread_t *thd, cherokee_connection_t *conn);
+
+ret_t cherokee_thread_deactive_to_sleeping (cherokee_thread_t *thd,
cherokee_connection_t *conn);
+ret_t cherokee_thread_reactive_from_sleeping (cherokee_thread_t *thd,
cherokee_connection_t *conn);
+ret_t cherokee_thread_move_connection_to_sleeping(cherokee_thread_t *thd,
cherokee_connection_t *conn);
+ret_t cherokee_thread_move_connection_from_sleeping_to_active
(cherokee_thread_t *thd, cherokee_connection_t *conn);
+
ret_t cherokee_thread_internal_request_rewrite (cherokee_thread_t *thd,
cherokee_connection_t *conn, cherokee_buffer_t *nr);
int cherokee_thread_connection_num (cherokee_thread_t *thd);
Index: cherokee/Makefile.am
===================================================================
--- cherokee/Makefile.am (revision 64)
+++ cherokee/Makefile.am (working copy)
@@ -721,7 +721,9 @@
nonce.h \
nonce.c \
post.h \
-post.c
+post.c \
+throttler.h \
+throttler.c
libcherokee_config_la_SOURCES = \
$(config_common) \
Index: cherokee/read_config_scanner.l
===================================================================
--- cherokee/read_config_scanner.l (revision 64)
+++ cherokee/read_config_scanner.l (working copy)
@@ -113,6 +113,7 @@
"MaxConnectionReuse" { return T_MAX_CONNECTION_REUSE; }
"IOCache" { return T_IO_CACHE; }
"Env" { return T_ENV; }
+"Limit" { return T_LIMIT; }
"On" { yylval.number = 1; return T_NUMBER; }
"Off" { yylval.number = 0; return T_NUMBER; }
Index: cherokee/connection.c
===================================================================
--- cherokee/connection.c (revision 64)
+++ cherokee/connection.c (working copy)
@@ -115,6 +115,7 @@
n->tx_partial = 0;
n->traffic_next = 0;
n->validator = NULL;
+ n->throttler = NULL;
cherokee_buffer_new (&n->buffer);
cherokee_buffer_new (&n->header_buffer);
@@ -639,6 +640,7 @@
}
cherokee_connection_tx_add (cnt, re);
+ cherokee_throttler_update (CONN_THROTTLER(cnt), re);
cnt->mmaped_len -= re;
cnt->mmaped += re;
@@ -670,6 +672,14 @@
default:
RET_UNKNOWN(ret);
}
+
+ /* Add to the connection traffic counter
+ */
+ cherokee_connection_tx_add (cnt, re);
+
+ /* Update the bandwidth throttler
+ */
+ cherokee_throttler_update (CONN_THROTTLER(cnt), re);
/* If writev() has not sent all data
*/
@@ -690,10 +700,6 @@
return ret_eagain;
}
- /* Add to the connection traffic counter
- */
- cherokee_connection_tx_add (cnt, re);
-
return ret_ok;
}
@@ -725,6 +731,7 @@
switch (ret) {
case ret_ok:
cherokee_connection_rx_add (cnt, readed);
+ cherokee_throttler_update (CONN_THROTTLER(cnt), readed);
*len = readed;
return ret_ok;
@@ -797,6 +804,10 @@
*/
cherokee_connection_tx_add (cnt, sent);
+ /* Update the bandwidth throttler
+ */
+ cherokee_throttler_update (CONN_THROTTLER(cnt), sent);
+
/* Drop out the sent info
*/
if (sent == cnt->buffer->len) {
@@ -826,6 +837,10 @@
*/
cherokee_connection_tx_add (cnt, sent);
+ /* Update the bandwidth throttler
+ */
+ cherokee_throttler_update (CONN_THROTTLER(cnt), sent);
+
/* Drop out the sent info
*/
if (sent == cnt->buffer->len) {
Index: cherokee/server.c
===================================================================
--- cherokee/server.c (revision 64)
+++ cherokee/server.c (working copy)
@@ -130,7 +130,7 @@
n->tls_enabled = false;
n->listen_to = NULL;
- n->fdwatch_msecs = 999;
+ n->fdwatch_msecs = SRV_DEFAULT_LATENCY;
n->start_time = time(NULL);
@@ -417,13 +417,12 @@
}
-void
+void
cherokee_server_set_min_latency (cherokee_server_t *srv, int msecs)
{
srv->fdwatch_msecs = msecs;
}
-
static ret_t
set_server_socket_opts (int socket)
{
Index: cherokee/server.h
===================================================================
--- cherokee/server.h (revision 64)
+++ cherokee/server.h (working copy)
@@ -47,6 +47,7 @@
typedef void (* cherokee_server_reinit_cb_t) (cherokee_server_t *new_srv);
#define SRV(x) ((cherokee_server_t *)(x))
+#define SRV_DEFAULT_LATENCY 10
ret_t cherokee_server_new (cherokee_server_t **srv);
ret_t cherokee_server_init (cherokee_server_t *srv);
Index: cherokee/connection-protected.h
===================================================================
--- cherokee/connection-protected.h (revision 64)
+++ cherokee/connection-protected.h (working copy)
@@ -61,6 +61,7 @@
#include "iocache.h"
#include "encoder_table.h"
#include "post.h"
+#include "throttler.h"
typedef enum {
phase_nothing,
@@ -158,6 +159,11 @@
size_t tx_partial; /* TX partial
counter */
time_t traffic_next; /* Time to update
traffic */
+ /* Bandwidth throttler
+ */
+ cherokee_throttler_t *throttler;
+ struct timeval throttler_time;
+
/* Post info
*/
cherokee_post_t post;
@@ -177,13 +183,13 @@
cherokee_iocache_entry_t *io_entry_ref;
};
-#define CONN_SRV(c) (SRV(CONN(c)->server))
-#define CONN_HDR(c) (HDR(CONN(c)->header))
-#define CONN_SOCK(c) (SOCKET(CONN(c)->socket))
-#define CONN_VSRV(c) (VSERVER(CONN(c)->vserver))
-#define CONN_THREAD(c) (THREAD(CONN(c)->thread))
+#define CONN_SRV(c) (SRV(CONN(c)->server))
+#define CONN_HDR(c) (HDR(CONN(c)->header))
+#define CONN_SOCK(c) (SOCKET(CONN(c)->socket))
+#define CONN_VSRV(c) (VSERVER(CONN(c)->vserver))
+#define CONN_THREAD(c) (THREAD(CONN(c)->thread))
+#define CONN_THROTTLER(c) (THROTTLER(CONN(c)->throttler))
-
/* Basic functions
*/
ret_t cherokee_connection_new (cherokee_connection_t **cnt);
_______________________________________________
Cherokee mailing list
[email protected]
http://www.alobbs.com/cgi-bin/mailman/listinfo/cherokee
