Changeset 576 for Tests

Timestamp:

02/02/13 12:27:11 (9 years ago)

Author:

BegemoT

Message:

 

File:

• Tests/JAVA/test/src/main/java/test/threads/experiment2/Main.java (modified) (9 diffs)

Tests/JAVA/test/src/main/java/test/threads/experiment2/Main.java

@@ -10 +10 @@
 import com.lmax.disruptor.dsl.Disruptor;
 
+import static com.google.common.base.Preconditions.checkElementIndex;
 import static test.threads.queue.stressed.ITask.BenchmarkResult;
 
 /**
- * IDEA:
- * use common cyclic buffer with small capacity
- * in single-threaded case just go around,
- * in multithreaded use lzaySet cursors
- *
  * @author ruslan
  *         created 27.01.13 at 23:00
  */
 public class Main {
-
-
-        private static final int RUNS = 16;
-        private static final int TURNS = 64;
-
-        private static final int MESSAGES = 1 << 14;
-
-        private static final int REPOSITORY_SIZE = 1 << 15;
-        private static final int SCAN_STEP = 64;
+        private static final boolean PREDICABLE_SCAN = Boolean.getBoolean( "predictable" );
+
+        private static final int RUNS = Integer.getInteger( "runs", 16 );
+        private static final int TURNS = Integer.getInteger( "turns", 512 );
+
+        private static final int MESSAGES = 1 << 8;
+
+        private static final int REPOSITORY_SIZE = Integer.getInteger( "repository-size", 1 << 16 );
+        private static final int READ_PER_MESSAGE = 128;
+        private static final int READ_STEP = 1025;
+
         private static final ThreadAffinityService AFFINITY_SERVICE = ThreadAffinityUtils.defaultAffinityService();
         private static final CPULayoutService LAYOUT_SERVICE = ThreadAffinityUtils.defaultLayoutService();
 
         public static void main( final String[] args ) {
+                System.out.printf(
+                                "Repo: %d, reads: %d/msg, step: %d %s\n" +
+                                                "%d runs by %d turns by %d messages each\n",
+                                REPOSITORY_SIZE, READ_PER_MESSAGE, READ_STEP,
+                                PREDICABLE_SCAN ? "predictable" : "random",
+                                RUNS, TURNS, MESSAGES
+                );
 
 
@@ -43 +48 @@
                 Arrays.fill( repositoryB, 1.0 );
 
-                final TaskA a = new TaskA( repositoryA );
-                final TaskB b = new TaskB( repositoryB );
+
+                final Phase a = new Phase( repositoryA );
+                final Phase b = new Phase( repositoryB );
 
                 final Function<Message, Message> function = composeSync( a, b );
 
-                final BenchmarkResult[] results = benchmark( function, messages, TURNS, RUNS );
+                final BenchmarkResult[] results = benchmark(
+                                function,
+                                messages,
+                                TURNS,
+                                RUNS
+                );
 
                 for( final BenchmarkResult result : results ) {
@@ -63 +74 @@
                                            final int repositorySize ) {
                 final Message[] messages = new Message[size];
+                final int step = REPOSITORY_SIZE / size;
                 for( int i = 0; i < messages.length; i++ ) {
-                        messages[i] = new Message();
-                        messages[i].from = i % repositorySize;
-                        messages[i].to = ( size - 1 - i ) & repositorySize;
-                        messages[i].amount = i / 2 - size;
+                        final Message message = new Message();
+
+                        final int index = i * step % repositorySize;
+                        message.index = index;
+
+                        checkElementIndex( message.index, repositorySize );
+
+                        message.amount = i / 2 - size;
+
+                        messages[i] = message;
                 }
                 return messages;
         }
+
+        private static void putStressOnMemory( final double[] repository,
+                                               final int offset,
+                                               final double amount ) {
+                if( PREDICABLE_SCAN ) {
+                        for( int i = 0; i < READ_PER_MESSAGE; i++ ) {
+                                final int index = ( i * READ_STEP + offset ) % repository.length;
+                                repository[index] += amount;
+                        }
+                } else {
+                        int index = offset;
+                        for( int i = 0; i < READ_PER_MESSAGE; i++ ) {
+                                index = Math.abs( index * offset ) % repository.length;
+                                repository[index] += amount;
+                        }
+                }
+        }
+
 
         private static BenchmarkResult[] benchmark( final Function<Message, Message> function,
@@ -77 +113 @@
                                                     final int runs ) {
                 final BenchmarkResult[] results = new BenchmarkResult[runs];
-                for( int j = 0; j < runs; j++ ) {
+                for( int run = 0; run < runs; run++ ) {
                         final long startedAt = System.nanoTime();
-                        for( int i = 0; i < turns; i++ ) {
+                        for( int turn = 0; turn < turns; turn++ ) {
                                 benchmarkTurn( messages, function );
                         }
                         final long finishedAt = System.nanoTime();
 
-                        results[j] = new BenchmarkResult(
+                        results[run] = new BenchmarkResult(
                                         -1,
                                         finishedAt - startedAt,
@@ -93 +129 @@
         }
 
-
         private static void benchmarkTurn( final Message[] messages,
                                            final Function<Message, Message> f ) {
                 for( final Message message : messages ) {
+                        message.index++;
                         f.apply( message );
                 }
         }
 
-        private static Function<Message, Message> composeSync( final TaskA a,
-                                                               final TaskB b ) {
+
+        private static Function<Message, Message> composeSync( final Function<Message, Message> a,
+                                                               final Function<Message, Message> b ) {
                 return Functions.compose( b, a );
         }
@@ -118 +155 @@
                                 },
                                 Executors.newSingleThreadExecutor(),
-                                new SingleThreadedClaimStrategy( 32 ),
+                                new SingleThreadedClaimStrategy( 16 ),
                                 new BusySpinWaitStrategy()
                 );
@@ -137 +174 @@
         }
 
-
-        public static class TaskA implements Function<Message, Message> {
+        public static class Phase implements Function<Message, Message> {
+
                 private final double[] repository;
 
-                public TaskA( final double[] repository ) {
+                public Phase( final double[] repository ) {
                         this.repository = repository;
                 }
@@ -147 +184 @@
                 @Override
                 public Message apply( final Message message ) {
-                        final double amount = -message.amount;
-                        final int offset = message.to;
+                        final double amount = message.amount;
+                        final int offset = message.index;
                         putStressOnMemory( repository, offset, amount );
+                        //actually, message just pass through
                         return message;
                 }
 
-        }
-
-        public static class TaskB implements Function<Message, Message> {
-
-                private final double[] repository;
-
-                public TaskB( final double[] repository ) {
-                        this.repository = repository;
-                }
-
-                @Override
-                public Message apply( final Message message ) {
-                        final int offset = message.from;
-                        final double amount = message.amount;
-                        putStressOnMemory( repository, offset, amount );
-                        return message;
-                }
-
-        }
-
-        private static void putStressOnMemory( final double[] repository,
-                                               final int offset,
-                                               final double amount ) {
-                for( int i = 0; i < repository.length; i += SCAN_STEP ) {
-                        final int index = ( i + offset ) % repository.length;
-                        repository[index] += amount;
-                }
+
         }
 
         private static class Message {
-                public int from;
-                public int to;
+                public int index;
                 public double amount;
         }
@@ -198 +209 @@
                 public Message translateTo( final Message event,
                                             final long sequence ) {
-                        event.from = message.from;
-                        event.to = message.to;
+                        event.index = message.index;
                         event.amount = message.amount;
                         return event;