<commit_before>package com.hazelcast.simulator.tests.network;

import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.core.ILock;
import com.hazelcast.core.Member;
import com.hazelcast.instance.HazelcastThreadGroup;
import com.hazelcast.instance.Node;
import com.hazelcast.internal.metrics.MetricsRegistry;
import com.hazelcast.logging.ILogger;
import com.hazelcast.logging.Logger;
import com.hazelcast.logging.LoggingService;
import com.hazelcast.nio.Address;
import com.hazelcast.nio.Connection;
import com.hazelcast.nio.Packet;
import com.hazelcast.nio.tcp.IOThreadingModel;
import com.hazelcast.nio.tcp.PacketWriter;
import com.hazelcast.nio.tcp.TcpIpConnectionManager;
import com.hazelcast.nio.tcp.nonblocking.NonBlockingIOThreadingModel;
import com.hazelcast.nio.tcp.spinning.SpinningIOThreadingModel;
import com.hazelcast.simulator.test.TestContext;
import com.hazelcast.simulator.test.TestRunner;
import com.hazelcast.simulator.test.annotations.RunWithWorker;
import com.hazelcast.simulator.test.annotations.Setup;
import com.hazelcast.simulator.test.annotations.Teardown;
import com.hazelcast.simulator.test.annotations.Verify;
import com.hazelcast.simulator.test.annotations.Warmup;
import com.hazelcast.simulator.tests.helpers.HazelcastTestUtils;
import com.hazelcast.simulator.worker.tasks.AbstractMonotonicWorker;
import com.hazelcast.spi.impl.PacketHandler;

import java.nio.ByteBuffer;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import static com.hazelcast.simulator.tests.network.NetworkTest.IOThreadingModelEnum.NonBlocking;
import static com.hazelcast.simulator.utils.CommonUtils.sleepMillis;


public class NetworkTest {

    private static final int PORT_OFFSET = 1000;

    private static final ILogger LOGGER = Logger.getLogger(NetworkTest.class);

    public int payloadSize = 0;
    public long requestTimeout = 60;
    public TimeUnit requestTimeUnit = TimeUnit.SECONDS;
    public int inputThreadCount = 1;
    public int outputThreadCount = 1;
    public boolean inputSelectNow = false;
    public boolean outputSelectNow = false;
    public boolean socketNoDelay = true;
    public int socketReceiveBufferSize = 32;
    public int socketSendBufferSize = 32;
    public IOThreadingModelEnum ioThreadingModel = NonBlocking;
    public boolean trackSequenceId = false;

    private final AtomicInteger workerIdGenerator = new AtomicInteger();
    private HazelcastInstance hz;
    private ILock networkCreateLock;
    private TcpIpConnectionManager connectionManager;
    private DummyPacketHandler packetHandler;
    private List<Connection> connections = new LinkedList<Connection>();

    public enum IOThreadingModelEnum {
        NonBlocking,
        Spinning
    }

    @Setup
    public void setup(TestContext context) throws Exception {
        hz = context.getTargetInstance();

        Node node = HazelcastTestUtils.getNode(hz);
        if (node == null) {
            throw new IllegalStateException("node is null");
        }
        MetricsRegistry metricsRegistry = node.nodeEngine.getMetricsRegistry();
        LoggingService loggingService = node.loggingService;
        HazelcastThreadGroup threadGroup = node.getHazelcastThreadGroup();

        // we don't know the number of worker threads (damn hidden property), so lets assume 1000.. that should be enough
        packetHandler = new DummyPacketHandler(1000);

        Address thisAddress = node.getThisAddress();
        Address newThisAddress = new Address(thisAddress.getHost(), thisAddress.getPort() + PORT_OFFSET);
        LOGGER.info("ThisAddress: " + newThisAddress);
        MockIOService ioService = new MockIOService(newThisAddress, loggingService);
        ioService.inputThreadCount = inputThreadCount;
        ioService.outputThreadCount = outputThreadCount;
        ioService.socketNoDelay = socketNoDelay;
        ioService.packetHandler = packetHandler;
        ioService.socketSendBufferSize = socketSendBufferSize;
        ioService.socketReceiveBufferSize = socketReceiveBufferSize;

        if (trackSequenceId) {
            ioService.packetWriterFactory = new TaggingPacketWriterFactory();
        }

        IOThreadingModel threadingModel;
        switch (ioThreadingModel) {
            case NonBlocking:
                NonBlockingIOThreadingModel nonBlockingIOThreadingModel = new NonBlockingIOThreadingModel(
                        ioService, loggingService, metricsRegistry, threadGroup);
                nonBlockingIOThreadingModel.setInputSelectNow(inputSelectNow);
                nonBlockingIOThreadingModel.setOutputSelectNow(outputSelectNow);
                threadingModel = nonBlockingIOThreadingModel;
                break;
            case Spinning:
                threadingModel = new SpinningIOThreadingModel(ioService, loggingService, metricsRegistry, threadGroup);
                break;
            default:
                throw new IllegalStateException("Unrecognized threading model:" + ioThreadingModel);
        }

        connectionManager = new TcpIpConnectionManager(
                ioService, ioService.serverSocketChannel, loggingService, metricsRegistry, threadingModel);
        connectionManager.start();
        networkCreateLock = hz.getLock("connectionCreateLock");
    }

