nanomsg / mangos

mangos is a pure Golang implementation of nanomsg's "Scalablilty Protocols"

Welcome to nanomsg

The nanomsg library is a simple high-performance implementation of several "scalability protocols". These scalability protocols are light-weight messaging protocols which can be used to solve a number of very common messaging patterns, such as request/reply, publish/subscribe, surveyor/respondent, and so forth. These protocols can run over a variety of transports such as TCP, UNIX sockets, and even WebSocket.

For more information check the website.

Prerequisites

1. Windows.

   • Windows Vista or newer (Windows XP and 2003 are NOT supported)
   • Microsoft Visual Studio 2010 (including C++) or newer, or mingw-w64. (Specifically mingw and older Microsoft compilers are NOT supported, and we do not test mingw-w64 at all, so YMMV.)
   • CMake 2.8.12 or newer, available in $PATH as cmake

2. POSIX (Linux, MacOS X, UNIX)

   • ANSI C compiler supporting C89
   • POSIX pthreads (should be present on all modern POSIX systems)
   • BSD sockets support for both TCP and UNIX domain sockets
   • CMake (http://cmake.org) 2.8.12 or newer, available in $PATH as cmake

3. Documentation (optional)

   • asciidoctor (http://asciidoctor.org/) available as asciidoctor
   • If not present, docs are not formatted, but left in readable ASCII
   • Available on-line at http://nanomsg.org/documentation

Quick Build Instructions

These steps here are the minimum steps to get a default Debug build. Using CMake you can do many other things, including setting additional variables, setting up for static builds, or generation project or solution files for different development environments. Please check the CMake website for all the various options that CMake supports.

POSIX

This assumes you have a shell in the project directory, and have the cmake and suitable compilers (and any required supporting tools like linkers or archivers) on your path.

1. % mkdir build
2. % cd build
3. % cmake ..
4. % cmake --build .
5. % ctest .
6. % sudo cmake --build . --target install
7. % sudo ldconfig (if on Linux)

Windows

This assumes you are in a command or powershell window and have the appropriate variables setup to support Visual Studio, typically by running vcvarsall.bat or similar with the appropriate argument(s). It also assumes you are in the project directory.

1. md build
2. cd build
3. cmake ..
4. cmake --build . --config Debug
5. ctest -C Debug .
6. cmake --build . --config Debug --target install NB: This may have to be done using an Administrator account.

Alternatively, you can build and install nanomsg using vcpkg dependency manager:

1. git clone https://github.com/Microsoft/vcpkg.git
2. cd vcpkg
3. ./bootstrap-vcpkg.bat
4. ./vcpkg integrate install
5. ./vcpkg install nanomsg

The nanomsg port in vcpkg is kept up to date by microsoft team members and community contributors. If the version is out of date, please create an issue or pull request on the vcpkg repository.

Static Library

We normally build a dynamic library (.so or .DLL) by default.

If you want a static library (.a or .LIB), configure by passing -DNN_STATIC_LIB=ON to the first cmake command.

POSIX

POSIX systems will need to link with the libraries normally used when building network applications. For some systems this might mean -lnsl or -lsocket.

Windows

You will also need to define NN_STATIC_LIB in your compilation environment when building programs that use this library. This is required because of the way Windows changes symbol names depending on whether the symbols should be exported in a DLL or not.

When using the .LIB on Windows, you will also need to link with the ws2_32, mswsock, and advapi32 libraries, as nanomsg depends on them.

Support

This library is considered to be in "sustaining" mode, which means that new feature development has ended, and bug fixes are made only when strictly necessary for severe issues.

New development is now occurring in the NNG project, which offers both protocol and API compatibility with this project. Please consider using NNG for new projects.

Please see the file SUPPORT for more details.

mangos's Issues

Flogo examples

Hi guys,

Hope you are all well !

It would be awesome to create some examples (surveyor, pub-sub) or a wrapper for mangos for Flogo and Mashling. It would make easier to create flows and pipelines scenarios. And more people would give a try for an implementation of nanomsg/mangos I guess.

Official Websites

• http://www.flogo.io/
• https://www.mashling.io/

Mashling:

• https://github.com/TIBCOSoftware/mashling
• https://www.mashling.io/home

Docker - Flogo web

• https://hub.docker.com/r/flogo/flogo-docker/
• https://github.com/retgits/dockerfiles/tree/master/flogoweb

Flogo - Components

• https://github.com/retgits/flogo-components/tree/master/activity
• https://github.com/TIBCOSoftware/flogo-contrib/tree/master/activity
• https://github.com/retgits/flogo-components/tree/master/trigger

Flogo - Contrib Example

• NSQ producer https://github.com/abasse/folgonsqproducer
• NSQ trigger https://github.com/abasse/flogonsqtrigger
• Stream of UDP data https://github.com/ayh20/flogo-components/tree/master/trigger/udp

What do you think about that ? some basic "hello world" examples would be just great ^^

Cheers,
Rosco

coverage build is busted

The coverage test incorrectly returns success on CircleCI even if it fails:

warning: no packages being tested depend on matches for pattern nanomsg.org/go-mangos/...
ok nanomsg.org/go/mangos/v2/transport/ipc 0.204s coverage: 0.0% of statements in nanomsg.org/go-mangos/...
Doing test in ./transport/inproc
warning: no packages being tested depend on matches for pattern nanomsg.org/go-mangos/...
ok nanomsg.org/go/mangos/v2/transport/inproc 0.204s coverage: 0.0% of statements in nanomsg.org/go-mangos/...
Doing test in ./transport/tlstcp
warning: no packages being tested depend on matches for pattern nanomsg.org/go-mangos/...
panic: interface conversion: interface {} is int64, not int

goroutine 71 [running]:
nanomsg.org/go/mangos/v2/transport.NewConnPipe(0x6a92a0, 0xc4200dc380, 0x310030, 0x67a923, 0x3, 0x67a920, 0x3, 0xc4201d7bb0, 0x1, 0x1, ...)
	/home/circleci/go/src/nanomsg.org/go/mangos/v2/transport/conn.go:152 +0x474
nanomsg.org/go/mangos/v2/transport/tlstcp.(*dialer).Dial(0xc4200a86c0, 0x67bbf9, 0xa, 0x664280, 0xc420001800)
	/home/circleci/go/src/nanomsg.org/go/mangos/v2/transport/tlstcp/tlstcp.go:134 +0x43e
nanomsg.org/go/mangos/v2/test.(*TranTest).TestListenAndAccept.func1(0xc420462da0, 0xc4204e00f0, 0xc4200d2140)
	/home/circleci/go/src/nanomsg.org/go/mangos/v2/test/transport.go:92 +0x1db
created by nanomsg.org/go/mangos/v2/test.(*TranTest).TestListenAndAccept
	/home/circleci/go/src/nanomsg.org/go/mangos/v2/test/transport.go:78 +0x2df
FAIL	nanomsg.org/go/mangos/v2/transport/tlstcp	0.582s

Add logging to ease debugging connection reattempts

This issue is imported from: #252

Essentially we need to have better logging for things like TLS errors.

Windows build broken

go get -t ./...
# nanomsg.org/go/mangos/v2/transport
transport\connipc_windows.go:36:23: too many arguments in call to p.conn.handshake
	have ([]interface {})
	want ()
transport\connipc_windows.go:85:15: undefined: ErrTooLong
Command exited with code 2

Applications should not have to call socket.AddTransport()

For most stock transports, it should be sufficient for an application to import the package (with a "_" import) to register the transport. There is really little use in having explicit transport registration for each socket.

The AddTransport() API will still need to be available for user supplied transports though.

WS and WSS should permit empty path

An empty websocket path is the same same as "/" per the HTTP RFC. We should canonicalize the URL as specified.

Port IsClient/IsServer redundant?

These APIs add little value, since we can just test against the listener or dialer. Just remove it.

Consider "DialContext" to match other golang net pkgs

It might be nice to have a Context that can be used to coordinate synchronous Dialing (done in a goroutine). Now that golang's context.Context is pretty much mainstream Go, we should look at using it more.

Abort Req retries on timeout

When the Recv deadline fires, we don't stop the underlying Req socket from continuing to retry requests behind the scenes. We should fix that -- probably we need to add an Reset() method to the protocols to reset them to their initial state...

pubsub: Filter messages on the sender side

Originally: https://bitbucket.org/gdamore/mangos/issue/18/pubsub-filter-messages-on-the-sender-side

Ondrej Kupka created an issue 2014-04-27
Just checked the source code for pubsub out of curiosity and as far as I understand it, the messages are filtered on the receiver side. If that is true, I would propose to use something like go-patricia (radix tree written in Go) to filter messages on the sender side.
Actually I am the one who wrote go-patricia, needed message filtering as well :-)
Regards, Ondrej Kupka
Comments (4)
Garrett D'Amore
This is an interesting idea; but note that sender filtering is not part of nanomsg itself. So we'd need to have some new protocol exchange for the subscriber to tell the publisher what it wants. I propose that this could be done using a simple message format along the lines of + to add a sub, - to remove a sub, and no notification from the publisher. (The subscriber would intrinsically know.) I'll discuss this on the mailing list.
Edit Mark as spam Delete 2014-05-03
Ondrej Kupka
Sure. Hmm, I thought that nanomsg is doing sender-side filtering...
Mark as spam Delete 2014-05-03
Ondrej Kupka
It is true that this complicates things, of course. You need to handle subscriptions, you need to clean them up when the subscriber is not there any more etc. But in any case I guess that you could use something else than just a list of subscriptions in mangos, perhaps this trie. But that is also questionable, I am wondering how many subscriptions you would have to manage for the subscriptions list to become a bottleneck...
Mark as spam Delete 2014-05-03
Garrett D'Amore
I sent a message to the mailing list detailing my thoughts on this. There are some non-trivial challenges. The biggest problem is ensuring that you handle subscription requests reliably -- you cannot simply have a best-effort solution like we have today.

Basically, until we get a standard protocol for this in nanomsg, we can't do this. We could invent a protocol or pattern only for our use, but that seems of limited value.

Add QUIC transport

Originally filed here; https://github.com/nanomsg/mangos/issues/308

lthibault did this very nice package, which will need adaptation for mangos v2.

The underlying transport provider API needs to stabilize first though.

Remove stale msg.Expired

We are no longer doing this expiration stuff anymore, so we should remove that from the Message API, and from the transports.

Changing queue depths should just discard queue contents

We have a bunch of logic intended to try to keep queue contents when a program changes the queue depth. This is silly.

Normally this is only done when the socket is first created. This is extra complexity that is not needed.

Furthermore, the new queue size might be smaller anyway, so queue contents can be dropped as a result of that. And generally any of the queues are in "best effort" cases anyway, where it is perfectly fine to drop messages on the floor.

Statistics desired

It may be nice to have methods to track statistics. There are probably a great number of potentially useful stats. We need to think about them.

Coverage test does not seem to exercise transport tests

I'm seeing that the transport tests don't seem to be included in our test run... investigation needed.

Update warnings about release

We are in a pre-release state, but once #3 is integrated, we will have achieved the main API breaking changes we want, and it will be time to tag.

There are other internal improvements to make still, but as those won't be breaking, we can probably tag a v2 release soon as well.

Ditch gorilla websocket

See also from mangos v1: #211

Probably we want to just switch to using the standard library now.

CircleCI working directory busted

The working directory is not ~ for circleci 2.0, which means we wind up having mismatches between what we check out and what we test.

This also is responsible for the confusion calculating coverage.

Use net.Buffers

See also #133

We can use go 1.9 stuff now, and net.Buffers would give a nice boost for TCP users.

Checking for compatible peer protocols MIA?

We will apparently happily allow incompatible peer protocols to connect and add to the socket.

This is a major oops!

add circleci link

Exclude examples and perf from coverage

The perf and examples code shouldn't be part of our coverage, because we do nothing with them in the test runs.

Failure in websocket test

There's a race, where the http server might not have started by the time we start trying to connect to it.

Eliminate stale GetID() in impl/pipe.go

surveyor busted on AppVeyor

Builds on AppVeyor seem to be failing in the Surveyor pattern.

Probably there is something special about the Windows scheduling order that is causing this; we're seeing very late message arrivals.

Revise PLANS into TODO.

Our PLANS were notes for 2.0. We've accomplished the big stuff, so lets turn this into a TODO list. Or possibly we should just make issues for this stuff.

Go Report Card has numerous minor issues

We have a bunch of minor issues, which we should address (ineff assign, fmt, etc.)

Coverage results are still deficient

The coverage analysis is woefully incomplete. It would be best to run a top-level go test I think.

Want pairv1

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.

Refactor the PortHook API.

PortHooks were horrible.

These are really PipeEvents, and we should also ditch the Endpoint legacy while we're at it because mangos Endpoints have nothing in common with Endpoints as used in nanomsg or NNG.

We might want to have PipeEvent event support for both pre-add, and post-add, as well as post remove, to make this more like NNG.

Debug TLS issue?

I've been using rep/req over TCP just fine, and am trying to switch to tls+tcp, but the req (client) end can't Recv.

I do something like (err handling skipped for brevity, but no errors generated except at the end):

sock, _ := req.NewSocket()
sock.SetOption(mangos.OptionMaxRecvSize, 0)
sock.SetOption(mangos.OptionRecvDeadline, timeout)
sock.AddTransport(tlstcp.NewTransport())
tlsConfig := &tls.Config{}
caCert, _ := ioutil.ReadFile(caFile)
certPool := x509.NewCertPool()
certPool.AppendCertsFromPEM(caCert)
tlsConfig.RootCAs = certPool
// tlsConfig.InsecureSkipVerify = true
dialOpts := make(map[string]interface{})
dialOpts[mangos.OptionTLSConfig] = tlsConfig
sock.DialOptions("tls+tcp://localhost:11301", dialOpts)
[...]
resp, err := sock.Recv()

With InsecureSkipVerify line uncommented it works, but with it commented out, the Recv() err is a generic receive time out. How do I get more information on what is really going wrong?

The rep (server) end seems fine. I do something like (again, err handling excluded):

sock, _ := rep.NewSocket()
sock.SetOption(mangos.OptionMaxRecvSize, 0)
sock.SetOption(mangos.OptionRaw, true)
sock.SetOption(mangos.OptionRecvDeadline, ServerInterruptTime)
sock.AddTransport(tlstcp.NewTransport())
cer, _ := tls.LoadX509KeyPair(serverPemFile, serverKeyFile)
tlsConfig := &tls.Config{Certificates: []tls.Certificate{cer}}
listenOpts := make(map[string]interface{})
listenOpts[mangos.OptionTLSConfig] = tlsConfig
sock.ListenOptions("tls+tcp://0.0.0.0:11301", listenOpts)

I create the certs like this:

func GenerateCerts(caFile, serverPemFile, serverKeyFile string) error {
	// key for root CA
	rootKey, err := rsa.GenerateKey(rand.Reader, 2048)
	if err != nil {
		return err
	}

	// cert for root CA, self-signed
	rootCertTmpl, err := certTemplate()
	if err != nil {
		return err
	}
	rootCertTmpl.IsCA = true
	rootCertTmpl.KeyUsage = x509.KeyUsageCertSign | x509.KeyUsageDigitalSignature
	rootCertTmpl.ExtKeyUsage = []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth, x509.ExtKeyUsageClientAuth}
	rootCert, err := createCert(rootCertTmpl, rootCertTmpl, &rootKey.PublicKey, rootKey, caFile)
	if err != nil {
		return err
	}

	// key for server
	serverKey, err := rsa.GenerateKey(rand.Reader, 2048)
	if err != nil {
		return err
	}

	// cert for server, signed by root CA
	servCertTmpl, err := certTemplate()
	if err != nil {
		return err
	}
	servCertTmpl.KeyUsage = x509.KeyUsageDigitalSignature
	servCertTmpl.ExtKeyUsage = []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth}
	_, err = createCert(servCertTmpl, rootCert, &serverKey.PublicKey, rootKey, serverPemFile)
	if err != nil {
		return err
	}

	// store the server's key
	keyOut, err := os.OpenFile(serverKeyFile, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0600)
	if err != nil {
		return err
	}
	err = pem.Encode(keyOut, &pem.Block{
		Type: "RSA PRIVATE KEY", Bytes: x509.MarshalPKCS1PrivateKey(serverKey),
	})
	if err != nil {
		return err
	}
	return keyOut.Close()
}

