PAIR 소켓¶

이 장의 계약 소유 문서 — PAIR socket 스펙이 다룬다. 이 챕터는 그 계약을 언어별 예제로 보여준다.

1. 개요¶

PAIR 소켓은 정확히 하나의 peer와 1:1 양방향 독점 연결을 맺는다. 두 번째 peer가 연결하면 그 나중 연결이 거부되고, 첫 번째 peer가 pipe를 유지한다.

핵심 특성: - 단일 파이프만 허용 (1:1 독점) - 양방향 자유 메시징 (send/recv 순서 무관) - 가장 단순한 소켓 타입

유효한 소켓 조합: PAIR ↔ PAIR

%%{init: {'flowchart': {'nodeSpacing': 32, 'rankSpacing': 40, 'padding': 8, 'wrappingWidth': 180}, 'themeVariables': {'fontSize': '18px'}}}%%
flowchart LR
    A[PAIR A] <-->|Bidirectional| B[PAIR B]

2. 기본 사용법¶

생성 및 연결¶

void *ctx = zlink_ctx_new();

/* Server side */
void *server = zlink_socket(ctx, ZLINK_SOCKET_PAIR);
zlink_bind(server, "tcp://*:5555");

/* Client side */
void *client = zlink_socket(ctx, ZLINK_SOCKET_PAIR);
zlink_connect(client, "tcp://127.0.0.1:5555");

메시지 교환¶

PAIR의 공개 수신 API는 recv/poller 전용이다. zlink_recv()로 한 번에 record 전체를 수신하며, 보통 poller 루프 안에서 호출한다. 양쪽 모두 send와 recv를 자유롭게 호출할 수 있다. PAIR는 peer가 정확히 하나이므로 선택할 routing id가 없다 — source_rid_out_에는 NULL을 넘긴다.

/* Client → Server */
zlink_msg_t msg;
zlink_msg_init_size(&msg, 5);
memcpy(zlink_msg_data(&msg), "Hello", 5);
zlink_send(client, &msg, 1, ZLINK_SEND_FLAGS_NONE, NULL, NULL);

/* Server receives with zlink_recv() (typically inside a poller loop) */
zlink_msg_t parts[8];
size_t part_count = 0;
if (zlink_recv(server, NULL, parts, 8, &part_count, ZLINK_RECV_FLAGS_NONE) == ZLINK_RECV_OK) {
    printf("Received: %.*s\n",
           (int)zlink_msg_size(&parts[0]),
           (char *)zlink_msg_data(&parts[0]));
    zlink_multipart_close(parts, part_count);
}

/* Server → Client (bidirectional; client uses the same recv+poller pattern) */
zlink_msg_t reply;
zlink_msg_init_size(&reply, 5);
memcpy(zlink_msg_data(&reply), "World", 5);
zlink_send(server, &reply, 1, ZLINK_SEND_FLAGS_NONE, NULL, NULL);

멀티파트 데이터 전송¶

멀티파트 데이터는 모든 part를 배열에 순서대로 두고 zlink_send()를 한 번 호출해 전송한다. Core는 배열 전체를 하나의 record로 원자적으로 제출한다.

zlink_msg_t parts[2];
zlink_msg_init_size(&parts[0], 3);
memcpy(zlink_msg_data(&parts[0]), "foo", 3);
zlink_msg_init_size(&parts[1], 6);
memcpy(zlink_msg_data(&parts[1]), "foobar", 6);

zlink_submit_result_t rc = zlink_send(
    server, parts, 2, ZLINK_SEND_FLAGS_NONE, NULL, NULL);
/* 수신자는 zlink_recv() 한 번으로 같은 순서의 part 두 개를 받는다. */

참고: core/tests/integration/test_public_inproc_multipart_send.cpp — test_public_inproc_pair_send_multipart_blocking() 테스트

수신 모드¶

PAIR의 공개 수신 API는 recv/poller 전용이다. zlink_recv()로 record 전체를 동기 수신한다.

void *pair = zlink_socket(ctx, ZLINK_SOCKET_PAIR);
zlink_bind(pair, "tcp://*:5556");

zlink_msg_t parts[8];
size_t part_count = 0;
zlink_recv_result_t rc = zlink_recv(
    pair, NULL, parts, 8, &part_count, ZLINK_RECV_FLAGS_NONE);
if (rc == ZLINK_RECV_OK) {
    /* parts[0..part_count)를 처리한 뒤 배열 전체를 닫는다. */
    zlink_multipart_close(parts, part_count);
}

