Hi All,
This is just a resend of my timeout patch to prevent it from fading from
our combined attention.
Rationale:
----------
The patch is a revert of my previous timeout approach following well
deserved criticism from Michael. A g_timeout_source based approach is
used instead.
What it does:
-------------
It introduced a new ORB parameter GIOPTimeoutMSEC which is the timeout
value in milliseconds. The default timeout value is 60 seconds unless
modified using this ORB parameter.
A timeout source is added to the link main context for all IPv4 and IPv6
connections unless said connection is a oneway.
The timeout source will be removed when any of the following happens:
1) The timeout triggered (link state = LINK_TIMEOUT)
2) A respons was received on the connection
3) The connection was destroyed
A CORBA timeout exception (ex_CORBA_TIMEOUT) is thrown if the timeout
triggered.
Please:
-------
Please review the patch so that we can get the current broken timeout
behavior replaced with something that actually has a fighting chance of
working properly. I would really like to commit sooner rather than later
considering how bad the old timeout approach turned out to be.
Thanks,
jules
Index: src/orb/orb-core/corba-orb.c
===================================================================
--- src/orb/orb-core/corba-orb.c (revision 2003)
+++ src/orb/orb-core/corba-orb.c (working copy)
@@ -61,7 +61,7 @@
static char *orbit_naming_ref = NULL;
static GSList *orbit_initref_list = NULL;
static gboolean orbit_use_corbaloc = FALSE;
-static gint orbit_timeout_limit = -1;
+static guint orbit_timeout_msec = 60000; /* 60 seconds - 0 will
disable timeouts altogether */
void
ORBit_ORB_start_servers (CORBA_ORB orb)
{
@@ -418,7 +418,7 @@
#endif /* G_ENABLE_DEBUG */
giop_recv_set_limit (orbit_initial_recv_limit);
- giop_recv_set_timeout (orbit_timeout_limit);
+ giop_set_timeout (orbit_timeout_msec);
giop_init (thread_safe,
orbit_use_ipv4 || orbit_use_ipv6 ||
orbit_use_irda || orbit_use_ssl);
@@ -1468,7 +1468,7 @@
{ "ORBDebugFlags", ORBIT_OPTION_STRING, &orbit_debug_options },
{ "ORBInitRef", ORBIT_OPTION_KEY_VALUE, &orbit_initref_list},
{ "ORBCorbaloc", ORBIT_OPTION_BOOLEAN, &orbit_use_corbaloc},
- { "GIOPTimeoutLimit", ORBIT_OPTION_INT, &orbit_timeout_limit },
+ { "GIOPTimeoutMSEC", ORBIT_OPTION_ULONG, &orbit_timeout_msec },
{ NULL, 0, NULL }
};
Index: src/orb/orb-core/corba-object.c
===================================================================
--- src/orb/orb-core/corba-object.c (revision 2003)
+++ src/orb/orb-core/corba-object.c (working copy)
@@ -270,6 +270,7 @@
retval = TRUE;
break;
case LINK_DISCONNECTED:
+ case LINK_TIMEOUT:
/* Have a go at reviving it */
dprintf (MESSAGES, "re-connecting dropped cnx %p: ", cnx);
if (giop_connection_try_reconnect (GIOP_CONNECTION (cnx)) ==
LINK_CONNECTED)
Index: src/orb/util/orbit-options.c
===================================================================
--- src/orb/util/orbit-options.c (revision 2003)
+++ src/orb/util/orbit-options.c (working copy)
@@ -53,6 +53,9 @@
case ORBIT_OPTION_INT:
*(gint *)option->arg = atoi (val);
break;
+ case ORBIT_OPTION_ULONG:
+ *(guint *)option->arg = strtoul(val, (char **)NULL, 10);
+ break;
case ORBIT_OPTION_STRING: {
gchar **str_arg = (char **) option->arg;
Index: src/orb/util/orbit-options.h
===================================================================
--- src/orb/util/orbit-options.h (revision 2003)
+++ src/orb/util/orbit-options.h (working copy)
@@ -10,7 +10,8 @@
ORBIT_OPTION_STRING,
ORBIT_OPTION_INT,
ORBIT_OPTION_BOOLEAN,
- ORBIT_OPTION_KEY_VALUE /* returns GSList of ORBit_option_key_value */
+ ORBIT_OPTION_KEY_VALUE, /* returns GSList of ORBit_option_key_value */
+ ORBIT_OPTION_ULONG,
} ORBit_option_type;
typedef struct {
Index: src/orb/GIOP/giop-recv-buffer.c
===================================================================
--- src/orb/GIOP/giop-recv-buffer.c (revision 2003)
+++ src/orb/GIOP/giop-recv-buffer.c (working copy)
@@ -2,6 +2,8 @@
#include <string.h>
#include <stdio.h>
#include <errno.h>
+#include <glib.h>
+#include <glib/gprintf.h>
#include "giop-private.h"
#include "giop-debug.h"
@@ -686,26 +688,20 @@
static inline gboolean
check_got (GIOPMessageQueueEntry *ent)
{
- return (ent->buffer || !ent->cnx ||
- (ent->cnx->parent.status == LINK_DISCONNECTED));
+ return (ent->buffer ||
+ !ent->cnx ||
+ (ent->cnx->parent.status == LINK_DISCONNECTED) ||
+ (ent->cnx->parent.status == LINK_TIMEOUT));
}
-static glong giop_initial_timeout_limit = GIOP_INITIAL_TIMEOUT_LIMIT;
-
-void
-giop_recv_set_timeout (const glong timeout)
-{
- if (0 < timeout) /* We really do not want (timeout <= 0) as that would
potentially block forever */
- giop_initial_timeout_limit = timeout;
-}
-
GIOPRecvBuffer *
giop_recv_buffer_get (GIOPMessageQueueEntry *ent,
gboolean *timeout)
{
GIOPThread *tdata = giop_thread_self ();
- GTimeVal tval;
+ *timeout = FALSE;
+
thread_switch:
if (giop_thread_io ()) {
ent_lock (ent);
@@ -715,17 +711,8 @@
ent_unlock (ent);
giop_thread_queue_process (tdata);
ent_lock (ent);
- } else {
- if (0 < giop_initial_timeout_limit) {
- g_get_current_time (&tval);
- g_time_val_add (&tval,
giop_initial_timeout_limit);
- }
- if (!g_cond_timed_wait (tdata->incoming,
tdata->lock, ((0 < giop_initial_timeout_limit) ? &tval : NULL))) {
- *timeout = TRUE;
- break;
- } else
- *timeout = FALSE;
- }
+ } else
+ g_cond_wait (tdata->incoming, tdata->lock);
}
ent_unlock (ent);
@@ -734,6 +721,7 @@
while (!ent->buffer && ent->cnx &&
(ent->cnx->parent.status != LINK_DISCONNECTED) &&
+ (ent->cnx->parent.status != LINK_TIMEOUT) &&
!giop_thread_io())
link_main_iteration (TRUE);
@@ -741,6 +729,17 @@
goto thread_switch;
}
+ if (ent->cnx->parent.timeout_mutex) {
+ g_mutex_lock (ent->cnx->parent.timeout_mutex);
+ if (ent->cnx->parent.timeout_status == LINK_TIMEOUT_UNKNOWN) {
+ link_io_thread_remove_timeout
(ent->cnx->parent.timeout_source_id);
+ ent->cnx->parent.timeout_source_id = 0;
+ ent->cnx->parent.timeout_status = LINK_TIMEOUT_NO;
+ } else
+ *timeout = TRUE;
+ g_mutex_unlock (ent->cnx->parent.timeout_mutex);
+ }
+
giop_thread_queue_tail_wakeup (tdata);
giop_recv_list_destroy_queue_entry (ent);
@@ -1355,6 +1354,74 @@
return TRUE;
}
+struct timeout_thread_data {
+ GIOPThread *tdata;
+ LinkConnection *lcnx;
+};
+
+static gboolean
+giop_timeout(gpointer data)
+{
+ gboolean retv = FALSE;
+ LinkConnection *lcnx = ((struct timeout_thread_data*)data)->lcnx;
+ GIOPThread *tdata = ((struct timeout_thread_data*)data)->tdata;
+
+ g_printf("Timeout function invoked\n");
+
+ g_assert (lcnx->timeout_mutex);
+
+ g_mutex_lock (lcnx->timeout_mutex);
+ if (lcnx->timeout_status == LINK_TIMEOUT_UNKNOWN) {
+ lcnx->timeout_source_id = 0;
+ lcnx->timeout_status = LINK_TIMEOUT_YES;
+ } else {
+ g_mutex_unlock (lcnx->timeout_mutex);
+ retv = TRUE; // do not remove the source - the one who sets
timeout_status will do that
+ goto out;
+ }
+ g_mutex_unlock (lcnx->timeout_mutex);
+
+ link_connection_state_changed (lcnx, LINK_TIMEOUT);
+
+ g_mutex_lock (tdata->lock); /* ent_lock */
+ giop_incoming_signal_T (tdata, GIOP_CLOSECONNECTION);
+ g_mutex_unlock (tdata->lock); /* ent_lock */
+
+out:
+ g_object_unref (lcnx);
+ g_free (data);
+
+ return retv;
+}
+
+void
+giop_timeout_add(GIOPConnection *cnx)
+{
+ struct timeout_thread_data *data = NULL;
+ LinkConnection *lcnx = LINK_CONNECTION (cnx);
+ GSource *timeout_source = NULL;
+
+ if (!lcnx->timeout_msec)
+ return;
+
+ g_object_ref (lcnx);
+
+ if (!lcnx->timeout_mutex)
+ lcnx->timeout_mutex = g_mutex_new ();
+
+ g_mutex_lock (lcnx->timeout_mutex);
+ lcnx->timeout_status = LINK_TIMEOUT_UNKNOWN;
+ g_mutex_unlock (lcnx->timeout_mutex);
+
+ data = g_new0 (struct timeout_thread_data, 1);
+ data->tdata = giop_thread_self ();
+ data->lcnx = lcnx;
+
+ g_printf("Adding timeout for %d milliseconds\n", lcnx->timeout_msec);
+
+ lcnx->timeout_source_id = link_io_thread_add_timeout
(lcnx->timeout_msec, giop_timeout, data);
+}
+
GIOPRecvBuffer *
giop_recv_buffer_use_buf (GIOPConnection *cnx)
{
@@ -1372,4 +1439,3 @@
return buf;
}
-
Index: src/orb/GIOP/giop-connection.c
===================================================================
--- src/orb/GIOP/giop-connection.c (revision 2003)
+++ src/orb/GIOP/giop-connection.c (working copy)
@@ -108,6 +108,7 @@
}
}
+
static void
giop_connection_class_init (GIOPConnectionClass *klass)
{
@@ -192,3 +193,9 @@
return link_connection_try_reconnect (LINK_CONNECTION (cnx));
}
+void
+giop_set_timeout (guint msec)
+{
+ link_set_timeout (msec);
+}
+
Index: src/orb/GIOP/giop-send-buffer.c
===================================================================
--- src/orb/GIOP/giop-send-buffer.c (revision 2003)
+++ src/orb/GIOP/giop-send-buffer.c (working copy)
@@ -45,6 +45,25 @@
static const GIOP_AddressingDisposition giop_1_2_target_type = GIOP_KeyAddr;
+static gboolean
+giop_send_buffer_is_oneway(const GIOPSendBuffer *buf)
+{
+ g_assert (buf);
+
+ switch (buf->giop_version) {
+ case GIOP_1_0:
+ case GIOP_1_1:
+ return (buf->msg.u.request_1_0.response_expected ? FALSE :
TRUE);
+ case GIOP_1_2:
+ return (buf->msg.u.request_1_2.response_flags ? FALSE : TRUE);
+ default:
+ break;
+ }
+ g_assert_not_reached();
+
+ return TRUE;
+}
+
GIOPSendBuffer *
giop_send_buffer_use_request (GIOPVersion giop_version,
CORBA_unsigned_long request_id,
@@ -426,6 +445,7 @@
gboolean blocking)
{
int retval;
+ LinkConnection *lcnx = LINK_CONNECTION (cnx);
static LinkWriteOpts *non_block = NULL;
if (!non_block)
@@ -434,11 +454,15 @@
/* FIXME: if a FRAGMENT, assert the 8 byte tail align,
&&|| giop_send_buffer_align (buf, 8); */
- retval = link_connection_writev (
- (LinkConnection *) cnx, buf->iovecs,
- buf->num_used,
- blocking ? NULL : non_block);
+ if (lcnx->timeout_msec && !giop_send_buffer_is_oneway (buf)) {
+ giop_timeout_add (cnx);
+ }
+ retval = link_connection_writev (lcnx,
+ buf->iovecs,
+ buf->num_used,
+ blocking ? NULL : non_block);
+
if (!blocking && retval == LINK_IO_QUEUED_DATA)
retval = 0;
Index: linc2/include/linc/linc-connection.h
===================================================================
--- linc2/include/linc/linc-connection.h (revision 2003)
+++ linc2/include/linc/linc-connection.h (working copy)
@@ -39,7 +39,8 @@
#define LINK_IS_CONNECTION(obj) (G_TYPE_CHECK_INSTANCE_TYPE ((obj),
LINK_TYPE_CONNECTION))
#define LINK_IS_CONNECTION_CLASS(klass) (G_TYPE_CHECK_CLASS_TYPE ((klass),
LINK_TYPE_CONNECTION))
-typedef enum { LINK_CONNECTING, LINK_CONNECTED, LINK_DISCONNECTED }
LinkConnectionStatus;
+typedef enum { LINK_CONNECTING, LINK_CONNECTED, LINK_DISCONNECTED,
LINK_TIMEOUT } LinkConnectionStatus;
+typedef enum { LINK_TIMEOUT_UNKNOWN, LINK_TIMEOUT_YES, LINK_TIMEOUT_NO }
LinkTimeoutStatus;
typedef struct _LinkWriteOpts LinkWriteOpts;
typedef struct _LinkConnectionPrivate LinkConnectionPrivate;
@@ -61,6 +62,11 @@
LinkConnectionPrivate *priv;
GSList *idle_broken_callbacks;
+
+ GMutex *timeout_mutex;
+ guint timeout_msec;
+ guint timeout_source_id; // protected by timeout_mutex
+ LinkTimeoutStatus timeout_status; // protected by timeout_mutex
} LinkConnection;
typedef struct {
@@ -153,6 +159,9 @@
void link_connections_close (void);
+/* set the link timeout in miliseconds */
+extern void link_set_timeout (guint msec);
+
G_END_DECLS
#endif /* _LINK_CONNECTION_H */
Index: linc2/include/linc/linc.h
===================================================================
--- linc2/include/linc/linc.h (revision 2003)
+++ linc2/include/linc/linc.h (working copy)
@@ -33,12 +33,17 @@
guint link_main_idle_add (GSourceFunc function,
gpointer data);
-gboolean link_wait (void);
+void link_wait (void);
void link_signal (void);
gboolean link_thread_io (void);
gboolean link_thread_safe (void);
+guint link_io_thread_add_timeout (guint interval,
+ GSourceFunc function,
+ gpointer data);
+void link_io_thread_remove_timeout (guint source_id);
+
#ifdef G_OS_WIN32
void link_map_winsock_error_to_errno (void);
#endif
Index: linc2/src/linc-debug.h
===================================================================
--- linc2/src/linc-debug.h (revision 2003)
+++ linc2/src/linc-debug.h (working copy)
@@ -25,6 +25,7 @@
# define STATE_NAME(s) (((s) == LINK_CONNECTED) ? "Connected" : \
((s) == LINK_CONNECTING) ? "Connecting" : \
((s) == LINK_DISCONNECTED) ? "Disconnected" : \
+ ((s) == LINK_TIMEOUT) ? "Timeout" : \
"Invalid state")
# ifdef CONNECTION_DEBUG_FLAG
extern gboolean link_connection_debug_flag;
Index: linc2/src/linc-connection.c
===================================================================
--- linc2/src/linc-connection.c (revision 2003)
+++ linc2/src/linc-connection.c (working copy)
@@ -27,6 +27,7 @@
#include <linc/linc-connection.h>
static GObjectClass *parent_class = NULL;
+static guint _link_timeout = 0;
enum {
BROKEN,
@@ -288,6 +289,7 @@
break;
case LINK_DISCONNECTED:
+ case LINK_TIMEOUT:
link_source_remove (cnx);
link_close_fd (cnx);
queue_free (cnx);
@@ -440,6 +442,16 @@
g_free (cnx->remote_serv_info);
cnx->remote_serv_info = remote_serv_info;
+ switch (cnx->proto->family) {
+ case AF_INET:
+ case AF_INET6:
+ if (_link_timeout && !cnx->timeout_msec) /* this should'nt
happen twice but I'm always paranoid... */
+ cnx->timeout_msec = _link_timeout;
+ break;
+ default:
+ break;
+ }
+
d_printf ("Cnx from fd (%d) '%s', '%s', '%s'\n",
fd, proto->name,
remote_host_info ? remote_host_info : "<Null>",
@@ -635,10 +647,8 @@
static LinkConnectionStatus
link_connection_wait_connected_T (LinkConnection *cnx)
{
- while (cnx && cnx->status == LINK_CONNECTING) {
- if (!link_wait ())
- link_connection_disconnect (cnx);
- }
+ while (cnx && cnx->status == LINK_CONNECTING)
+ link_wait ();
return cnx ? cnx->status : LINK_DISCONNECTED;
}
@@ -661,12 +671,8 @@
cnx->inhibit_reconnect = FALSE;
dispatch_callbacks_drop_lock (cnx);
g_main_context_release (NULL);
- } else {
- if (!link_wait ()) {
- link_connection_disconnect (cnx);
- break;
- }
- }
+ } else
+ link_wait ();
}
if (cnx->status != LINK_DISCONNECTED)
@@ -1254,6 +1260,13 @@
g_free (cnx->priv);
+ if (cnx->timeout_mutex)
+ g_mutex_free (cnx->timeout_mutex);
+
+ if (cnx->timeout_source_id)
+ link_io_thread_remove_timeout (cnx->timeout_source_id);
+
+
#ifdef G_ENABLE_DEBUG
g_assert (g_list_find(cnx_list, cnx) == NULL);
#endif
@@ -1269,6 +1282,12 @@
cnx->priv = g_new0 (LinkConnectionPrivate, 1);
cnx->priv->fd = -1;
cnx->priv->was_disconnected = FALSE;
+
+ cnx->timeout_mutex = NULL;
+ cnx->timeout_msec = 0;
+ cnx->timeout_source_id = 0;
+ cnx->timeout_status = LINK_TIMEOUT_UNKNOWN;
+
#ifdef CONNECTION_DEBUG
cnx->priv->total_read_bytes = 0;
cnx->priv->total_written_bytes = 0;
@@ -1568,3 +1587,10 @@
g_list_free (cnx);
}
+
+void
+link_set_timeout (guint msec)
+{
+ _link_timeout = msec;
+}
+
Index: linc2/src/linc.c
===================================================================
--- linc2/src/linc.c (revision 2003)
+++ linc2/src/linc.c (working copy)
@@ -52,9 +52,6 @@
SSL_CTX *link_ssl_ctx;
#endif
-/* max time to wait for the link condition to get signaled - 10 seconds */
-#define LINK_WAIT_TIMEOUT_USEC (10000000)
-
static void link_dispatch_command (gpointer data, gboolean immediate);
gboolean
@@ -537,28 +534,17 @@
}
}
-gboolean
+void
link_wait (void)
{
- GTimeVal gtime;
-
if (!(link_is_thread_safe && link_is_io_in_thread)) {
link_unlock ();
link_main_iteration (TRUE);
link_lock ();
} else {
g_assert (link_main_cond != NULL);
-
- g_get_current_time (>ime);
- g_time_val_add (>ime, LINK_WAIT_TIMEOUT_USEC);
- if (!g_cond_timed_wait (link_main_cond, link_main_lock,
>ime)) {
- if (link_is_locked ())
- link_unlock ();
- return FALSE;
- }
+ g_cond_wait (link_main_cond, link_main_lock);
}
-
- return TRUE;
}
@@ -568,3 +554,34 @@
{
return link_mutex_is_locked (link_main_lock);
}
+
+/* Hack */
+guint
+link_io_thread_add_timeout (guint interval,
+ GSourceFunc function,
+ gpointer data)
+{
+ guint id;
+ GSource *tsrc;
+
+ tsrc = g_timeout_source_new (interval);
+ g_source_set_priority (tsrc, G_PRIORITY_HIGH_IDLE);
+ g_source_set_callback (tsrc, function, data, NULL);
+ g_source_set_can_recurse (tsrc, TRUE);
+ id = g_source_attach (tsrc, link_thread_context);
+ g_source_unref (tsrc);
+
+ return id;
+}
+
+void
+link_io_thread_remove_timeout (guint source_id)
+{
+ GSource *tsrc;
+
+ if (!source_id)
+ return;
+
+ tsrc = g_main_context_find_source_by_id (link_thread_context,
source_id);
+ g_source_destroy (tsrc);
+}
Index: linc2/ChangeLog
===================================================================
--- linc2/ChangeLog (revision 2003)
+++ linc2/ChangeLog (working copy)
@@ -1,3 +1,24 @@
+2007-06-19 Jules Colding <[EMAIL PROTECTED]>
+
+ * src/linc-connection.c (link_connection_init): Initialize
+ timeout members in the LinkConnection structure.
+ (link_set_timeout): New function to set the timeout value.
+
+2007-06-18 Jules Colding <[EMAIL PROTECTED]>
+
+ * src/linc-connection.c (link_connection_from_fd_T): Initiate
+ timeout members of the link connection iff:
+ 1) The connection is IPv4 or IPv6
+ 2) It is not a oneway
+ 3) The timeout parameter is non-zero
+
+ * include/linc/linc-connection.h (struct):
+ 1) timeout mutex
+ 2) timeout value in milliseconds
+ 3) timeout status
+
+ Furthermore declare link_set_timeout()
+
========================== ORBit2-2.14.8 ========================
2007-02-27 Kjartan Maraas <[EMAIL PROTECTED]>
Index: include/orbit/GIOP/giop-connection.h
===================================================================
--- include/orbit/GIOP/giop-connection.h (revision 2003)
+++ include/orbit/GIOP/giop-connection.h (working copy)
@@ -53,6 +53,9 @@
#define giop_connection_ref(cnx) link_connection_ref(cnx)
#define giop_connection_unref(cnx) link_connection_unref(cnx)
+/* set the link timeout in milliseconds */
+extern void giop_set_timeout (guint msec);
+
#endif /* ORBIT2_INTERNAL_API */
G_END_DECLS
Index: include/orbit/GIOP/giop.h
===================================================================
--- include/orbit/GIOP/giop.h (revision 2003)
+++ include/orbit/GIOP/giop.h (working copy)
@@ -23,7 +23,6 @@
gboolean giop_thread_io (void);
GIOPThread *giop_thread_self (void);
void giop_invoke_async (GIOPMessageQueueEntry *ent);
-void giop_recv_set_timeout (const glong timeout);
void giop_recv_set_limit (glong limit);
glong giop_recv_get_limit (void);
void giop_incoming_signal_T (GIOPThread *tdata, GIOPMsgType t);
@@ -47,7 +46,6 @@
gboolean giop_thread_queue_empty_T (GIOPThread *tdata);
void giop_thread_queue_tail_wakeup(GIOPThread *tdata);
-
#endif /* ORBIT2_INTERNAL_API */
G_END_DECLS
Index: include/orbit/GIOP/giop-types.h
===================================================================
--- include/orbit/GIOP/giop-types.h (revision 2003)
+++ include/orbit/GIOP/giop-types.h (working copy)
@@ -35,10 +35,8 @@
gpointer dummy);
};
-#define GIOP_INITIAL_TIMEOUT_LIMIT (30000000) /* 30 seconds */
+#define GIOP_INITIAL_MSG_SIZE_LIMIT (256*1024)
-#define GIOP_INITIAL_MSG_SIZE_LIMIT (256*1024) /* in bytes */
-
typedef enum {
GIOP_CONNECTION_SSL
} GIOPConnectionOptions;
Index: include/orbit/GIOP/giop-recv-buffer.h
===================================================================
--- include/orbit/GIOP/giop-recv-buffer.h (revision 2003)
+++ include/orbit/GIOP/giop-recv-buffer.h (working copy)
@@ -74,7 +74,9 @@
void giop_recv_list_zap (GIOPConnection
*cnx);
gboolean giop_connection_handle_input (LinkConnection
*lcnx);
void giop_connection_destroy_frags (GIOPConnection
*cnx);
+extern void giop_timeout_add (GIOPConnection
*cnx);
+
#endif /* ORBIT2_INTERNAL_API */
G_END_DECLS
_______________________________________________
orbit-list mailing list
[email protected]
http://mail.gnome.org/mailman/listinfo/orbit-list
