Description:fix dinput thread

Match-id-4c6907c6a22d833d6308051f091a6dd7cc8aa6da

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.

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_) {

common/include/input_hub.h

@@ -16,6 +16,7 @@
 #ifndef INPUT_HUB_H
 #define INPUT_HUB_H
 
+#include <atomic>
 #include <mutex>
 #include <unordered_map>
 
@@ -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<InputDevice> GetAllInputDevices();
     void SetSupportInputType(const uint32_t& inputType);
@@ -126,6 +129,8 @@ private:
     std::mutex visitMutex_;
     bool deviceChanged_;
     uint32_t inputTypes_;
+    std::atomic<bool> isStartCollectEvent_;
+    std::atomic<bool> isStartCollectHandler_;
 };
 } // namespace DistributedInput
 } // namespace DistributedHardware

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);

services/sink/inputcollector/include/distributed_input_collector.h

@@ -16,6 +16,7 @@
 #ifndef DISTRIBUTED_INPUT_COLLECTOR_H
 #define DISTRIBUTED_INPUT_COLLECTOR_H
 
+#include <atomic>
 #include <map>
 #include <mutex>
 #include <string>
@@ -35,6 +36,7 @@ class DistributedInputCollector {
 public:
     static DistributedInputCollector &GetInstance();
     int32_t Init(std::shared_ptr<AppExecFwk::EventHandler> 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_;

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

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);