HWM(High-Water Mark, queue가 보관할 수 있는 accounted byte 상한) 도달 시 zlink_send()는 대기(기본) 또는 ZLINK_DONTWAIT로 ZLINK_SUBMIT_BACKPRESSURED를 반환한다. 고급 배압(backpressure) 패턴은 socket option 가이드를 참고.

3. 메시지 형식¶

PAIR 소켓의 메시지 프레임에는 애플리케이션 데이터만 들어간다.

Single frame:     [data]
Multipart frame:  [frame1][frame2]...[frameN]

source_rid 등 공통 수신 인터페이스는 소켓 패턴 개요를 참고.

멀티파트 전송:

zlink_msg_t parts[2];
zlink_msg_init_size(&parts[0], 6);
memcpy(zlink_msg_data(&parts[0]), "header", 6);
zlink_msg_init_size(&parts[1], 4);
memcpy(zlink_msg_data(&parts[1]), "body", 4);

zlink_submit_result_t rc = zlink_send(
    server, parts, 2, ZLINK_SEND_FLAGS_NONE, NULL, NULL);

4. 소켓 옵션¶

옵션	타입	기본값	설명
`ZLINK_OPT_SNDHWM`	`uint64_t` bytes	자동	PAIR의 peer-queue 역할에 맞춰 계산한 자동 HWM. 수동 설정이 우선하며 `0`은 무제한
`ZLINK_OPT_RCVHWM`	`uint64_t` bytes	자동	PAIR의 peer-queue 역할에 맞춰 계산한 자동 HWM. 수동 설정이 우선하며 `0`은 무제한
`ZLINK_OPT_LINGER`	int	-1	close 시 미전송 메시지 대기 시간 (ms), -1=무한
`ZLINK_OPT_SNDTIMEO`	int	1000	송신 타임아웃(ms). 무한 대기는 `-1`을 명시적으로 설정
`ZLINK_OPT_RCVTIMEO`	int	1000	수신 타임아웃(ms). 무한 대기는 `-1`을 명시적으로 설정

uint64_t hwm_bytes = 5 * 1024 * 1024;  /* HWM은 byte이고 정확히 8 byte로 전달한다 */
zlink_set_option(socket, ZLINK_OPT_SNDHWM, &hwm_bytes, sizeof(hwm_bytes));

int linger = 0;  /* return immediately on close */
zlink_set_option(socket, ZLINK_OPT_LINGER, &linger, sizeof(linger));

5. 사용 패턴¶

패턴 1: 스레드 간 시그널링 (inproc)¶

가장 일반적인 PAIR 사용 사례. inproc transport로 스레드 간 제로카피(zero-copy, 메모리 복사 없이 전달) 통신.

/* Main thread */
void *signal = zlink_socket(ctx, ZLINK_SOCKET_PAIR);
zlink_bind(signal, "inproc://signal");

/* Worker thread */
void *worker_signal = zlink_socket(ctx, ZLINK_SOCKET_PAIR);
zlink_connect(worker_signal, "inproc://signal");

/* Worker → Main: task completion signal */
zlink_msg_t msg;
zlink_msg_init_size(&msg, 4);
memcpy(zlink_msg_data(&msg), "DONE", 4);
zlink_send(worker_signal, &msg, 1, ZLINK_SEND_FLAGS_NONE, NULL, NULL);

/* Main: poller 루프(zlink_recv)로 "DONE" 수신 */

참고: core/tests/integration/test_pair_inproc.cpp — bind → connect → bounce 패턴

패턴 2: TCP 통신¶

네트워크를 통한 1:1 통신. 와일드카드 바인드로 포트를 자동 할당할 수 있다.

/* Server: wildcard port */
void *server = zlink_socket(ctx, ZLINK_SOCKET_PAIR);
zlink_bind(server, "tcp://127.0.0.1:*");

/* Query the assigned endpoint */
char endpoint[256];
size_t len = sizeof(endpoint);
zlink_get_option(server, ZLINK_OPT_LAST_ENDPOINT, endpoint, &len);

/* Client: connect using the queried endpoint */
void *client = zlink_socket(ctx, ZLINK_SOCKET_PAIR);
zlink_connect(client, endpoint);

참고: core/tests/integration/test_pair_tcp.cpp — bind_loopback_ipv4() + 와일드카드 바인드

패턴 3: DNS 이름 연결¶

호스트명으로도 연결할 수 있다.

void *client = zlink_socket(ctx, ZLINK_SOCKET_PAIR);
zlink_connect(client, "tcp://localhost:5555");

