For networks with high latencies and for certain messaging scenarios where a reliable underlying transport protocol is not required, UDP transport support is desirable.

In our REQ/REP scenario, with a worst-case RTT of 0.3 seconds, the initial TCP handshake alone would take 1 second, which is a lot of overhead for short exchanges.

Question: whether UDP support is planned for go-mangos?

For nanomessage, this was originally specified in sp-udp-mapping-01.txt.

During a proof-of-concept, there were 2 challenges I ran into:

1. asynchronous errors - in TCP, socket errors are instantaneously visible; UDP errors (via ICMP) are reported on the next socket call (which leads to e.g. 'connection refused' on send). Mitigations are
   • poll socket error queue (IP_RECVERR, Linux only);
   • retry operations multiple times (i.e. multiple send calls)
2. different socket semantics: go-mangos PipeListener requires Accept. This can be mimicked in the REQ/REP case using a single listener socket, which de-multiplexes incoming connections. In the PUB/SUB case, this would not work, since the SUB sockets require a passive connection setup.

As an alternative, TCP Fast Open might be an option. Go does not yet support it, but they are working on it.

I'm attempting to upgrade my project to use mangos v2 and am having issues compiling it. My project is using the go modules feature. When I compile my application from a simlinked direectory in the normal $GOPATH/src/github.com/... location it can compile just fine.

When I try to compile the project with go modules turned on from my ~/projects/... directory I see the following error: cannot find module for path nanomsg.org/go/mangos/v2

Is this library incompatible with go modules at this time? v1 works just fine which is confusing.

Hello, first and thank you for this great project.

I'm working with mangos as Req(master)/Resp(agents) protocol with 1 master listening for connections and distributing jobs and N agents connecting to the master and doing distributed work.

I did a PoC as I commented here https://github.com/nanomsg/mangos/issues/189 , in the poc all working fine.

But with real data we have Panics in the master, after agents have been restarted and trying to reconnect again.

[signal SIGSEGV: segmentation violation code=0x1 addr=0x20 pc=0xb81f89]

goroutine 39 [running]:
go.nanomsg.org/mangos/v3.(*Message).Dup(0x0, 0x419ff6)
	/home/vant/go/pkg/mod/go.nanomsg.org/mangos/[email protected]/message.go:157 +0x29
go.nanomsg.org/mangos/v3/protocol/req.(*socket).send(0xc0001f0240)
	/home/vant/go/pkg/mod/go.nanomsg.org/mangos/[email protected]/protocol/req/req.go:89 +0x13d
go.nanomsg.org/mangos/v3/protocol/req.(*socket).AddPipe(0xc0001f0240, 0x1004300, 0xc0000974a0, 0x0, 0x0)
	/home/vant/go/pkg/mod/go.nanomsg.org/mangos/[email protected]/protocol/req/req.go:489 +0x13a
go.nanomsg.org/mangos/v3/internal/core.(*socket).addPipe(0xc00032a000, 0x7f82e205bea0, 0xc000096960, 0x0, 0xc000290100)
	/home/vant/go/pkg/mod/go.nanomsg.org/mangos/[email protected]/internal/core/socket.go:78 +0x176
go.nanomsg.org/mangos/v3/internal/core.(*listener).serve(0xc000290100)
	/home/vant/go/pkg/mod/go.nanomsg.org/mangos/[email protected]/internal/core/listener.go:59 +0xfe
created by go.nanomsg.org/mangos/v3/internal/core.(*listener).Listen
	/home/vant/go/pkg/mod/go.nanomsg.org/mangos/[email protected]/internal/core/listener.go:92 +0x1aa
[Bra] 03-18 18:43:53 [ WARN] Fail to execute command: ./bin/synthetix-manager [] - exit status 2

There is any workaround to avoid this error?

Hi,

It's not clear to me whether Mangos supports sending multiple messages over one socket after receiving one single request. I was under the impression that using a OpenContext on a socket that uses a protocol that supports it, it would be possible to achieve something like:

ctx, _ := sock.OpenContext()

req, _ = ctx.Recv()

// imaginary function that returns a response in multiple chunks from a request
resp := getResponseChunks(req)

for _, chunk := range resp {
	ctx.Send(chunk)
}

But when doing so using a rep socket over tcp, the second call to ctx.Send returns an incorrect protocol state error, so I imagine that it's not possible.

