diff --git a/common/include/constants_dinput.h b/common/include/constants_dinput.h index 9c4a22b..22d182a 100644 --- a/common/include/constants_dinput.h +++ b/common/include/constants_dinput.h @@ -86,7 +86,12 @@ namespace DistributedInput { /** * Maximum number of signalled FDs to handle at a time. */ - const uint32_t EPOLL_MAX_EVENTS = 16; + constexpr uint32_t EPOLL_MAX_EVENTS = 16; + + /** + * Maximum number of event buffer size. + */ + constexpr uint32_t INPUT_EVENT_BUFFER_SIZE = 256; /** * Input Type ALL. diff --git a/common/include/input_hub.cpp b/common/include/input_hub.cpp index 7bb8b6a..0907bcb 100644 --- a/common/include/input_hub.cpp +++ b/common/include/input_hub.cpp @@ -44,7 +44,7 @@ const uint32_t ERROR_MSG_MAX_LEN = 256; InputHub::InputHub() : epollFd_(0), iNotifyFd_(0), inputWd_(0), needToScanDevices_(true), nextDeviceId_(1), mPendingEventItems{0x00}, pendingEventCount_(0), pendingEventIndex_(0), pendingINotify_(false), - deviceChanged_(false) + deviceChanged_(false), isStartCollectEvent_(false), isStartCollectHandler_(false) { Initialize(); } @@ -96,13 +96,16 @@ int32_t InputHub::Release() ::close(epollFd_); ::close(iNotifyFd_); + StopCollectInputEvents(); + StopCollectInputHandler(); return DH_SUCCESS; } -size_t InputHub::CollectInputEvents(RawEvent* buffer, size_t bufferSize) +size_t InputHub::StartCollectInputEvents(RawEvent* buffer, size_t bufferSize) { size_t count; - for (;;) { + isStartCollectEvent_.store(true); + while (isStartCollectEvent_.load()) { if (needToScanDevices_) { needToScanDevices_ = false; ScanInputDevices(DEVICE_PATH); @@ -147,6 +150,11 @@ size_t InputHub::CollectInputEvents(RawEvent* buffer, size_t bufferSize) return count; } +void InputHub::StopCollectInputEvents() +{ + isStartCollectEvent_.store(false); +} + size_t InputHub::GetEvents(RawEvent* buffer, size_t bufferSize) { RawEvent* event = buffer; @@ -261,10 +269,11 @@ size_t InputHub::DeviceIsExists(InputDeviceEvent* buffer, size_t bufferSize) return event - buffer; } -size_t InputHub::CollectInputHandler(InputDeviceEvent* buffer, size_t bufferSize) +size_t InputHub::StartCollectInputHandler(InputDeviceEvent* buffer, size_t bufferSize) { size_t count; - for (;;) { + isStartCollectHandler_.store(true); + while (isStartCollectHandler_.load()) { count = DeviceIsExists(buffer, bufferSize); deviceChanged_ = false; GetDeviceHandler(); @@ -291,6 +300,11 @@ size_t InputHub::CollectInputHandler(InputDeviceEvent* buffer, size_t bufferSize return count; } +void InputHub::StopCollectInputHandler() +{ + isStartCollectHandler_.store(false); +} + void InputHub::GetDeviceHandler() { while (pendingEventIndex_ < pendingEventCount_) { diff --git a/common/include/input_hub.h b/common/include/input_hub.h index 46db599..5795148 100644 --- a/common/include/input_hub.h +++ b/common/include/input_hub.h @@ -16,6 +16,7 @@ #ifndef INPUT_HUB_H #define INPUT_HUB_H +#include #include #include @@ -33,8 +34,10 @@ class InputHub { public: InputHub(); ~InputHub(); - size_t CollectInputEvents(RawEvent* buffer, size_t bufferSize); - size_t CollectInputHandler(InputDeviceEvent* buffer, size_t bufferSize); + size_t StartCollectInputEvents(RawEvent* buffer, size_t bufferSize); + size_t StartCollectInputHandler(InputDeviceEvent* buffer, size_t bufferSize); + void StopCollectInputEvents(); + void StopCollectInputHandler(); size_t DeviceIsExists(InputDeviceEvent* event, size_t capacity); std::vector GetAllInputDevices(); void SetSupportInputType(const uint32_t& inputType); @@ -126,6 +129,8 @@ private: std::mutex visitMutex_; bool deviceChanged_; uint32_t inputTypes_; + std::atomic isStartCollectEvent_; + std::atomic isStartCollectHandler_; }; } // namespace DistributedInput } // namespace DistributedHardware diff --git a/inputdevicehandler/src/distributed_input_handler.cpp b/inputdevicehandler/src/distributed_input_handler.cpp index 453cb6c..b62781a 100644 --- a/inputdevicehandler/src/distributed_input_handler.cpp +++ b/inputdevicehandler/src/distributed_input_handler.cpp @@ -167,7 +167,7 @@ void *DistributedInputHandler::CollectEventsThread(void *param) void DistributedInputHandler::StartInputMonitorDeviceThread(const std::string deviceId) { while (isCollectingEvents_) { - size_t count = inputHub_->CollectInputHandler(mEventBuffer, INPUT_DEVICR_BUFFER_SIZE); + size_t count = inputHub_->StartCollectInputHandler(mEventBuffer, INPUT_DEVICR_BUFFER_SIZE); if (count > 0) { DHLOGI("Count: %zu", count); for (size_t iCnt = 0; iCnt < count; iCnt++) { @@ -205,6 +205,7 @@ void DistributedInputHandler::StopInputMonitorDeviceThread() { isCollectingEvents_ = false; isStartCollectEventThread = false; + inputHub_->StopCollectInputHandler(); if (collectThreadID_ != (pthread_t)(-1)) { DHLOGI("DistributedInputHandler::Wait collect thread exit"); pthread_join(collectThreadID_, NULL); diff --git a/services/sink/inputcollector/include/distributed_input_collector.h b/services/sink/inputcollector/include/distributed_input_collector.h index e637d18..188145a 100644 --- a/services/sink/inputcollector/include/distributed_input_collector.h +++ b/services/sink/inputcollector/include/distributed_input_collector.h @@ -16,6 +16,7 @@ #ifndef DISTRIBUTED_INPUT_COLLECTOR_H #define DISTRIBUTED_INPUT_COLLECTOR_H +#include #include #include #include @@ -35,6 +36,7 @@ class DistributedInputCollector { public: static DistributedInputCollector &GetInstance(); int32_t Init(std::shared_ptr sinkHandler); + void Release(); void SetInputTypes(const uint32_t& inputType); private: @@ -46,8 +48,6 @@ private: void StartCollectEventsThread(); void StopCollectEventsThread(); - // The event queue. - static const int INPUT_EVENT_BUFFER_SIZE = 256; RawEvent mEventBuffer[INPUT_EVENT_BUFFER_SIZE]; pthread_t collectThreadID_; bool isCollectingEvents_; diff --git a/services/sink/inputcollector/src/distributed_input_collector.cpp b/services/sink/inputcollector/src/distributed_input_collector.cpp index 57f33ae..20c8f70 100644 --- a/services/sink/inputcollector/src/distributed_input_collector.cpp +++ b/services/sink/inputcollector/src/distributed_input_collector.cpp @@ -96,7 +96,7 @@ void *DistributedInputCollector::CollectEventsThread(void *param) void DistributedInputCollector::StartCollectEventsThread() { while (isCollectingEvents_) { - size_t count = inputHub_->CollectInputEvents(mEventBuffer, INPUT_EVENT_BUFFER_SIZE); + size_t count = inputHub_->StartCollectInputEvents(mEventBuffer, INPUT_EVENT_BUFFER_SIZE); if (count > 0) { DHLOGI("Count: %zu", count); } else { @@ -130,6 +130,7 @@ void DistributedInputCollector::StopCollectEventsThread() { isCollectingEvents_ = false; isStartGetDeviceHandlerThread = false; + inputHub_->StopCollectInputEvents(); if (collectThreadID_ != (pthread_t)(-1)) { DHLOGI("DistributedInputCollector::Wait collect thread exit"); pthread_join(collectThreadID_, NULL); @@ -151,6 +152,11 @@ void DistributedInputCollector::SetInputTypes(const uint32_t& inputType) } inputHub_->SetSupportInputType(inputTypes_); } + +void DistributedInputCollector::Release() +{ + StopCollectEventsThread(); +} } // namespace DistributedInput } // namespace DistributedHardware } // namespace OHOS diff --git a/services/sink/sinkmanager/src/distributed_input_sink_manager.cpp b/services/sink/sinkmanager/src/distributed_input_sink_manager.cpp index ee1f814..6f39f8a 100644 --- a/services/sink/sinkmanager/src/distributed_input_sink_manager.cpp +++ b/services/sink/sinkmanager/src/distributed_input_sink_manager.cpp @@ -298,6 +298,9 @@ int32_t DistributedInputSinkManager::Release() startServerCB->OnResult(0, 0); } + // 4.Release input collect resource + DistributedInputCollector::GetInstance().Release(); + serviceRunningState_ = ServiceSinkRunningState::STATE_NOT_START; DHLOGI("exit dinput sink sa."); exit(0);