참고: core/tests/integration/test_pair_tcp.cpp — test_pair_tcp_connect_by_name()

패턴 4: IPC 통신¶

같은 머신의 프로세스 간 통신 (Linux/macOS).

void *server = zlink_socket(ctx, ZLINK_SOCKET_PAIR);
zlink_bind(server, "ipc:///tmp/myapp.ipc");

void *client = zlink_socket(ctx, ZLINK_SOCKET_PAIR);
zlink_connect(client, "ipc:///tmp/myapp.ipc");

참고: core/tests/integration/test_pair_ipc.cpp — IPC 경로 길이 검증 포함

6. 주의사항¶

단일 peer만 허용¶

PAIR 소켓은 하나의 연결만 유지한다. 두 번째 peer가 연결하면 그 나중 연결이 거부되고, 첫 번째 peer가 pipe를 유지한다.

 Allowed:  PAIR A ↔ PAIR B      (1:1)
 Invalid:  PAIR A ← PAIR B      (N:1 attempt: later peers rejected)
               ← PAIR C

N:1 통신이 필요하면 DEALER/ROUTER를 사용한다.

inproc connect/bind 순서¶

inproc transport는 보통 bind를 먼저 호출하지만, connect를 먼저 해도 된다 — bind 이전의 connect는 pending connection으로 보관되었다가 bind 시점에 연결된다.

/* 권장 순서 */
zlink_bind(a, "inproc://signal");     /* 1. bind */
zlink_connect(b, "inproc://signal");  /* 2. connect */

/* connect-before-bind 도 동작 — connect는 pending으로 보관됐다가 bind에서 연결됨 */
zlink_connect(b, "inproc://signal");  /* pending connection으로 보관 */
zlink_bind(a, "inproc://signal");     /* 이 시점에 b의 pending connect가 연결됨 */

IPC 경로 길이¶

IPC endpoint의 파일 경로는 시스템 제한(보통 108자)을 넘을 수 없다.

/* Path too long → ENAMETOOLONG error */
zlink_bind(socket, "ipc:///very/long/path/.../endpoint.ipc");

참고: core/tests/integration/test_pair_ipc.cpp — test_endpoint_too_long()

HWM 동작¶

peer가 연결되지 않았으면 PAIR 송신은 queue에 쌓이지 않고 곧바로 backpressure로 처리된다(ZLINK_DONTWAIT면 ZLINK_SUBMIT_BACKPRESSURED, 아니면 sndtimeo까지 대기). peer가 연결돼 있고 느릴 때는 HWM까지 queue에 쌓이고, HWM을 넘으면 zlink_send()가 대기(기본) 또는 ZLINK_SUBMIT_BACKPRESSURED(ZLINK_DONTWAIT)를 반환한다.

LINGER 설정¶

zlink_close()를 호출할 때 미전송 메시지가 남아 있으면 LINGER 시간만큼 대기한다. 테스트나 빠른 종료가 필요한 경우:

int linger = 0;
zlink_set_option(socket, ZLINK_OPT_LINGER, &linger, sizeof(linger));

← 소켓 패턴 | PUB/SUB →

언어별 완전한 예제¶

PAIR 소켓으로 메시지를 주고받는 자립형 예제다(모든 바인딩, 빌드·실행 검증됨).

C++C#/.NETJavaKotlinPythonNode/TypeScriptJavaScriptGoRust

zlink::context_t ctx;
zlink::pair_socket_t server (ctx);
zlink::pair_socket_t client (ctx);
zlink::socket_monitor_t server_monitor = server.monitor_open ();
zlink::socket_monitor_t client_monitor = client.monitor_open ();

server.bind ("tcp://127.0.0.1:0");
const std::string endpoint = server.options ().last_endpoint ();
assert (!endpoint.empty ());
client.connect (endpoint);
assert (detail::wait_connected (server_monitor, client_monitor));

const std::string sent = detail::k_pair_payload;
zlink::message_t outbound = detail::make_message (sent);
client.send ().message (outbound).submit ();

zlink::received_t inbound;
assert (server.recv (inbound) == 0);
assert (inbound.parts ().size () == 1);
const std::string received = inbound.parts ()[0].to_string ();
assert (received == detail::k_pair_payload);
inbound.close ();
std::printf ("[pair/recv] send: \"%s\" → recv: \"%s\"\n", sent.c_str (), received.c_str ());
return 0;

if (!SampleSupport.IsNativeAvailable())
    return;

