1 биллион челлендж

Можно и не нормализовывать, но вот чтоб у тебя буквы шли в нужном порядке - придётся парсить всё и использовать многомегабайтные таблицы юникодных спецификаций. Юникод сам по себе неконсистентный и о свойствах конкретного code pointa (он может быть вообще не буквой а модификатором к следующей или предыдущей букве около него) без таблиц не догадаться.

Да, мне вот тоже интересно, например, как отсортировать эмоджи, какие-нибудь геометрические фигуры или вовсе кодепоинты из private use area «по алфавиту». Получается, что это невозможно.

UTF-8 - это вообще просто кодирвка Юникода. И пока набор символов и «алфавитный порядок» не определены, конкурсная задача является не полной.

Ну вот, у них даже в тестах алфавитный порядок не соблюдается, строчные буквы идут после прописных:

https://github.com/gunnarmorling/1brc/blob/main/src/test/resources/samples/measurements-complex-utf8.out

B=8.9/8.9/8.9
C=38.9/38.9/38.9
...
aniCartagoEṭ Ṭīra...
burgazAl Ḩawīyah...

И что? Это не та сортировка которую ожидают. В юникоде буквы много где не по порядку идут, и, хуже того, где-то вообще понятие порядка отсутствует.

С этим можно разобраться в конце. Идея в том что тратить на это precious CPU cycles во время чтения основного объема - смысла не много.

Надо будет допилить да off-by-one ошибки повыловить

Fukk yeah

;; using multiple threads to read the file (divided to logical segments) and
;; compute stats for each segment, then combining the results

(def ^:dynamic *end-of-file*
  "Integer value returned by a file input stream when end of file is reached."  -1)

(def ^:dynamic *end-of-line*
  "Integer value that corresponds to the end of line character." 10)

(defn file-segment
  "Returns a logical segment of `file`, defined by start and end
  positions in bytes. The segment will begin at `start` and end after
  newline nearest to `start` + `initial-length` position (inclusive),
  i.e., segment always ends between lines (or at the end of file);
  `start` defaults to 0; `initial-length` defaults to the length of
  the entire file."
  ([^File file]
   (file-segment file 0))
  ([^File file start]
   (file-segment file start (.length file)))
  ([^File file start initial-length]
   (when (pos? initial-length)
     (with-open [file-input-stream (FileInputStream. file)]
       (let [file-channel         (.getChannel file-input-stream)
             end-of-line-or-file? #{*end-of-file* *end-of-line*}
             at-initial-end       (+ start initial-length)
             before-initial-end   (dec at-initial-end)]
         {:file  file
          :start start
          :end   (do
                   (.position file-channel ^int before-initial-end)
                   (let [initial-end-byte (.read file-input-stream)]
                     (if (end-of-line-or-file? initial-end-byte)
                       (min (.position file-channel) (.length file))
                       (do
                         (.position file-channel ^int at-initial-end)
                         (while (not (end-of-line-or-file? (.read file-input-stream))))
                         (min (.position file-channel) (.length file))))))})))))

(defn file-segment-seq
  "Returns a lazy sequence of at most `n` successive logical segments of
  `file`. (As segments may be expanded to align with line boundaries,
  there may be fewer of them, especially if segments are small and
  lines are long.)"
  ([^java.io.File file n]
   (file-segment-seq file 0 n))
  ([^java.io.File file start n]
   (when (> n 0)
     (let [remaining-file-length  (- (.length file) start)
           initial-segment-length (int (/ remaining-file-length n))]
       (when-let [segment (file-segment file start initial-segment-length)]
         (cons segment (lazy-seq (file-segment-seq file (:end segment) (dec n)))))))))

(defn file-segment-line-seq
  "Returns a lazy sequence of lines from the `segment`. Reading begins
  at `start` and ends after `end`. (The line that contains the end
  position will be the last read. Segment constructor guarantees that
  segment will always end at a line boundary, but manually created
  segments may not.)"
  ([{:keys [file start end]}]
   (let [file-input-stream (FileInputStream. ^File file)
         _                 (.position (.getChannel file-input-stream) ^int start)
         reader            (io/reader file-input-stream)]
     (file-segment-line-seq reader 0 (- end start))))
  ([^BufferedReader reader bytes-read bytes-max]
   (when (not (>= bytes-read bytes-max))
     (when-let [line (.readLine reader)]
       (let [line-length-bytes (inc (count (.getBytes line)))]
         (cons line (lazy-seq (file-segment-line-seq reader
                                                     (+ bytes-read line-length-bytes)
                                                     bytes-max))))))))

