{ kind: Interrupted, error: "Mailbox has closed" } #7

xudesheng · 2020-03-06T05:10:37Z

First step, I added a loop to continuously publish message and it works:
`
let mut total_count = 0;
let mut rng = rand::thread_rng();

            loop {
                total_count += 1;
                if total_count>1000 {break;}

                println!("Publish: {}", total_count);
                let payload = json!({
                    "temperature": 707 + rng.gen_range(0, 70),
                    "humidity": 202 + rng.gen_range(0, 20),
                });

                let payload = serde_json::to_string(&payload).unwrap();
                println!("payload:{}", payload);
                client
                    .publish(
                        //String::from("test"),
                        TOPIC.to_owned(),
                        QualityOfService::Level0,
                        Vec::from(payload.as_bytes()),
                    )
                    .await?;
                println!("Sending");
                println!("Wait for 5s");
                let delay_time = Instant::now() + Duration::new(1, 0);
                delay_until(delay_time).await;
                println!("Sent successfully, delayed 5s");
            }

`

second step is to simulate multiple client by adding a for loop

`
for client in clients {

let future = ...

   let result = async move {

       new mqtt-client

       loop {sending msg}

   }

Arbiter::spawn(future);

}

`

After adding a for loop, all clients will only be able to send out first message and then Actor will stop. Error message is Custom { kind: Interrupted, error: "Mailbox has closed" }. I couldn't figure out why Actor was stopped.

Appreciate your help.

The text was updated successfully, but these errors were encountered:

Syndim · 2020-03-06T12:53:44Z

Hi, I did a quick test on my side but I can't repro this issue. Can you provide more info about:

Which OS are you running?
How did you create multiple clients? Do you just clone the client or connect to the MQTT server for multiple times?

In the mainwhile I created 2 tests:
This test clones the MQTT client for 5 times and send the random packet

actix-mqtt-client/src/lib.rs

Line 305 in fe77cf0