using var ctx = Zlink.CreateContext();
using var sender = ctx.CreatePairSocket();
using var receiver = ctx.CreatePairSocket();
string endpoint = SampleSupport.NewEndpoint("tcp", "sample");
using var senderMonitor = sender.MonitorOpen(SocketEvent.ConnectionReady);
using var receiverMonitor = receiver.MonitorOpen(SocketEvent.ConnectionReady);
sender.Bind(endpoint);
receiver.Connect(endpoint);
SampleSupport.WaitConnected(senderMonitor, receiverMonitor);

const string payload = "hello-pair";
using (Message message = Message.From(payload))
    sender.Send().Message(message).Submit();
string receivedPayload = SampleSupport.ReceiveUtf8(receiver, 2000);
Console.WriteLine(
    $"[pair/recv] send: \"hello-pair\" -> recv: \"{receivedPayload}\"");

SampleSupport.ensureNative();
String endpoint = SampleSupport.tcpEndpoint();

try (Context ctx = Zlink.createContext();
     PairSocket server = ctx.createPairSocket();
     PairSocket client = ctx.createPairSocket();
     var serverMonitor = server.monitorOpen(
         systems.zlink.contracts.eventing.MonitorEventType.CONNECTION_READY);
     var clientMonitor = client.monitorOpen(
         systems.zlink.contracts.eventing.MonitorEventType.CONNECTION_READY)) {
    server.bind(endpoint);
    client.connect(endpoint);
    SampleSupport.waitConnected(serverMonitor, clientMonitor);

    try (Message outbound = Message.from(SampleSupport.PAIR_PAYLOAD)) {
        client.send().message(outbound).submit().admitted()
            .toCompletableFuture().join();
    }

    try (systems.zlink.contracts.messaging.Received received = new systems.zlink.contracts.messaging.Received()) {
        server.recv(received, systems.zlink.contracts.sockets.RecvFlags.NONE);
        String value = SampleSupport.singleUtf8(received);
        if (!SampleSupport.PAIR_PAYLOAD.equals(value)) {
            throw new IllegalStateException("unexpected payload: " + value);
        }
        System.out.println("[pair/recv] send: \"" + SampleSupport.PAIR_PAYLOAD
            + "\" \u2192 recv: \"" + value + "\"");
    }
}

SampleSupport.ensureNative()
val endpoint = SampleSupport.tcpEndpoint()

Zlink.createContext().use { ctx ->
    ctx.createPairSocket().use { server ->
        ctx.createPairSocket().use { client ->
            server.monitorOpen(MonitorEventType.CONNECTION_READY).use { serverMonitor ->
                client.monitorOpen(MonitorEventType.CONNECTION_READY).use { clientMonitor ->
                    server.bind(endpoint)
                    client.connect(endpoint)
                    SampleSupport.waitConnected(serverMonitor, clientMonitor)

                    Message.from(SampleSupport.PAIR_PAYLOAD).use { outbound ->
                        client.send().message(outbound).submit()
                    }

                    Received().use { received ->
                        server.recv(received, RecvFlags.NONE)
                        val value = SampleSupport.singleUtf8(received)
                        check(SampleSupport.PAIR_PAYLOAD == value) { "unexpected payload: $value" }
                        println(
                            "[pair/recv] send: \"${SampleSupport.PAIR_PAYLOAD}\"" +
                                " → recv: \"$value\""
                        )
                    }
                }
            }
        }
    }
}

_, endpoint = tcp_endpoint()

with zlink.create_context() as ctx:
    with zlink.create_pair_socket(ctx) as server:
        with zlink.create_pair_socket(ctx) as client:
            with server.monitor_open(zlink.MonitorEventMask.CONNECTION_READY) as server_monitor:
                with client.monitor_open(zlink.MonitorEventMask.CONNECTION_READY) as client_monitor:
                    server.bind(endpoint)
                    client.connect(endpoint)
                    wait_connected(server_monitor, client_monitor)

            async def send():
                submission = client.send().message(b"hello-pair").submit()
                await submission.admitted

            asyncio.run(send())
            received = zlink.create_received()
            if not server.recv_into(received):
                raise AssertionError("expected pair payload")
            with received:
                payload = received.to_bytes_list()
                if payload != [b"hello-pair"]:
                    raise AssertionError(f"unexpected pair payload: {payload!r}")
            print('[pair/recv] send: "hello-pair" → recv: "hello-pair"')

const endpoint = await tcpEndpoint();
const ctx = zlink.createContext();
const server = zlink.createPairSocket(ctx);
const client = zlink.createPairSocket(ctx);

