Сети и потоки

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

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

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

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

Простой пример 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 для отключения сокета.

Разрешение 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 с использованием протокола User Datagram Protocol (UDP) в качестве транспортного протокола.

В отличие от протокола Transmission Control Protocol (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)