How to clear the queue in publisher?

[jerry-xie-000]

I use publisher.loan( 1000 ).and_then([&](auto& userPayload) { .... }; to publish quickly, so there maybe lots of memory blocks in the publish queue. Sometimes, I need to clear the whole publish queue, and restart to loan again.
Is there some function like publisher.clear() to do this thing?

[elfenpiff]

@jerry-xie-000 The publisher has a queue for the history and every subscriber has a buffer/queue. I am unsure what kind of publish queue you mean?

[jerry-xie-000]

I make a mistake, maybe is not about the publisher or subscriber queue. I try to describe the workflow below:

1. subscriber wait() and then take(), sleep(2) to do something, and then release(), I guess this will cause the Mempool to hold several data.
2. publisher loan() and publish() 10 times very quickly;
3. when the subscriber is handling the third data, something happen, and I need to stop the subscriber to handle the remaining data. Don't
   destroy the subscriber process, because it need to wait() again to receive the new data.

Is there any method to clear all the data in the Mempool?

[elfenpiff]

Ah, now I got it!

Is there any method to clear all the data in the Mempool?

No, since this would cause data races in all subscribers. Just think about the case where a subscriber reads the data while the publishers clear the data in the mempool and reuse it for something completely different. The subscriber would read garbage, and by accident, you may kill it.

But your problem can be solved completely; what you require is

1. safe overflow (which is enabled by default unless you have defined the QueueFullPolicy::BLOCK_PRODUCER on the subscriber side)
2. A memory configuration providing much more samples so that you never run out of chunks. This is described in detail here:https://github.com/eclipse-iceoryx/iceoryx/blob/main/doc/website/advanced/configuration-guide.md#configuring-mempools-for-roudi

Others, besides you, have experienced the same issue in the past, and it is always a bit cumbersome to overcome. With iceoryx2, we solved this issue once and for all and a lot of other restrictions. So, if you are interested, check it out. It comes with C and C++ language bindings, and the only feature missing so that it has feature parity with good old iceoryx is request-response, which will come in the following months.

But it has a lot of other cool features like:

• it no longer requires a central daemon (RouDi)
• no more mempool configuration
• up to 10 times faster compared to iceoryx
• dynamic memory support - so the payload side can vary after the service has been created
• better tooling
• completely decentralized and much more robust
• written in Rust but C, C++ and Python will become first class language bindings so that you can use it without having to care about Rust (if you don't want to)

[jerry-xie-000]

I remove the QueueFullPolicy::BLOCK_PRODUCER on the subscriber side. But if the subscriber sleep 2 seconds to do something, the publisher will print the below error every time:

Unable to loan sample, error code: AllocationError::RUNNING_OUT_OF_CHUNKS
iox-cpp-publisher-untyped sent two times value: 261
2021-01-23 11:09:53.097 [Warn ]: Mempool [m_chunkSize = 524336, numberOfChunks = 50, used_chunks = 50 ] has no more space left
2021-01-23 11:09:53.097 [Error]: MemoryManager: unable to acquire a chunk with a chunk-payload size of 133666The following mempools are available:  MemPool [ ChunkSize = 176, ChunkPayloadSize = 128, ChunkCount = 50 ]  MemPool [ ChunkSize = 1072, ChunkPayloadSize = 1024, ChunkCount = 50 ]  MemPool [ ChunkSize = 16432, ChunkPayloadSize = 16384, ChunkCount = 50 ]  MemPool [ ChunkSize = 131120, ChunkPayloadSize = 131072, ChunkCount = 50 ]  MemPool [ ChunkSize = 524336, ChunkPayloadSize = 524288, ChunkCount = 50 ]  MemPool [ ChunkSize = 1048624, ChunkPayloadSize = 1048576, ChunkCount = 50 ]  MemPool [ ChunkSize = 4194352, ChunkPayloadSize = 4194304, ChunkCount = 10 ]
2021-01-23 11:09:53.098 [Error]: /mnt/sda1/xzy/Workspace/device_controller/third_party/iceoryx/iceoryx_posh/source/mepoo/memory_manager.cpp:197 [MEPOO__MEMPOOL_GETCHUNK_POOL_IS_RUNNING_OUT_OF_CHUNKS (code = 84)] in module [iceoryx_posh (id = 2)]
Unable to loan sample, error code: AllocationError::RUNNING_OUT_OF_CHUNKS
iox-cpp-publisher-untyped sent two times value: 262
2021-01-23 11:09:53.118 [Warn ]: Mempool [m_chunkSize = 524336, numberOfChunks = 50, used_chunks = 50 ] has no more space left
2021-01-23 11:09:53.118 [Error]: MemoryManager: unable to acquire a chunk with a chunk-payload size of 133666The following mempools are available:  MemPool [ ChunkSize = 176, ChunkPayloadSize = 128, ChunkCount = 50 ]  MemPool [ ChunkSize = 1072, ChunkPayloadSize = 1024, ChunkCount = 50 ]  MemPool [ ChunkSize = 16432, ChunkPayloadSize = 16384, ChunkCount = 50 ]  MemPool [ ChunkSize = 131120, ChunkPayloadSize = 131072, ChunkCount = 50 ]  MemPool [ ChunkSize = 524336, ChunkPayloadSize = 524288, ChunkCount = 50 ]  MemPool [ ChunkSize = 1048624, ChunkPayloadSize = 1048576, ChunkCount = 50 ]  MemPool [ ChunkSize = 4194352, ChunkPayloadSize = 4194304, ChunkCount = 10 ]
2021-01-23 11:09:53.119 [Error]: /mnt/sda1/xzy/Workspace/device_controller/third_party/iceoryx/iceoryx_posh/source/mepoo/memory_manager.cpp:197 [MEPOO__MEMPOOL_GETCHUNK_POOL_IS_RUNNING_OUT_OF_CHUNKS (code = 84)] in module [iceoryx_posh (id = 2)]
Unable to loan sample, error code: AllocationError::RUNNING_OUT_OF_CHUNKS
iox-cpp-publisher-untyped sent two times value: 263

iox_publisher_untyped.cpp:

// Copyright (c) 2020 by Robert Bosch GmbH. All rights reserved.
// Copyright (c) 2020 - 2021 by Apex.AI Inc. All rights reserved.
//
// 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.
//
// SPDX-License-Identifier: Apache-2.0

//! [include topic data]
#include "topic_data.hpp"
//! [include topic data]

//! [includes]
#include "iceoryx_posh/popo/untyped_publisher.hpp"
#include "iceoryx_posh/runtime/posh_runtime.hpp"
#include "iox/signal_watcher.hpp"
//! [includes]

#include <iostream>

int main()
{
    //! [runtime initialization]
    constexpr char APP_NAME[] = "iox-cpp-publisher-untyped";
    iox::runtime::PoshRuntime::initRuntime(APP_NAME);
    //! [runtime initialization]

    //! [create untyped publisher]
    iox::popo::UntypedPublisher publisher({"Radar", "FrontLeft", "Object"});
    //! [create untyped publisher]

    double ct = 0.0;
    while (!iox::hasTerminationRequested())
    {
        ++ct;

        //! [Loan chunk and provide logic to populate it via a lambda]
        publisher.loan(133666)                       // <=================   change!
            .and_then([&](auto& userPayload) {
                //! [construct RadarObject]
                RadarObject* data = new (userPayload) RadarObject(ct, ct, ct);
                //! [construct RadarObject]

                //! [write data]
                data->x = ct;
                data->y = ct;
                data->z = ct;
                //! [write data]
  
                //! [publish]
                publisher.publish(userPayload);
                //! [publish]
            })
            .or_else([&](auto& error) {
                //! [print error]
                std::cerr << "Unable to loan sample, error code: " << error << std::endl;
                //! [print error]
            });
        //! [Loan chunk and provide logic to populate it via a lambda]

        std::cout << APP_NAME << " sent two times value: " << ct << std::endl;

        std::this_thread::sleep_for(std::chrono::milliseconds(20));       // <=================   change!
    }

    return 0;
}

iox_subscriber_untyped.cpp:

// Copyright (c) 2020 by Robert Bosch GmbH. All rights reserved.
// Copyright (c) 2020 - 2021 by Apex.AI Inc. All rights reserved.
//
// 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.
//
// SPDX-License-Identifier: Apache-2.0

//! [includes]
#include "topic_data.hpp"

#include "iceoryx_posh/popo/untyped_subscriber.hpp"
#include "iceoryx_posh/runtime/posh_runtime.hpp"
#include "iox/signal_watcher.hpp"
//! [includes]

#include <iostream>

iox::concurrent::Atomic<bool> keepRunning{true};


using WaitSet = iox::popo::WaitSet<>;

volatile WaitSet* waitsetSigHandlerAccess{nullptr};

static void sigHandler(int f_sig [[maybe_unused]])
{
    printf("Sign\n");
    keepRunning = false;
    if (waitsetSigHandlerAccess)
    {
        waitsetSigHandlerAccess->markForDestruction();
    }
}

int main()
{
        // register sigHandler
    auto signalIntGuard =
        iox::registerSignalHandler(iox::PosixSignal::INT, sigHandler).expect("failed to register SIGINT");
    auto signalTermGuard =
        iox::registerSignalHandler(iox::PosixSignal::TERM, sigHandler).expect("failed to register SIGTERM");


    //! [initialize runtime]
    constexpr char APP_NAME[] = "iox-cpp-subscriber-untyped";
    iox::runtime::PoshRuntime::initRuntime(APP_NAME);
    //! [initialize runtime]

    //! [create untyped subscriber]
    iox::popo::UntypedSubscriber subscriber({"Radar", "FrontLeft", "Object"});
    //! [create untyped subscriber]
    iox::popo::WaitSet<> waitset_;
    waitsetSigHandlerAccess = &waitset_;

    waitset_.attachState(subscriber, iox::popo::SubscriberState::HAS_DATA).or_else([](auto) {
        std::cerr << "failed to attach subscriber" << std::endl;
        std::exit(EXIT_FAILURE);
    });

                

    // run until interrupted by Ctrl-C
    //! [loop]
    while (keepRunning)
    {
         auto notificationVector = waitset_.wait();

        for (auto& notification : notificationVector)
        {
            // We woke up and hence there must be at least one sample. When the sigHandler has called
            // markForDestruction the notificationVector is empty otherwise we know which subscriber received samples
            // since we only attached one.
            // Best practice is to always acquire the notificationVector and iterate over all elements and then react
            // accordingly. When this is not done and more elements are attached to the WaitSet it can cause
            // problems since we either miss events or handle events for objects which never occurred.
            if (notification->doesOriginateFrom(&subscriber))
            {
                        subscriber
            .take()
            //! [chunk happy path]
            .and_then([&](const void* userPayload) {
                //! [chunk received]
                auto object = static_cast<const RadarObject*>(userPayload);
                std::cout << APP_NAME << " got value: " << object->x << std::endl;
                //! [chunk received]
 

                 sleep(2);                   // <=====================  do something here!

                //! [release]
                // note that we explicitly have to release the sample
                // and afterwards the pointer access is undefined behavior
                subscriber.release(userPayload);
                //! [release]

            })
            //! [chunk happy path]
            .or_else([](auto& result) {
                if (result != iox::popo::ChunkReceiveResult::NO_CHUNK_AVAILABLE)
                {
                    std::cout << "Error receiving chunk." << std::endl;
                }
            });
            }
        }

    }
    //! [loop]

    return (EXIT_SUCCESS);
}

It seems not safe overflow when new data coming.
Is there something wrong on my code?

[elfenpiff]

@jerry-xie-000 The underlying memory pool is running out of memory and you need to reconfigure it so that it can hold more chunks:

2021-01-23 11:09:53.097 [Warn ]: Mempool [m_chunkSize = 524336, numberOfChunks = 50, used_chunks = 50 ] has no more space left

So you require maybe 100 chunks of size 524336. This document explains how to realize this.

Btw. with iceoryx2 you no longer need to configure your memory - it guarantees you that it will never run out of memory, and it does not require you to use a central daemon like RouDi.

[jerry-xie-000]

I would like to try the iceoryx2, but our platform is arm64. Do you have the install and compile guild for arm64? I try to compile, but the rust show some error.

[elfenpiff]

@jerry-xie-000 Just create an issue on iceoryx2 with the error output you get.

iceoryx2 officially supports arm64 and we also have a CI target that continuously runs, so any kind of problem you have there should be some kind of bug on iceoryx2 side.

[jerry-xie-000]

OK, I post the issue for iceoryx2.

Let's back to my example above, I require maybe 100 chunks of size 524336. However, if I make a sleep 30 seconds in subscriber, the chunks of size may be huge.
My question is why the new coming data will not be dropped when the chunks is full.

[elfenpiff]

@jerry-xie-000 I think I found the issue, take a look at this here: https://github.com/eclipse-iceoryx/iceoryx/blob/main/iceoryx_examples/icedelivery/iox_subscriber_untyped.cpp

The untyped subscriber requires you to release the samples manually with subscriber.release(userPayload);

[jerry-xie-000]

            .and_then([&](const void* userPayload) {
                //! [chunk received]
                auto object = static_cast<const RadarObject*>(userPayload);
                std::cout << APP_NAME << " got value: " << object->x << std::endl;
                //! [chunk received]
 

                 sleep(2);                   // <=====================  do something here!

                //! [release]
                // note that we explicitly have to release the sample
                // and afterwards the pointer access is undefined behavior
                subscriber.release(userPayload);
                //! [release]

            })

Yes, I know. I do the subscriber.release(userPayload); after handling something that maybe take a few time, maybe 2 seconds. But the the chunks will show these errors when full.

The whole source example I had show above. It's easy to reproduce the problem.

[elfenpiff]

@jerry-xie-000 Sorry, I overlooked it. Could you take a look at this example: https://github.com/eclipse-iceoryx/iceoryx/tree/main/iceoryx_examples/iceoptions

and try to reduce the subscriber buffer.

[jerry-xie-000]

I set the subscriber queue by 10, and the errors gone.

    iox::popo::SubscriberOptions subscriberOptions;
    subscriberOptions.queueCapacity = 10U;
    iox::popo::UntypedSubscriber subscriber({"Radar", "FrontLeft", "Object"}, subscriberOptions);

I found the default value is SubscriberChunkQueueData_t::MAX_CAPACITY. It seems it will never overflow when full.

Thank you.

[elBoberido]

Right, the queue is quite large by default. We should maybe enforce the setting of the queue size on construction of the subscriber. It would probably be a braking change though. I need to check how it's handled by iceoryx2