    @Teardown
    public void teardown() throws Exception {
        connectionManager.shutdown();
    }

    @Warmup
    public void warmup() throws Exception {
        networkCreateLock.lock();
        try {
            LOGGER.info("Starting connections: " + (hz.getCluster().getMembers().size() - 1));
            for (Member member : hz.getCluster().getMembers()) {

                if (member.localMember()) {
                    continue;
                }

                Address memberAddress = member.getAddress();
                Address targetAddress = new Address(memberAddress.getHost(), memberAddress.getPort() + PORT_OFFSET);

                LOGGER.info("Connecting to: " + targetAddress);

                connectionManager.getOrConnect(targetAddress);

                Connection connection;
                for (; ; ) {
                    connection = connectionManager.getConnection(targetAddress);
                    if (connection != null) {
                        break;
                    }

                    LOGGER.info("Waiting for connection to: " + targetAddress);
                    sleepMillis(100);
                }

                connections.add(connection);

                LOGGER.info("Successfully created connection to: " + targetAddress);
            }

            LOGGER.info("Successfully started all connections");
        } finally {
            networkCreateLock.unlock();
        }
    }

    class TaggingPacketWriterFactory implements PacketWriterFactory {

        @Override
        public PacketWriter create() {
            return new TaggingPacketWriter();
        }
    }

    class TaggingPacketWriter implements PacketWriter {

        private long sequenceId = 0;

