JdbcSemaphore implements AutoCloseable, Semaphore { @SuppressFBWarnings("UW_UNCOND_WAIT") @CheckReturnValue @Override public boolean tryAcquire(final int nrPermits, final long deadlineNanos) throws InterruptedException { if (nrPermits < 1) { throw new IllegalArgumentException("You should try to acquire something! not " + nrPermits); } synchronized (syncObj) { boolean acquired = false; final MutableHolder<Boolean> beat = MutableHolder.of(Boolean.FALSE); do { checkClosed(); validate(); try { acquired = jdbc.transactOnConnection(new HandlerNano<Connection, Boolean, SQLException>() { @Override public Boolean handle(final Connection conn, final long deadlineNanos) throws SQLException { try (PreparedStatement stmt = conn.prepareStatement(acquireSql)) { stmt.setQueryTimeout(Math.min(JdbcTemplate.getTimeoutToDeadlineSeconds(deadlineNanos), jdbcTimeoutSeconds)); stmt.setInt(1, nrPermits); stmt.setNString(2, org.spf4j.base.Runtime.PROCESS_ID); stmt.setNString(3, semName); stmt.setInt(4, nrPermits); int rowsUpdated = stmt.executeUpdate(); Boolean acquired; if (rowsUpdated == 1) { try (PreparedStatement ostmt = conn.prepareStatement(acquireByOwnerSql)) { ostmt.setInt(1, nrPermits); ostmt.setNString(2, org.spf4j.base.Runtime.PROCESS_ID); ostmt.setNString(3, semName); ostmt.setQueryTimeout(Math.min(JdbcTemplate.getTimeoutToDeadlineSeconds(deadlineNanos), jdbcTimeoutSeconds)); int nrUpdated = ostmt.executeUpdate(); if (nrUpdated != 1) { throw new IllegalStateException("Updated " + nrUpdated + " is incorrect for " + ostmt); } } acquired = Boolean.TRUE; } else { if (rowsUpdated > 1) { throw new IllegalStateException("Too many rows updated! when trying to acquire " + nrPermits); } acquired = Boolean.FALSE; } long currNanoTime = TimeSource.nanoTime(); if (deadlineNanos - currNanoTime > heartBeat.getBeatDurationNanos()) { beat.setValue(heartBeat.tryBeat(conn, currNanoTime, deadlineNanos)); } return acquired; } } }, deadlineNanos); } catch (SQLTimeoutException ex) { return false; } catch (SQLException ex) { throw new LockRuntimeException(ex); } if (beat.getValue()) { heartBeat.updateLastRunNanos(TimeSource.nanoTime()); } if (!acquired) { long secondsLeft = JdbcTemplate.getTimeoutToDeadlineSecondsNoEx(deadlineNanos); if (secondsLeft < 0) { return false; } if (secondsLeft < CLEANUP_TIMEOUT_SECONDS) { Future<Integer> fut = DefaultExecutor.INSTANCE.submit( () -> removeDeadHeartBeatAndNotOwnerRows(CLEANUP_TIMEOUT_SECONDS)); try { fut.get(secondsLeft, TimeUnit.SECONDS); } catch (TimeoutException ex) { break; } catch (ExecutionException ex) { throw new LockRuntimeException(ex); } } else { try { removeDeadHeartBeatAndNotOwnerRows(secondsLeft); } catch (SQLTimeoutException ex) { return false; } catch (SQLException ex) { throw new LockRuntimeException(ex); } } try { if (releaseDeadOwnerPermits(nrPermits) <= 0) { long wtimeMilis = Math.min(TimeUnit.NANOSECONDS.toMillis(deadlineNanos - TimeSource.nanoTime()), ThreadLocalRandom.current().nextLong(acquirePollMillis)); if (wtimeMilis > 0) { syncObj.wait(wtimeMilis); } else { break; } } } catch (SQLException ex) { throw new LockRuntimeException(ex); } } } while (!acquired && deadlineNanos > TimeSource.nanoTime()); if (acquired) { ownedReservations += nrPermits; } return acquired; } } JdbcSemaphore(final DataSource dataSource, final String semaphoreName, final int nrPermits);  JdbcSemaphore(final DataSource dataSource, final String semaphoreName,
          final int nrPermits, final boolean strict);  JdbcSemaphore(final DataSource dataSource, final SemaphoreTablesDesc semTableDesc,
          final String semaphoreName, final int nrPermits, final int jdbcTimeoutSeconds,
          final boolean strictReservations); @SuppressFBWarnings({"CBX_CUSTOM_BUILT_XML", "STT_TOSTRING_STORED_IN_FIELD"}) // no sql builder (yet)  JdbcSemaphore(final DataSource dataSource, final SemaphoreTablesDesc semTableDesc,
          final String semaphoreName, final int nrPermits, final int jdbcTimeoutSeconds,
          final boolean strictReservations, final int acquirePollMillis); void registerJmx(); void unregisterJmx(); @SuppressFBWarnings("UW_UNCOND_WAIT") @CheckReturnValue @Override boolean tryAcquire(final int nrPermits, final long deadlineNanos); @Override @SuppressFBWarnings("EXS_EXCEPTION_SOFTENING_NO_CHECKED") void release(final int nrReservations); void releaseAll(); @JmxExport(description = "Get the available semaphore permits") int availablePermits(); @JmxExport(description = "get the number of permits owned by this process") int permitsOwned(); @JmxExport(description = "Get the total permits this semaphore can hand out") int totalPermits(); @JmxExport(description = "get a list of all dead owners which hold permits") @Nonnull List<OwnerPermits> getDeadOwnerPermits(final int wishPermits); @JmxExport(description = "release dead owner permits") @CheckReturnValue int releaseDeadOwnerPermits(@JmxExport(value = "wishPermits",
          description = "how many we whish to release") final int wishPermits); @JmxExport(description = "Change the total available permits to the provided number") void updatePermits(final int nrPermits); @JmxExport(description = "Reduce the total available permits by the provided number") void reducePermits(final int nrPermits); @JmxExport(description = "Increase the total available permits by the provided number") void increasePermits(final int nrPermits); int removeDeadHeartBeatAndNotOwnerRows(final long timeoutSeconds); @Override String toString(); @Override void close(); @JmxExport int getJdbcTimeoutSeconds(); @JmxExport boolean isIsHealthy();  }
