Message may arrive out of order in mmap mode?

[chengguizi]

I understand system like ROS1 which relies on TCP transport naturally guarantee the subscriber receive the messages in order. This is particularly important for high frequency sequential data like IMU messages at several hundred Hz.

However, I experience regression of timestamps when I turn memfile_buffer_count to be more than 1 (e.g. 2), and with a small memfile_ack_timeout (say 0 or 2).

The issue seems to disappear when I increase timeout such that memfile_buffer_count = 2 and memfile_ack_timeout = 10.

Could help explain what is going on under the hood? And is there anyway I can tweak eCAL to guarantee in order messaging on the same machine? Thanks!

A bit more info:

• Running with eCAL 5.11.3
• Running on ARM64 edge device, Jetson Xavier NX, ubuntu 20.04
• Seems that on my amd64 machine, the issue is not really observable, probably due to high performance?

[KerstinKeller]

Hi, thanks for reporting this!
First of all, the messages shouldn't arrive out of order, but second of all, I fear that the memfile_buffer_count != 1 is not very well tested. (We're usually not using this feature, instead we usually add internal buffers which are being filled in the callbacks - and which is what I would recommend).

Do you have a minimal sample which can reproduce the issue on an x86 architecture?

[chengguizi]

Emm not yet. The deployment platform I currently testing on is arm64. I will see if I can reproduce on a x86 machine soonest possible. But I think the issue is partially masked out when the CPU is powerful enough.. Let's see

[chengguizi]

If I do get a minimal reproducible sample, will raise an issue then

[FlorianReimold]

• Yes, TCP guarantees the correct order of messages. This guarantee however is only given for one connection (so basically 1:1). If you have multiple clients connected to the same endpoint, the messages of different clients may arrive out of order, e.g. because the route of one client is just longer and messages take more time to be transferred. This is also how the eCAL TCP Layer works.
• For eCAL UDP, there are no measures to bring messages back into order, as eCAL is tuned to reduce latency. So - in theory - even messages coming from one publisher can be received out of order. In practice this is however unlikely, as eCAL targets local networks, where usually only 1 route to any host exists.
• We (the eCAL Devs) have talked about SHM with mult-buffer. It is important for us to know more about your setup in order to make an educated statement here. If you have multiple publishers, then it is expected for messages to arrive out of order from time to time for the same reason like TCP. If not, then we may want to investigate a little further. As eCAL however doesn't buffer messages, the only way eCAL could react to that case is dropping messages that are too old.

What you can always do in your own code is to store the last timestamp and drop messages with timestamps smaller than that.

[chengguizi]

I have a single publisher. Yes dropping messages that is older is an expected behaviour. Having out of order data generally is quite undesirable for sensors

[KerstinKeller]

With the following test and settings of

memfile_ack_timeout = 30
memfile_buffer_count = 2

I can reproduce the out of order messages:

In my opinion, the only thing which can really be done to avoid this, is to have shared locks of the SHM for reading?

However, what I don't understand, even with acknowledge, shouldn't the memfile be blocked only while the read is taking place, not while the whole callback is? (zerocopy is NOT enabled)

Setup:
1 Publisher, Sending at 10 Hz
1 Subscriber, Sleeping for 20ms in the callbacks
1 Subscriber, Checking if messages arrive out of order. (-> They do)

#1 - Expected: (last_clock) <= (clock_), actual: 192 vs 191
#2 - Expected: (last_clock) <= (clock_), actual: 247 vs 246
#3 - Expected: (last_clock) <= (clock_), actual: 291 vs 290
#4 - Expected: (last_clock) <= (clock_), actual: 309 vs 308
#5 - Expected: (last_clock) <= (clock_), actual: 357 vs 356
...

-> Not every (other) message arrives out of order, maybe 1 / 50 ? It seems like this could be quite hard to locate, we might need more logging....

@rex-schilasky

/* ========================= eCAL LICENSE =================================
 *
 * Copyright (C) 2016 - 2019 Continental Corporation
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 * ========================= eCAL LICENSE =================================
*/

#include <ecal/ecal.h>
#include <ecal/msg/string/publisher.h>
#include <ecal/msg/string/subscriber.h>
#include <algorithm>
#include <atomic>
#include <thread>
#include <vector>
#include <gtest/gtest.h>

namespace {
  // subscriber callback function
  void OnReceiveLazy(const char* topic_name_, const std::string& msg_, long long time_, long long clock_, long long id_)
  {
    std::this_thread::sleep_for(std::chrono::milliseconds(20));
  }

  void OnReceiveCrazy(const char* topic_name_, const std::string& msg_, long long time_, long long clock_, long long id_)
  {
    static long long last_clock = 0;
    EXPECT_LE(last_clock, clock_);
    last_clock = clock_;
  }
}


TEST(Core, OutOfOrderMessages)
{
  // initialize eCAL API
  EXPECT_EQ(0, eCAL::Initialize(0, nullptr, "callback destruction"));

  // enable loop back communication in the same thread
  eCAL::Util::EnableLoopback(true);

  // start publishing thread
  eCAL::string::CPublisher<std::string> pub("foo");
  eCAL::string::CSubscriber<std::string> sub_lazy("foo");
  sub_lazy.AddReceiveCallback(OnReceiveLazy);
  eCAL::string::CSubscriber<std::string> sub_crazy("foo");
  sub_crazy.AddReceiveCallback(OnReceiveCrazy);


  std::atomic<bool> pub_stop(false);
  std::thread pub_t([&]() {
    while (!pub_stop)
    {
      pub.Send("Hello World");
      // some kind of busy waiting....
      std::this_thread::sleep_for(std::chrono::milliseconds(10));
    }
    });

  std::atomic<bool> sub_lazy_stop(false);
  std::thread sub_lazy_t([&]() {
    while (!sub_lazy_stop)
    {
      std::this_thread::sleep_for(std::chrono::milliseconds(100));
    }
    });

  std::atomic<bool> sub_crazy_stop(false);
  std::thread sub_crazy_t([&]() {
    while (!sub_crazy_stop)
    {
      std::this_thread::sleep_for(std::chrono::milliseconds(100));
    }
    });
  
  // let them work together
  std::this_thread::sleep_for(std::chrono::seconds(10));

  // stop publishing thread
  pub_stop = true;
  pub_t.join();

  sub_lazy_stop = true;
  sub_lazy_t.join();

  sub_crazy_stop = true;
  sub_crazy_t.join();

  // finalize eCAL API
  // without destroying any pub / sub
  EXPECT_EQ(0, eCAL::Finalize());

}