(defn merge-station-stats
  "Returns two station stats maps merged."
  [{cnt1 :count min1 :minimum max1 :maximum avg1 :average}
   {cnt2 :count min2 :minimum max2 :maximum avg2 :average}]
  {:count   (+ cnt1 cnt2)
   :minimum (min min1 min2)
   :maximum (max max1 max2)
   :average (/ (+ (* cnt1 avg1) (* cnt2 avg2)) (+ cnt1 cnt2))})

(defn parallel-stats
  "Returns the map of station stats keyed by station names. Reads the
  measurements data from `filename`, that is a text file containing
  semicolon-separated values. Uses multiple threads to read the
  file (divided to `n` logical segments) and computes stats for each
  segment, then combines the results. Default number of segments is the
  number of available cores + 2, same as the number of threads used by
  `pmap`."
  ([filename]
   (parallel-stats filename (+ 2 (.. Runtime getRuntime availableProcessors))))
  ([filename n]
   (let [file (io/file filename)]
     (reduce #(merge-with merge-station-stats %1 %2)
             (pmap (fn [segment]
                     (transducing-stats (file-segment-line-seq segment)))
                   (file-segment-seq file n))))))

(comment

  ;; 80% better than baseline (4.5 min vs 22 min), with 4 physical (8 logical) cores (Ryzen 5 3400G)

  ;; 1e6 lines: 0.3 sec
  (time (format-stats (parallel-stats "dev/resources/measurements1M.txt")))

  ;; 1e9 lines: 263 sec (~4.5 min)
  (time (format-stats (parallel-stats "dev/resources/measurements.txt")))

)

И пока набор символов и «алфавитный порядок» не определены

Всё там определено, но, повторю, эта информация в огромных таблицах. Можно пользоваться libicu которая сама с ними взаимодействует.

как отсортировать эмоджи

А главное не забывать, что и эмоджи и даже буквы - это не обязательно кодепоинт. Одна буква может быть последовательностью хоть из 10 кодепоинтов (и соответственно 20+ байт utf-8), но сортировать её надо именно как одну букву, а не как первый кодепоинт из неё.

Разумеется в конце, а во время чтения сортировать ничего не надо, надо хеш-таблицу сделать с шириной в 10000 элементов чтоб она с 1 попытки всегда находила нужное.

Но это не отменяет необходимость этого неприятного аспекта - т.к. вместо оптимизации алгоритмов придётся разбираться с особенностями кривого юникода.

Так и должно быть.

Всё там определено

Я имею в виду, что в задаче не определён алфавит. В зависимости от языка и культуры сортировка может отличаться же. Но для некоторых групп Юникода алфавитный порядок вообще невозможен.

Смайликов в названиях станций очевидно не бывает. А те названия, что есть - вполне сортируются.

Так и должно быть.

А мне кажется, что нет.

Смайликов в названиях станций очевидно не бывает.

Не очевидно.

В ридми написано, что названием может быть любая UFT-8 строка от 1 до 100 байт.

Но это не отменяет необходимость этого неприятного аспекта - т.к. вместо оптимизации алгоритмов придётся разбираться с особенностями кривого юникода.

Мне так думается что финальная сортировка 10k строк вне зависимости от их внутреннего представления - это о-малое от прожёвывания 1B записей.

Тривиальная версия на Guile (никто не постил ещё?)

#!/usr/bin/env -S guile
!#

(import (srfi srfi-1)
        (srfi srfi-26)
        (ice-9 textual-ports))

