Patchwork [17,of,21,V2] speedy: fall back to local query if the history server times out

login
register
mail settings
Submitter Tomasz Kleczek
Date Dec. 14, 2012, 2:52 a.m.
Message ID <3a2c9270ca9847f9ae4f.1355453549@dev408.prn1.facebook.com>
Download mbox | patch
Permalink /patch/99/
State Deferred, archived
Headers show

Comments

Tomasz Kleczek - Dec. 14, 2012, 2:52 a.m.
# HG changeset patch
# User Tomasz Kleczek <tkleczek at fb.com>
# Date 1355360384 28800
# Node ID 3a2c9270ca9847f9ae4f09bfce2b1e112e567afc
# Parent  a8066abf4b9d2a1d751ce9eeeb2b334d86eb18c1
speedy: fall back to local query if the history server times out

The fact that history server is being used should ideally be transparent
to the user. In case of server being unreachable or slow we should just
compute the answer locally as if the server doesn't exist.

Patch

diff --git a/hgext/speedy/client.py b/hgext/speedy/client.py
--- a/hgext/speedy/client.py
+++ b/hgext/speedy/client.py
@@ -207,18 +207,21 @@ 
     revs.update(revset.date(repo, lrevsall, ('symbol', datestr)))
     return [ r for r in subset if r in revs ]
 
+ at tcptransport.timeout(revset.author)
 def patchedauthor(metapeer, repo, subset, x):
     """Used to monkey patch revset.author function."""
     # We want to catch errors early and on client, if possible
     pat = revset.getstring(x, _("author requires a string"))
     return _patchedauthor(metapeer, repo, subset, [pat])
 
+ at tcptransport.timeout(revset.date)
 def patcheddate(metapeer, repo, subset, d):
     """Used to monkey patch revset.date function."""
     # We want to catch errors early and on client, if possible
     ds = revset.getstring(d, _("date requires a string"))
     return _patcheddate(metapeer, repo, subset, ds)
 
+ at tcptransport.timeout(cmdutil.filterrevs)
 def patchedfilterrevs(metapeer, repo, revs, match):
     """Used to monkey patch cmdutil.filterrevs function."""
     wanted, fncache = metapeer.path(match)
@@ -228,6 +231,7 @@ 
     fncache.update(lfncache)
     return (wanted & set(revs)), fncache
 
+ at tcptransport.timeout(cmdutil.filterrevsopts)
 def patchedfilterrevsopts(metapeer, repo, revs, opts):
     users = opts.get('user')
     date = opts.get('date')
@@ -249,11 +253,12 @@ 
 
     serverrepopath = ui.config('speedy', 'serverrepo', '')
     host = ui.config('speedy', 'host', '')
+    timeout = ui.configint('speedy', 'timeout', 10)
 
     if host:
         if not serverrepopath:
             raise util.Abort(_("config option 'serverrepo' required by option 'host'"))
-        proxy = tcptransport.tcpclient(host, protocol.wireprotocol)
+        proxy = tcptransport.tcpclient(host, protocol.wireprotocol, timeout)
     else:
         if not serverrepopath:
             serverrepopath = repo.root
diff --git a/hgext/speedy/tcptransport.py b/hgext/speedy/tcptransport.py
--- a/hgext/speedy/tcptransport.py
+++ b/hgext/speedy/tcptransport.py
@@ -11,6 +11,25 @@ 
 import urlparse
 import transport
 
+def timeout(fallback):
+    """A decorator that makes a call to fallback if the function times out.
+
+    As far as this decorator is concerned the function times out if the
+    call raises `socket.timeout` or `socket.error` exception.
+
+    fallback: a callable to fall back to. It is called with the same
+        parameters as the original function.
+    """
+    def decorator(f):
+        def wrapper(metapeer, *args, **kwargs):
+            try:
+                return f(metapeer, *args, **kwargs)
+            except (socket.timeout, socket.error):
+                return fallback(*args, **kwargs)
+        wrapper.__name__ = f.__name__
+        return wrapper
+    return decorator
+
 def exactreader(read):
     """Return a function that reads and returns a string of the specified
     length.
@@ -44,12 +63,14 @@ 
 class tcpclient(transport.clientproxy):
     """Sends queries to server using TCP sockets directly."""
 
-    def __init__(self, uri, protoclass):
+    def __init__(self, uri, protoclass, timeout):
         parsed = urlparse.urlparse(uri, scheme='http')
         self.port = parsed.port
         self.host = parsed.hostname
         self.protoclass = protoclass
+        self.timeout = timeout
         self._sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        self._sock.settimeout(timeout)
 
     def request(self, queryname, args):
         """Send a single query to the server and return the response.
@@ -57,7 +78,10 @@ 
         The arguments (de)serialization is done behind the scenes by the
         provided protocol.
 
-        Blocks until the complete response is returned.
+        Blocks until the complete response is returned or the socket
+        times out.
+
+        Raises `socket.timeout` on timeout.
         """
         self._sock.connect((self.host, self.port))
         try:
diff --git a/tests/test-speedy.t b/tests/test-speedy.t
--- a/tests/test-speedy.t
+++ b/tests/test-speedy.t
@@ -328,3 +328,37 @@ 
   $ cat out
   listening on port 8123
   killed!
+
+  $ cd $TESTTMP/localrepo
+
+Testing unreachable server
+
+  $ hg log d1 --config 'speedy.host=http://localhost:15341'
+  chg6
+  chg8
+  chgl6
+  chg2
+  chg1
+  chg0
+
+Testing server timeout
+
+  $ (
+  > python $TESTDIR/timeouthistserver.py 2>err &
+  > SLOW_SERVER_PID=$!
+  > echo $SLOW_SERVER_PID > pidfile
+  > )
+  $ sleep 1
+
+  $ hg log d1 --config 'speedy.host=http://localhost:8877' \
+  > --config 'speedy.timeout=1'
+  chg6
+  chg8
+  chgl6
+  chg2
+  chg1
+  chg0
+
+  $ kill `cat pidfile` 2> /dev/null
+  $ cat err
+  handling request
diff --git a/tests/timeouthistserver.py b/tests/timeouthistserver.py
new file mode 100644
--- /dev/null
+++ b/tests/timeouthistserver.py
@@ -0,0 +1,25 @@ 
+import time
+import SocketServer
+import errno
+import sys
+
+
+class SlowHandler(SocketServer.StreamRequestHandler):
+    def setup(self):
+        sys.stderr.write('handling request\n')
+    def handle(self):
+        try:
+            time.sleep(4)
+        except:
+            pass
+
+host = 'localhost'
+port = 8877
+server = SocketServer.TCPServer((host, port), SlowHandler)
+
+# Just handle one request
+try:
+    server.handle_request()
+except IOError, e:
+    if e.error == errno.EPIPE:
+        pass