3
\$\begingroup\$

I'm working on a legacy C++ project, which is using GLib for its Main Event Loop. This has caused a lot of ugliness because it mixes C++ and C paradigms (no C++ wrapper for GLib had been used).

Anyway, I'd like to generalize some classes that internally use the Asynchronous Queue provided by GLib.

This is what I've done:

// File AsyncQueue.h - C++11 required 
#include "glib.h"

template<typename T>
class AsyncQueue
{
    static gboolean prepare(GSource *source, gint *timeout)
    {
        auto async_queue_source = reinterpret_cast<AsyncQueueSource*>(source);

        if( nullptr == async_queue_source || nullptr == async_queue_source->async_queue )
            return FALSE;

        if( nullptr != timeout )
            *timeout = -1; // Poll timeout

        return g_async_queue_length(async_queue_source->async_queue) > 0;
    }

    static gboolean check(GSource *source)
    {
        auto async_queue_source = reinterpret_cast<AsyncQueueSource*>(source);

        if( nullptr == async_queue_source || nullptr == async_queue_source->async_queue )
            return FALSE;

        return g_async_queue_length(async_queue_source->async_queue) > 0;
    }

    static gboolean dispatch(GSource *source, GSourceFunc callback, gpointer user_data)
    {
        auto async_queue_source = reinterpret_cast<AsyncQueueSource*>(source);

        if( nullptr == async_queue_source || nullptr == async_queue_source->async_queue )
            return G_SOURCE_REMOVE;

        auto item = reinterpret_cast<T*>(g_async_queue_pop(async_queue_source->async_queue));

        if( nullptr != item )
        {
            if( nullptr != async_queue_source->this_object )
                async_queue_source->this_object->OnItemPopped(*item);
            delete item;
        }

        return G_SOURCE_CONTINUE;
    }

    struct AsyncQueueSource
    {
        GSource        source_object;
        AsyncQueue<T>* this_object;
        GAsyncQueue*   async_queue;
    };

    static_assert(std::is_trivial<AsyncQueueSource>::value, "AsyncQueueSource must be a trivial type");

    GSourceFuncs source_funcs;
    GMainContext* context;
    guint source_id_within_context;
    AsyncQueueSource* async_queue_source;

public:
    // noncopyable
    AsyncQueue(const AsyncQueue&) = delete;
    AsyncQueue& operator=(const AsyncQueue&) = delete;

    AsyncQueue()
        : source_funcs{&AsyncQueue::prepare, nullptr, &AsyncQueue::dispatch, nullptr, nullptr, nullptr}
        , context{nullptr}
        , source_id_within_context{0}
    {
        async_queue_source = reinterpret_cast<AsyncQueueSource*>(g_source_new(&source_funcs, sizeof(AsyncQueueSource)));

        if( nullptr != async_queue_source )
        {
            async_queue_source->this_object = this;
            async_queue_source->async_queue = g_async_queue_new();
        }
    }

    virtual ~AsyncQueue()
    {
        if( nullptr != async_queue_source )
        {
            if( nullptr != async_queue_source->async_queue )
            {
                g_async_queue_unref(async_queue_source->async_queue);
            }
            async_queue_source->this_object = nullptr;
            g_source_unref(reinterpret_cast<GSource*>(async_queue_source));
        }
    }

    bool Attach(GMainContext* main_context)
    {
        bool result = false;

        if( nullptr != async_queue_source )
        {
            source_id_within_context = g_source_attach(&async_queue_source->source_object, main_context);
            if( source_id_within_context > 0 )
                result = true;
        }

        return result;
    }

    bool IsAttached() const { return source_id_within_context > 0; }

    virtual void OnItemPopped(T& item) = 0;

    // item must be allocated with new() and will be owned by AsyncQueue
    void PushItem(T* item)
    {
        if( nullptr != item )
        {
            if( nullptr != async_queue_source && nullptr != async_queue_source->async_queue )
            {
                g_async_queue_push(async_queue_source->async_queue, item);
                if( IsAttached() )
                    g_main_context_wakeup(context);
            }
            else
            {
                delete item;
            }
        }
    }

    void PushItem(const T* item)
    {
        if( nullptr != item )
        {
            auto copied_item = new T(*item);
            PushItem(copied_item);
        }
    }

    void PushItem(const T& item)
    {
        auto copied_item = new T(item);
        PushItem(copied_item);
    }

    void PushItem(T&& item)
    {
        auto moved_item = new T(std::move(item));
        PushItem(moved_item);
    }
};

Pointer misuse

This class seems reasonably clean and simple to use, but my main concern is misuse of AsyncQueue::PushItem(T* item) because the caller could pass a pointer to an object on the heap or something allocated with g_new or malloc.

I could give the responsibility to free the memory to the implementer of virtual void OnItemPopped(T& item) by passing a T* instead, but I don't like this option.

What if I accept only smart pointers of type std::unique_ptr<T> and therefore I change signatures as AsyncQueue::PushItem(std::unique_ptr<T> item) and virtual void OnItemPopped(std::unique_ptr<T> item)? Anyway, I would still have to pass naked pointers to the asynchronous queue because it works only with trivial types (aka POD types).

Thread safety

Multiple threads should be able to call without issues AsyncQueue::PushItem, all the synchronization is done by GLib inside the async queue. The only thread able to pop elements is the thread running the Main Loop on the GMainContext passed to AsyncQueue::Attach.

Besides this, there are some obvious multithreading flaws that I missed?

Items leaked on class destruction

If the class or the main loop is destroyed when the queue still holds data they will be leaked (see https://stackoverflow.com/a/52175388/297373). I'm not sure how to solve this issue.

\$\endgroup\$

Your Answer

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

Browse other questions tagged or ask your own question.