(define (parse port cities)
  (let ((line (get-line port)))
    (when (not (eof-object? line))
      (let* ((l (string-split line #\;))
             (city-name (first l))
             (temperature (string->number (second l)))
             (city (hash-ref cities city-name)))
        (hash-set! cities city-name
                   (if city
                       `(,(+ (first city) 1)
                         ,(min (second city) temperature)
                         ,(+ (third city) temperature)
                         ,(max (fourth city) temperature))
                       `(1 ,temperature ,temperature ,temperature))))
      (parse port cities))))

(define (parse-file file)
  (let ((cities (make-hash-table)))
    (call-with-input-file file (cut parse <> cities))
    (sort (map (lambda (c)
                 `(,(first c)
                   ,(third c)
                   ,(/ (fourth c) (second c))
                   ,(fifth c)))
               (hash-map->list cons cities))
          (lambda (a b) (string< (car a) (car b))))))

(parse-file "./measurements.txt")

Результат (Ryzen 7 4700U):

$ time ./1brc.scm

________________________________________________________
Executed in   23.20 mins    fish           external
   usr time   34.63 mins  421.00 micros   34.63 mins
   sys time    1.15 mins  274.00 micros    1.15 mins

Для скорости пробовал делать hash-get-handle и set-cdr!, но время получилось то же самое. Да и нечего тут оптимизировать, потому что простое чтение строк без обработки занимает почти 10 минут.

PS: второй запуск выполнялся 20 минут.

Я не про процессорное время, а про то что программисту надо с этим возиться.

Я не про процессорное время, а про то что программисту надо с этим возиться.

Это связанные вещи. Даже самое кривенькое решение «в лоб» погоды не сделает. Кмк, это не то на чём нужно фокусироваться.

простое чтение строк без обработки занимает почти 10 минут.

HDD шоле?

Блин, кот на лоре практически невозможно читать с телефона. Даже в лежачей ориентации всего символов 56 помещается, не говоря уже про торчковую.

С точки зрения читабельности текста как такового это окей, там около 65 символов оптимальная длина строки. Но под кот-то надо хотя бы 80 иметь. (120-то это ересь новомодная, конечно.)

HDD шоле?

Нет, nvme :) Самый быстрый вариант на Си, который с SSE и многопотоком, выполняется за 1.2 сек.

Нужна горизонтальная прокрутка.

Наврал немного. Ноут на Ryzen 7 4700U отработал код на Си за 4 сек. Комп на Зионе за 1.2 сек. Код на Guile оба примерно за одно время выполняют.

Чтением в bytevector блоками по мегабайту и многопоточной обработкой (через futures) удалось ускориться до 4 минут.

А что это мы все на проце да на проце. Побуду немного python-макакой и посчитаю все на видеокарте:

#!/usr/bin/env python3

# Installation:
# conda create --solver=libmamba -n rapids-23.12 -c rapidsai -c conda-forge -c nvidia rapids=23.12 python=3.9 cuda-version=11.2 tensorflow
# conda activate rapids-23.12

from dask.distributed import Client, wait
from dask_cuda import LocalCUDACluster
from dask.utils import parse_bytes
from time import time
import dask_cudf
import warnings

warnings.simplefilter("ignore")

if __name__ == '__main__':
    # prepare environment for RTX A6000 with 48GB of video memory
    memory = "46GB"
    cluster = LocalCUDACluster(
        CUDA_VISIBLE_DEVICES="0",
        rmm_pool_size=parse_bytes(memory), 
        device_memory_limit=parse_bytes(memory),
        threads_per_worker=16,
        scheduler_port=12345
    )
    client = Client(cluster)
    print("Started cluster: ", cluster)

    # ready steady go
    start = time()
    df = dask_cudf.read_csv('measurements.txt', sep=';', header=None, names=['station', 'temp'])
    df = (df.set_index('station')
            .groupby('station')           
            .agg(['min','mean','max']))
    rows = df.compute()
    wait(rows)
    result = client.gather(rows)
    result.sort_index(inplace=True)
    print(result.to_string())
    print("Time: {} sec".format(time() - start))

Пробуем:

Started cluster:  LocalCUDACluster(2cc990ac, 'tcp://127.0.0.1:12345', workers=1, threads=16, memory=125.61 GiB)
               (temp, min)  (temp, mean)  (temp, max)
station                                              
Abha                 -37.4     18.003349         65.3
Abidjan              -23.3     25.995575         73.9
Abéché               -20.2     29.393758         78.6
Accra                -21.3     26.402094         73.9
Addis Ababa          -35.8     16.007987         64.0
...                    ...           ...          ...
Zagreb               -40.8     10.700093         62.5
Zanzibar City        -20.1     25.999807         73.5
Zürich               -42.1      9.291377         62.3
Ürümqi               -41.0      7.400875         58.7
İzmir                -35.9     17.915624         69.1

[413 rows x 3 columns]
Time: 40.325023889541626 sec

Итого, 40 сек на 10^9 записей (10^6 переваривает за 1.5 сек). 2 строки кода на развертывание кластера и 6 строк само решение.

Интересно, тут кто-нибудь еще из минуты смог выйти со своим решением?

Круто! Хорошая у тебя видеакарта) А точно 40 секунд? Если 1e6 за 1.5, то 1e9 должно в 1000 раз дольше считать, мне кажется. Либо 1.5 для 1e6 - слишком медленно.

Интересно, тут кто-нибудь еще из минуты смог выйти со своим решением?

На питоне - никого. AntonI делал простой вариант, который считал 13+ минут.

На Си у firkax самое адекватное решение, в один поток за 25 сек.

На Си у firkax самое адекватное решение, в один поток за 25 сек.

единственно полезный результат из всего этого мегачелленджа.

руст не пробегал?

А точно 40 секунд?

Я запускал с десяток раз, время прыгало от 39 до 42 сек.

Если у вас RTX карта то можете изменить строку memory = "46GB" под вашу видеокарту и перепроверить, результат будет хуже, но сопоставимый.

Вот здесь есть сравнения по скорости и notebook где датасет NYC Taxi 2017 на 8 видеокартах обрабатывают за 10.9 сек.

Если 1e6 за 1.5, то 1e9 должно в 1000 раз дольше считать, мне кажется.

Там самое долгое это загрузка данных в видеопамять, из 40 секунд ~32 сек это загрузка и ~8 сек сами вычисления. Если взять карту с быстрой памятью, например, Nvidia V100 c HBM2, то результат можно улучшить раза в полтора.

Либо 1.5 для 1e6 - слишком медленно.

Без загрузки данных в видеопамять для 1e6 (т.е. CPU+RAM) будет 1.8-1.9 сек, с загрузкой в видеопамять - 1.5 сек. Там нелинейная зависимость.

На Си у firkax самое адекватное решение, в один поток за 25 сек.

Это отличный результат, мое решение можно еще оптимизировать переписав с Python на Nvidia CUDA C, но вряд ли получится догнать до 25 сек, да и тратить на это несколько вечеров мне лень. Код на питоне написал примерно за полчаса просто проверить идею т.к. исходные данные прямо кричат что это csv.

Вообще-то челлендж по ссылке и там есть варианты и быстрее. А я тут просто так запостил свой вариант.

На самом деле там оптимизатор компиляторный странно себя ведёт (в т.ч. -O2 существенно быстрее чем -O3 в ряда вариантов кода получалось, или разница в скорости в полтора раза в зависимости от наличия или отсутствия какой-то не особо значимой строчки в коде) и если с ним всё нормально сделать то будет наверно ещё быстрее. Указанные странности я выяснил когда попытался разбить затраченное время на части кода, которы его тратят и обнаружил что это весьма сложно и отключение отдельных блоков ведёт себя не так как ожидалось.

Варианты по 10 минут это другое конечно, считай конкурс «как выжать максимум из тормозных скриптов». Можно даже на баше попробовать для прикола, тоже интересно будет.

в один поток за 25 сек

А в задаче кстати указано сколько у нас ядер доступно? А то можно распараллелить хоть в 1000 раз и упрётся в память (архитектура которой зависит от материнки), ну и в i/o чтение данных с диска.

А что это мы все на проце да на проце.

Хотел на плисе запилить, а потом подумал, что ну его наhер. К тому же у меня плата с pcie gen2 x4, быстро передать данные не получится.

тут кто-нибудь еще из минуты смог выйти со своим решением?

Тут нет, а по ссылке в главпосте есть вариант меньше секунды (на моём компе 1.2 сек).

Тут нет, а по ссылке в главпосте есть вариант меньше секунды (на моём компе 1.2 сек).

По ссылке я видел, там очевидные решения от «засунуть все в clickhouse» до «распараллелить разбив на куски и хешем его, хешем». Интересно было посмотреть как местные справятся предложив свои собственные варианты. Например, все еще жду Lisp вариант от lovesan который накажет python.

единственно полезный результат из всего этого мегачелленджа.

Le Huy Duc парсит написанные текстом float'ы в настоящие int'ы через любопытное:

  // PhD code from curiouscoding.nl 
  int value;  
  memcpy(&value, data + pos + 1, 4);
  value <<= 8 * (data[pos + 2] == '.');
  constexpr uint64_t C = 1 + (10 << 16) + (100 << 24); // holy hell
  value &= 0x0f000f0f; // new mask just dropped
  value = ((value * C) >> 24) & ((1 << 10) - 1); // actual branchless
  value *= sign;

Таки проверил (правда только через tcc -run, без настоящей компиляции) - на миллиарде итераций парсинга одной и той же строки "-23.1" в int=-231

• через посимвольный перебор с while получается real 0m8,697s
• через этот «PhD code» получается real 0m6,829s

Там самое долгое это загрузка данных в видеопамять, из 40 секунд ~32 сек это загрузка

OMG, это много! Если файл 13ГБ, то 13/32 ~= 0.41 ГБ/с, что как-то совсем уныло, даже проигрывает по скорости чтению с диска. Скорее всего 32 сек - это не только загрузка, но и парсинг CSV.

В доке пишут, что dask_cudf.read_csv() параллелится только если на вход подаются несколько файлов, либо указан параметр blocksize.

руст не пробегал?

Здесь - нет, а там - да. Хочешь попробовать?)

-O2 существенно быстрее чем -O3 в ряда вариантов кода получалось

На таких объёмах там любая мелочь может быть существенной, например, вравнивнивание инструкций, их размеры, размеры переменных. Компилял как-то одну прожку с SIMD, так она с -Oz быстрее работала, чем с -Ofast, так как там был ботлнек по скорости чтения инструкций, а ещё менял 256-битные регистры на 128-битные - тоже было ощутимо лучше.

Пока на конкретной машине не протестишь, неизвестно как будет лучше.

А в задаче кстати указано сколько у нас ядер доступно?

Да, сейчас тестируют на 8 ядрах.

В доке пишут, что dask_cudf.read_csv() параллелится только если на вход подаются несколько файлов, либо указан параметр blocksize.

Да, я не читерил и не разбивал предварительно на кучу файлов как в некоторых опубликованных решениях. Так то можно доподготавливливать их до того что они окажутся сразу в оперативной памяти готовые к быстрой загрузке в видеопамять.

Возможно, мне стоило отключить хром с кучей открытых вкладок при тестах. Потом перемеряю если не забуду.

из копытца воды напиться?… ой плохо кончаются такие сказочки.

😄

Если 1e6 за 1.5, то 1e9 должно в 1000 раз дольше считать, мне кажется

Наверное, там постоянный оверхед большой на раскочегаривание всего этого великолепия (и эта часть на петоне), поэтому такая нелинейность.

Побуду немного python-макакой

Погоди-ка, это что, весь файл в память сначала помещается?

Погоди-ка, это что, весь файл в память сначала помещается?

Да, в видеопамять.

жду Lisp вариант от lovesan который накажет python

Не питон, а невидию RTX с 48Гб памяти %) Ты же тупо закидал задачу железом.

Я думаю, в районе 3-4 минут должно получиться, если в несколько потоков, одна реализация тут уже была. Питон в этом треде пока только 13 смог.

Быстрее 4 минут никак не получается. Основное время тратится на парсинг строки.

Ну я, если честно, тоже забил на произвольную длину станций. Хотя явно это кажется на алгоритм не влияет.

Вообще, как по мне, при всей абсурдности задачи (никому нафиг не надо вычислять среднее само по себе), требование на длину станции от 1 до 100 символов самое абсурдное.

Я сделал branchless разбор температуры и еще немного кое-где чего. Теперь на горячем старте получаю 8.5 с.

sergeis@sergeis-pc1850:~/Projects/fortran_file_mmap$ time ./test_mmap
Storage sizes for char, int(1), int(2), int, int(4), real [bits]: 
                    8     8      16      64    32     32
 File 'measurements.txt' has size           13795786022  bytes
 ******************* Using C functions to open and mmap the file *******************
FSTAT status: -1
Our C file descriptor FD1 is: 3

 **********  Here we have mmap-ed the file in C and now doing Fortran stuff **********


real	0m8.718s
user	1m55.854s
sys	0m0.912s

Из любопытного и подводных камней: файловый дескриптор в С никак не связан с логическим номером файла в фортране (хотя выглядят и присутствуют в функциях абсолютно одинаково); стандартная функция считывания строки в число жутко медленная. Даже с if-then-else вышло 5-ти кратное ускорение. Еще судя по выхлопу компилятора с -fopt-info он сам делает инлайн хеш-функции (тестовая замена вызова хэш-функции на «что-то попроще, типа суммы» замедляет) и векторизует вычисления. При этом следов sse не видно.

Ты же тупо закидал задачу железом.

С чего это вдруг? Можно взять видеокарту попроще с достаточным объемом видеопамяти чтобы загрузить весь файл и с cuda ядрами. У меня просто нет под рукой карты попроще.

Остальные реализации используют ядра процессора и достаточный объем оперативной памяти, если я правильно понял там вообще на AMD Epyc тестили со 128Gb памяти. Но это «закидать железом» почему-то не считается :)

Профилирование показало что кластер этой конечно хорошо, но с одной картой смысла не имеет тк мы сейчас на скорость загружаем, а там много времени тратится на построение индексов при загрузке данных и прочих не нужных в этой задаче вещей. Поэтому я выбросил кластер и стал использовать libcudf через dask напрямую. Это ускорило выполнение в 5 раз, с 40 до 8 секунд.

Код стал совсем неприлично маленьким и работающим на любой RTX видеокарте:

#!/usr/bin/env python3

# Installation:
# conda create --solver=libmamba -n rapids-23.12 -c rapidsai -c conda-forge -c nvidia rapids=23.12 python=3.9 cuda-version=11.2 tensorflow
# conda activate rapids-23.12

from time import time
import dask_cudf

start = time()

df = dask_cudf.read_csv('measurements.txt', sep=';', header=None, names=['station', 'temp'], dtype=['str','float64'], blocksize="64MB")
result = df.groupby('station')
           .agg({'temp': ['min', 'mean', 'max']})
           .compute()
result.sort_index(inplace=True)

print(result.to_string())
print("Total time: {} sec".format(time() - start))

Результат:

               temp                 
                min       mean   max
station                             
Abha          -37.4  18.003349  65.3
Abidjan       -23.3  25.995575  73.9
Abéché        -20.2  29.393758  78.6
Accra         -21.3  26.402094  73.9
Addis Ababa   -35.8  16.007987  64.0
...             ...        ...   ...
Zagreb        -40.8  10.700093  62.5
Zanzibar City -20.1  25.999807  73.5
Zürich        -42.1   9.291377  62.3
Ürümqi        -41.0   7.400875  58.7
İzmir         -35.9  17.915624  69.1

[413 rows x 3 columns]
Total time: 7.936687469482422 sec

с достаточным объемом видеопамяти чтобы загрузить весь файл

Ну вот это и есть «закидать железом» aka вертикальное масштабирование. Горы памяти, специальное оборудование. Но ведь обязательно придёт момент, когда набор данных вырастет настолько, что в памяти уже никак не поместится. И что тогда? Всё равно придётся обратиться к потоковой реализации.

там вообще на AMD Epyc тестили со 128Gb памяти. Но это «закидать железом» почему-то не считается :)

Красиво жыть не запретишь %)

И что тогда?

Тогда как все, делить файлы на куски, обрабатывать по отдельности, затем сливать результат. Ну, всё равно же круто.

P.S. А вот, что immutable data structure и software transaction memory в данном случае ни разу не помогают и только мешаются меня огорчило.

Ну вот это и есть «закидать железом» aka вертикальное масштабирование. Горы памяти, специальное оборудование. Но ведь обязательно придёт момент, когда набор данных вырастет настолько, что в памяти уже никак не поместится. И что тогда? Всё равно придётся обратиться к потоковой реализации.

Посмотрите на новый вариант, там как раз потоковая реализация, нет загрузки всего файла в память и не указан даже объем этой самой памяти. Работает на обычных игровых картах, например, на RTX 4090 будет даже на 0.5-1 сек быстрее чем у меня.

что immutable data structure и software transaction memory в данном случае ни разу не помогают и только мешаются меня огорчило

Просто выжимание последних соков из железа — не та задача, для решения которой они предназначены. Они предназначены для разрешения проблем, создаваемых shared mutable state, особенно в многопоточных средах — гонки данных, дедлоки и всё такое.

Immutable data structures делают из shared mutable state просто shared state и одним этим полностью исключают огромный шмат проблем; управление изменением состояния без явных блокировок — atoms, refs (где требуется координация изменения нескольких кусков shared state — собственно STM), agents для асинхронных изменений, — исключают другой немаленький шмат проблем.

У нас тут нет вообще никакого shared state, даже многопоточные реализации не используют никаких общих (между потоками) данных, каждый поток работает со своим куском.

dask_cudf.read_csv() - оно по сути ничем не отличается от clickhouse.read_csv() или duckdb.read_csv() или ДругаяЧорнаяКоробочка.read_csv(), плюс еще и спец.оборудование требует.

Это не программирование, это пользование.

-------

Тем не менее - по мотивам вашего кода, но без CUDA

from time import time
import dask.dataframe as dd

start = time()
df = dd.read_csv(
    '/home/Sources/1brc/measurements.txt'
    , sep=';'
    , header=None
    , names=['station', 'temp']
    , dtype={'station': 'str', 'temp': 'float64'}
    , blocksize="64MB"
    , engine='c'
)
result = df.groupby('station').agg({'temp': ['min', 'mean', 'max']}).compute()
result.sort_index(inplace=True)

print(result.to_string())
print("Total time: {} sec".format(time() - start))

)