/*
 * Copyright 2016-2022 chronicle.software
 *
 *       https://chronicle.software
 *
 * Licensed under Contract only
 */

package net.openhft.chronicle.wire.channel;

import net.openhft.chronicle.bytes.MethodReader;
import net.openhft.chronicle.wire.channel.echo.DummyData;
import net.openhft.chronicle.wire.channel.echo.EchoHandler;

import java.util.concurrent.atomic.AtomicLong;

public class PerfThroughputMain {
    static final int RUN_TIME = Integer.getInteger("runTime", 5);
    static final String URL = System.getProperty("url", "tcp://:1248");

    public static void main(String[] args) {
        try (ChronicleContext context = ChronicleContext.newContext(URL)) {
            EchoHandler echoHandler = new EchoHandler();
            final ChronicleChannelSupplier supplier = context.newChannelSupplier(echoHandler);
            ChronicleChannel channel = supplier.get();

            doTest("unbuffered", channel);

            echoHandler.buffered(true);
            ChronicleChannel channel2 = supplier.buffered(true).get();

            doTest("buffered", channel2);
        }
    }

    private static void doTest(String desc, ChronicleChannel channel) {
        DummyData data = new DummyData();
        for (int size = 32 << 10; size >= 8; size /= 2) {
            data.data(new byte[size - Long.BYTES]);
            long start = System.currentTimeMillis();
            long end = start + RUN_TIME * 1000L;
            final Echoing echoing = channel.methodWriter(Echoing.class);
            long count = 0;
            AtomicLong unread = new AtomicLong(0);
            MethodReader reader = channel.methodReader((Echoing) d -> {
                unread.getAndDecrement();
            });
            int window = (128 << 20) / size;
            do {
                echoing.echo(data);
                unread.getAndIncrement();
                count++;
                do {
                    reader.readOne();
                } while (unread.get() > window);
            } while (System.currentTimeMillis() < end);
            do {
                reader.readOne();
            } while (unread.get() > 0);
            long time = System.currentTimeMillis() - start;
            long totalBytes = size * count;
            long MBps = totalBytes / time / (1_000_000 / 1_000);
            long rate = count * 1000 / time;
            System.out.println("desc: " + desc + ", size: " + size + ", MBps: " + MBps+", mps: "+rate);
        }
    }
}

@@ -197,28 +197,203 @@
 e.bytes.
-MethodReader
+Bytes;%0Aimport net.openhft.chronicle.bytes.MethodReader;%0Aimport net.openhft.chronicle.core.Jvm;%0Aimport net.openhft.chronicle.wire.DocumentContext;%0Aimport net.openhft.chronicle.wire.WireOut
 ;%0Aimport
@@ -724,16 +724,82 @@
 :1248%22);
+%0A    static final boolean METHOD_RW = Jvm.getBoolean(%22method.rw%22);
 %0A%0A    pu
@@ -1810,24 +1810,102 @@
 micLong(0);%0A
+            int window = (128 %3C%3C 20) / size;%0A            if (METHOD_RW) %7B%0A    
             
@@ -1972,32 +1972,36 @@
                 
+    
 unread.getAndDec
@@ -2022,16 +2022,20 @@
         
+    
 %7D);%0A    
@@ -2046,94 +2046,796 @@
     
-int window = (128 %3C%3C 20) / size;%0A            do %7B%0A                echoing.echo(data);%0A
+    do %7B%0A                    echoing.echo(data);%0A                    unread.getAndIncrement();%0A                    count++;%0A                    do %7B%0A                        reader.readOne();%0A                    %7D while (unread.get() %3E window);%0A                %7D while (System.currentTimeMillis() %3C end);%0A%0A                do %7B%0A                    reader.readOne();%0A                %7D while (unread.get() %3E 0);%0A%0A            %7D else %7B%0A                InternalChronicleChannel icc = (InternalChronicleChannel) channel;%0A                do %7B%0A                    final Bytes%3C?%3E bytes = icc.acquireProducer().bytes();%0A                    bytes.writeInt(size);%0A                    for(int i=0;i%3Csize;i+=8)%0A                        bytes.writeLong(0L);%0A                    icc.releaseProducer();%0A%0A    
     
@@ -2880,32 +2880,36 @@
                 
+    
 count++;%0A       
@@ -2909,32 +2909,36 @@
                 
+  
+  
 do %7B%0A           
@@ -2946,34 +2946,210 @@
         
-reader.readOne();%0A
+    try (DocumentContext dc = channel.readingDocument())%7B%0A                            if (dc.isPresent())%0A                                unread.getAndDecrement();%0A                        %7D%0A    
         
@@ -3181,32 +3181,36 @@
 et() %3E window);%0A
+    
             %7D wh
@@ -3249,16 +3249,17 @@
 %3C end);%0A
+%0A
         
@@ -3254,32 +3254,36 @@
 );%0A%0A            
+    
 do %7B%0A           
@@ -3287,34 +3287,198 @@
         
-reader.readOne();%0A
+    try (DocumentContext dc = channel.readingDocument())%7B%0A                        if (dc.isPresent())%0A                            unread.getAndDecrement();%0A                    %7D%0A    
         
@@ -3505,24 +3505,38 @@
 get() %3E 0);%0A
+            %7D%0A
             

Add a single channel throughput test