@Test public void testAllocateAndClear() throws Exception { assertWithPool(pool -> { int n = 2; final CyclicBarrier barrier = new CyclicBarrier(n); final CountDownLatch halt = new CountDownLatch(n); final AtomicInteger errors1 = new AtomicInteger(); final AtomicInteger errors2 = new AtomicInteger(); final AtomicInteger writerCount = new AtomicInteger(); new Thread(() -> { try { for (int i = 0; i < 10000; i++) { try (TableWriter ignored = pool.get("z")) { writerCount.incrementAndGet(); } catch (EntryUnavailableException ignored) { } if (i == 1) { barrier.await(); } else { LockSupport.parkNanos(1L); } } } catch (Exception e) { e.printStackTrace(); errors1.incrementAndGet(); } finally { halt.countDown(); } }).start(); new Thread(() -> { try { barrier.await(); for (int i = 0; i < 10000; i++) { pool.releaseInactive(); LockSupport.parkNanos(1L); } } catch (Exception e) { e.printStackTrace(); errors2.incrementAndGet(); } finally { halt.countDown(); } }).start(); halt.await(); Assert.assertTrue(writerCount.get() > 0); Assert.assertEquals(0, errors1.get()); Assert.assertEquals(0, errors2.get()); }); }
@Override public TableWriter get(CharSequence tableName) { checkClosed(); long thread = Thread.currentThread().getId(); Entry e = entries.get(tableName); if (e == null) { e = new Entry(clock.getTicks()); Entry other = entries.putIfAbsent(tableName, e); if (other == null) { return createWriter(tableName, e, thread); } else { e = other; } } long owner = e.owner; if (Unsafe.cas(e, ENTRY_OWNER, UNALLOCATED, thread)) { if (e.writer == null) { return createWriter(tableName, e, thread); } return checkClosedAndGetWriter(tableName, e); } else { if (e.owner == thread) { if (e.lockFd != -1L) { throw EntryLockedException.INSTANCE; } if (e.ex != null) { notifyListener(thread, tableName, PoolListener.EV_EX_RESEND); throw e.ex; } return checkClosedAndGetWriter(tableName, e); } LOG.error().$("busy [table=`").utf8(tableName).$("`, owner=").$(owner).$(']').$(); throw EntryUnavailableException.INSTANCE; } }
WriterPool extends AbstractPool implements ResourcePool<TableWriter> { @Override public TableWriter get(CharSequence tableName) { checkClosed(); long thread = Thread.currentThread().getId(); Entry e = entries.get(tableName); if (e == null) { e = new Entry(clock.getTicks()); Entry other = entries.putIfAbsent(tableName, e); if (other == null) { return createWriter(tableName, e, thread); } else { e = other; } } long owner = e.owner; if (Unsafe.cas(e, ENTRY_OWNER, UNALLOCATED, thread)) { if (e.writer == null) { return createWriter(tableName, e, thread); } return checkClosedAndGetWriter(tableName, e); } else { if (e.owner == thread) { if (e.lockFd != -1L) { throw EntryLockedException.INSTANCE; } if (e.ex != null) { notifyListener(thread, tableName, PoolListener.EV_EX_RESEND); throw e.ex; } return checkClosedAndGetWriter(tableName, e); } LOG.error().$("busy [table=`").utf8(tableName).$("`, owner=").$(owner).$(']').$(); throw EntryUnavailableException.INSTANCE; } } }
WriterPool extends AbstractPool implements ResourcePool<TableWriter> { @Override public TableWriter get(CharSequence tableName) { checkClosed(); long thread = Thread.currentThread().getId(); Entry e = entries.get(tableName); if (e == null) { e = new Entry(clock.getTicks()); Entry other = entries.putIfAbsent(tableName, e); if (other == null) { return createWriter(tableName, e, thread); } else { e = other; } } long owner = e.owner; if (Unsafe.cas(e, ENTRY_OWNER, UNALLOCATED, thread)) { if (e.writer == null) { return createWriter(tableName, e, thread); } return checkClosedAndGetWriter(tableName, e); } else { if (e.owner == thread) { if (e.lockFd != -1L) { throw EntryLockedException.INSTANCE; } if (e.ex != null) { notifyListener(thread, tableName, PoolListener.EV_EX_RESEND); throw e.ex; } return checkClosedAndGetWriter(tableName, e); } LOG.error().$("busy [table=`").utf8(tableName).$("`, owner=").$(owner).$(']').$(); throw EntryUnavailableException.INSTANCE; } } WriterPool(CairoConfiguration configuration, @Nullable MessageBus messageBus); }
WriterPool extends AbstractPool implements ResourcePool<TableWriter> { @Override public TableWriter get(CharSequence tableName) { checkClosed(); long thread = Thread.currentThread().getId(); Entry e = entries.get(tableName); if (e == null) { e = new Entry(clock.getTicks()); Entry other = entries.putIfAbsent(tableName, e); if (other == null) { return createWriter(tableName, e, thread); } else { e = other; } } long owner = e.owner; if (Unsafe.cas(e, ENTRY_OWNER, UNALLOCATED, thread)) { if (e.writer == null) { return createWriter(tableName, e, thread); } return checkClosedAndGetWriter(tableName, e); } else { if (e.owner == thread) { if (e.lockFd != -1L) { throw EntryLockedException.INSTANCE; } if (e.ex != null) { notifyListener(thread, tableName, PoolListener.EV_EX_RESEND); throw e.ex; } return checkClosedAndGetWriter(tableName, e); } LOG.error().$("busy [table=`").utf8(tableName).$("`, owner=").$(owner).$(']').$(); throw EntryUnavailableException.INSTANCE; } } WriterPool(CairoConfiguration configuration, @Nullable MessageBus messageBus); @Override TableWriter get(CharSequence tableName); boolean lock(CharSequence tableName); int getBusyCount(); int size(); void unlock(CharSequence name); void unlock(CharSequence name, @Nullable TableWriter writer); }
WriterPool extends AbstractPool implements ResourcePool<TableWriter> { @Override public TableWriter get(CharSequence tableName) { checkClosed(); long thread = Thread.currentThread().getId(); Entry e = entries.get(tableName); if (e == null) { e = new Entry(clock.getTicks()); Entry other = entries.putIfAbsent(tableName, e); if (other == null) { return createWriter(tableName, e, thread); } else { e = other; } } long owner = e.owner; if (Unsafe.cas(e, ENTRY_OWNER, UNALLOCATED, thread)) { if (e.writer == null) { return createWriter(tableName, e, thread); } return checkClosedAndGetWriter(tableName, e); } else { if (e.owner == thread) { if (e.lockFd != -1L) { throw EntryLockedException.INSTANCE; } if (e.ex != null) { notifyListener(thread, tableName, PoolListener.EV_EX_RESEND); throw e.ex; } return checkClosedAndGetWriter(tableName, e); } LOG.error().$("busy [table=`").utf8(tableName).$("`, owner=").$(owner).$(']').$(); throw EntryUnavailableException.INSTANCE; } } WriterPool(CairoConfiguration configuration, @Nullable MessageBus messageBus); @Override TableWriter get(CharSequence tableName); boolean lock(CharSequence tableName); int getBusyCount(); int size(); void unlock(CharSequence name); void unlock(CharSequence name, @Nullable TableWriter writer);  }