@Test public void testChangeNode() throws Exception { waitForFullMesh(2000); IStoreClient<String, String> client0 = syncManagers[0].getStoreClient("global", String.class, String.class); IStoreClient<String, String> client2 = syncManagers[2].getStoreClient("global", String.class, String.class); client0.put("key", "value"); waitForValue(client2, "key", "value", 2000, "client2"); nodes.set(2, new Node("localhost", 50103, (short)3, (short)1)); nodeString = mapper.writeValueAsString(nodes); for(int i = 0; i < syncManagers.length; i++) { moduleContexts[i].addConfigParam(syncManagers[i], "nodes", nodeString); syncManagers[i].doUpdateConfiguration(); } waitForFullMesh(2000); waitForValue(client2, "key", "value", 2000, "client2"); client2 = syncManagers[2].getStoreClient("global", String.class, String.class); client0.put("key", "newvalue"); waitForValue(client2, "key", "newvalue", 2000, "client2"); }
protected void doUpdateConfiguration() throws FloodlightModuleException { try { ClusterConfig oldConfig = clusterConfig; clusterConfig = clusterConfigProvider.getConfig(); if (clusterConfig.equals(oldConfig)) return; logger.info("[{}] Updating sync configuration {}", clusterConfig.getNode().getNodeId(), clusterConfig); if (oldConfig.getNode() != null && !clusterConfig.getNode().equals(oldConfig.getNode())) { logger.info("[{}] Local node configuration changed; restarting sync" + "service", oldConfig.getNode().getNodeId()); shutdown(); startUp(null); } for (Node n : clusterConfig.getNodes()) { Node existing = oldConfig.getNode(n.getNodeId()); if (existing != null && !n.equals(existing)) { logger.info("[{}->{}] Configuration for node has changed", getLocalNodeId(), n.getNodeId()); rpcService.disconnectNode(n.getNodeId()); } } for (Node n : oldConfig.getNodes()) { Node nn = clusterConfig.getNode(n.getNodeId()); if (nn == null) { logger.info("[{}->{}] Disconnecting deconfigured node", getLocalNodeId(), n.getNodeId()); rpcService.disconnectNode(n.getNodeId()); } } } catch (Exception e) { throw new FloodlightModuleException("Could not update " + "configuration", e); } }
SyncManager extends AbstractSyncManager { protected void doUpdateConfiguration() throws FloodlightModuleException { try { ClusterConfig oldConfig = clusterConfig; clusterConfig = clusterConfigProvider.getConfig(); if (clusterConfig.equals(oldConfig)) return; logger.info("[{}] Updating sync configuration {}", clusterConfig.getNode().getNodeId(), clusterConfig); if (oldConfig.getNode() != null && !clusterConfig.getNode().equals(oldConfig.getNode())) { logger.info("[{}] Local node configuration changed; restarting sync" + "service", oldConfig.getNode().getNodeId()); shutdown(); startUp(null); } for (Node n : clusterConfig.getNodes()) { Node existing = oldConfig.getNode(n.getNodeId()); if (existing != null && !n.equals(existing)) { logger.info("[{}->{}] Configuration for node has changed", getLocalNodeId(), n.getNodeId()); rpcService.disconnectNode(n.getNodeId()); } } for (Node n : oldConfig.getNodes()) { Node nn = clusterConfig.getNode(n.getNodeId()); if (nn == null) { logger.info("[{}->{}] Disconnecting deconfigured node", getLocalNodeId(), n.getNodeId()); rpcService.disconnectNode(n.getNodeId()); } } } catch (Exception e) { throw new FloodlightModuleException("Could not update " + "configuration", e); } } }
SyncManager extends AbstractSyncManager { protected void doUpdateConfiguration() throws FloodlightModuleException { try { ClusterConfig oldConfig = clusterConfig; clusterConfig = clusterConfigProvider.getConfig(); if (clusterConfig.equals(oldConfig)) return; logger.info("[{}] Updating sync configuration {}", clusterConfig.getNode().getNodeId(), clusterConfig); if (oldConfig.getNode() != null && !clusterConfig.getNode().equals(oldConfig.getNode())) { logger.info("[{}] Local node configuration changed; restarting sync" + "service", oldConfig.getNode().getNodeId()); shutdown(); startUp(null); } for (Node n : clusterConfig.getNodes()) { Node existing = oldConfig.getNode(n.getNodeId()); if (existing != null && !n.equals(existing)) { logger.info("[{}->{}] Configuration for node has changed", getLocalNodeId(), n.getNodeId()); rpcService.disconnectNode(n.getNodeId()); } } for (Node n : oldConfig.getNodes()) { Node nn = clusterConfig.getNode(n.getNodeId()); if (nn == null) { logger.info("[{}->{}] Disconnecting deconfigured node", getLocalNodeId(), n.getNodeId()); rpcService.disconnectNode(n.getNodeId()); } } } catch (Exception e) { throw new FloodlightModuleException("Could not update " + "configuration", e); } }  }
SyncManager extends AbstractSyncManager { protected void doUpdateConfiguration() throws FloodlightModuleException { try { ClusterConfig oldConfig = clusterConfig; clusterConfig = clusterConfigProvider.getConfig(); if (clusterConfig.equals(oldConfig)) return; logger.info("[{}] Updating sync configuration {}", clusterConfig.getNode().getNodeId(), clusterConfig); if (oldConfig.getNode() != null && !clusterConfig.getNode().equals(oldConfig.getNode())) { logger.info("[{}] Local node configuration changed; restarting sync" + "service", oldConfig.getNode().getNodeId()); shutdown(); startUp(null); } for (Node n : clusterConfig.getNodes()) { Node existing = oldConfig.getNode(n.getNodeId()); if (existing != null && !n.equals(existing)) { logger.info("[{}->{}] Configuration for node has changed", getLocalNodeId(), n.getNodeId()); rpcService.disconnectNode(n.getNodeId()); } } for (Node n : oldConfig.getNodes()) { Node nn = clusterConfig.getNode(n.getNodeId()); if (nn == null) { logger.info("[{}->{}] Disconnecting deconfigured node", getLocalNodeId(), n.getNodeId()); rpcService.disconnectNode(n.getNodeId()); } } } catch (Exception e) { throw new FloodlightModuleException("Could not update " + "configuration", e); } }  @Override void registerStore(String storeName, Scope scope); @Override void registerPersistentStore(String storeName, Scope scope); ClusterConfig getClusterConfig(); void cleanup(); void antientropy(Node node); void antientropy(); void writeSyncValue(String storeName, Scope scope,
			boolean persist,
			byte[] key, Iterable<Versioned<byte[]>> values); boolean handleSyncOffer(String storeName,
			byte[] key,
			Iterable<VectorClock> versions); IStorageEngine<ByteArray, byte[]> getRawStore(String storeName); IThreadPoolService getThreadPool(); void queueSyncTask(SynchronizingStorageEngine e,
			ByteArray key, Versioned<byte[]> value); @Override void addListener(String storeName, MappingStoreListener listener); void updateConfiguration(); Cursor getCursor(int cursorId); Cursor newCursor(String storeName); void closeCursor(Cursor cursor); @Override IStore<ByteArray,byte[]> getStore(String storeName); @Override short getLocalNodeId(); @Override void shutdown(); @Override void init(FloodlightModuleContext context); @Override void startUp(FloodlightModuleContext context); @Override Collection<Class<? extends IFloodlightService>> getModuleDependencies(); }
