2
\$\begingroup\$

As I was writing, I realized I have multiple classes polling for file descriptors in its own thread - lots of duplicate code like, synchronization, creating array of pollfds etc.

I've decided to remove the duplicate code by writing the following class, which takes care of polling fds in the background. Any thoughts?

template<typename Callback, typename ReadFn>
class Poller
{
public:
    using storage_type = std::vector<struct pollfd>;
    using callbacks_type = std::map<int, Callback>;

    explicit Poller(ReadFn&& readFn) : readFn(std::forward(readFn)) {}
    ~Poller()
    {
        if (worker.joinable())
        {
            executeInContext([](storage_type& storage, callbacks_type& callbacks){
                storage.erase(storage.begin()+1, storage.end());
                callbacks.clear();
            });

            worker.join();
            close(_fd[0]);
            close(_fd[1]);
        }
    }

    void addDescriptor(int fd, Callback cb)
    {
        if (!worker.joinable())
        {
            pipe(_fd);
            _storage.push_back({_fd[0], POLLPRI | POLLIN, 0});
            _storage.push_back({fd, POLLPRI | POLLIN, 0});
            _callbacks[fd] = cb;

            worker = std::thread([this]{threadFunc();});
        }
        else
        {
            executeInContext([fd, cb](storage_type& storage, callbacks_type& callbacks){
                auto it = std::find_if(storage.begin(), storage.end(),
                                       [fd](struct pollfd i){ return fd == i.fd; });
                if (it == storage.end())
                {
                    storage.push_back({fd, POLLPRI | POLLIN, 0});
                }
                callbacks[fd] = cb;
            });
        }
    }

    void removeDescriptor(int fd)
    {
        executeInContext([fd](storage_type& storage, callbacks_type& callbacks){
            if (auto it = std::find_if(storage.begin(), storage.end(), [fd](struct pollfd i){
                                            return i.fd == fd;
                                        }); it != storage.end())
            {
                callbacks.erase(it->fd);
                storage.erase(it);
            }
        });

        if (_storage.size() == 1 && worker.joinable())
        {
            worker.join();

            close(_fd[0]);
            close(_fd[1]);
            _storage.clear();
        }
    }

    template<typename Callable>
    void executeInContext(Callable&& func)
    {
        ::write(_fd[1], "suse\0", 5);
        {
            std::unique_lock lock(mutex);
            std::invoke(std::forward(func),
                        std::ref(_storage), std::ref(_callbacks));
        }
        cv.notify_all();
    }

protected:
    void threadFunc()
    {
        std::unique_lock lock(mutex);
        while (true)
        {
            if (int rv = poll(_storage.data(), _storage.size(), -1); rv > 0)
            {
                /* Pipe sync request */
                if (_storage.begin()->revents)
                {
                    char buff[0x10];
                    ::read(_fd[0], buff, sizeof(buff));

                    /* Wait for main thread to execute its stuff */
                    cv.wait(lock);

                    if (_storage.size() == 1)
                    {
                        /* All fds have been removed except the pipe, exit */
                        break;
                    }
                }

                for (auto& pfd : _storage)
                {
                    if (pfd.revents)
                    {
                        std::invoke(readFn,
                                    pfd.fd, _callbacks.at(pfd.fd));
                    }
                }
            }
            else if (errno == EINTR)
            {
                continue;
            }
            else
            { /* Debug trace breakpoint */ }
        }
    }

private:
    std::thread worker{};
    std::vector<struct pollfd> _storage{};
    std::map<int, Callback> _callbacks{};
    std::condition_variable cv{};
    std::mutex mutex{};
    int _fd[2]{};

    ReadFn readFn;
};

Example usage:

using callback_type = std::function<void(int)>;
auto readfn = [](int fd, callback_type cb){
    struct event_structure ev{};
    ::read(fd, &ev, sizeof(ev));

    std::invoke(cb, ev.property);
};
Poller<callback_type, decltype(readfn)> poller(readfn);

int fd = /* open FD */;
poller.addDescriptor(fd, [](int){/* Handle property change */});
/* More descriptors with unique callbacks added/removed during program execution */
\$\endgroup\$
1
  • 1
    \$\begingroup\$ @TobySpeight I've updated the question with example usage. Please note I've just written it from top of my head - might not compile. \$\endgroup\$
    – Quest
    Sep 4 at 8:40
2
\$\begingroup\$

Start the thread in the constructor

Starting and stopping the thread on demand requires being very careful not to trigger any race conditions. But there is no reason not to have the thread running all the time, if it's just blocked waiting on a command on the sync fd, it doesn't use any CPU time. So I recommend you start the thread in the constructor, and stop it in the destructor.

Missing error handling

The calls to ::read() and ::write() can fail; be sure to handle failures correctly.

Use a std::vector for _callbacks

It should be possible to store the callbacks in a std::vector, that matches its order with that of _storage. That way, you don't have to do a rather expensive lookup to get the callback; you just iterate over _storage and _callbacks simultaneously in threadFunc().

Thread safety

The code you have might work for your application, but it is quite fragile. You've written it in such a way that it's only safe to add and remove descriptors to a Poller in the same thread where you instantiated it. Most importantly, a callback cannot remove itself safely. It would be possible to make it more robust at the cost of adding a bit more complexity.

One reason for this is that adding items to or removing items from _storage will invalidate any iterators, including the ones used behind the scenes by the for-loop in threadFunc(). A possible solution is to set a flag when adding or removing items from the loop, and after invoking the callback, immediately exit the loop if the flag is set.

Another issue is that calling executeInContext() from inside the polling thread will result in a deadlock. A potential solution is to let detect if executeInContext() detect if it is called from the same thread as threadFunc(), and if so not try to lock the mutex.

\$\endgroup\$
4
  • \$\begingroup\$ I do agree callback cannot remove itself, but how come I can only add items from the thread that instantiated it? \$\endgroup\$
    – Quest
    Sep 4 at 19:11
  • \$\begingroup\$ The reason is that iterators that are still in use are being invalidated that way, and there's a deadlock situation. I've updated the answer. \$\endgroup\$
    – G. Sliepen
    Sep 4 at 20:32
  • \$\begingroup\$ I don't agree about invalidating the iterators. If we take into account that we are not able to add/remove elements from the callback, the iterators are not used during modify operation on vector/map. Only after this function has taken place, the for loop will begin to execute. Although I still like your answer as it made me think of these scenarios. Which brings me to the last question. Is there any error (besides EINTR), which might be set during write on pipes (when both ends are still valid)? \$\endgroup\$
    – Quest
    Sep 5 at 15:11
  • \$\begingroup\$ If it's done outside the callback, it's indeed fine, as long as adding/removing items is all done from the same thread, or you have some way to ensure no two threads can add/remove at the same time. This might be a reasonable assumption to make. If you document your code, be sure to include a note about the thread safety of your functions. As for pipe errors: if you make them non-blocking, EAGAIN or EWOULDBLOCK can happen. Apart from that not much. EPIPE when one end closes the pipe, which you could make use of in your code. \$\endgroup\$
    – G. Sliepen
    Sep 5 at 15:46

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.