From c4e2e20d488c8a816c2d6c1c9f94564491cd4673 Mon Sep 17 00:00:00 2001 From: xxxx Date: Thu, 30 Jun 2022 21:42:43 +0800 Subject: [PATCH] Description:add clock Match-id-430c480195ada4b4c1404cbcb15efe7e7c43f0d5 --- common/include/input_hub.cpp | 159 +++++++++++------- common/include/input_hub.h | 25 +-- .../include/distributed_input_sink_manager.h | 2 +- .../include/distributed_input_sink_switch.h | 1 - .../distributed_input_sink_transport.h | 2 - .../src/distributed_input_sink_transport.cpp | 8 +- .../distributed_input_source_manager.h | 1 + .../src/distributed_input_source_manager.cpp | 7 +- 8 files changed, 119 insertions(+), 86 deletions(-) diff --git a/common/include/input_hub.cpp b/common/include/input_hub.cpp index dc4330d..fc9a538 100644 --- a/common/include/input_hub.cpp +++ b/common/include/input_hub.cpp @@ -103,24 +103,28 @@ int32_t InputHub::Release() size_t InputHub::StartCollectInputEvents(RawEvent* buffer, size_t bufferSize) { - size_t count; - isStartCollectEvent_.store(true); - while (isStartCollectEvent_.load()) { + size_t count = 0; + isStartCollectEvent_ = true; + while (isStartCollectEvent_) { if (needToScanDevices_) { needToScanDevices_ = false; ScanInputDevices(DEVICE_PATH); } - while (!openingDevices_.empty()) { - std::unique_lock deviceLock(visitMutex_); - std::unique_ptr device = std::move(*openingDevices_.rbegin()); - openingDevices_.pop_back(); - DHLOGI("Reporting device opened: id=%s, name=%s\n", - GetAnonyInt32(device->id).c_str(), device->path.c_str()); - auto [dev_it, inserted] = devices_.insert_or_assign(device->id, std::move(device)); - if (!inserted) { - DHLOGI("Device id %s exists, replaced. \n", GetAnonyInt32(device->id).c_str()); + + { + std::unique_lock deviceLock(openingDevicesMutex_); + while (!openingDevices_.empty()) { + std::unique_ptr device = std::move(*openingDevices_.rbegin()); + openingDevices_.pop_back(); + DHLOGI("Reporting device opened: id=%s, name=%s\n", + GetAnonyInt32(device->id).c_str(), device->path.c_str()); + auto [dev_it, inserted] = devices_.insert_or_assign(device->id, std::move(device)); + if (!inserted) { + DHLOGI("Device id %s exists, replaced. \n", GetAnonyInt32(device->id).c_str()); + } } } + deviceChanged_ = false; count = GetEvents(buffer, bufferSize); // readNotify() will modify the list of devices so this must be done after @@ -152,7 +156,8 @@ size_t InputHub::StartCollectInputEvents(RawEvent* buffer, size_t bufferSize) void InputHub::StopCollectInputEvents() { - isStartCollectEvent_.store(false); + DHLOGI("Stop Collect Input Events Thread"); + isStartCollectEvent_ = false; } size_t InputHub::GetEvents(RawEvent* buffer, size_t bufferSize) @@ -170,7 +175,7 @@ size_t InputHub::GetEvents(RawEvent* buffer, size_t bufferSize) } continue; } - Device* device = GetDeviceByFd(eventItem.data.fd); + Device* device = GetSupportDeviceByFd(eventItem.data.fd); if (!device) { continue; } @@ -232,48 +237,57 @@ size_t InputHub::DeviceIsExists(InputDeviceEvent* buffer, size_t bufferSize) InputDeviceEvent* event = buffer; size_t capacity = bufferSize; // Report any devices that had last been added/removed. - for (auto it = closingDevices_.begin(); it != closingDevices_.end();) { - std::unique_ptr device = std::move(*it); - DHLOGI("Reporting device closed: id=%s, name=%s\n", GetAnonyInt32(device->id).c_str(), device->path.c_str()); - event->type = DeviceType::DEVICE_REMOVED; - event->deviceInfo = device->identifier; - event += 1; - it = closingDevices_.erase(it); - if (capacity == 0) { - break; + { + std::unique_lock my_lock(closingDevicesMutex_); + for (auto it = closingDevices_.begin(); it != closingDevices_.end();) { + std::unique_ptr device = std::move(*it); + DHLOGI("Reporting device closed: id=%s, name=%s\n", + GetAnonyInt32(device->id).c_str(), device->path.c_str()); + event->type = DeviceType::DEVICE_REMOVED; + event->deviceInfo = device->identifier; + event += 1; + it = closingDevices_.erase(it); + if (capacity == 0) { + break; + } + capacity--; } - capacity--; } + if (needToScanDevices_) { needToScanDevices_ = false; ScanInputDevices(DEVICE_PATH); } - while (!openingDevices_.empty()) { - std::unique_lock deviceLock(visitMutex_); - std::unique_ptr device = std::move(*openingDevices_.rbegin()); - openingDevices_.pop_back(); - DHLOGI("Reporting device opened: id=%s, name=%s\n", GetAnonyInt32(device->id).c_str(), device->path.c_str()); - event->type = DeviceType::DEVICE_ADDED; - event->deviceInfo = device->identifier; - event += 1; - auto [dev_it, inserted] = devices_.insert_or_assign(device->id, std::move(device)); - if (!inserted) { - DHLOGI("Device id %s exists, replaced. \n", GetAnonyInt32(device->id).c_str()); + { + std::unique_lock deviceLock(openingDevicesMutex_); + while (!openingDevices_.empty()) { + std::unique_ptr device = std::move(*openingDevices_.rbegin()); + openingDevices_.pop_back(); + DHLOGI("Reporting device opened: id=%s, name=%s\n", + GetAnonyInt32(device->id).c_str(), device->path.c_str()); + event->type = DeviceType::DEVICE_ADDED; + event->deviceInfo = device->identifier; + event += 1; + + auto [dev_it, inserted] = devices_.insert_or_assign(device->id, std::move(device)); + if (!inserted) { + DHLOGI("Device id %s exists, replaced. \n", GetAnonyInt32(device->id).c_str()); + } + if (capacity == 0) { + break; + } + capacity--; } - if (capacity == 0) { - break; - } - capacity--; } return event - buffer; } size_t InputHub::StartCollectInputHandler(InputDeviceEvent* buffer, size_t bufferSize) { - size_t count; - isStartCollectHandler_.store(true); - while (isStartCollectHandler_.load()) { + size_t count = 0; + isStartCollectHandler_ = true; + while (isStartCollectHandler_) { count = DeviceIsExists(buffer, bufferSize); deviceChanged_ = false; GetDeviceHandler(); @@ -302,7 +316,8 @@ size_t InputHub::StartCollectInputHandler(InputDeviceEvent* buffer, size_t buffe void InputHub::StopCollectInputHandler() { - isStartCollectHandler_.store(false); + DHLOGI("Stop Collect Input Handler Thread"); + isStartCollectHandler_ = false; } void InputHub::GetDeviceHandler() @@ -359,14 +374,14 @@ int32_t InputHub::RefreshEpollItem() } } else { // Some events occurred. - pendingEventCount_ = size_t(pollResult); + pendingEventCount_ = pollResult; } return DH_SUCCESS; } std::vector InputHub::GetAllInputDevices() { - std::unique_lock deviceLock(visitMutex_); + std::unique_lock deviceLock(devicesMutex_); std::vector vecDevice; for (const auto& [id, device] : devices_) { vecDevice.push_back(device->identifier); @@ -380,7 +395,7 @@ void InputHub::ScanInputDevices(const std::string& dirname) struct dirent *de; dir = opendir(dirname.c_str()); if (dir == nullptr) { - DHLOGE("error opendir dev/input :%{public}s\n", ConvertErrNo().c_str()); + DHLOGE("error opendir /dev/input :%{public}s\n", ConvertErrNo().c_str()); return; } size_t dirNameFirstPos = 0; @@ -404,7 +419,7 @@ void InputHub::ScanInputDevices(const std::string& dirname) int32_t InputHub::OpenInputDeviceLocked(const std::string& devicePath) { { - std::unique_lock deviceLock(visitMutex_); + std::unique_lock deviceLock(devicesMutex_); for (const auto& [deviceId, device] : devices_) { if (device->path == devicePath) { return DH_SUCCESS; // device was already registered @@ -653,20 +668,35 @@ int32_t InputHub::RegisterFdForEpoll(int fd) void InputHub::AddDeviceLocked(std::unique_ptr device) { - std::unique_lock deviceLock(visitMutex_); + std::unique_lock deviceLock(openingDevicesMutex_); openingDevices_.push_back(std::move(device)); } void InputHub::CloseDeviceLocked(Device& device) { - DHLOGI( - "Removed device: path=%s name=%s id=%s fd=%d classes=0x%x", + DHLOGI("Removed device: path=%s name=%s id=%s fd=%d classes=0x%x", device.path.c_str(), device.identifier.name.c_str(), GetAnonyInt32(device.id).c_str(), device.fd, device.classes); UnregisterDeviceFromEpollLocked(device); device.Close(); - std::unique_lock deviceLock(visitMutex_); + { + std::unique_lock devicesLock(devicesMutex_); + std::unique_lock closingDevicesLock(closingDevicesMutex_); + closingDevices_.push_back(std::move(devices_[device.id])); + devices_.erase(device.id); + } +} + +void InputHub::CloseDeviceForAllLocked(Device& device) +{ + DHLOGI("Removed device: path=%s name=%s id=%s fd=%d classes=0x%x", + device.path.c_str(), device.identifier.name.c_str(), GetAnonyInt32(device.id).c_str(), + device.fd, device.classes); + + UnregisterDeviceFromEpollLocked(device); + device.Close(); + std::unique_lock deviceLock(closingDevicesMutex_); closingDevices_.push_back(std::move(devices_[device.id])); devices_.erase(device.id); } @@ -676,8 +706,8 @@ int32_t InputHub::UnregisterDeviceFromEpollLocked(const Device& device) const if (device.HasValidFd()) { int32_t result = UnregisterFdFromEpoll(device.fd); if (result != DH_SUCCESS) { - DHLOGE( - "Could not remove input device fd from epoll for device %s", GetAnonyInt32(device.id).c_str()); + DHLOGE("Could not remove input device fd from epoll for device %s", + GetAnonyInt32(device.id).c_str()); return result; } } @@ -687,8 +717,7 @@ int32_t InputHub::UnregisterDeviceFromEpollLocked(const Device& device) const int32_t InputHub::UnregisterFdFromEpoll(int fd) const { if (epoll_ctl(epollFd_, EPOLL_CTL_DEL, fd, nullptr)) { - DHLOGE( - "Could not remove fd from epoll instance: %s", ConvertErrNo().c_str()); + DHLOGE("Could not remove fd from epoll instance: %s", ConvertErrNo().c_str()); return ERR_DH_INPUT_HUB_UNREGISTER_FD_FAIL; } return DH_SUCCESS; @@ -752,14 +781,16 @@ void InputHub::CloseDeviceByPathLocked(const std::string& devicePath) void InputHub::CloseAllDevicesLocked() { + DHLOGI("Close All Devices"); + std::unique_lock deviceLock(devicesMutex_); while (!devices_.empty()) { - CloseDeviceLocked(*(devices_.begin()->second)); + CloseDeviceForAllLocked(*(devices_.begin()->second)); } } InputHub::Device* InputHub::GetDeviceByDescriptorLocked(const std::string& descriptor) { - std::unique_lock deviceLock(visitMutex_); + std::unique_lock deviceLock(devicesMutex_); for (const auto& [id, device] : devices_) { if (descriptor == device->identifier.descriptor) { return device.get(); @@ -770,7 +801,7 @@ InputHub::Device* InputHub::GetDeviceByDescriptorLocked(const std::string& descr InputHub::Device* InputHub::GetDeviceByPathLocked(const std::string& devicePath) { - std::unique_lock deviceLock(visitMutex_); + std::unique_lock deviceLock(devicesMutex_); for (const auto& [id, device] : devices_) { if (device->path == devicePath) { return device.get(); @@ -781,7 +812,7 @@ InputHub::Device* InputHub::GetDeviceByPathLocked(const std::string& devicePath) InputHub::Device* InputHub::GetDeviceByFdLocked(int fd) { - std::unique_lock deviceLock(visitMutex_); + std::unique_lock deviceLock(devicesMutex_); for (const auto& [id, device] : devices_) { if (device->fd == fd) { return device.get(); @@ -790,12 +821,12 @@ InputHub::Device* InputHub::GetDeviceByFdLocked(int fd) return nullptr; } -InputHub::Device* InputHub::GetDeviceByFd(int fd) +InputHub::Device* InputHub::GetSupportDeviceByFd(int fd) { - std::unique_lock deviceLock(visitMutex_); + std::unique_lock deviceLock(devicesMutex_); for (const auto& [id, device] : devices_) { if (device->fd == fd) { - if (GetIsSupportInputTypes(device->classes)) { + if (IsSupportInputTypes(device->classes)) { return device.get(); } } @@ -834,14 +865,14 @@ uint32_t InputHub::SizeofBitArray(uint32_t bit) return ((bit) + round) / divisor; } -bool InputHub::GetIsSupportInputTypes(uint32_t classes) +bool InputHub::IsSupportInputTypes(uint32_t classes) { return classes & inputTypes_; } void InputHub::SetSupportInputType(const uint32_t& inputTypes) { inputTypes_ = inputTypes; - DHLOGI("SetSupportInputType: inputTypes=0x%x,", inputTypes_); + DHLOGI("SetSupportInputType: inputTypes=0x%x,", inputTypes_.load()); } void InputHub::RecordEventLog(const RawEvent* event) diff --git a/common/include/input_hub.h b/common/include/input_hub.h index 5795148..3a9c162 100644 --- a/common/include/input_hub.h +++ b/common/include/input_hub.h @@ -85,6 +85,7 @@ private: int32_t RegisterDeviceForEpollLocked(const Device& device); void AddDeviceLocked(std::unique_ptr device); void CloseDeviceLocked(Device& device); + void CloseDeviceForAllLocked(Device& device); int32_t UnregisterDeviceFromEpollLocked(const Device& device) const; int32_t UnregisterFdFromEpoll(int fd) const; int32_t ReadNotifyLocked(); @@ -94,7 +95,7 @@ private: Device* GetDeviceByDescriptorLocked(const std::string& descriptor); Device* GetDeviceByPathLocked(const std::string& devicePath); Device* GetDeviceByFdLocked(int fd); - Device* GetDeviceByFd(int fd); + Device* GetSupportDeviceByFd(int fd); bool ContainsNonZeroByte(const uint8_t* array, uint32_t startIndex, uint32_t endIndex); int64_t ProcessEventTimestamp(const input_event& event); @@ -106,7 +107,7 @@ private: bool TestBit(uint32_t bit, const uint8_t* array); /* this macro computes the number of bytes needed to represent a bit array of the specified size */ uint32_t SizeofBitArray(uint32_t bit); - bool GetIsSupportInputTypes(uint32_t classes); + bool IsSupportInputTypes(uint32_t classes); void RecordEventLog(const RawEvent* event); int epollFd_; @@ -114,21 +115,25 @@ private: int inputWd_; std::vector> openingDevices_; + std::mutex openingDevicesMutex_; std::vector> closingDevices_; + std::mutex closingDevicesMutex_; std::unordered_map> devices_; - bool needToScanDevices_; + std::mutex devicesMutex_; + + std::atomic needToScanDevices_; std::string deviceId_; - int32_t nextDeviceId_; + std::atomic nextDeviceId_; // The array of pending epoll events and the index of the next event to be handled. struct epoll_event mPendingEventItems[EPOLL_MAX_EVENTS]; - size_t pendingEventCount_; - size_t pendingEventIndex_; - bool pendingINotify_; + std::atomic pendingEventCount_; + std::atomic pendingEventIndex_; + std::atomic pendingINotify_; std::mutex operationMutex_; - std::mutex visitMutex_; - bool deviceChanged_; - uint32_t inputTypes_; + + std::atomic deviceChanged_; + std::atomic inputTypes_; std::atomic isStartCollectEvent_; std::atomic isStartCollectHandler_; }; diff --git a/services/sink/sinkmanager/include/distributed_input_sink_manager.h b/services/sink/sinkmanager/include/distributed_input_sink_manager.h index 0be5160..d545361 100644 --- a/services/sink/sinkmanager/include/distributed_input_sink_manager.h +++ b/services/sink/sinkmanager/include/distributed_input_sink_manager.h @@ -94,7 +94,7 @@ private: ServiceSinkRunningState serviceRunningState_ = ServiceSinkRunningState::STATE_NOT_START; DInputServerType isStartTrans_ = DInputServerType::NULL_SERVER_TYPE; std::shared_ptr statuslistener_; - sptr startServerCallback_ = nullptr; + sptr startServerCallback_ = nullptr; std::shared_ptr runner_; std::shared_ptr handler_; diff --git a/services/sink/transport/include/distributed_input_sink_switch.h b/services/sink/transport/include/distributed_input_sink_switch.h index 4d7a68f..1af5b68 100644 --- a/services/sink/transport/include/distributed_input_sink_switch.h +++ b/services/sink/transport/include/distributed_input_sink_switch.h @@ -42,7 +42,6 @@ public: int32_t GetSwitchOpenedSession(); private: - std::mutex operationMutex_; std::vector switchVector_; }; diff --git a/services/sink/transport/include/distributed_input_sink_transport.h b/services/sink/transport/include/distributed_input_sink_transport.h index 0be7299..e6eb250 100644 --- a/services/sink/transport/include/distributed_input_sink_transport.h +++ b/services/sink/transport/include/distributed_input_sink_transport.h @@ -72,8 +72,6 @@ private: private: std::string deviceId_; - std::set sessionIdSet_; - std::map sessionDevMap_; std::string mySessionName_; std::shared_ptr eventHandler_; diff --git a/services/sink/transport/src/distributed_input_sink_transport.cpp b/services/sink/transport/src/distributed_input_sink_transport.cpp index c8645b6..cd0cac3 100644 --- a/services/sink/transport/src/distributed_input_sink_transport.cpp +++ b/services/sink/transport/src/distributed_input_sink_transport.cpp @@ -43,7 +43,6 @@ DistributedInputSinkTransport::DistributedInputSinkTransport() DistributedInputSinkTransport::~DistributedInputSinkTransport() { DHLOGI("~DistributedInputSinkTransport"); - sessionDevMap_.clear(); (void)RemoveSessionServer(DINPUT_PKG_NAME.c_str(), mySessionName_.c_str()); } @@ -253,9 +252,6 @@ int32_t DistributedInputSinkTransport::OnSessionOpened(int32_t sessionId, int32_ { if (result != DH_SUCCESS) { DHLOGE("session open failed, sessionId %s", GetAnonyInt32(sessionId).c_str()); - if (sessionIdSet_.count(sessionId) > 0) { - sessionIdSet_.erase(sessionId); - } return DH_SUCCESS; } @@ -290,9 +286,7 @@ int32_t DistributedInputSinkTransport::OnSessionOpened(int32_t sessionId, int32_ void DistributedInputSinkTransport::OnSessionClosed(int32_t sessionId) { DHLOGI("OnSessionClosed, sessionId:%s", GetAnonyInt32(sessionId).c_str()); - if (sessionIdSet_.count(sessionId) > 0) { - sessionIdSet_.erase(sessionId); - } + char peerDevId[DEVICE_ID_SIZE_MAX] = ""; int ret = GetPeerDeviceId(sessionId, peerDevId, sizeof(peerDevId)); if (ret != DH_SUCCESS) { diff --git a/services/source/sourcemanager/include/distributed_input_source_manager.h b/services/source/sourcemanager/include/distributed_input_source_manager.h index 6069eff..d8e9607 100644 --- a/services/source/sourcemanager/include/distributed_input_source_manager.h +++ b/services/source/sourcemanager/include/distributed_input_source_manager.h @@ -236,6 +236,7 @@ private: std::vector inputDevice_; bool InitAuto(); void handleStartServerCallback(const std::string& devId); + std::mutex operationMutex_; }; } // namespace DistributedInput } // namespace DistributedHardware diff --git a/services/source/sourcemanager/src/distributed_input_source_manager.cpp b/services/source/sourcemanager/src/distributed_input_source_manager.cpp index a515ffd..afa65fb 100644 --- a/services/source/sourcemanager/src/distributed_input_source_manager.cpp +++ b/services/source/sourcemanager/src/distributed_input_source_manager.cpp @@ -540,7 +540,6 @@ int32_t DistributedInputSourceManager::RegisterDistributedHardware(const std::st HisyseventUtil::GetInstance().SysEventWriteBehavior(DINPUT_REGISTER, devId, dhId, "dinput register call."); DHLOGI("%s called, deviceId: %s, dhId: %s, parameters: %s", __func__, GetAnonyString(devId).c_str(), GetAnonyString(dhId).c_str(), parameters.c_str()); - if (callback == nullptr) { DHLOGE( "%s called, deviceId: %s callback is null.", @@ -551,6 +550,8 @@ int32_t DistributedInputSourceManager::RegisterDistributedHardware(const std::st return ERR_DH_INPUT_SERVER_SOURCE_MANAGER_REGISTER_FAIL; } + std::lock_guard lock(operationMutex_); + DInputClientRegistInfo info; info.devId = devId; info.dhId = dhId; @@ -679,6 +680,8 @@ int32_t DistributedInputSourceManager::UnregisterDistributedHardware(const std:: return ERR_DH_INPUT_SERVER_SOURCE_MANAGER_UNREGISTER_FAIL; } + std::lock_guard lock(operationMutex_); + DInputClientUnregistInfo info; info.devId = devId; info.dhId = dhId; @@ -950,6 +953,7 @@ int32_t DistributedInputSourceManager::IsStartDistributedInput( void DistributedInputSourceManager::RunRegisterCallback( const std::string& devId, const std::string& dhId, const int32_t& status) { + std::lock_guard lock(operationMutex_); for (std::vector::iterator iter = regCallbacks_.begin(); iter != regCallbacks_.end(); iter++) { if (iter->devId == devId && iter->dhId == dhId) { @@ -966,6 +970,7 @@ void DistributedInputSourceManager::RunRegisterCallback( void DistributedInputSourceManager::RunUnregisterCallback( const std::string& devId, const std::string& dhId, const int32_t& status) { + std::lock_guard lock(operationMutex_); for (std::vector::iterator iter = unregCallbacks_.begin(); iter != unregCallbacks_.end(); iter++) { if (iter->devId == devId && iter->dhId == dhId) {