ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From agoncha...@apache.org
Subject [06/39] incubator-ignite git commit: # Removed shmem from communication
Date Mon, 16 Mar 2015 05:33:10 GMT
# Removed shmem from communication


Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/c014b001
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/c014b001
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/c014b001

Branch: refs/heads/ignite-45
Commit: c014b0017cc271a5942e242f0d28d075355e94a6
Parents: 3f23de3
Author: Valentin Kulichenko <vkulichenko@gridgain.com>
Authored: Fri Mar 13 17:20:36 2015 -0700
Committer: Valentin Kulichenko <vkulichenko@gridgain.com>
Committed: Fri Mar 13 17:20:36 2015 -0700

----------------------------------------------------------------------
 .../util/nio/GridShmemCommunicationClient.java  | 146 -------------------
 .../communication/tcp/TcpCommunicationSpi.java  |  44 ++----
 2 files changed, 10 insertions(+), 180 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/c014b001/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridShmemCommunicationClient.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridShmemCommunicationClient.java
b/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridShmemCommunicationClient.java
deleted file mode 100644
index b0a6df3..0000000
--- a/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridShmemCommunicationClient.java
+++ /dev/null
@@ -1,146 +0,0 @@
-/*
- * 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.util.nio;
-
-import org.apache.ignite.*;
-import org.apache.ignite.internal.util.ipc.shmem.*;
-import org.apache.ignite.internal.util.lang.*;
-import org.apache.ignite.internal.util.typedef.internal.*;
-import org.apache.ignite.plugin.extensions.communication.*;
-import org.jetbrains.annotations.*;
-
-import java.io.*;
-import java.nio.*;
-import java.util.*;
-
-/**
- *
- */
-public class GridShmemCommunicationClient extends GridAbstractCommunicationClient {
-    /** */
-    private final IpcSharedMemoryClientEndpoint shmem;
-
-    /** */
-    private final ByteBuffer writeBuf;
-
-    /** */
-    private final MessageFormatter formatter;
-
-    /**
-     * @param metricsLsnr Metrics listener.
-     * @param port Shared memory IPC server port.
-     * @param connTimeout Connection timeout.
-     * @param log Logger.
-     * @param formatter Message formatter.
-     * @throws IgniteCheckedException If failed.
-     */
-    public GridShmemCommunicationClient(GridNioMetricsListener metricsLsnr, int port, long
connTimeout,
-        IgniteLogger log, MessageFormatter formatter) throws IgniteCheckedException {
-        super(metricsLsnr);
-
-        assert metricsLsnr != null;
-        assert port > 0 && port < 0xffff;
-        assert connTimeout >= 0;
-
-        shmem = new IpcSharedMemoryClientEndpoint(port, (int)connTimeout, log);
-
-        writeBuf = ByteBuffer.allocate(8 << 10);
-
-        writeBuf.order(ByteOrder.nativeOrder());
-
-        this.formatter = formatter;
-    }
-
-    /** {@inheritDoc} */
-    @Override public synchronized void doHandshake(IgniteInClosure2X<InputStream, OutputStream>
handshakeC)
-        throws IgniteCheckedException {
-        handshakeC.applyx(shmem.inputStream(), shmem.outputStream());
-    }
-
-    /** {@inheritDoc} */
-    @Override public boolean close() {
-        boolean res = super.close();
-
-        if (res)
-            shmem.close();
-
-        return res;
-    }
-
-    /** {@inheritDoc} */
-    @Override public void forceClose() {
-        super.forceClose();
-
-        // Do not call forceClose() here.
-        shmem.close();
-    }
-
-    /** {@inheritDoc} */
-    @Override public synchronized void sendMessage(byte[] data, int len) throws IgniteCheckedException
{
-        if (closed())
-            throw new IgniteCheckedException("Communication client was closed: " + this);
-
-        try {
-            shmem.outputStream().write(data, 0, len);
-
-            metricsLsnr.onBytesSent(len);
-        }
-        catch (IOException e) {
-            throw new IgniteCheckedException("Failed to send message to remote node: " +
shmem, e);
-        }
-
-        markUsed();
-    }
-
-    /** {@inheritDoc} */
-    @Override public synchronized boolean sendMessage(@Nullable UUID nodeId, Message msg)
-        throws IgniteCheckedException {
-        if (closed())
-            throw new IgniteCheckedException("Communication client was closed: " + this);
-
-        assert writeBuf.hasArray();
-
-        try {
-            int cnt = U.writeMessageFully(msg, shmem.outputStream(), writeBuf, formatter.writer());
-
-            metricsLsnr.onBytesSent(cnt);
-        }
-        catch (IOException e) {
-            throw new IgniteCheckedException("Failed to send message to remote node: " +
shmem, e);
-        }
-
-        markUsed();
-
-        return false;
-    }
-
-    /** {@inheritDoc} */
-    @Override public void sendMessage(ByteBuffer data) throws IgniteCheckedException {
-        throw new UnsupportedOperationException();
-    }
-
-    /** {@inheritDoc} */
-    @Override public void flushIfNeeded(long timeout) throws IOException {
-        // No-op.
-    }
-
-    /** {@inheritDoc} */
-    @Override public String toString() {
-        return S.toString(GridShmemCommunicationClient.class, this, super.toString());
-    }
-}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/c014b001/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpi.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpi.java
b/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpi.java
index b38dc15..65eed6a 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpi.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpi.java
@@ -319,8 +319,6 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter
 
                     GridCommunicationClient oldClient = clients.get(sndId);
 
