Merge pull request #305 from zhenlineo/1.1-routing

Remove server from routingTable if failed to establish connection

Author: Zhen Li

driver/src/main/java/org/neo4j/driver/internal/cluster/ClusterRoutingTable.java

@@ -0,0 +1,129 @@
+/*
+ * Copyright (c) 2002-2017 "Neo Technology,"
+ * Network Engine for Objects in Lund AB [http://neotechnology.com]
+ *
+ * This file is part of Neo4j.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.neo4j.driver.internal.cluster;
+
+import java.util.HashSet;
+import java.util.Set;
+
+import org.neo4j.driver.internal.net.BoltServerAddress;
+import org.neo4j.driver.internal.util.Clock;
+
+import static java.lang.String.format;
+import static java.util.Arrays.asList;
+
+public class ClusterRoutingTable implements RoutingTable
+{
+    private static final int MIN_ROUTERS = 1;
+
+    private final Clock clock;
+    private long expirationTimeout;
+    private final RoundRobinAddressSet readers;
+    private final RoundRobinAddressSet writers;
+    private final RoundRobinAddressSet routers;
+
+    public ClusterRoutingTable( Clock clock, BoltServerAddress... routingAddresses )
+    {
+        this( clock );
+        routers.update( new HashSet<>( asList( routingAddresses ) ), new HashSet<BoltServerAddress>() );
+    }
+
+    private ClusterRoutingTable( Clock clock )
+    {
+        this.clock = clock;
+        this.expirationTimeout = clock.millis() - 1;
+
+        this.readers = new RoundRobinAddressSet();
+        this.writers = new RoundRobinAddressSet();
+        this.routers = new RoundRobinAddressSet();
+    }
+
+    @Override
+    public boolean isStale()
+    {
+        return expirationTimeout < clock.millis() || // the expiration timeout has been reached
+               routers.size() <= MIN_ROUTERS || // we need to discover more routing servers
+               readers.size() == 0 || // we need to discover more read servers
+               writers.size() == 0; // we need to discover more write servers
+    }
+
+    @Override
+    public synchronized Set<BoltServerAddress> update( ClusterComposition cluster )
+    {
+        expirationTimeout = cluster.expirationTimestamp;
+        HashSet<BoltServerAddress> removed = new HashSet<>();
+        readers.update( cluster.readers(), removed );
+        writers.update( cluster.writers(), removed );
+        routers.update( cluster.routers(), removed );
+        return removed;
+    }
+
+    @Override
+    public synchronized void forget( BoltServerAddress address )
+    {
+        // Don't remove it from the set of routers, since that might mean we lose our ability to re-discover,
+        // just remove it from the set of readers and writers, so that we don't use it for actual work without
+        // performing discovery first.
+        readers.remove( address );
+        writers.remove( address );
+    }
+
+    @Override
+    public RoundRobinAddressSet readers()
+    {
+        return readers;
+    }
+
+    @Override
+    public RoundRobinAddressSet writers()
+    {
+        return writers;
+    }
+
+    @Override
+    public BoltServerAddress nextRouter()
+    {
+        return routers.next();
+    }
+
+    @Override
+    public int routerSize()
+    {
+        return routers.size();
+    }
+
+    @Override
+    public void removeWriter( BoltServerAddress toRemove )
+    {
+        writers.remove( toRemove );
+    }
+
+    @Override
+    public void removeRouter( BoltServerAddress toRemove )
+    {
+        routers.remove( toRemove );
+    }
+
+    @Override
+    public String toString()
+    {
+        return format( "Ttl %s, currentTime %s, routers %s, writers %s, readers %s",
+                expirationTimeout, clock.millis(), routers, writers, readers );
+    }
+}

driver/src/main/java/org/neo4j/driver/internal/cluster/LoadBalancer.java

@@ -18,7 +18,7 @@
  */
 package org.neo4j.driver.internal.cluster;
 
-import java.util.HashSet;
+import java.util.Set;
 
 import org.neo4j.driver.internal.RoutingErrorHandler;
 import org.neo4j.driver.internal.net.BoltServerAddress;
