···

       
51
       
 
       
                try {

     

       
52
       
 
       
                    var buffer = channel.map(FileChannel.MapMode.READ_ONLY, pStart, pSize);

     

       
53
       
 
       
                    partitions[pIdx] = processBuffer(buffer, pIdx == 0);

     

       
54
       
-
       
                } catch (IOException e) {

     

       
0
       
 
       

     

       
55
       
 
       
                    throw new RuntimeException(e);

     

       
56
       
 
       
                }

     

       
57
       
 
       
            };

     
···

       
94
       
 
       
                    var t = target[j];

     

       
95
       
 
       
                    if (t == null) {

     

       
96
       
 
       
                        target[j] = current[j]; // copy ref from current to target

     

       
97
       
-
       
                    } else {

     

       
0
       
 
       

     

       
98
       
 
       
                        t.min = Math.min(t.min, current[j].min);

     

       
99
       
 
       
                        t.max = Math.max(t.max, current[j].max);

     

       
100
       
 
       
                        t.sum += current[j].sum;

     
···

       
154
       
 
       
                        st = stats[idx] = new Stats(key);

     

       
155
       
 
       
                    }

     

       
156
       
 
       
                    readingKey = false;

     

       
157
       
-
       
                } else {

     

       
0
       
 
       

     

       
158
       
 
       
                    keyHash = HASH_FACTOR * keyHash + b;

     

       
159
       
 
       
                }

     

       
160
       
-
       
            } else {

     

       
0
       
 
       

     

       
161
       
 
       
                if (b == '\n') {

     

       
162
       
 
       
                    var v = negative ? -val : val;

     

       
163
       
 
       
                    st.min = Math.min(st.min, v);

     
···

       
169
       
 
       
                    val = 0;

     

       
170
       
 
       
                    negative = false;

     

       
171
       
 
       
                    keyStart = buffer.position();

     

       
172
       
-
       
                } else if (b == '-') {

     

       
0
       
 
       

     

       
173
       
 
       
                    negative = true;

     

       
174
       
-
       
                } else if (b != '.') { // skip '.' since fractional tenth unit after decimal point is assumed

     

       
0
       
 
       

     

       
175
       
 
       
                    val = val * 10 + (b - '0');

     

       
176
       
 
       
                }

     

       
177
       
 
       
            }

     
···

       
187
       
 
       
    }

     

       
188
       
 
       


     

       
189
       
 
       
    private static byte[] readHeader(ByteBuffer buffer) { // read up to and including first newline (or end-of-input)

     

       
190
       
-
       
        while (buffer.hasRemaining() && buffer.get() != '\n') ;

     

       
0
       
 
       

     

       
191
       
 
       
        var header = new byte[buffer.position()];

     

       
192
       
 
       
        buffer.get(0, header, 0, header.length);

     

       
193
       
 
       
        return header;

     

       
194
       
 
       
    }

     

       
195
       
 
       


     

       
196
       
-
       
    record Partition(byte[] header, byte[] footer, Stats[] stats) {}

     

       
0
       
 
       

     

       
197
       
 
       


     

       
198
       
 
       
    private static class Stats { // min, max, and sum values are modeled with integral types that represent tenths of a unit

     

       
199
       
 
       
        final byte[] key;

     

···

       
51
       
 
       
                try {

     

       
52
       
 
       
                    var buffer = channel.map(FileChannel.MapMode.READ_ONLY, pStart, pSize);

     

       
53
       
 
       
                    partitions[pIdx] = processBuffer(buffer, pIdx == 0);

     

       
54
       
+
       
                }

     

       
55
       
+
       
                catch (IOException e) {

     

       
56
       
 
       
                    throw new RuntimeException(e);

     

       
57
       
 
       
                }

     

       
58
       
 
       
            };

     
···

       
95
       
 
       
                    var t = target[j];

     

       
96
       
 
       
                    if (t == null) {

     

       
97
       
 
       
                        target[j] = current[j]; // copy ref from current to target

     

       
98
       
+
       
                    }

     

       
99
       
+
       
                    else {

     

       
100
       
 
       
                        t.min = Math.min(t.min, current[j].min);

     

       
101
       
 
       
                        t.max = Math.max(t.max, current[j].max);

     

       
102
       
 
       
                        t.sum += current[j].sum;

     
···

       
156
       
 
       
                        st = stats[idx] = new Stats(key);

     

       
157
       
 
       
                    }

     

       
158
       
 
       
                    readingKey = false;

     

       
