네트워킹과 스트림

네트워킹과 스트림 (Networking and Streams)

Julia는 터미널, 파이프, TCP 소켓 같은 스트리밍 I/O 객체를 다루는 풍부한 인터페이스를 제공해요. 이런 객체들은 데이터를 스트림 방식으로 주고받게 해주는데, 즉 데이터가 도착하는 대로 순차적으로 처리된다는 뜻이에요.

출처: Julia 공식 매뉴얼 · Networking and Streams

본문

이 인터페이스는 시스템 레벨에서는 비동기지만, 프로그래머에게는 동기(synchronous) 방식으로 보여요. Julia의 협력적 스레딩(cooperative threading, 즉 코루틴) 기능을 적극 활용해서 그렇게 만든 거예요.

기본 스트림 I/O (Basic Stream I/O)

모든 Julia 스트림은 최소한 read와 write 메서드를 제공하는데, 둘 다 스트림을 첫 번째 인자로 받아요. 예를 들면:

julia> write(stdout, "Hello World");  # ;로 반환값 11을 숨긴다
Hello World

julia> read(stdin, Char)

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

...

터미널 설정에 따라 TTY("teletype terminal")가 라인 버퍼링이라서, stdin 데이터가 Julia로 전달되기 전에 추가로 Enter를 눌러야 할 수도 있어요. TTY에서 커맨드라인으로 Julia를 실행하면 출력은 기본적으로 콘솔로 나가고, 표준 입력은 키보드에서 읽어요.

...

텍스트 I/O (Text I/O)

앞서 말한 write 메서드는 이진 스트림 위에서 동작한다는 점을 유의하세요. 특히 값이 어떤 표준 텍스트 표현으로 변환되지 않고 있는 그대로 쓰여요.

julia> write(stdout, 0x61);  # ;로 반환값 1을 숨긴다
a

a가 write 함수에 의해 stdout에 쓰였고 반환값이 1이라는 점(0x61은 1바이트라서)을 확인해 보세요. 텍스트 I/O에는 필요에 따라 print나 show 메서드를 쓰면 돼요(두 메서드의 차이에 대한 자세한 논의는 각각의 문서를 참고하세요).

...

I/O 출력 컨텍스트 속성 (IO Output Contextual Properties)

예를 들어 :compact => true는 호출된 show 메서드가 (가능하면) 더 짧은 출력을 내도록 힌트를 주는 매개변수를 IO 객체에 추가해요. 자주 쓰는 속성 목록은 IOContext 문서에서 확인할 수 있어요.

...

파일 다루기 (Working with Files)

심화: 파일 스트리밍

위의 read·write 메서드로 파일 내용을 읽고 쓸 수 있어요. 다른 많은 환경처럼 Julia에도 open 함수가 있는데, 파일 이름을 받아 그 파일에서 읽고 쓸 수 있는 IOStream 객체를 반환해요.

...

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

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

파일에 쓰고 싶다면 쓰기("w") 플래그로 열면 돼요.

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

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

이 시점에서 hello.txt의 내용을 확인해 보면 비어 있을 거예요. 아직 아무것도 디스크에 쓰이지 않았거든요. IOStream을 닫아야 쓰기가 실제로 디스크에 반영(flush)되기 때문이에요.

julia> close(f)

...

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."

stdout을 파일로 리다이렉트하고 싶다면:

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

# stdout을 파일로 리다이렉트
redirect_stdout(out_file) do
    # 여기에 코드 작성
    println("This output goes to `out_file` via the `stdout` variable.")
end

# 파일 닫기
close(out_file)

...

간단한 TCP 예제 (A simple TCP example)

TCP 소켓을 다루는 간단한 예제로 바로 들어가 볼게요. 이 기능은 표준 라이브러리 패키지인 Sockets에 있어요. 먼저 간단한 서버를 만들어 보죠.

julia> using Sockets

julia> errormonitor(Threads.@spawn begin
           server = listen(2000)        # 2000번 포트에서 연결을 기다리는 서버 생성
           while true
               sock = accept(server)
               println("Hello World\n")
           end
       end)
Task (runnable) @0x00007fd31dc11ae0

Unix 소켓 API를 아는 사람이라면 메서드 이름이 익숙할 거예요. 다만 사용법은 원시 Unix 소켓 API보다 훨씬 간단하죠. listen을 처음 호출하면 지정된 포트(여기서는 2000)에서 들어오는 연결을 기다리는 서버를 만듭니다.

...

julia> listen(2000)              # localhost:2000 (IPv4)에서 수신
Sockets.TCPServer(active)

julia> listen(ip"127.0.0.1",2000)  # 첫 번째와 동일
Sockets.TCPServer(active)

julia> listen(ip"::1",2000)      # localhost:2000 (IPv6)에서 수신
Sockets.TCPServer(active)

julia> listen(IPv4(0),2001)      # 모든 IPv4 인터페이스의 2001번 포트에서 수신
Sockets.TCPServer(active)

julia> listen(IPv6(0),2001)      # 모든 IPv6 인터페이스의 2001번 포트에서 수신
Sockets.TCPServer(active)