        @Override
        public boolean write(Packet packet, ByteBuffer dst) throws Exception {
            if (packet.dataSize() > 100) {
                byte[] payload = packet.toByteArray();
                // we also stuff in a sequence id at the beginning
                long s = sequenceId;
                for (int i = 7; i >= 0; i--) {
                    payload[i + 3] = (byte) (s & 0xFF);
                    s >>= 8;
                }<commit_msg>Cleanup in adding tag to the packet<commit_after>package com.hazelcast.simulator.tests.network;

import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.core.ILock;
import com.hazelcast.core.Member;
import com.hazelcast.instance.HazelcastThreadGroup;
import com.hazelcast.instance.Node;
import com.hazelcast.internal.metrics.MetricsRegistry;
import com.hazelcast.logging.ILogger;
import com.hazelcast.logging.Logger;
import com.hazelcast.logging.LoggingService;
import com.hazelcast.nio.Address;
import com.hazelcast.nio.Connection;
import com.hazelcast.nio.Packet;
import com.hazelcast.nio.tcp.IOThreadingModel;
import com.hazelcast.nio.tcp.PacketWriter;
import com.hazelcast.nio.tcp.TcpIpConnectionManager;
import com.hazelcast.nio.tcp.nonblocking.NonBlockingIOThreadingModel;
import com.hazelcast.nio.tcp.spinning.SpinningIOThreadingModel;
import com.hazelcast.simulator.test.TestContext;
import com.hazelcast.simulator.test.TestRunner;
import com.hazelcast.simulator.test.annotations.RunWithWorker;
import com.hazelcast.simulator.test.annotations.Setup;
import com.hazelcast.simulator.test.annotations.Teardown;
import com.hazelcast.simulator.test.annotations.Verify;
import com.hazelcast.simulator.test.annotations.Warmup;
import com.hazelcast.simulator.tests.helpers.HazelcastTestUtils;
import com.hazelcast.simulator.worker.tasks.AbstractMonotonicWorker;
import com.hazelcast.spi.impl.PacketHandler;

import java.nio.ByteBuffer;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import static com.hazelcast.simulator.tests.network.NetworkTest.IOThreadingModelEnum.NonBlocking;
import static com.hazelcast.simulator.utils.CommonUtils.sleepMillis;


public class NetworkTest {

    private static final int PORT_OFFSET = 1000;

    private static final ILogger LOGGER = Logger.getLogger(NetworkTest.class);

    public int payloadSize = 0;
    public long requestTimeout = 60;
    public TimeUnit requestTimeUnit = TimeUnit.SECONDS;
    public int inputThreadCount = 1;
    public int outputThreadCount = 1;
    public boolean inputSelectNow = false;
    public boolean outputSelectNow = false;
    public boolean socketNoDelay = true;
    public int socketReceiveBufferSize = 32;
    public int socketSendBufferSize = 32;
    public IOThreadingModelEnum ioThreadingModel = NonBlocking;
    public boolean trackSequenceId = false;

    private final AtomicInteger workerIdGenerator = new AtomicInteger();
    private HazelcastInstance hz;
    private ILock networkCreateLock;
    private TcpIpConnectionManager connectionManager;
    private DummyPacketHandler packetHandler;
    private List<Connection> connections = new LinkedList<Connection>();

    public enum IOThreadingModelEnum {
        NonBlocking,
        Spinning
    }

    @Setup
    public void setup(TestContext context) throws Exception {
        hz = context.getTargetInstance();

        Node node = HazelcastTestUtils.getNode(hz);
        if (node == null) {
            throw new IllegalStateException("node is null");
        }
        MetricsRegistry metricsRegistry = node.nodeEngine.getMetricsRegistry();
        LoggingService loggingService = node.loggingService;
        HazelcastThreadGroup threadGroup = node.getHazelcastThreadGroup();

        // we don't know the number of worker threads (damn hidden property), so lets assume 1000.. that should be enough
        packetHandler = new DummyPacketHandler(1000);

        Address thisAddress = node.getThisAddress();
        Address newThisAddress = new Address(thisAddress.getHost(), thisAddress.getPort() + PORT_OFFSET);
        LOGGER.info("ThisAddress: " + newThisAddress);
        MockIOService ioService = new MockIOService(newThisAddress, loggingService);
        ioService.inputThreadCount = inputThreadCount;
        ioService.outputThreadCount = outputThreadCount;
        ioService.socketNoDelay = socketNoDelay;
        ioService.packetHandler = packetHandler;
        ioService.socketSendBufferSize = socketSendBufferSize;
        ioService.socketReceiveBufferSize = socketReceiveBufferSize;

        if (trackSequenceId) {
            ioService.packetWriterFactory = new TaggingPacketWriterFactory();
        }

        IOThreadingModel threadingModel;
        switch (ioThreadingModel) {
            case NonBlocking:
                NonBlockingIOThreadingModel nonBlockingIOThreadingModel = new NonBlockingIOThreadingModel(
                        ioService, loggingService, metricsRegistry, threadGroup);
                nonBlockingIOThreadingModel.setInputSelectNow(inputSelectNow);
                nonBlockingIOThreadingModel.setOutputSelectNow(outputSelectNow);
                threadingModel = nonBlockingIOThreadingModel;
                break;
            case Spinning:
                threadingModel = new SpinningIOThreadingModel(ioService, loggingService, metricsRegistry, threadGroup);
                break;
            default:
                throw new IllegalStateException("Unrecognized threading model:" + ioThreadingModel);
        }

        connectionManager = new TcpIpConnectionManager(
                ioService, ioService.serverSocketChannel, loggingService, metricsRegistry, threadingModel);
        connectionManager.start();
        networkCreateLock = hz.getLock("connectionCreateLock");
    }

    @Teardown
    public void teardown() throws Exception {
        connectionManager.shutdown();
    }

    @Warmup
    public void warmup() throws Exception {
        networkCreateLock.lock();
        try {
            LOGGER.info("Starting connections: " + (hz.getCluster().getMembers().size() - 1));
            for (Member member : hz.getCluster().getMembers()) {

                if (member.localMember()) {
                    continue;
                }

                Address memberAddress = member.getAddress();
                Address targetAddress = new Address(memberAddress.getHost(), memberAddress.getPort() + PORT_OFFSET);

                LOGGER.info("Connecting to: " + targetAddress);

                connectionManager.getOrConnect(targetAddress);

                Connection connection;
                for (; ; ) {
                    connection = connectionManager.getConnection(targetAddress);
                    if (connection != null) {
                        break;
                    }

                    LOGGER.info("Waiting for connection to: " + targetAddress);
                    sleepMillis(100);
                }

                connections.add(connection);

                LOGGER.info("Successfully created connection to: " + targetAddress);
            }

            LOGGER.info("Successfully started all connections");
        } finally {
            networkCreateLock.unlock();
        }
    }

    class TaggingPacketWriterFactory implements PacketWriterFactory {

        @Override
        public PacketWriter create() {
            return new TaggingPacketWriter();
        }
    }

    class TaggingPacketWriter implements PacketWriter {

        private long sequenceId = 1;

        @Override
        public boolean write(Packet packet, ByteBuffer dst) throws Exception {
            if (packet.dataSize() > 100) {
                byte[] payload = packet.toByteArray();
                // we also stuff in a sequence id at the beginning
                long s = sequenceId;
                for (int i = 7; i >= 0; i--) {
                    payload[i + 3] = (byte) (s & 0xFF);
                    s >>= 8;
                }
{'fraction_non_alphanumeric': 0.08034923446792357, 'fraction_numerical': 0.004302163735290396, 'mean_word_length': 2.9389484176426612, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 15, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '681418', 'n_tokens_mistral': 4250, 'n_tokens_neox': 4082, 'n_words': 878}