// certTemplate creates a certificate template with a random serial number,
// valid from now until validFor.
func certTemplate() (*x509.Certificate, error) {
	serialNumberLimit := new(big.Int).Lsh(big.NewInt(1), 128)
	serialNumber, err := rand.Int(rand.Reader, serialNumberLimit)
	if err != nil {
		return nil, errors.New("failed to generate serial number: " + err.Error())
	}

	template := x509.Certificate{
		SerialNumber:          serialNumber,
		Subject:               pkix.Name{Organization: []string{"my org"}},
		SignatureAlgorithm:    x509.SHA256WithRSA,
		NotBefore:             time.Now(),
		NotAfter:              time.Now().Add(validFor),
		IPAddresses:           []net.IP{net.ParseIP("0.0.0.0"), net.ParseIP("127.0.0.1")},
		DNSNames:              []string{"localhost"},
		BasicConstraintsValid: true,
	}
	return &template, nil
}

// createCert creates a certificate given a template, signing it against its
// parent, and saving the cert in PEM format to certPath.
func createCert(template, parentCert *x509.Certificate, publicKey interface{}, parentPrivateKey interface{}, certPath string) (*x509.Certificate, error) {
	certDER, err := x509.CreateCertificate(rand.Reader, template, parentCert, publicKey, parentPrivateKey)
	if err != nil {
		return nil, err
	}

	// parse the resulting certificate so we can use it again
	cert, err := x509.ParseCertificate(certDER)
	if err != nil {
		return nil, err
	}

	// save in PEM format
	b := pem.Block{Type: "CERTIFICATE", Bytes: certDER}
	certOut, err := os.Create(certPath)
	if err != nil {
		err = fmt.Errorf("creation of certificate file [%s] failed: %s", certPath, err)
		return nil, err
	}
	err = pem.Encode(certOut, &b)
	if err != nil {
		return nil, err
	}
	err = certOut.Close()
	return cert, err
}