I haven't found detailed documentation on which protocol supports which operations, using contexts, so I might be doing something wrong. My use case is that I need to send a large amount of data (potentially multiple gigabytes) over the socket, so I want to split it into multiple responses. Maybe there's another, better way to achieve that using mangos?

While trying to build a go project, which uses mangos-v2, I get the following error:

go: finding nanomsg.org/go/mangos/v2 v2.0.1
go: nanomsg.org/go/mangos/[email protected]: unknown revision nanomsg.org/go/mangos/v2.0.1
go: error loading module requirements

This is the require statement in my go.mod:

...
require nanomsg.org/go/mangos/v2 v2.0.1
...

I am using go-1.11.

Not sure what I did wrong.

This fixes an important security and reliability concern -- a some of the early negotiation logic was on a synchronous path for dialing, and this could allow a bad peer to either slow or completely stall out any new connections from being established.

The Dialer side is still synchronous, by necessity, as it only opens a single connection at a time.

In https://github.com/nanomsg/nng - we have a new PAIRv1 protocol, which adds better resilience for loops in device plumbing.

We should go ahead and add that. We might also use this opportunity to try to support polyamorous mode from nng in our new v1 pipes.

Hi,

I am looking for some suggestions to connect 0MQ client to 0MQ server behind ELB/Proxy.

I have a client and server code as containers and not able to connect with docker. Same issue with connecting via ELB.

Any suggestions?

Hey guys,

Great work, love the project, but when I try to install the application using our CI pipeline I run into a problem from my build script. The build process pulls the code from our private repo and then uses go get -t -v "./..." to pull go dependencies. There are problems with pulling the code, though, and that causes an exit code of 1 to be generated in our build script. This fails the build entirely, and I don't want to special case skip that check for all go packages because if Mangos. I'm hoping you can advise.

The build script:

#!/bin/bash -e

# Ted: contact me when you make any changes

PREFIX="${PREFIX:-$(pwd)/build}"

SRC_DIR="$(dirname "${BASH_SOURCE[0]}")"
source "$SRC_DIR/env.sh"

GECKO_PKG=github.com/ava-labs/gecko
GECKO_PATH="$GOPATH/src/$GECKO_PKG"
if [[ -d "$GECKO_PATH/.git" ]]; then
    cd "$GECKO_PATH"
    go get -t -v "./..."
    cd -
else
    go get -t -v "$GECKO_PKG/..."
fi
go build -o "$PREFIX/ava" "$GECKO_PATH/main/"*.go
go build -o "$PREFIX/xputtest" "$GECKO_PATH/xputtest/"*.go

The package DOES pull down, but it also generates errors which results in an exit code being produced. On my local environment, running the build script once fails. Running it a second time passes because the library is pulled down already and no errors are generated.

The errors produced are as follows:

