Сети и потоки

Julia предоставляет богатый интерфейс для работы с потоковыми объектами ввода-вывода, такими как терминалы, конвейеры и TCP-сокеты. Эти объекты позволяют отправлять и получать данные в виде потока, что означает, что данные обрабатываются последовательно по мере их появления. Этот интерфейс, хотя асинхронный на уровне системы, представлен программисту синхронным образом. Это достигается за счет активного использования кооперативного многопоточия Julia (корутины).

Основные операции ввода-вывода с потоками

Все потоки Julia предоставляют как минимум метод read и метод write, принимающие поток в качестве первого аргумента, например:

julia> write(stdout, "Hello World");  # suppress return value 11 with ;
Hello World
julia> read(stdin, Char)

'\n': ASCII/Unicode U+000a (category Cc: Other, control)

Обратите внимание, что write возвращает 11, количество байт (в "Hello World") записанных в stdout, но это значение возврата подавляется с помощью ;.

Здесь была нажата клавиша Enter, чтобы Julia прочитала перевод строки. Как видно из этого примера, write принимает данные для записи в качестве второго аргумента, а read принимает тип данных для чтения в качестве второго аргумента.

Например, для чтения простого массива байтов можно сделать следующее:

julia> x = zeros(UInt8, 4)
4-element Array{UInt8,1}:
 0x00
 0x00
 0x00
 0x00

julia> read!(stdin, x)
abcd
4-element Array{UInt8,1}:
 0x61
 0x62
 0x63
 0x64

Однако, поскольку это немного неудобно, предоставляется несколько удобных методов. Например, вышеприведенное можно записать так:

julia> read(stdin, 4)
abcd
4-element Array{UInt8,1}:
 0x61
 0x62
 0x63
 0x64

или если бы мы хотели прочитать всю строку:

julia> readline(stdin)
abcd
"abcd"

Обратите внимание, что в зависимости от настроек вашего терминала, ваш TTY ("телетайп-терминал") может быть буферизован по строкам и, следовательно, может потребовать дополнительного нажатия Enter перед тем, как stdin данные будут отправлены в Julia. При запуске Julia из командной строки в TTY вывод по умолчанию направляется в консоль, а стандартный ввод считывается с клавиатуры.

Для чтения каждой строки из stdin можно использовать eachline:

for line in eachline(stdin)
    print("Found $line")
end

или read, если вы хотите читать по символам:

while !eof(stdin)
    x = read(stdin, Char)
    println("Found: $x")
end

Ввод-вывод текста

Обратите внимание, что метод write, упомянутый выше, работает с бинарными потоками. В частности, значения не преобразуются в какое-либо каноническое текстовое представление, а записываются как есть:

julia> write(stdout, 0x61);  # suppress return value 1 with ;
a

Обратите внимание, что a записывается в stdout функцией write и что возвращаемое значение равно 1, так как 0x61 является одним байтом.

Для текстового ввода-вывода используйте методы print или show, в зависимости от ваших потребностей (см. документацию по этим двум методам для подробного обсуждения различий между ними):

julia> print(stdout, 0x61)
97

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

Свойства контекста вывода IO

Иногда вывод IO может выиграть от возможности передавать контекстную информацию в методы отображения. Объект IOContext предоставляет этот механизм для привязки произвольных метаданных к объекту IO. Например, :compact => true добавляет параметр подсказки к объекту IO, что вызванный метод отображения должен вывести более короткий вывод (если применимо). Обратитесь к документации IOContext для списка общих свойств.

Работа с файлами

Вы можете записать содержимое в файл с помощью метода write(filename::String, content):

julia> write("hello.txt", "Hello, World!")
13

(13 — количество записанных байтов.)

Вы можете прочитать содержимое файла с помощью метода read(filename::String) или read(filename::String, String) для получения содержимого как строки:

julia> read("hello.txt", String)
"Hello, World!"

Расширенная работа: потоковая передача файлов

Методы read и write позволяют читать и записывать содержимое файла. Как и во многих других средах, Julia также имеет функцию open, которая принимает имя файла и возвращает объект IOStream, который можно использовать для чтения и записи из файла. Например, если у нас есть файл hello.txt, содержимое которого Hello, World!:

julia> f = open("hello.txt")
IOStream(<file hello.txt>)

julia> readlines(f)
1-element Array{String,1}:
 "Hello, World!"

Если вы хотите записать в файл, вы можете открыть его с флагом записи ("w"):

julia> f = open("hello.txt","w")
IOStream(<file hello.txt>)

julia> write(f,"Hello again.")
12

Если вы сейчас просмотрите содержимое hello.txt, вы заметите, что оно пустое; ничего еще не было записано на диск. Это потому, что IOStream должен быть закрыт, прежде чем запись действительно будет записана на диск:

