Are nng sockets thread-safe?

[kevinjqiu]

Hi 👋

I'm new to nng. I'd like to know if the sockets in nng are thread-safe. Here's a simple surveyor/respondent client/server setup. The server is written in go with mangos and the client is in Python with pynng.

server.go

package main

import (
	"fmt"
	"log"

	"go.nanomsg.org/mangos/v3/protocol/respondent"
	_ "go.nanomsg.org/mangos/v3/transport/all"
)

func main() {
	// Create a respondent socket
	socket, err := respondent.NewSocket()
	if err != nil {
		log.Fatalf("Failed to create socket: %v", err)
	}
	defer socket.Close()

	// Listen on a Unix domain socket
	if err := socket.Listen("ipc:///tmp/nng.ipc"); err != nil {
		log.Fatalf("Failed to listen: %v", err)
	}

	fmt.Println("Server (respondent) started on ipc:///tmp/nng.ipc")

	for {
		// Wait for survey
		msg, err := socket.Recv()
		if err != nil {
			log.Printf("Failed to receive: %v", err)
			continue
		}

		fmt.Printf("Received survey: %s\n", string(msg))

		// Send response
		response := "Response from Go server"
		err = socket.Send([]byte(response))
		if err != nil {
			log.Printf("Failed to send response: %v", err)
		}
	}
}

client.py

import uuid
import pynng
from concurrent.futures import ThreadPoolExecutor
import traceback

class Client:
    def __init__(self):
        self.socket = pynng.Surveyor0()
        self.socket.dial("ipc:///tmp/nng.ipc")
        print("Client connected to ipc:///tmp/nng.ipc")

    def request(self, msg: str):
        try:
            self.socket.send(msg.encode())
            return self.socket.recv().decode()
        except Exception as e:
            print(f"Error in request {msg}: {e}")
            traceback.print_exc()
            return None

    def __del__(self):
        self.socket.__exit__(None, None, None)


def main():
    client = Client()
    messages = [uuid.uuid4().hex for _ in range(10000)]
    
    # Use a thread pool with 10 workers
    with ThreadPoolExecutor(max_workers=10) as executor:
        # Map the request function to all messages
        results = list(executor.map(client.request, messages))
    
    # Print successful responses (optional)
    successful = [r for r in results if r is not None]
    print(f"Completed {len(successful)} successful requests out of {len(messages)}")


if __name__ == "__main__":
    main()

I'm getting a lot of pynng.exceptions.Canceled: Operation canceled (about half out of the 10000 requests). When I ran the requests in serial, everything is fine so I assume the error was b/c of running the requests in a multithreading environment?

I know I don't have to use multithreading but I'm doing a POC for using nng to write a client which can be run in a multithreaded environment (e.g., gunicorn gthread with 1+ threads). Previously I'm using ZMQ's DEALER-ROUTER sockets but it was clear in ZMQ's documentation that the socket isn't thread-safe. Someone suggested me to take a look at nanomsg and nng which are thread-safe.

Is there anything I'm missing? Thanks!

[kevinjqiu]

Asking in discord instead.