private static record Measurement(String station, double value) {
    private Measurement(String[] parts) {
        this(parts[0], Double.parseDouble(parts[1]));
    }
}
        
Collector<Measurement, MeasurementAggregator, ResultRow> collector = Collector.of(
        MeasurementAggregator::new,
        (a, m) -> {
            a.min = Math.min(a.min, m.value);
            a.max = Math.max(a.max, m.value);
            a.sum += m.value;
            a.count++;
        },
        (agg1, agg2) -> {
            var res = new MeasurementAggregator();
            res.min = Math.min(agg1.min, agg2.min);
            res.max = Math.max(agg1.max, agg2.max);
            res.sum = agg1.sum + agg2.sum;
            res.count = agg1.count + agg2.count;

            return res;
        },
        agg -> {
            return new ResultRow(agg.count, agg.sum, agg.min, agg.max);
        });

Map<String, ResultRow> measurements = 
        Files.lines(Paths.get(fileName))            
        .map(l -> l.split(";"))
        .map(a -> new Measurement(a))
        .collect(Collectors.groupingBy(m -> m.station(), collector));

return new TreeMap<>(measurements).toString();
    

org.rschwietzke.parallel.BRC001_Baseline
github.com/rschwietzke/1brc-the-first-80-meters.git

The Results

Numbers between slides might vary.
github.com/rschwietzke/1brc-the-first-80-meters.git

Map<String, ResultRow> measurements = 
        Files.lines(Paths.get(fileName)).parallel()
        .map(l -> l.split(";"))
        .map(a -> new Measurement(a))
        .collect(
                Collectors.groupingBy(m -> m.station(), collector));

Map<String, ResultRow> measurements = 
        Files.lines(Paths.get(fileName))
        .map(l -> l.split(";"))
        .map(a -> new Measurement(a)).parallel()
        .collect(
                Collectors.groupingBy(m -> m.station(), collector));

Map<String, ResultRow> measurements = 
        Files.lines(Paths.get(fileName)).parallel()
        .map(l -> l.split(";")).parallel()
        .map(a -> new Measurement(a)).parallel()
        .collect(
                Collectors.groupingBy(m -> m.station(), collector));

...                
        .collect(
                Collectors.groupingByConcurrent(m -> m.station(), collector));

Map<String, ResultRow> measurements = 
        Files.lines(Paths.get(fileName)).parallel()
        .map(l -> new Measurement(l.split(";")))
        .collect(
                Collectors.groupingByConcurrent(m -> m.station(), collector));                
    

org.rschwietzke.parallel.BRC010* to BRC016*

Results

// BRC014_ParallelGrouping
Map<String, ResultRow> measurements = 
        Files.lines(Paths.get(fileName)).parallel()
        .map(l -> l.split(";")).parallel()
        .map(a -> new Measurement(a))
        .collect(
                Collectors.groupingByConcurrent(
                    m -> m.station(), collector));


// BRC016_Compact                
Map<String, ResultRow> measurements = 
    Files.lines(Paths.get(fileName)).parallel()
    .map(l -> new Measurement(l.split(";")))
    .collect(
            Collectors.groupingByConcurrent(
                m -> m.station(), collector));
    

@Override
public String run(final String fileName) throws IOException {
    // our storage
    final Map<String, Temperatures> cities = new HashMap<>();

    // open the file
    try (var reader = Files.newBufferedReader(Paths.get(fileName))) {
        String line;
        
        // read all lines until end of file
        while ((line = reader.readLine()) != null) {
            // split the line
            final String[] cols = line.split(";");

            // first is the city
            final String city = cols[0];
            // second our double temperature
            final double temperature = Double.parseDouble(cols[1]);

            // get us our measurement record
            var measurement = new Temperatures(temperature);
            
            // store it, when it exists, merge both measurements otherwise
            cities.merge(city, measurement, (t1, t2) -> t1.merge(t2));
        }
    }

    // ok, we got everything, now we need to order it
    return new TreeMap<String, Temperatures>(cities).toString();
}
    

org.rschwietzke.parallel.BRC020_NoStream

The Results

public String run(final String fileName)
{
    // let's do a raw and basic old-school implementation without
    // most modern Java help
    BlockingQueue<String> splittingQueueInput = new LinkedBlockingDeque<>(100);
    BlockingQueue<String[]> measureQueueInput = new LinkedBlockingDeque<>(100);
    BlockingQueue<Temperatures> mappingQueueInput = new LinkedBlockingDeque<>(100);

    int consumerCount = Math.max(1, (this.getThreadCount() - 2) / 2);

    var readerThread = new ReaderThread(Paths.get(fileName), splittingQueueInput, consumerCount);
    readerThread.start();

    for (int i = 0; i < consumerCount; i++)
    {
        var sThread = new SplittingThread(splittingQueueInput, measureQueueInput);
        sThread.start();
        
        var mThread = new MeasurementThread(measureQueueInput, mappingQueueInput);
        mThread.start();
    }

    var mapperThread = new MapperThread(mappingQueueInput, consumerCount);
    mapperThread.start();

    // wait till done
    try 
    {
        mapperThread.join();
    } 
    catch (InterruptedException e) 
    {
        throw new RuntimeException(e);
    }

    // get our result
    var cities = mapperThread.getCities();

    // ok, we got everything, now we need to order it
    return new TreeMap<String, Temperatures>(cities).toString();
}        
    

