multi-thread Client Request

[cyzhou]

When I create one client in one process, and send msgs at the same time in multi-threads(using lock in sending part), then server can't send response to client after receiving one request. The error msg is "Could not deliver to client!Client is not available anymore!". I guess it may can't send response to client because client is sending msg.? So what's the right way to send msg in multi-threads, is there any demo can show this?

[mossmaurice]

@cyzhou Thanks for raising this question!

In general, the APIs of Server's or Clients are not thread-safe, but using a lock should be fine. I fear there is no example available for this particular use case.

I don't fully understand your case yet. Few questions:

1. Are you using sequences numbers to check which thread sent the request?
2. Why didn't you consider to use two different Client / Server pairs?
3. The error

   iceoryx/iceoryx_posh/source/popo/ports/server_port_user.cpp

   Line 153 in fbdae45

   LogWarn() << "Could not deliver to client! Client not available anymore!";

   indicates that either the error POPO__CHUNK_SENDER_INVALID_CHUNK_TO_SEND_FROM_USER must have occurred or Server / Client are not connected. Did you see this one?

Could you copy and paste your example code here or to a GitHub repository so it can be reproduced? Thanks!

[cyzhou]

//Client
void A::init() {
client_ = std::make_shared(iox::capro::ServiceDescription(
iox::capro::IdString_t(iox::cxx::TruncateToCapacity, serviceName),
iox::capro::IdString_t(iox::cxx::TruncateToCapacity, instanceName),
iox::capro::IdString_t(iox::cxx::TruncateToCapacity, methodName)));
listener_.attachEvent(
*newClient, iox::popo::ClientEvent::RESPONSE_RECEIVED,
iox::popo::createNotificationCallback(OnResponseReceived));
}

void A::Request(const std::vector<uint8_t>& data) {
std::lock_guardstd::mutex guard(lock_);
auto requestID = getID(); //increased num
auto loanResult = client_->loan();
//! [loan]
if (!loanResult.has_error()) {
auto& request = loanResult.value();
request.getRequestHeader().setSequenceId(requestID);
request->size = data.size();
memcpy(request->body, data.data(), data.size());
auto re = request.send();
if (re.has_error()) {
utils::GetLogger().LogError() << "send request failed, error code: " << iox::popo::asStringLiteral(re.get_error());
}
} else {
auto error = loanResult.get_error();
utils::GetLogger().LogError() << "Unable to loan sample, error code: " << iox::popo::asStringLiteral(error);
}
}

void methodCall(A* p) {
while(true) {
p->Request("TEST TEST");
std::this_thread::sleep_for(std::chrono::milliseconds(500));
}
}

int main() {
A a;
a.init();
std::thread tds[LEN];
for (int i = 0; i < LEN; ++i) {
tds[i] = std::thread(methodCall, &a);
}
}

//server
void B::init() {
server_ = std::make_shared(
iox::capro::ServiceDescription(service_, instance_,
iox::capro::IdString_t(iox::cxx::TruncateToCapacity, method)));
listener_.attachEvent(
*server, iox::popo::ServerEvent::REQUEST_RECEIVED,
iox::popo::createNotificationCallback(OnRequestReceived));
}

void OnRequestReceived(ServerIns* server) {
server->take().and_then([&](const auto& request) {
//! [send response]
server->loan(request)
.and_then([&](auto& response) {
//request
std::vector<uint8_t> reqBody(request->body, request->body + request->size);
std::vector<uint8_t> replyBody = reqBody;
replyBody.push_back('B');
response->size = replyBody.size();
memcpy(response->body, replyBody.data(), replyBody.size());
response.send().or_else([&](auto& error) {
utils::GetLogger().LogWarn() << "Could not send Response! Error: " << iox::popo::asStringLiteral(error);
});
})
.or_else([](auto& error) {
utils::GetLogger().LogWarn() << "Could not allocate Response! Error: " << iox::popo::asStringLiteral(error);
});
});
}

[cyzhou]

@mossmaurice see my example above, could u understand it and reproduce it? Thanks a lot~

[mossmaurice]

@elBoberido As I'm on vacation, could you have a look at the above code? Thanks!

[elBoberido]

@mossmaurice @cyzhou the problem is that the Listener also has a thread and accesses the Client without the lock. @elfenpiff has a more detailed analysis and a potential solution.

[elfenpiff]

@cyzhou I looked at your example but I was unable to reproduce or understand it in its entirety - I think there is still some code missing.

But here some general suggestions.

• If you use threads anyway I wouldn't use the Listener use the Waitset instead in one thread, collect all the requests and then distribute the work loads over multiple threads.
• Every class can be made very easily threadsafe with the smart_lock. For the client you could write something like:

#include "iceoryx_hoofs/internal/concurrent/smart_lock.hpp"

using namespace iox::capro; // for ServiceDescription and IdString_t
using namespace iox::concurrent; // for smart_lock and ForwardArgsToCtor
using namespace iox::cxx; // for TruncateToCapacity
using namespace iox::popo; // for Client

smart_lock<Client<AddRequest, AddResponse>> 
  client(ForwardArgsToCTor, ServiceDescription{IdString_t(TruncateToCapacity, serviceName), 
                                               IdString_t(TruncateToCapacity, instanceName), 
                                               IdString_t(TruncateToCapacity, eventName)});

auto loanResult = client->loan(); // works like a shared_ptr now -> ensures that the method loan is called inside a locked mutex
if (!loanResult.has_error()) {
   auto &request = loanResult.value(); 
   // remaining code like in your example
}

// attaching to a waitset works like
myWaitset->attachEvent(client.operator->(), ClientEvent::RESPONSE_RECEIVED);

// remark client.operator->(); is really bad style but there is not yet a dereferencing operator implemented therefore it is the best solution

To narrow down the deadlock you can also try to use helgrind (https://valgrind.org/docs/manual/hg-manual.html) or the thread sanitizer from gcc.

[cyzhou]

@elfenpiff Leave multi-threads problem alone, I found that if I only send msgs from client to server in main thread too frequently，then server will miss msgs. If I send one msg, then sleep 100ms, there will be no problem. So I guess there maybe some restriction when send msgs or publish msgs. Can u guys tell us more details? Thanks a lot.

use example:
//client
class Client {
public:
void send(const char* str) {
auto loanResult = client.loan(); // works like a shared_ptr now -> ensures that the method loan is called inside a locked mutex
if (!loanResult.has_error()) {
auto &request = loanResult.value();
// remaining code like in your example
}
}
private:
Client<AddRequest, AddResponse> client_;
};
//init client
Client proxy;
for (auto i = 0; i < 1000; ++i) {
proxy.send("TEST");
}

[elBoberido]

@cyzhou there is a bounded queue between the client and the server. By default, if the queue is full, the oldest request is discarded. You can change this behavior with the ClientOptions and ServerOptions