@Test public void testSerialOpenClose() throws Exception { assertWithPool(pool -> { TableReader firstReader = null; for (int i = 0; i < 1000; i++) { try (TableReader reader = pool.get("u")) { if (firstReader == null) { firstReader = reader; } Assert.assertNotNull(reader); Assert.assertSame(firstReader, reader); } } }); }
@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);  }