@Test public void testConcurrentReadsAndWrites() throws Exception { int segmentCacheCapacity = 100; long segmentSizeThreshold = 1000L; ByteBuffer byteBuffer = ByteBuffer.allocate(1000); PartitionInfo partitionInfo = new PartitionInfo(byteBuffer, 0, null, PARTITION_ID, true); partitionInfo.setFlags(PartitionInfo.Flags.PARTITION_IS_ASSIGNED | PartitionInfo.Flags.PARTITION_IS_AVAILABLE); TestPartitionClass partition = new TestPartitionClass(UUID.randomUUID(), partitionDir, partitionInfo, segmentSizeThreshold, segmentCacheCapacity); partition.open(); CompletableFuture<Boolean> future = new CompletableFuture<>(); WriteThread writeThread = new WriteThread(partition, future); writeThread.start(); ReadThread[] readThreads = new ReadThread[1]; for (int i = 0; i < readThreads.length; i++) { readThreads[i] = new ReadThread(partition, writeThread, future); } for (ReadThread readThread : readThreads) { readThread.start(); } for (int i = 0; i < 10 && !future.isDone(); i++) { Uninterruptibly.sleep(1000); } future.complete(true); writeThread.join(); for (ReadThread readThread : readThreads) { readThread.join(); } try { future.get(); } catch (ExecutionException ex) { ex.getCause().printStackTrace(); fail(ex.getCause().getMessage()); } assertTrue(partition.evictedEntrylist.size() > 1); }
void open() throws StorageException, IOException { synchronized (segments) { try (DirectoryStream<Path> stream = Files.newDirectoryStream(directory, "*.seg")) { for (Path segPath : stream) { Path segFileName = segPath.getFileName(); if (segFileName == null) { throw new IllegalStateException("found no segment file name: " + segPath.toString()); } String idxName = segFileName.toString().replace(".seg", ".idx"); Path idxPath = directory.resolve(idxName); segments.add(new Segment(key, segPath, idxPath, partitionInfo, segmentSizeThreshold)); } } segments.sort(Segment.FIRST_TRANSACTION_ID_COMPARATOR); if (segments.size() == 0) { Path segPath = directory.resolve(String.format(FILE_NAME_FORMAT, 0L, "seg")); Path idxPath = directory.resolve(String.format(FILE_NAME_FORMAT, 0L, "idx")); Segment.create(key, segPath, idxPath, partitionInfo.partitionId, 0L); segments.add(new Segment(key, segPath, idxPath, partitionInfo, segmentSizeThreshold)); } int size = segments.size(); Segment segment = segments.get(size - 1); segment.setWritable(); currentSegment = segment; for (int i = 0; i < size - 1; i++) { segmentLRUCache.put(segments.get(i), null); } } task.start(); }
Partition { void open() throws StorageException, IOException { synchronized (segments) { try (DirectoryStream<Path> stream = Files.newDirectoryStream(directory, "*.seg")) { for (Path segPath : stream) { Path segFileName = segPath.getFileName(); if (segFileName == null) { throw new IllegalStateException("found no segment file name: " + segPath.toString()); } String idxName = segFileName.toString().replace(".seg", ".idx"); Path idxPath = directory.resolve(idxName); segments.add(new Segment(key, segPath, idxPath, partitionInfo, segmentSizeThreshold)); } } segments.sort(Segment.FIRST_TRANSACTION_ID_COMPARATOR); if (segments.size() == 0) { Path segPath = directory.resolve(String.format(FILE_NAME_FORMAT, 0L, "seg")); Path idxPath = directory.resolve(String.format(FILE_NAME_FORMAT, 0L, "idx")); Segment.create(key, segPath, idxPath, partitionInfo.partitionId, 0L); segments.add(new Segment(key, segPath, idxPath, partitionInfo, segmentSizeThreshold)); } int size = segments.size(); Segment segment = segments.get(size - 1); segment.setWritable(); currentSegment = segment; for (int i = 0; i < size - 1; i++) { segmentLRUCache.put(segments.get(i), null); } } task.start(); } }
Partition { void open() throws StorageException, IOException { synchronized (segments) { try (DirectoryStream<Path> stream = Files.newDirectoryStream(directory, "*.seg")) { for (Path segPath : stream) { Path segFileName = segPath.getFileName(); if (segFileName == null) { throw new IllegalStateException("found no segment file name: " + segPath.toString()); } String idxName = segFileName.toString().replace(".seg", ".idx"); Path idxPath = directory.resolve(idxName); segments.add(new Segment(key, segPath, idxPath, partitionInfo, segmentSizeThreshold)); } } segments.sort(Segment.FIRST_TRANSACTION_ID_COMPARATOR); if (segments.size() == 0) { Path segPath = directory.resolve(String.format(FILE_NAME_FORMAT, 0L, "seg")); Path idxPath = directory.resolve(String.format(FILE_NAME_FORMAT, 0L, "idx")); Segment.create(key, segPath, idxPath, partitionInfo.partitionId, 0L); segments.add(new Segment(key, segPath, idxPath, partitionInfo, segmentSizeThreshold)); } int size = segments.size(); Segment segment = segments.get(size - 1); segment.setWritable(); currentSegment = segment; for (int i = 0; i < size - 1; i++) { segmentLRUCache.put(segments.get(i), null); } } task.start(); } Partition(UUID key, Path directory, PartitionInfo partitionInfo, long segmentSizeThreshold, int segmentCacheCapacity); }
Partition { void open() throws StorageException, IOException { synchronized (segments) { try (DirectoryStream<Path> stream = Files.newDirectoryStream(directory, "*.seg")) { for (Path segPath : stream) { Path segFileName = segPath.getFileName(); if (segFileName == null) { throw new IllegalStateException("found no segment file name: " + segPath.toString()); } String idxName = segFileName.toString().replace(".seg", ".idx"); Path idxPath = directory.resolve(idxName); segments.add(new Segment(key, segPath, idxPath, partitionInfo, segmentSizeThreshold)); } } segments.sort(Segment.FIRST_TRANSACTION_ID_COMPARATOR); if (segments.size() == 0) { Path segPath = directory.resolve(String.format(FILE_NAME_FORMAT, 0L, "seg")); Path idxPath = directory.resolve(String.format(FILE_NAME_FORMAT, 0L, "idx")); Segment.create(key, segPath, idxPath, partitionInfo.partitionId, 0L); segments.add(new Segment(key, segPath, idxPath, partitionInfo, segmentSizeThreshold)); } int size = segments.size(); Segment segment = segments.get(size - 1); segment.setWritable(); currentSegment = segment; for (int i = 0; i < size - 1; i++) { segmentLRUCache.put(segments.get(i), null); } } task.start(); } Partition(UUID key, Path directory, PartitionInfo partitionInfo, long segmentSizeThreshold, int segmentCacheCapacity); SessionInfo getLastSessionInfo(); ArrayList<Record> getRecords(long transactionId, int maxNumRecords); }
Partition { void open() throws StorageException, IOException { synchronized (segments) { try (DirectoryStream<Path> stream = Files.newDirectoryStream(directory, "*.seg")) { for (Path segPath : stream) { Path segFileName = segPath.getFileName(); if (segFileName == null) { throw new IllegalStateException("found no segment file name: " + segPath.toString()); } String idxName = segFileName.toString().replace(".seg", ".idx"); Path idxPath = directory.resolve(idxName); segments.add(new Segment(key, segPath, idxPath, partitionInfo, segmentSizeThreshold)); } } segments.sort(Segment.FIRST_TRANSACTION_ID_COMPARATOR); if (segments.size() == 0) { Path segPath = directory.resolve(String.format(FILE_NAME_FORMAT, 0L, "seg")); Path idxPath = directory.resolve(String.format(FILE_NAME_FORMAT, 0L, "idx")); Segment.create(key, segPath, idxPath, partitionInfo.partitionId, 0L); segments.add(new Segment(key, segPath, idxPath, partitionInfo, segmentSizeThreshold)); } int size = segments.size(); Segment segment = segments.get(size - 1); segment.setWritable(); currentSegment = segment; for (int i = 0; i < size - 1; i++) { segmentLRUCache.put(segments.get(i), null); } } task.start(); } Partition(UUID key, Path directory, PartitionInfo partitionInfo, long segmentSizeThreshold, int segmentCacheCapacity); SessionInfo getLastSessionInfo(); ArrayList<Record> getRecords(long transactionId, int maxNumRecords);  }