ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From sboi...@apache.org
Subject [48/50] [abbrv] incubator-ignite git commit: # ignite-63
Date Thu, 22 Jan 2015 22:04:43 GMT
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/b89b472d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedLockRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedLockRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedLockRequest.java
new file mode 100644
index 0000000..30c2330
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedLockRequest.java
@@ -0,0 +1,833 @@
+/*
+ * 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.ignite.internal.processors.cache.distributed;
+
+import org.apache.ignite.*;
+import org.apache.ignite.internal.*;
+import org.apache.ignite.lang.*;
+import org.apache.ignite.transactions.*;
+import org.gridgain.grid.kernal.processors.cache.*;
+import org.apache.ignite.internal.processors.cache.transactions.*;
+import org.apache.ignite.internal.util.direct.*;
+import org.apache.ignite.internal.util.tostring.*;
+import org.apache.ignite.internal.util.typedef.internal.*;
+import org.jetbrains.annotations.*;
+
+import java.nio.*;
+import java.util.*;
+
+/**
+ * Lock request message.
+ */
+public class GridDistributedLockRequest<K, V> extends GridDistributedBaseMessage<K, V> {
+    /** */
+    private static final long serialVersionUID = 0L;
+
+    /** Sender node ID. */
+    private UUID nodeId;
+
+    /** Near transaction version. */
+    private GridCacheVersion nearXidVer;
+
+    /** Thread ID. */
+    private long threadId;
+
+    /** Future ID. */
+    private IgniteUuid futId;
+
+    /** Max wait timeout. */
+    private long timeout;
+
+    /** Indicates whether lock is obtained within a scope of transaction. */
+    private boolean isInTx;
+
+    /** Invalidate flag for transactions. */
+    private boolean isInvalidate;
+
+    /** Indicates whether implicit lock so for read or write operation. */
+    private boolean isRead;
+
+    /** Transaction isolation. */
+    private IgniteTxIsolation isolation;
+
+    /** Key bytes for keys to lock. */
+    @GridDirectCollection(byte[].class)
+    private List<byte[]> keyBytes;
+
+    /** Keys. */
+    @GridDirectTransient
+    private List<K> keys;
+
+    /** Write entries. */
+    @GridToStringInclude
+    @GridDirectTransient
+    private List<IgniteTxEntry<K, V>> writeEntries;
+
+    /** Serialized write entries. */
+    private byte[] writeEntriesBytes;
+
+    /** Array indicating whether value should be returned for a key. */
+    @GridToStringInclude
+    private boolean[] retVals;
+
+    /** Key-bytes index. */
+    @GridDirectTransient
+    protected int idx;
+
+    /** Key count. */
+    private int txSize;
+
+    /** Group lock key if this is a group-lock transaction. */
+    @GridDirectTransient
+    private IgniteTxKey grpLockKey;
+
+    /** Group lock key bytes. */
+    private byte[] grpLockKeyBytes;
+
+    /** Partition lock flag. Only if group-lock transaction. */
+    private boolean partLock;
+
+    /** DR versions. */
+    @GridToStringInclude
+    private GridCacheVersion[] drVersByIdx;
+
+    /**
+     * Empty constructor.
+     */
+    public GridDistributedLockRequest() {
+        /* No-op. */
+    }
+
+    /**
+     * @param nodeId Node ID.
+     * @param nearXidVer Near transaction ID.
+     * @param threadId Thread ID.
+     * @param futId Future ID.
+     * @param lockVer Cache version.
+     * @param isInTx {@code True} if implicit transaction lock.
+     * @param isRead Indicates whether implicit lock is for read or write operation.
+     * @param isolation Transaction isolation.
+     * @param isInvalidate Invalidation flag.
+     * @param timeout Lock timeout.
+     * @param keyCnt Number of keys.
+     * @param txSize Expected transaction size.
+     * @param grpLockKey Group lock key if this is a group-lock transaction.
+     * @param partLock {@code True} if this is a group-lock transaction request and whole partition is
+     *      locked.
+     */
+    public GridDistributedLockRequest(
+        int cacheId,
+        UUID nodeId,
+        @Nullable GridCacheVersion nearXidVer,
+        long threadId,
+        IgniteUuid futId,
+        GridCacheVersion lockVer,
+        boolean isInTx,
+        boolean isRead,
+        IgniteTxIsolation isolation,
+        boolean isInvalidate,
+        long timeout,
+        int keyCnt,
+        int txSize,
+        @Nullable IgniteTxKey grpLockKey,
+        boolean partLock
+    ) {
+        super(lockVer, keyCnt);
+
+        assert keyCnt > 0;
+        assert futId != null;
+        assert !isInTx || isolation != null;
+
+        this.cacheId = cacheId;
+        this.nodeId = nodeId;
+        this.nearXidVer = nearXidVer;
+        this.threadId = threadId;
+        this.futId = futId;
+        this.isInTx = isInTx;
+        this.isRead = isRead;
+        this.isolation = isolation;
+        this.isInvalidate = isInvalidate;
+        this.timeout = timeout;
+        this.txSize = txSize;
+        this.grpLockKey = grpLockKey;
+        this.partLock = partLock;
+
+        retVals = new boolean[keyCnt];
+    }
+
+    /**
+     *
+     * @return Node ID.
+     */
+    public UUID nodeId() {
+        return nodeId;
+    }
+
+    /**
+     * @return Near transaction ID.
+     */
+    public GridCacheVersion nearXidVersion() {
+        return nearXidVer;
+    }
+
+    /**
+     *
+     * @return Owner node thread ID.
+     */
+    public long threadId() {
+        return threadId;
+    }
+
+    /**
+     * @return Future ID.
+     */
+    public IgniteUuid futureId() {
+        return futId;
+    }
+
+    /**
+     * @return {@code True} if implicit transaction lock.
+     */
+    public boolean inTx() {
+        return isInTx;
+    }
+
+    /**
+     * @return Invalidate flag.
+     */
+    public boolean isInvalidate() {
+        return isInvalidate;
+    }
+
+    /**
+     * @return {@code True} if lock is implicit and for a read operation.
+     */
+    public boolean txRead() {
+        return isRead;
+    }
+
+    /**
+     * @param idx Key index.
+     * @return Flag indicating whether a value should be returned.
+     */
+    public boolean returnValue(int idx) {
+        return retVals[idx];
+    }
+
+    /**
+     * @return Return flags.
+     */
+    public boolean[] returnFlags() {
+        return retVals;
+    }
+
+    /**
+     * @return Transaction isolation or <tt>null</tt> if not in transaction.
+     */
+    public IgniteTxIsolation isolation() {
+        return isolation;
+    }
+
+    /**
+     *
+     * @return Key to lock.
+     */
+    public List<byte[]> keyBytes() {
+        return keyBytes;
+    }
+
+    /**
+     * @return Write entries list.
+     */
+    public List<IgniteTxEntry<K, V>> writeEntries() {
+        return writeEntries;
+    }
+
+    /**
+     * @return Tx size.
+     */
+    public int txSize() {
+        return txSize;
+    }
+
+    /**
+     * Adds a key.
+     *
+     * @param key Key.
+     * @param retVal Flag indicating whether value should be returned.
+     * @param keyBytes Key bytes.
+     * @param writeEntry Write entry.
+     * @param cands Candidates.
+     * @param drVer DR version.
+     * @param ctx Context.
+     * @throws IgniteCheckedException If failed.
+     */
+    public void addKeyBytes(
+        K key,
+        @Nullable byte[] keyBytes,
+        @Nullable IgniteTxEntry<K, V> writeEntry,
+        boolean retVal,
+        @Nullable Collection<GridCacheMvccCandidate<K>> cands,
+        @Nullable GridCacheVersion drVer,
+        GridCacheContext<K, V> ctx
+    ) throws IgniteCheckedException {
+        if (ctx.deploymentEnabled())
+            prepareObject(key, ctx.shared());
+
+        if (keyBytes != null) {
+            if (this.keyBytes == null)
+                this.keyBytes = new ArrayList<>(keysCount());
+
+            this.keyBytes.add(keyBytes);
+        }
+
+        if (keys == null)
+            keys = new ArrayList<>(keysCount());
+
+        keys.add(key);
+
+        candidatesByIndex(idx, cands);
+        drVersionByIndex(idx, drVer);
+
+        retVals[idx] = retVal;
+
+        if (writeEntry != null) {
+            if (writeEntries == null) {
+                assert idx == 0 : "Cannot start adding write entries in the middle of lock message [idx=" + idx +
+                    ", writeEntry=" + writeEntry + ']';
+
+                writeEntries = new ArrayList<>(keysCount());
+            }
+
+            writeEntries.add(writeEntry);
+        }
+
+        idx++;
+    }
+
+    /**
+     * @return Unmarshalled keys.
+     */
+    public List<K> keys() {
+        return keys;
+    }
+
+    /**
+     * @return {@code True} if lock request for group-lock transaction.
+     */
+    public boolean groupLock() {
+        return grpLockKey != null;
+    }
+
+    /**
+     * @return Group lock key.
+     */
+    @Nullable public IgniteTxKey groupLockKey() {
+        return grpLockKey;
+    }
+
+    /**
+     * @return {@code True} if partition is locked in group-lock transaction.
+     */
+    public boolean partitionLock() {
+        return partLock;
+    }
+
+    /**
+     * @return Max lock wait time.
+     */
+    public long timeout() {
+        return timeout;
+    }
+
+    /**
+     * @param idx Key index.
+     * @param drVer DR version.
+     */
+    @SuppressWarnings({"unchecked"})
+    public void drVersionByIndex(int idx, GridCacheVersion drVer) {
+        assert idx < keysCount();
+
+        // If nothing to add.
+        if (drVer == null)
+            return;
+
+        if (drVersByIdx == null)
+            drVersByIdx = new GridCacheVersion[keysCount()];
+
+        drVersByIdx[idx] = drVer;
+    }
+
+    /**
+     * @param idx Key index.
+     * @return DR versions for given key.
+     */
+    public GridCacheVersion drVersionByIndex(int idx) {
+        return drVersByIdx == null ? null : drVersByIdx[idx];
+    }
+
+    /**
+     * @return All DR versions.
+     */
+    public GridCacheVersion[] drVersions() {
+        return drVersByIdx;
+    }
+
+    /** {@inheritDoc}
+     * @param ctx*/
+    @Override public void prepareMarshal(GridCacheSharedContext<K, V> ctx) throws IgniteCheckedException {
+        super.prepareMarshal(ctx);
+
+        if (grpLockKey != null && grpLockKeyBytes == null) {
+            if (ctx.deploymentEnabled())
+                prepareObject(grpLockKey, ctx);
+
+            grpLockKeyBytes = CU.marshal(ctx, grpLockKey);
+        }
+
+        if (writeEntries != null) {
+            marshalTx(writeEntries, ctx);
+
+            writeEntriesBytes = ctx.marshaller().marshal(writeEntries);
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override public void finishUnmarshal(GridCacheSharedContext<K, V> ctx, ClassLoader ldr) throws IgniteCheckedException {
+        super.finishUnmarshal(ctx, ldr);
+
+        if (keys == null)
+            keys = unmarshalCollection(keyBytes, ctx, ldr);
+
+        if (grpLockKey == null && grpLockKeyBytes != null)
+            grpLockKey = ctx.marshaller().unmarshal(grpLockKeyBytes, ldr);
+
+        if (writeEntriesBytes != null) {
+            writeEntries = ctx.marshaller().unmarshal(writeEntriesBytes, ldr);
+
+            unmarshalTx(writeEntries, false, ctx, ldr);
+        }
+    }
+
+    /** {@inheritDoc} */
+    @SuppressWarnings({"CloneCallsConstructors", "OverriddenMethodCallDuringObjectConstruction",
+        "CloneDoesntCallSuperClone"})
+    @Override public GridTcpCommunicationMessageAdapter clone() {
+        GridDistributedLockRequest _clone = new GridDistributedLockRequest();
+
+        clone0(_clone);
+
+        return _clone;
+    }
+
+    /** {@inheritDoc} */
+    @Override protected void clone0(GridTcpCommunicationMessageAdapter _msg) {
+        super.clone0(_msg);
+
+        GridDistributedLockRequest _clone = (GridDistributedLockRequest)_msg;
+
+        _clone.nodeId = nodeId;
+        _clone.nearXidVer = nearXidVer;
+        _clone.threadId = threadId;
+        _clone.futId = futId;
+        _clone.timeout = timeout;
+        _clone.isInTx = isInTx;
+        _clone.isInvalidate = isInvalidate;
+        _clone.isRead = isRead;
+        _clone.isolation = isolation;
+        _clone.keyBytes = keyBytes;
+        _clone.keys = keys;
+        _clone.writeEntries = writeEntries;
+        _clone.writeEntriesBytes = writeEntriesBytes;
+        _clone.retVals = retVals;
+        _clone.idx = idx;
+        _clone.txSize = txSize;
+        _clone.grpLockKey = grpLockKey;
+        _clone.grpLockKeyBytes = grpLockKeyBytes;
+        _clone.partLock = partLock;
+        _clone.drVersByIdx = drVersByIdx;
+    }
+
+    /** {@inheritDoc} */
+    @SuppressWarnings("all")
+    @Override public boolean writeTo(ByteBuffer buf) {
+        commState.setBuffer(buf);
+
+        if (!super.writeTo(buf))
+            return false;
+
+        if (!commState.typeWritten) {
+            if (!commState.putByte(directType()))
+                return false;
+
+            commState.typeWritten = true;
+        }
+
+        switch (commState.idx) {
+            case 8:
+                if (drVersByIdx != null) {
+                    if (commState.it == null) {
+                        if (!commState.putInt(drVersByIdx.length))
+                            return false;
+
+                        commState.it = arrayIterator(drVersByIdx);
+                    }
+
+                    while (commState.it.hasNext() || commState.cur != NULL) {
+                        if (commState.cur == NULL)
+                            commState.cur = commState.it.next();
+
+                        if (!commState.putCacheVersion((GridCacheVersion)commState.cur))
+                            return false;
+
+                        commState.cur = NULL;
+                    }
+
+                    commState.it = null;
+                } else {
+                    if (!commState.putInt(-1))
+                        return false;
+                }
+
+                commState.idx++;
+
+            case 9:
+                if (!commState.putGridUuid(futId))
+                    return false;
+
+                commState.idx++;
+
+            case 10:
+                if (!commState.putByteArray(grpLockKeyBytes))
+                    return false;
+
+                commState.idx++;
+
+            case 11:
+                if (!commState.putBoolean(isInTx))
+                    return false;
+
+                commState.idx++;
+
+            case 12:
+                if (!commState.putBoolean(isInvalidate))
+                    return false;
+
+                commState.idx++;
+
+            case 13:
+                if (!commState.putBoolean(isRead))
+                    return false;
+
+                commState.idx++;
+
+            case 14:
+                if (!commState.putEnum(isolation))
+                    return false;
+
+                commState.idx++;
+
+            case 15:
+                if (keyBytes != null) {
+                    if (commState.it == null) {
+                        if (!commState.putInt(keyBytes.size()))
+                            return false;
+
+                        commState.it = keyBytes.iterator();
+                    }
+
+                    while (commState.it.hasNext() || commState.cur != NULL) {
+                        if (commState.cur == NULL)
+                            commState.cur = commState.it.next();
+
+                        if (!commState.putByteArray((byte[])commState.cur))
+                            return false;
+
+                        commState.cur = NULL;
+                    }
+
+                    commState.it = null;
+                } else {
+                    if (!commState.putInt(-1))
+                        return false;
+                }
+
+                commState.idx++;
+
+            case 16:
+                if (!commState.putCacheVersion(nearXidVer))
+                    return false;
+
+                commState.idx++;
+
+            case 17:
+                if (!commState.putUuid(nodeId))
+                    return false;
+
+                commState.idx++;
+
+            case 18:
+                if (!commState.putBoolean(partLock))
+                    return false;
+
+                commState.idx++;
+
+            case 19:
+                if (!commState.putBooleanArray(retVals))
+                    return false;
+
+                commState.idx++;
+
+            case 20:
+                if (!commState.putLong(threadId))
+                    return false;
+
+                commState.idx++;
+
+            case 21:
+                if (!commState.putLong(timeout))
+                    return false;
+
+                commState.idx++;
+
+            case 22:
+                if (!commState.putInt(txSize))
+                    return false;
+
+                commState.idx++;
+
+            case 23:
+                if (!commState.putByteArray(writeEntriesBytes))
+                    return false;
+
+                commState.idx++;
+
+        }
+
+        return true;
+    }
+
+    /** {@inheritDoc} */
+    @SuppressWarnings("all")
+    @Override public boolean readFrom(ByteBuffer buf) {
+        commState.setBuffer(buf);
+
+        if (!super.readFrom(buf))
+            return false;
+
+        switch (commState.idx) {
+            case 8:
+                if (commState.readSize == -1) {
+                    if (buf.remaining() < 4)
+                        return false;
+
+                    commState.readSize = commState.getInt();
+                }
+
+                if (commState.readSize >= 0) {
+                    if (drVersByIdx == null)
+                        drVersByIdx = new GridCacheVersion[commState.readSize];
+
+                    for (int i = commState.readItems; i < commState.readSize; i++) {
+                        GridCacheVersion _val = commState.getCacheVersion();
+
+                        if (_val == CACHE_VER_NOT_READ)
+                            return false;
+
+                        drVersByIdx[i] = (GridCacheVersion)_val;
+
+                        commState.readItems++;
+                    }
+                }
+
+                commState.readSize = -1;
+                commState.readItems = 0;
+
+                commState.idx++;
+
+            case 9:
+                IgniteUuid futId0 = commState.getGridUuid();
+
+                if (futId0 == GRID_UUID_NOT_READ)
+                    return false;
+
+                futId = futId0;
+
+                commState.idx++;
+
+            case 10:
+                byte[] grpLockKeyBytes0 = commState.getByteArray();
+
+                if (grpLockKeyBytes0 == BYTE_ARR_NOT_READ)
+                    return false;
+
+                grpLockKeyBytes = grpLockKeyBytes0;
+
+                commState.idx++;
+
+            case 11:
+                if (buf.remaining() < 1)
+                    return false;
+
+                isInTx = commState.getBoolean();
+
+                commState.idx++;
+
+            case 12:
+                if (buf.remaining() < 1)
+                    return false;
+
+                isInvalidate = commState.getBoolean();
+
+                commState.idx++;
+
+            case 13:
+                if (buf.remaining() < 1)
+                    return false;
+
+                isRead = commState.getBoolean();
+
+                commState.idx++;
+
+            case 14:
+                if (buf.remaining() < 1)
+                    return false;
+
+                byte isolation0 = commState.getByte();
+
+                isolation = IgniteTxIsolation.fromOrdinal(isolation0);
+
+                commState.idx++;
+
+            case 15:
+                if (commState.readSize == -1) {
+                    if (buf.remaining() < 4)
+                        return false;
+
+                    commState.readSize = commState.getInt();
+                }
+
+                if (commState.readSize >= 0) {
+                    if (keyBytes == null)
+                        keyBytes = new ArrayList<>(commState.readSize);
+
+                    for (int i = commState.readItems; i < commState.readSize; i++) {
+                        byte[] _val = commState.getByteArray();
+
+                        if (_val == BYTE_ARR_NOT_READ)
+                            return false;
+
+                        keyBytes.add((byte[])_val);
+
+                        commState.readItems++;
+                    }
+                }
+
+                commState.readSize = -1;
+                commState.readItems = 0;
+
+                commState.idx++;
+
+            case 16:
+                GridCacheVersion nearXidVer0 = commState.getCacheVersion();
+
+                if (nearXidVer0 == CACHE_VER_NOT_READ)
+                    return false;
+
+                nearXidVer = nearXidVer0;
+
+                commState.idx++;
+
+            case 17:
+                UUID nodeId0 = commState.getUuid();
+
+                if (nodeId0 == UUID_NOT_READ)
+                    return false;
+
+                nodeId = nodeId0;
+
+                commState.idx++;
+
+            case 18:
+                if (buf.remaining() < 1)
+                    return false;
+
+                partLock = commState.getBoolean();
+
+                commState.idx++;
+
+            case 19:
+                boolean[] retVals0 = commState.getBooleanArray();
+
+                if (retVals0 == BOOLEAN_ARR_NOT_READ)
+                    return false;
+
+                retVals = retVals0;
+
+                commState.idx++;
+
+            case 20:
+                if (buf.remaining() < 8)
+                    return false;
+
+                threadId = commState.getLong();
+
+                commState.idx++;
+
+            case 21:
+                if (buf.remaining() < 8)
+                    return false;
+
+                timeout = commState.getLong();
+
+                commState.idx++;
+
+            case 22:
+                if (buf.remaining() < 4)
+                    return false;
+
+                txSize = commState.getInt();
+
+                commState.idx++;
+
+            case 23:
+                byte[] writeEntriesBytes0 = commState.getByteArray();
+
+                if (writeEntriesBytes0 == BYTE_ARR_NOT_READ)
+                    return false;
+
+                writeEntriesBytes = writeEntriesBytes0;
+
+                commState.idx++;
+
+        }
+
+        return true;
+    }
+
+    /** {@inheritDoc} */
+    @Override public byte directType() {
+        return 22;
+    }
+
+    /** {@inheritDoc} */
+    @Override public String toString() {
+        return S.toString(GridDistributedLockRequest.class, this, "keysCnt", retVals.length,
+            "super", super.toString());
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/b89b472d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedLockResponse.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedLockResponse.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedLockResponse.java
new file mode 100644
index 0000000..99d5997
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedLockResponse.java
@@ -0,0 +1,436 @@
+/*
+ * 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.ignite.internal.processors.cache.distributed;
+
+import org.apache.ignite.*;
+import org.apache.ignite.internal.*;
+import org.apache.ignite.lang.*;
+import org.gridgain.grid.kernal.processors.cache.*;
+import org.apache.ignite.internal.util.direct.*;
+import org.apache.ignite.internal.util.tostring.*;
+import org.apache.ignite.internal.util.typedef.*;
+import org.apache.ignite.internal.util.typedef.internal.*;
+import org.jetbrains.annotations.*;
+
+import java.io.*;
+import java.nio.*;
+import java.util.*;
+
+/**
+ * Lock response message.
+ */
+public class GridDistributedLockResponse<K, V> extends GridDistributedBaseMessage<K, V> {
+    /** */
+    private static final long serialVersionUID = 0L;
+
+    /** Future ID. */
+    private IgniteUuid futId;
+
+    /** Error. */
+    @GridDirectTransient
+    private Throwable err;
+
+    /** Serialized error. */
+    private byte[] errBytes;
+
+    /** Value bytes. */
+    @GridDirectCollection(GridCacheValueBytes.class)
+    private List<GridCacheValueBytes> valBytes;
+
+    /** Values. */
+    @GridToStringInclude
+    @GridDirectTransient
+    private List<V> vals;
+
+    /**
+     * Empty constructor (required by {@link Externalizable}).
+     */
+    public GridDistributedLockResponse() {
+        /* No-op. */
+    }
+
+    /**
+     * @param cacheId Cache ID.
+     * @param lockVer Lock version.
+     * @param futId Future ID.
+     * @param cnt Key count.
+     */
+    public GridDistributedLockResponse(int cacheId,
+        GridCacheVersion lockVer,
+        IgniteUuid futId,
+        int cnt) {
+        super(lockVer, cnt);
+
+        assert futId != null;
+
+        this.cacheId = cacheId;
+        this.futId = futId;
+
+        vals = new ArrayList<>(cnt);
+        valBytes = new ArrayList<>(cnt);
+    }
+
+    /**
+     * @param cacheId Cache ID.
+     * @param lockVer Lock ID.
+     * @param futId Future ID.
+     * @param err Error.
+     */
+    public GridDistributedLockResponse(int cacheId,
+        GridCacheVersion lockVer,
+        IgniteUuid futId,
+        Throwable err) {
+        super(lockVer, 0);
+
+        assert futId != null;
+
+        this.cacheId = cacheId;
+        this.futId = futId;
+        this.err = err;
+    }
+
+    /**
+     * @param cacheId Cache ID.
+     * @param lockVer Lock ID.
+     * @param futId Future ID.
+     * @param cnt Count.
+     * @param err Error.
+     */
+    public GridDistributedLockResponse(int cacheId,
+        GridCacheVersion lockVer,
+        IgniteUuid futId,
+        int cnt,
+        Throwable err) {
+        super(lockVer, cnt);
+
+        assert futId != null;
+
+        this.cacheId = cacheId;
+        this.futId = futId;
+        this.err = err;
+
+        vals = new ArrayList<>(cnt);
+        valBytes = new ArrayList<>(cnt);
+    }
+
+    /**
+     *
+     * @return Future ID.
+     */
+    public IgniteUuid futureId() {
+        return futId;
+    }
+
+    /**
+     * @return Error.
+     */
+    public Throwable error() {
+        return err;
+    }
+
+    /**
+     * @param err Error to set.
+     */
+    public void error(Throwable err) {
+        this.err = err;
+    }
+
+    /**
+     * @param idx Index of locked flag.
+     * @return Value of locked flag at given index.
+     */
+    public boolean isCurrentlyLocked(int idx) {
+        assert idx >= 0;
+
+        Collection<GridCacheMvccCandidate<K>> cands = candidatesByIndex(idx);
+
+        for (GridCacheMvccCandidate<K> cand : cands)
+            if (cand.owner())
+                return true;
+
+        return false;
+    }
+
+    /**
+     * @param idx Candidates index.
+     * @param cands Collection of candidates.
+     * @param committedVers Committed versions relative to lock version.
+     * @param rolledbackVers Rolled back versions relative to lock version.
+     */
+    public void setCandidates(int idx, Collection<GridCacheMvccCandidate<K>> cands,
+        Collection<GridCacheVersion> committedVers, Collection<GridCacheVersion> rolledbackVers) {
+        assert idx >= 0;
+
+        completedVersions(committedVers, rolledbackVers);
+
+        candidatesByIndex(idx, cands);
+    }
+
+    /**
+     * @param idx Value index.
+     *
+     * @return Value bytes (possibly {@code null}).
+     */
+    @Nullable public byte[] valueBytes(int idx) {
+        if (!F.isEmpty(valBytes)) {
+            GridCacheValueBytes res = valBytes.get(idx);
+
+            if (res != null && !res.isPlain())
+                return res.get();
+        }
+
+        return null;
+    }
+
+    /**
+     * @param val Value.
+     * @param valBytes Value bytes (possibly {@code null}).
+     * @param ctx Context.
+     * @throws IgniteCheckedException If failed.
+     */
+    public void addValueBytes(V val, @Nullable byte[] valBytes, GridCacheContext<K, V> ctx) throws IgniteCheckedException {
+        if (ctx.deploymentEnabled())
+            prepareObject(val, ctx.shared());
+
+        GridCacheValueBytes vb = null;
+
+        if (val != null) {
+            vb = val instanceof byte[] ? GridCacheValueBytes.plain(val) : valBytes != null ?
+                GridCacheValueBytes.marshaled(valBytes) : null;
+        }
+        else if (valBytes != null)
+            vb = GridCacheValueBytes.marshaled(valBytes);
+
+        this.valBytes.add(vb);
+
+        vals.add(val);
+    }
+
+    /**
+     * @return Values size.
+     */
+    protected int valuesSize() {
+        return vals.size();
+    }
+
+    /**
+     * @param idx Index.
+     * @return Value for given index.
+     */
+    @Nullable public V value(int idx) {
+        if (!F.isEmpty(vals)) {
+            V res = vals.get(idx);
+
+            if (res != null)
+                return res;
+        }
+
+        // If there was no value in values collection, then it could be in value bytes collection in case of byte[].
+        if (!F.isEmpty(valBytes)) {
+            GridCacheValueBytes res = valBytes.get(idx);
+
+            if (res != null && res.isPlain())
+                return (V)res.get();
+        }
+
+        // Value is not found in both value and value bytes collections.
+        return null;
+    }
+
+    /** {@inheritDoc}
+     * @param ctx*/
+    @Override public void prepareMarshal(GridCacheSharedContext<K, V> ctx) throws IgniteCheckedException {
+        super.prepareMarshal(ctx);
+
+        if (F.isEmpty(valBytes) && !F.isEmpty(vals))
+            valBytes = marshalValuesCollection(vals, ctx);
+
+        if (err != null)
+            errBytes = ctx.marshaller().marshal(err);
+    }
+
+    /** {@inheritDoc} */
+    @Override public void finishUnmarshal(GridCacheSharedContext<K, V> ctx, ClassLoader ldr) throws IgniteCheckedException {
+        super.finishUnmarshal(ctx, ldr);
+
+        if (F.isEmpty(vals) && !F.isEmpty(valBytes))
+            vals = unmarshalValueBytesCollection(valBytes, ctx, ldr);
+
+        if (errBytes != null)
+            err = ctx.marshaller().unmarshal(errBytes, ldr);
+    }
+
+    /** {@inheritDoc} */
+    @SuppressWarnings({"CloneDoesntCallSuperClone", "CloneCallsConstructors",
+        "OverriddenMethodCallDuringObjectConstruction"})
+    @Override public GridTcpCommunicationMessageAdapter clone() {
+        GridDistributedLockResponse _clone = new GridDistributedLockResponse();
+
+        clone0(_clone);
+
+        return _clone;
+    }
+
+    /** {@inheritDoc} */
+    @Override protected void clone0(GridTcpCommunicationMessageAdapter _msg) {
+        super.clone0(_msg);
+
+        GridDistributedLockResponse _clone = (GridDistributedLockResponse)_msg;
+
+        _clone.futId = futId;
+        _clone.err = err;
+        _clone.errBytes = errBytes;
+        _clone.valBytes = valBytes;
+        _clone.vals = vals;
+    }
+
+    /** {@inheritDoc} */
+    @SuppressWarnings("all")
+    @Override public boolean writeTo(ByteBuffer buf) {
+        commState.setBuffer(buf);
+
+        if (!super.writeTo(buf))
+            return false;
+
+        if (!commState.typeWritten) {
+            if (!commState.putByte(directType()))
+                return false;
+
+            commState.typeWritten = true;
+        }
+
+        switch (commState.idx) {
+            case 8:
+                if (!commState.putByteArray(errBytes))
+                    return false;
+
+                commState.idx++;
+
+            case 9:
+                if (!commState.putGridUuid(futId))
+                    return false;
+
+                commState.idx++;
+
+            case 10:
+                if (valBytes != null) {
+                    if (commState.it == null) {
+                        if (!commState.putInt(valBytes.size()))
+                            return false;
+
+                        commState.it = valBytes.iterator();
+                    }
+
+                    while (commState.it.hasNext() || commState.cur != NULL) {
+                        if (commState.cur == NULL)
+                            commState.cur = commState.it.next();
+
+                        if (!commState.putValueBytes((GridCacheValueBytes)commState.cur))
+                            return false;
+
+                        commState.cur = NULL;
+                    }
+
+                    commState.it = null;
+                } else {
+                    if (!commState.putInt(-1))
+                        return false;
+                }
+
+                commState.idx++;
+
+        }
+
+        return true;
+    }
+
+    /** {@inheritDoc} */
+    @SuppressWarnings("all")
+    @Override public boolean readFrom(ByteBuffer buf) {
+        commState.setBuffer(buf);
+
+        if (!super.readFrom(buf))
+            return false;
+
+        switch (commState.idx) {
+            case 8:
+                byte[] errBytes0 = commState.getByteArray();
+
+                if (errBytes0 == BYTE_ARR_NOT_READ)
+                    return false;
+
+                errBytes = errBytes0;
+
+                commState.idx++;
+
+            case 9:
+                IgniteUuid futId0 = commState.getGridUuid();
+
+                if (futId0 == GRID_UUID_NOT_READ)
+                    return false;
+
+                futId = futId0;
+
+                commState.idx++;
+
+            case 10:
+                if (commState.readSize == -1) {
+                    if (buf.remaining() < 4)
+                        return false;
+
+                    commState.readSize = commState.getInt();
+                }
+
+                if (commState.readSize >= 0) {
+                    if (valBytes == null)
+                        valBytes = new ArrayList<>(commState.readSize);
+
+                    for (int i = commState.readItems; i < commState.readSize; i++) {
+                        GridCacheValueBytes _val = commState.getValueBytes();
+
+                        if (_val == VAL_BYTES_NOT_READ)
+                            return false;
+
+                        valBytes.add((GridCacheValueBytes)_val);
+
+                        commState.readItems++;
+                    }
+                }
+
+                commState.readSize = -1;
+                commState.readItems = 0;
+
+                commState.idx++;
+
+        }
+
+        return true;
+    }
+
+    /** {@inheritDoc} */
+    @Override public byte directType() {
+        return 23;
+    }
+
+    /** {@inheritDoc} */
+    @Override public String toString() {
+        return S.toString(GridDistributedLockResponse.class, this,
+            "valBytesLen", valBytes == null ? 0 : valBytes.size(),
+            "super", super.toString());
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/b89b472d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxFinishRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxFinishRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxFinishRequest.java
new file mode 100644
index 0000000..d34b2b6
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxFinishRequest.java
@@ -0,0 +1,695 @@
+/*
+ * 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.ignite.internal.processors.cache.distributed;
+
+import org.apache.ignite.*;
+import org.apache.ignite.internal.*;
+import org.apache.ignite.lang.*;
+import org.gridgain.grid.kernal.processors.cache.*;
+import org.apache.ignite.internal.processors.cache.transactions.*;
+import org.apache.ignite.internal.util.direct.*;
+import org.apache.ignite.internal.util.tostring.*;
+import org.apache.ignite.internal.util.typedef.*;
+import org.apache.ignite.internal.util.typedef.internal.*;
+import org.jetbrains.annotations.*;
+
+import java.io.*;
+import java.nio.*;
+import java.util.*;
+
+/**
+ * Transaction completion message.
+ */
+public class GridDistributedTxFinishRequest<K, V> extends GridDistributedBaseMessage<K, V> {
+    /** */
+    private static final long serialVersionUID = 0L;
+
+    /** Future ID. */
+    private IgniteUuid futId;
+
+    /** Thread ID. */
+    private long threadId;
+
+    /** Commit version. */
+    private GridCacheVersion commitVer;
+
+    /** Invalidate flag. */
+    private boolean invalidate;
+
+    /** Commit flag. */
+    private boolean commit;
+
+    /** Sync commit flag. */
+    private boolean syncCommit;
+
+    /** Sync commit flag. */
+    private boolean syncRollback;
+
+    /** Min version used as base for completed versions. */
+    private GridCacheVersion baseVer;
+
+    /** Transaction write entries. */
+    @GridToStringInclude
+    @GridDirectTransient
+    private Collection<IgniteTxEntry<K, V>> writeEntries;
+
+    /** */
+    @GridDirectCollection(byte[].class)
+    private Collection<byte[]> writeEntriesBytes;
+
+    /** Write entries which have not been transferred to nodes during lock request. */
+    @GridToStringInclude
+    @GridDirectTransient
+    private Collection<IgniteTxEntry<K, V>> recoveryWrites;
+
+    /** */
+    @GridDirectCollection(byte[].class)
+    private Collection<byte[]> recoveryWritesBytes;
+
+    /** Expected txSize. */
+    private int txSize;
+
+    /** Group lock key. */
+    @GridDirectTransient
+    private IgniteTxKey grpLockKey;
+
+    /** Group lock key bytes. */
+    private byte[] grpLockKeyBytes;
+
+    /** System flag. */
+    private boolean sys;
+
+    /**
+     * Empty constructor required by {@link Externalizable}.
+     */
+    public GridDistributedTxFinishRequest() {
+        /* No-op. */
+    }
+
+    /**
+     * @param xidVer Transaction ID.
+     * @param futId future ID.
+     * @param threadId Thread ID.
+     * @param commitVer Commit version.
+     * @param commit Commit flag.
+     * @param invalidate Invalidate flag.
+     * @param sys System flag.
+     * @param baseVer Base version.
+     * @param committedVers Committed versions.
+     * @param rolledbackVers Rolled back versions.
+     * @param txSize Expected transaction size.
+     * @param writeEntries Write entries.
+     * @param recoveryWrites Recover entries. In pessimistic mode entries which were not transferred to remote nodes
+     *      with lock requests. {@code Null} for optimistic mode.
+     * @param grpLockKey Group lock key if this is a group-lock transaction.
+     */
+    public GridDistributedTxFinishRequest(
+        GridCacheVersion xidVer,
+        IgniteUuid futId,
+        @Nullable GridCacheVersion commitVer,
+        long threadId,
+        boolean commit,
+        boolean invalidate,
+        boolean sys,
+        boolean syncCommit,
+        boolean syncRollback,
+        GridCacheVersion baseVer,
+        Collection<GridCacheVersion> committedVers,
+        Collection<GridCacheVersion> rolledbackVers,
+        int txSize,
+        Collection<IgniteTxEntry<K, V>> writeEntries,
+        Collection<IgniteTxEntry<K, V>> recoveryWrites,
+        @Nullable IgniteTxKey grpLockKey
+    ) {
+        super(xidVer, writeEntries == null ? 0 : writeEntries.size());
+        assert xidVer != null;
+
+        this.futId = futId;
+        this.commitVer = commitVer;
+        this.threadId = threadId;
+        this.commit = commit;
+        this.invalidate = invalidate;
+        this.sys = sys;
+        this.syncCommit = syncCommit;
+        this.syncRollback = syncRollback;
+        this.baseVer = baseVer;
+        this.txSize = txSize;
+        this.writeEntries = writeEntries;
+        this.recoveryWrites = recoveryWrites;
+        this.grpLockKey = grpLockKey;
+
+        completedVersions(committedVers, rolledbackVers);
+    }
+
+    /**
+     * Clones write entries so that near entries are not passed to DHT cache.
+     */
+    public void cloneEntries() {
+        if (F.isEmpty(writeEntries))
+            return;
+
+        Collection<IgniteTxEntry<K, V>> cp = new ArrayList<>(writeEntries.size());
+
+        for (IgniteTxEntry<K, V> e : writeEntries) {
+            GridCacheContext<K, V> cacheCtx = e.context();
+
+            // Clone only if it is a near cache.
+            if (cacheCtx.isNear())
+                cp.add(e.cleanCopy(cacheCtx.nearTx().dht().context()));
+            else
+                cp.add(e);
+        }
+
+        writeEntries = cp;
+    }
+
+    /**
+     * @return System flag.
+     */
+    public boolean system() {
+        return sys;
+    }
+
+    /**
+     * @return Future ID.
+     */
+    public IgniteUuid futureId() {
+        return futId;
+    }
+
+    /**
+     * @return Thread ID.
+     */
+    public long threadId() {
+        return threadId;
+    }
+
+    /**
+     * @return Commit version.
+     */
+    public GridCacheVersion commitVersion() {
+        return commitVer;
+    }
+
+    /**
+     * @return Commit flag.
+     */
+    public boolean commit() {
+        return commit;
+    }
+
+    /**
+     *
+     * @return Invalidate flag.
+     */
+    public boolean isInvalidate() {
+        return invalidate;
+    }
+
+    /**
+     * @return Sync commit flag.
+     */
+    public boolean syncCommit() {
+        return syncCommit;
+    }
+
+    /**
+     * @return Sync rollback flag.
+     */
+    public boolean syncRollback() {
+        return syncRollback;
+    }
+
+    /**
+     * @return Base version.
+     */
+    public GridCacheVersion baseVersion() {
+        return baseVer;
+    }
+
+    /**
+     * @return Write entries.
+     */
+    public Collection<IgniteTxEntry<K, V>> writes() {
+        return writeEntries;
+    }
+
+    /**
+     * @return Recover entries.
+     */
+    public Collection<IgniteTxEntry<K, V>> recoveryWrites() {
+        return recoveryWrites;
+    }
+
+    /**
+     * @return Expected tx size.
+     */
+    public int txSize() {
+        return txSize;
+    }
+
+    /**
+     *
+     * @return {@code True} if reply is required.
+     */
+    public boolean replyRequired() {
+        return commit ? syncCommit : syncRollback;
+    }
+
+    /**
+     * @return {@code True} if group lock transaction.
+     */
+    public boolean groupLock() {
+        return grpLockKey != null;
+    }
+
+    /**
+     * @return Group lock key.
+     */
+    @Nullable public IgniteTxKey groupLockKey() {
+        return grpLockKey;
+    }
+
+    /** {@inheritDoc}
+     * @param ctx*/
+    @Override public void prepareMarshal(GridCacheSharedContext<K, V> ctx) throws IgniteCheckedException {
+        super.prepareMarshal(ctx);
+
+        if (writeEntries != null) {
+            marshalTx(writeEntries, ctx);
+
+            writeEntriesBytes = new ArrayList<>(writeEntries.size());
+
+            for (IgniteTxEntry<K, V> e : writeEntries)
+                writeEntriesBytes.add(ctx.marshaller().marshal(e));
+        }
+
+        if (recoveryWrites != null) {
+            marshalTx(recoveryWrites, ctx);
+
+            recoveryWritesBytes = new ArrayList<>(recoveryWrites.size());
+
+            for (IgniteTxEntry<K, V> e : recoveryWrites)
+                recoveryWritesBytes.add(ctx.marshaller().marshal(e));
+        }
+
+        if (grpLockKey != null && grpLockKeyBytes == null) {
+            if (ctx.deploymentEnabled())
+                prepareObject(grpLockKey, ctx);
+
+            grpLockKeyBytes = CU.marshal(ctx, grpLockKey);
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override public void finishUnmarshal(GridCacheSharedContext<K, V> ctx, ClassLoader ldr) throws IgniteCheckedException {
+        super.finishUnmarshal(ctx, ldr);
+
+        if (writeEntriesBytes != null) {
+            writeEntries = new ArrayList<>(writeEntriesBytes.size());
+
+            for (byte[] arr : writeEntriesBytes)
+                writeEntries.add(ctx.marshaller().<IgniteTxEntry<K, V>>unmarshal(arr, ldr));
+
+            unmarshalTx(writeEntries, false, ctx, ldr);
+        }
+
+        if (recoveryWritesBytes != null) {
+            recoveryWrites = new ArrayList<>(recoveryWritesBytes.size());
+
+            for (byte[] arr : recoveryWritesBytes)
+                recoveryWrites.add(ctx.marshaller().<IgniteTxEntry<K, V>>unmarshal(arr, ldr));
+
+            unmarshalTx(recoveryWrites, false, ctx, ldr);
+        }
+
+        if (grpLockKeyBytes != null && grpLockKey == null)
+            grpLockKey = ctx.marshaller().unmarshal(grpLockKeyBytes, ldr);
+    }
+
+    /** {@inheritDoc} */
+    @SuppressWarnings({"CloneDoesntCallSuperClone", "CloneCallsConstructors",
+        "OverriddenMethodCallDuringObjectConstruction"})
+    @Override public GridTcpCommunicationMessageAdapter clone() {
+        GridDistributedTxFinishRequest _clone = new GridDistributedTxFinishRequest();
+
+        clone0(_clone);
+
+        return _clone;
+    }
+
+    /** {@inheritDoc} */
+    @Override protected void clone0(GridTcpCommunicationMessageAdapter _msg) {
+        super.clone0(_msg);
+
+        GridDistributedTxFinishRequest _clone = (GridDistributedTxFinishRequest)_msg;
+
+        _clone.futId = futId;
+        _clone.threadId = threadId;
+        _clone.commitVer = commitVer;
+        _clone.invalidate = invalidate;
+        _clone.commit = commit;
+        _clone.baseVer = baseVer;
+        _clone.writeEntries = writeEntries;
+        _clone.writeEntriesBytes = writeEntriesBytes;
+        _clone.recoveryWrites = recoveryWrites;
+        _clone.recoveryWritesBytes = recoveryWritesBytes;
+        _clone.txSize = txSize;
+        _clone.grpLockKey = grpLockKey;
+        _clone.grpLockKeyBytes = grpLockKeyBytes;
+        _clone.sys = sys;
+    }
+
+    /** {@inheritDoc} */
+    @SuppressWarnings("all")
+    @Override public boolean writeTo(ByteBuffer buf) {
+        commState.setBuffer(buf);
+
+        if (!super.writeTo(buf))
+            return false;
+
+        if (!commState.typeWritten) {
+            if (!commState.putByte(directType()))
+                return false;
+
+            commState.typeWritten = true;
+        }
+
+        switch (commState.idx) {
+            case 8:
+                if (!commState.putCacheVersion(baseVer))
+                    return false;
+
+                commState.idx++;
+
+            case 9:
+                if (!commState.putBoolean(commit))
+                    return false;
+
+                commState.idx++;
+
+            case 10:
+                if (!commState.putCacheVersion(commitVer))
+                    return false;
+
+                commState.idx++;
+
+            case 11:
+                if (!commState.putGridUuid(futId))
+                    return false;
+
+                commState.idx++;
+
+            case 12:
+                if (!commState.putByteArray(grpLockKeyBytes))
+                    return false;
+
+                commState.idx++;
+
+            case 13:
+                if (!commState.putBoolean(invalidate))
+                    return false;
+
+                commState.idx++;
+
+            case 14:
+                if (!commState.putBoolean(syncCommit))
+                    return false;
+
+                commState.idx++;
+
+            case 15:
+                if (!commState.putBoolean(syncRollback))
+                    return false;
+
+                commState.idx++;
+
+            case 16:
+                if (recoveryWritesBytes != null) {
+                    if (commState.it == null) {
+                        if (!commState.putInt(recoveryWritesBytes.size()))
+                            return false;
+
+                        commState.it = recoveryWritesBytes.iterator();
+                    }
+
+                    while (commState.it.hasNext() || commState.cur != NULL) {
+                        if (commState.cur == NULL)
+                            commState.cur = commState.it.next();
+
+                        if (!commState.putByteArray((byte[])commState.cur))
+                            return false;
+
+                        commState.cur = NULL;
+                    }
+
+                    commState.it = null;
+                } else {
+                    if (!commState.putInt(-1))
+                        return false;
+                }
+
+                commState.idx++;
+
+            case 17:
+                if (!commState.putLong(threadId))
+                    return false;
+
+                commState.idx++;
+
+            case 18:
+                if (!commState.putInt(txSize))
+                    return false;
+
+                commState.idx++;
+
+            case 19:
+                if (writeEntriesBytes != null) {
+                    if (commState.it == null) {
+                        if (!commState.putInt(writeEntriesBytes.size()))
+                            return false;
+
+                        commState.it = writeEntriesBytes.iterator();
+                    }
+
+                    while (commState.it.hasNext() || commState.cur != NULL) {
+                        if (commState.cur == NULL)
+                            commState.cur = commState.it.next();
+
+                        if (!commState.putByteArray((byte[])commState.cur))
+                            return false;
+
+                        commState.cur = NULL;
+                    }
+
+                    commState.it = null;
+                } else {
+                    if (!commState.putInt(-1))
+                        return false;
+                }
+
+                commState.idx++;
+
+            case 20:
+                if (!commState.putBoolean(sys))
+                    return false;
+
+                commState.idx++;
+        }
+
+        return true;
+    }
+
+    /** {@inheritDoc} */
+    @SuppressWarnings("all")
+    @Override public boolean readFrom(ByteBuffer buf) {
+        commState.setBuffer(buf);
+
+        if (!super.readFrom(buf))
+            return false;
+
+        switch (commState.idx) {
+            case 8:
+                GridCacheVersion baseVer0 = commState.getCacheVersion();
+
+                if (baseVer0 == CACHE_VER_NOT_READ)
+                    return false;
+
+                baseVer = baseVer0;
+
+                commState.idx++;
+
+            case 9:
+                if (buf.remaining() < 1)
+                    return false;
+
+                commit = commState.getBoolean();
+
+                commState.idx++;
+
+            case 10:
+                GridCacheVersion commitVer0 = commState.getCacheVersion();
+
+                if (commitVer0 == CACHE_VER_NOT_READ)
+                    return false;
+
+                commitVer = commitVer0;
+
+                commState.idx++;
+
+            case 11:
+                IgniteUuid futId0 = commState.getGridUuid();
+
+                if (futId0 == GRID_UUID_NOT_READ)
+                    return false;
+
+                futId = futId0;
+
+                commState.idx++;
+
+            case 12:
+                byte[] grpLockKeyBytes0 = commState.getByteArray();
+
+                if (grpLockKeyBytes0 == BYTE_ARR_NOT_READ)
+                    return false;
+
+                grpLockKeyBytes = grpLockKeyBytes0;
+
+                commState.idx++;
+
+            case 13:
+                if (buf.remaining() < 1)
+                    return false;
+
+                invalidate = commState.getBoolean();
+
+                commState.idx++;
+
+            case 14:
+                if (buf.remaining() < 1)
+                    return false;
+
+                syncCommit = commState.getBoolean();
+
+                commState.idx++;
+
+            case 15:
+                if (buf.remaining() < 1)
+                    return false;
+
+                syncRollback = commState.getBoolean();
+
+                commState.idx++;
+
+            case 16:
+                if (commState.readSize == -1) {
+                    if (buf.remaining() < 4)
+                        return false;
+
+                    commState.readSize = commState.getInt();
+                }
+
+                if (commState.readSize >= 0) {
+                    if (recoveryWritesBytes == null)
+                        recoveryWritesBytes = new ArrayList<>(commState.readSize);
+
+                    for (int i = commState.readItems; i < commState.readSize; i++) {
+                        byte[] _val = commState.getByteArray();
+
+                        if (_val == BYTE_ARR_NOT_READ)
+                            return false;
+
+                        recoveryWritesBytes.add((byte[])_val);
+
+                        commState.readItems++;
+                    }
+                }
+
+                commState.readSize = -1;
+                commState.readItems = 0;
+
+                commState.idx++;
+
+            case 17:
+                if (buf.remaining() < 8)
+                    return false;
+
+                threadId = commState.getLong();
+
+                commState.idx++;
+
+            case 18:
+                if (buf.remaining() < 4)
+                    return false;
+
+                txSize = commState.getInt();
+
+                commState.idx++;
+
+            case 19:
+                if (commState.readSize == -1) {
+                    if (buf.remaining() < 4)
+                        return false;
+
+                    commState.readSize = commState.getInt();
+                }
+
+                if (commState.readSize >= 0) {
+                    if (writeEntriesBytes == null)
+                        writeEntriesBytes = new ArrayList<>(commState.readSize);
+
+                    for (int i = commState.readItems; i < commState.readSize; i++) {
+                        byte[] _val = commState.getByteArray();
+
+                        if (_val == BYTE_ARR_NOT_READ)
+                            return false;
+
+                        writeEntriesBytes.add((byte[])_val);
+
+                        commState.readItems++;
+                    }
+                }
+
+                commState.readSize = -1;
+                commState.readItems = 0;
+
+                commState.idx++;
+
+            case 20:
+                if (buf.remaining() < 1)
+                    return false;
+
+                sys = commState.getBoolean();
+
+                commState.idx++;
+        }
+
+        return true;
+    }
+
+    /** {@inheritDoc} */
+    @Override public byte directType() {
+        return 24;
+    }
+
+    /** {@inheritDoc} */
+    @Override public String toString() {
+        return GridToStringBuilder.toString(GridDistributedTxFinishRequest.class, this,
+            "super", super.toString());
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/b89b472d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxFinishResponse.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxFinishResponse.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxFinishResponse.java
new file mode 100644
index 0000000..b6ce1c1
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxFinishResponse.java
@@ -0,0 +1,172 @@
+/*
+ * 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.ignite.internal.processors.cache.distributed;
+
+import org.apache.ignite.lang.*;
+import org.gridgain.grid.kernal.processors.cache.*;
+import org.apache.ignite.internal.util.direct.*;
+import org.apache.ignite.internal.util.tostring.*;
+
+import java.io.*;
+import java.nio.*;
+
+/**
+ * Transaction finish response.
+ */
+public class GridDistributedTxFinishResponse<K, V> extends GridCacheMessage<K, V> {
+    /** */
+    private static final long serialVersionUID = 0L;
+
+    /** */
+    private GridCacheVersion txId;
+
+    /** Future ID. */
+    private IgniteUuid futId;
+
+    /**
+     * Empty constructor required by {@link Externalizable}.
+     */
+    public GridDistributedTxFinishResponse() {
+        /* No-op. */
+    }
+
+    /**
+     * @param txId Transaction id.
+     * @param futId Future ID.
+     */
+    public GridDistributedTxFinishResponse(GridCacheVersion txId, IgniteUuid futId) {
+        assert txId != null;
+        assert futId != null;
+
+        this.txId = txId;
+        this.futId = futId;
+    }
+
+    /**
+     *
+     * @return Transaction id.
+     */
+    public GridCacheVersion xid() {
+        return txId;
+    }
+
+    /**
+     * @return Future ID.
+     */
+    public IgniteUuid futureId() {
+        return futId;
+    }
+
+    /** {@inheritDoc} */
+    @SuppressWarnings({"CloneDoesntCallSuperClone", "CloneCallsConstructors",
+        "OverriddenMethodCallDuringObjectConstruction"})
+    @Override public GridTcpCommunicationMessageAdapter clone() {
+        GridDistributedTxFinishResponse _clone = new GridDistributedTxFinishResponse();
+
+        clone0(_clone);
+
+        return _clone;
+    }
+
+    /** {@inheritDoc} */
+    @Override protected void clone0(GridTcpCommunicationMessageAdapter _msg) {
+        super.clone0(_msg);
+
+        GridDistributedTxFinishResponse _clone = (GridDistributedTxFinishResponse)_msg;
+
+        _clone.txId = txId;
+        _clone.futId = futId;
+    }
+
+    /** {@inheritDoc} */
+    @SuppressWarnings("all")
+    @Override public boolean writeTo(ByteBuffer buf) {
+        commState.setBuffer(buf);
+
+        if (!super.writeTo(buf))
+            return false;
+
+        if (!commState.typeWritten) {
+            if (!commState.putByte(directType()))
+                return false;
+
+            commState.typeWritten = true;
+        }
+
+        switch (commState.idx) {
+            case 3:
+                if (!commState.putGridUuid(futId))
+                    return false;
+
+                commState.idx++;
+
+            case 4:
+                if (!commState.putCacheVersion(txId))
+                    return false;
+
+                commState.idx++;
+
+        }
+
+        return true;
+    }
+
+    /** {@inheritDoc} */
+    @SuppressWarnings("all")
+    @Override public boolean readFrom(ByteBuffer buf) {
+        commState.setBuffer(buf);
+
+        if (!super.readFrom(buf))
+            return false;
+
+        switch (commState.idx) {
+            case 3:
+                IgniteUuid futId0 = commState.getGridUuid();
+
+                if (futId0 == GRID_UUID_NOT_READ)
+                    return false;
+
+                futId = futId0;
+
+                commState.idx++;
+
+            case 4:
+                GridCacheVersion txId0 = commState.getCacheVersion();
+
+                if (txId0 == CACHE_VER_NOT_READ)
+                    return false;
+
+                txId = txId0;
+
+                commState.idx++;
+
+        }
+
+        return true;
+    }
+
+    /** {@inheritDoc} */
+    @Override public byte directType() {
+        return 25;
+    }
+
+    /** {@inheritDoc} */
+    @Override public String toString() {
+        return GridToStringBuilder.toString(GridDistributedTxFinishResponse.class, this);
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/b89b472d/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxMapping.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxMapping.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxMapping.java
new file mode 100644
index 0000000..0fc45c7
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxMapping.java
@@ -0,0 +1,302 @@
+/*
+ * 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.ignite.internal.processors.cache.distributed;
+
+import org.apache.ignite.cluster.*;
+import org.gridgain.grid.kernal.processors.cache.*;
+import org.apache.ignite.internal.processors.cache.transactions.*;
+import org.apache.ignite.internal.util.typedef.*;
+import org.apache.ignite.internal.util.typedef.internal.*;
+import org.apache.ignite.internal.util.tostring.*;
+import org.jetbrains.annotations.*;
+
+import java.io.*;
+import java.util.*;
+import java.util.concurrent.*;
+
+/**
+ * Transaction node mapping.
+ */
+public class GridDistributedTxMapping<K, V> implements Externalizable {
+    /** */
+    private static final long serialVersionUID = 0L;
+
+    /** Mapped node. */
+    @GridToStringExclude
+    private ClusterNode node;
+
+    /** Entries. */
+    @GridToStringInclude
+    private Collection<IgniteTxEntry<K, V>> entries;
+
+    /** Explicit lock flag. */
+    private boolean explicitLock;
+
+    /** DHT version. */
+    private GridCacheVersion dhtVer;
+
+    /** Copy on remove flag. */
+    private boolean readOnly;
+
+    /** {@code True} if this is last mapping for node. */
+    private boolean last;
+
+    /** IDs of backup nodes receiving last prepare request during this mapping. */
+    private Collection<UUID> lastBackups;
+
+    /** {@code True} if mapping is for near caches, {@code false} otherwise. */
+    private boolean near;
+
+    /**
+     * Empty constructor required for {@link Externalizable}.
+     */
+    public GridDistributedTxMapping() {
+        // No-op.
+    }
+
+    /**
+     * @param node Mapped node.
+     */
+    public GridDistributedTxMapping(ClusterNode node) {
+        this.node = node;
+
+        entries = new ConcurrentLinkedQueue<>();
+    }
+
+    /**
+     * @return IDs of backup nodes receiving last prepare request during this mapping.
+     */
+    @Nullable public Collection<UUID> lastBackups() {
+        return lastBackups;
+    }
+
+    /**
+     * @param lastBackups IDs of backup nodes receiving last prepare request during this mapping.
+     */
+    public void lastBackups(@Nullable Collection<UUID> lastBackups) {
+        this.lastBackups = lastBackups;
+    }
+
+    /**
+     * @return {@code True} if this is last mapping for node.
+     */
+    public boolean last() {
+        return last;
+    }
+
+    /**
+     * @param last If {@code True} this is last mapping for node.
+     */
+    public void last(boolean last) {
+        this.last = last;
+    }
+
+    /**
+     * @return {@code True} if mapping is for near caches, {@code false} otherwise.
+     */
+    public boolean near() {
+        return near;
+    }
+
+    /**
+     * @param near {@code True} if mapping is for near caches, {@code false} otherwise.
+     */
+    public void near(boolean near) {
+        this.near = near;
+    }
+
+    /**
+     * @return Node.
+     */
+    public ClusterNode node() {
+        return node;
+    }
+
+    /**
+     * @return Entries.
+     */
+    public Collection<IgniteTxEntry<K, V>> entries() {
+        return entries;
+    }
+
+    /**
+     * @param entries Mapped entries.
+     * @param readOnly Flag indicating that passed in collection is read-only.
+     */
+    public void entries(Collection<IgniteTxEntry<K, V>> entries, boolean readOnly) {
+        this.entries = entries;
+
+        // Set copy on remove flag as passed in collection is unmodifiable.
+        this.readOnly = true;
+    }
+
+    /**
+     * @return {@code True} if lock is explicit.
+     */
+    public boolean explicitLock() {
+        return explicitLock;
+    }
+
+    /**
+     * Sets explicit flag to {@code true}.
+     */
+    public void markExplicitLock() {
+        explicitLock = true;
+    }
+
+    /**
+     * @return DHT version.
+     */
+    public GridCacheVersion dhtVersion() {
+        return dhtVer;
+    }
+
+    /**
+     * @param dhtVer DHT version.
+     */
+    public void dhtVersion(GridCacheVersion dhtVer) {
+        this.dhtVer = dhtVer;
+
+        for (IgniteTxEntry<K, V> e : entries)
+            e.dhtVersion(dhtVer);
+    }
+
+    /**
+     * @return Reads.
+     */
+    public Collection<IgniteTxEntry<K, V>> reads() {
+        return F.view(entries, CU.<K, V>reads());
+    }
+
+    /**
+     * @return Writes.
+     */
+    public Collection<IgniteTxEntry<K, V>> writes() {
+        return F.view(entries, CU.<K, V>writes());
+    }
+
+    /**
+     * @param entry Adds entry.
+     */
+    public void add(IgniteTxEntry<K, V> entry) {
+        ensureModifiable();
+
+        entries.add(entry);
+    }
+
+    /**
+     * @param entry Entry to remove.
+     * @return {@code True} if entry was removed.
+     */
+    public boolean removeEntry(IgniteTxEntry<K, V> entry) {
+        ensureModifiable();
+
+        return entries.remove(entry);
+    }
+
+    /**
+     * @param parts Evicts partitions from mapping.
+     */
+    public void evictPartitions(@Nullable int[] parts) {
+        if (!F.isEmpty(parts)) {
+            ensureModifiable();
+
+            evictPartitions(parts, entries);
+        }
+    }
+
+    /**
+     * @param parts Partitions.
+     * @param c Collection.
+     */
+    private void evictPartitions(int[] parts, Collection<IgniteTxEntry<K, V>> c) {
+        assert parts != null;
+
+        for (Iterator<IgniteTxEntry<K, V>> it = c.iterator(); it.hasNext();) {
+            IgniteTxEntry<K, V> e = it.next();
+
+            GridCacheEntryEx<K,V> cached = e.cached();
+
+            if (U.containsIntArray(parts, cached.partition()))
+                it.remove();
+        }
+    }
+
+    /**
+     * @param keys Keys to evict readers for.
+     */
+    public void evictReaders(@Nullable Collection<IgniteTxKey<K>> keys) {
+        if (keys == null || keys.isEmpty())
+            return;
+
+        ensureModifiable();
+
+        evictReaders(keys, entries);
+    }
+
+    /**
+     * @param keys Keys to evict readers for.
+     * @param entries Entries to check.
+     */
+    private void evictReaders(Collection<IgniteTxKey<K>> keys, @Nullable Collection<IgniteTxEntry<K, V>> entries) {
+        if (entries == null || entries.isEmpty())
+            return;
+
+        for (Iterator<IgniteTxEntry<K, V>> it = entries.iterator(); it.hasNext();) {
+            IgniteTxEntry<K, V> entry = it.next();
+
+            if (keys.contains(entry.txKey()))
+                it.remove();
+        }
+    }
+
+    /**
+     * Copies collection of entries if it is read-only.
+     */
+    private void ensureModifiable() {
+        if (readOnly) {
+            entries = new ConcurrentLinkedQueue<>(entries);
+
+            readOnly = false;
+        }
+    }
+
+    /** {@inheritDoc} */
+    public boolean empty() {
+        return entries.isEmpty();
+    }
+
+    /** {@inheritDoc} */
+    @Override public void writeExternal(ObjectOutput out) throws IOException {
+        out.writeObject(node);
+
+        U.writeCollection(out, entries);
+    }
+
+    /** {@inheritDoc} */
+    @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
+        node = (ClusterNode)in.readObject();
+
+        entries = U.readCollection(in);
+    }
+
+    /** {@inheritDoc} */
+    @Override public String toString() {
+        return S.toString(GridDistributedTxMapping.class, this, "node", node.id());
+    }
+}


Mime
View raw message