org.rschwietzke.parallel.BRC025_1_N_N_1

Results

Results

Results

Results

Performance counter stats for 'java -Xms2g -Xmx2g -cp target/classes/ org.rschwietzke.parallel.BRC001_Baseline -f data-1000m.txt -wc 0 -mc 1 -t 8':

         130,376.50 msec task-clock                       #    1.020 CPUs utilized             
             18,943      context-switches                 #  145.295 /sec                      
              4,788      cpu-migrations                   #   36.724 /sec                      
            326,998      page-faults                      #    2.508 K/sec                     
👉  464,798,376,203      cycles                           #    3.565 GHz                       
     54,941,360,494      stalled-cycles-frontend          #   11.82% frontend cycles idle      
👉2,380,905,128,702      instructions                     #    5.12  insn per cycle            
                                                  #    0.02  stalled cycles per insn   
    549,265,356,816      branches                         #    4.213 G/sec                     
      2,342,277,605      branch-misses                    #    0.43% of all branches           

    ⏱127.773261850 seconds time elapsed

      124.145161000 seconds user
        5.642008000 seconds sys


Performance counter stats for 'java -Xms2g -Xmx2g -cp target/classes/ org.rschwietzke.parallel.BRC044_ForkJoinPool_SelfSplitting -f data-1000m.txt -wc 0 -mc 1 -t 8':

         195,689.07 msec task-clock                       #    7.527 CPUs utilized             
             26,837      context-switches                 #  137.141 /sec                      
              4,385      cpu-migrations                   #   22.408 /sec                      
            333,597      page-faults                      #    1.705 K/sec                     
👉  694,817,257,247      cycles                           #    3.551 GHz                       
    124,646,585,103      stalled-cycles-frontend          #   17.94% frontend cycles idle      
👉2,160,182,371,314      instructions                     #    3.11  insn per cycle            
                                                  #    0.06  stalled cycles per insn   
    495,875,578,633      branches                         #    2.534 G/sec                     
      2,251,135,482      branch-misses                    #    0.45% of all branches           

    ⏱ 25.997951402 seconds time elapsed

      189.993214000 seconds user
        4.836329000 seconds sys
 
    

org.rschwietzke.parallel.BRC0

Looking into CPU Time

Asking a Profiler
🔳🕒🕵️

CPU - Async-profiler to the rescue

Checking CPU time of BRC048_VT_ChunkSize_10MB

Source

Allocations - Async-profiler to the rescue

Checking allocations of BRC048_VT_ChunkSize_10MB

Source

Split is Slow

String::split does a lot of work

while ((line = r.readln()) != null) {
    var splitData = line.split(";");

    // first is the city
    var city = splitData[0];

    // second our double temperature
    var temperature = Double.parseDouble(splitData[1]);

    // get us our measurement record
    var measurement = new Temperatures(city, temperature);                

    cities.merge(city, measurement, (t1, t2) -> t1.merge(t2));
}
    

while ((line = r.readln()) != null) {
    // split the line using indexOf
    final int semicolon = line.indexOf(';');
    final String city = line.substring(0, semicolon);
    final String temperatureString = line.substring(semicolon + 1);

    // second our double temperature
    var temperature = Double.parseDouble(temperatureString);

    // get us our measurement record
    var measurement = new Temperatures(city, temperature);                

    cities.merge(city, measurement, (t1, t2) -> t1.merge(t2));
}
    

Results

Double::parseDouble

String to double Parsing Is Heavy Too

private static final int DIGITOFFSET = 48;

private static final double[] multipliers = {
    1, 1, 0.1, 0.01, 0.001, 0.000_1, 0.000_01, 0.000_001, 0.000_000_1, 0.000_000_01,
    0.000_000_001, 0.000_000_000_1, 0.000_000_000_01, 0.000_000_000_001, 0.000_000_000_000_1,
    0.000_000_000_000_01, 0.000_000_000_000_001, 0.000_000_000_000_000_1, 0.000_000_000_000_000_01};

public static double parseDouble(final String s)
{
    final int negative = s.charAt(0) == '-' ? 1 : 0;

    long value = 0;
    final int end = s.length();
    int decimalPos = end;

    for (int i = negative; i < end; i++)
    {
        final int d = s.charAt(i);
        if (d == '.')
        {
            decimalPos = i;
            continue;
        }
        final int v = d - DIGITOFFSET;
        value = value * 10 + v;
    }

    // set the sign
    value = negative == 1 ? -value : value;

    // adjust the decimal places
    return value * multipliers[end - decimalPos];
}

    

Results

Get rid of Garbage

We Don't Need a Fresh String for Parsing

while ((line = r.readln()) != null)
{
    // split the line using indexOf
    final int semicolon = line.indexOf(';');
    final String city = line.substring(0, semicolon);
    final String temperatureString = line.substring(semicolon + 1);

    // second our double temperature
    var temperature = parseDouble(temperatureString);

    // get us our measurement record
    var measurement = new Temperatures(city, temperature);                

    cities.merge(city, measurement, (t1, t2) -> t1.merge(t2));
}
    