get "go.nanomsg.org/mangos": found meta tag get.metaImport{Prefix:"go.nanomsg.org/mangos", VCS:"git", RepoRoot:"https://github.com/nanomsg/mangos"} at //go.nanomsg.org/mangos?go-get=1
go.nanomsg.org/mangos (download)
package go.nanomsg.org/mangos/protocol/pub: unrecognized import path "go.nanomsg.org/mangos/protocol/pub" (parse https://go.nanomsg.org/mangos/protocol/pub?go-get=1: no go-import meta tags ())
package go.nanomsg.org/mangos/transport/ipc: unrecognized import path "go.nanomsg.org/mangos/transport/ipc" (parse https://go.nanomsg.org/mangos/transport/ipc?go-get=1: no go-import meta tags ())

Any suggestions?

I'm getting a panic in a client when I restart a server it is connected to:

panic: runtime error: invalid memory address or nil pointer dereference
[signal SIGSEGV: segmentation violation code=0x1 addr=0x20 pc=0x950c39]

goroutine 151 [running]:
nanomsg.org/go/mangos/v2.(*Message).Dup(0x0, 0x40c1c6)
	/Users/nick/go/pkg/mod/nanomsg.org/go/mangos/[email protected]/message.go:124 +0x29
nanomsg.org/go/mangos/v2/protocol/req.(*socket).send(0xc000251680)
	/Users/nick/go/pkg/mod/nanomsg.org/go/mangos/[email protected]/protocol/req/req.go:92 +0x189
nanomsg.org/go/mangos/v2/protocol/req.(*socket).AddPipe(0xc000251680, 0xc3e1a0, 0xc000061bc0, 0x0, 0x0)
	/Users/nick/go/pkg/mod/nanomsg.org/go/mangos/[email protected]/protocol/req/req.go:487 +0x13a
nanomsg.org/go/mangos/v2/internal/core.(*socket).addPipe(0xc00023e360, 0x7f595fadaca0, 0xc000061b60, 0xc000251920, 0x0)
	/Users/nick/go/pkg/mod/nanomsg.org/go/mangos/[email protected]/internal/core/socket.go:78 +0x176
nanomsg.org/go/mangos/v2/internal/core.(*dialer).dial(0xc000251920, 0xc000061401, 0x0, 0x0)
	/Users/nick/go/pkg/mod/nanomsg.org/go/mangos/[email protected]/internal/core/dialer.go:173 +0x337
nanomsg.org/go/mangos/v2/internal/core.(*dialer).redial(...)
	/Users/nick/go/pkg/mod/nanomsg.org/go/mangos/[email protected]/internal/core/dialer.go:215
created by time.goFunc
	/usr/local/Cellar/go/1.13.5/libexec/src/time/sleep.go:168 +0x44

Setup

• Kubernetes
• Two pods with REP socket to load balance requests
• Client REQ socket dials to headless service e.g. mystatefulset.default.svc.cluster.local:8000

Requests properly alternate between each replica in the stateful set. When one server pod is deleted, the client dumps the above panic.

• Is this the proper way to setup load balancing? Should I list the exact endpoint instead? dig would show multiple IPs for this address.
• How can I get a status log when a client connects/disconnects in the Dialer during a redial attempt?
• How would you handle removing members of a load balancing set? I only see socket.Dial(url) to connect to more.

This is an amazing project! Thank you 💯

The documentation on godoc.org for the custom import url seems to be missing packages (this page is referenced by the badge in README):

• https://godoc.org/nanomsg.org/go/mangos/v2

Godoc seems to properly detect everything when using the GitHub repo as the import path:

• https://godoc.org/github.com/nanomsg/mangos

I noticed that the go-get info page at https://nanomsg.org/go/mangos/v2?go-get=1 still references the old repository url. But I don't know whether that is the cause of the issue (go get generally works fine):

<meta name="go-import" content="nanomsg.org/go/mangos/v2 git https://github.com/nanomsg/mangos-v2">
<meta name="go-source" content="nanomsg.org/go/mangos/v2 https://github.com/nanomsg/mangos-v2 _ https://github.com/nanomsg/mangos-v2/tree/master{/dir} https://github.com/nanomsg/mangos-v2/blob/master{/dir}{file}#L{line}">

This actually affected pretty much all the protocols and both send and receive deadlines. While here we've also made sure that a non-blocking check will not fail if a message can immediately be sent or received.

We code that uses timeout options, but these APIs were designed back before context.Context existed.

It would be good to introduce APIs like

RecvContext() and RecvMsgContext() that took a context as their first argument. Arguably, the former receive functions could then be implemented in terms of these. We would then deprecate the legacy options.

This would also potentially provide a nice API for cancellation.

Forgive me if I missed a way to do this.

I know it's possible to upgrade from a "http.ResponseWriter, *http.Request" pair in Gorilla to a gorilla.Conn. I also understand that mangos uses gorilla underneath for it's websockets, right? It there a way I can upgrade "http.ResponseWriter, *http.Request" pair to a mangos.Socket somehow?

If not, should I consider writing a PR ?

If OptionRecvDeadline is set to a negative number on a sub socket (could apply to other protocols too, I did not test), calling the the Recv method will block until a data is received. According to the documentation, this should cause Recv to be a non-blocking operation.

Hello! everybody working in this great library.

I'm evaluating mangos as base library to spread load from "master" (who sends jobs ) to a lot of "agents".

Agents right now are all in the same category and any of the agents, will run any of the messages in the master queue.

master && agents Execution

without routing

• Master

go run pipeline.go master tcp://127.0.0.1:40899 

• Agent 1

go run pipeline.go agent tcp://127.0.0.1:40899  AGENT1

• Agent 2

go run pipeline.go agent tcp://127.0.0.1:40899  AGENT2

• Agent 3

go run pipeline.go agent tcp://127.0.0.1:40899  AGENT3

• Agent 4

go run pipeline.go agent tcp://127.0.0.1:40899  AGENT4

I would like to add labels to agents in any way to route messages by example by environment=production, testing and master will send messages only to the agents with its proper label. ( agents will only dequeue messages routed with its own label )

This new feature will apply on req/rep and pull/push protocols

Execution examples...

with routing

• Master

go run pipeline.go master tcp://127.0.0.1:40899 

• Agent 1

go run pipeline.go agent tcp://127.0.0.1:40899  AGENT1 env=production

• Agent 2

go run pipeline.go agent tcp://127.0.0.1:40899  AGENT2 env=production

• Agent 3

go run pipeline.go agent tcp://127.0.0.1:40899  AGENT3  env=testing

• Agent 4

go run pipeline.go agent tcp://127.0.0.1:40899  AGENT4 env=testing

PoC Code

Without Routing ( as currently is working)

I did the PoC with this code.

package main

import (
	"fmt"
	"os"
	"time"

	"go.nanomsg.org/mangos/v3"

	// register transports

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

func die(format string, v ...interface{}) {
	fmt.Fprintln(os.Stderr, fmt.Sprintf(format, v...))
	os.Exit(1)
}

func agent(url string, agentName string) {
	var sock mangos.Socket
	var err error
	var msg []byte
	if sock, err = rep.NewSocket(); err != nil {
		die("can't get new pull socket: %s", err)
	}
	if err = sock.Dial(url); err != nil {
		die("can't dial on push socket: %s", err.Error())
	}
	for {
		// Could also use sock.RecvMsg to get header
		msg, err = sock.Recv()
		fmt.Printf("AGENT [%s]: RECEIVED \"%s\"\n", agentName, msg)
		err = sock.Send([]byte(agentName))
		if err != nil {
			die("can't send reply: %s", err.Error())
		}

	}
}

func master(url string) {
	var sock mangos.Socket
	var err error

	if sock, err = req.NewSocket(); err != nil {
		die("can't get new push socket: %s", err.Error())
	}
	defer sock.Close()
	if err = sock.Listen(url); err != nil {
		die("can't listen on pull socket: %s", err.Error())
	}
	for {
		var resp []byte
		time.Sleep(time.Second / 2)
		msg := "MSG: " + time.Now().String()

		if err = sock.Send([]byte(msg)); err != nil {
			die("can't send message on push socket: %s", err.Error())
		}
		if resp, err = sock.Recv(); err != nil {
			die("can't receive resp: %s", err.Error())
		}
		fmt.Printf("MASTER: SENDED [%s] RECEIVED [%s] \n", msg, string(resp))
	}

}

func main() {
	if len(os.Args) > 2 && os.Args[1] == "agent" {
		agent(os.Args[2], os.Args[3])
		os.Exit(0)
	}
	if len(os.Args) > 3 && os.Args[1] == "master" {
		master(os.Args[2])
		os.Exit(0)
	}
	fmt.Fprintf(os.Stderr,
		"Usage: pipeline agent|master <URL> <ARG> ...\n")
	os.Exit(1)
}

With Routing in Agent (pseudocode)

func agent(url string, agentName string,environment string) {
	var sock mangos.Socket
	var err error
	var msg []byte
	if sock, err = rep.NewSocket(); err != nil {
		die("can't get new pull socket: %s", err)
	}
	if err = sock.Dial(url); err != nil {
		die("can't dial on push socket: %s", err.Error())
	}
	for {
		// Will add labels to Recv to dequeue only messages with this label.
		msg, err = sock.Recv(environment)  // <-------------CHANGE HERE
		fmt.Printf("AGENT [%s]: RECEIVED \"%s\"\n", agentName, msg)
		err = sock.Send([]byte(agentName))
		if err != nil {
			die("can't send reply: %s", err.Error())
		}

	}
}

With Routing in Master(pseudocode)

var sock mangos.Socket

func procesJobs(env string) {
var err error
 for {
		var resp []byte
		time.Sleep(time.Second / 2)
		msg := "MSG: " + time.Now().String()

		if err = sock.Send(env,[]byte(msg)); err != nil { //<-----------------CHANGED HERE
			die("can't send message on push socket: %s", err.Error())
		}
		if resp, err = sock.Recv(); err != nil {
			die("can't receive resp: %s", err.Error())
		}
		fmt.Printf("MASTER: SENDED [%s] RECEIVED [%s] \n", msg, string(resp))
	 }
}

func master(url string) {

	var err error

	if sock, err = req.NewSocket(); err != nil {
		die("can't get new push socket: %s", err.Error())
	}
	defer sock.Close()
	if err = sock.Listen(url); err != nil {
		die("can't listen on pull socket: %s", err.Error())
	}
        //jobs for production
        go procesJobs("production")
        go procesJobs("testing")
}