PDG_EventQueue.h

Go to the documentation of this file.

/*
 * PROPRIETARY INFORMATION.  This software is proprietary to
 * Side Effects Software Inc., and is not to be reproduced,
 * transmitted, or disclosed in any way without written permission.
 *
 * COMMENTS:
 */

#ifndef __PDG_EventQueue_h__
#define __PDG_EventQueue_h__

#include "PDG_API.h"
#include "PDG_Event.h"
#include "PDG_EventTypes.h"

#include <UT/UT_Array.h>
#include <UT/UT_ConcurrentHashMap.h>
#include <UT/UT_ConcurrentQueue.h>
#include <UT/UT_ConcurrentVector.h>
#include <UT/UT_ParallelUtil.h>
#include <UT/UT_ThreadSpecificValue.h>
#include <UT/UT_UniquePtr.h>

#include <SYS/SYS_AtomicInt.h>

#define PDG_EVENT_QUEUE_STATS 0

class PDG_EventEmitter;
class PDG_EventHandler;

class UT_Thread;

/* Event queue that dispatches PDG_Events to the handlers registered for
 * the PDG_EventEmitter that produced them.
 */
class PDG_API PDG_EventQueue
{
public:
    /// Scoped object that pauses event handling. Pausing is implemented with
    /// an atomic counter on the queue, so nested ScopedPause objects work as
    /// expected. Each scoped pause increments the pause request counter, so
    /// event processing will only proceed once all ScopedPause objects are
    /// unwound.
    class ScopedPause
    {
    public:
        ScopedPause(PDG_EventQueue* queue)
            : myQueue(queue)
        {
            queue->pause();
        }

        ~ScopedPause()
        {
            myQueue->resume();
        }

    private:
        PDG_EventQueue* myQueue;
    };

public:
    /// Constructs a new event queue
                        PDG_EventQueue(bool background);
                        ~PDG_EventQueue();

    /// Queues an event that will eventually be dispatched by the background
    /// thread running in this event queue instance.
    void                queueEvent(
                            const PDG_EventEmitter* emitter,
                            const PDG_Event& event);

    /// Emits an event immediately on the calling thread
    void                emitEvent(
                            const PDG_EventEmitter* emitter,
                            const PDG_Event& event) const;

    /// Waits for all queued events to finish, and prevents new events from
    /// being added until that condition is reached. Blocks the caller. If
    /// an event emitter is passed into this function, it gets deregistered
    /// before unblocking the event queue.
    void                waitAllEvents(PDG_EventEmitter* emitter=nullptr);

    /// Waits until the specified emitter is registered, to ensure that all
    /// events for that emitter have been flushed
    void                waitEmitter(const PDG_EventEmitter* emitter);

    /// Returns whether or not the event queue attempts to consolidate
    /// consecutive events of the same type
    inline bool         isConsolidateEvents() const
                            { return myIsConsolidateEvents; }

    /// Configures the event consolidation flag
    inline void         setConsolidateEvents(bool consolidate)
                            { myIsConsolidateEvents = consolidate; }

private:
    friend class PDG_EventEmitter;
    friend class PDG_GraphContext;

    struct EmitterInfo
    {
        PDG_EventHandlerArray   myHandlers;
        PDG_EventFilterArray    myFilters;

        void                    add(
                                    PDG_EventHandler* handler,
                                    const PDG_EventFilter& filter,
                                    PDG_EventFilterMap& filter_map);
        bool                    remove(
                                    PDG_EventHandler* handler,
                                    PDG_EventFilterMap& filter_map);
        void                    removeAll(
                                    PDG_EventHandlerArray& removed,
                                    PDG_EventFilterMap& filter_map,
                                    bool user_only);
        void                    emitEvent(const PDG_Event& event) const;
    };

    enum EmitterUpdate
    {
        eHandlerAdd,
        eHandlerRemove,
        eHandlerRemoveAll
    };

    struct DeferredEmitterUpdate
    {
        PDG_EventEmitter*       myEmitter;
        PDG_EventHandler*       myHandler;

        PDG_EventFilter         myFilter;

        EmitterUpdate           myType;
        bool                    myUserOnly;
    };

    using EventInfo             = std::pair<
                                    const PDG_EventEmitter*, PDG_Event>;
    using EventArray            = UT_Array<EventInfo>;
    using EventQueue            = UT_ConcurrentQueue<EventInfo>;
    using EmitterMap            = UT_ConcurrentHashMap<
                                    PDG_EventEmitter*, EmitterInfo>;
    using EmitterArrayMap       = UT_UniquePtr<EmitterMap[]>;
    using EmitterUpdateArray    = UT_ConcurrentVector<DeferredEmitterUpdate>;
    using ActiveEmitter         = UT_ThreadSpecificValue<
                                    const PDG_EventEmitter*>;

    class EventFunctor
    {
    public:
                            EventFunctor(
                                PDG_EventQueue* queue,
                                const EventArray& events);
                            ~EventFunctor();

        void                operator()(
                                const UT_BlockedRange<int64> &range) const;

    private:
        PDG_EventQueue*     myQueue;
        const EventArray&   myEvents;
    };
    
private:
    void                terminate();

    exint               emitterIndex(const PDG_EventEmitter* emitter) const;

    void                registerEmitter(PDG_EventEmitter* emitter);
    void                deregisterEmitter(PDG_EventEmitter* emitter);

    bool                addEventHandler(
                            PDG_EventEmitter* emitter,
                            PDG_EventHandler* handler,
                            const PDG_EventFilter& filter);
    void                removeEventHandler(
                            PDG_EventEmitter* emitter,
                            PDG_EventHandler* handler);
    void                removeAllEventHandlers(
                            PDG_EventEmitter* emitter,
                            bool user_only);
    void                findEventHandlers(
                            const PDG_EventEmitter* emitter,
                            PDG_EventHandlerArray& handlers) const;

    void                processQueueStats(int print_limit);
    void                processEvent(
                            const EventInfo& info,
                            const PDG_EventEmitter*& active_emitter);
    void                processEvents();

    inline void         pause()
                            { myPauseRequests.add(1); }
    inline void         resume()
                            { myPauseRequests.add(-1); }

    static void*        runThread(void* param);

private:
    EventQueue          myEvents;
    EmitterArrayMap     myEmitters;
    EmitterUpdateArray  myEmitterUpdates;
    ActiveEmitter       myActiveEmitter;

    UT_Thread*          myThread;

#if PDG_EVENT_QUEUE_STATS
    exint               myDropEventCount;
    exint               myQueuedEventCount;
    exint               myPoppedEventCount;
    exint               myProcessedEventCount;
    exint               myQueueLogTime;
#endif

    exint               myEmitterMapSize;
    exint               myProcessSleepTime;
    SYS_AtomicInt32     myPauseRequests;

    bool                myIsTerminating;
    bool                myIsWaiting;
    bool                myIsProcessing;
    bool                myIsConsolidateEvents;
};

#endif