@@ -28,21 +28,15 @@
 import org.neo4j.driver.v1.Logger;
 import org.neo4j.driver.v1.exceptions.ServiceUnavailableException;
 
-import static java.util.Arrays.asList;
+import static java.lang.String.format;
 
 public final class LoadBalancer implements RoutingErrorHandler, AutoCloseable
 {
-    private static final int MIN_ROUTERS = 1;
-    private static final String NO_ROUTERS_AVAILABLE = "Could not perform discovery. No routing servers available.";
-    // dependencies
-    private final RoutingSettings settings;
-    private final Clock clock;
     private final Logger log;
+
     private final ConnectionPool connections;
-    private final ClusterComposition.Provider provider;
-    // state
-    private long expirationTimeout;
-    private final RoundRobinAddressSet readers, writers, routers;
+    private final RoutingTable routingTable;
+    private final Rediscovery rediscovery;
 
     public LoadBalancer(
             RoutingSettings settings,
@@ -51,40 +45,41 @@ public LoadBalancer(
             ConnectionPool connections,
             BoltServerAddress... routingAddresses ) throws ServiceUnavailableException
     {
-        this( settings, clock, log, connections, new ClusterComposition.Provider.Default( clock, log ),
-                routingAddresses );
+        this( settings, clock, log, connections, new ClusterRoutingTable( clock, routingAddresses ),
+                new ClusterComposition.Provider.Default( clock, log ) );
     }
 
-    LoadBalancer(
+    private LoadBalancer(
             RoutingSettings settings,
             Clock clock,
             Logger log,
             ConnectionPool connections,
-            ClusterComposition.Provider provider,
-            BoltServerAddress... routingAddresses ) throws ServiceUnavailableException
+            RoutingTable routingTable,
+            ClusterComposition.Provider provider ) throws ServiceUnavailableException
+    {
+        this( log, connections, routingTable, new Rediscovery( settings, clock, log, provider ) );
+    }
+
+    LoadBalancer( Logger log, ConnectionPool connections, RoutingTable routingTable, Rediscovery rediscovery )
+            throws ServiceUnavailableException
     {
-        this.clock = clock;
         this.log = log;
         this.connections = connections;
-        this.expirationTimeout = clock.millis() - 1;
-        this.provider = provider;
-        this.settings = settings;
-        this.readers = new RoundRobinAddressSet();
-        this.writers = new RoundRobinAddressSet();
-        this.routers = new RoundRobinAddressSet();
-        routers.update( new HashSet<>( asList( routingAddresses ) ), new HashSet<BoltServerAddress>() );
+        this.routingTable = routingTable;
+        this.rediscovery = rediscovery;
+
         // initialize the routing table
         ensureRouting();
     }
 
     public Connection acquireReadConnection() throws ServiceUnavailableException
     {
-        return acquireConnection( readers );
+        return acquireConnection( routingTable.readers() );
     }
 
     public Connection acquireWriteConnection() throws ServiceUnavailableException
     {
-        return acquireConnection( writers );
+        return acquireConnection( routingTable.writers() );
     }
 
     @Override
@@ -96,7 +91,7 @@ public void onConnectionFailure( BoltServerAddress address )
     @Override
     public void onWriteFailure( BoltServerAddress address )
     {
-        writers.remove( address );
+        routingTable.removeWriter( address );
     }
 
     @Override
@@ -119,7 +114,7 @@ private Connection acquireConnection( RoundRobinAddressSet servers ) throws Serv
                 }
                 catch ( ServiceUnavailableException e )
                 {
-                    log.error( String.format( "Failed to refresh routing information using routing address %s",
+                    log.error( format( "Failed to refresh routing information using routing address %s",
                             address ), e );
 
                     forget( address );
@@ -129,29 +124,31 @@ private Connection acquireConnection( RoundRobinAddressSet servers ) throws Serv
         }
     }
 
+    private synchronized void forget( BoltServerAddress address )
+    {
+        // First remove from the load balancer, to prevent concurrent threads from making connections to them.
+        routingTable.forget( address );
+        // drop all current connections to the address
+        connections.purge( address );
+    }
+
     private synchronized void ensureRouting() throws ServiceUnavailableException
     {
-        if ( stale() )
+        if ( routingTable.isStale() )
         {
-            log.info( "Routing information is stale. Ttl %s, currentTime %s, routers %s, writers %s, readers %s",
-                    expirationTimeout, clock.millis(), routers, writers, readers );
+            log.info( "Routing information is stale. %s", routingTable );
             try
             {
                 // get a new routing table
-                ClusterComposition cluster = lookupRoutingTable();
-                expirationTimeout = cluster.expirationTimestamp;
-                HashSet<BoltServerAddress> removed = new HashSet<>();
-                readers.update( cluster.readers(), removed );
-                writers.update( cluster.writers(), removed );
-                routers.update( cluster.routers(), removed );
+                ClusterComposition cluster = rediscovery.lookupRoutingTable( connections, routingTable );
+                Set<BoltServerAddress> removed = routingTable.update( cluster );
                 // purge connections to removed addresses
                 for ( BoltServerAddress address : removed )
                 {
                     connections.purge( address );
                 }
 
-                log.info( "Refreshed routing information. Ttl %s, routers %s, writers %s, readers %s",
-                        expirationTimeout, routers, writers, readers );
+                log.info( "Refreshed routing information. %s", routingTable );
             }
             catch ( InterruptedException e )
             {
@@ -160,79 +157,5 @@ private synchronized void ensureRouting() throws ServiceUnavailableException
         }
     }
 
-    private ClusterComposition lookupRoutingTable() throws InterruptedException, ServiceUnavailableException
-    {
-        int size = routers.size(), failures = 0;
-        if ( size == 0 )
-        {
-            throw new ServiceUnavailableException( NO_ROUTERS_AVAILABLE );
-        }
-        for ( long start = clock.millis(), delay = 0; ; delay = Math.max( settings.retryTimeoutDelay, delay * 2 ) )
-        {
-            long waitTime = start + delay - clock.millis();
-            if ( waitTime > 0 )
-            {
-                clock.sleep( waitTime );
-            }
-            start = clock.millis();
-            for ( int i = 0; i < size; i++ )
-            {
-                BoltServerAddress address = routers.next();
-                if ( address == null )
-                {
-                    throw new ServiceUnavailableException( NO_ROUTERS_AVAILABLE );
-                }
-                ClusterComposition cluster;
-                try ( Connection connection = connections.acquire( address ) )
-                {
-                    cluster = provider.getClusterComposition( connection );
-                    log.info( "Got cluster composition %s", cluster );
-                }
-                catch ( Exception e )
-                {
-                    log.error( String.format( "Failed to connect to routing server '%s'.", address ), e );
-                    continue;
-                }
-                if ( cluster == null || !cluster.isValid() )
-                {
-                    log.info(
-                            "Server <%s> unable to perform routing capability, dropping from list of routers.",
-                            address );
-                    routers.remove( address );
-                    if ( --size == 0 )
-                    {
-                        throw new ServiceUnavailableException( NO_ROUTERS_AVAILABLE );
-                    }
-                }
-                else
-                {
-                    return cluster;
-                }
-            }
-            if ( ++failures >= settings.maxRoutingFailures )
-            {
-                throw new ServiceUnavailableException( NO_ROUTERS_AVAILABLE );
-            }
-        }
-    }
-
-    private synchronized void forget( BoltServerAddress address )
-    {
-        // First remove from the load balancer, to prevent concurrent threads from making connections to them.
-        // Don't remove it from the set of routers, since that might mean we lose our ability to re-discover,
-        // just remove it from the set of readers and writers, so that we don't use it for actual work without
-        // performing discovery first.
-        readers.remove( address );
-        writers.remove( address );
-        // drop all current connections to the address
-        connections.purge( address );
-    }
 
-    private boolean stale()
-    {
-        return expirationTimeout < clock.millis() || // the expiration timeout has been reached
-                routers.size() <= MIN_ROUTERS || // we need to discover more routing servers
-                readers.size() == 0 || // we need to discover more read servers
-                writers.size() == 0; // we need to discover more write servers
-    }
 }