/*
 * 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.loadtests.communication;
//
//import org.apache.ignite.*;
//import org.apache.ignite.lang.*;
//import org.apache.ignite.lang.utils.*;
//import org.apache.ignite.spi.*;
//import org.apache.ignite.spi.communication.*;
//import org.apache.ignite.spi.communication.tcp.*;
//import org.apache.ignite.typedef.*;
//import org.apache.ignite.typedef.internal.*;
//import org.apache.ignite.testframework.*;
//import org.apache.ignite.testframework.junits.*;
//import org.apache.ignite.testframework.junits.spi.*;
//
//import java.util.*;
//import java.util.concurrent.*;
//import java.util.concurrent.atomic.*;
//
///** */
//@GridSpiTest(spi = GridTcpCommunicationSpi.class, group = "TCP communication SPI benchmark.")
//public class GridTcpCommunicationBenchmark extends GridSpiAbstractTest<GridTcpCommunicationSpi> {
//    /** */
//    public static final int CONCUR_MSGS = 10 * 1024;
//
//    /** */
//    private static final int THREADS = 1;
//
//    /** */
//    private static final long TEST_TIMEOUT = 3 * 60 * 1000;
//
//    /** */
//    private final Collection<GridTestResources> spiRsrcs = new ArrayList<>();
//
//    /** */
//    private final Map<UUID, GridCommunicationSpi> spis = new HashMap<>();
//
//    /** */
//    private final Collection<ClusterNode> nodes = new ArrayList<>();
//
//    /**
//     * Disable automatic test SPI start.
//     */
//    public GridTcpCommunicationBenchmark() {
//        super(false);
//    }
//
//    /** {@inheritDoc} */
//    @Override protected void beforeTest() throws Exception {
//        Map<ClusterNode, GridSpiTestContext> ctxs = new HashMap<>();
//
//        for (int i = 0; i < 2; i++) {
//            GridCommunicationSpi spi = getCommunication();
//
//            GridTestResources rsrcs = new GridTestResources();
//
//            GridTestNode node = new GridTestNode(rsrcs.getNodeId());
//
//            GridSpiTestContext ctx = initSpiContext();
//
//            ctx.setLocalNode(node);
//
//            spiRsrcs.add(rsrcs);
//
//            rsrcs.inject(spi);
//
//            node.setAttributes(spi.getNodeAttributes());
//
//            nodes.add(node);
//
//            spi.spiStart(getTestIgniteInstanceName() + (i + 1));
//
//            spis.put(rsrcs.getNodeId(), spi);
//
//            spi.onContextInitialized(ctx);
//
//            ctxs.put(node, ctx);
//        }
//
//        // For each context set remote nodes.
//        for (Map.Entry<ClusterNode, GridSpiTestContext> e : ctxs.entrySet()) {
//            for (ClusterNode n : nodes) {
//                if (!n.equals(e.getKey()))
//                    e.getValue().remoteNodes().add(n);
//            }
//        }
//    }
//
//    /** {@inheritDoc} */
//    @Override protected void afterTest() throws Exception {
//        for (GridCommunicationSpi spi : spis.values()) {
//            spi.setListener(null);
//
//            spi.spiStop();
//        }
//
//        for (GridTestResources rsrcs : spiRsrcs)
//            rsrcs.stopThreads();
//    }
//
//    /**
//     * @param len Length.
//     * @return Test string.
//     */
//    private static String generateTestString(int len) {
//        assert len > 0;
//        SB sb = new SB();
//
//        for (int i = 0; i < len; i++)
//            sb.a(Character.forDigit(i % 10, 10));
//
//        return sb.toString();
//    }
//
//    /**
//     * @throws Exception If failed.
//     */
//    @SuppressWarnings("deprecation")
//    public void testThroughput() throws Exception {
//        assert spis.size() == 2;
//        assert nodes.size() == 2;
//
//        Iterator<ClusterNode> it = nodes.iterator();
//
//        final ClusterNode sndNode = it.next();
//        final ClusterNode rcvNode = it.next();
//
//        final GridCommunicationSpi sndComm = spis.get(sndNode.id());
//        final GridCommunicationSpi rcvComm = spis.get(rcvNode.id());
//
//        final String testStr = generateTestString(66);
//
//        info("Test string length: " + testStr.length());
//        info("Senders: " + THREADS);
//        info("Messages: " + CONCUR_MSGS);
//
//        final Semaphore sem = new Semaphore(CONCUR_MSGS);
//        final LongAdder8 msgCntr = new LongAdder8();
//
//        rcvComm.setListener(new GridCommunicationListener() {
//            @Override public void onMessage(UUID nodeId, byte[] msg, GridAbsClosure msgC) {
//                try {
//                    byte[] res = U.join(U.intToBytes(msg.length), msg);
//
//                    rcvComm.sendMessage(sndNode, res, 0, res.length);
//                }
//                catch (GridSpiException e) {
//                    log.error("Message echo failed.", e);
//                }
//                finally {
//                    msgC.apply();
//                }
//            }
//        });
//
//        sndComm.setListener(new GridCommunicationListener() {
//            @Override public void onMessage(UUID nodeId, byte[] msg, GridAbsClosure msgC) {
//                msgCntr.increment();
//
//                sem.release();
//
//                msgC.apply();
//            }
//        });
//
//        Timer t = new Timer("results-reporter");
//
//        t.schedule(new TimerTask() {
//            private long ts = System.currentTimeMillis();
//
//            @Override public void run() {
//                long newTs = System.currentTimeMillis();
//                long qrys = msgCntr.sumThenReset();
//
//                long time = newTs - ts;
//
//                X.println("Communication benchmark [qps=" + qrys * 1000 / time +
//                    ", executed=" + qrys + ", time=" + time + ']');
//
//                ts = newTs;
//            }
//        }, 10000, 10000);
//
//        final AtomicBoolean finish = new AtomicBoolean();
//
//        GridFuture<?> f = GridTestUtils.runMultiThreadedAsync(new Callable<Object>() {
//            @Override public Object call() throws Exception {
//                try {
//                    while (!finish.get()) {
//                        IgniteUuid msgId = IgniteUuid.randomUuid();
//
//                        sem.acquire();
//
//                        // Loading message with additional data, to get results,
//                        // comparable with IoManager benchmark.
//                        GridTuple2<byte[], Integer> t = getTestResources().getMarshaller().
//                            marshalNoCopy(new GridTestMessage(msgId, testStr), 4);
//
//                        byte[] buf = t.get1();
//                        int len = t.get2();
//
//                        U.intToBytes(len - 4, buf, 0);
//
//                        sndComm.sendMessage(rcvNode, buf, 0, len);
//                    }
//                }
//                catch (IgniteCheckedException e) {
//                    X.println("Message send failed", e);
//                }
//                catch (InterruptedException ignored) {
//                    // No-op.
//                }
//
//                return null;
//            }
//        }, THREADS, "send-thread");
//
//        Thread.sleep(TEST_TIMEOUT);
//
//        finish.set(true);
//
//        sem.release(CONCUR_MSGS * 2);
//
//        t.cancel();
//
//        f.get();
//    }
//
//    /**
//     * @throws Exception If failed.
//     */
//    @SuppressWarnings("deprecation")
//    public void testLatency() throws Exception {
//        assert spis.size() == 2;
//        assert nodes.size() == 2;
//
//        Iterator<ClusterNode> it = nodes.iterator();
//
//        final ClusterNode sndNode = it.next();
//        final ClusterNode rcvNode = it.next();
//
//        final GridCommunicationSpi sndComm = spis.get(sndNode.id());
//        final GridCommunicationSpi rcvComm = spis.get(rcvNode.id());
//
//        final String testStr = generateTestString(66);
//
//        info("Test string length: " + testStr.length());
//
//        final LongAdder8 msgCntr = new LongAdder8();
//
//        final Map<IgniteUuid, CountDownLatch> map = new ConcurrentHashMap8<>();
//
//        rcvComm.setListener(new GridCommunicationListener() {
//            @Override public void onMessage(UUID nodeId, byte[] msg, GridAbsClosure msgC) {
//                try {
//                    byte[] res = U.join(U.intToBytes(msg.length), msg);
//
//                    rcvComm.sendMessage(sndNode, res, 0, res.length);
//                }
//                catch (GridSpiException e) {
//                    log.error("Message echo failed.", e);
//                }
//                finally {
//                    msgC.apply();
//                }
//            }
//        });
//
//        final ClassLoader clsLdr = getClass().getClassLoader();
//
//        sndComm.setListener(new GridCommunicationListener() {
//            @Override public void onMessage(UUID nodeId, byte[] msg, GridAbsClosure msgC) {
//                try {
//                    GridTestMessage testMsg = getTestResources().getMarshaller().unmarshal(msg, clsLdr);
//
//                    map.get(testMsg.id()).countDown();
//                }
//                catch (IgniteCheckedException e) {
//                    U.error(log, "Failed to ", e);
//                }
//                finally {
//                    msgC.apply();
//                }
//            }
//        });
//
//        Timer t = new Timer("results-reporter");
//
//        t.schedule(new TimerTask() {
//            private long ts = System.currentTimeMillis();
//
//            @Override public void run() {
//                long newTs = System.currentTimeMillis();
//                long qrys = msgCntr.sumThenReset();
//
//                long time = newTs - ts;
//
//                X.println("Communication benchmark [qps=" + qrys * 1000 / time +
//                    ", executed=" + qrys + ", time=" + time + ']');
//
//                ts = newTs;
//            }
//        }, 10000, 10000);
//
//        final AtomicBoolean finish = new AtomicBoolean();
//
//        GridFuture<?> f = GridTestUtils.runMultiThreadedAsync(new Callable<Object>() {
//            @Override public Object call() throws Exception {
//                info("Test thread started.");
//
//                try {
//                    IgniteUuid msgId = IgniteUuid.randomUuid();
//
//                    GridTuple2<byte[], Integer> t = getTestResources().getMarshaller().
//                        marshalNoCopy(new GridTestMessage(msgId, testStr), 4);
//
//                    byte[] buf = t.get1();
//                    int len = t.get2();
//
//                    U.intToBytes(len - 4, buf, 0);
//
//                    while (!finish.get()) {
//                        // Loading message with additional data, to get results,
//                        // comparable with IoManager benchmark.
//                        CountDownLatch latch = new CountDownLatch(1);
//
//                        map.put(msgId, latch);
//
//                        sndComm.sendMessage(rcvNode, buf, 0, len);
//
//                        latch.await();
//
//                        msgCntr.increment();
//                    }
//                }
//                catch (IgniteCheckedException e) {
//                    X.println("Message send failed", e);
//                }
//                catch (InterruptedException ignored) {
//                    // No-op.
//                }
//
//                return null;
//            }
//        }, 2, "send-thread");
//
//        Thread.sleep(TEST_TIMEOUT);
//
//        finish.set(true);
//
//        t.cancel();
//
//        f.get();
//    }
//
//    /**
//     * @throws Exception If failed.
//     */
//    @SuppressWarnings("deprecation")
//    public void testVariableLoad() throws Exception {
//        assert spis.size() == 2;
//        assert nodes.size() == 2;
//
//        Iterator<ClusterNode> it = nodes.iterator();
//
//        final ClusterNode sndNode = it.next();
//        final ClusterNode rcvNode = it.next();
//
//        final GridCommunicationSpi sndComm = spis.get(sndNode.id());
//        final GridCommunicationSpi rcvComm = spis.get(rcvNode.id());
//
//        final String testStr = generateTestString(16);
//
//        info("Test string length: " + testStr.length());
//        info("Senders: " + THREADS);
//        info("Messages: " + CONCUR_MSGS);
//
//        final Semaphore sem = new Semaphore(CONCUR_MSGS);
//        final LongAdder8 msgCntr = new LongAdder8();
//
//        final Map<IgniteUuid, CountDownLatch> latches = new ConcurrentHashMap8<>();
//
//        rcvComm.setListener(new GridCommunicationListener() {
//            @Override public void onMessage(UUID nodeId, byte[] msg, GridAbsClosure msgC) {
//                try {
//                    byte[] res = U.join(U.intToBytes(msg.length), msg);
//
//                    rcvComm.sendMessage(sndNode, res, 0, res.length);
//                }
//                catch (GridSpiException e) {
//                    log.error("Message echo failed.", e);
//                }
//                finally {
//                    msgC.apply();
//                }
//            }
//        });
//
//        sndComm.setListener(new GridCommunicationListener() {
//            @Override public void onMessage(UUID nodeId, byte[] buf, GridAbsClosure msgC) {
//                msgCntr.increment();
//
//                sem.release();
//
//                GridTestMessage msg = null;
//
//                try {
//                    msg = getTestResources().getMarshaller().unmarshal(buf, U.gridClassLoader());
//                }
//                catch (IgniteCheckedException e) {
//                    U.error(log, "Failed to unmarshal message.", e);
//
//                    fail();
//                }
//                finally {
//                    msgC.apply();
//                }
//
//                CountDownLatch latch = latches.get(msg.id());
//
//                if (latch != null)
//                    latch.countDown();
//            }
//        });
//
//        final AtomicBoolean finish = new AtomicBoolean();
//        final AtomicReference<CountDownLatch> latchRef = new AtomicReference<>();
//
//        GridFuture<?> f = GridTestUtils.runMultiThreadedAsync(new Callable<Object>() {
//            @Override public Object call() throws Exception {
//                while (!finish.get()) {
//                    CountDownLatch latch = latchRef.get();
//
//                    if (latch != null)
//                        U.await(latch);
//
//                    IgniteUuid msgId = IgniteUuid.randomUuid();
//
//                    sem.acquire();
//
//                    // Loading message with additional data, to get results,
//                    // comparable with IoManager benchmark.
//                    GridTuple2<byte[], Integer> t = getTestResources().getMarshaller().
//                        marshalNoCopy(new GridTestMessage(msgId, testStr), 4);
//
//                    byte[] buf = t.get1();
//                    int len = t.get2();
//
//                    U.intToBytes(len - 4, buf, 0);
//
//                    sndComm.sendMessage(rcvNode, buf, 0, len);
//                }
//
//                return null;
//            }
//        }, THREADS, "send-thread");
//
//        GridFuture<?> f1 = GridTestUtils.runMultiThreadedAsync(new Callable<Object>() {
//            private long ts = System.currentTimeMillis();
//
//            @Override public Object call() throws Exception {
//                try {
//                    while (!finish.get()) {
//                        info(U.nl() + ">>>" + U.nl() + ">>> High load." + U.nl() + ">>>");
//
//                        U.sleep(15 * 1000);
//
//                        reportNumbers();
//
//                        info(U.nl() + ">>>" + U.nl() + ">>> Low load." + U.nl() + ">>>");
//
//                        CountDownLatch latch = new CountDownLatch(1);
//
//                        try {
//                            // Here will be a pause.
//                            latchRef.set(latch);
//
//                            U.sleep(7 * 1000);
//
//                            reportNumbers();
//                        }
//                        finally {
//                            latch.countDown();
//                        }
//                    }
//                }
//                catch (IgniteCheckedException e) {
//                    X.println("Message send failed", e);
//                }
//
//                return null;
//            }
//
//            /**
//             *
//             */
//            void reportNumbers() {
//                long newTs = System.currentTimeMillis();
//                long qrys = msgCntr.sumThenReset();
//
//                long time = newTs - ts;
//
//                X.println("Communication benchmark [qps=" + qrys * 1000 / time +
//                    ", executed=" + qrys + ", time=" + time + ']');
//
//                ts = newTs;
//
//            }
//        }, 1, "load-dispatcher");
//
//        GridFuture<?> f2 = GridTestUtils.runMultiThreadedAsync(new Callable<Object>() {
//            @Override public Object call() throws Exception {
//                while (!finish.get()) {
//                    U.sleep(1000);
//
//                    IgniteUuid msgId = IgniteUuid.randomUuid();
//                    CountDownLatch latch = new CountDownLatch(1);
//
//                    latches.put(msgId, latch);
//
//                    GridTuple2<byte[], Integer> t = getTestResources().getMarshaller().
//                        marshalNoCopy(new GridTestMessage(msgId, testStr), 4);
//
//                    byte[] buf = t.get1();
//                    int len = t.get2();
//
//                    U.intToBytes(len - 4, buf, 0);
//
//                    sndComm.sendMessage(rcvNode, buf, 0, len);
//
//                    long start = System.currentTimeMillis();
//
//                    latch.await();
//
//                    info("Response time: " + (System.currentTimeMillis() - start));
//                }
//
//                return null;
//            }
//        }, THREADS, "low-loader");
//
//        Thread.sleep(TEST_TIMEOUT);
//
//        finish.set(true);
//
//        sem.release(CONCUR_MSGS * 2);
//
//        f.get();
//        f1.get();
//        f2.get();
//    }
//
//    /**
//     * @return SPI instance.
//     */
//    private GridCommunicationSpi getCommunication() {
//        GridTcpCommunicationSpi spi = new GridTcpCommunicationSpi();
//
//        spi.setSharedMemoryPort(-1);
//        spi.setNoDelay(true);
//        spi.setLocalAddress("127.0.0.1");
//
//        return spi;
//    }
//
//    /** {@inheritDoc} */
//    @Override protected long getTestTimeout() {
//        return TEST_TIMEOUT + 60 * 1000;
//    }
//}