while ((line = r.readln()) != null)
{
    // split the line using indexOf
    final int semicolon = line.indexOf(';');
    final String city = line.substring(0, semicolon);

    // second our double temperature
    final double temperature = parseDouble(line, 
        semicolon + 1, line.length() - 1);

    // get us our measurement record
    var measurement = new Temperatures(city, temperature);                

    cities.merge(city, measurement, (t1, t2) -> t1.merge(t2));
}
    

Results

Simplify the Data

Who Needs double Anyway?

private static final int DIGITOFFSET = 48;

/**
    * Parses a double but ends up with an int, only because we know
    * the format of the results -99.9 to 99.9
    */
public static int parseInteger(final String s, final int offset, final int end)
{
    final int negative = s.charAt(offset) == '-' ? offset + 1 : offset;

    int value = 0;

    for (int i = negative; i < end; i++)
    {
        final int d = s.charAt(i);
        if (d == '.')
        {
            continue;
        }
        final int v = d - DIGITOFFSET;
        value = value * 10 + v;
    }

    value = negative != offset ? -value : value;
    return value;
}
    

Results

Making it Nicer for the Hardware

private static final int DIGITOFFSET = 48;

/**
    * Parses a double but ends up with an int, only because we know
    * the format of the results -99.9 to 99.9
    * 
    * Best case, two branches
    * Worst case, two branches
    */
public static int parseInteger(final String s, int offset, final int end)
{
    int l = end - offset;
    var p0 = s.charAt(end - 1);
    var p2 = s.charAt(end - 3) * 10;
    var value = p2 + p0 - (DIGITOFFSET * 10 + DIGITOFFSET);
    
    final char firstChar = s.charAt(offset);
    if (firstChar == '-') {
        if (l == 5) {
            // -99.9
            var p3 = s.charAt(end - 4) * 100;
            value = p3 + value - (DIGITOFFSET * 100);
        }
        else {
            // -9.9
        }
        return -value;
    }
    else {
        if (l == 4) {
            // 99.9
            // we can use firstChar directly as we know its not '-'
            var p3 = firstChar * 100;
            value = p3 + value - (DIGITOFFSET * 100);
        }
        else {
            // 9.9
        }
        return value;
    }
}
    

Results

The Best Memory Is No Memory

Immutability Is Good But Mutations Are Better

/**
    * Holds our temperature data without the station, because the
    * map already knows that
    */
private static class Temperatures
{
    private int min;
    private int max;
    private long total;
    private long count;

    public Temperatures(final int temperature)
    {
        this.min = temperature;
        this.max = temperature;
        this.total = temperature;
        this.count = 1;
    }

    public Temperatures merge(final int temperature)
    {
        this.min = Math.min(this.min, temperature);
        this.max = Math.max(this.max, temperature);
        this.total += temperature;
        this.count += 1;
        return this;
    }    
    ...
}

Results

Why Strings?

private static class City {
    private int min;
    private int max;
    private long total;
    private long count;
    private final byte[] name;
    private final int hashCode;
    
    public City(byte[] bytes, int semicolon, int temperature, int hashCode) {
        this.name = Arrays.copyOfRange(bytes, 0, semicolon);
        this.max = temperature;
        this.min = temperature;
        this.total = temperature;
        this.count = 1;
        this.hashCode = hashCode;
    }

    public City update(int temperature) {
        this.min = Math.min(this.min, temperature);
        this.max = Math.max(this.max, temperature);
        count++;
        this.total += temperature;
        
        return this;
    }
    
    public City merge(final City city) {
        this.min = Math.min(this.min, city.min);
        this.max = Math.max(this.max, city.max);
        this.total += city.total;
        this.count += city.count;
        
        return this;
    }
...
} 

Results

Reduce ForkJoinPool Logic

protected Map<City, City> compute() {
    // split only when we should
    if (taskCount > 1)
    {
        final List<Mapper> tasks = new ArrayList<>(taskCount);
        
        final long size = this.to - this.from;
        final long chunkSize = size / taskCount;

        long from = -chunkSize;
        long to = 0;
        while (to < size) {
            from += chunkSize;
            to = from + chunkSize;
            
            // when close to the end, make one chunk larger 
            // than really small... otherwise we lose data
            to = (size - to) < chunkSize ? size : to;

            tasks.add(new Mapper(filePath, from, to, 1));
        }
        ForkJoinTask.invokeAll(tasks);

        // reduce result
        final Map<City, City> cities = new HashMap<>();

        for (var t : tasks) {
            try {
                final var result = t.get();
                
                result.forEach((k, v) -> 
                    cities.compute(k, (_, v2) -> v2 == null ? v : v2.merge(v)));
            } 
            catch (InterruptedException e) {
                throw new RuntimeException(e);
            } 
            catch (ExecutionException e) {
                throw new RuntimeException(e);
            }
        }

        return cities;
    }
    else {
        return map();
    }
}
    

Results