Any thoughts on how I can get it to work?

Dead code in dialer

The dialer implementation has an "xredial" function still hanging around.

Add PAIRv1 protocol + Polyamorous mode

NNG has a PAIRv1 protocol that supports what I'm calling "Polyamorous" mode. This protocol lets us get a 1:N communication pattern.

Remove protocol.ProtocolName()

We have other ways to get this, so we don't need a global map for this anymore.

Add goreport card and godocs badges.

These work now, so lets restore them.

Dial should be synchronous by default

In NNG, we have elected to make Dial() operations synchronous by default, at least on their first attempt. We find that this is superior for debugging and diagnosability, and most applications actually prefer this.

Request: Example with TLS Transport and PAIR Protocol

Hello!

I've been reading through the documentation and various examples extensively, but have been having trouble getting an example working with using TLS over TCP as the transport and PAIR as the protocol. Would it be possible to provide a simple example?

Cheers

Add logging to ease debugging connection reattempts

Background: I was setting up tls+tcp connection with mutual authentication. I didn't realize that I needed to provide a ServerName in the tls.Config, so mangos kept trying to reconnect, but I just saw this as a timeout occurring from my code. I eventually had to throw in print statements throughout core.go to find the cause.

I propose adding a very simple logging solution.

My idea is to let users register loggers (like how database/sql registers drivers). A default one that logs to stderr could be provided in a mangos subpackage. I'll submit a PR with a proof of concept in just a minute. Using it would look like this:

// Not a working example
package mangos_test

import (
        "crypto/tls"
        "crypto/x509"
        "log"

        "github.com/go-mangos/mangos"
        "github.com/go-mangos/mangos/protocol/push"
        "github.com/go-mangos/mangos/transport/tlstcp"

        _ "github.com/go-mangos/mangos/log"
)

var (
        url  = "localhost:5555"
        conf = &tls.Config{
                Certificates: []tls.Certificate{},
                RootCAs:      x509.NewCertPool(),
        }
)

func init() {
        mangos.SetLogLevel(2)
}

func Example() {
        sock, err = push.NewSocket()
        if err != nil {
                log.Fatalf("can't get new push socket: %v", err)
        }
        sock.AddTransport(tlstcp.NewTransport())
        if err = sock.DialOption("tls+tcp://"+url, conf); err != nil {
                log.Fatalf("can't dial on push socket: %v", err)
        }
        defer sock.Close()

        if err = sock.Send([]byte("Hello world!")); err != nil {
                log.Fatalf("can't send message on push socket: %v", err)
        }
}

Examples need to be updated with import paths

Kill off the impl.pipe.IsOpen

Nothing is using this anymore -- but maybe some tests. Review and clean this up.

Statistics desired

See also #67

We can roll our own here, but there may be 3rd party stats packages we want to use.

