Locks made transaction termination aware

This commit makes community and forseti locks react on transaction termination
by throwing TransactionTerminatedException. Previously when a terminated
transaction was waiting to grab the lock this waiting did not end.

Feature will allow us to develop kill-transaction functionality for
transactions that are basically starved on locks. Feature is currently guarded
by a boolean system property 'tx_termination_aware_locks' and off by default.

community/kernel/src/main/java/org/neo4j/kernel/impl/api/KernelTransactionImplementation.java

@@ -52,6 +52,8 @@
 import org.neo4j.kernel.impl.api.state.TxState;
 import org.neo4j.kernel.impl.api.store.PersistenceCache;
 import org.neo4j.kernel.impl.api.store.StoreReadLayer;
+import org.neo4j.kernel.impl.api.tx.TxTermination;
+import org.neo4j.kernel.impl.api.tx.TxTerminationImpl;
 import org.neo4j.kernel.impl.index.IndexEntityType;
 import org.neo4j.kernel.impl.locking.LockGroup;
 import org.neo4j.kernel.impl.locking.Locks;
@@ -120,6 +122,8 @@ TransactionType upgradeToSchemaTransaction() throws InvalidTransactionTypeKernel
         }
     }
 
+    private static final boolean TX_TERMINATION_AWARE_LOCKS = Boolean.getBoolean( "tx_termination_aware_locks" );
+
     // Logic
     private final SchemaWriteGuard schemaWriteGuard;
     private final IndexingService indexService;
@@ -153,7 +157,7 @@ TransactionType upgradeToSchemaTransaction() throws InvalidTransactionTypeKernel
     private Locks.Client locks;
     private boolean closing, closed;
     private boolean failure, success;
-    private volatile boolean terminated;
+    private final TxTerminationImpl termination = new TxTerminationImpl();
     // Some header information
     private long startTimeMillis;
     private long lastTransactionIdWhenStarted;
@@ -212,9 +216,10 @@ public KernelTransactionImplementation( StatementOperationParts operations,
     /** Reset this transaction to a vanilla state, turning it into a logically new transaction. */
     public KernelTransactionImplementation initialize( long lastCommittedTx )
     {
-        this.locks = locksManager.newClient();
+        this.locks = newLocksClient();
         this.context.bind( locks );
-        this.closing = closed = failure = success = terminated = false;
+        this.closing = closed = failure = success = false;
+        this.termination.reset();
         this.transactionType = TransactionType.ANY;
         this.beforeHookInvoked = false;
         this.recordState.initialize( lastCommittedTx );
@@ -240,16 +245,16 @@ public void failure()
     @Override
     public boolean shouldBeTerminated()
     {
-        return terminated;
+        return termination.shouldBeTerminated();
     }
 
     @Override
     public void markForTermination()
     {
-        if ( !terminated )
+        if ( !termination.shouldBeTerminated() )
         {
             failure = true;
-            terminated = true;
+            termination.markForTermination();
             transactionMonitor.transactionTerminated();
         }
     }
@@ -968,4 +973,13 @@ public String toString()
     {
         return "KernelTransaction[" + this.locks.getLockSessionId() + "]";
     }
+
+    private Locks.Client newLocksClient()
+    {
+        if ( TX_TERMINATION_AWARE_LOCKS )
+        {
+            return locksManager.newClient( termination );
+        }
+        return locksManager.newClient( TxTermination.NONE );
+    }
 }

community/kernel/src/main/java/org/neo4j/kernel/impl/api/tx/TxTermination.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright (c) 2002-2016 "Neo Technology,"
+ * Network Engine for Objects in Lund AB [http://neotechnology.com]
+ *
+ * This file is part of Neo4j.
+ *
+ * Neo4j is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+package org.neo4j.kernel.impl.api.tx;
+
+public interface TxTermination
+{
+    TxTermination NONE = new TxTermination()
+    {
+        @Override
+        public boolean shouldBeTerminated()
+        {
+            return false;
+        }
+    };
+
+    boolean shouldBeTerminated();
+}

community/kernel/src/main/java/org/neo4j/kernel/impl/api/tx/TxTerminationImpl.java

@@ -0,0 +1,40 @@
+/*
+ * Copyright (c) 2002-2016 "Neo Technology,"
+ * Network Engine for Objects in Lund AB [http://neotechnology.com]
+ *
+ * This file is part of Neo4j.
+ *
+ * Neo4j is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+package org.neo4j.kernel.impl.api.tx;
+
+public class TxTerminationImpl implements TxTermination
+{
+    private volatile boolean terminated;
+
+    public void markForTermination()
+    {
+        terminated = true;
+    }
+
+    public boolean shouldBeTerminated()
+    {
+        return terminated;
+    }
+
+    public void reset()
+    {
+        terminated = false;
+    }
+}

community/kernel/src/main/java/org/neo4j/kernel/impl/locking/Locks.java

@@ -20,6 +20,7 @@
 package org.neo4j.kernel.impl.locking;
 
 import org.neo4j.helpers.Service;
+import org.neo4j.kernel.impl.api.tx.TxTermination;
 import org.neo4j.kernel.impl.util.concurrent.WaitStrategy;
 import org.neo4j.kernel.lifecycle.Lifecycle;
 
@@ -114,8 +115,9 @@ interface Client extends AutoCloseable
      * you call {@link Locks.Client#close()}.
      *
      * @throws IllegalStateException if this instance has been closed, i.e has had {@link #shutdown()} called.
+     * @param txTermination shows if transaction owning the client should be terminated
      */
-    Client newClient();
+    Client newClient( TxTermination txTermination );
 
     /** Visit all held locks. */
     void accept(Visitor visitor);

community/kernel/src/main/java/org/neo4j/kernel/impl/locking/community/CommunityLockClient.java

@@ -24,21 +24,24 @@
 import org.neo4j.collection.primitive.PrimitiveIntObjectVisitor;
 import org.neo4j.collection.primitive.PrimitiveLongObjectMap;
 import org.neo4j.collection.primitive.PrimitiveLongObjectVisitor;
+import org.neo4j.kernel.impl.api.tx.TxTermination;
 import org.neo4j.kernel.impl.locking.Locks;
 
 import static java.lang.String.format;
 
 public class CommunityLockClient implements Locks.Client
 {
     private final LockManagerImpl manager;
+    private final TxTermination txTermination;
     private final LockTransaction lockTransaction = new LockTransaction();
 
     private final PrimitiveIntObjectMap<PrimitiveLongObjectMap<LockResource>> sharedLocks = Primitive.intObjectMap();
     private final PrimitiveIntObjectMap<PrimitiveLongObjectMap<LockResource>> exclusiveLocks = Primitive.intObjectMap();
 
-    public CommunityLockClient( LockManagerImpl manager )
+    public CommunityLockClient( LockManagerImpl manager, TxTermination txTermination )
     {
         this.manager = manager;
+        this.txTermination = txTermination;
     }
 
     @Override
@@ -55,7 +58,7 @@ public void acquireShared( Locks.ResourceType resourceType, long... resourceIds
             }
 
             resource = new LockResource( resourceType, resourceId );
-            manager.getReadLock( resource, lockTransaction );
+            manager.getReadLock( resource, lockTransaction, txTermination );
             localLocks.put(resourceId, resource);
         }
     }
@@ -74,7 +77,7 @@ public void acquireExclusive( Locks.ResourceType resourceType, long... resourceI
             }
 
             resource = new LockResource( resourceType, resourceId );
-            manager.getWriteLock( resource, lockTransaction );
+            manager.getWriteLock( resource, lockTransaction, txTermination );
             localLocks.put(resourceId, resource);
         }
     }

