/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

package org.apache.asterix.transaction.management.service.locking;

import java.io.IOException;
import java.io.OutputStream;
import java.util.ArrayList;
import java.util.concurrent.TimeUnit;

import org.apache.asterix.common.exceptions.ACIDException;
import org.apache.asterix.common.transactions.DatasetId;
import org.apache.asterix.common.transactions.ILockManager;
import org.apache.asterix.common.transactions.ITransactionContext;
import org.apache.asterix.common.transactions.ITransactionManager;
import org.apache.asterix.transaction.management.service.transaction.TransactionManagementConstants.LockManagerConstants.LockMode;
import org.apache.hyracks.api.lifecycle.ILifeCycleComponent;
import org.apache.logging.log4j.Level;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import it.unimi.dsi.fastutil.longs.Long2LongMap;
import it.unimi.dsi.fastutil.longs.Long2LongMaps;
import it.unimi.dsi.fastutil.longs.Long2LongOpenHashMap;
import it.unimi.dsi.fastutil.longs.LongArrayList;
import it.unimi.dsi.fastutil.longs.LongList;

/**
 * A concurrent implementation of the ILockManager interface.
 *
 * @see ResourceGroupTable
 * @see ResourceGroup
 */
@SuppressWarnings("squid:RedundantThrowsDeclarationCheck") // throws ACIDException
public class ConcurrentLockManager implements ILockManager, ILifeCycleComponent {

    static final Logger LOGGER = LogManager.getLogger();
    static final Level LVL = Level.TRACE;
    private static final boolean ENABLED_DEADLOCK_FREE_LOCKING_PROTOCOL = true;

    private static final int NIL = -1;
    static final long NILL = -1L;

    private static final boolean DEBUG_MODE = false;//true
    private static final boolean CHECK_CONSISTENCY = false;

    private final ResourceGroupTable table;
    private final ResourceArenaManager resArenaMgr;
    private final RequestArenaManager reqArenaMgr;
    private final JobArenaManager jobArenaMgr;
    private final Long2LongMap txnId2TxnSlotMap;
    private final LockManagerStats stats = new LockManagerStats(10000);

    enum LockAction {
        ERR(false, false),
        GET(false, false),
        UPD(false, true), // version of GET that updates the max lock mode
        WAIT(true, false),
        CONV(true, true) // convert (upgrade) a lock (e.g. from S to X)
        ;

        boolean wait;
        boolean modify;

        LockAction(boolean wait, boolean modify) {
            this.wait = wait;
            this.modify = modify;
        }
    }

    private static final LockAction[][] ACTION_MATRIX = {
            // new    NL              IS               IX                S                X
            { LockAction.ERR, LockAction.UPD, LockAction.UPD, LockAction.UPD, LockAction.UPD }, // NL
            { LockAction.ERR, LockAction.GET, LockAction.UPD, LockAction.UPD, LockAction.WAIT }, // IS
            { LockAction.ERR, LockAction.GET, LockAction.GET, LockAction.WAIT, LockAction.WAIT }, // IX
            { LockAction.ERR, LockAction.GET, LockAction.WAIT, LockAction.GET, LockAction.WAIT }, // S
            { LockAction.ERR, LockAction.WAIT, LockAction.WAIT, LockAction.WAIT, LockAction.WAIT } // X
    };

    public ConcurrentLockManager(final int lockManagerShrinkTimer, int tableSize) throws ACIDException {
        this(lockManagerShrinkTimer, Runtime.getRuntime().availableProcessors() * 2, tableSize);
    }

    public ConcurrentLockManager(final int lockManagerShrinkTimer, final int noArenas, final int tableSize)
            throws ACIDException {
        table = new ResourceGroupTable(tableSize);
        resArenaMgr = new ResourceArenaManager(noArenas, lockManagerShrinkTimer);
        reqArenaMgr = new RequestArenaManager(noArenas, lockManagerShrinkTimer);
        jobArenaMgr = new JobArenaManager(noArenas, lockManagerShrinkTimer);
        txnId2TxnSlotMap = Long2LongMaps.synchronize(new Long2LongOpenHashMap());
    }

