threadsafe_zmq 2.0.0

Thread-safe ZeroMQ wrapper with sync and async support
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117

# Thread-Safe ZeroMQ

[![Crates.io](https://img.shields.io/crates/v/threadsafe_zmq.svg)](https://crates.io/crates/threadsafe_zmq)
[![Documentation](https://docs.rs/threadsafe_zmq/badge.svg)](https://docs.rs/threadsafe_zmq)
[![License](https://img.shields.io/crates/l/threadsafe_zmq.svg)](LICENSE)

Thread-safe wrapper for ZeroMQ sockets.

## The Problem

ZeroMQ sockets are not thread-safe. From the ZMQ Guide:

> "Do not use or close sockets except in the thread that created them."

If you have a multi-threaded server where workers need to send through a shared socket, you'll get SIGSEGV crashes. Common workarounds like mutexes or proxies have significant overhead.

This library wraps ZMQ sockets in a dedicated thread and exposes channel handles that are `Send + Sync`.

Thread-safe ZeroMQ wrapper for multi-threaded servers.

Trade-off: ~9x latency overhead per message, but enables parallel sending from multiple threads - which can result in higher total throughput than single-threaded raw ZMQ.

## Installation

```toml
[dependencies]
threadsafe_zmq = "2.0"

# For async support
threadsafe_zmq = { version = "2.0", features = ["async"] }
```

Requires ZeroMQ on your system:

```bash
# macOS
brew install zeromq pkg-config

# Ubuntu/Debian
apt-get install libzmq3-dev pkg-config
```

Or use Nix: `just install-nix && just shell`

## Usage

Server with worker threads:

```rust
use threadsafe_zmq::ChannelPairBuilder;
use crossbeam_channel::bounded;
use std::sync::Arc;
use std::thread;

let ctx = zmq::Context::new();
let socket = ctx.socket(zmq::ROUTER)?;
socket.bind("tcp://*:5555")?;

let channel = ChannelPairBuilder::new(&ctx, socket)
    .with_bounded_queue(1000)
    .build()?;

// Work queue for distributing messages to workers
let (work_tx, work_rx) = bounded(1000);

// Spawn workers - each can send responses through the shared channel
for _ in 0..4 {
    let ch = Arc::clone(&channel);
    let rx = work_rx.clone();
    thread::spawn(move || {
        while let Ok(msg) = rx.recv() {
            // Process and respond - safe from any thread
            ch.send(vec![b"response".to_vec()]).unwrap();
        }
    });
}

// Main loop receives and dispatches to workers
loop {
    let msg = channel.recv()?;
    work_tx.send(msg)?;
}
```

## Async

```rust
use threadsafe_zmq::AsyncChannelPair;

let channel = AsyncChannelPair::new(&ctx, socket)?;

channel.send(vec![b"hello".to_vec()]).await?;
let response = channel.recv().await?;

channel.shutdown().await;
```

## Examples

See the `example/` directory for a complete fibonacci server/client:

```bash
just example       # sync version
just example-async # async version
```

## Benchmarks

```bash
just bench
```

Compares ChannelPair against raw ZMQ with mutex and proxy patterns.

## License

Apache-2.0