try {
  const serverMonitor = server.monitorOpen([zlink.MonitorEventType.ConnectionReady]);
  const clientMonitor = client.monitorOpen([zlink.MonitorEventType.ConnectionReady]);
  try {
    server.bind(endpoint);
    client.connect(endpoint);
    serverMonitor.recv();
    clientMonitor.recv();
  } finally {
    serverMonitor.close();
    clientMonitor.close();
  }

  const sent = 'hello-pair';
  await client.send().message(Buffer.from(sent)).submit().admitted;

  const received = new zlink.Received();
  server.recv(received);
  try {
    const recv = received.parts[0].data().toString();
    assert.equal(recv, sent);
    console.log(`[pair/recv] send: "${sent}" \u2192 recv: "${recv}"`);
  } finally {
    received.close();
  }
} finally {
  client.close();
  server.close();
  ctx.close();
}

const port = await reservePort();
const endpoint = `tcp://127.0.0.1:${port}`;
const ctx = zlink.createContext();
const server = zlink.createPairSocket(ctx);
const client = zlink.createPairSocket(ctx);

try {
  const serverMonitor = server.monitorOpen([zlink.MonitorEventType.ConnectionReady]);
  const clientMonitor = client.monitorOpen([zlink.MonitorEventType.ConnectionReady]);
  try {
    server.bind(endpoint);
    client.connect(endpoint);
    serverMonitor.recv();
    clientMonitor.recv();
  } finally {
    serverMonitor.close();
    clientMonitor.close();
  }

  const sent = 'hello-pair';
  client.send().message(Buffer.from(sent)).submit();

  const received = new zlink.Received();
  server.recv(received);
  try {
    const recv = received.parts[0].data().toString();
    assert.equal(recv, sent);
    console.log(`[pair/recv] send: "${sent}" → recv: "${recv}"`);
  } finally {
    received.close();
  }
} finally {
  client.close();
  server.close();
  ctx.close();
}

ctx, err := zlink.NewContext()
samplecommon.Must(err)
defer ctx.Close()

server, err := ctx.PairSocket()
samplecommon.Must(err)
defer server.Close()
client, err := ctx.PairSocket()
samplecommon.Must(err)
defer client.Close()

serverMon := samplecommon.OpenMonitor(server)
defer serverMon.Close()
clientMon := samplecommon.OpenMonitor(client)
defer clientMon.Close()

endpoint := samplecommon.UniqueTCP("pair-recv")
samplecommon.Must(server.Bind(endpoint))
samplecommon.Must(client.Connect(endpoint))
samplecommon.WaitConnected(serverMon, clientMon)

sent := "hello-pair"
submission, err := client.Send().Message(samplecommon.Message(sent)).Submit(context.Background())
if err == nil {
    err = submission.Admitted(context.Background())
}
samplecommon.Must(err)

var received zlink.Received
_, err = server.Recv(&received, zlink.RecvFlagsNone)
samplecommon.Must(err)
defer received.Close()
part, err := received.SinglePartOrError()
samplecommon.Must(err)
if !bytes.Equal(part.Data(), []byte(sent)) {
    samplecommon.Must(fmt.Errorf("unexpected payload %q", string(part.Data())))
}

fmt.Printf("[pair/recv] send: %q -> recv: %q\n", sent, string(part.Data()))

let ctx = Context::new().expect("context creation failed");
let endpoint = sample_support::tcp_endpoint();

let server = ctx.pair_socket().expect("server socket failed");
let client = ctx.pair_socket().expect("client socket failed");

let server_mon = SocketMonitor::open(&server).expect("server monitor open failed");
let client_mon = SocketMonitor::open(&client).expect("client monitor open failed");

server.bind(&endpoint).expect("bind failed");
client.connect(&endpoint).expect("connect failed");

sample_support::wait_connected(&[&server_mon, &client_mon]);
drop(server_mon);
drop(client_mon);

let msg = Message::try_from(b"hello-pair").expect("message creation failed");
let submission = client.send().message(msg).submit().expect("send failed");
sample_support::block_on(submission.admitted).expect("send admission failed");

let mut received = zlink::Received::empty();
server
    .recv(&mut received, zlink::RecvFlags::NONE)
    .expect("recv failed");
let payload = received.parts()[0].as_str().expect("utf8 error");
assert_eq!(payload, "hello-pair");
println!("[pair/recv] send: \"hello-pair\" → recv: \"{}\"", payload);

← 소켓 패턴 | PUB/SUB →