fn test_random_publish_level0_cloned_client() {

This test connects to the MQTT server(I'm using mosquitto) for 5 times and send random packet

actix-mqtt-client/src/lib.rs

Line 410 in fe77cf0

fn test_random_publish_level0_created_client() {

Both tests are running fine on my side(MacOS 10.15.3), can you also try to run those tests on your side and see if it works?

xudesheng · 2020-03-06T14:15:32Z

Thank you for your quick response:

I'm on Mac 10.15
I created new client, instead of clone. I print out peer_addr and local_addr as:
devicemqtt0001,Remote:V4(104.46.115.xxx:8883), Local:V4(192.168.1.9:57784) devicemqtt0000,Remote:V4(104.46.115.xxx:8883), Local:V4(192.168.1.9:57785)
Because I have to simulate client pressure, so I want to use more clients instead of clone.

appreciate you shared your test code.

let me follow your pattern exactly and try it.

xudesheng · 2020-03-06T14:56:47Z

let mut client = MqttClient::new( r, w, format!("test_{}", client_id), MqttOptions::default(), MessageActor(**sender.clone()**).start().recipient(), ErrorActor.start().recipient(), None, );
I feel this is the only difference: sender.clone(). I don't understand why it needed here and I couldn't find it out from doc. may I have missed anything?

after look at your code more closely, I understood why you add this sender, but it seems like my issue will not be impacted by this one.

let me install a broker in my local computer and test your code.

xudesheng · 2020-03-06T15:56:09Z

running 4 tests
test random_test::test_random_publish_level0_cloned_client ... FAILED
test tests::test_client ... FAILED
test random_test::test_random_publish_level2 ... FAILED
test random_test::test_random_publish_level0_created_client ... test random_test::test_random_publish_level0_created_client has been running for over 60 seconds

Your test code seems like failed on my local.

message from mosquitto console:
1583510015: New client connected from 127.0.0.1 as test_0 (p2, c0, k0). 1583510015: New client connected from 127.0.0.1 as test_1 (p2, c0, k0). 1583510015: New client connected from 127.0.0.1 as test_2 (p2, c0, k0). 1583510015: New client connected from 127.0.0.1 as test_3 (p2, c0, k0). 1583510015: New client connected from 127.0.0.1 as test_4 (p2, c0, k0).

xudesheng · 2020-03-06T18:59:54Z

if I only run test_client case, it's success.

cargo test test_client
Finished test [unoptimized + debuginfo] target(s) in 0.12s
Running target/debug/deps/actix_mqtt_client-21c06d40e896069d

running 1 test
test tests::test_client ... ok

test result: ok. 1 passed; 0 failed; 0 ignored; 0 measured; 3 filtered out

but if I only run `cargo test random_test', it will fail:

cargo test random_test
Finished test [unoptimized + debuginfo] target(s) in 0.10s
Running target/debug/deps/actix_mqtt_client-21c06d40e896069d

running 3 tests
test random_test::test_random_publish_level2 ... FAILED
test random_test::test_random_publish_level0_cloned_client ... FAILED
test random_test::test_random_publish_level0_created_client ... test random_test::test_random_publish_level0_created_client has been running for over 60 seconds

Syndim · 2020-03-07T09:04:41Z

The random tests are actually infinite loops which sends the random packets forever so they won't actually "pass" the tests :).

Looks like you are doing exactly the same as test_random_publish_level0_created_client is doing. Can you try to run the test by:

RUST_LOG=info cargo test test_random_publish_level0_created_client -- --nocapture

And see if there are any error logs? If everything goes fine you will only get "INFO" logs like this:

[2020-03-07T08:58:47Z INFO  actix_mqtt_client::random_test] Got message: id:0, topic: test, payload: [226, 104, 107, 32, 142, 174, 59, 194, 156, 86, 244, 56, 129, 94, 60, 62, 131, 60, 122, 247, 12, 187, 214, 64, 128, 176, 1, 119, 1, 139, 150, 66]
[2020-03-07T08:58:47Z INFO  actix_mqtt_client::random_test] Got message: id:0, topic: test, payload: [87, 50, 65, 193, 210, 48, 17, 234, 167, 240, 219, 169, 173, 46, 207, 207, 149, 234, 103, 177, 125, 115, 105, 113, 156, 174, 160, 96, 213, 176, 141, 129]
[2020-03-07T08:58:47Z INFO  actix_mqtt_client::random_test] Pending recv items: 0
[2020-03-07T08:58:47Z INFO  actix_mqtt_client::random_test] Pending recv items: 0
[2020-03-07T08:58:47Z INFO  actix_mqtt_client::actors::packets::publish] Handle message for RecvPublishActor
[2020-03-07T08:58:47Z INFO  actix_mqtt_client::actors::packets::publish] Handle message for RecvPublishActor

If you find any error logs please paste them here and I can take a look.

Or if you can share the full code snippet which you use to connect to local MQTT server I can also check what's happening.

xudesheng · 2020-03-07T16:30:02Z

The random tests are actually infinite loops which sends the random packets forever so they won't actually "pass" the tests :).

Looks like you are doing exactly the same as test_random_publish_level0_created_client is doing. Can you try to run the test by:
RUST_LOG=info cargo test test_random_publish_level0_created_client -- --nocapture
And see if there are any error logs? If everything goes fine you will only get "INFO" logs like this:
[2020-03-07T08:58:47Z INFO  actix_mqtt_client::random_test] Got message: id:0, topic: test, payload: [226, 104, 107, 32, 142, 174, 59, 194, 156, 86, 244, 56, 129, 94, 60, 62, 131, 60, 122, 247, 12, 187, 214, 64, 128, 176, 1, 119, 1, 139, 150, 66]
[2020-03-07T08:58:47Z INFO  actix_mqtt_client::random_test] Got message: id:0, topic: test, payload: [87, 50, 65, 193, 210, 48, 17, 234, 167, 240, 219, 169, 173, 46, 207, 207, 149, 234, 103, 177, 125, 115, 105, 113, 156, 174, 160, 96, 213, 176, 141, 129]
[2020-03-07T08:58:47Z INFO  actix_mqtt_client::random_test] Pending recv items: 0
[2020-03-07T08:58:47Z INFO  actix_mqtt_client::random_test] Pending recv items: 0
[2020-03-07T08:58:47Z INFO  actix_mqtt_client::actors::packets::publish] Handle message for RecvPublishActor
[2020-03-07T08:58:47Z INFO  actix_mqtt_client::actors::packets::publish] Handle message for RecvPublishActor
If you find any error logs please paste them here and I can take a look.

Or if you can share the full code snippet which you use to connect to local MQTT server I can also check what's happening.

Thank you. I didn't find any error log.

let me clean up my code and narrow down issue scope. if issue persists, I will send to you and appreciate you help in advance.

xudesheng · 2020-03-07T19:03:45Z

The random tests are actually infinite loops which sends the random packets forever so they won't actually "pass" the tests :).

Looks like you are doing exactly the same as test_random_publish_level0_created_client is doing. Can you try to run the test by:
RUST_LOG=info cargo test test_random_publish_level0_created_client -- --nocapture
And see if there are any error logs? If everything goes fine you will only get "INFO" logs like this:
[2020-03-07T08:58:47Z INFO  actix_mqtt_client::random_test] Got message: id:0, topic: test, payload: [226, 104, 107, 32, 142, 174, 59, 194, 156, 86, 244, 56, 129, 94, 60, 62, 131, 60, 122, 247, 12, 187, 214, 64, 128, 176, 1, 119, 1, 139, 150, 66]
[2020-03-07T08:58:47Z INFO  actix_mqtt_client::random_test] Got message: id:0, topic: test, payload: [87, 50, 65, 193, 210, 48, 17, 234, 167, 240, 219, 169, 173, 46, 207, 207, 149, 234, 103, 177, 125, 115, 105, 113, 156, 174, 160, 96, 213, 176, 141, 129]
[2020-03-07T08:58:47Z INFO  actix_mqtt_client::random_test] Pending recv items: 0
[2020-03-07T08:58:47Z INFO  actix_mqtt_client::random_test] Pending recv items: 0
[2020-03-07T08:58:47Z INFO  actix_mqtt_client::actors::packets::publish] Handle message for RecvPublishActor
[2020-03-07T08:58:47Z INFO  actix_mqtt_client::actors::packets::publish] Handle message for RecvPublishActor
If you find any error logs please paste them here and I can take a look.

Or if you can share the full code snippet which you use to connect to local MQTT server I can also check what's happening.

Hi, I created a repo: https://github.com/xudesheng/actixmqtttest
since it includes credential for test, so, I have to make it to be private.

main binary: actixmqtttest, I modified your sample code but it always fail.
another binary: single, which is also modified from your sample code and it works.

I also run single in two separated instances and point to different devices, it works as expected.
RUST_LOG=info cargo run --bin single 0

RUST_LOG=info cargo run --bin single 1

I included error log in repo too.

appreciate your help.

Syndim · 2020-03-08T08:51:49Z

hi, I compared your working/not working code and find that the client name of the not working one is wrong:

let mut client = MqttClient::new(
            r,
            w,
            format!("test_{}", client_id),                // THIS LINE
            mqtt_option,
            MessageActor(sender.clone()).start().recipient(),
            ErrorActor.start().recipient(),
            None,
        );

It should just be client_id. After updating the client name the code is running fine.

I also made some improvement for multi-client support of the library and published v0.4.0, you can also have a try :).

xudesheng · 2020-03-08T22:06:53Z

hi, I compared your working/not working code and find that the client name of the not working one is wrong:
let mut client = MqttClient::new(
            r,
            w,
            format!("test_{}", client_id),                // THIS LINE
            mqtt_option,
            MessageActor(sender.clone()).start().recipient(),
            ErrorActor.start().recipient(),
            None,
        );
It should just be client_id. After updating the client name the code is running fine.

I also made some improvement for multi-client support of the library and published v0.4.0, you can also have a try :).

Shame! I wasted whole day there.
really appreciate and thank you very much. I will come back to this lib in my next release very soon.

xudesheng · 2020-03-09T06:08:08Z

@Syndim , sorry, one more question needs your help. Appreciate.
I have to simulate up to 1000 MQTT client. If I understand correctly, one Arbiter is just one thread, which obviously will be a bottle neck.
I'm thinking about two approaches:

lunch multiple threads, each thread has a System::run to drive all futures.
Use multiple Arbiters? never see examples.

appreciate you can give some hit.

Syndim · 2020-03-09T14:42:12Z

@xudesheng sorry I don't have much experience about it. I think the event loop itself is single threaded but the executors aren't so I guess it may not be a big issue?

xudesheng · 2020-03-09T16:57:11Z

@xudesheng sorry I don't have much experience about it. I think the event loop itself is single threaded but the executors aren't so I guess it may not be a big issue?

I will run some benchmark test. I have used another lib 'mqtt-async-client' and have to lunch multiple threads and use tokio::local in each thread.
New build is based your lib and I will use system executor. Hopefully I can find enough resource to run benchmark test sometime later.

xudesheng closed this as completed Mar 8, 2020

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

{ kind: Interrupted, error: "Mailbox has closed" } #7

{ kind: Interrupted, error: "Mailbox has closed" } #7

xudesheng commented Mar 6, 2020 •

edited

Loading

Syndim commented Mar 6, 2020

xudesheng commented Mar 6, 2020 •

edited

Loading

xudesheng commented Mar 6, 2020 •

edited

Loading

xudesheng commented Mar 6, 2020 •

edited

Loading

xudesheng commented Mar 6, 2020 •

edited

Loading

Syndim commented Mar 7, 2020

xudesheng commented Mar 7, 2020

xudesheng commented Mar 7, 2020

Syndim commented Mar 8, 2020

xudesheng commented Mar 8, 2020

xudesheng commented Mar 9, 2020

Syndim commented Mar 9, 2020

xudesheng commented Mar 9, 2020

{ kind: Interrupted, error: "Mailbox has closed" } #7

{ kind: Interrupted, error: "Mailbox has closed" } #7

Comments

xudesheng commented Mar 6, 2020 • edited Loading

Syndim commented Mar 6, 2020

xudesheng commented Mar 6, 2020 • edited Loading

xudesheng commented Mar 6, 2020 • edited Loading

xudesheng commented Mar 6, 2020 • edited Loading

xudesheng commented Mar 6, 2020 • edited Loading

Syndim commented Mar 7, 2020

xudesheng commented Mar 7, 2020

xudesheng commented Mar 7, 2020

Syndim commented Mar 8, 2020

xudesheng commented Mar 8, 2020

xudesheng commented Mar 9, 2020

Syndim commented Mar 9, 2020

xudesheng commented Mar 9, 2020

xudesheng commented Mar 6, 2020 •

edited

Loading

xudesheng commented Mar 6, 2020 •

edited

Loading

xudesheng commented Mar 6, 2020 •

edited

Loading

xudesheng commented Mar 6, 2020 •

edited

Loading

xudesheng commented Mar 6, 2020 •

edited

Loading