@Test public void testConcurrentRead() throws Exception { final int readerCount = 5; int threadCount = 2; final int iterations = 1000000; Rnd dataRnd = new Rnd(); final String[] names = new String[readerCount]; final String[] expectedRows = new String[readerCount]; final CharSequenceObjHashMap<String> expectedRowMap = new CharSequenceObjHashMap<>(); for (int i = 0; i < readerCount; i++) { names[i] = "x" + i; try (TableModel model = new TableModel(configuration, names[i], PartitionBy.NONE).col("ts", ColumnType.DATE)) { CairoTestUtils.create(model); } try (TableWriter w = new TableWriter(configuration, names[i])) { for (int k = 0; k < 10; k++) { TableWriter.Row r = w.newRow(); r.putDate(0, dataRnd.nextLong()); r.append(); } w.commit(); } sink.clear(); try (TableReader r = new TableReader(configuration, names[i])) { printer.print(r.getCursor(), r.getMetadata(), true); } expectedRows[i] = sink.toString(); expectedRowMap.put(names[i], expectedRows[i]); } assertWithPool((ReaderPool pool) -> { final CyclicBarrier barrier = new CyclicBarrier(threadCount); final CountDownLatch halt = new CountDownLatch(threadCount); final AtomicInteger errors = new AtomicInteger(); for (int k = 0; k < threadCount; k++) { new Thread(new Runnable() { final ObjHashSet<TableReader> readers = new ObjHashSet<>(); final StringSink sink = new StringSink(); final RecordCursorPrinter printer = new RecordCursorPrinter(sink); @Override public void run() { Rnd rnd = new Rnd(); try { barrier.await(); String name; for (int i = 0; i < iterations; i++) { if (readers.size() == 0 || (readers.size() < 40 && rnd.nextPositiveInt() % 4 == 0)) { name = names[rnd.nextPositiveInt() % readerCount]; try { Assert.assertTrue(readers.add(pool.get(name))); } catch (EntryUnavailableException ignore) { } } Thread.yield(); if (readers.size() == 0) { continue; } int index = rnd.nextPositiveInt() % readers.size(); TableReader reader = readers.get(index); Assert.assertTrue(reader.isOpen()); sink.clear(); printer.print(reader.getCursor(), reader.getMetadata(), true); TestUtils.assertEquals(expectedRowMap.get(reader.getTableName()), sink); Thread.yield(); if (readers.size() > 0 && rnd.nextPositiveInt() % 4 == 0) { TableReader r2 = readers.get(rnd.nextPositiveInt() % readers.size()); Assert.assertTrue(r2.isOpen()); r2.close(); Assert.assertTrue(readers.remove(r2)); } Thread.yield(); } } catch (Exception e) { errors.incrementAndGet(); e.printStackTrace(); } finally { for (int i = 0; i < readers.size(); i++) { readers.get(i).close(); } halt.countDown(); } } }).start(); } halt.await(); Assert.assertEquals(0, halt.getCount()); Assert.assertEquals(0, errors.get()); }); }
@Override public TableReader get(CharSequence name) { Entry e = getEntry(name); long lockOwner = e.lockOwner; long thread = Thread.currentThread().getId(); if (lockOwner != UNLOCKED) { LOG.info().$('\'').utf8(name).$("' is locked [owner=").$(lockOwner).$(']').$(); throw EntryLockedException.INSTANCE; } do { for (int i = 0; i < ENTRY_SIZE; i++) { if (Unsafe.cas(e.allocations, i, UNALLOCATED, thread)) { R r = e.readers[i]; if (r == null) { try { LOG.info().$("open '").utf8(name).$("' [at=").$(e.index).$(':').$(i).$(']').$(); r = new R(this, e, i, name); } catch (CairoException ex) { Unsafe.arrayPutOrdered(e.allocations, i, UNALLOCATED); throw ex; } e.readers[i] = r; notifyListener(thread, name, PoolListener.EV_CREATE, e.index, i); } else { r.reload(); notifyListener(thread, name, PoolListener.EV_GET, e.index, i); } if (isClosed()) { e.readers[i] = null; r.goodby(); LOG.info().$('\'').utf8(name).$("' born free").$(); return r; } LOG.debug().$('\'').utf8(name).$("' is assigned [at=").$(e.index).$(':').$(i).$(", thread=").$(thread).$(']').$(); return r; } } LOG.debug().$("Thread ").$(thread).$(" is moving to entry ").$(e.index + 1).$(); if (Unsafe.getUnsafe().compareAndSwapInt(e, NEXT_STATUS, NEXT_OPEN, NEXT_ALLOCATED)) { LOG.debug().$("Thread ").$(thread).$(" allocated entry ").$(e.index + 1).$(); e.next = new Entry(e.index + 1, clock.getTicks()); } e = e.next; } while (e != null && e.index < maxSegments); notifyListener(thread, name, PoolListener.EV_FULL, -1, -1); LOG.info().$("could not get, busy [table=`").utf8(name).$("`, thread=").$(thread).$(", retries=").$(this.maxSegments).$(']').$(); throw EntryUnavailableException.INSTANCE; }
ReaderPool extends AbstractPool implements ResourcePool<TableReader> { @Override public TableReader get(CharSequence name) { Entry e = getEntry(name); long lockOwner = e.lockOwner; long thread = Thread.currentThread().getId(); if (lockOwner != UNLOCKED) { LOG.info().$('\'').utf8(name).$("' is locked [owner=").$(lockOwner).$(']').$(); throw EntryLockedException.INSTANCE; } do { for (int i = 0; i < ENTRY_SIZE; i++) { if (Unsafe.cas(e.allocations, i, UNALLOCATED, thread)) { R r = e.readers[i]; if (r == null) { try { LOG.info().$("open '").utf8(name).$("' [at=").$(e.index).$(':').$(i).$(']').$(); r = new R(this, e, i, name); } catch (CairoException ex) { Unsafe.arrayPutOrdered(e.allocations, i, UNALLOCATED); throw ex; } e.readers[i] = r; notifyListener(thread, name, PoolListener.EV_CREATE, e.index, i); } else { r.reload(); notifyListener(thread, name, PoolListener.EV_GET, e.index, i); } if (isClosed()) { e.readers[i] = null; r.goodby(); LOG.info().$('\'').utf8(name).$("' born free").$(); return r; } LOG.debug().$('\'').utf8(name).$("' is assigned [at=").$(e.index).$(':').$(i).$(", thread=").$(thread).$(']').$(); return r; } } LOG.debug().$("Thread ").$(thread).$(" is moving to entry ").$(e.index + 1).$(); if (Unsafe.getUnsafe().compareAndSwapInt(e, NEXT_STATUS, NEXT_OPEN, NEXT_ALLOCATED)) { LOG.debug().$("Thread ").$(thread).$(" allocated entry ").$(e.index + 1).$(); e.next = new Entry(e.index + 1, clock.getTicks()); } e = e.next; } while (e != null && e.index < maxSegments); notifyListener(thread, name, PoolListener.EV_FULL, -1, -1); LOG.info().$("could not get, busy [table=`").utf8(name).$("`, thread=").$(thread).$(", retries=").$(this.maxSegments).$(']').$(); throw EntryUnavailableException.INSTANCE; } }
ReaderPool extends AbstractPool implements ResourcePool<TableReader> { @Override public TableReader get(CharSequence name) { Entry e = getEntry(name); long lockOwner = e.lockOwner; long thread = Thread.currentThread().getId(); if (lockOwner != UNLOCKED) { LOG.info().$('\'').utf8(name).$("' is locked [owner=").$(lockOwner).$(']').$(); throw EntryLockedException.INSTANCE; } do { for (int i = 0; i < ENTRY_SIZE; i++) { if (Unsafe.cas(e.allocations, i, UNALLOCATED, thread)) { R r = e.readers[i]; if (r == null) { try { LOG.info().$("open '").utf8(name).$("' [at=").$(e.index).$(':').$(i).$(']').$(); r = new R(this, e, i, name); } catch (CairoException ex) { Unsafe.arrayPutOrdered(e.allocations, i, UNALLOCATED); throw ex; } e.readers[i] = r; notifyListener(thread, name, PoolListener.EV_CREATE, e.index, i); } else { r.reload(); notifyListener(thread, name, PoolListener.EV_GET, e.index, i); } if (isClosed()) { e.readers[i] = null; r.goodby(); LOG.info().$('\'').utf8(name).$("' born free").$(); return r; } LOG.debug().$('\'').utf8(name).$("' is assigned [at=").$(e.index).$(':').$(i).$(", thread=").$(thread).$(']').$(); return r; } } LOG.debug().$("Thread ").$(thread).$(" is moving to entry ").$(e.index + 1).$(); if (Unsafe.getUnsafe().compareAndSwapInt(e, NEXT_STATUS, NEXT_OPEN, NEXT_ALLOCATED)) { LOG.debug().$("Thread ").$(thread).$(" allocated entry ").$(e.index + 1).$(); e.next = new Entry(e.index + 1, clock.getTicks()); } e = e.next; } while (e != null && e.index < maxSegments); notifyListener(thread, name, PoolListener.EV_FULL, -1, -1); LOG.info().$("could not get, busy [table=`").utf8(name).$("`, thread=").$(thread).$(", retries=").$(this.maxSegments).$(']').$(); throw EntryUnavailableException.INSTANCE; } ReaderPool(CairoConfiguration configuration); }
ReaderPool extends AbstractPool implements ResourcePool<TableReader> { @Override public TableReader get(CharSequence name) { Entry e = getEntry(name); long lockOwner = e.lockOwner; long thread = Thread.currentThread().getId(); if (lockOwner != UNLOCKED) { LOG.info().$('\'').utf8(name).$("' is locked [owner=").$(lockOwner).$(']').$(); throw EntryLockedException.INSTANCE; } do { for (int i = 0; i < ENTRY_SIZE; i++) { if (Unsafe.cas(e.allocations, i, UNALLOCATED, thread)) { R r = e.readers[i]; if (r == null) { try { LOG.info().$("open '").utf8(name).$("' [at=").$(e.index).$(':').$(i).$(']').$(); r = new R(this, e, i, name); } catch (CairoException ex) { Unsafe.arrayPutOrdered(e.allocations, i, UNALLOCATED); throw ex; } e.readers[i] = r; notifyListener(thread, name, PoolListener.EV_CREATE, e.index, i); } else { r.reload(); notifyListener(thread, name, PoolListener.EV_GET, e.index, i); } if (isClosed()) { e.readers[i] = null; r.goodby(); LOG.info().$('\'').utf8(name).$("' born free").$(); return r; } LOG.debug().$('\'').utf8(name).$("' is assigned [at=").$(e.index).$(':').$(i).$(", thread=").$(thread).$(']').$(); return r; } } LOG.debug().$("Thread ").$(thread).$(" is moving to entry ").$(e.index + 1).$(); if (Unsafe.getUnsafe().compareAndSwapInt(e, NEXT_STATUS, NEXT_OPEN, NEXT_ALLOCATED)) { LOG.debug().$("Thread ").$(thread).$(" allocated entry ").$(e.index + 1).$(); e.next = new Entry(e.index + 1, clock.getTicks()); } e = e.next; } while (e != null && e.index < maxSegments); notifyListener(thread, name, PoolListener.EV_FULL, -1, -1); LOG.info().$("could not get, busy [table=`").utf8(name).$("`, thread=").$(thread).$(", retries=").$(this.maxSegments).$(']').$(); throw EntryUnavailableException.INSTANCE; } ReaderPool(CairoConfiguration configuration); @Override TableReader get(CharSequence name); int getBusyCount(); int getMaxEntries(); boolean lock(CharSequence name); void unlock(CharSequence name); }
ReaderPool extends AbstractPool implements ResourcePool<TableReader> { @Override public TableReader get(CharSequence name) { Entry e = getEntry(name); long lockOwner = e.lockOwner; long thread = Thread.currentThread().getId(); if (lockOwner != UNLOCKED) { LOG.info().$('\'').utf8(name).$("' is locked [owner=").$(lockOwner).$(']').$(); throw EntryLockedException.INSTANCE; } do { for (int i = 0; i < ENTRY_SIZE; i++) { if (Unsafe.cas(e.allocations, i, UNALLOCATED, thread)) { R r = e.readers[i]; if (r == null) { try { LOG.info().$("open '").utf8(name).$("' [at=").$(e.index).$(':').$(i).$(']').$(); r = new R(this, e, i, name); } catch (CairoException ex) { Unsafe.arrayPutOrdered(e.allocations, i, UNALLOCATED); throw ex; } e.readers[i] = r; notifyListener(thread, name, PoolListener.EV_CREATE, e.index, i); } else { r.reload(); notifyListener(thread, name, PoolListener.EV_GET, e.index, i); } if (isClosed()) { e.readers[i] = null; r.goodby(); LOG.info().$('\'').utf8(name).$("' born free").$(); return r; } LOG.debug().$('\'').utf8(name).$("' is assigned [at=").$(e.index).$(':').$(i).$(", thread=").$(thread).$(']').$(); return r; } } LOG.debug().$("Thread ").$(thread).$(" is moving to entry ").$(e.index + 1).$(); if (Unsafe.getUnsafe().compareAndSwapInt(e, NEXT_STATUS, NEXT_OPEN, NEXT_ALLOCATED)) { LOG.debug().$("Thread ").$(thread).$(" allocated entry ").$(e.index + 1).$(); e.next = new Entry(e.index + 1, clock.getTicks()); } e = e.next; } while (e != null && e.index < maxSegments); notifyListener(thread, name, PoolListener.EV_FULL, -1, -1); LOG.info().$("could not get, busy [table=`").utf8(name).$("`, thread=").$(thread).$(", retries=").$(this.maxSegments).$(']').$(); throw EntryUnavailableException.INSTANCE; } ReaderPool(CairoConfiguration configuration); @Override TableReader get(CharSequence name); int getBusyCount(); int getMaxEntries(); boolean lock(CharSequence name); void unlock(CharSequence name);  }