159
       
+
       
                }

     

       
160
       
+
       
                else {

     

       
161
       
 
       
                    keyHash = HASH_FACTOR * keyHash + b;

     

       
162
       
 
       
                }

     

       
163
       
+
       
            }

     

       
164
       
+
       
            else {

     

       
165
       
 
       
                if (b == '\n') {

     

       
166
       
 
       
                    var v = negative ? -val : val;

     

       
167
       
 
       
                    st.min = Math.min(st.min, v);

     
···

       
173
       
 
       
                    val = 0;

     

       
174
       
 
       
                    negative = false;

     

       
175
       
 
       
                    keyStart = buffer.position();

     

       
176
       
+
       
                }

     

       
177
       
+
       
                else if (b == '-') {

     

       
178
       
 
       
                    negative = true;

     

       
179
       
+
       
                }

     

       
180
       
+
       
                else if (b != '.') { // skip '.' since fractional tenth unit after decimal point is assumed

     

       
181
       
 
       
                    val = val * 10 + (b - '0');

     

       
182
       
 
       
                }

     

       
183
       
 
       
            }

     
···

       
193
       
 
       
    }

     

       
194
       
 
       


     

       
195
       
 
       
    private static byte[] readHeader(ByteBuffer buffer) { // read up to and including first newline (or end-of-input)

     

       
196
       
+
       
        while (buffer.hasRemaining() && buffer.get() != '\n')

     

       
197
       
+
       
            ;

     

       
198
       
 
       
        var header = new byte[buffer.position()];

     

       
199
       
 
       
        buffer.get(0, header, 0, header.length);

     

       
200
       
 
       
        return header;

     

       
201
       
 
       
    }

     

       
202
       
 
       


     

       
203
       
+
       
    record Partition(byte[] header, byte[] footer, Stats[] stats) {

     

       
204
       
+
       
    }

     

       
205
       
 
       


     

       
206
       
 
       
    private static class Stats { // min, max, and sum values are modeled with integral types that represent tenths of a unit

     

       
207
       
 
       
        final byte[] key;

     

···

       
26
       
 
       


     

       
27
       
 
       
    private static final String FILE = "./measurements.txt";

     

       
28
       
 
       


     

       
29
       
-
       


     

       
30
       
 
       
    private static class MeasurementAggregator {

     

       
31
       
 
       
        private double min = Double.POSITIVE_INFINITY;

     

       
32
       
 
       
        private double max = Double.NEGATIVE_INFINITY;

     
···

       
179
       
 
       
            chunkStartPtr = chunkEndPtr;

     

       
180
       
 
       
        }

     

       
181
       
 
       


     

       
182
       
-
       
        try(var executor = Executors.newFixedThreadPool(jobCnt)) {

     

       
183
       
 
       
            for (int i = 0; i < jobCnt; ++i) {

     

       
184
       
 
       
                executor.submit(chunks[i]);

     

       
185
       
 
       
            }

     

···

       
26
       
 
       


     

       
27
       
 
       
    private static final String FILE = "./measurements.txt";

     

       
28
       
 
       


     

       
0
       
 
       

     

       
29
       
 
       
    private static class MeasurementAggregator {

     

       
30
       
 
       
        private double min = Double.POSITIVE_INFINITY;

     

       
31
       
 
       
        private double max = Double.NEGATIVE_INFINITY;

     
···

       
178
       
 
       
            chunkStartPtr = chunkEndPtr;

     

       
179
       
 
       
        }

     

       
180
       
 
       


     

       
181
       
+
       
        try (var executor = Executors.newFixedThreadPool(jobCnt)) {

     

       
182
       
 
       
            for (int i = 0; i < jobCnt; ++i) {

     

       
183
       
 
       
                executor.submit(chunks[i]);

     

       
184
       
 
       
            }

     

···

       
105
       
 
       
    }

     

       
106
       
 
       


     

       
107
       
 
       
    private static Map<String, Measurement> combineMaps(

     

       
108
       
-
       
            final Map<String, Measurement> map1,

     

       
109
       
-
       
            final Map<String, Measurement> map2

     

       
110
       
-
       
    ) {

     

       
111
       
 
       
        for (final var entry : map2.entrySet()) {

     

       
112
       
 
       
            map1.merge(entry.getKey(), entry.getValue(), Measurement::combineWith);

     

       
113
       
 
       
        }

     
···

       
116
       
 
       
    }

     

       
117
       
 
       


     

       
118
       
 
       
    public static void main(final String[] args) throws IOException {

     

       
119
       
-
       
        //long before = System.currentTimeMillis();

     

       
120
       
 
       
        /**

     

       
121
       
 
       
         * Shoutout to bjhara

     

       
122
       
 
       
         */

     
···

       
136
       
 
       
                    final MappedByteBuffer mbb = in.map(

     

       
137
       
 
       
                            FileChannel.MapMode.READ_ONLY,

     

       
138
       
 
       
                            start,

     

       
139
       
-
       
                            Math.min(CHUNK_SIZE, total - start)

     

       
140
       
-
       
                    );

     

       
141
       
 
       
                    int realEnd = mbb.limit() - 1;

     

       
142
       
 
       
                    while (mbb.get(realEnd) != '\n') {

     

       
143
       
 
       
                        realEnd--;

     
···

       
148
       
 
       
                    start += realEnd;

     

       
149
       
 
       


     

       
150
       
 
       
                    return mbb;

     

       
151
       
-
       
                } catch (final IOException e) {

     

       
0
       
 
       

     

       
152
       
 
       
                    throw new RuntimeException(e);

     

       
153
       
 
       
                }

     

       
154
       
 
       
            }

     

       
155
       
 
       
        };

     

       
156
       
 
       
        final Map<String, Measurement> reduce = StreamSupport.stream(Spliterators.spliteratorUnknownSize(

     

       
157
       
-
       
                                                               iterator, Spliterator.IMMUTABLE), true)

     

       
158
       
-
       
                                                             .parallel()

     

       
159
       
-
       
                                                             .map(CalculateAverage_truelive::parseBuffer)

     

       
160
       
-
       
                                                             .reduce(CalculateAverage_truelive::combineMaps).get();

     

       
161
       
 
       


     

       
162
       
 
       
        System.out.print("{");

     

       
163
       
 
       
        System.out.print(

     
···

       
169
       
 
       
                        .collect(Collectors.joining(", ")));

     

       
170
       
 
       
        System.out.println("}");

     

       
171
       
 
       


     

       
172
       
-
       
        //System.out.println("Took: " + (System.currentTimeMillis() - before));

     

       
173
       
 
       


     

       
174
       
 
       
    }

     

       