-                    boolean hasShmemClient = false;
-
                     if (oldClient != null) {
                         if (oldClient instanceof GridTcpNioCommunicationClient) {
                             if (log.isDebugEnabled())
@@ -332,11 +330,6 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter
 
                             return;
                         }
-                        else {
-                            assert oldClient instanceof GridShmemCommunicationClient;
-
-                            hasShmemClient = true;
-                        }
                     }
 
                     GridFutureAdapter<GridCommunicationClient> fut = new GridFutureAdapter<>();
@@ -363,15 +356,10 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter
 
                                 return;
                             }
-                            else {
-                                assert oldClient instanceof GridShmemCommunicationClient;
-
-                                hasShmemClient = true;
-                            }
                         }
 
                         boolean reserved = recoveryDesc.tryReserve(msg0.connectCount(),
-                            new ConnectClosure(ses, recoveryDesc, rmtNode, msg0, !hasShmemClient,
fut));
+                            new ConnectClosure(ses, recoveryDesc, rmtNode, msg0, fut));
 
                         if (log.isDebugEnabled())
                             log.debug("Received incoming connection from remote node " +
@@ -380,7 +368,7 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter
                         if (reserved) {
                             try {
                                 GridTcpNioCommunicationClient client =
-                                    connected(recoveryDesc, ses, rmtNode, msg0.received(),
true, !hasShmemClient);
+                                    connected(recoveryDesc, ses, rmtNode, msg0.received(),
true);
 
                                 fut.onDone(client);
                             }
@@ -402,11 +390,11 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter
                         }
                         else {
                             boolean reserved = recoveryDesc.tryReserve(msg0.connectCount(),
-                                new ConnectClosure(ses, recoveryDesc, rmtNode, msg0, !hasShmemClient,
fut));
+                                new ConnectClosure(ses, recoveryDesc, rmtNode, msg0, fut));
 
                             if (reserved) {
                                 GridTcpNioCommunicationClient client =
-                                    connected(recoveryDesc, ses, rmtNode, msg0.received(),
true, !hasShmemClient);
+                                    connected(recoveryDesc, ses, rmtNode, msg0.received(),
true);
 
                                 fut.onDone(client);
                             }
@@ -474,7 +462,6 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter
              * @param node Node.
              * @param rcvCnt Number of received messages..
              * @param sndRes If {@code true} sends response for recovery handshake.
-             * @param createClient If {@code true} creates NIO communication client.
              * @return Client.
              */
             private GridTcpNioCommunicationClient connected(
@@ -482,8 +469,7 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter
                 GridNioSession ses,
                 ClusterNode node,
                 long rcvCnt,
-                boolean sndRes,
-                boolean createClient) {
+                boolean sndRes) {
                 recovery.onHandshake(rcvCnt);
 
                 ses.recoveryDescriptor(recovery);
@@ -495,16 +481,12 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter
 
                 recovery.connected();
 
-                GridTcpNioCommunicationClient client = null;
-
-                if (createClient) {
-                    client = new GridTcpNioCommunicationClient(ses, log);
+                GridTcpNioCommunicationClient client = new GridTcpNioCommunicationClient(ses,
log);
 
-                    GridCommunicationClient oldClient = clients.putIfAbsent(node.id(), client);
+                GridCommunicationClient oldClient = clients.putIfAbsent(node.id(), client);
 
-                    assert oldClient == null : "Client already created [node=" + node + ",
client=" + client +
-                            ", oldClient=" + oldClient + ", recoveryDesc=" + recovery + ']';
-                }
+                assert oldClient == null : "Client already created [node=" + node + ", client="
+ client +
+                        ", oldClient=" + oldClient + ", recoveryDesc=" + recovery + ']';
 
                 return client;
             }
@@ -532,28 +514,22 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter
                 /** */
                 private final GridFutureAdapter<GridCommunicationClient> fut;
 
-                /** */
-                private final boolean createClient;
-
                 /**
                  * @param ses Incoming session.
                  * @param recoveryDesc Recovery descriptor.
                  * @param rmtNode Remote node.
                  * @param msg Handshake message.
-                 * @param createClient If {@code true} creates NIO communication client..
                  * @param fut Connect future.
                  */
                 ConnectClosure(GridNioSession ses,
                     GridNioRecoveryDescriptor recoveryDesc,
                     ClusterNode rmtNode,
                     HandshakeMessage msg,
-                    boolean createClient,
                     GridFutureAdapter<GridCommunicationClient> fut) {
                     this.ses = ses;
                     this.recoveryDesc = recoveryDesc;
                     this.rmtNode = rmtNode;
                     this.msg = msg;
-                    this.createClient = createClient;
                     this.fut = fut;
                 }
 
@@ -566,7 +542,7 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter
                                     msgFut.get();
 
                                     GridTcpNioCommunicationClient client =
-                                        connected(recoveryDesc, ses, rmtNode, msg.received(),
false, createClient);
+                                        connected(recoveryDesc, ses, rmtNode, msg.received(),
false);
 
                                     fut.onDone(client);
                                 }


Mime
View raw message