@Test @SuppressFBWarnings("AFBR_ABNORMAL_FINALLY_BLOCK_RETURN") public void testMultiProcess() throws SQLException, IOException, InterruptedException, ExecutionException, TimeoutException { Server server = Server.createTcpServer(new String[]{"-tcpPort", "9123", "-tcpAllowOthers"}).start(); File tempDB = File.createTempFile("test", "h2db"); String connStr = "jdbc:h2:tcp: try { JdbcDataSource ds = new JdbcDataSource(); ds.setURL(connStr); ds.setUser("sa"); ds.setPassword("sa"); createSchemaObjects(ds); testReleaseAck(ds, "testSem", 2); JdbcSemaphore semaphore = new JdbcSemaphore(ds, "test_sem2", 3); org.spf4j.base.Runtime.jrun(BadSemaphoreHandler.class, 10000, connStr, "test_sem2"); org.spf4j.base.Runtime.jrun(BadSemaphoreHandler.class, 10000, connStr, "test_sem2"); Assert.assertTrue(semaphore.tryAcquire(1, TimeUnit.SECONDS)); Assert.assertTrue(semaphore.tryAcquire(10, TimeUnit.SECONDS)); JdbcHeartBeat.stopHeartBeats(); server.stop(); } finally { if (!tempDB.delete()) { throw new IOException("Cannot delete " + tempDB); } } }