Сети и потоки

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 перед отправкой данных в Julia.

Чтобы прочитать каждую строку из 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 для списка общих свойств.

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

Как и во многих других средах, в 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."

Пример простого 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, поэтому, если окружение (т.е. хост, cwd и т.д.) одинаково, вы должны иметь возможность передать те же аргументы в 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)