julia> close(f)

Проверка содержимого hello.txt снова покажет, что оно изменилось.

Открытие файла, выполнение с ним операций и его закрытие — очень распространенный шаблон. Чтобы упростить этот процесс, существует еще один вызов open, который принимает функцию в качестве первого аргумента и имя файла в качестве второго, открывает файл, вызывает функцию с файлом в качестве аргумента, а затем снова закрывает его. Например, задана функция:

function read_and_capitalize(f::IOStream)
    return uppercase(read(f, String))
end

Вы можете вызвать:

julia> open(read_and_capitalize, "hello.txt")
"HELLO AGAIN."

для открытия hello.txt, вызова read_and_capitalize для него, закрытия hello.txt и возврата прописного содержимого.

Чтобы даже не нужно было определять именованную функцию, можно использовать синтаксис do, который создает анонимную функцию на лету:

julia> open("hello.txt") do f
           uppercase(read(f, String))
       end
"HELLO AGAIN."

Если вы хотите перенаправить стандартный вывод в файл

out_file = open("output.txt", "w")

# Redirect stdout to file
redirect_stdout(out_file) do
    # Your code here
    println("This output goes to `out_file` via the `stdout` variable.")
end

# Close file
close(out_file)

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

Простой пример TCP

Давайте сразу перейдем к простому примеру, связанному с TCP-сокетами. Эта функциональность находится в стандартной библиотеке под названием Sockets. Давайте сначала создадим простой сервер:

julia> using Sockets

julia> errormonitor(@async begin
           server = listen(2000)
           while true
               sock = accept(server)
               println("Hello World\n")
           end
       end)
Task (runnable) @0x00007fd31dc11ae0

Для тех, кто знаком с API сокетов Unix, имена методов будут знакомы, хотя их использование несколько проще, чем использование API сырых сокетов Unix. Первый вызов listen создаст сервер, ожидающий входящих подключений на указанном порту (2000) в данном случае. Такую же функцию можно использовать для создания различных других типов серверов:

julia> listen(2000) # Listens on localhost:2000 (IPv4)
Sockets.TCPServer(active)

julia> listen(ip"127.0.0.1",2000) # Equivalent to the first
Sockets.TCPServer(active)

julia> listen(ip"::1",2000) # Listens on localhost:2000 (IPv6)
Sockets.TCPServer(active)

julia> listen(IPv4(0),2001) # Listens on port 2001 on all IPv4 interfaces
Sockets.TCPServer(active)

julia> listen(IPv6(0),2001) # Listens on port 2001 on all IPv6 interfaces
Sockets.TCPServer(active)

julia> listen("testsocket") # Listens on a UNIX domain socket
Sockets.PipeServer(active)

julia> listen("\\\\.\\pipe\\testsocket") # Listens on a Windows named pipe
Sockets.PipeServer(active)

Обратите внимание, что тип возвращаемого значения последнего вызова отличается. Это потому, что этот сервер не слушает TCP, а слушает именованную трубу (Windows) или сокет доменной области Unix. Также обратите внимание, что в формате именованных труб Windows должен быть определенный шаблон, такой что префикс имени (\\.\pipe\) однозначно идентифицирует тип файла. Разница между TCP и именованными трубами или сокетами доменной области Unix незначительна и связана с методами accept и connect. Метод accept получает подключение к клиенту, подключающемуся к серверу, который мы только что создали, в то время как функция connect подключается к серверу с использованием указанного метода. Функция connect принимает те же аргументы, что и listen, поэтому, при условии, что среда (т. е. хост, текущий каталог и т. д.) одинакова, вы должны иметь возможность передать те же аргументы в connect, что и для прослушивания, чтобы установить соединение. Давайте попробуем это (после создания сервера выше):

julia> connect(2000)
TCPSocket(open, 0 bytes waiting)

julia> Hello World

Как ожидалось, мы увидели "Hello World". Итак, давайте проанализируем, что произошло за кулисами. При вызове connect мы подключаемся к серверу, который мы только что создали. Тем временем функция accept возвращает соединение со стороны сервера с недавно созданным сокетом и выводит "Hello World", чтобы указать, что подключение было успешно.

Отличительной чертой Julia является то, что, поскольку API представлен синхронно, хотя ввод-вывод фактически происходит асинхронно, нам не нужно было беспокоиться о обратных вызовах или даже о том, чтобы убедиться, что сервер запустится. Когда мы вызывали connect, текущая задача ожидала установления соединения и продолжила выполнение только после того, как это было сделано. Во время этой паузы задача сервера возобновляла выполнение (потому что теперь доступна запрос соединения), принимала подключение, выводила сообщение и ожидала следующего клиента. Чтение и запись работают аналогичным образом. Для этого рассмотрим следующий простой эхо-сервер:

julia> errormonitor(@async begin
           server = listen(2001)
           while true
               sock = accept(server)
               @async while isopen(sock)
                   write(sock, readline(sock, keep=true))
               end
           end
       end)
Task (runnable) @0x00007fd31dc12e60

julia> clientside = connect(2001)
TCPSocket(RawFD(28) open, 0 bytes waiting)

julia> errormonitor(@async while isopen(clientside)
           write(stdout, readline(clientside, keep=true))
       end)
Task (runnable) @0x00007fd31dc11870

julia> println(clientside,"Hello World from the Echo Server")
Hello World from the Echo Server

Как и в других потоках, используйте close для разрыва соединения сокета:

julia> close(clientside)

Разрешение IP-адресов

Один из методов connect, который не следует методам listen, это connect(host::String,port), который попытается подключиться к хосту, заданному параметром host, на порту, заданном параметром port. Он позволяет выполнять такие действия:

julia> connect("google.com", 80)
TCPSocket(RawFD(30) open, 0 bytes waiting)

В основе этой функциональности лежит getaddrinfo, которая выполнит соответствующее разрешение адреса:

julia> getaddrinfo("google.com")
ip"74.125.226.225"

Асинхронный ввод-вывод

Все операции ввода-вывода, предоставляемые Base.read и Base.write, могут выполняться асинхронно с помощью корутин. Вы можете создать новую корутину для чтения или записи в поток, используя макрос @async:

julia> task = @async open("foo.txt", "w") do io
           write(io, "Hello, World!")
       end;

julia> wait(task)

julia> readlines("foo.txt")
1-element Array{String,1}:
 "Hello, World!"

Часто возникает ситуация, когда необходимо выполнить несколько асинхронных операций одновременно и дождаться их завершения. Вы можете использовать макрос @sync, чтобы заставить вашу программу заблокироваться, пока все корутины, которые он обертывает, не завершатся:

julia> using Sockets

julia> @sync for hostname in ("google.com", "github.com", "julialang.org")
           @async begin
               conn = connect(hostname, 80)
               write(conn, "GET / HTTP/1.1\r\nHost:$(hostname)\r\n\r\n")
               readline(conn, keep=true)
               println("Finished connection to $(hostname)")
           end
       end
Finished connection to google.com
Finished connection to julialang.org
Finished connection to github.com

Мультипоток

Julia поддерживает мультипоток по IPv4 и IPv6 с использованием протокола пользовательских датаграмм (UDP) в качестве транспорта.

В отличие от протокола управления передачей (TCP), UDP почти не делает предположений о потребностях приложения. TCP обеспечивает управление потоком (он ускоряет и замедляет для максимальной пропускной способности), надёжность (потерянные или повреждённые пакеты автоматически повторно передаются), упорядочение (пакеты упорядочиваются операционной системой перед передачей приложению), размер сегмента и установку и разрыв сеанса. UDP не предоставляет таких функций.

Распространённое применение UDP — в приложениях с мультипотоком. TCP — это протокол состояния для связи ровно между двумя устройствами. UDP может использовать специальные адреса мультипотока, чтобы обеспечить одновременную связь между многими устройствами.

Приём пакетов IP-мультипотока

Для передачи данных по UDP-мультипотоку просто recv на сокете, и первый полученный пакет будет возвращён. Обратите внимание, что это может быть не первый пакет, который вы отправили!

using Sockets
group = ip"228.5.6.7"
socket = Sockets.UDPSocket()
bind(socket, ip"0.0.0.0", 6789)
join_multicast_group(socket, group)
println(String(recv(socket)))
leave_multicast_group(socket, group)
close(socket)

Отправка пакетов IP-мультипотока

Для передачи данных по UDP-мультипотоку просто send на сокете. Заметьте, что отправителю не обязательно присоединяться к группе мультипотока.

using Sockets
group = ip"228.5.6.7"
socket = Sockets.UDPSocket()
send(socket, group, 6789, "Hello over IPv4")
close(socket)

Пример IPv6

Этот пример предоставляет ту же функциональность, что и предыдущая программа, но использует IPv6 в качестве сетевого протокола.

Слушатель:

using Sockets
group = Sockets.IPv6("ff05::5:6:7")
socket = Sockets.UDPSocket()
bind(socket, Sockets.IPv6("::"), 6789)
join_multicast_group(socket, group)
println(String(recv(socket)))
leave_multicast_group(socket, group)
close(socket)

Отправитель:

using Sockets
group = Sockets.IPv6("ff05::5:6:7")
socket = Sockets.UDPSocket()
send(socket, group, 6789, "Hello over IPv6")
close(socket)