175
       
 
       


     
···

       
189
       
 
       
                name = new String(arr, 0, len);

     

       
190
       
 
       
                bug.position(pos);

     

       
191
       
 
       
                bug.mark();

     

       
192
       
-
       
            } else if (c == '\n') {

     

       
0
       
 
       

     

       
193
       
 
       
                final int pos = bug.position();

     

       
194
       
 
       
                bug.reset();

     

       
195
       
 
       
                final int len = pos - bug.position();

     

···

       
105
       
 
       
    }

     

       
106
       
 
       


     

       
107
       
 
       
    private static Map<String, Measurement> combineMaps(

     

       
108
       
+
       
                                                        final Map<String, Measurement> map1,

     

       
109
       
+
       
                                                        final Map<String, Measurement> map2) {

     

       
0
       
 
       

     

       
110
       
 
       
        for (final var entry : map2.entrySet()) {

     

       
111
       
 
       
            map1.merge(entry.getKey(), entry.getValue(), Measurement::combineWith);

     

       
112
       
 
       
        }

     
···

       
115
       
 
       
    }

     

       
116
       
 
       


     

       
117
       
 
       
    public static void main(final String[] args) throws IOException {

     

       
118
       
+
       
        // long before = System.currentTimeMillis();

     

       
119
       
 
       
        /**

     

       
120
       
 
       
         * Shoutout to bjhara

     

       
121
       
 
       
         */

     
···

       
135
       
 
       
                    final MappedByteBuffer mbb = in.map(

     

       
136
       
 
       
                            FileChannel.MapMode.READ_ONLY,

     

       
137
       
 
       
                            start,

     

       
138
       
+
       
                            Math.min(CHUNK_SIZE, total - start));

     

       
0
       
 
       

     

       
139
       
 
       
                    int realEnd = mbb.limit() - 1;

     

       
140
       
 
       
                    while (mbb.get(realEnd) != '\n') {

     

       
141
       
 
       
                        realEnd--;

     
···

       
146
       
 
       
                    start += realEnd;

     

       
147
       
 
       


     

       
148
       
 
       
                    return mbb;

     

       
149
       
+
       
                }

     

       
150
       
+
       
                catch (final IOException e) {

     

       
151
       
 
       
                    throw new RuntimeException(e);

     

       
152
       
 
       
                }

     

       
153
       
 
       
            }

     

       