community/kernel/src/main/java/org/neo4j/kernel/impl/locking/community/CommunityLockManger.java

@@ -19,6 +19,7 @@
  */
 package org.neo4j.kernel.impl.locking.community;
 
+import org.neo4j.kernel.impl.api.tx.TxTermination;
 import org.neo4j.kernel.impl.locking.Locks;
 import org.neo4j.kernel.lifecycle.LifecycleAdapter;
 
@@ -28,7 +29,7 @@ public class CommunityLockManger extends LifecycleAdapter implements Locks
     private volatile boolean closed;
 
     @Override
-    public Client newClient()
+    public Client newClient( TxTermination txTermination )
     {
         // We check this volatile closed flag here, which may seem like a contention overhead, but as the time
         // of writing we apply pooling of transactions and in extension pooling of lock clients,
@@ -37,7 +38,7 @@ public Client newClient()
         {
             throw new IllegalStateException( this + " already closed" );
         }
-        return new CommunityLockClient( manager );
+        return new CommunityLockClient( manager, txTermination );
     }
 
     @Override

community/kernel/src/main/java/org/neo4j/kernel/impl/locking/community/LockManagerImpl.java

@@ -24,6 +24,7 @@
 
 import org.neo4j.helpers.collection.Visitor;
 import org.neo4j.kernel.DeadlockDetectedException;
+import org.neo4j.kernel.impl.api.tx.TxTermination;
 import org.neo4j.kernel.impl.transaction.IllegalResourceException;
 import org.neo4j.kernel.impl.util.StringLogger;
 import org.neo4j.kernel.logging.Logging;
@@ -43,10 +44,10 @@ public long getDetectedDeadlockCount()
         return ragManager.getDeadlockCount();
     }
 
-    public void getReadLock( Object resource, Object tx )
+    public void getReadLock( Object resource, Object tx, TxTermination txTermination )
         throws DeadlockDetectedException, IllegalResourceException
     {
-        getRWLockForAcquiring( resource, tx ).acquireReadLock( tx );
+        getRWLockForAcquiring( resource, tx ).acquireReadLock( tx, txTermination );
     }
 
     public boolean tryReadLock( Object resource, Object tx )
@@ -55,10 +56,10 @@ public boolean tryReadLock( Object resource, Object tx )
         return getRWLockForAcquiring( resource, tx ).tryAcquireReadLock( tx );
     }
 
-    public void getWriteLock( Object resource, Object tx )
+    public void getWriteLock( Object resource, Object tx, TxTermination txTermination )
         throws DeadlockDetectedException, IllegalResourceException
     {
-        getRWLockForAcquiring( resource, tx ).acquireWriteLock( tx );
+        getRWLockForAcquiring( resource, tx ).acquireWriteLock( tx, txTermination );
     }
 
     public boolean tryWriteLock( Object resource, Object tx )