tlstcp transport busted

goroutine 55 [running]:
nanomsg.org/go/mangos/v2/transport.NewConnPipe(0x6c60a0, 0xc00048ce00, 0x310030, 0x6861ad, 0x3, 0x6861aa, 0x3, 0xc0003fbba8, 0x1, 0x1, ...)
/go/src/nanomsg.org/go/mangos/v2/transport/conn.go:152 +0x463
_/home/circleci/go/src/nanomsg.org/go/mangos/v2/transport/tlstcp.(*dialer).Dial(0xc0000a2d00, 0x687481, 0xa, 0x671240, 0xc000001980)
/home/circleci/go/src/nanomsg.org/go/mangos/v2/transport/tlstcp/tlstcp.go:134 +0x419
nanomsg.org/go/mangos/v2/test.(*TranTest).TestListenAndAccept.func1(0xc000456e20, 0xc000232100, 0xc0000e0050)
/go/src/nanomsg.org/go/mangos/v2/test/transport.go:92 +0x1db
created by nanomsg.org/go/mangos/v2/test.(*TranTest).TestListenAndAccept
/go/src/nanomsg.org/go/mangos/v2/test/transport.go:78 +0x2df
FAIL _/home/circleci/go/src/nanomsg.org/go/mangos/v2/transport/tlstcp 0.678s

Figure out appveyor again

We used to have mangos testing under appveyor. We should set up v2 for appveyor coverage on Windows as well.

Protocol Info() could be global constants.

This would eliminate some function calls, and also eliminate code that is not called.

transport: add UDP support?

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.

UDP transport

Originally filed as #215

There was a lot of discussion there, but ultimately I believe we want a UDP transport at some time.

Context demos needed

We have this cool new context API, but there are no demo programs!

Probably we should follow the example of NNG here.

Move the "impl" package to an "internal" package

Modern go has an "internal" layout to hide implementation details. We should use that!

Port RemoteProtocol and LocalProtocol are mostly useless

We aren't using this stuff anywhere, and I can't think of any real need for it. Lets remove it.

Use asciidoc instead of markdown for docs