154
       
 
       
        };

     

       
155
       
 
       
        final Map<String, Measurement> reduce = StreamSupport.stream(Spliterators.spliteratorUnknownSize(

     

       
156
       
+
       
                iterator, Spliterator.IMMUTABLE), true)

     

       
157
       
+
       
                .parallel()

     

       
158
       
+
       
                .map(CalculateAverage_truelive::parseBuffer)

     

       
159
       
+
       
                .reduce(CalculateAverage_truelive::combineMaps).get();

     

       
160
       
 
       


     

       
161
       
 
       
        System.out.print("{");

     

       
162
       
 
       
        System.out.print(

     
···

       
168
       
 
       
                        .collect(Collectors.joining(", ")));

     

       
169
       
 
       
        System.out.println("}");

     

       
170
       
 
       


     

       
171
       
+
       
        // System.out.println("Took: " + (System.currentTimeMillis() - before));

     

       
172
       
 
       


     

       
173
       
 
       
    }

     

       
174
       
 
       


     
···

       
188
       
 
       
                name = new String(arr, 0, len);

     

       
189
       
 
       
                bug.position(pos);

     

       
190
       
 
       
                bug.mark();

     

       
191
       
+
       
            }

     

       
192
       
+
       
            else if (c == '\n') {

     

       
193
       
 
       
                final int pos = bug.position();

     

       
194
       
 
       
                bug.reset();

     

       
195
       
 
       
                final int len = pos - bug.position();

     

···

       
35
       
 
       
    private static final String FILE = "./measurements2.txt";

     

       
36
       
 
       


     

       
37
       
 
       
    static class WeatherStation {

     

       
38
       
-
       
        final static char[] NUMBERS = {'0', '1', '2', '3', '4', '5', '6', '7', '8', '9' };

     

       
39
       
 
       


     

       
40
       
 
       
        final String id;

     

       
41
       
 
       
        final int meanTemperature;

     
···

       
58
       
 
       
         * @param buffer the buffer to append to

     

       
59
       
 
       
         */

     

       
60
       
 
       
        void measurement(final CheaperCharBuffer buffer) {

     

       
61
       
-
       
        	// fake -10.9 to +10.9 variance without double operations and rounding

     

       
62
       
-
       
        	// gives us -10 to +10

     

       
63
       
 
       
            int m = meanTemperature + (r.nextInt(21) - 10);

     

       
64
       
 
       
            // gives us a decimal digit 0 to 9 as char

     

       
65
       
 
       
            char d = NUMBERS[r.nextInt(10)];

     
···

       
569
       
 
       


     

       
570
       
 
       
        // manual loop unroll for less jumps

     

       
571
       
 
       
        for (int i = 0; i < count; i = i + 8) {

     

       
572
       
-
       
        	{

     

       
573
       
-
       
        		// try to fill teh cpu pipeline as much as possible with

     

       
574
       
-
       
        		// independent operations

     

       
575
       
-
       
        		int s1 = r1.nextInt(stationCount);

     

       
576
       
-
       
        		int s2 = r2.nextInt(stationCount);

     

       
577
       
-
       
        		int s3 = r3.nextInt(stationCount);

     

       
578
       
-
       
        		int s4 = r4.nextInt(stationCount);

     

       
579
       
-
       
        		// get us the ojects one after the other to have the array

     

       
580
       
-
       
        		// in our L1 cache and not push it out with other data

     

       
581
       
-
       
        		var w1 = stations.get(s1);

     

       
582
       
-
       
        		var w2 = stations.get(s2);

     

       
583
       
-
       
        		var w3 = stations.get(s3);

     

       
584
       
-
       
        		var w4 = stations.get(s4);

     

       
585
       
-
       
        		// write our data to our buffer

     

       
586
       
-
       
        		w1.measurement(sb);

     

       
587
       
-
       
        		w2.measurement(sb);

     

       
588
       
-
       
        		w3.measurement(sb);

     

       
589
       
-
       
        		w4.measurement(sb);

     

       
590
       
-
       
        	}

     

       
591
       
-
       
        	{

     

       
592
       
-
       
        		int s1 = r1.nextInt(stationCount);

     

       
593
       
-
       
        		int s2 = r2.nextInt(stationCount);

     

       
594
       
-
       
        		int s3 = r3.nextInt(stationCount);

     

       
595
       
-
       
        		int s4 = r4.nextInt(stationCount);

     

       
596
       
-
       
        		var w1 = stations.get(s1);

     

       
597
       
-
       
        		var w2 = stations.get(s2);

     

       
598
       
-
       
        		var w3 = stations.get(s3);

     

       
599
       
-
       
        		var w4 = stations.get(s4);

     

       
600
       
-
       
        		w1.measurement(sb);

     

       
601
       
-
       
        		w2.measurement(sb);

     

       
602
       
-
       
        		w3.measurement(sb);

     

       
603
       
-
       
        		w4.measurement(sb);

     

       
604
       
-
       
        	}

     

       
605
       
 
       
            // write the buffer directly, no intermediate string copy

     

       
606
       
 
       
            bw.write(sb.data_, 0, sb.length_);

     

       
607
       
 
       


     

···

       
35
       
 
       
    private static final String FILE = "./measurements2.txt";

     

       
36
       
 
       


     

       
37
       
 
       
    static class WeatherStation {

     

       
38
       
+
       
        final static char[] NUMBERS = { '0', '1', '2', '3', '4', '5', '6', '7', '8', '9' };

     

       
39
       
 
       


     

       
40
       
 
       
        final String id;

     

       
41
       
 
       
        final int meanTemperature;

     
···

       
58
       
 
       
         * @param buffer the buffer to append to

     

       
59
       
 
       
         */

     

       
60
       
 
       
        void measurement(final CheaperCharBuffer buffer) {

     

       
61
       
+
       
            // fake -10.9 to +10.9 variance without double operations and rounding

     

       
62
       
+
       
            // gives us -10 to +10

     

       
63
       
 
       
            int m = meanTemperature + (r.nextInt(21) - 10);

     

       
64
       
 
       
            // gives us a decimal digit 0 to 9 as char

     

       
65
       
 
       
            char d = NUMBERS[r.nextInt(10)];

     
···

       
569
       
 
       


     

       
570
       
 
       
        // manual loop unroll for less jumps

     

       
571
       
 
       
        for (int i = 0; i < count; i = i + 8) {

     

       
572
       
+
       
            {

     

       
573
       
+
       
                // try to fill teh cpu pipeline as much as possible with

     

       
574
       
+
       
                // independent operations

     

       
575
       
+
       
                int s1 = r1.nextInt(stationCount);

     

       
576
       
+
       
                int s2 = r2.nextInt(stationCount);

     

       
577
       
+
       
                int s3 = r3.nextInt(stationCount);

     

       
578
       
+
       
                int s4 = r4.nextInt(stationCount);

     

       
579
       
+
       
                // get us the ojects one after the other to have the array

     

       
580
       
+
       
                // in our L1 cache and not push it out with other data

     

       
581
       
+
       
                var w1 = stations.get(s1);

     

       
582
       
+
       
                var w2 = stations.get(s2);

     

       
583
       
+
       
                var w3 = stations.get(s3);

     

       
584
       
+
       
                var w4 = stations.get(s4);

     

       
585
       
+
       
                // write our data to our buffer

     

       
586
       
+
       
                w1.measurement(sb);

     

       
587
       
+
       
                w2.measurement(sb);

     

       
588
       
+
       
                w3.measurement(sb);

     

       
589
       
+
       
                w4.measurement(sb);

     

       
590
       
+
       
            }

     

       
591
       
+
       
            {

     

       
592
       
+
       
                int s1 = r1.nextInt(stationCount);

     

       
593
       
+
       
                int s2 = r2.nextInt(stationCount);

     

       
594
       
+
       
                int s3 = r3.nextInt(stationCount);

     

       
595
       
+
       
                int s4 = r4.nextInt(stationCount);

     

       
596
       
+
       
                var w1 = stations.get(s1);

     

       
597
       
+
       
                var w2 = stations.get(s2);

     

       
598
       
+
       
                var w3 = stations.get(s3);

     

       
599
       
+
       
                var w4 = stations.get(s4);

     

       
600
       
+
       
                w1.measurement(sb);

     

       
601
       
+
       
                w2.measurement(sb);

     

       
602
       
+
       
                w3.measurement(sb);

     

       
603
       
+
       
                w4.measurement(sb);

     

       
604
       
+
       
            }

     

       
605
       
 
       
            // write the buffer directly, no intermediate string copy

     

       
606
       
 
       
            bw.write(sb.data_, 0, sb.length_);

     

       
607