jeikabu / runng Goto Github PK

The methods on NngMsg take things like *const u8. For creating a msg, I have the impression we have no choice but to build the body with append for example.

The API would be much more rust idiomatic and convenient if you took &[u8] and converted to pointer internally. Also &[u8] has a len method, so the user would not have to supply a second parameter for size.

From the source code:

struct WorkQueue {
    waiting: VecDeque<oneshot::Sender<Result<NngMsg>>>,
    ready: VecDeque<Result<NngMsg>>,
}

impl WorkQueue {
    fn push_back(&mut self, message: Result<NngMsg>) {
        if let Some(sender) = self.waiting.pop_front() {
            sender
                .send(message)
                .unwrap_or_else(|err| debug!("Dropping message: {:?}", err));
        } else {
            self.ready.push_back(message);
        }
    }

    fn pop_front(&mut self) -> AsyncMsg {
        // If a value is ready return it immediately.  Otherwise
        if let Some(item) = self.ready.pop_front() {
            Box::pin(future::ready(item))
        } else {
            let (sender, receiver) = oneshot::channel();
            self.waiting.push_back(sender);
            let receiver = receiver.map(result::flatten_result);
            Box::pin(receiver)
        }
    }
}

When you read from a socket. The pop_front function is called. Since there is no message ready, an oneshot channel is registered. If the read is cancelled (AsyncMsg is dropped), the oneshot stays registered.
Eventually, when a message actually is received, it will be sent to that invalid channel. The send will fail and the message will be dropped.

A simple workaround would be to put the body of the push_back within a loop statement. Trying to send the message until a valid channel is reached or until it is placed into the ready queue.

Cancelling a read is useful when you use the select! macro.

Provide wrapper for nng_url_* functions (details)
Implement stream alloc_url functions.

Right now if a struct contains for example a protocol::pair1::Pair1 it can't derive Debug because Pair1 doesn't implement it. It is generally accepted to more or less always derive or implement at least Debug for all pub structs.

Others that might be useful can be found in the rust book. Eq might also make sense for a socket. For the difference between Eq and PartialEq, see here.

Some traits are handy if it makes sense, like Copy, but they have implications for the future. If you later add a non-Copy property to your struct, you will have to break the API by removing Copy and instances will be implicitly copied on the stack which might not always be a good idea. However Clone is explicit and might be a good idea.

Bus

I would like to use runng stream as a transport for Remoc (https://crates.io/crates/remoc). Remoc supports two kind of interfaces, Sink/Stream or AsyncRead/AsyncWrite. See https://docs.rs/remoc/0.9.7/remoc/struct.Connect.html.

Would it be possible to support one of these with runng stream?

I have been looking at nng-rs and runng. What are the main differences?

Am I correct to state:

• nng-rs is a rust safe implementation that follows closely the nng-sys / nng API
• runng adds further logic, including async patterns, to best support Aio in NNG in Rust?

Is it possible to create a long running Aio reusing the same callback, to be used for workers, like in the example for nng-rs? See
https://github.com/neachdainn/nng-rs/blob/master/examples/async.rs

My understanding of runng leads me to believe that Aio's are one time use only, and cannot be rescheduled to do another receive. Is that correct, or is there a way?

Pin/Unpin are in 1.33.
https://doc.rust-lang.org/std/pin/index.html

Runng and Nng-rs currently use different crates for linking to the NNG library. If my understanding is correct, that means that anyone who has both Nng-rs and Runng in their dependency tree will have linker issues¹. I think it would be wise for us to pick a single "sys" crate that we both use. I don't particularly care if we use your code or my code but we should probably utilize the nng-sys crate name to match the community style.

Nng-rs repository

1: I was going to run a quick test of this but I ran into issues compiling Runng. It's probably something wrong with my environment - I didn't investigate because I don't think it impacts the point I'm trying to make.

Code like:

let pusher = factory.pusher_open()?.listen(&url)?;
    let puller = factory.puller_open()?.dial(&url)?;

    thread::spawn(move || -> NngReturn {
        let mut push_ctx = pusher.create_async_context()?;
        loop {
            let msg = msg::NngMsg::new()?;
            push_ctx.send(msg).wait().unwrap()?;
        }
        
        Ok(())
    });

    thread::spawn(move || -> NngReturn {
        let mut pull_ctx = puller.create_async_context()?;
        loop {
            pull_ctx.receive().take(1).wait();
        }
        
        Ok(())
    });

Will crash with:

pushpull_tests-7035aebbdce1069e(57332,0x700001ce8000) malloc: *** error for object 0xf: pointer being freed was not allocated
pushpull_tests-7035aebbdce1069e(57332,0x700001ce8000) malloc: *** set a breakpoint in malloc_error_break to debug

The problem is the rapid creation of the receive future:

loop {
    pull_ctx.receive() // this
}

Can be avoided by only creating the future once:

// No `loop{}`
pull_ctx.receive()
            .for_each(|_|{
                // Do stuff
                Ok(())
            }).wait().unwrap();

nanomsg/nng@5cf7506

std::future::Future was stabilised in Rust 1.36

reqrep_tests::nonblock fails, especially on Windows. Using NONBLOCK but doing reqrep can receive ESTATE error:

thread 'tests::reqrep_tests::nonblock' panicked at 'ESTATE', runng\tests\tests\reqrep_tests.rs:116:25

Either need to add appropriate blocking or change protocol.

Surveyor
Respondent

I see that structs like protocol::pair1::Pair1 have methods like dial and listen, but not close?

I wondered about this after running into this problem. Messages get dropped because the sender shuts down before they have finished sending. I wasn't using threads, but when sleeping 500ms before shutting the program down I can't reproduce the problem anymore, so I suppose it's what happens.

I wondered if a close method could flush the connection and then disconnect? And if that could prevent problems like this.

Look at replacing the channels being used in 0.3 (dev branch) with custom Future/Waker.

It would be nice to be able to use the ? operator in functions that can also throw other errors than NngFail. The most recent work on making error handling smoother in rust is the failure crate. It would be nice to provide compatibility with that. It can be done by deriving failure::Fail for your error, or if you don't want the extra dependency, implementing std::error::Error as described here in the docs.

If you're low on time I could make a pull request for this one, but you'd have to decide whether you prefer std::error or failure.

btw off topic: I am trying to see how I will use runng and actix in my project, so I made a small test repository where I try different ways to get it to work. It might at some point serve as example code for runng. Currently it's not very documented, but just wanted to let you know it exists. Every commit is a different working example. It will definitely evolve over the next few days, because it's not at all satisfying right now. Next I will look into the async api and improve the way I invoke actix.

I'm evaluating runng for a small project. I'm on macOS. When I do a fresh compilation of my project, everything is fine. Now, when I'm trying to compile a second time, I get this error:

$ cargo run  
error: checksum for `runng-sys v1.1.1+4` changed between lock files

this could be indicative of a few possible errors:

    * the lock file is corrupt
    * a replacement source in use (e.g. a mirror) returned a different checksum
    * the source itself may be corrupt in one way or another

unable to verify that `runng-sys v1.1.1+4` is the same as when the lockfile was generated

And it doesn't compile

IoVec is being moved to std

nng_iov is used by the upcoming "stream" API (e.g. nng_stream_send()).

nanomsg/nng@5803db0