hal/vehicle/2.0/GrpcVehicleServer.cpp - device/google/trout - Git at Google

 /*
  * Copyright (C) 2019 The Android Open Source Project
  *
  * 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.
  */
 #include "GrpcVehicleServer.h"

 #include <condition_variable>
 #include <mutex>
 #include <shared_mutex>

 #include <android-base/logging.h>
 #include <grpc++/grpc++.h>

 #include "GarageModeServerSideHandler.h"
 #include "PowerStateListener.h"
 #include "VehicleServer.grpc.pb.h"
 #include "VehicleServer.pb.h"
 #include "vhal_v2_0/DefaultConfig.h"
 #include "vhal_v2_0/ProtoMessageConverter.h"

 namespace android {
 namespace hardware {
 namespace automotive {
 namespace vehicle {
 namespace V2_0 {

 namespace impl {

 class GrpcVehicleServerImpl : public GrpcVehicleServer, public vhal_proto::VehicleServer::Service {
   public:
     explicit GrpcVehicleServerImpl(const VirtualizedVhalServerInfo& serverInfo)
         : mServiceAddr(serverInfo.getServerUri()),
           mGarageModeHandler(makeGarageModeServerSideHandler(this, &mValueObjectPool,
                                                              serverInfo.powerStateMarkerFilePath)),
           mPowerStateListener(serverInfo.powerStateSocket, serverInfo.powerStateMarkerFilePath) {
         setValuePool(&mValueObjectPool);
     }

     // method from GrpcVehicleServer
     GrpcVehicleServer& Start() override;

     void Wait() override;

     GrpcVehicleServer& Stop() override;

     uint32_t NumOfActivePropertyValueStream() override;

     // methods from IVehicleServer
     void onPropertyValueFromCar(const VehiclePropValue& value, bool updateStatus) override;

     StatusCode onSetProperty(const VehiclePropValue& value, bool updateStatus) override;

     // methods from vhal_proto::VehicleServer::Service

     ::grpc::Status GetAllPropertyConfig(
             ::grpc::ServerContext* context, const ::google::protobuf::Empty* request,
             ::grpc::ServerWriter<vhal_proto::VehiclePropConfig>* stream) override;

     ::grpc::Status SetProperty(::grpc::ServerContext* context,
                                const vhal_proto::WrappedVehiclePropValue* wrappedPropValue,
                                vhal_proto::VehicleHalCallStatus* status) override;

     ::grpc::Status StartPropertyValuesStream(
             ::grpc::ServerContext* context, const ::google::protobuf::Empty* request,
             ::grpc::ServerWriter<vhal_proto::WrappedVehiclePropValue>* stream) override;

   private:
     // We keep long-lasting connection for streaming the prop values.
     // For us, each connection can be represented as a function to send the new value, and
     // an ID to identify this connection
     struct ConnectionDescriptor {
         using ValueWriterType = std::function<bool(const vhal_proto::WrappedVehiclePropValue&)>;

         explicit ConnectionDescriptor(ValueWriterType&& value_writer)
             : mValueWriter(std::move(value_writer)),
               mConnectionID(CONNECTION_ID_COUNTER.fetch_add(1)) {}

         ConnectionDescriptor(const ConnectionDescriptor&) = delete;

         ConnectionDescriptor& operator=(const ConnectionDescriptor&) = delete;

         // This move constructor is NOT THREAD-SAFE, which means it cannot be moved
         // while using. Since the connection descriptors are pretected by mConnectionMutex
         // then we are fine here
         ConnectionDescriptor(ConnectionDescriptor&& cd)
             : mValueWriter(std::move(cd.mValueWriter)),
               mConnectionID(cd.mConnectionID),
               mIsAlive(cd.mIsAlive.load()) {
             cd.mIsAlive.store(false);
         }

         ValueWriterType mValueWriter;
         uint64_t mConnectionID;
         std::atomic<bool> mIsAlive{true};

         static std::atomic<uint64_t> CONNECTION_ID_COUNTER;
     };

     std::string mServiceAddr;
     std::unique_ptr<::grpc::Server> mServer{nullptr};
     VehiclePropValuePool mValueObjectPool;
     std::unique_ptr<GarageModeServerSideHandler> mGarageModeHandler;
     PowerStateListener mPowerStateListener;
     std::thread mPowerStateListenerThread{};
     mutable std::shared_mutex mConnectionMutex;
     mutable std::shared_mutex mWriterMutex;
     std::list<ConnectionDescriptor> mValueStreamingConnections;
 };

 std::atomic<uint64_t> GrpcVehicleServerImpl::ConnectionDescriptor::CONNECTION_ID_COUNTER = 0;

 static std::shared_ptr<::grpc::ServerCredentials> getServerCredentials() {
     // TODO(chenhaosjtuacm): get secured credentials here
     return ::grpc::InsecureServerCredentials();
 }

 GrpcVehicleServerPtr makeGrpcVehicleServer(const VirtualizedVhalServerInfo& serverInfo) {
     return std::make_unique<GrpcVehicleServerImpl>(serverInfo);
 }

 GrpcVehicleServer& GrpcVehicleServerImpl::Start() {
     if (mServer) {
         LOG(WARNING) << __func__ << ": GrpcVehicleServer has already started.";
         return *this;
     }

     ::grpc::ServerBuilder builder;
     builder.RegisterService(this);
     builder.AddListeningPort(mServiceAddr, getServerCredentials());
     mServer = builder.BuildAndStart();

     CHECK(mServer) << __func__ << ": failed to create the GRPC server, "
                    << "please make sure the configuration and permissions are correct";

     mPowerStateListenerThread = std::thread([this]() { mPowerStateListener.Listen(); });
     return *this;
 }

 void GrpcVehicleServerImpl::Wait() {
     if (mServer) {
         mServer->Wait();
     }

     if (mPowerStateListenerThread.joinable()) {
         mPowerStateListenerThread.join();
     }

     mPowerStateListenerThread = {};
     mServer.reset();
 }

 GrpcVehicleServer& GrpcVehicleServerImpl::Stop() {
     if (!mServer) {
         LOG(WARNING) << __func__ << ": GrpcVehicleServer has not started.";
         return *this;
     }

     mServer->Shutdown();
     mPowerStateListener.Stop();
     return *this;
 }

 uint32_t GrpcVehicleServerImpl::NumOfActivePropertyValueStream() {
     std::shared_lock read_lock(mConnectionMutex);
     return mValueStreamingConnections.size();
 }

 void GrpcVehicleServerImpl::onPropertyValueFromCar(const VehiclePropValue& value,
                                                    bool updateStatus) {
     vhal_proto::WrappedVehiclePropValue wrappedPropValue;
     proto_msg_converter::toProto(wrappedPropValue.mutable_value(), value);
     wrappedPropValue.set_update_status(updateStatus);
     std::shared_lock read_lock(mConnectionMutex);

     bool has_terminated_connections = 0;

     for (auto& connection : mValueStreamingConnections) {
         auto writeOK = connection.mValueWriter(wrappedPropValue);
         if (!writeOK) {
             LOG(ERROR) << __func__ << ": Server Write failed, connection lost. ID: "
                        << connection.mConnectionID;
             has_terminated_connections = true;
             connection.mIsAlive.store(false);
         }
     }

     if (!has_terminated_connections) {
         return;
     }

     read_lock.unlock();

     std::unique_lock write_lock(mConnectionMutex);

     for (auto itr = mValueStreamingConnections.begin(); itr != mValueStreamingConnections.end();) {
         if (!itr->mIsAlive.load()) {
             itr = mValueStreamingConnections.erase(itr);
         } else {
             ++itr;
         }
     }
 }

 StatusCode GrpcVehicleServerImpl::onSetProperty(const VehiclePropValue& value, bool updateStatus) {
     if (value.prop == AP_POWER_STATE_REPORT &&
         value.value.int32Values[0] == toInt(VehicleApPowerStateReport::SHUTDOWN_POSTPONE)) {
         mGarageModeHandler->HandleHeartbeat();
     }
     return GrpcVehicleServer::onSetProperty(value, updateStatus);
 }

 ::grpc::Status GrpcVehicleServerImpl::GetAllPropertyConfig(
         ::grpc::ServerContext* context, const ::google::protobuf::Empty* request,
         ::grpc::ServerWriter<vhal_proto::VehiclePropConfig>* stream) {
     auto configs = onGetAllPropertyConfig();
     for (auto& config : configs) {
         vhal_proto::VehiclePropConfig protoConfig;
         proto_msg_converter::toProto(&protoConfig, config);
         if (!stream->Write(protoConfig)) {
             return ::grpc::Status(::grpc::StatusCode::ABORTED, "Connection lost.");
         }
     }

     return ::grpc::Status::OK;
 }

 ::grpc::Status GrpcVehicleServerImpl::SetProperty(
         ::grpc::ServerContext* context, const vhal_proto::WrappedVehiclePropValue* wrappedPropValue,
         vhal_proto::VehicleHalCallStatus* status) {
     VehiclePropValue value;
     proto_msg_converter::fromProto(&value, wrappedPropValue->value());

     auto set_status = static_cast<int32_t>(onSetProperty(value, wrappedPropValue->update_status()));
     if (!vhal_proto::VehicleHalStatusCode_IsValid(set_status)) {
         return ::grpc::Status(::grpc::StatusCode::INTERNAL, "Unknown status code");
     }

     status->set_status_code(static_cast<vhal_proto::VehicleHalStatusCode>(set_status));

     return ::grpc::Status::OK;
 }

 ::grpc::Status GrpcVehicleServerImpl::StartPropertyValuesStream(
         ::grpc::ServerContext* context, const ::google::protobuf::Empty* request,
         ::grpc::ServerWriter<vhal_proto::WrappedVehiclePropValue>* stream) {
     std::mutex terminateMutex;
     std::condition_variable terminateCV;
     std::unique_lock<std::mutex> terminateLock(terminateMutex);
     bool terminated{false};

     auto callBack = [stream, &terminateMutex, &terminateCV, &terminated,
                      this](const vhal_proto::WrappedVehiclePropValue& value) {
         std::unique_lock lock(mWriterMutex);
         if (!stream->Write(value)) {
             std::unique_lock<std::mutex> terminateLock(terminateMutex);
             terminated = true;
             terminateLock.unlock();
             terminateCV.notify_all();
             return false;
         }
         return true;
     };

     // Register connection
     std::unique_lock lock(mConnectionMutex);
     auto& conn = mValueStreamingConnections.emplace_back(std::move(callBack));
     lock.unlock();

     // Never stop until connection lost
     terminateCV.wait(terminateLock, [&terminated]() { return terminated; });

     LOG(ERROR) << __func__ << ": Stream lost, ID : " << conn.mConnectionID;

     return ::grpc::Status(::grpc::StatusCode::ABORTED, "Connection lost.");
 }

 }  // namespace impl

 }  // namespace V2_0
 }  // namespace vehicle
 }  // namespace automotive
 }  // namespace hardware
 }  // namespace android
	/*
	* Copyright (C) 2019 The Android Open Source Project
	*
	* 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.
	*/
	#include "GrpcVehicleServer.h"

	#include <condition_variable>
	#include <mutex>
	#include <shared_mutex>

	#include <android-base/logging.h>
	#include <grpc++/grpc++.h>

	#include "GarageModeServerSideHandler.h"
	#include "PowerStateListener.h"
	#include "VehicleServer.grpc.pb.h"
	#include "VehicleServer.pb.h"
	#include "vhal_v2_0/DefaultConfig.h"
	#include "vhal_v2_0/ProtoMessageConverter.h"

	namespace android {
	namespace hardware {
	namespace automotive {
	namespace vehicle {
	namespace V2_0 {

	namespace impl {

	class GrpcVehicleServerImpl : public GrpcVehicleServer, public vhal_proto::VehicleServer::Service {
	public:
	explicit GrpcVehicleServerImpl(const VirtualizedVhalServerInfo& serverInfo)
	: mServiceAddr(serverInfo.getServerUri()),
	mGarageModeHandler(makeGarageModeServerSideHandler(this, &mValueObjectPool,
	serverInfo.powerStateMarkerFilePath)),
	mPowerStateListener(serverInfo.powerStateSocket, serverInfo.powerStateMarkerFilePath) {
	setValuePool(&mValueObjectPool);
	}

	// method from GrpcVehicleServer
	GrpcVehicleServer& Start() override;

	void Wait() override;

	GrpcVehicleServer& Stop() override;

	uint32_t NumOfActivePropertyValueStream() override;

	// methods from IVehicleServer
	void onPropertyValueFromCar(const VehiclePropValue& value, bool updateStatus) override;

	StatusCode onSetProperty(const VehiclePropValue& value, bool updateStatus) override;

	// methods from vhal_proto::VehicleServer::Service

	::grpc::Status GetAllPropertyConfig(
	::grpc::ServerContext* context, const ::google::protobuf::Empty* request,
	::grpc::ServerWriter<vhal_proto::VehiclePropConfig>* stream) override;

	::grpc::Status SetProperty(::grpc::ServerContext* context,
	const vhal_proto::WrappedVehiclePropValue* wrappedPropValue,
	vhal_proto::VehicleHalCallStatus* status) override;

	::grpc::Status StartPropertyValuesStream(
	::grpc::ServerContext* context, const ::google::protobuf::Empty* request,
	::grpc::ServerWriter<vhal_proto::WrappedVehiclePropValue>* stream) override;

	private:
	// We keep long-lasting connection for streaming the prop values.
	// For us, each connection can be represented as a function to send the new value, and
	// an ID to identify this connection
	struct ConnectionDescriptor {
	using ValueWriterType = std::function<bool(const vhal_proto::WrappedVehiclePropValue&)>;

	explicit ConnectionDescriptor(ValueWriterType&& value_writer)
	: mValueWriter(std::move(value_writer)),
	mConnectionID(CONNECTION_ID_COUNTER.fetch_add(1)) {}

	ConnectionDescriptor(const ConnectionDescriptor&) = delete;

	ConnectionDescriptor& operator=(const ConnectionDescriptor&) = delete;

	// This move constructor is NOT THREAD-SAFE, which means it cannot be moved
	// while using. Since the connection descriptors are pretected by mConnectionMutex
	// then we are fine here
	ConnectionDescriptor(ConnectionDescriptor&& cd)
	: mValueWriter(std::move(cd.mValueWriter)),
	mConnectionID(cd.mConnectionID),
	mIsAlive(cd.mIsAlive.load()) {
	cd.mIsAlive.store(false);
	}

	ValueWriterType mValueWriter;
	uint64_t mConnectionID;
	std::atomic<bool> mIsAlive{true};

	static std::atomic<uint64_t> CONNECTION_ID_COUNTER;
	};

	std::string mServiceAddr;
	std::unique_ptr<::grpc::Server> mServer{nullptr};
	VehiclePropValuePool mValueObjectPool;
	std::unique_ptr<GarageModeServerSideHandler> mGarageModeHandler;
	PowerStateListener mPowerStateListener;
	std::thread mPowerStateListenerThread{};
	mutable std::shared_mutex mConnectionMutex;
	mutable std::shared_mutex mWriterMutex;
	std::list<ConnectionDescriptor> mValueStreamingConnections;
	};

	std::atomic<uint64_t> GrpcVehicleServerImpl::ConnectionDescriptor::CONNECTION_ID_COUNTER = 0;

	static std::shared_ptr<::grpc::ServerCredentials> getServerCredentials() {
	// TODO(chenhaosjtuacm): get secured credentials here
	return ::grpc::InsecureServerCredentials();
	}

	GrpcVehicleServerPtr makeGrpcVehicleServer(const VirtualizedVhalServerInfo& serverInfo) {
	return std::make_unique<GrpcVehicleServerImpl>(serverInfo);
	}

	GrpcVehicleServer& GrpcVehicleServerImpl::Start() {
	if (mServer) {
	LOG(WARNING) << __func__ << ": GrpcVehicleServer has already started.";
	return *this;
	}

	::grpc::ServerBuilder builder;
	builder.RegisterService(this);
	builder.AddListeningPort(mServiceAddr, getServerCredentials());
	mServer = builder.BuildAndStart();

	CHECK(mServer) << __func__ << ": failed to create the GRPC server, "
	<< "please make sure the configuration and permissions are correct";

	mPowerStateListenerThread = std::thread([this]() { mPowerStateListener.Listen(); });
	return *this;
	}

	void GrpcVehicleServerImpl::Wait() {
	if (mServer) {
	mServer->Wait();
	}

	if (mPowerStateListenerThread.joinable()) {
	mPowerStateListenerThread.join();
	}

	mPowerStateListenerThread = {};
	mServer.reset();
	}

	GrpcVehicleServer& GrpcVehicleServerImpl::Stop() {
	if (!mServer) {
	LOG(WARNING) << __func__ << ": GrpcVehicleServer has not started.";
	return *this;
	}

	mServer->Shutdown();
	mPowerStateListener.Stop();
	return *this;
	}

	uint32_t GrpcVehicleServerImpl::NumOfActivePropertyValueStream() {
	std::shared_lock read_lock(mConnectionMutex);
	return mValueStreamingConnections.size();
	}

	void GrpcVehicleServerImpl::onPropertyValueFromCar(const VehiclePropValue& value,
	bool updateStatus) {
	vhal_proto::WrappedVehiclePropValue wrappedPropValue;
	proto_msg_converter::toProto(wrappedPropValue.mutable_value(), value);
	wrappedPropValue.set_update_status(updateStatus);
	std::shared_lock read_lock(mConnectionMutex);

	bool has_terminated_connections = 0;

	for (auto& connection : mValueStreamingConnections) {
	auto writeOK = connection.mValueWriter(wrappedPropValue);
	if (!writeOK) {
	LOG(ERROR) << __func__ << ": Server Write failed, connection lost. ID: "
	<< connection.mConnectionID;
	has_terminated_connections = true;
	connection.mIsAlive.store(false);
	}
	}

	if (!has_terminated_connections) {
	return;
	}

	read_lock.unlock();

	std::unique_lock write_lock(mConnectionMutex);

	for (auto itr = mValueStreamingConnections.begin(); itr != mValueStreamingConnections.end();) {
	if (!itr->mIsAlive.load()) {
	itr = mValueStreamingConnections.erase(itr);
	} else {
	++itr;
	}
	}
	}

	StatusCode GrpcVehicleServerImpl::onSetProperty(const VehiclePropValue& value, bool updateStatus) {
	if (value.prop == AP_POWER_STATE_REPORT &&
	value.value.int32Values[0] == toInt(VehicleApPowerStateReport::SHUTDOWN_POSTPONE)) {
	mGarageModeHandler->HandleHeartbeat();
	}
	return GrpcVehicleServer::onSetProperty(value, updateStatus);
	}

	::grpc::Status GrpcVehicleServerImpl::GetAllPropertyConfig(
	::grpc::ServerContext* context, const ::google::protobuf::Empty* request,
	::grpc::ServerWriter<vhal_proto::VehiclePropConfig>* stream) {
	auto configs = onGetAllPropertyConfig();
	for (auto& config : configs) {
	vhal_proto::VehiclePropConfig protoConfig;
	proto_msg_converter::toProto(&protoConfig, config);
	if (!stream->Write(protoConfig)) {
	return ::grpc::Status(::grpc::StatusCode::ABORTED, "Connection lost.");
	}
	}

	return ::grpc::Status::OK;
	}

	::grpc::Status GrpcVehicleServerImpl::SetProperty(
	::grpc::ServerContext* context, const vhal_proto::WrappedVehiclePropValue* wrappedPropValue,
	vhal_proto::VehicleHalCallStatus* status) {
	VehiclePropValue value;
	proto_msg_converter::fromProto(&value, wrappedPropValue->value());

	auto set_status = static_cast<int32_t>(onSetProperty(value, wrappedPropValue->update_status()));
	if (!vhal_proto::VehicleHalStatusCode_IsValid(set_status)) {
	return ::grpc::Status(::grpc::StatusCode::INTERNAL, "Unknown status code");
	}

	status->set_status_code(static_cast<vhal_proto::VehicleHalStatusCode>(set_status));

	return ::grpc::Status::OK;
	}

	::grpc::Status GrpcVehicleServerImpl::StartPropertyValuesStream(
	::grpc::ServerContext* context, const ::google::protobuf::Empty* request,
	::grpc::ServerWriter<vhal_proto::WrappedVehiclePropValue>* stream) {
	std::mutex terminateMutex;
	std::condition_variable terminateCV;
	std::unique_lock<std::mutex> terminateLock(terminateMutex);
	bool terminated{false};

	auto callBack = [stream, &terminateMutex, &terminateCV, &terminated,
	this](const vhal_proto::WrappedVehiclePropValue& value) {
	std::unique_lock lock(mWriterMutex);
	if (!stream->Write(value)) {
	std::unique_lock<std::mutex> terminateLock(terminateMutex);
	terminated = true;
	terminateLock.unlock();
	terminateCV.notify_all();
	return false;
	}
	return true;
	};

	// Register connection
	std::unique_lock lock(mConnectionMutex);
	auto& conn = mValueStreamingConnections.emplace_back(std::move(callBack));
	lock.unlock();

	// Never stop until connection lost
	terminateCV.wait(terminateLock, [&terminated]() { return terminated; });

	LOG(ERROR) << __func__ << ": Stream lost, ID : " << conn.mConnectionID;

	return ::grpc::Status(::grpc::StatusCode::ABORTED, "Connection lost.");
	}

	} // namespace impl

	} // namespace V2_0
	} // namespace vehicle
	} // namespace automotive
	} // namespace hardware
	} // namespace android