@Test public void testStress() throws NoSessionException, GetFailedException, PutFailedException, IOException, NoPeerConnectionException, AbortModifyException { File root = tempFolder.newFolder(); UserProfileManager profileManager = new UserProfileManager(client.getDataManager(), userCredentials); for (int i = 0; i < 10; i++) { final boolean isFolder = rnd.nextBoolean(); final File file = new File(root, randomString()); if (!isFolder) { FileUtils.writeStringToFile(file, randomString()); } final byte[] hash = HashUtil.hash(file); final KeyPair fileKeys = generateRSAKeyPair(RSA_KEYLENGTH.BIT_512); while (true) { String pid = UUID.randomUUID().toString(); profileManager.modifyUserProfile(pid, new IUserProfileModification() { @Override public void modifyUserProfile(UserProfile userProfile) { List<FolderIndex> indexes = getIndexList(userProfile.getRoot()); if (isFolder) { new FolderIndex(indexes.get(rnd.nextInt(indexes.size())), fileKeys, randomString()); } else { new FileIndex(indexes.get(rnd.nextInt(indexes.size())), fileKeys, file.getName(), hash); } } }); break; } } }
public void modifyUserProfile(String pid, IUserProfileModification modifier) throws GetFailedException, PutFailedException, AbortModifyException { PutQueueEntry entry = new PutQueueEntry(pid); modifyQueue.add(entry); synchronized (queueWaiter) { queueWaiter.notify(); } UserProfile profile; try { profile = entry.getUserProfile(); if (profile == null) { throw new GetFailedException("User Profile not found"); } } catch (GetFailedException e) { if (modifying != null && modifying.getPid().equals(pid)) { modifying.abort(); } throw e; } boolean retryPut = true; int forkCounter = 0; int forkWaitTime = new Random().nextInt(1000) + 500; while (retryPut) { modifier.modifyUserProfile(profile); try { if (protectionKeys == null) { protectionKeys = profile.getProtectionKeys(); } if (modifying != null && modifying.getPid().equals(pid)) { modifying.setUserProfile(profile); modifying.readyToPut(); modifying.waitForPut(); retryPut = false; } else { throw new PutFailedException("Not allowed to put anymore"); } } catch (VersionForkAfterPutException e) { if (forkCounter++ > FORK_LIMIT) { logger.warn("Ignoring fork after {} rejects and retries.", forkCounter); retryPut = false; } else { logger.warn("Version fork after put detected. Rejecting and retrying put."); try { Thread.sleep(forkWaitTime); } catch (InterruptedException e1) { } forkWaitTime = forkWaitTime * 2; } } } }
UserProfileManager { public void modifyUserProfile(String pid, IUserProfileModification modifier) throws GetFailedException, PutFailedException, AbortModifyException { PutQueueEntry entry = new PutQueueEntry(pid); modifyQueue.add(entry); synchronized (queueWaiter) { queueWaiter.notify(); } UserProfile profile; try { profile = entry.getUserProfile(); if (profile == null) { throw new GetFailedException("User Profile not found"); } } catch (GetFailedException e) { if (modifying != null && modifying.getPid().equals(pid)) { modifying.abort(); } throw e; } boolean retryPut = true; int forkCounter = 0; int forkWaitTime = new Random().nextInt(1000) + 500; while (retryPut) { modifier.modifyUserProfile(profile); try { if (protectionKeys == null) { protectionKeys = profile.getProtectionKeys(); } if (modifying != null && modifying.getPid().equals(pid)) { modifying.setUserProfile(profile); modifying.readyToPut(); modifying.waitForPut(); retryPut = false; } else { throw new PutFailedException("Not allowed to put anymore"); } } catch (VersionForkAfterPutException e) { if (forkCounter++ > FORK_LIMIT) { logger.warn("Ignoring fork after {} rejects and retries.", forkCounter); retryPut = false; } else { logger.warn("Version fork after put detected. Rejecting and retrying put."); try { Thread.sleep(forkWaitTime); } catch (InterruptedException e1) { } forkWaitTime = forkWaitTime * 2; } } } } }
UserProfileManager { public void modifyUserProfile(String pid, IUserProfileModification modifier) throws GetFailedException, PutFailedException, AbortModifyException { PutQueueEntry entry = new PutQueueEntry(pid); modifyQueue.add(entry); synchronized (queueWaiter) { queueWaiter.notify(); } UserProfile profile; try { profile = entry.getUserProfile(); if (profile == null) { throw new GetFailedException("User Profile not found"); } } catch (GetFailedException e) { if (modifying != null && modifying.getPid().equals(pid)) { modifying.abort(); } throw e; } boolean retryPut = true; int forkCounter = 0; int forkWaitTime = new Random().nextInt(1000) + 500; while (retryPut) { modifier.modifyUserProfile(profile); try { if (protectionKeys == null) { protectionKeys = profile.getProtectionKeys(); } if (modifying != null && modifying.getPid().equals(pid)) { modifying.setUserProfile(profile); modifying.readyToPut(); modifying.waitForPut(); retryPut = false; } else { throw new PutFailedException("Not allowed to put anymore"); } } catch (VersionForkAfterPutException e) { if (forkCounter++ > FORK_LIMIT) { logger.warn("Ignoring fork after {} rejects and retries.", forkCounter); retryPut = false; } else { logger.warn("Version fork after put detected. Rejecting and retrying put."); try { Thread.sleep(forkWaitTime); } catch (InterruptedException e1) { } forkWaitTime = forkWaitTime * 2; } } } } UserProfileManager(DataManager dataManager, UserCredentials credentials); }
UserProfileManager { public void modifyUserProfile(String pid, IUserProfileModification modifier) throws GetFailedException, PutFailedException, AbortModifyException { PutQueueEntry entry = new PutQueueEntry(pid); modifyQueue.add(entry); synchronized (queueWaiter) { queueWaiter.notify(); } UserProfile profile; try { profile = entry.getUserProfile(); if (profile == null) { throw new GetFailedException("User Profile not found"); } } catch (GetFailedException e) { if (modifying != null && modifying.getPid().equals(pid)) { modifying.abort(); } throw e; } boolean retryPut = true; int forkCounter = 0; int forkWaitTime = new Random().nextInt(1000) + 500; while (retryPut) { modifier.modifyUserProfile(profile); try { if (protectionKeys == null) { protectionKeys = profile.getProtectionKeys(); } if (modifying != null && modifying.getPid().equals(pid)) { modifying.setUserProfile(profile); modifying.readyToPut(); modifying.waitForPut(); retryPut = false; } else { throw new PutFailedException("Not allowed to put anymore"); } } catch (VersionForkAfterPutException e) { if (forkCounter++ > FORK_LIMIT) { logger.warn("Ignoring fork after {} rejects and retries.", forkCounter); retryPut = false; } else { logger.warn("Version fork after put detected. Rejecting and retrying put."); try { Thread.sleep(forkWaitTime); } catch (InterruptedException e1) { } forkWaitTime = forkWaitTime * 2; } } } } UserProfileManager(DataManager dataManager, UserCredentials credentials); void stopQueueWorker(); void startQueueWorker(); UserCredentials getUserCredentials(); UserProfile readUserProfile(); void modifyUserProfile(String pid, IUserProfileModification modifier); }
UserProfileManager { public void modifyUserProfile(String pid, IUserProfileModification modifier) throws GetFailedException, PutFailedException, AbortModifyException { PutQueueEntry entry = new PutQueueEntry(pid); modifyQueue.add(entry); synchronized (queueWaiter) { queueWaiter.notify(); } UserProfile profile; try { profile = entry.getUserProfile(); if (profile == null) { throw new GetFailedException("User Profile not found"); } } catch (GetFailedException e) { if (modifying != null && modifying.getPid().equals(pid)) { modifying.abort(); } throw e; } boolean retryPut = true; int forkCounter = 0; int forkWaitTime = new Random().nextInt(1000) + 500; while (retryPut) { modifier.modifyUserProfile(profile); try { if (protectionKeys == null) { protectionKeys = profile.getProtectionKeys(); } if (modifying != null && modifying.getPid().equals(pid)) { modifying.setUserProfile(profile); modifying.readyToPut(); modifying.waitForPut(); retryPut = false; } else { throw new PutFailedException("Not allowed to put anymore"); } } catch (VersionForkAfterPutException e) { if (forkCounter++ > FORK_LIMIT) { logger.warn("Ignoring fork after {} rejects and retries.", forkCounter); retryPut = false; } else { logger.warn("Version fork after put detected. Rejecting and retrying put."); try { Thread.sleep(forkWaitTime); } catch (InterruptedException e1) { } forkWaitTime = forkWaitTime * 2; } } } } UserProfileManager(DataManager dataManager, UserCredentials credentials); void stopQueueWorker(); void startQueueWorker(); UserCredentials getUserCredentials(); UserProfile readUserProfile(); void modifyUserProfile(String pid, IUserProfileModification modifier);  }