From 4295a66c70e595d02e14f384432d8eee8dfef013 Mon Sep 17 00:00:00 2001 From: Thom Nichols Date: Wed, 2 Jun 2010 14:18:09 -0400 Subject: reconnection quiesce logic --- sleekxmpp/xmlstream/xmlstream.py | 36 ++++++++++++++++++++++++++++++------ 1 file changed, 30 insertions(+), 6 deletions(-) (limited to 'sleekxmpp/xmlstream/xmlstream.py') diff --git a/sleekxmpp/xmlstream/xmlstream.py b/sleekxmpp/xmlstream/xmlstream.py index 3bcb3412..e0b3a23c 100644 --- a/sleekxmpp/xmlstream/xmlstream.py +++ b/sleekxmpp/xmlstream/xmlstream.py @@ -16,6 +16,7 @@ from . stanzabase import StanzaBase from xml.etree import cElementTree from xml.parsers import expat import logging +import random import socket import threading import time @@ -46,6 +47,10 @@ class CloseStream(Exception): stanza_extensions = {} +RECONNECT_MAX_DELAY = 3600 +RECONNECT_QUIESCE_FACTOR = 1.6180339887498948 # Phi +RECONNECT_QUIESCE_JITTER = 0.11962656472 # molar Planck constant times c, joule meter/mole + class XMLStream(object): "A connection manager with XML events." @@ -101,7 +106,12 @@ class XMLStream(object): def connectTCP(self, host='', port=0, use_ssl=None, use_tls=None, reattempt=True): "Connect and create socket" - while reattempt and not self.state['connected']: # the self.state part is redundant. + + # Note that this is thread-safe by merit of being called solely from connect() which + # holds the state lock. + + delay = 1.0 # reconnection delay + while self.run: logging.debug('connecting....') try: if host and port: @@ -115,21 +125,35 @@ class XMLStream(object): else: self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.socket.settimeout(None) #10) + if self.use_ssl and self.ssl_support: logging.debug("Socket Wrapped for SSL") self.socket = ssl.wrap_socket(self.socket,ca_certs=self.ca_certs) - except: - logging.exception("Connection error") - try: + self.socket.connect(self.address) self.filesocket = self.socket.makefile('rb', 0) + if not self.state.transition('connecting','connected'): logging.error( "State transition error!!!! Shouldn't have happened" ) logging.debug('connect complete.') return True + except socket.error as serr: - logging.error("Could not connect. Socket Error #%s: %s" % (serr.errno, serr.strerror)) - time.sleep(1) # TODO proper quiesce if connection attempt fails + logging.exception("Socket Error #%s: %s", serr.errno, serr.strerror) + if not reattempt: return False + except: + logging.exception("Connection error") + if not reattempt: return False + + # quiesce if rconnection fails: + # This code based loosely on Twisted internet.protocol + # http://twistedmatrix.com/trac/browser/trunk/twisted/internet/protocol.py#L310 + delay = min(delay * RECONNECT_QUIESCE_JITTER, RECONNECT_MAX_DELAY) + delay = random.normalvariate(delay, delay * RECONNECT_QUIESCE_JITTER) + logging.debug('Waiting %fs until next reconnect attempt...', delay) + time.sleep(delay) + + def connectUnix(self, filepath): "Connect to Unix file and create socket" -- cgit v1.2.3 From 1c32668e18c0cbf840694c66a6f724b2d3cb8f29 Mon Sep 17 00:00:00 2001 From: Thom Nichols Date: Thu, 3 Jun 2010 07:47:27 -0400 Subject: fixed quiesce algorithm; state transition if connect fails; note about use_tls instance variable. --- sleekxmpp/xmlstream/xmlstream.py | 17 +++++++++++++---- 1 file changed, 13 insertions(+), 4 deletions(-) (limited to 'sleekxmpp/xmlstream/xmlstream.py') diff --git a/sleekxmpp/xmlstream/xmlstream.py b/sleekxmpp/xmlstream/xmlstream.py index e0b3a23c..af95c2bf 100644 --- a/sleekxmpp/xmlstream/xmlstream.py +++ b/sleekxmpp/xmlstream/xmlstream.py @@ -101,8 +101,16 @@ class XMLStream(object): def connect(self, host='', port=0, use_ssl=None, use_tls=None): "Link to connectTCP" - if self.state.transition('disconnected', 'connecting'): - return self.connectTCP(host, port, use_ssl, use_tls) + if not self.connectTCP(host, port, use_ssl, use_tls): + # return to the 'disconnected' state if connect failed: + # otherwise the connect method is not reentrant + if not self.state.transition('connecting','disconnected'): + logging.error("Couldn't transition to the 'disconnected' state!") + return False + return True + + # TODO currently a caller can't distinguish between "connection failed" and + # "we're already trying to connect from another thread" def connectTCP(self, host='', port=0, use_ssl=None, use_tls=None, reattempt=True): "Connect and create socket" @@ -119,6 +127,7 @@ class XMLStream(object): if use_ssl is not None: self.use_ssl = use_ssl if use_tls is not None: + # TODO this variable doesn't seem to be used for anything! self.use_tls = use_tls if sys.version_info < (3, 0): self.socket = filesocket.Socket26(socket.AF_INET, socket.SOCK_STREAM) @@ -146,9 +155,9 @@ class XMLStream(object): if not reattempt: return False # quiesce if rconnection fails: - # This code based loosely on Twisted internet.protocol + # This algorithm based loosely on Twisted internet.protocol # http://twistedmatrix.com/trac/browser/trunk/twisted/internet/protocol.py#L310 - delay = min(delay * RECONNECT_QUIESCE_JITTER, RECONNECT_MAX_DELAY) + delay = min(delay * RECONNECT_QUIESCE_FACTOR, RECONNECT_MAX_DELAY) delay = random.normalvariate(delay, delay * RECONNECT_QUIESCE_JITTER) logging.debug('Waiting %fs until next reconnect attempt...', delay) time.sleep(delay) -- cgit v1.2.3 From da6e1e47dc81f5f9579201644d7c18dd85510368 Mon Sep 17 00:00:00 2001 From: Thom Nichols Date: Thu, 3 Jun 2010 08:09:09 -0400 Subject: whups, somehow I lost the 'connecting' lock in connect() --- sleekxmpp/xmlstream/xmlstream.py | 4 ++++ 1 file changed, 4 insertions(+) (limited to 'sleekxmpp/xmlstream/xmlstream.py') diff --git a/sleekxmpp/xmlstream/xmlstream.py b/sleekxmpp/xmlstream/xmlstream.py index af95c2bf..76aecee4 100644 --- a/sleekxmpp/xmlstream/xmlstream.py +++ b/sleekxmpp/xmlstream/xmlstream.py @@ -101,6 +101,10 @@ class XMLStream(object): def connect(self, host='', port=0, use_ssl=None, use_tls=None): "Link to connectTCP" + if not self.state.transition('disconnected','connecting'): + logging.warning("Can't connect now; Already in state %s", self.state.current_state()) + return False + if not self.connectTCP(host, port, use_ssl, use_tls): # return to the 'disconnected' state if connect failed: # otherwise the connect method is not reentrant -- cgit v1.2.3 From d20cd6b3e6a7edacfef6010339d7fb7b587a9def Mon Sep 17 00:00:00 2001 From: Thom Nichols Date: Thu, 3 Jun 2010 13:51:11 -0400 Subject: added function execution on transition, and more unit tests. --- sleekxmpp/xmlstream/xmlstream.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) (limited to 'sleekxmpp/xmlstream/xmlstream.py') diff --git a/sleekxmpp/xmlstream/xmlstream.py b/sleekxmpp/xmlstream/xmlstream.py index 76aecee4..c407a335 100644 --- a/sleekxmpp/xmlstream/xmlstream.py +++ b/sleekxmpp/xmlstream/xmlstream.py @@ -281,7 +281,7 @@ class XMLStream(object): data = None try: - data = self.sendqueue.get(True,10) + data = self.sendqueue.get(True,5) logging.debug("SEND: %s" % data) self.socket.sendall(data.encode('utf-8')) except queue.Empty: @@ -372,6 +372,7 @@ class XMLStream(object): try: event = self.eventqueue.get(True, timeout=5) except queue.Empty: + logging.debug('Nothing on event queue') event = None if event is not None: etype = event[0] -- cgit v1.2.3 From 919c8c5633acd0d638f39c4a7ec8794fc5e94fff Mon Sep 17 00:00:00 2001 From: Thom Nichols Date: Thu, 3 Jun 2010 15:21:26 -0400 Subject: tweaked connectTCP call slightly to reduce possibility of 'connecting' state limbo --- sleekxmpp/xmlstream/xmlstream.py | 23 +++++++++++++---------- 1 file changed, 13 insertions(+), 10 deletions(-) (limited to 'sleekxmpp/xmlstream/xmlstream.py') diff --git a/sleekxmpp/xmlstream/xmlstream.py b/sleekxmpp/xmlstream/xmlstream.py index c407a335..cc5e1ec8 100644 --- a/sleekxmpp/xmlstream/xmlstream.py +++ b/sleekxmpp/xmlstream/xmlstream.py @@ -100,18 +100,20 @@ class XMLStream(object): self.filesocket = filesocket def connect(self, host='', port=0, use_ssl=None, use_tls=None): - "Link to connectTCP" + "Establish a socket connection to the given XMPP server." if not self.state.transition('disconnected','connecting'): logging.warning("Can't connect now; Already in state %s", self.state.current_state()) return False - if not self.connectTCP(host, port, use_ssl, use_tls): - # return to the 'disconnected' state if connect failed: - # otherwise the connect method is not reentrant - if not self.state.transition('connecting','disconnected'): - logging.error("Couldn't transition to the 'disconnected' state!") - return False - return True + try: + return self.connectTCP(host, port, use_ssl, use_tls) + finally: + # attempt to ensure once a connection attempt starts, we leave either in the + # 'connected' or 'disconnected' state. Otherwise the connect method is not reentrant + if self.state['connecting']: + if not self.state.transition('connecting','disconnected'): + logging.error("Couldn't return to the 'disconnected' state after connection failure!") + # TODO currently a caller can't distinguish between "connection failed" and # "we're already trying to connect from another thread" @@ -285,7 +287,8 @@ class XMLStream(object): logging.debug("SEND: %s" % data) self.socket.sendall(data.encode('utf-8')) except queue.Empty: - logging.debug('nothing on send queue') +# logging.debug('Nothing on send queue') + pass except socket.timeout: # this is to prevent a thread blocked indefinitely logging.debug('timeout sending packet data') @@ -372,7 +375,7 @@ class XMLStream(object): try: event = self.eventqueue.get(True, timeout=5) except queue.Empty: - logging.debug('Nothing on event queue') +# logging.debug('Nothing on event queue') event = None if event is not None: etype = event[0] -- cgit v1.2.3 From e7c37c4ec5a31402ba4d3fd95f2ee31d72183a83 Mon Sep 17 00:00:00 2001 From: Thom Nichols Date: Fri, 4 Jun 2010 17:00:51 -0400 Subject: connect uses the new function-on-state-transition so when the connect method returns you are guaranteed to be either in the 'connected' or 'disconnected' state. Could remove the 'connecting' state except uses it. --- sleekxmpp/xmlstream/xmlstream.py | 22 ++++++++-------------- 1 file changed, 8 insertions(+), 14 deletions(-) (limited to 'sleekxmpp/xmlstream/xmlstream.py') diff --git a/sleekxmpp/xmlstream/xmlstream.py b/sleekxmpp/xmlstream/xmlstream.py index cc5e1ec8..6dbe7b30 100644 --- a/sleekxmpp/xmlstream/xmlstream.py +++ b/sleekxmpp/xmlstream/xmlstream.py @@ -101,19 +101,16 @@ class XMLStream(object): def connect(self, host='', port=0, use_ssl=None, use_tls=None): "Establish a socket connection to the given XMPP server." - if not self.state.transition('disconnected','connecting'): - logging.warning("Can't connect now; Already in state %s", self.state.current_state()) + + if not self.state.transition('disconnected','connected', + func=self.connectTCP, args=[host, port, use_ssl, use_tls] ): + + if self.state['connected']: logging.debug('Already connected') + else: logging.warning("Connection failed" ) return False - try: - return self.connectTCP(host, port, use_ssl, use_tls) - finally: - # attempt to ensure once a connection attempt starts, we leave either in the - # 'connected' or 'disconnected' state. Otherwise the connect method is not reentrant - if self.state['connecting']: - if not self.state.transition('connecting','disconnected'): - logging.error("Couldn't return to the 'disconnected' state after connection failure!") - + logging.debug('Connection complete.') + return True # TODO currently a caller can't distinguish between "connection failed" and # "we're already trying to connect from another thread" @@ -148,9 +145,6 @@ class XMLStream(object): self.socket.connect(self.address) self.filesocket = self.socket.makefile('rb', 0) - if not self.state.transition('connecting','connected'): - logging.error( "State transition error!!!! Shouldn't have happened" ) - logging.debug('connect complete.') return True except socket.error as serr: -- cgit v1.2.3