<commit_before>package org.nd4j.aeron.ipc;

import io.aeron.Aeron;
import io.aeron.driver.MediaDriver;
import io.aeron.driver.ThreadingMode;
import org.agrona.CloseHelper;
import org.agrona.concurrent.BusySpinIdleStrategy;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.nd4j.linalg.api.ndarray.INDArray;
import org.nd4j.linalg.factory.Nd4j;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;

import static org.junit.Assert.assertFalse;

/**
 * Created by agibsonccc on 9/22/16.
 */
public class NdArrayIpcTest {
    private MediaDriver mediaDriver;
    private static Logger log = LoggerFactory.getLogger(NdArrayIpcTest.class);
    private Aeron.Context ctx;
    private String channel = "aeron:udp?endpoint=localhost:40123";
    private int streamId = 10;

    @Before
    public void before() {
        final MediaDriver.Context ctx = new MediaDriver.Context()
                .threadingMode(ThreadingMode.DEDICATED)
                .dirsDeleteOnStart(true)
                .termBufferSparseFile(false)
                .conductorIdleStrategy(new BusySpinIdleStrategy())
                .receiverIdleStrategy(new BusySpinIdleStrategy())
                .senderIdleStrategy(new BusySpinIdleStrategy());
        mediaDriver = MediaDriver.launchEmbedded(ctx);
        System.out.println("Using media driver directory " + mediaDriver.aeronDirectoryName());
        System.out.println("Launched media driver");
    }

    @After
    public void after() {
        CloseHelper.quietClose(mediaDriver);
    }

    @Test
    public void testMultiThreadedIpcBig() throws Exception {
        ExecutorService executorService = Executors.newFixedThreadPool(4);
        INDArray arr = Nd4j.ones((int) 1e7);

        final AtomicBoolean running = new AtomicBoolean(true);
        Aeron aeron = Aeron.connect(getContext());
        int numSubscribers = 10;
        AeronNDArraySubscriber[] subscribers = new AeronNDArraySubscriber[numSubscribers];
        for(int i = 0; i < numSubscribers; i++) {
            AeronNDArraySubscriber subscriber = AeronNDArraySubscriber.builder()
                    .streamId(streamId)
                    .ctx(getContext()).channel(channel).aeron(aeron)
                    .running(running)
                    .ndArrayCallback(new NDArrayCallback() {
                        @Override
                        public void onNDArrayPartial(INDArray arr, long idx, int... dimensions) {

                        }

                        @Override
                        public void onNDArray(INDArray arr) {
                            running.set(false);
                        }
                    }).build();


            Thread t = new Thread(() -> {
                try {
                    subscriber.launch();
                } catch (Exception e) {
                    e.printStackTrace();
                }

            });<commit_msg>add higher ipc length<commit_after>package org.nd4j.aeron.ipc;

import io.aeron.Aeron;
import io.aeron.driver.MediaDriver;
import io.aeron.driver.ThreadingMode;
import org.agrona.BitUtil;
import org.agrona.CloseHelper;
import org.agrona.concurrent.BusySpinIdleStrategy;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.nd4j.linalg.api.ndarray.INDArray;
import org.nd4j.linalg.factory.Nd4j;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;

import static org.junit.Assert.assertFalse;

/**
 * Created by agibsonccc on 9/22/16.
 */
public class NdArrayIpcTest {
    private MediaDriver mediaDriver;
    private static Logger log = LoggerFactory.getLogger(NdArrayIpcTest.class);
    private Aeron.Context ctx;
    private String channel = "aeron:udp?endpoint=localhost:40123";
    private int streamId = 10;
    private  int length = (int) 1e7;

    @Before
    public void before() {
        //length of array * sizeof(float)
        int ipcLength =  length * 16;
        //padding for NDArrayMessage
        ipcLength += 64;
        //must be a power of 2
        ipcLength *= 2;
        //ipc length must be positive power of 2
        while(!BitUtil.isPowerOfTwo(ipcLength))
            ipcLength += 2;
        //Length in bytes for the SO_RCVBUF, 0 means use OS default. This needs to be larger than Receiver Window.
        System.setProperty("aeron.socket.so_rcvbuf",String.valueOf(ipcLength));

        final MediaDriver.Context ctx = new MediaDriver.Context()
                .threadingMode(ThreadingMode.DEDICATED)
                .dirsDeleteOnStart(true)
                   .ipcTermBufferLength(ipcLength)
                    .publicationTermBufferLength(ipcLength)
                    .maxTermBufferLength(ipcLength)
                .termBufferSparseFile(false)
                .conductorIdleStrategy(new BusySpinIdleStrategy())
                .receiverIdleStrategy(new BusySpinIdleStrategy())
                .senderIdleStrategy(new BusySpinIdleStrategy());
        mediaDriver = MediaDriver.launchEmbedded(ctx);
        System.out.println("Using media driver directory " + mediaDriver.aeronDirectoryName());
        System.out.println("Launched media driver");
    }

    @After
    public void after() {
        CloseHelper.quietClose(mediaDriver);
    }

    @Test
    public void testMultiThreadedIpcBig() throws Exception {
        int length = (int) 1e7;
        ExecutorService executorService = Executors.newFixedThreadPool(4);
        INDArray arr = Nd4j.ones((int) length);

        //length of array * sizeof(float)
        int ipcLength =  length * 4;
        //padding for NDArrayMessage
        ipcLength += 64;
        //must be a power of 2
        ipcLength *= 2;
        //ipc length must be positive power of 2
        while(!BitUtil.isPowerOfTwo(ipcLength))
            ipcLength += 2;
        //Length in bytes for the SO_RCVBUF, 0 means use OS default. This needs to be larger than Receiver Window.
        System.setProperty("aeron.socket.so_rcvbuf",String.valueOf(ipcLength));
        final AtomicBoolean running = new AtomicBoolean(true);
        Aeron aeron = Aeron.connect(getContext());
        int numSubscribers = 10;
        AeronNDArraySubscriber[] subscribers = new AeronNDArraySubscriber[numSubscribers];
        for(int i = 0; i < numSubscribers; i++) {
            AeronNDArraySubscriber subscriber = AeronNDArraySubscriber.builder()
                    .streamId(streamId)
                    .ctx(getContext()).channel(channel).aeron(aeron)
                    .running(running)
                    .ndArrayCallback(new NDArrayCallback() {
                        @Override
                        public void onNDArrayPartial(INDArray arr, long idx, int... dimensions) {

                        }

                        @Override
                        public void onNDArray(INDArray arr) {
                            running.set(false);
                        }
                    }).build();


            Thread t = new Thread(() -> {
                try {
                    subscriber.launch();
                } catch (Exception e) {
                    e.printStackTrace();
                }

            });
{'fraction_non_alphanumeric': 0.09194159004867496, 'fraction_numerical': 0.009329367225527312, 'mean_word_length': 2.1801375752364573, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, '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': '7393880', 'n_tokens_mistral': 2140, 'n_tokens_neox': 2001, 'n_words': 447}