    @Override
    public void lock(DatasetId datasetId, int entityHashValue, byte lockMode, ITransactionContext txnContext)
            throws ACIDException {
        log("lock", datasetId.getId(), entityHashValue, lockMode, txnContext);
        stats.lock();

        final long txnId = txnContext.getTxnId().getId();
        final long jobSlot = findOrAllocJobSlot(txnId);
        final ResourceGroup group = table.get(datasetId.getId(), entityHashValue);
        group.getLatch();
        try {
            validateJob(txnContext);
            final long resSlot = findOrAllocResourceSlot(group, datasetId.getId(), entityHashValue);
            final long reqSlot = allocRequestSlot(resSlot, jobSlot, lockMode);
            boolean locked = false;
            while (!locked) {
                final LockAction act = determineLockAction(resSlot, jobSlot, lockMode);
                switch (act) {
                    case CONV:
                        if (introducesDeadlock(resSlot, jobSlot, NOPTracker.INSTANCE)) {
                            DeadlockTracker tracker = new CollectingTracker();
                            tracker.pushJob(jobSlot);
                            introducesDeadlock(resSlot, jobSlot, tracker);
                            requestAbort(txnContext, tracker.toString());
                            break;
                        } else if (hasOtherHolders(resSlot, jobSlot)) {
                            enqueueWaiter(group, reqSlot, resSlot, jobSlot, act, txnContext);
                            break;
                        }
                        //fall-through
                    case UPD:
                        resArenaMgr.setMaxMode(resSlot, lockMode);
                        //fall-through
                    case GET:
                        addHolder(reqSlot, resSlot, jobSlot);
                        locked = true;
                        break;
                    case WAIT:
                        enqueueWaiter(group, reqSlot, resSlot, jobSlot, act, txnContext);
                        break;
                    case ERR:
                    default:
                        throw new IllegalStateException();
                }
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new ACIDException(e);
        } finally {
            group.releaseLatch();
        }

        if (CHECK_CONSISTENCY) {
            assertLocksCanBefoundInJobQueue();
        }
    }

    private void enqueueWaiter(final ResourceGroup group, final long reqSlot, final long resSlot, final long jobSlot,
            final LockAction act, ITransactionContext txnContext) throws ACIDException, InterruptedException {
        final Queue queue = act.modify ? upgrader : waiter;
        if (introducesDeadlock(resSlot, jobSlot, NOPTracker.INSTANCE)) {
            DeadlockTracker tracker = new CollectingTracker();
            tracker.pushJob(jobSlot);
            introducesDeadlock(resSlot, jobSlot, tracker);
            requestAbort(txnContext, tracker.toString());
        } else {
            queue.add(reqSlot, resSlot, jobSlot);
        }
        try {
            group.await(txnContext);
        } finally {
            queue.remove(reqSlot, resSlot, jobSlot);
        }
    }

    interface DeadlockTracker {
        void pushResource(long resSlot);

        void pushRequest(long reqSlot);

        void pushJob(long jobSlot);

        void pop();
    }

    private static class NOPTracker implements DeadlockTracker {
        static final DeadlockTracker INSTANCE = new NOPTracker();

        @Override
        public void pushResource(long resSlot) {
            // no-op
        }

        @Override
        public void pushRequest(long reqSlot) {
            // no-op
        }

        @Override
        public void pushJob(long jobSlot) {
            // no-op
        }

        @Override
        public void pop() {
            // no-op
        }
    }

    private static class CollectingTracker implements DeadlockTracker {

        static final boolean DEBUG = false;

        LongList slots = new LongArrayList();
        ArrayList<String> types = new ArrayList<>();

        @Override
        public void pushResource(long resSlot) {
            types.add("Resource");
            slots.add(resSlot);
            if (DEBUG) {
                LOGGER.info("push " + types.get(types.size() - 1) + " " + slots.getLong(slots.size() - 1));
            }
        }

        @Override
        public void pushRequest(long reqSlot) {
            types.add("Request");
            slots.add(reqSlot);
            if (DEBUG) {
                LOGGER.info("push " + types.get(types.size() - 1) + " " + slots.getLong(slots.size() - 1));
            }
        }

        @Override
        public void pushJob(long jobSlot) {
            types.add("Job");
            slots.add(jobSlot);
            if (DEBUG) {
                LOGGER.info("push " + types.get(types.size() - 1) + " " + slots.getLong(slots.size() - 1));
            }
        }

        @Override
        public void pop() {
            if (DEBUG) {
                LOGGER.info("pop " + types.get(types.size() - 1) + " " + slots.getLong(slots.size() - 1));
            }
            types.remove(types.size() - 1);
            slots.removeLong(slots.size() - 1);
        }

        @Override
        public String toString() {
            StringBuilder sb = new StringBuilder();
            for (int i = 0; i < slots.size(); ++i) {
                sb.append(types.get(i)).append(" ").append(TypeUtil.Global.toString(slots.getLong(i))).append("\n");
            }
            return sb.toString();
        }
    }

    /**
     * determine if adding a job to the waiters of a resource will introduce a
     * cycle in the wait-graph where the job waits on itself - but not directly on itself (which happens e.g. in the
     * case of upgrading a lock from S to X).
     *
     * @param resSlot
     *            the slot that contains the information about the resource
     * @param jobSlot
     *            the slot that contains the information about the job
     * @return true if a cycle would be introduced, false otherwise
     */
    private boolean introducesDeadlock(final long resSlot, final long jobSlot, final DeadlockTracker tracker) {
        /*
         * Due to the deadlock-free locking protocol, deadlock is not possible.
         * So, this method always returns false in that case
         */
        return !ENABLED_DEADLOCK_FREE_LOCKING_PROTOCOL && introducesDeadlock(resSlot, jobSlot, tracker, 0);
    }

    private boolean introducesDeadlock(final long resSlot, final long jobSlot, final DeadlockTracker tracker,
            final int depth) {
        synchronized (jobArenaMgr) {
            tracker.pushResource(resSlot);
            long reqSlot = resArenaMgr.getLastHolder(resSlot);
            while (reqSlot >= 0) {
                tracker.pushRequest(reqSlot);
                final long holderJobSlot = reqArenaMgr.getJobSlot(reqSlot);
                tracker.pushJob(holderJobSlot);
                if (holderJobSlot == jobSlot && depth != 0) {
                    return true;
                }

                // To determine if we have a deadlock we need to look at the waiters and at the upgraders.
                // The scanWaiters flag indicates if we are currently scanning the waiters (true) or the upgraders
                // (false).
                boolean scanWaiters = true;
                long jobWaiter = jobArenaMgr.getLastWaiter(holderJobSlot);
                if (jobWaiter < 0) {
                    scanWaiters = false;
                    jobWaiter = jobArenaMgr.getLastUpgrader(holderJobSlot);
                }
                while (jobWaiter >= 0) {
                    long waitingOnResSlot = reqArenaMgr.getResourceId(jobWaiter);
                    if (introducesDeadlock(waitingOnResSlot, jobSlot, tracker, depth + 1)) {
                        return true;
                    }
                    jobWaiter = reqArenaMgr.getNextJobRequest(jobWaiter);
                    if (jobWaiter < 0 && scanWaiters) {
                        scanWaiters = false;
                        jobWaiter = jobArenaMgr.getLastUpgrader(holderJobSlot);
                    }
                }

                tracker.pop(); // job
                tracker.pop(); // request
                reqSlot = reqArenaMgr.getNextRequest(reqSlot);
            }
            tracker.pop(); // resource
            return false;
        }
    }

    @Override
    public void instantLock(DatasetId datasetId, int entityHashValue, byte lockMode, ITransactionContext txnContext)
            throws ACIDException {
        log("instantLock", datasetId.getId(), entityHashValue, lockMode, txnContext);
        stats.instantLock();

        final long txnId = txnContext.getTxnId().getId();
        final ResourceGroup group = table.get(datasetId.getId(), entityHashValue);
        if (group.firstResourceIndex == NILL) {
            validateJob(txnContext);
            // if we do not have a resource in the group, we know that the
            // resource that we are looking for is not locked
            return;
        }

        // we only allocate a request slot if we actually have to wait
        long reqSlot = NILL;

        group.getLatch();
        try {
            validateJob(txnContext);
            final long resSlot = findResourceInGroup(group, datasetId.getId(), entityHashValue);
            if (resSlot < 0) {
                // if we don't find the resource, there are no locks on it.
                return;
            }

            final long jobSlot = findOrAllocJobSlot(txnId);

            while (true) {
                final LockAction act = determineLockAction(resSlot, jobSlot, lockMode);
                switch (act) {
                    case UPD:
                    case GET:
                        return;
                    case WAIT:
                    case CONV:
                        if (reqSlot == NILL) {
                            reqSlot = allocRequestSlot(resSlot, jobSlot, lockMode);
                        }
                        enqueueWaiter(group, reqSlot, resSlot, jobSlot, act, txnContext);
                        break;
                    case ERR:
                    default:
                        throw new IllegalStateException();
                }
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new ACIDException(e);
        } finally {
            if (reqSlot != NILL) {
                // deallocate request, if we allocated one earlier
                if (DEBUG_MODE) {
                    LOGGER.trace("del req slot " + TypeUtil.Global.toString(reqSlot));
                }
                reqArenaMgr.deallocate(reqSlot);
            }
            group.releaseLatch();
        }
    }

    @Override
    public boolean tryLock(DatasetId datasetId, int entityHashValue, byte lockMode, ITransactionContext txnContext)
            throws ACIDException {
        log("tryLock", datasetId.getId(), entityHashValue, lockMode, txnContext);
        stats.tryLock();

        final long txnId = txnContext.getTxnId().getId();
        final long jobSlot = findOrAllocJobSlot(txnId);
        final ResourceGroup group = table.get(datasetId.getId(), entityHashValue);
        group.getLatch();

        try {
            validateJob(txnContext);

            final long resSlot = findOrAllocResourceSlot(group, datasetId.getId(), entityHashValue);
            final long reqSlot = allocRequestSlot(resSlot, jobSlot, lockMode);

            final LockAction act = determineLockAction(resSlot, jobSlot, lockMode);
            switch (act) {
                case UPD:
                    resArenaMgr.setMaxMode(resSlot, lockMode);
                    //fall-through
                case GET:
                    addHolder(reqSlot, resSlot, jobSlot);
                    return true;
                case WAIT:
                case CONV:
                    return false;
                default:
                    throw new IllegalStateException();
            }
        } finally {
            group.releaseLatch();
        }
    }

    @Override
    public boolean instantTryLock(DatasetId datasetId, int entityHashValue, byte lockMode,
            ITransactionContext txnContext) throws ACIDException {
        log("instantTryLock", datasetId.getId(), entityHashValue, lockMode, txnContext);
        stats.instantTryLock();

        final long txnId = txnContext.getTxnId().getId();
        final ResourceGroup group = table.get(datasetId.getId(), entityHashValue);
        if (group.firstResourceIndex == NILL) {
            validateJob(txnContext);
            // if we do not have a resource in the group, we know that the
            // resource that we are looking for is not locked
            return true;
        }

        group.getLatch();
        try {
            validateJob(txnContext);

            final long resSlot = findResourceInGroup(group, datasetId.getId(), entityHashValue);
            if (resSlot < 0) {
                // if we don't find the resource, there are no locks on it.
                return true;
            }

            final long jobSlot = findOrAllocJobSlot(txnId);

            LockAction act = determineLockAction(resSlot, jobSlot, lockMode);
            switch (act) {
                case UPD:
                case GET:
                    return true;
                case WAIT:
                case CONV:
                    return false;
                case ERR:
                default:
                    throw new IllegalStateException();
            }
        } finally {
            group.releaseLatch();
        }
    }

    @Override
    public void unlock(DatasetId datasetId, int entityHashValue, byte lockMode, ITransactionContext txnContext)
            throws ACIDException {
        log("unlock", datasetId.getId(), entityHashValue, lockMode, txnContext);
        final long txnId = txnContext.getTxnId().getId();
        final long jobSlot = txnId2TxnSlotMap.get(txnId);

        unlock(datasetId.getId(), entityHashValue, lockMode, jobSlot);
    }

    private void unlock(int dsId, int entityHashValue, byte lockMode, long jobSlot) throws ACIDException {
        log("unlock", dsId, entityHashValue, lockMode, null);
        stats.unlock();

        ResourceGroup group = table.get(dsId, entityHashValue);
        group.getLatch();
        try {

            long resource = findResourceInGroup(group, dsId, entityHashValue);
            if (resource < 0) {
                throw new IllegalStateException("resource (" + dsId + ",  " + entityHashValue + ") not found");
            }

            if (CHECK_CONSISTENCY) {
                assertLocksCanBefoundInJobQueue();
            }

            long holder = removeLastHolder(resource, jobSlot, lockMode);

            // deallocate request
            if (DEBUG_MODE) {
                LOGGER.trace("del req slot " + TypeUtil.Global.toString(holder));
            }
            reqArenaMgr.deallocate(holder);
            // deallocate resource or fix max lock mode
            if (resourceNotUsed(resource)) {
                long prev = group.firstResourceIndex;
                if (prev == resource) {
                    group.firstResourceIndex = resArenaMgr.getNext(resource);
                } else {
                    while (resArenaMgr.getNext(prev) != resource) {
                        prev = resArenaMgr.getNext(prev);
                    }
                    resArenaMgr.setNext(prev, resArenaMgr.getNext(resource));
                }
                if (DEBUG_MODE) {
                    LOGGER.trace("del res slot " + TypeUtil.Global.toString(resource));
                }
                resArenaMgr.deallocate(resource);
            } else {
                final int oldMaxMode = resArenaMgr.getMaxMode(resource);
                final int newMaxMode = determineNewMaxMode(resource, oldMaxMode);
                resArenaMgr.setMaxMode(resource, newMaxMode);
                group.wakeUp();
            }
        } finally {
            group.releaseLatch();
        }
    }

    @Override
    public void releaseLocks(ITransactionContext txnContext) throws ACIDException {
        log("releaseLocks", NIL, NIL, LockMode.ANY, txnContext);
        stats.releaseLocks();

        long txnId = txnContext.getTxnId().getId();
        long jobSlot = txnId2TxnSlotMap.get(txnId);
        if (jobSlot == 0) {
            // we don't know the job, so there are no locks for it - we're done
            return;
        }
        //System.err.println(table.append(new StringBuilder(), true).toString());
        if (LOGGER.isEnabled(LVL)) {
            LOGGER.log(LVL, "jobArenaMgr " + jobArenaMgr.addTo(new RecordManagerStats()).toString());
            LOGGER.log(LVL, "resArenaMgr " + resArenaMgr.addTo(new RecordManagerStats()).toString());
            LOGGER.log(LVL, "reqArenaMgr " + reqArenaMgr.addTo(new RecordManagerStats()).toString());
        }
        long holder;
        synchronized (jobArenaMgr) {
            holder = jobArenaMgr.getLastHolder(jobSlot);
        }
        while (holder != NILL) {
            long resource = reqArenaMgr.getResourceId(holder);
            int dsId = resArenaMgr.getDatasetId(resource);
            int pkHashVal = resArenaMgr.getPkHashVal(resource);
            unlock(dsId, pkHashVal, LockMode.ANY, jobSlot);
            synchronized (jobArenaMgr) {
                holder = jobArenaMgr.getLastHolder(jobSlot);
            }
        }
        if (DEBUG_MODE) {
            LOGGER.trace("del job slot " + TypeUtil.Global.toString(jobSlot));
        }
        jobArenaMgr.deallocate(jobSlot);
        txnId2TxnSlotMap.remove(txnId);
        stats.logCounters(LOGGER, Level.DEBUG, true);
    }

    private long findOrAllocJobSlot(long txnId) {
        long jobSlot = txnId2TxnSlotMap.get(txnId);
        if (jobSlot == 0) {
            jobSlot = jobArenaMgr.allocate();
            if (DEBUG_MODE) {
                LOGGER.trace("new job slot " + TypeUtil.Global.toString(jobSlot) + " (" + txnId + ")");
            }
            jobArenaMgr.setTxnId(jobSlot, txnId);
            long oldSlot = txnId2TxnSlotMap.putIfAbsent(txnId, jobSlot);
            if (oldSlot != 0) {
                // if another thread allocated a slot for this jobThreadId between
                // get(..) and putIfAbsent(..), we'll use that slot and
                // deallocate the one we allocated
                if (DEBUG_MODE) {
                    LOGGER.trace("del job slot " + TypeUtil.Global.toString(jobSlot) + " due to conflict");
                }
                jobArenaMgr.deallocate(jobSlot);
                jobSlot = oldSlot;
            }
        }
        assert jobSlot > 0;
        return jobSlot;
    }

    private long findOrAllocResourceSlot(ResourceGroup group, int dsId, int entityHashValue) {
        long resSlot = findResourceInGroup(group, dsId, entityHashValue);

        if (resSlot == NILL) {
            // we don't know about this resource, let's alloc a slot
            resSlot = resArenaMgr.allocate();
            resArenaMgr.setDatasetId(resSlot, dsId);
            resArenaMgr.setPkHashVal(resSlot, entityHashValue);
            resArenaMgr.setNext(resSlot, group.firstResourceIndex);
            group.firstResourceIndex = resSlot;
            if (DEBUG_MODE) {
                LOGGER.trace("new res slot " + TypeUtil.Global.toString(resSlot) + " (" + dsId + ", " + entityHashValue
                        + ")");
            }
        } else {
            if (DEBUG_MODE) {
                LOGGER.trace("fnd res slot " + TypeUtil.Global.toString(resSlot) + " (" + dsId + ", " + entityHashValue
                        + ")");
            }
        }
        return resSlot;
    }

    private long allocRequestSlot(long resSlot, long jobSlot, byte lockMode) {
        long reqSlot = reqArenaMgr.allocate();
        reqArenaMgr.setResourceId(reqSlot, resSlot);
        reqArenaMgr.setLockMode(reqSlot, lockMode); // lock mode is a byte!!
        reqArenaMgr.setJobSlot(reqSlot, jobSlot);
        if (DEBUG_MODE) {
            LOGGER.trace("new req slot " + TypeUtil.Global.toString(reqSlot) + " (" + TypeUtil.Global.toString(resSlot)
                    + ", " + TypeUtil.Global.toString(jobSlot) + ", " + LockMode.toString(lockMode) + ")");
        }
        return reqSlot;
    }

    private LockAction determineLockAction(long resSlot, long jobSlot, byte lockMode) {
        final int curLockMode = resArenaMgr.getMaxMode(resSlot);
        final LockAction act = ACTION_MATRIX[curLockMode][lockMode];
        if (act == LockAction.WAIT) {
            return updateActionForSameJob(resSlot, jobSlot, lockMode);
        }
        return act;
    }

    /**
     * when we've got a lock conflict for a different job, we always have to
     * wait, if it is for the same job we either have to
     * a) (wait and) convert the lock once conversion becomes viable or
     * b) acquire the lock if we want to lock the same resource with the same
     * lock mode for the same job.
     *
     * @param resource
     *            the resource slot that's being locked
     * @param job
     *            the job slot of the job locking the resource
     * @param lockMode
     *            the lock mode that the resource should be locked with
     * @return
     */
    private LockAction updateActionForSameJob(long resource, long job, byte lockMode) {
        // TODO we can reduce the number of things we have to look at by
        // carefully distinguishing the different lock modes
        long holder = resArenaMgr.getLastHolder(resource);
        LockAction res = LockAction.WAIT;
        while (holder != NILL) {
            if (job == reqArenaMgr.getJobSlot(holder)) {
                if (reqArenaMgr.getLockMode(holder) == lockMode) {
                    return LockAction.GET;
                } else {
                    if (ENABLED_DEADLOCK_FREE_LOCKING_PROTOCOL) {
                        throw new IllegalStateException(
                                "Lock conversion is not supported when deadlock-free locking protocol is enabled!");
                    }
                    res = LockAction.CONV;
                }
            }
            holder = reqArenaMgr.getNextRequest(holder);
        }
        return res;
    }

    private long findResourceInGroup(ResourceGroup group, int dsId, int entityHashValue) {
        stats.logCounters(LOGGER, LVL, false);
        long resSlot = group.firstResourceIndex;
        while (resSlot != NILL) {
            // either we already have a lock on this resource or we have a
            // hash collision
            if (resArenaMgr.getDatasetId(resSlot) == dsId && resArenaMgr.getPkHashVal(resSlot) == entityHashValue) {
                return resSlot;
            } else {
                resSlot = resArenaMgr.getNext(resSlot);
            }
        }
        return NILL;
    }

    private void addHolder(long request, long resource, long job) {
        long lastHolder = resArenaMgr.getLastHolder(resource);
        reqArenaMgr.setNextRequest(request, lastHolder);
        resArenaMgr.setLastHolder(resource, request);

        synchronized (jobArenaMgr) {
            long lastJobHolder = jobArenaMgr.getLastHolder(job);
            insertIntoJobQueue(request, lastJobHolder);
            jobArenaMgr.setLastHolder(job, request);
        }
    }

    private boolean hasOtherHolders(long resSlot, long jobSlot) {
        long holder = resArenaMgr.getLastHolder(resSlot);
        while (holder != NILL) {
            if (reqArenaMgr.getJobSlot(holder) != jobSlot) {
                return true;
            }
            holder = reqArenaMgr.getNextRequest(holder);
        }
        return false;
    }

    private long removeLastHolder(long resource, long jobSlot, byte lockMode) {
        long holder = resArenaMgr.getLastHolder(resource);
        if (holder < 0) {
            throw new IllegalStateException("no holder for resource " + resource);
        }
        // remove from the list of holders for a resource
        if (requestMatches(holder, jobSlot, lockMode)) {
            // if the head of the queue matches, we need to update the resource
            long next = reqArenaMgr.getNextRequest(holder);
            resArenaMgr.setLastHolder(resource, next);
        } else {
            holder = removeRequestFromQueueForJob(holder, jobSlot, lockMode);
        }

        synchronized (jobArenaMgr) {
            // remove from the list of requests for a job
            long newHead = removeRequestFromJob(holder, jobArenaMgr.getLastHolder(jobSlot));
            jobArenaMgr.setLastHolder(jobSlot, newHead);
        }
        return holder;
    }

    private boolean requestMatches(long holder, long jobSlot, byte lockMode) {
        return jobSlot == reqArenaMgr.getJobSlot(holder)
                && (lockMode == LockMode.ANY || lockMode == reqArenaMgr.getLockMode(holder));
    }

    private long removeRequestFromJob(long holder, long unmodified) {
        long prevForJob = reqArenaMgr.getPrevJobRequest(holder);
        long nextForJob = reqArenaMgr.getNextJobRequest(holder);
        if (nextForJob != NILL) {
            reqArenaMgr.setPrevJobRequest(nextForJob, prevForJob);
        }
        if (prevForJob == NILL) {
            return nextForJob;
        } else {
            reqArenaMgr.setNextJobRequest(prevForJob, nextForJob);
            return unmodified;
        }
    }

    interface Queue {
        void add(long request, long resource, long job);

        void remove(long request, long resource, long job);
    }

    private final Queue waiter = new Queue() {
        @Override
        public void add(long request, long resource, long job) {
            long waiter = resArenaMgr.getFirstWaiter(resource);
            reqArenaMgr.setNextRequest(request, NILL);
            if (waiter == NILL) {
                resArenaMgr.setFirstWaiter(resource, request);
            } else {
                appendToRequestQueue(waiter, request);
            }
            synchronized (jobArenaMgr) {
                waiter = jobArenaMgr.getLastWaiter(job);
                insertIntoJobQueue(request, waiter);
                jobArenaMgr.setLastWaiter(job, request);
            }
        }

        @Override
        public void remove(long request, long resource, long job) {
            long waiter = resArenaMgr.getFirstWaiter(resource);
            if (waiter == request) {
                long next = reqArenaMgr.getNextRequest(waiter);
                resArenaMgr.setFirstWaiter(resource, next);
            } else {
                waiter = removeRequestFromQueueForSlot(waiter, request);
            }
            synchronized (jobArenaMgr) {
                // remove from the list of requests for a job
                long newHead = removeRequestFromJob(waiter, jobArenaMgr.getLastWaiter(job));
                jobArenaMgr.setLastWaiter(job, newHead);
            }
        }
    };

    private final Queue upgrader = new Queue() {
        @Override
        public void add(long request, long resource, long job) {
            long upgrader = resArenaMgr.getFirstUpgrader(resource);
            reqArenaMgr.setNextRequest(request, NILL);
            if (upgrader == NILL) {
                resArenaMgr.setFirstUpgrader(resource, request);
            } else {
                appendToRequestQueue(upgrader, request);
            }
            synchronized (jobArenaMgr) {
                upgrader = jobArenaMgr.getLastUpgrader(job);
                insertIntoJobQueue(request, upgrader);
                jobArenaMgr.setLastUpgrader(job, request);
            }
        }

        @Override
        public void remove(long request, long resource, long job) {
            long upgrader = resArenaMgr.getFirstUpgrader(resource);
            if (upgrader == request) {
                long next = reqArenaMgr.getNextRequest(upgrader);
                resArenaMgr.setFirstUpgrader(resource, next);
            } else {
                upgrader = removeRequestFromQueueForSlot(upgrader, request);
            }
            synchronized (jobArenaMgr) {
                // remove from the list of requests for a job
                long newHead = removeRequestFromJob(upgrader, jobArenaMgr.getLastUpgrader(job));
                jobArenaMgr.setLastUpgrader(job, newHead);
            }
        }
    };

    private void insertIntoJobQueue(long newRequest, long oldRequest) {
        reqArenaMgr.setNextJobRequest(newRequest, oldRequest);
        reqArenaMgr.setPrevJobRequest(newRequest, NILL);
        if (oldRequest >= 0) {
            reqArenaMgr.setPrevJobRequest(oldRequest, newRequest);
        }
    }

    private void appendToRequestQueue(long head, long appendee) {
        long next = reqArenaMgr.getNextRequest(head);
        while (next != NILL) {
            head = next;
            next = reqArenaMgr.getNextRequest(head);
        }
        reqArenaMgr.setNextRequest(head, appendee);
    }

    private long removeRequestFromQueueForSlot(long head, long reqSlot) {
        long cur = head;
        long prev = cur;
        while (prev != NILL) {
            cur = reqArenaMgr.getNextRequest(prev);
            if (cur == NILL) {
                throw new IllegalStateException("request " + reqSlot + " not in queue");
            }
            if (cur == reqSlot) {
                break;
            }
            prev = cur;
        }
        long next = reqArenaMgr.getNextRequest(cur);
        reqArenaMgr.setNextRequest(prev, next);
        return cur;
    }

    /**
     * remove the first request for a given job and lock mode from a request queue.
     * If the value of the parameter lockMode is LockMode.ANY the first request
     * for the job is removed - independent of the LockMode.
     *
     * @param head
     *            the head of the request queue
     * @param jobSlot
     *            the job slot
     * @param lockMode
     *            the lock mode
     * @return the slot of the first request that matched the given job
     */
    private long removeRequestFromQueueForJob(long head, long jobSlot, byte lockMode) {
        long holder = head;
        long prev = holder;
        while (prev != NILL) {
            holder = reqArenaMgr.getNextRequest(prev);
            if (holder == NILL) {
                throw new IllegalStateException("no entry for job " + jobSlot + " in queue");
            }
            if (requestMatches(holder, jobSlot, lockMode)) {
                break;
            }
            prev = holder;
        }
        long next = reqArenaMgr.getNextRequest(holder);
        reqArenaMgr.setNextRequest(prev, next);
        return holder;
    }

    private int determineNewMaxMode(long resource, int oldMaxMode) {
        int newMaxMode = LockMode.NL;
        long holder = resArenaMgr.getLastHolder(resource);
        while (holder != NILL) {
            int curLockMode = reqArenaMgr.getLockMode(holder);
            if (curLockMode == oldMaxMode) {
                // we have another lock of the same mode - we're done
                return oldMaxMode;
            }
            switch (ACTION_MATRIX[newMaxMode][curLockMode]) {
                case UPD:
                    newMaxMode = curLockMode;
                    break;
                case GET:
                    break;
                case WAIT:
                case CONV:
                case ERR:
                    throw new IllegalStateException("incompatible locks in holder queue");
            }
            holder = reqArenaMgr.getNextRequest(holder);
        }
        return newMaxMode;
    }

    private boolean resourceNotUsed(long resource) {
        return resArenaMgr.getLastHolder(resource) == NILL && resArenaMgr.getFirstUpgrader(resource) == NILL
                && resArenaMgr.getFirstWaiter(resource) == NILL;
    }

    private void validateJob(ITransactionContext txnContext) throws ACIDException {
        if (txnContext.getTxnState() == ITransactionManager.ABORTED) {
            throw new ACIDException("" + txnContext.getTxnId() + " is in ABORTED state.");
        } else if (txnContext.isTimeout()) {
            requestAbort(txnContext, "timeout");
        }
    }

    private void requestAbort(ITransactionContext txnContext, String msg) throws ACIDException {
        txnContext.setTimeout(true);
        throw new ACIDException(
                "Transaction " + txnContext.getTxnId() + " should abort (requested by the Lock Manager)" + ":\n" + msg);
    }

    /*
     * Debugging support
     */

    private void log(String string, int id, int entityHashValue, byte lockMode, ITransactionContext txnContext) {
        if (!LOGGER.isEnabled(LVL)) {
            return;
        }
        StringBuilder sb = new StringBuilder();
        sb.append("{ op : ").append(string);
        if (id != NIL) {
            sb.append(" , dataset : ").append(id);
        }
        if (entityHashValue != NIL) {
            sb.append(" , entity : ").append(entityHashValue);
        }
        if (lockMode != LockMode.NL) {
            sb.append(" , mode : ").append(LockMode.toString(lockMode));
        }
        if (txnContext != null) {
            sb.append(" , txnId : ").append(txnContext.getTxnId());
        }
        sb.append(" , thread : ").append(Thread.currentThread().getName());
        sb.append(" }");
        LOGGER.log(LVL, sb.toString());
    }

    private void assertLocksCanBefoundInJobQueue() throws ACIDException {
        try {
            for (int i = 0; i < table.size; ++i) {
                final ResourceGroup group = table.get(i);
                if (group.tryLatch(100, TimeUnit.MILLISECONDS)) {
                    try {
                        long resSlot = group.firstResourceIndex;
                        while (resSlot != NILL) {
                            int dsId = resArenaMgr.getDatasetId(resSlot);
                            int entityHashValue = resArenaMgr.getPkHashVal(resSlot);
                            long reqSlot = resArenaMgr.getLastHolder(resSlot);
                            while (reqSlot != NILL) {
                                byte lockMode = (byte) reqArenaMgr.getLockMode(reqSlot);
                                long jobSlot = reqArenaMgr.getJobSlot(reqSlot);
                                long txnId = jobArenaMgr.getTxnId(jobSlot);
                                assertLockCanBeFoundInJobQueue(dsId, entityHashValue, lockMode, txnId);
                                reqSlot = reqArenaMgr.getNextRequest(reqSlot);
                            }
                            resSlot = resArenaMgr.getNext(resSlot);
                        }
                    } finally {
                        group.releaseLatch();
                    }
                } else {
                    LOGGER.warn("Could not check locks for " + group);
                }
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new IllegalStateException("interrupted", e);
        }
    }

    private void assertLockCanBeFoundInJobQueue(int dsId, int entityHashValue, byte lockMode, long txnId) {
        if (findLockInJobQueue(dsId, entityHashValue, txnId, lockMode) == NILL) {
            String msg = "request for " + LockMode.toString(lockMode) + " lock on dataset " + dsId + " entity "
                    + entityHashValue + " not found for txn " + txnId + " in thread "
                    + Thread.currentThread().getName();
            LOGGER.error(msg);
            throw new IllegalStateException(msg);
        }
    }

    /**
     * tries to find a lock request searching though the job queue
     *
     * @param dsId
     *            dataset id
     * @param entityHashValue
     *            primary key hash value
     * @param txnId
     *            job id
     * @param lockMode
     *            lock mode
     * @return the slot of the request, if the lock request is found, NILL otherwise
     */
    private long findLockInJobQueue(final int dsId, final int entityHashValue, final long txnId, byte lockMode) {
        long jobSlot = txnId2TxnSlotMap.get(txnId);
        if (jobSlot == 0) {
            return NILL;
        }

        long holder;
        synchronized (jobArenaMgr) {
            holder = jobArenaMgr.getLastHolder(jobSlot);
        }
        while (holder != NILL) {
            long resource = reqArenaMgr.getResourceId(holder);
            if (dsId == resArenaMgr.getDatasetId(resource) && entityHashValue == resArenaMgr.getPkHashVal(resource)
                    && jobSlot == reqArenaMgr.getJobSlot(holder)
                    && (lockMode == reqArenaMgr.getLockMode(holder) || lockMode == LockMode.ANY)) {
                return holder;
            }
            synchronized (jobArenaMgr) {
                holder = reqArenaMgr.getNextJobRequest(holder);
            }
        }
        return NILL;
    }

    private TablePrinter getResourceTablePrinter() {
        return new ResourceTablePrinter(table, resArenaMgr, reqArenaMgr, jobArenaMgr);
    }

    private TablePrinter getDumpTablePrinter() {
        return new DumpTablePrinter(table, resArenaMgr, reqArenaMgr, jobArenaMgr, txnId2TxnSlotMap);
    }

    public String printByResource() {
        return getResourceTablePrinter().append(new StringBuilder()).append("\n").toString();
    }

    @Override
    public String toString() {
        return printByResource();
    }

    public String dump() {
        return getDumpTablePrinter().append(new StringBuilder()).toString();
    }

    @Override
    public String prettyPrint() throws ACIDException {
        StringBuilder s = new StringBuilder("\n########### LockManager Status #############\n");
        return getDumpTablePrinter().append(s).toString() + "\n";
    }

    @Override
    public void start() {
        //no op
    }

    @Override
    public void dumpState(OutputStream os) throws IOException {
        os.write(dump().getBytes());
    }

    @Override
    public void stop(boolean dumpState, OutputStream os) throws IOException {
        if (dumpState) {
            dumpState(os);
        }
    }
}
