@Test public void testReconnect() throws Exception { IStoreClient<String, String> client0 = syncManagers[0].getStoreClient("global", String.class, String.class); IStoreClient<String, String> client1 = syncManagers[1].getStoreClient("global", String.class, String.class); IStoreClient<String, String> client2 = syncManagers[2].getStoreClient("global", String.class, String.class); client0.put("key0", "value0"); waitForValue(client2, "key0", "value0", 1000, "client0"); logger.info("Shutting down server ID 1"); syncManagers[0].shutdown(); client1.put("newkey1", "newvalue1"); client2.put("newkey2", "newvalue2"); client1.put("key0", "newvalue0"); client2.put("key2", "newvalue2"); for (int i = 0; i < 500; i++) { client2.put("largetest" + i, "largetestvalue"); } logger.info("Initializing server ID 1"); syncManagers[0] = new SyncManager(); setupSyncManager(moduleContexts[0], syncManagers[0], nodes.get(0)); waitForFullMesh(2000); client0 = syncManagers[0].getStoreClient("global", String.class, String.class); waitForValue(client0, "newkey1", "newvalue1", 1000, "client0"); waitForValue(client0, "newkey2", "newvalue2", 1000, "client0"); waitForValue(client0, "key0", "newvalue0", 1000, "client0"); waitForValue(client0, "key2", "newvalue2", 1000, "client0"); for (int i = 0; i < 500; i++) { waitForValue(client0, "largetest" + i, "largetestvalue", 1000, "client0"); } }
@Override public void shutdown() { logger.debug("Shutting down Sync Manager: {} {}", clusterConfig.getNode().getHostname(), clusterConfig.getNode().getPort()); if (rpcService != null) { rpcService.shutdown(); } if (hintThreadPool != null) { hintThreadPool.shutdown(); } if (storeRegistry != null) { storeRegistry.shutdown(); } hintThreadPool = null; rpcService = null; }
SyncManager extends AbstractSyncManager { @Override public void shutdown() { logger.debug("Shutting down Sync Manager: {} {}", clusterConfig.getNode().getHostname(), clusterConfig.getNode().getPort()); if (rpcService != null) { rpcService.shutdown(); } if (hintThreadPool != null) { hintThreadPool.shutdown(); } if (storeRegistry != null) { storeRegistry.shutdown(); } hintThreadPool = null; rpcService = null; } }
SyncManager extends AbstractSyncManager { @Override public void shutdown() { logger.debug("Shutting down Sync Manager: {} {}", clusterConfig.getNode().getHostname(), clusterConfig.getNode().getPort()); if (rpcService != null) { rpcService.shutdown(); } if (hintThreadPool != null) { hintThreadPool.shutdown(); } if (storeRegistry != null) { storeRegistry.shutdown(); } hintThreadPool = null; rpcService = null; }  }
SyncManager extends AbstractSyncManager { @Override public void shutdown() { logger.debug("Shutting down Sync Manager: {} {}", clusterConfig.getNode().getHostname(), clusterConfig.getNode().getPort()); if (rpcService != null) { rpcService.shutdown(); } if (hintThreadPool != null) { hintThreadPool.shutdown(); } if (storeRegistry != null) { storeRegistry.shutdown(); } hintThreadPool = null; rpcService = null; }  @Override void registerStore(String storeName, Scope scope); @Override void registerPersistentStore(String storeName, Scope scope); ClusterConfig getClusterConfig(); void cleanup(); @LogMessageDoc(level="INFO", message="[{id}->{id}] Synchronizing local state to remote node", explanation="Normal state resynchronization is occurring") 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(); @LogMessageDoc(level="WARN", message="Sync task queue full and not emptying", explanation="The synchronization service is overloaded", recommendation=LogMessageDoc.CHECK_CONTROLLER) 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 { @Override public void shutdown() { logger.debug("Shutting down Sync Manager: {} {}", clusterConfig.getNode().getHostname(), clusterConfig.getNode().getPort()); if (rpcService != null) { rpcService.shutdown(); } if (hintThreadPool != null) { hintThreadPool.shutdown(); } if (storeRegistry != null) { storeRegistry.shutdown(); } hintThreadPool = null; rpcService = null; }  @Override void registerStore(String storeName, Scope scope); @Override void registerPersistentStore(String storeName, Scope scope); ClusterConfig getClusterConfig(); void cleanup(); @LogMessageDoc(level="INFO", message="[{id}->{id}] Synchronizing local state to remote node", explanation="Normal state resynchronization is occurring") 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(); @LogMessageDoc(level="WARN", message="Sync task queue full and not emptying", explanation="The synchronization service is overloaded", recommendation=LogMessageDoc.CHECK_CONTROLLER) 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; }