Merge pull request #265 from lutovich/1.5-lb-infra

Extract load balancing logic in a separate class

Author: lutovich

src/v1/internal/connection-providers.js

@@ -20,12 +20,12 @@
 import {newError, SERVICE_UNAVAILABLE, SESSION_EXPIRED} from '../error';
 import {READ, WRITE} from '../driver';
 import Session from '../session';
-import RoundRobinArray from './round-robin-array';
 import RoutingTable from './routing-table';
 import Rediscovery from './rediscovery';
 import hasFeature from './features';
 import {DnsHostNameResolver, DummyHostNameResolver} from './host-name-resolvers';
 import RoutingUtil from './routing-util';
+import RoundRobinLoadBalancingStrategy from './round-robin-load-balancing-strategy';
 
 class ConnectionProvider {
 
@@ -65,20 +65,23 @@ export class LoadBalancer extends ConnectionProvider {
   constructor(address, routingContext, connectionPool, driverOnErrorCallback) {
     super();
     this._seedRouter = address;
-    this._routingTable = new RoutingTable(new RoundRobinArray([this._seedRouter]));
+    this._routingTable = new RoutingTable([this._seedRouter]);
     this._rediscovery = new Rediscovery(new RoutingUtil(routingContext));
     this._connectionPool = connectionPool;
     this._driverOnErrorCallback = driverOnErrorCallback;
     this._hostNameResolver = LoadBalancer._createHostNameResolver();
+    this._loadBalancingStrategy = new RoundRobinLoadBalancingStrategy();
     this._useSeedRouter = false;
   }
 
   acquireConnection(accessMode) {
     const connectionPromise = this._freshRoutingTable(accessMode).then(routingTable => {
       if (accessMode === READ) {
-        return this._acquireConnectionToServer(routingTable.readers, 'read');
+        const address = this._loadBalancingStrategy.selectReader(routingTable.readers);
+        return this._acquireConnectionToServer(address, 'read');
       } else if (accessMode === WRITE) {
-        return this._acquireConnectionToServer(routingTable.writers, 'write');
+        const address = this._loadBalancingStrategy.selectWriter(routingTable.writers);
+        return this._acquireConnectionToServer(address, 'write');
       } else {
         throw newError('Illegal mode ' + accessMode);
       }
@@ -95,8 +98,7 @@ export class LoadBalancer extends ConnectionProvider {
     this._routingTable.forgetWriter(address);
   }
 
-  _acquireConnectionToServer(serversRoundRobinArray, serverName) {
-    const address = serversRoundRobinArray.next();
+  _acquireConnectionToServer(address, serverName) {
     if (!address) {
       return Promise.reject(newError(
         `Failed to obtain connection towards ${serverName} server. Known routing table is: ${this._routingTable}`,
@@ -115,7 +117,7 @@ export class LoadBalancer extends ConnectionProvider {
   }
 
   _refreshRoutingTable(currentRoutingTable) {
-    const knownRouters = currentRoutingTable.routers.toArray();
+    const knownRouters = currentRoutingTable.routers;
 
     if (this._useSeedRouter) {
       return this._fetchRoutingTableFromSeedRouterFallbackToKnownRouters(knownRouters, currentRoutingTable);
@@ -215,7 +217,7 @@ export class LoadBalancer extends ConnectionProvider {
         SERVICE_UNAVAILABLE);
     }
 
-    if (newRoutingTable.writers.isEmpty()) {
+    if (newRoutingTable.writers.length === 0) {
       // use seed router next time. this is important when cluster is partitioned. it tries to make sure driver
       // does not always get routing table without writers because it talks exclusively to a minority partition
       this._useSeedRouter = true;

src/v1/internal/load-balancing-strategy.js

@@ -0,0 +1,43 @@
+/**
+ * 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.
+ */
+
+
+/**
+ * A facility to select most appropriate reader or writer among the given addresses for request processing.
+ */
+export default class LoadBalancingStrategy {
+
+  /**
+   * Select next most appropriate reader from the list of given readers.
+   * @param {string[]} knownReaders an array of currently known readers to select from.
+   * @return {string} most appropriate reader or <code>null</code> if given array is empty.
+   */
+  selectReader(knownReaders) {
+    throw new Error('Abstract function');
+  }
+
+  /**
+   * Select next most appropriate writer from the list of given writers.
+   * @param {string[]} knownWriters an array of currently known writers to select from.
+   * @return {string} most appropriate writer or <code>null</code> if given array is empty.
+   */
+  selectWriter(knownWriters) {
+    throw new Error('Abstract function');
+  }
+}

src/v1/internal/rediscovery.js

@@ -63,8 +63,8 @@ export default class Rediscovery {
     });
   }
 
-  static _assertNonEmpty(serversRoundRobinArray, serversName, routerAddress) {
-    if (serversRoundRobinArray.isEmpty()) {
+  static _assertNonEmpty(serverAddressesArray, serversName, routerAddress) {
+    if (serverAddressesArray.length === 0) {
       throw newError('Received no ' + serversName + ' from router ' + routerAddress, PROTOCOL_ERROR);
     }
   }

src/v1/internal/round-robin-array-index.js

@@ -0,0 +1,48 @@
+/**
+ * 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.
+ */
+
+export default class RoundRobinArrayIndex {
+
+  /**
+   * @constructor
+   * @param {number} [initialOffset=0] the initial offset for round robin.
+   */
+  constructor(initialOffset) {
+    this._offset = initialOffset || 0;
+  }
+
+  /**
+   * Get next index for an array with given length.
+   * @param {number} arrayLength the array length.
+   * @return {number} index in the array.
+   */
+  next(arrayLength) {
+    if (arrayLength === 0) {
+      return -1;
+    }
+
+    const nextOffset = this._offset;
+    this._offset += 1;
+    if (this._offset === Number.MAX_SAFE_INTEGER) {
+      this._offset = 0;
+    }
+
+    return nextOffset % arrayLength;
+  }
+}

src/v1/internal/round-robin-array.js

@@ -0,0 +1,52 @@
+/**
+ * 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.
+ */
+import RoundRobinArrayIndex from './round-robin-array-index';
+import LoadBalancingStrategy from './load-balancing-strategy';
+
+export default class RoundRobinLoadBalancingStrategy extends LoadBalancingStrategy {
+
+  constructor() {
+    super();
+    this._readersIndex = new RoundRobinArrayIndex();
+    this._writersIndex = new RoundRobinArrayIndex();
+  }
+
+  /**
+   * @inheritDoc
+   */
+  selectReader(knownReaders) {
+    return this._select(knownReaders, this._readersIndex);
+  }
+
+  /**
+   * @inheritDoc
+   */
+  selectWriter(knownWriters) {
+    return this._select(knownWriters, this._writersIndex);
+  }
+
+  _select(addresses, roundRobinIndex) {
+    const length = addresses.length;
+    if (length === 0) {
+      return null;
+    }
+    const index = roundRobinIndex.next(length);
+    return addresses[index];
+  }
+}

src/v1/internal/round-robin-load-balancing-strategy.js

@@ -17,34 +17,34 @@
  * limitations under the License.
  */
 import {int} from '../integer';
-import RoundRobinArray from './round-robin-array';
 import {READ, WRITE} from '../driver';
 
 const MIN_ROUTERS = 1;
 
 export default class RoutingTable {
 
   constructor(routers, readers, writers, expirationTime) {
-    this.routers = routers || new RoundRobinArray();
-    this.readers = readers || new RoundRobinArray();
-    this.writers = writers || new RoundRobinArray();
+    this.routers = routers || [];
+    this.readers = readers || [];
+    this.writers = writers || [];
     this.expirationTime = expirationTime || int(0);
   }
 
   forget(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.
-    this.readers.remove(address);
-    this.writers.remove(address);
+
+    this.readers = removeFromArray(this.readers, address);
+    this.writers = removeFromArray(this.writers, address);
   }
 
   forgetRouter(address) {
-    this.routers.remove(address);
+    this.routers = removeFromArray(this.routers, address);
   }
 
   forgetWriter(address) {
-    this.writers.remove(address);
+    this.writers = removeFromArray(this.writers, address);
   }
 
   serversDiff(otherRoutingTable) {
@@ -62,20 +62,30 @@ export default class RoutingTable {
    */
   isStaleFor(accessMode) {
     return this.expirationTime.lessThan(Date.now()) ||
-      this.routers.size() < MIN_ROUTERS ||
-      accessMode === READ && this.readers.isEmpty() ||
-      accessMode === WRITE && this.writers.isEmpty();
+      this.routers.length < MIN_ROUTERS ||
+      accessMode === READ && this.readers.length === 0 ||
+      accessMode === WRITE && this.writers.length === 0;
   }
 
   _allServers() {
-    return [...this.routers.toArray(), ...this.readers.toArray(), ...this.writers.toArray()];
+    return [...this.routers, ...this.readers, ...this.writers];
   }
 
   toString() {
     return `RoutingTable[` +
       `expirationTime=${this.expirationTime}, ` +
-      `routers=${this.routers}, ` +
-      `readers=${this.readers}, ` +
-      `writers=${this.writers}]`;
+      `routers=[${this.routers}], ` +
+      `readers=[${this.readers}], ` +
+      `writers=[${this.writers}]]`;
   }
 }
+
+/**
+ * Remove all occurrences of the element in the array.
+ * @param {Array} array the array to filter.
+ * @param {object} element the element to remove.
+ * @return {Array} new filtered array.
+ */
+function removeFromArray(array, element) {
+  return array.filter(item => item !== element);
+}