Как безопасно дождаться новых данных в реакторе потокового сервера GRPC C ++, используя API обратного вызова? ⇐ C++

[Anonymous]

Я реализовал реактор потокового сервера GRPC, используя API обратного вызова C ++. Вот основная идея: отдельный поток приложения создает данные каждые 10 мс и вызывает метод ответа () на реакторе. Я толкаю данные в очередь, и поток GRPC и отправляет их клиенту. Поток GRPC ждет данных, используя std :: condition_variable . Это работает надежно на практике. < /P>
Однако, просмотрев официальные лучшие практики GRPC C ++, я нашел следующее руководство: < /p>

Реакции должны быть быстрыми. Не выполняйте блокирующие или длительные/тяжелые задачи или сон. Это может повлиять на другие RPC в рамках процесса. В результате я не уверен, какие потоки разрешено вызывать методы API GRPC, поэтому, чтобы оставаться в безопасности, я вызываю их только из потоков GRPC. Контекст. Реализация: < /p>

Код: Выделить всё

class EcatStreamReactor final
: public grpc::ServerWriteReactor, public IBusClient
{
public:
EcatStreamReactor(
grpc::CallbackServerContext *context,
const ecat::ControlMessage *request
)
: _request(request)
, _context(context)
{}

void set_bus_controller(std::weak_ptr bus_controller);   // Implementation omitted

/**
* Called from application thread every 10 ms
* Implements IBusClient
*/
void response(const EcatResponse &response) override
{
std::unique_lock lock(_data_mutex);
_data_queue.push(response);
_data_cv.notify_one();
}

// Called once from service handler after reactor construction
void Work()
{
EcatResponse data;
grpc::Status status;
bool has_data = false;
auto deadline = std::chrono::system_clock::now() + std::chrono::milliseconds(100);
{
std::unique_lock lock(_data_mutex);
_data_cv.wait_until(lock, deadline, [this] {
return !_data_queue.empty() || _cancelled.load();
});

if (_cancelled.load()) {
status = grpc::Status::OK;
} else if (!_data_queue.empty()) {
data = _data_queue.front();
_data_queue.pop();
has_data = true;
} else {
status = grpc::Status(grpc::StatusCode::DEADLINE_EXCEEDED,
"No data received from bus controller");
}
}

if (has_data) {
_res = ecat_conv::convert(data);
StartWrite(&_res);
} else {
Finish(status);
}
}

void OnWriteDone(bool ok) override
{
if (ok)
Work();
else
Finish(grpc::Status::CANCELLED);
}

void OnCancel() override; // Implementation omitted

void OnDone() override;   // Implementation omitted

private:
ecat::EcatResponse _res;
std::queue _data_queue;
std::mutex _data_mutex;
std::condition_variable_any _data_cv;
std::weak_ptr _bus_controller;
std::atomic _cancelled{false};
grpc::CallbackServerContext *_context;
const ecat::ControlMessage *_request;
};

Любые советы или шаблоны для реализации этого правильно и безопасно будут оценены.


Подробнее здесь: https://stackoverflow.com/questions/796 ... -using-the