SyncManager extends AbstractSyncManager { protected void doUpdateConfiguration() throws FloodlightModuleException { try { ClusterConfig oldConfig = clusterConfig; clusterConfig = clusterConfigProvider.getConfig(); if (clusterConfig.equals(oldConfig)) return; logger.info("[{}] Updating sync configuration {}", clusterConfig.getNode().getNodeId(), clusterConfig); if (oldConfig.getNode() != null && !clusterConfig.getNode().equals(oldConfig.getNode())) { logger.info("[{}] Local node configuration changed; restarting sync" + "service", oldConfig.getNode().getNodeId()); shutdown(); startUp(null); } for (Node n : clusterConfig.getNodes()) { Node existing = oldConfig.getNode(n.getNodeId()); if (existing != null && !n.equals(existing)) { logger.info("[{}->{}] Configuration for node has changed", getLocalNodeId(), n.getNodeId()); rpcService.disconnectNode(n.getNodeId()); } } for (Node n : oldConfig.getNodes()) { Node nn = clusterConfig.getNode(n.getNodeId()); if (nn == null) { logger.info("[{}->{}] Disconnecting deconfigured node", getLocalNodeId(), n.getNodeId()); rpcService.disconnectNode(n.getNodeId()); } } } catch (Exception e) { throw new FloodlightModuleException("Could not update " + "configuration", e); } }  @Override void registerStore(String storeName, Scope scope); @Override void registerPersistentStore(String storeName, Scope scope); ClusterConfig getClusterConfig(); void cleanup(); void antientropy(Node node); void antientropy(); void writeSyncValue(String storeName, Scope scope,
			boolean persist,
			byte[] key, Iterable<Versioned<byte[]>> values); boolean handleSyncOffer(String storeName,
			byte[] key,
			Iterable<VectorClock> versions); IStorageEngine<ByteArray, byte[]> getRawStore(String storeName); IThreadPoolService getThreadPool(); void queueSyncTask(SynchronizingStorageEngine e,
			ByteArray key, Versioned<byte[]> value); @Override void addListener(String storeName, MappingStoreListener listener); void updateConfiguration(); Cursor getCursor(int cursorId); Cursor newCursor(String storeName); void closeCursor(Cursor cursor); @Override IStore<ByteArray,byte[]> getStore(String storeName); @Override short getLocalNodeId(); @Override void shutdown(); @Override void init(FloodlightModuleContext context); @Override void startUp(FloodlightModuleContext context); @Override Collection<Class<? extends IFloodlightService>> getModuleDependencies(); static IDebugCounter counterHints; static IDebugCounter counterSentValues; static IDebugCounter counterReceivedValues; static IDebugCounter counterPuts; static IDebugCounter counterGets; static IDebugCounter counterIterators; static IDebugCounter counterErrorRemote; static IDebugCounter counterErrorProcessing; }