@Test public void testGetAndCloseRace() throws Exception { try (TableModel model = new TableModel(configuration, "xyz", PartitionBy.NONE).col("ts", ColumnType.DATE)) { CairoTestUtils.create(model); } for (int i = 0; i < 100; i++) { assertWithPool(pool -> { AtomicInteger exceptionCount = new AtomicInteger(); CyclicBarrier barrier = new CyclicBarrier(2); CountDownLatch stopLatch = new CountDownLatch(2); try (TableWriter writer = pool.get("xyz")) { Assert.assertNotNull(writer); } new Thread(() -> { try { barrier.await(); pool.close(); } catch (Exception e) { exceptionCount.incrementAndGet(); e.printStackTrace(); } finally { stopLatch.countDown(); } }).start(); new Thread(() -> { try { barrier.await(); try (TableWriter writer = pool.get("xyz")) { Assert.assertNotNull(writer); } catch (PoolClosedException | EntryUnavailableException ignore) { } } catch (Exception e) { exceptionCount.incrementAndGet(); e.printStackTrace(); } finally { stopLatch.countDown(); } }).start(); Assert.assertTrue(stopLatch.await(2, TimeUnit.SECONDS)); Assert.assertEquals(0, exceptionCount.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);  }