xbmc/platform/linux/FDEventMonitor.cpp

   1 /*
   2  *  Copyright (C) 2014-2018 Team Kodi
   3  *  This file is part of Kodi - https://kodi.tv
   4  *
   5  *  SPDX-License-Identifier: GPL-2.0-or-later
   6  *  See LICENSES/README.md for more information.
   7  */
   8
   9 #include "FDEventMonitor.h"
  10
  11 #include "threads/SingleLock.h"
  12 #include "utils/log.h"
  13
  14 #include <errno.h>
  15 #include <mutex>
  16
  17 #include <poll.h>
  18 #include <sys/eventfd.h>
  19 #include <unistd.h>
  20
  21 #include "PlatformDefs.h"
  22
  23 CFDEventMonitor::CFDEventMonitor() :
  24   CThread("FDEventMonitor")
  25 {
  26 }
  27
  28 CFDEventMonitor::~CFDEventMonitor()
  29 {
  30   std::unique_lock<CCriticalSection> lock(m_mutex);
  31   InterruptPoll();
  32
  33   if (m_wakeupfd >= 0)
  34   {
  35     /* sets m_bStop */
  36     StopThread(false);
  37
  38     /* wake up the poll() call */
  39     eventfd_write(m_wakeupfd, 1);
  40
  41     /* Wait for the thread to stop */
  42     {
  43       CSingleExit exit(m_mutex);
  44       StopThread(true);
  45     }
  46
  47     close(m_wakeupfd);
  48   }
  49 }
  50
  51 void CFDEventMonitor::AddFD(const MonitoredFD& monitoredFD, int& id)
  52 {
  53   std::unique_lock<CCriticalSection> lock(m_mutex);
  54   InterruptPoll();
  55
  56   AddFDLocked(monitoredFD, id);
  57
  58   StartMonitoring();
  59 }
  60
  61 void CFDEventMonitor::AddFDs(const std::vector<MonitoredFD>& monitoredFDs,
  62                              std::vector<int>& ids)
  63 {
  64   std::unique_lock<CCriticalSection> lock(m_mutex);
  65   InterruptPoll();
  66
  67   for (unsigned int i = 0; i < monitoredFDs.size(); ++i)
  68   {
  69     int id;
  70     AddFDLocked(monitoredFDs[i], id);
  71     ids.push_back(id);
  72   }
  73
  74   StartMonitoring();
  75 }
  76
  77 void CFDEventMonitor::RemoveFD(int id)
  78 {
  79   std::unique_lock<CCriticalSection> lock(m_mutex);
  80   InterruptPoll();
  81
  82   if (m_monitoredFDs.erase(id) != 1)
  83   {
  84     CLog::Log(LOGERROR, "CFDEventMonitor::RemoveFD - Tried to remove non-existing monitoredFD {}",
  85               id);
  86   }
  87
  88   UpdatePollDescs();
  89 }
  90
  91 void CFDEventMonitor::RemoveFDs(const std::vector<int>& ids)
  92 {
  93   std::unique_lock<CCriticalSection> lock(m_mutex);
  94   InterruptPoll();
  95
  96   for (unsigned int i = 0; i < ids.size(); ++i)
  97   {
  98     if (m_monitoredFDs.erase(ids[i]) != 1)
  99     {
 100       CLog::Log(LOGERROR,
 101                 "CFDEventMonitor::RemoveFDs - Tried to remove non-existing monitoredFD {} while "
 102                 "removing {} FDs",
 103                 ids[i], (unsigned)ids.size());
 104     }
 105   }
 106
 107   UpdatePollDescs();
 108 }
 109
 110 void CFDEventMonitor::Process()
 111 {
 112   eventfd_t dummy;
 113
 114   while (!m_bStop)
 115   {
 116     std::unique_lock<CCriticalSection> lock(m_mutex);
 117     std::unique_lock<CCriticalSection> pollLock(m_pollMutex);
 118
 119     /*
 120      * Leave the main mutex here to allow another thread to
 121      * lock it while we are in poll().
 122      * By then calling InterruptPoll() the other thread can
 123      * wake up poll and wait for the processing to pause at
 124      * the above lock(m_mutex).
 125      */
 126     lock.unlock();
 127
 128     int err = poll(&m_pollDescs[0], m_pollDescs.size(), -1);
 129
 130     if (err < 0 && errno != EINTR)
 131     {
 132       CLog::Log(LOGERROR, "CFDEventMonitor::Process - poll() failed, error {}, stopping monitoring",
 133                 errno);
 134       StopThread(false);
 135     }
 136
 137     // Something woke us up - either there is data available or we are being
 138     // paused/stopped via m_wakeupfd.
 139
 140     for (unsigned int i = 0; i < m_pollDescs.size(); ++i)
 141     {
 142       struct pollfd& pollDesc = m_pollDescs[i];
 143       int id = m_monitoredFDbyPollDescs[i];
 144       const MonitoredFD& monitoredFD = m_monitoredFDs[id];
 145
 146       if (pollDesc.revents)
 147       {
 148         if (monitoredFD.callback)
 149         {
 150           monitoredFD.callback(id, pollDesc.fd, pollDesc.revents,
 151                                monitoredFD.callbackData);
 152         }
 153
 154         if (pollDesc.revents & (POLLERR | POLLHUP | POLLNVAL))
 155         {
 156           CLog::Log(LOGERROR,
 157                     "CFDEventMonitor::Process - polled fd {} got revents 0x{:x}, removing it",
 158                     pollDesc.fd, pollDesc.revents);
 159
 160           /* Probably would be nice to inform our caller that their FD was
 161            * dropped, but oh well... */
 162           m_monitoredFDs.erase(id);
 163           UpdatePollDescs();
 164         }
 165
 166         pollDesc.revents = 0;
 167       }
 168     }
 169
 170     /* flush wakeup fd */
 171     eventfd_read(m_wakeupfd, &dummy);
 172
 173   }
 174 }
 175
 176 void CFDEventMonitor::AddFDLocked(const MonitoredFD& monitoredFD, int& id)
 177 {
 178   id = m_nextID;
 179
 180   while (m_monitoredFDs.count(id))
 181   {
 182     ++id;
 183   }
 184   m_nextID = id + 1;
 185
 186   m_monitoredFDs[id] = monitoredFD;
 187
 188   AddPollDesc(id, monitoredFD.fd, monitoredFD.events);
 189 }
 190
 191 void CFDEventMonitor::AddPollDesc(int id, int fd, short events)
 192 {
 193   struct pollfd newPollFD;
 194   newPollFD.fd = fd;
 195   newPollFD.events = events;
 196   newPollFD.revents = 0;
 197
 198   m_pollDescs.push_back(newPollFD);
 199   m_monitoredFDbyPollDescs.push_back(id);
 200 }
 201
 202 void CFDEventMonitor::UpdatePollDescs()
 203 {
 204   m_monitoredFDbyPollDescs.clear();
 205   m_pollDescs.clear();
 206
 207   for (const auto& it : m_monitoredFDs)
 208   {
 209     AddPollDesc(it.first, it.second.fd, it.second.events);
 210   }
 211 }
 212
 213 void CFDEventMonitor::StartMonitoring()
 214 {
 215   if (!IsRunning())
 216   {
 217     /* Start the monitoring thread */
 218
 219     m_wakeupfd = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK);
 220     if (m_wakeupfd < 0)
 221     {
 222       CLog::Log(LOGERROR, "CFDEventMonitor::StartMonitoring - Failed to create eventfd, error {}",
 223                 errno);
 224       return;
 225     }
 226
 227     /* Add wakeup fd to the fd list */
 228     int id;
 229     AddFDLocked(MonitoredFD(m_wakeupfd, POLLIN, NULL, NULL), id);
 230
 231     Create(false);
 232   }
 233 }
 234
 235 void CFDEventMonitor::InterruptPoll()
 236 {
 237   if (m_wakeupfd >= 0)
 238   {
 239     eventfd_write(m_wakeupfd, 1);
 240     /* wait for the poll() result handling (if any) to end */
 241     std::unique_lock<CCriticalSection> pollLock(m_pollMutex);
 242   }
 243 }