julia> listen("testsocket")      # UNIX 도메인 소켓에서 수신
Sockets.PipeServer(active)

julia> listen("\\\\.\\pipe\\testsocket")  # Windows 네임드 파이프에서 수신
Sockets.PipeServer(active)

마지막 호출의 반환 타입이 다르다는 점에 주목하세요. 이 서버는 TCP가 아니라 네임드 파이프(Windows)나 UNIX 도메인 소켓에서 수신하기 때문이에요. 또 Windows 네임드 파이프 형식은 이름 접두사(\\.\pipe\)가 파일 타입을 유일하게 식별하도록 특정 패턴이어야 합니다.

TCP와 네임드 파이프·UNIX 도메인 소켓의 차이는 미묘한데, accept와 connect 메서드와 관련이 있어요. accept 메서드는 방금 만든 서버에 연결하는 클라이언트에 대한 서버 쪽 연결을 가져오고, connect 함수는 지정된 방법으로 서버에 연결해요.

connect 함수는 listen과 같은 인자를 받아요. 그래서 환경(호스트, cwd 등)이 같다면 listen에 넘긴 것과 같은 인자를 connect에도 넘겨 연결을 성립할 수 있어요. (위에서 서버를 만든 뒤에) 직접 시도해 볼게요.

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

julia> Hello World

예상대로 "Hello World"가 출력됐어요. 이제 뒤에서 실제로 무슨 일이 벌어졌는지 분석해 볼게요. connect를 호출했을 때 방금 만든 서버에 연결했어요. 한편 accept 함수는 새로 생긴 소켓에 대한 서버 쪽 연결을 반환하고, 연결이 성공했음을 알리려고 "Hello World"를 출력했죠.

Julia의 큰 강점 중 하나는, 실제 I/O가 비동기로 일어나는데도 API가 동기로 노출되어서 콜백을 신경 쓸 필요도, 서버가 실행되도록 보장할 필요도 없다는 점이에요. connect를 호출하면 현재 태스크가 연결이 성립할 때까지 기다렸다가 그게 끝난 뒤에야 실행을 이어갔어요. 이 지연 동안 서버 태스크는 (연결 요청이 이제 가능해졌으므로) 실행을 재개해 연결을 받아들이고, 메시지를 출력한 다음 다음 클라이언트를 기다렸죠.

...

julia> errormonitor(Threads.@spawn begin
           server = listen(2001)
           while true
               sock = accept(server)
               Threads.@spawn 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(Threads.@spawn 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 주소 해석 (Resolving IP Addresses)

listen 메서드들을 따르지 않는 connect 메서드가 하나 있어요. 바로 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"

비동기 I/O (Asynchronous I/O)

Base.read와 Base.write가 노출하는 모든 I/O 연산은 코루틴을 이용해 비동기로 수행할 수 있어요. Threads.@spawn 매크로로 스트림에서 읽거나 쓰는 새 코루틴을 만들 수 있죠.

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

julia> wait(task)

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

여러 비동기 연산을 동시에 수행하고 모두 끝날 때까지 기다려야 하는 상황도 흔해요. @sync 매크로를 쓰면 감싸고 있는 코루틴들이 전부 끝날 때까지 프로그램이 블록되도록 할 수 있어요.

julia> using Sockets

julia> @sync for hostname in ("google.com", "github.com", "julialang.org")
           Threads.@spawn 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

멀티캐스트 (Multicast)

Julia는 전송 계층으로 사용자 데이터그램 프로토콜(UDP)을 사용해 IPv4와 IPv6를 통한 멀티캐스트를 지원해요.

전송 제어 프로토콜(TCP)과 달리 UDP는 애플리케이션의 요구에 대해 거의 아무 가정도 하지 않아요. TCP는 흐름 제어(처리량을 최대화하려고 속도를 높였다 낮췄다 함), 신뢰성(잃어버리거나 손상된 패킷 자동 재전송), 순서화(애플리케이션에 주기 전에 운영체제가 패킷 순서를 정리함), 세그먼트 크기, 세션 설정·해제를 제공해요. UDP는 그런 기능이 전혀 없죠.

UDP의 흔한 용도는 멀티캐스트 애플리케이션이에요. TCP는 정확히 두 장치 사이의 통신을 위한 상태 저장(stateful) 프로토콜이지만, UDP는 특수한 멀티캐스트 주소를 사용해 많은 장치가 동시에 통신할 수 있게 해줘요.

IP 멀티캐스트 패킷 수신 (Receiving IP Multicast Packets)

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 멀티캐스트 패킷 전송 (Sending IP Multicast Packets)

UDP 멀티캐스트로 데이터를 전송하려면 소켓에 send 하면 돼요. 송신자가 멀티캐스트 그룹에 참여할 필요는 없다는 점을 기억하세요.

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

IPv6 예제 (IPv6 Example)

이 예제는 앞선 프로그램과 같은 기능인데, 네트워크 계층 프로토콜로 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)

더 알아보기

Missing Values (이전 장)
Parallel Computing (다음 장)
Sockets 표준 라이브러리 문서