text:
#ifndef ThreadState_h
#define ThreadState_h

#include "platform/PlatformExport.h"
#include "platform/heap/AddressSanitizer.h"
#include "public/platform/WebThread.h"
#include "wtf/HashSet.h"
#include "wtf/OwnPtr.h"
#include "wtf/PassOwnPtr.h"
#include "wtf/ThreadSpecific.h"
#include "wtf/Threading.h"
#include "wtf/ThreadingPrimitives.h"
#include "wtf/Vector.h"

#if ENABLE(GC_PROFILE_HEAP)
#include "wtf/HashMap.h"
#endif

namespace blink {

class BaseHeap;
class BaseHeapPage;
class FinalizedHeapObjectHeader;
struct GCInfo;
class HeapContainsCache;
class HeapObjectHeader;
class PageMemory;
class PersistentNode;
class WrapperPersistentRegion;
class Visitor;
class SafePointBarrier;
class SafePointAwareMutexLocker;
template<typename Header> class ThreadHeap;
class CallbackStack;

typedef uint8_t* Address;

typedef void (*FinalizationCallback)(void*);
typedef void (*VisitorCallback)(Visitor*, void* self);
typedef VisitorCallback TraceCallback;
typedef VisitorCallback WeakPointerCallback;
typedef VisitorCallback EphemeronCallback;

// ThreadAffinity indicates which threads objects can be used on. We
// distinguish between objects that can be used on the main thread
// only and objects that can be used on any thread.
//
// For objects that can only be used on the main thread we avoid going
// through thread-local storage to get to the thread state.
//
// FIXME: We should evaluate the performance gain. Having
// ThreadAffinity is complicating the implementation and we should get
// rid of it if it is fast enough to go through thread-local storage
// always.
enum ThreadAffinity {
    AnyThread,
    MainThreadOnly,
};

class Node;
class CSSValue;

template<typename T, bool derivesNode = WTF::IsSubclass<typename WTF::RemoveConst<T>::Type, Node>::value> struct DefaultThreadingTrait;

template<typename T>
struct DefaultThreadingTrait<T, false> {
    static const ThreadAffinity Affinity = AnyThread;
};

template<typename T>
struct DefaultThreadingTrait<T, true> {
    static const ThreadAffinity Affinity = MainThreadOnly;
};

template<typename T>
struct ThreadingTrait {
    static const ThreadAffinity Affinity = DefaultThreadingTrait<T>::Affinity;
};

// Marks the specified class as being used from multiple threads. When
// a class is used from multiple threads we go through thread local
// storage to get the heap in which to allocate an object of that type
// and when allocating a Persistent handle for an object with that
// type. Notice that marking the base class does not automatically
// mark its descendants and they have to be explicitly marked.
#define USED_FROM_MULTIPLE_THREADS(Class)                 \
    class Class;                                          \
    template<> struct ThreadingTrait<Class> {             \
        static const ThreadAffinity Affinity = AnyThread; \
    }

#define USED_FROM_MULTIPLE_THREADS_NAMESPACE(Namespace, Class)          \
    namespace Namespace {                                               \
        class Class;                                                    \
    }                                                                   \
    namespace blink {                                                 \
        template<> struct ThreadingTrait<Namespace::Class> {            \
            static const ThreadAffinity Affinity = AnyThread;           \
        };                                                              \
    }

template<typename U> class ThreadingTrait<const U> : public ThreadingTrait<U> { };

// List of typed heaps. The list is used to generate the implementation
// of typed heap related methods.
//
// To create a new typed heap add a H(<ClassName>) to the
// FOR_EACH_TYPED_HEAP macro below.
#define FOR_EACH_TYPED_HEAP(H)  \
    H(Node)

#define TypedHeapEnumName(Type) Type##Heap,
#define TypedHeapEnumNameNonFinalized(Type) Type##HeapNonFinalized,

enum TypedHeaps {
    GeneralHeap = 0,
    CollectionBackingHeap,
    FOR_EACH_TYPED_HEAP(TypedHeapEnumName)
    GeneralHeapNonFinalized,
    CollectionBackingHeapNonFinalized,
    FOR_EACH_TYPED_HEAP(TypedHeapEnumNameNonFinalized)
    // Values used for iteration of heap segments.
    NumberOfHeaps,
    FirstFinalizedHeap = GeneralHeap,
    FirstNonFinalizedHeap = GeneralHeapNonFinalized,
    NumberOfFinalizedHeaps = GeneralHeapNonFinalized,
    NumberOfNonFinalizedHeaps = NumberOfHeaps - NumberOfFinalizedHeaps,
    NonFinalizedHeapOffset = FirstNonFinalizedHeap
};

// Base implementation for HeapIndexTrait found below.
template<int heapIndex>
struct HeapIndexTraitBase {
    typedef FinalizedHeapObjectHeader HeaderType;
    typedef ThreadHeap<HeaderType> HeapType;
    static const int finalizedIndex = heapIndex;
    static const int nonFinalizedIndex = heapIndex + static_cast<int>(NonFinalizedHeapOffset);
    static int index(bool isFinalized)
    {
        return isFinalized ? finalizedIndex : nonFinalizedIndex;
    }
};

// HeapIndexTrait defines properties for each heap in the TypesHeaps enum.
template<int index>
struct HeapIndexTrait;

template<>
struct HeapIndexTrait<GeneralHeap> : public HeapIndexTraitBase<GeneralHeap> { };
template<>
struct HeapIndexTrait<GeneralHeapNonFinalized> : public HeapIndexTrait<GeneralHeap> { };

template<>
struct HeapIndexTrait<CollectionBackingHeap> : public HeapIndexTraitBase<CollectionBackingHeap> { };
template<>
struct HeapIndexTrait<CollectionBackingHeapNonFinalized> : public HeapIndexTrait<CollectionBackingHeap> { };

#define DEFINE_TYPED_HEAP_INDEX_TRAIT(Type)                                     \
    template<>                                                                  \
    struct HeapIndexTrait<Type##Heap> : public HeapIndexTraitBase<Type##Heap> { \
        typedef HeapObjectHeader HeaderType;                                    \
        typedef ThreadHeap<HeaderType> HeapType;                                \
    };                                                                          \
    template<>                                                                  \
    struct HeapIndexTrait<Type##HeapNonFinalized> : public HeapIndexTrait<Type##Heap> { };
FOR_EACH_TYPED_HEAP(DEFINE_TYPED_HEAP_INDEX_TRAIT)
#undef DEFINE_TYPED_HEAP_INDEX_TRAIT

// HeapTypeTrait defines which heap to use for particular types.
// By default objects are allocated in the GeneralHeap.
template<typename T>
struct HeapTypeTrait : public HeapIndexTrait<GeneralHeap> { };

// We don't have any type-based mappings to the CollectionBackingHeap.

// Each typed-heap maps the respective type to its heap.
#define DEFINE_TYPED_HEAP_TRAIT(Type)                                   \
    class Type;                                                         \
    template<>                                                          \
    struct HeapTypeTrait<class Type> : public HeapIndexTrait<Type##Heap> { };
FOR_EACH_TYPED_HEAP(DEFINE_TYPED_HEAP_TRAIT)
#undef DEFINE_TYPED_HEAP_TRAIT

// A HeapStats structure keeps track of the amount of memory allocated
// for a Blink heap and how much of that memory is used for actual
// Blink objects. These stats are used in the heuristics to determine
// when to perform garbage collections.
class HeapStats {
public:
    HeapStats() : m_totalObjectSpace(0), m_totalAllocatedSpace(0) { }

    size_t totalObjectSpace() const { return m_totalObjectSpace; }
    size_t totalAllocatedSpace() const { return m_totalAllocatedSpace; }

    void add(HeapStats* other)
    {
        m_totalObjectSpace += other->m_totalObjectSpace;
        m_totalAllocatedSpace += other->m_totalAllocatedSpace;
    }

    void inline increaseObjectSpace(size_t newObjectSpace)
    {
        m_totalObjectSpace += newObjectSpace;
    }

    void inline decreaseObjectSpace(size_t deadObjectSpace)
    {
        m_totalObjectSpace -= deadObjectSpace;
    }

    void inline increaseAllocatedSpace(size_t newAllocatedSpace)
    {
        m_totalAllocatedSpace += newAllocatedSpace;
    }

    void inline decreaseAllocatedSpace(size_t deadAllocatedSpace)
    {
        m_totalAllocatedSpace -= deadAllocatedSpace;
    }

    void clear()
    {
        m_totalObjectSpace = 0;
        m_totalAllocatedSpace = 0;
    }

    bool operator==(const HeapStats& other)
    {
        return m_totalAllocatedSpace == other.m_totalAllocatedSpace
            && m_totalObjectSpace == other.m_totalObjectSpace;
    }

private:
    size_t m_totalObjectSpace; // Actually contains objects that may be live, not including headers.
    size_t m_totalAllocatedSpace; // Allocated from the OS.

    friend class HeapTester;
};

class PLATFORM_EXPORT ThreadState {
    WTF_MAKE_NONCOPYABLE(ThreadState);
public:
    // When garbage collecting we need to know whether or not there
    // can be pointers to Blink GC managed objects on the stack for
    // each thread. When threads reach a safe point they record
    // whether or not they have pointers on the stack.
    enum StackState {
        NoHeapPointersOnStack,
        HeapPointersOnStack
    };

    // When profiling we would like to identify forced GC requests.
    enum CauseOfGC {
        NormalGC,
        ForcedGC
    };

    class NoSweepScope {
    public:
        explicit NoSweepScope(ThreadState* state) : m_state(state)
        {
            ASSERT(!m_state->m_sweepInProgress);
            m_state->m_sweepInProgress = true;
        }
        ~NoSweepScope()
        {
            ASSERT(m_state->m_sweepInProgress);
            m_state->m_sweepInProgress = false;
        }
    private:
        ThreadState* m_state;
    };

    // The set of ThreadStates for all threads attached to the Blink
    // garbage collector.
    typedef HashSet<ThreadState*> AttachedThreadStateSet;
    static AttachedThreadStateSet& attachedThreads();

    // Initialize threading infrastructure. Should be called from the main
    // thread.
    static void init();
    static void shutdown();
    static void shutdownHeapIfNecessary();
    bool isTerminating() { return m_isTerminating; }

    static void attachMainThread();
    static void detachMainThread();

    // Trace all persistent roots, called when marking the managed heap objects.
    static void visitPersistentRoots(Visitor*);

    // Trace all objects found on the stack, used when doing conservative GCs.
    static void visitStackRoots(Visitor*);

    // Associate ThreadState object with the current thread. After this
    // call thread can start using the garbage collected heap infrastructure.
    // It also has to periodically check for safepoints.
    static void attach();

    // Disassociate attached ThreadState from the current thread. The thread
    // can no longer use the garbage collected heap after this call.
    static void detach();

    static ThreadState* current() { return **s_threadSpecific; }
    static ThreadState* mainThreadState()
    {
        return reinterpret_cast<ThreadState*>(s_mainThreadStateStorage);
    }

    bool isMainThread() const { return this == mainThreadState(); }
    inline bool checkThread() const
    {
        ASSERT(m_thread == currentThread());
        return true;
    }

    // shouldGC and shouldForceConservativeGC implement the heuristics
    // that are used to determine when to collect garbage. If
    // shouldForceConservativeGC returns true, we force the garbage
    // collection immediately. Otherwise, if shouldGC returns true, we
    // record that we should garbage collect the next time we return
    // to the event loop. If both return false, we don't need to
    // collect garbage at this point.
    bool shouldGC();
    bool shouldForceConservativeGC();
    bool increasedEnoughToGC(size_t, size_t);
    bool increasedEnoughToForceConservativeGC(size_t, size_t);

    // If gcRequested returns true when a thread returns to its event
    // loop the thread will initiate a garbage collection.
    bool gcRequested();
    void setGCRequested();
    void clearGCRequested();

    // Was the last GC forced for testing? This is set when garbage collection
    // is forced for testing and there are pointers on the stack. It remains
    // set until a garbage collection is triggered with no pointers on the stack.
    // This is used for layout tests that trigger GCs and check if objects are
    // dead at a given point in time. That only reliably works when we get
    // precise GCs with no conservative stack scanning.
    void setForcePreciseGCForTesting(bool);
    bool forcePreciseGCForTesting();

    bool sweepRequested();
    void setSweepRequested();
    void clearSweepRequested();
    void performPendingSweep();

    // Support for disallowing allocation. Mainly used for sanity
    // checks asserts.
    bool isAllocationAllowed() const { return !isAtSafePoint() && !m_noAllocationCount; }
    void enterNoAllocationScope() { m_noAllocationCount++; }
    void leaveNoAllocationScope() { m_noAllocationCount--; }

    // Before performing GC the thread-specific heap state should be
    // made consistent for sweeping.
    void makeConsistentForSweeping();
#if ENABLE(ASSERT)
    bool isConsistentForSweeping();
#endif

    // Is the thread corresponding to this thread state currently
    // performing GC?
    bool isInGC() const { return m_inGC; }

    // Is any of the threads registered with the blink garbage collection
    // infrastructure currently performing GC?
    static bool isAnyThreadInGC() { return s_inGC; }

    void enterGC()
    {
        ASSERT(!m_inGC);
        ASSERT(!s_inGC);
        m_inGC = true;
        s_inGC = true;
    }

    void leaveGC()
    {
        m_inGC = false;
        s_inGC = false;
    }

    // Is the thread corresponding to this thread state currently
    // sweeping?
    bool isSweepInProgress() const { return m_sweepInProgress; }

    void prepareForGC();

    // Safepoint related functionality.
    //
    // When a thread attempts to perform GC it needs to stop all other threads
    // that use the heap or at least guarantee that they will not touch any
    // heap allocated object until GC is complete.
    //
    // We say that a thread is at a safepoint if this thread is guaranteed to
    // not touch any heap allocated object or any heap related functionality until
    // it leaves the safepoint.
    //
    // Notice that a thread does not have to be paused if it is at safepoint it
    // can continue to run and perform tasks that do not require interaction
    // with the heap. It will be paused if it attempts to leave the safepoint and
    // there is a GC in progress.
    //
    // Each thread that has ThreadState attached must:
    //   - periodically check if GC is requested from another thread by calling a safePoint() method;
    //   - use SafePointScope around long running loops that have no safePoint() invocation inside,
    //     such loops must not touch any heap object;
    //   - register an Interruptor that can interrupt long running loops that have no calls to safePoint and
    //     are not wrapped in a SafePointScope (e.g. Interruptor for JavaScript code)
    //

    // Request all other threads to stop. Must only be called if the current thread is at safepoint.
    static bool stopThreads();
    static void resumeThreads();

    // Check if GC is requested by another thread and pause this thread if this is the case.
    // Can only be called when current thread is in a consistent state.
    void safePoint(StackState);

    // Mark current thread as running inside safepoint.
    void enterSafePointWithoutPointers() { enterSafePoint(NoHeapPointersOnStack, 0); }
    void enterSafePointWithPointers(void* scopeMarker) { enterSafePoint(HeapPointersOnStack, scopeMarker); }
    void leaveSafePoint(SafePointAwareMutexLocker* = 0);
    bool isAtSafePoint() const { return m_atSafePoint; }

    class SafePointScope {
    public:
        enum ScopeNesting {
            NoNesting,
            AllowNesting
        };

        explicit SafePointScope(StackState stackState, ScopeNesting nesting = NoNesting)
            : m_state(ThreadState::current())
        {
            if (m_state->isAtSafePoint()) {
                RELEASE_ASSERT(nesting == AllowNesting);
                // We can ignore stackState because there should be no heap object
                // pointers manipulation after outermost safepoint was entered.
                m_state = 0;
            } else {
                m_state->enterSafePoint(stackState, this);
            }
        }

        ~SafePointScope()
        {
            if (m_state)
                m_state->leaveSafePoint();
        }

    private:
        ThreadState* m_state;
    };

    // If attached thread enters long running loop that can call back
    // into Blink and leaving and reentering safepoint at every
    // transition between this loop and Blink is deemed too expensive
    // then instead of marking this loop as a GC safepoint thread
    // can provide an interruptor object which would allow GC
    // to temporarily interrupt and pause this long running loop at
    // an arbitrary moment creating a safepoint for a GC.
    class PLATFORM_EXPORT Interruptor {
    public:
        virtual ~Interruptor() { }

        // Request the interruptor to interrupt the thread and
        // call onInterrupted on that thread once interruption
        // succeeds.
        virtual void requestInterrupt() = 0;

        // Clear previous interrupt request.
        virtual void clearInterrupt() = 0;

    protected:
        // This method is called on the interrupted thread to
        // create a safepoint for a GC.
        void onInterrupted();
    };

    void addInterruptor(Interruptor*);
    void removeInterruptor(Interruptor*);

    // CleanupTasks are executed when ThreadState performs
    // cleanup before detaching.
    class CleanupTask {
    public:
        virtual ~CleanupTask() { }

        // Executed before the final GC.
        virtual void preCleanup() { }

        // Executed after the final GC. Thread heap is empty at this point.
        virtual void postCleanup() { }
    };

    void addCleanupTask(PassOwnPtr<CleanupTask> cleanupTask)
    {
        m_cleanupTasks.append(cleanupTask);
    }

    // Should only be called under protection of threadAttachMutex().
    const Vector<Interruptor*>& interruptors() const { return m_interruptors; }

    void recordStackEnd(intptr_t* endOfStack)
    {
        m_endOfStack = endOfStack;
    }

    // Get one of the heap structures for this thread.
    //
    // The heap is split into multiple heap parts based on object
    // types. To get the index for a given type, use
    // HeapTypeTrait<Type>::index.
    BaseHeap* heap(int index) const { return m_heaps[index]; }

    // Infrastructure to determine if an address is within one of the
    // address ranges for the Blink heap. If the address is in the Blink
    // heap the containing heap page is returned.
    HeapContainsCache* heapContainsCache() { return m_heapContainsCache.get(); }
    BaseHeapPage* contains(Address address) { return heapPageFromAddress(address); }
    BaseHeapPage* contains(void* pointer) { return contains(reinterpret_cast<Address>(pointer)); }
    BaseHeapPage* contains(const void* pointer) { return contains(const_cast<void*>(pointer)); }

    WrapperPersistentRegion* wrapperRoots() const
    {
        ASSERT(m_liveWrapperPersistents);
        return m_liveWrapperPersistents;
    }
    WrapperPersistentRegion* takeWrapperPersistentRegion();
    void freeWrapperPersistentRegion(WrapperPersistentRegion*);

    // List of persistent roots allocated on the given thread.
    PersistentNode* roots() const { return m_persistents.get(); }

    // List of global persistent roots not owned by any particular thread.
    // globalRootsMutex must be acquired before any modifications.
    static PersistentNode* globalRoots();
    static Mutex& globalRootsMutex();

    // Visit local thread stack and trace all pointers conservatively.
    void visitStack(Visitor*);

    // Visit the asan fake stack frame corresponding to a slot on the
    // real machine stack if there is one.
    void visitAsanFakeStackForPointer(Visitor*, Address);

    // Visit all persistents allocated on this thread.
    void visitPersistents(Visitor*);

    // Checks a given address and if a pointer into the oilpan heap marks
    // the object to which it points.
    bool checkAndMarkPointer(Visitor*, Address);

#if ENABLE(GC_PROFILE_MARKING)
    const GCInfo* findGCInfo(Address);
    static const GCInfo* findGCInfoFromAllThreads(Address);
#endif

#if ENABLE(GC_PROFILE_HEAP)
    struct SnapshotInfo {
        ThreadState* state;

        size_t freeSize;
        size_t pageCount;

        // Map from base-classes to a snapshot class-ids (used as index below).
        HashMap<const GCInfo*, size_t> classTags;

        // Map from class-id (index) to count/size.
        Vector<int> liveCount;
        Vector<int> deadCount;
        Vector<size_t> liveSize;
        Vector<size_t> deadSize;

        // Map from class-id (index) to a vector of generation counts.
        // For i < 7, the count is the number of objects that died after surviving |i| GCs.
        // For i == 7, the count is the number of objects that survived at least 7 GCs.
        Vector<Vector<int, 8> > generations;

        explicit SnapshotInfo(ThreadState* state) : state(state), freeSize(0), pageCount(0) { }

        size_t getClassTag(const GCInfo*);
    };

    void snapshot();
#endif

    void pushWeakObjectPointerCallback(void*, WeakPointerCallback);
    bool popAndInvokeWeakPointerCallback(Visitor*);

    void getStats(HeapStats&);
    HeapStats& stats() { return m_stats; }
    HeapStats& statsAfterLastGC() { return m_statsAfterLastGC; }

    void setupHeapsForTermination();

    void registerSweepingTask();
    void unregisterSweepingTask();

    Mutex& sweepMutex() { return m_sweepMutex; }

private:
    explicit ThreadState();
    ~ThreadState();

    friend class SafePointBarrier;
    friend class SafePointAwareMutexLocker;

    void enterSafePoint(StackState, void*);
    NO_SANITIZE_ADDRESS void copyStackUntilSafePointScope();
    void clearSafePointScopeMarker()
    {
        m_safePointStackCopy.clear();
        m_safePointScopeMarker = 0;
    }

    void performPendingGC(StackState);

    // Finds the Blink HeapPage in this thread-specific heap
    // corresponding to a given address. Return 0 if the address is
    // not contained in any of the pages. This does not consider
    // large objects.
    BaseHeapPage* heapPageFromAddress(Address);

    // When ThreadState is detaching from non-main thread its
    // heap is expected to be empty (because it is going away).
    // Perform registered cleanup tasks and garbage collection
    // to sweep away any objects that are left on this heap.
    // We assert that nothing must remain after this cleanup.
    // If assertion does not hold we crash as we are potentially
    // in the dangling pointer situation.
    void cleanup();
    void cleanupPages();

    void setLowCollectionRate(bool value) { m_lowCollectionRate = value; }

    void waitUntilSweepersDone();

    static WTF::ThreadSpecific<ThreadState*>* s_threadSpecific;
    static SafePointBarrier* s_safePointBarrier;

    // This variable is flipped to true after all threads are stoped
    // and outermost GC has started.
    static bool s_inGC;

    // We can't create a static member of type ThreadState here
    // because it will introduce global constructor and destructor.
    // We would like to manage lifetime of the ThreadState attached
    // to the main thread explicitly instead and still use normal
    // constructor and destructor for the ThreadState class.
    // For this we reserve static storage for the main ThreadState
    // and lazily construct ThreadState in it using placement new.
    static uint8_t s_mainThreadStateStorage[];

    ThreadIdentifier m_thread;
    WrapperPersistentRegion* m_liveWrapperPersistents;
    WrapperPersistentRegion* m_pooledWrapperPersistents;
    size_t m_pooledWrapperPersistentRegionCount;
    OwnPtr<PersistentNode> m_persistents;
    StackState m_stackState;
    intptr_t* m_startOfStack;
    intptr_t* m_endOfStack;
    void* m_safePointScopeMarker;
    Vector<Address> m_safePointStackCopy;
    bool m_atSafePoint;
    Vector<Interruptor*> m_interruptors;
    bool m_gcRequested;
    bool m_forcePreciseGCForTesting;
    volatile int m_sweepRequested;
    bool m_sweepInProgress;
    size_t m_noAllocationCount;
    bool m_inGC;
    BaseHeap* m_heaps[NumberOfHeaps];
    OwnPtr<HeapContainsCache> m_heapContainsCache;
    HeapStats m_stats;
    HeapStats m_statsAfterLastGC;

    Vector<OwnPtr<CleanupTask> > m_cleanupTasks;
    bool m_isTerminating;

    bool m_lowCollectionRate;

    OwnPtr<blink::WebThread> m_sweeperThread;
    int m_numberOfSweeperTasks;
    Mutex m_sweepMutex;
    ThreadCondition m_sweepThreadCondition;

    CallbackStack* m_weakCallbackStack;

#if defined(ADDRESS_SANITIZER)
    void* m_asanFakeStack;
#endif
};

template<ThreadAffinity affinity> class ThreadStateFor;

template<> class ThreadStateFor<MainThreadOnly> {
public:
    static ThreadState* state()
    {
        // This specialization must only be used from the main thread.
        ASSERT(ThreadState::current()->isMainThread());
        return ThreadState::mainThreadState();
    }
};

template<> class ThreadStateFor<AnyThread> {
public:
    static ThreadState* state() { return ThreadState::current(); }
};

// The SafePointAwareMutexLocker is used to enter a safepoint while waiting for
// a mutex lock. It also ensures that the lock is not held while waiting for a GC
// to complete in the leaveSafePoint method, by releasing the lock if the
// leaveSafePoint method cannot complete without blocking, see
// SafePointBarrier::checkAndPark.
class SafePointAwareMutexLocker {
    WTF_MAKE_NONCOPYABLE(SafePointAwareMutexLocker);
public:
    explicit SafePointAwareMutexLocker(MutexBase& mutex, ThreadState::StackState stackState = ThreadState::HeapPointersOnStack)
        : m_mutex(mutex)
        , m_locked(false)
    {
        ThreadState* state = ThreadState::current();
        do {
            bool leaveSafePoint = false;
            // We cannot enter a safepoint if we are currently sweeping. In that
            // case we just try to acquire the lock without being at a safepoint.
            // If another thread tries to do a GC at that time it might time out
            // due to this thread not being at a safepoint and waiting on the lock.
            if (!state->isSweepInProgress() && !state->isAtSafePoint()) {
                state->enterSafePoint(stackState, this);
                leaveSafePoint = true;
            }
            m_mutex.lock();
            m_locked = true;
            if (leaveSafePoint) {
                // When leaving the safepoint we might end up release the mutex
                // if another thread is requesting a GC, see
                // SafePointBarrier::checkAndPark. This is the case where we
                // loop around to reacquire the lock.
                state->leaveSafePoint(this);
            }
        } while (!m_locked);
    }

    ~SafePointAwareMutexLocker()
    {
        ASSERT(m_locked);
        m_mutex.unlock();
    }

private:
    friend class SafePointBarrier;

    void reset()
    {
        ASSERT(m_locked);
        m_mutex.unlock();
        m_locked = false;
    }

    MutexBase& m_mutex;
    bool m_locked;
};

// Common header for heap pages. Needs to be defined before class Visitor.
class BaseHeapPage {
public:
    BaseHeapPage(PageMemory*, const GCInfo*, ThreadState*);
    virtual ~BaseHeapPage() { }

    // Check if the given address points to an object in this
    // heap page. If so, find the start of that object and mark it
    // using the given Visitor. Otherwise do nothing. The pointer must
    // be within the same aligned blinkPageSize as the this-pointer.
    //
    // This is used during conservative stack scanning to
    // conservatively mark all objects that could be referenced from
    // the stack.
    virtual void checkAndMarkPointer(Visitor*, Address) = 0;
    virtual bool contains(Address) = 0;

#if ENABLE(GC_PROFILE_MARKING)
    virtual const GCInfo* findGCInfo(Address) = 0;
#endif

    Address address() { return reinterpret_cast<Address>(this); }
    PageMemory* storage() const { return m_storage; }
    ThreadState* threadState() const { return m_threadState; }
    const GCInfo* gcInfo() { return m_gcInfo; }
    virtual bool isLargeObject() { return false; }
    virtual void markOrphaned()
    {
        m_threadState = 0;
        m_gcInfo = 0;
        m_terminating = false;
        m_tracedAfterOrphaned = false;
    }
    bool orphaned() { return !m_threadState; }
    bool terminating() { return m_terminating; }
    void setTerminating() { m_terminating = true; }
    bool tracedAfterOrphaned() { return m_tracedAfterOrphaned; }
    void setTracedAfterOrphaned() { m_tracedAfterOrphaned = true; }
    size_t promptlyFreedSize() { return m_promptlyFreedSize; }
    void resetPromptlyFreedSize() { m_promptlyFreedSize = 0; }
    void addToPromptlyFreedSize(size_t size) { m_promptlyFreedSize += size; }

private:
    PageMemory* m_storage;
    const GCInfo* m_gcInfo;
    ThreadState* m_threadState;
    // Pointer sized integer to ensure proper alignment of the
    // HeapPage header. We use some of the bits to determine
    // whether the page is part of a terminting thread or
    // if the page is traced after being terminated (orphaned).
    uintptr_t m_terminating : 1;
    uintptr_t m_tracedAfterOrphaned : 1;
    uintptr_t m_promptlyFreedSize : 17; // == blinkPageSizeLog2
};

}

#endif // ThreadState_h


text:
namespace Merchello.Bazaar.Install
{
    using System;

    using Umbraco.Web.UI.Controls;

    /// <summary>
    /// The installer.
    /// </summary>
    public class Installer : UmbracoUserControl
    {
        /// <summary>
        /// Gets the store url.
        /// </summary>
        public string StoreUrl { get; private set; }

        /// <summary>
        /// Handles the initialize event.
        /// </summary>
        /// <param name="e">
        /// The <see cref="EventArgs"/>.
        /// </param>
        protected override void OnInit(EventArgs e)
        {
            base.OnInit(e);

            var dataInstaller = new BazaarDataInstaller();
            var root = dataInstaller.Execute();

            StoreUrl = Umbraco.TypedContent(root.Id).Url;
        }
    }
}

redpajama_set_name: RedPajamaGithub


text:
module Agent #:nodoc:

  # To be mixed into an element via the RindaRunner...
  module RindaService

    cattr_accessor :tuple_space, :tuple_uri

    # How do I mix this in with another initialize.  I'm not sure...
    def initialize
      require 'rinda/rinda'
      require 'rinda/ring'
      include Rinda
      DRb.start_service
      @tuple_space = RingFinger.primary

      @services = []
    end

  end
end


text:
from __future__ import unicode_literals

from django.db import migrations, models

class Migration(migrations.Migration):

    dependencies = [
        ('jobs', '0001_initial'),
    ]

    operations = [
        migrations.AlterField(
            model_name='job',
            name='content',
            field=models.TextField(max_length=800),
        ),
        migrations.AlterField(
            model_name='job',
            name='slug',
            field=models.SlugField(unique=True, max_length=200),
        ),
    ]


text:
using System;

namespace DigitallyCreated.ExpressionTreeRewriter
{
	/// <summary>
	/// Marks a method as a method that should be replaced by the <see cref="ExpressionTreeRewriter"/> using an
	/// instance of the class specified by <see cref="RewriterClass"/>.
	/// </summary>
	public class RewriteUsingRewriterClassAttribute : RewriterMarkerMethodAttribute
	{
		private readonly Type _rewriterClass;

		/// <summary>
		/// The type of the class that can rewrite the method call into another expression.
		/// This type will implement <see cref="IExpressionRewriter"/>
		/// </summary>
		public Type RewriterClass
		{
			get { return _rewriterClass; }
		}

		/// <summary>
		/// Constructor, creates a <see cref="RewriteUsingRewriterClassAttribute"/>
		/// </summary>
		/// <param name="rewriterClass">
		/// The type of the class that can rewrite the method call into another expression.
		/// This type must implement <see cref="IExpressionRewriter"/>
		/// </param>
		public RewriteUsingRewriterClassAttribute(Type rewriterClass)
		{
			if (typeof(IExpressionRewriter).IsAssignableFrom(rewriterClass) == false)
				throw new ArgumentException("rewriterClass must implement IExpressionRewriter", "rewriterClass");

			_rewriterClass = rewriterClass;
		}

		/// <summary>
		/// Creates an <see cref="IExpressionRewriter"/> that can rewrite out the marker method
		/// this attribute is applied to.
		/// </summary>
		/// <returns>The <see cref="IExpressionRewriter"/></returns>
		public override IExpressionRewriter CreateRewriter()
		{
			return (IExpressionRewriter)Activator.CreateInstance(_rewriterClass);
		}
	}
}


text:
// ==================================================================================================================
//        ,::i                                                           BBB
//       BBBBBi                                                         EBBB
//      MBBNBBU                                                         BBB,
//     BBB. BBB     BBB,BBBBM   BBB   UBBB   MBB,  LBBBBBO,   :BBG,BBB :BBB  .BBBU  kBBBBBF
//    BBB,  BBB    7BBBBS2BBBO  BBB  iBBBB  YBBJ :BBBMYNBBB:  FBBBBBB: OBB: 5BBB,  BBBi ,M,
//   MBBY   BBB.   8BBB   :BBB  BBB .BBUBB  BB1  BBBi   kBBB  BBBM     BBBjBBBr    BBB1
//  BBBBBBBBBBBu   BBB    FBBP  MBM BB. BB BBM  7BBB    MBBY .BBB     7BBGkBB1      JBBBBi
// PBBBFE0GkBBBB  7BBX   uBBB   MBBMBu .BBOBB   rBBB   kBBB  ZBBq     BBB: BBBJ   .   iBBB
//BBBB      iBBB  BBBBBBBBBE    EBBBB  ,BBBB     MBBBBBBBM   BBB,    iBBB  .BBB2 :BBBBBBB7
//vr7        777  BBBu8O5:      .77r    Lr7       .7EZk;     L77     .Y7r   irLY  JNMMF:
//               LBBj
//
// Apworks Application Development Framework
// Copyright (C) 2009-2018 by daxnet.
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//    http://www.apache.org/licenses/LICENSE-2.0
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
// ==================================================================================================================

using System;
using System.Threading;
using System.Threading.Tasks;

namespace Apworks.Messaging
{
    /// <summary>
    /// Represents the message handler execution context, in which the message handlers are going
    /// to handle the received messages in a particular context, so that the dependent resources
    /// can be easily managed and maintained.
    /// </summary>
    public interface IMessageHandlerExecutionContext
    {
        /// <summary>
        /// Registers the message handler to the current execution context.
        /// </summary>
        /// <typeparam name="TMessage">The type of the message whose handler is going to be registered.</typeparam>
        /// <typeparam name="THandler">The type of the message handler that is going to be registered.</typeparam>
        void RegisterHandler<TMessage, THandler>()
            where TMessage : IMessage
            where THandler : IMessageHandler<TMessage>;

        /// <summary>
        /// Checks whether the message handler of the specified message type has been registered.
        /// </summary>
        /// <typeparam name="TMessage">The type of the message whose handler's existence should be checked.</typeparam>
        /// <typeparam name="THandler">The type of the message handler whose existence should be checked.</typeparam>
        /// <returns></returns>
        bool HandlerRegistered<TMessage, THandler>()
            where TMessage : IMessage
            where THandler : IMessageHandler<TMessage>;

        void RegisterHandler(Type messageType, Type handlerType);

        bool HandlerRegistered(Type messageType, Type handlerType);

        Task HandleMessageAsync(IMessage message, CancellationToken cancellationToken = default(CancellationToken));
    }
}


text:
#ifndef __SLAVE_SLAVE_LOG_EVENT_H
#define __SLAVE_SLAVE_LOG_EVENT_H

#include "relayloginfo.h"

namespace slave {

/*
 *
 */

enum Log_event_type
{

  UNKNOWN_EVENT= 0,
  START_EVENT_V3= 1,
  QUERY_EVENT= 2,
  STOP_EVENT= 3,
  ROTATE_EVENT= 4,
  INTVAR_EVENT= 5,
  LOAD_EVENT= 6,
  SLAVE_EVENT= 7,
  CREATE_FILE_EVENT= 8,
  APPEND_BLOCK_EVENT= 9,
  EXEC_LOAD_EVENT= 10,
  DELETE_FILE_EVENT= 11,
  NEW_LOAD_EVENT= 12,
  RAND_EVENT= 13,
  USER_VAR_EVENT= 14,
  FORMAT_DESCRIPTION_EVENT= 15,
  XID_EVENT= 16,
  BEGIN_LOAD_QUERY_EVENT= 17,
  EXECUTE_LOAD_QUERY_EVENT= 18,

  TABLE_MAP_EVENT = 19,

  PRE_GA_WRITE_ROWS_EVENT = 20,
  PRE_GA_UPDATE_ROWS_EVENT = 21,
  PRE_GA_DELETE_ROWS_EVENT = 22,

  WRITE_ROWS_EVENT = 23,
  UPDATE_ROWS_EVENT = 24,
  DELETE_ROWS_EVENT = 25,

  INCIDENT_EVENT = 26,

  HEARTBEAT_LOG_EVENT= 27,

  ENUM_END_EVENT
};

#define LOG_EVENT_TYPES (ENUM_END_EVENT-1)

#define EVENT_TYPE_OFFSET    4
#define SERVER_ID_OFFSET     5
#define EVENT_LEN_OFFSET     9
#define LOG_POS_OFFSET       13

#define LOG_EVENT_HEADER_LEN 19

#define ROTATE_HEADER_LEN    8
#define R_POS_OFFSET       0

#define QUERY_HEADER_MINIMAL_LEN     (4 + 4 + 1 + 2)
#define QUERY_HEADER_LEN     (QUERY_HEADER_MINIMAL_LEN + 2)
#define Q_STATUS_VARS_LEN_OFFSET 11
#define Q_DB_LEN_OFFSET             8

#define TABLE_MAP_HEADER_LEN   8
#define TM_MAPID_OFFSET    0

#define ROWS_HEADER_LEN        8
#define RW_MAPID_OFFSET    0
#define ROWS_HEADER_LEN        8

#define LOG_EVENT_MINIMAL_HEADER_LEN 19

#define ST_BINLOG_VER_LEN           2
#define ST_BINLOG_VER_OFFSET        0
#define ST_SERVER_VER_LEN 50
#define ST_SERVER_VER_OFFSET        (ST_BINLOG_VER_OFFSET + 2)
#define ST_CREATED_OFFSET     (ST_SERVER_VER_OFFSET + ST_SERVER_VER_LEN)
#define ST_COMMON_HEADER_LEN_OFFSET (ST_CREATED_OFFSET + 4)

#define START_V3_HEADER_LEN     (2 + ST_SERVER_VER_LEN + 4)

//-----------------------------------------------------------------------------------------

struct Basic_event_info {

    slave::Log_event_type type;
    size_t log_pos;
    time_t when;
    unsigned int server_id;

    const char* buf;
    unsigned int event_len;

    Basic_event_info() : type(UNKNOWN_EVENT), log_pos(0), when(0), server_id(0), buf(NULL), event_len(0)
        {}

    void parse(const char* b, unsigned int el);
};

struct Rotate_event_info {

    unsigned int ident_len;
    std::string new_log_ident;
    size_t pos;

    Rotate_event_info(const char* buf, unsigned int event_len);
};

struct Query_event_info {

    std::string query;

    Query_event_info(const char* buf, unsigned int event_len);
};

struct Table_map_event_info {

    unsigned long m_table_id;
    std::string m_tblnam;
    std::string m_dbnam;

    Table_map_event_info(const char* buf, unsigned int event_len);
};

struct Row_event_info {

    unsigned long m_width;
    unsigned long m_table_id;

    std::vector<unsigned char> m_cols;
    std::vector<unsigned char> m_cols_ai;

    unsigned char* m_rows_buf;
    unsigned char* m_rows_end;

    bool has_after_image;

    Row_event_info(const char* buf, unsigned int event_len, bool do_update);
};

bool read_log_event(const char* buf, unsigned int event_len, Basic_event_info& info);

void apply_row_event(slave::RelayLogInfo& rli, const Basic_event_info& bei, const Row_event_info& roi, ExtStateIface &ext_state);

//------------------------------------------------------------------------------------------

}; //namespace

#endif


text:
<!DOCTYPE html>

    <meta charset="utf-8">
    <meta http-equiv="X-UA-Compatible" content="IE=edge">
    <meta name="viewport" content="width=device-width, initial-scale=1">

    <meta property="og:site_name" content="Łukasz Chrząszcz">
    <meta property="og:type" content="article">

    <meta property="og:image" content="http://chrzaszcz.dev/img/2019-09-kafka-streams-store-basics/background.jpg">
    <meta property="twitter:image" content="http://chrzaszcz.dev/img/2019-09-kafka-streams-store-basics/background.jpg" />

    <meta name="title" content="How to count events in Kafka Streams?" />
    <meta property="og:title" content="How to count events in Kafka Streams?" />
    <meta property="twitter:title" content="How to count events in Kafka Streams?" />

    <meta name="description" content="Kafka Streams enables you to use stateful processing via Stores. Let's go through a simple example of order-counting app and discover how Stores fit in the stream processing world">
    <meta property="og:description" content="Kafka Streams enables you to use stateful processing via Stores. Let's go through a simple example of order-counting app and discover how Stores fit in the stream processing world" />
    <meta property="twitter:description" content="Kafka Streams enables you to use stateful processing via Stores. Let's go through a simple example of order-counting app and discover how Stores fit in the stream processing world" />

    <meta property="twitter:card" content="summary" />

    <meta name="keyword"  content="">
    <link rel="shortcut icon" href="/img/favicon.ico">

    <title>How to count events in Kafka Streams? - Łukasz Chrząszcz Blog</title>

    <link rel="canonical" href="/2019/09/kafka-streams-store-basics/">

    <link rel="stylesheet" href="/css/iDisqus.min.css"/>

    <link rel="stylesheet" href="/css/bootstrap.min.css">

    <link rel="stylesheet" href="/css/hux-blog.min.css">

    <link rel="stylesheet" href="/css/zanshang.css">

    <link rel="stylesheet" href="/css/main.css">

    <link href="//cdnjs.cloudflare.com/ajax/libs/font-awesome/4.6.3/css/font-awesome.min.css" rel="stylesheet" type="text/css">

            <button type="button" class="navbar-toggle">
                Toggle navigation

            </button>
            Łukasz Chrząszcz

                        Home

ABOUT

                            kafka

How to count events in Kafka Streams?

Stores and aggregation in Kafka Streams

			Posted by

			    Łukasz Chrząszcz

			on
			Saturday, September 28, 2019

Stateful?

In the previous post, we have discussed how to define topologies in Kafka Streams to apply our processing logic to every record and send it to another topic. Examples we have run through, revolved around stateless processing, but moving forward, what if we want to aggregate some events? What if we want to calculate the average value of users’ orders in the last hour or calculate how many times did users order pizza in our pizzeria?

Since in Kafka we are living in an endless stream world, to achieve such aggregations we have to maintain some state to quickly apply newly arrived events and update our data. So the question is, can we introduce state to our Kafka Streams application?

Yes, we can, and stores is a mechanism enabling us to do so.

Today, we will analyze an example of the Kafka Streams application that leverages stores to count how many times did each user order pizza in our imaginary restaurant.

Our App

Let’s think of a simple scenario. You run a restaurant, and you offer your customers a selection of meals including among other pasta, burgers, and pizza. You want especially to improve your pizza delivery experience, so you thought of giving something extra for customers that order a lot of pizza in your restaurant.

Your employees could add extra sauces, coupons, etc. to every package, but they need to know how many pizzas did the particular client order since the existence of your company, to treat better the regular customers.

Let’s assume we have a stream of order events. Every event contains customerId and meal (for simplicity one order is one meal). We will create a Kafka Streams application, that selects only pizza orders and counts them based on customerId. As a result, we expect to receive events about the current total pizza orders for each user.

We can achieve logic like that with just a few lines in Kafka Streams. Take a look at the snippet below. This is the code we will be dealing with today. If you want to check the whole project, head over to Github.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

val builder = StreamsBuilder()
val messagesStream = builder.stream<String, Order>("Orders", Consumed.with(Serdes.String(), orderSerdes))

messagesStream
    .filter { _, value -> value.meal == "pizza" }
    .groupBy { _, value -> value.customerId }
    .count(Materialized
        .`as`<String, Long, KeyValueStore<Bytes, ByteArray>> ("TotalPizzaOrdersStore")
        .withCachingDisabled())
    .toStream()
    .to("TotalPizzaOrders", Produced.with(Serdes.String(), Serdes.Long()))

val topology = builder.build()

logger.info { topology.describe() }

val streams = KafkaStreams(topology, props)
streams.start()

What is important here? Let’s see.

Line 1-2 - We are creating a stream builder as usual. We will be fetching records from Orders topic. The only difference is that we have provided custom Serdes. The incoming events will be in JSON format, so we need a custom serializer/deserializer. It is pretty straightforward, but if you are interested in the implementation, head over to the repository and check it out.

Line 5 - We are interested only in pizza orders, hence filtering here.

Line 6 - All the incoming messages will be grouped based on a customer who ordered them. So all Bob’s orders will end up in one group, while all Alice’s orders will end up in another group.

Line 7 - Each group will be counted separately, and the result will be stored in a store called TotalPizzaOrdersStore. You might wonder what is with .withCachingDisabled method. By default, Kafka Streams uses RocksDB for stores, and every update requires I/O since it is stored on a disk. So to prevent the heavy load on I/O, Kafka Streams caches incoming updates on a store and occasionally flushes writes to a DB. This would result in our employees receiving an update only once in a while. We would like to receive every update immediately, so we are disabling this caching. This is also good for demonstration purposes.

Line 10 - count method returns a KTable, so if we want to operate on a stream of updates to the table, we have to convert it to KStream again.

Line 11 - We are taking our stream of pizza orders count updates and publish it to the TotalPizzaOrders topic.

Let’s run this example and see how it works. The short video below presents how does our application count pizza orders. In the top terminal, I am writing some orders for users Bob and Alice, and in the bottom terminal, you can see the updating pizza order count for them.

How does it work?

The most important concept we are dealing with today is a store. It is a key-value store holding some aggregated data derived from a stream. In Kafka Streams, you can have 2 kinds of stores: local store, and global store.

Store types

Local Store is a data about one particular partition from an input topic. So since every instance of your service has different partitions assigned, it has different stores with different data gathered from different partitions.

On the other hand, there is a global store, which - as you might have guessed - is present on all Kafka Streams application instances and contains all the data across all the partitions.

In our case, we do not need shared data between instances, so our store is a simple local store. Alice’s data might end up in one store and Bob’s in a totally different store.

Repartitioning

If that is the case, and the local store is closely connected to one of the input partitions, then what happens if you publish orders for Bob with different keys? You still need to count them as Bob’s orders, but they might end up in different instances of your service. If that would happen we could not be able to count them. Look at the picture below.

Each instance would know about a fraction of all Bob’s orders, and could not say globally, how many times did he order pizza. That is why one particular user’s order events should end up in the same instance to correctly count them, right?

To gather the same values in the same instance, Kafka Streams employs repartitioning. This is a mechanism that splits a topology in half. The first part sends records with a new key to an intermediate topic, and the second one fetches records from this topic and follows up with the rest of the logic. Doing this ensures that all the records with the same key end up in the same partition of the intermediate topic, so it is fetched by the same consumer.

Take a look at the diagram below, where the producer publishes 3 Bob’s orders with keys 111, 222, and 333. As you see, any instance can receive Bob’s order, but as soon as they do, they change the key to the customer name and send it back to Kafka.

As I said above, we can split topology in half, and it is exactly what Kafka Streams does. Let’s print the topology and find out for ourselves:

   Sub-topology: 0
    Source: KSTREAM-SOURCE-0000000000 (topics: [Orders])
      --> KSTREAM-FILTER-0000000001
    Processor: KSTREAM-FILTER-0000000001 (stores: [])
      --> KSTREAM-KEY-SELECT-0000000002
      <-- KSTREAM-SOURCE-0000000000
    Processor: KSTREAM-KEY-SELECT-0000000002 (stores: [])
      --> KSTREAM-FILTER-0000000005
      <-- KSTREAM-FILTER-0000000001
    Processor: KSTREAM-FILTER-0000000005 (stores: [])
      --> KSTREAM-SINK-0000000004
      <-- KSTREAM-KEY-SELECT-0000000002
    Sink: KSTREAM-SINK-0000000004 (topic: TotalPizzaOrdersStore-repartition)
      <-- KSTREAM-FILTER-0000000005

  Sub-topology: 1
    Source: KSTREAM-SOURCE-0000000006 (topics: [TotalPizzaOrdersStore-repartition])
      --> KSTREAM-AGGREGATE-0000000003
    Processor: KSTREAM-AGGREGATE-0000000003 (stores: [TotalPizzaOrdersStore])
      --> KTABLE-TOSTREAM-0000000007
      <-- KSTREAM-SOURCE-0000000006
    Processor: KTABLE-TOSTREAM-0000000007 (stores: [])
      --> KSTREAM-SINK-0000000008
      <-- KSTREAM-AGGREGATE-0000000003
    Sink: KSTREAM-SINK-0000000008 (topic: TotalPizzaOrders)
      <-- KTABLE-TOSTREAM-0000000007

At first, you see that there are two sub-topologies. As we stated previously, the first one ends up in publishing records to TotalPizzaOrdersStore-repartition topic, while the other one starts from fetching from TotalPizzaOrdersStore-repartition topic.

Let’s subscribe to this internal topic to have a pick at what is sent there. The internal topic’s name is store-sample-dsl-application-TotalPizzaOrdersStore-repartition. Prefix store-sample-dsl-application is inherited from Kafka Stream’s application name, to prevent conflicts with other application’s topics.

Take a look at the video below to check what is being sent to the repartitioning topic. As you see it is the original event, but with a key changed to the customerId.

What about memory?

So far so good. We have pizza orders counts for all our users.

Along with being enormously satisfied with what we have achieved, we should be equally worried about the whole order history fitting into our memory, right?

Keeping everything in RAM, would not only be problematic but also wasteful, as a lot of customers order only once in a while. That is why Kafka Streams by default leverages RocksDB to store all the data on a disk.

I would not be myself if I would not dig into the database itself to see what is stored there. Let’s go there together and check!

Usually Kafka Streams stores databases in /tmp directory. In our case, for application named store-sample-dsl-application all the databases are stored in /tmp/kafka-streams/store-sample-dsl-application directory.

Listing all the directories there shows us all the tasks that were created for our application.

[0] % ls /tmp/kafka-streams/store-sample-dsl-application
1_0  1_1  1_2  1_3  1_4  1_5  1_6  1_7

We can check the content of those directories, and in my case two of them actually contained some data. That is reasonable since we have been sending orders for Alice and Bob (and those are two different keys).

We can use an ldb tool shipped with RocksDB to scan the database. The printed values are a hex representation of order count with a timestamp. The first 8 bytes are a timestamp, and the rest is a count. If you are curious about the implementation, you can check the Kafka Streams source here

theer@Theerpad : /tmp/kafka-streams/store-sample-dsl-application/1_7/rocksdb/TotalPizzaOrdersStore
[0] % ldb --db=`pwd` scan --column_family=keyValueWithTimestamp --value_hex
Bob : 0x0000016D787ABAE80000000000000022

theer@Theerpad : /tmp/kafka-streams/store-sample-dsl-application/1_1/rocksdb/TotalPizzaOrdersStore
[0] % ldb --db=`pwd` scan --column_family=keyValueWithTimestamp --value_hex
Alice : 0x0000016D78A7D984000000000000000C

I was playing for a while with the application, and the last update I have received for Bob was 34 and for Alice 12. This is exactly what we can see in the snippet above in an output from RocksDB.

Durability

A crucial thing to remember is that RocksDB is not used for durability. It is entirely disposable, so if you run your services on Kubernetes or similar technology you would not have to run it as a StatefulSet or provision any shared storage.

Does that mean in case of failure your application has to process and aggregate all events from the beginning?

Fortunately, no. Kafka Streams sends all values from the store after to another topic. This topic is compacted, so it keeps at least the last value for each of the keys. Given that, in case of your service failure, the new instance will just scan this changelog topic and rebuild the store.

In our case, this changelog topic is named store-sample-dsl-application-TotalPizzaOrdersStore-changelog. We can query it from beginning to see what has been sent there so far. Take a look at the snippet below (for readability, I skipped some parts with “…").

bash-4.4# ./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic store-sample-dsl-application-TotalPizzaOrdersStore-changelog --property print.key=true --from-beginning --value-deserializer org.apache.kafka.common.serialization.LongDeserializer
Bob	1
Bob	2
Bob	3
...
Bob	32
Bob	33
Bob	34
Alice	1
Alice	2
Alice	3
...
Alice	11
Alice	12

Additionally, if you want your processing to be up-and-running even sooner, you can enable standby mode in which, every store is followed by a standby instance that is ready to take over the active one as soon as it goes down, but that is a material for future posts.

Summary

I think we got a pretty decent grasp of how stateful processing looks like in Kafka Streams. Even though it was only a simple counting example, we could observe a lot of mechanisms underneath.

You can achieve much more in terms of aggregation in Kafka Streams by introducing moving windowing over the stream of data or joining multiple streams. We will definitely dig into that in later posts.

However, for today let’s summarize what we have learned:

You can easily count everything since the beginning of your stream.

You can receive all the updates on a particular count value on another Kafka topic.

There are two store types: global store, and local store.

The Global store is present in all instances and has the same data.

The Local store is unique for every input partition, and instances have different stores, due to different partition assignment.

Counting requires gathering all events that need to be counted together in one instance, so if the input topic has a different partitioning key than the required grouping key, Kafka Streams has to perform repartitioning.

Repartitioning is basically just changing the key of the record to the chosen grouping key, and sending it to intermediate Kafka topic, so the events that have the same grouping key will end up on the same partition, hence the same instance.

Stores are not stored entirely in RAM. They are stored on a disk leveraging RocksDB.

RocksDB is not used for increased durability. It is simply a way to fit the huge datasets next to the application.

To prevent data loss, Kafka Streams sends all updates to the store value to the changelog topic. In case of a failure, another instance will take over the task, and restore the database from this changelog topic.

Thank you for reading the post. Please share your thoughts and comments, so I can improve future articles, and share knowledge with you in the best way I can 😄

                        ←
                            Previous Post

                        Next
                            Post →

Please enable JavaScript to view the comments powered by Disqus.
comments powered by Disqus

FEATURED TAGS

                            docker

                            kafka

                    Copyright © Łukasz Chrząszcz 2020

                    CleanWhite Hugo Theme by Huabing |


text:
using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Runtime.InteropServices;
using Microsoft.Build.Evaluation;
using Microsoft.VisualStudio;
using MSBuild = Microsoft.Build.Evaluation;

namespace Microsoft.VisualStudioTools.Project
{
    internal class MsBuildProjectElement : ProjectElement
    {
        private MSBuild.ProjectItem _item;
        private string _url; // cached Url

        /// <summary>
        /// Constructor to create a new MSBuild.ProjectItem and add it to the project
        /// Only have internal constructors as the only one who should be creating
        /// such object is the project itself (see Project.CreateFileNode()).
        /// </summary>
        internal MsBuildProjectElement(ProjectNode project, string itemPath, string itemType)
            : base(project)
        {
            Utilities.ArgumentNotNullOrEmpty("itemPath", itemPath);
            Utilities.ArgumentNotNullOrEmpty("itemType", itemType);

            // create and add the item to the project

            this._item = project.BuildProject.AddItem(itemType, Microsoft.Build.Evaluation.ProjectCollection.Escape(itemPath))[0];
            this._url = base.Url;
        }

        /// <summary>
        /// Constructor to Wrap an existing MSBuild.ProjectItem
        /// Only have internal constructors as the only one who should be creating
        /// such object is the project itself (see Project.CreateFileNode()).
        /// </summary>
        /// <param name="project">Project that owns this item</param>
        /// <param name="existingItem">an MSBuild.ProjectItem; can be null if virtualFolder is true</param>
        /// <param name="virtualFolder">Is this item virtual (such as reference folder)</param>
        internal MsBuildProjectElement(ProjectNode project, MSBuild.ProjectItem existingItem)
            : base(project)
        {
            Utilities.ArgumentNotNull("existingItem", existingItem);

            // Keep a reference to project and item
            this._item = existingItem;
            this._url = base.Url;
        }

        protected override string ItemType
        {
            get
            {
                return this._item.ItemType;
            }
            set
            {
                this._item.ItemType = value;
                OnItemTypeChanged();
            }
        }

        /// <summary>
        /// Set an attribute on the project element
        /// </summary>
        /// <param name="attributeName">Name of the attribute to set</param>
        /// <param name="attributeValue">Value to give to the attribute</param>
        public override void SetMetadata(string attributeName, string attributeValue)
        {
            Debug.Assert(!StringComparer.OrdinalIgnoreCase.Equals(attributeName, ProjectFileConstants.Include), "Use rename as this won't work");

            // Build Action is the type, not a property, so intercept
            if (StringComparer.OrdinalIgnoreCase.Equals(attributeName, ProjectFileConstants.BuildAction))
            {
                this._item.ItemType = attributeValue;
                return;
            }

            // Check out the project file.
            if (!this.ItemProject.QueryEditProjectFile(false))
            {
                throw Marshal.GetExceptionForHR(VSConstants.OLE_E_PROMPTSAVECANCELLED);
            }

            if (attributeValue == null)
            {
                this._item.RemoveMetadata(attributeName);
            }
            else
            {
                this._item.SetMetadataValue(attributeName, attributeValue);
            }
        }

        /// <summary>
        /// Get the value of an attribute on a project element
        /// </summary>
        /// <param name="attributeName">Name of the attribute to get the value for</param>
        /// <returns>Value of the attribute</returns>
        public override string GetMetadata(string attributeName)
        {
            // cannot ask MSBuild for Include, so intercept it and return the corresponding property
            if (StringComparer.OrdinalIgnoreCase.Equals(attributeName, ProjectFileConstants.Include))
            {
                return this._item.EvaluatedInclude;
            }

            // Build Action is the type, not a property, so intercept this one as well
            if (StringComparer.OrdinalIgnoreCase.Equals(attributeName, ProjectFileConstants.BuildAction))
            {
                return this._item.ItemType;
            }

            return this._item.GetMetadataValue(attributeName);
        }

        public override void Rename(string newPath)
        {
            var escapedPath = Microsoft.Build.Evaluation.ProjectCollection.Escape(newPath);

            this._item.Rename(escapedPath);
            this.RefreshProperties();
        }

        public override void RefreshProperties()
        {
            this.ItemProject.BuildProject.ReevaluateIfNecessary();

            this._url = base.Url;

            IEnumerable<ProjectItem> items = this.ItemProject.BuildProject.GetItems(this._item.ItemType);
            foreach (var projectItem in items)
            {
                if (projectItem != null && projectItem.UnevaluatedInclude.Equals(this._item.UnevaluatedInclude))
                {
                    this._item = projectItem;
                    return;
                }
            }
        }

        /// <summary>
        /// Calling this method remove this item from the project file.
        /// Once the item is delete, you should not longer be using it.
        /// Note that the item should be removed from the hierarchy prior to this call.
        /// </summary>
        public override void RemoveFromProjectFile()
        {
            if (!this.Deleted)
            {
                this.ItemProject.BuildProject.RemoveItem(this._item);
            }

            base.RemoveFromProjectFile();
        }

        internal MSBuild.ProjectItem Item => this._item;

        public override string Url => this._url;

        public override bool Equals(object obj)
        {
            // Do they reference the same element?
            if (object.ReferenceEquals(this, obj))
            {
                return true;
            }

            var msBuildProjElem = obj as MsBuildProjectElement;
            if (object.ReferenceEquals(msBuildProjElem, null))
            {
                return false;
            }

            // Do they reference the same project?
            if (!this.ItemProject.Equals(msBuildProjElem.ItemProject))
            {
                return false;
            }

            // Do they have the same include?
            var include1 = GetMetadata(ProjectFileConstants.Include);
            var include2 = msBuildProjElem.GetMetadata(ProjectFileConstants.Include);

            if (include1 == null || include2 == null)
            {
                return false;
            }

            return StringComparer.OrdinalIgnoreCase.Equals(include1, include2);
        }

        public override int GetHashCode()
        {
            return StringComparer.OrdinalIgnoreCase.GetHashCode(GetMetadata(ProjectFileConstants.Include));
        }
    }
}


text:
package com.healthcit.cacure.model.breadcrumb;

import com.healthcit.cacure.utils.Constants;

/**
 * Breadcrumb for displaying link to Manage Library page.
 *
 */
public class ManageLibraryBreadCrumb extends HomeBreadCrumb {

	@Override
	public Link getLink() {
		Link link = super.getLink();
		Link currentLink = new Link("Manage Library", Constants.LIBRARY_MANAGE_URI, this);
		this.addLastChild(link, currentLink);
		return link;
	}
}


text:
"""The tests for the  MQTT device_tracker discovery platform."""

import pytest

from homeassistant.components import device_tracker
from homeassistant.components.mqtt.const import DOMAIN as MQTT_DOMAIN
from homeassistant.components.mqtt.discovery import ALREADY_DISCOVERED
from homeassistant.const import STATE_HOME, STATE_NOT_HOME, STATE_UNKNOWN
from homeassistant.setup import async_setup_component

from .test_common import help_test_setting_blocked_attribute_via_mqtt_json_message

from tests.common import async_fire_mqtt_message, mock_device_registry, mock_registry

DEFAULT_CONFIG = {
    device_tracker.DOMAIN: {
        "platform": "mqtt",
        "name": "test",
        "state_topic": "test-topic",
    }
}

@pytest.fixture
def device_reg(hass):
    """Return an empty, loaded, registry."""
    return mock_device_registry(hass)

@pytest.fixture
def entity_reg(hass):
    """Return an empty, loaded, registry."""
    return mock_registry(hass)

async def test_discover_device_tracker(hass, mqtt_mock_entry_no_yaml_config, caplog):
    """Test discovering an MQTT device tracker component."""
    await mqtt_mock_entry_no_yaml_config()
    async_fire_mqtt_message(
        hass,
        "homeassistant/device_tracker/bla/config",
        '{ "name": "test", "state_topic": "test_topic" }',
    )
    await hass.async_block_till_done()

    state = hass.states.get("device_tracker.test")

    assert state is not None
    assert state.name == "test"
    assert ("device_tracker", "bla") in hass.data[ALREADY_DISCOVERED]

@pytest.mark.no_fail_on_log_exception
async def test_discovery_broken(hass, mqtt_mock_entry_no_yaml_config, caplog):
    """Test handling of bad discovery message."""
    await mqtt_mock_entry_no_yaml_config()
    async_fire_mqtt_message(
        hass,
        "homeassistant/device_tracker/bla/config",
        '{ "name": "Beer" }',
    )
    await hass.async_block_till_done()

    state = hass.states.get("device_tracker.beer")
    assert state is None

    async_fire_mqtt_message(
        hass,
        "homeassistant/device_tracker/bla/config",
        '{ "name": "Beer", "state_topic": "required-topic" }',
    )
    await hass.async_block_till_done()

    state = hass.states.get("device_tracker.beer")
    assert state is not None
    assert state.name == "Beer"

async def test_non_duplicate_device_tracker_discovery(
    hass, mqtt_mock_entry_no_yaml_config, caplog
):
    """Test for a non duplicate component."""
    await mqtt_mock_entry_no_yaml_config()
    async_fire_mqtt_message(
        hass,
        "homeassistant/device_tracker/bla/config",
        '{ "name": "Beer", "state_topic": "test-topic" }',
    )
    async_fire_mqtt_message(
        hass,
        "homeassistant/device_tracker/bla/config",
        '{ "name": "Beer", "state_topic": "test-topic" }',
    )
    await hass.async_block_till_done()

    state = hass.states.get("device_tracker.beer")
    state_duplicate = hass.states.get("device_tracker.beer1")

    assert state is not None
    assert state.name == "Beer"
    assert state_duplicate is None
    assert "Component has already been discovered: device_tracker bla" in caplog.text

async def test_device_tracker_removal(hass, mqtt_mock_entry_no_yaml_config, caplog):
    """Test removal of component through empty discovery message."""
    await mqtt_mock_entry_no_yaml_config()
    async_fire_mqtt_message(
        hass,
        "homeassistant/device_tracker/bla/config",
        '{ "name": "Beer", "state_topic": "test-topic" }',
    )
    await hass.async_block_till_done()
    state = hass.states.get("device_tracker.beer")
    assert state is not None

    async_fire_mqtt_message(hass, "homeassistant/device_tracker/bla/config", "")
    await hass.async_block_till_done()
    state = hass.states.get("device_tracker.beer")
    assert state is None

async def test_device_tracker_rediscover(hass, mqtt_mock_entry_no_yaml_config, caplog):
    """Test rediscover of removed component."""
    await mqtt_mock_entry_no_yaml_config()
    async_fire_mqtt_message(
        hass,
        "homeassistant/device_tracker/bla/config",
        '{ "name": "Beer", "state_topic": "test-topic" }',
    )
    await hass.async_block_till_done()
    state = hass.states.get("device_tracker.beer")
    assert state is not None

    async_fire_mqtt_message(hass, "homeassistant/device_tracker/bla/config", "")
    await hass.async_block_till_done()
    state = hass.states.get("device_tracker.beer")
    assert state is None

    async_fire_mqtt_message(
        hass,
        "homeassistant/device_tracker/bla/config",
        '{ "name": "Beer", "state_topic": "test-topic" }',
    )
    await hass.async_block_till_done()
    state = hass.states.get("device_tracker.beer")
    assert state is not None

async def test_duplicate_device_tracker_removal(
    hass, mqtt_mock_entry_no_yaml_config, caplog
):
    """Test for a non duplicate component."""
    await mqtt_mock_entry_no_yaml_config()
    async_fire_mqtt_message(
        hass,
        "homeassistant/device_tracker/bla/config",
        '{ "name": "Beer", "state_topic": "test-topic" }',
    )
    await hass.async_block_till_done()
    async_fire_mqtt_message(hass, "homeassistant/device_tracker/bla/config", "")
    await hass.async_block_till_done()
    assert "Component has already been discovered: device_tracker bla" in caplog.text
    caplog.clear()
    async_fire_mqtt_message(hass, "homeassistant/device_tracker/bla/config", "")
    await hass.async_block_till_done()

    assert (
        "Component has already been discovered: device_tracker bla" not in caplog.text
    )

async def test_device_tracker_discovery_update(
    hass, mqtt_mock_entry_no_yaml_config, caplog
):
    """Test for a discovery update event."""
    await mqtt_mock_entry_no_yaml_config()
    async_fire_mqtt_message(
        hass,
        "homeassistant/device_tracker/bla/config",
        '{ "name": "Beer", "state_topic": "test-topic" }',
    )
    await hass.async_block_till_done()

    state = hass.states.get("device_tracker.beer")
    assert state is not None
    assert state.name == "Beer"

    async_fire_mqtt_message(
        hass,
        "homeassistant/device_tracker/bla/config",
        '{ "name": "Cider", "state_topic": "test-topic" }',
    )
    await hass.async_block_till_done()

    state = hass.states.get("device_tracker.beer")
    assert state is not None
    assert state.name == "Cider"

async def test_cleanup_device_tracker(
    hass, hass_ws_client, device_reg, entity_reg, mqtt_mock_entry_no_yaml_config
):
    """Test discovered device is cleaned up when removed from registry."""
    assert await async_setup_component(hass, "config", {})
    await hass.async_block_till_done()
    mqtt_mock = await mqtt_mock_entry_no_yaml_config()
    ws_client = await hass_ws_client(hass)

    async_fire_mqtt_message(
        hass,
        "homeassistant/device_tracker/bla/config",
        '{ "device":{"identifiers":["0AFFD2"]},'
        '  "state_topic": "foobar/tracker",'
        '  "unique_id": "unique" }',
    )
    await hass.async_block_till_done()

    # Verify device and registry entries are created
    device_entry = device_reg.async_get_device({("mqtt", "0AFFD2")})
    assert device_entry is not None
    entity_entry = entity_reg.async_get("device_tracker.mqtt_unique")
    assert entity_entry is not None

    state = hass.states.get("device_tracker.mqtt_unique")
    assert state is not None

    # Remove MQTT from the device
    mqtt_config_entry = hass.config_entries.async_entries(MQTT_DOMAIN)[0]
    await ws_client.send_json(
        {
            "id": 6,
            "type": "config/device_registry/remove_config_entry",
            "config_entry_id": mqtt_config_entry.entry_id,
            "device_id": device_entry.id,
        }
    )
    response = await ws_client.receive_json()
    assert response["success"]
    await hass.async_block_till_done()
    await hass.async_block_till_done()

    # Verify device and registry entries are cleared
    device_entry = device_reg.async_get_device({("mqtt", "0AFFD2")})
    assert device_entry is None
    entity_entry = entity_reg.async_get("device_tracker.mqtt_unique")
    assert entity_entry is None

    # Verify state is removed
    state = hass.states.get("device_tracker.mqtt_unique")
    assert state is None
    await hass.async_block_till_done()

    # Verify retained discovery topic has been cleared
    mqtt_mock.async_publish.assert_called_once_with(
        "homeassistant/device_tracker/bla/config", "", 0, True
    )

async def test_setting_device_tracker_value_via_mqtt_message(
    hass, mqtt_mock_entry_no_yaml_config, caplog
):
    """Test the setting of the value via MQTT."""
    await mqtt_mock_entry_no_yaml_config()
    async_fire_mqtt_message(
        hass,
        "homeassistant/device_tracker/bla/config",
        '{ "name": "test", "state_topic": "test-topic" }',
    )

    await hass.async_block_till_done()

    state = hass.states.get("device_tracker.test")

    assert state.state == STATE_UNKNOWN

    async_fire_mqtt_message(hass, "test-topic", "home")
    state = hass.states.get("device_tracker.test")
    assert state.state == STATE_HOME

    async_fire_mqtt_message(hass, "test-topic", "not_home")
    state = hass.states.get("device_tracker.test")
    assert state.state == STATE_NOT_HOME

async def test_setting_device_tracker_value_via_mqtt_message_and_template(
    hass, mqtt_mock_entry_no_yaml_config, caplog
):
    """Test the setting of the value via MQTT."""
    await mqtt_mock_entry_no_yaml_config()
    async_fire_mqtt_message(
        hass,
        "homeassistant/device_tracker/bla/config",
        "{"
        '"name": "test", '
        '"state_topic": "test-topic", '
        '"value_template": "{% if value is equalto \\"proxy_for_home\\" %}home{% else %}not_home{% endif %}" '
        "}",
    )
    await hass.async_block_till_done()

    async_fire_mqtt_message(hass, "test-topic", "proxy_for_home")
    state = hass.states.get("device_tracker.test")
    assert state.state == STATE_HOME

    async_fire_mqtt_message(hass, "test-topic", "anything_for_not_home")
    state = hass.states.get("device_tracker.test")
    assert state.state == STATE_NOT_HOME

async def test_setting_device_tracker_value_via_mqtt_message_and_template2(
    hass, mqtt_mock_entry_no_yaml_config, caplog
):
    """Test the setting of the value via MQTT."""
    await mqtt_mock_entry_no_yaml_config()
    async_fire_mqtt_message(
        hass,
        "homeassistant/device_tracker/bla/config",
        "{"
        '"name": "test", '
        '"state_topic": "test-topic", '
        '"value_template": "{{ value | lower }}" '
        "}",
    )
    await hass.async_block_till_done()

    state = hass.states.get("device_tracker.test")
    assert state.state == STATE_UNKNOWN

    async_fire_mqtt_message(hass, "test-topic", "HOME")
    state = hass.states.get("device_Tracker.test")
    assert state.state == STATE_HOME

    async_fire_mqtt_message(hass, "test-topic", "NOT_HOME")
    state = hass.states.get("device_tracker.test")
    assert state.state == STATE_NOT_HOME

async def test_setting_device_tracker_location_via_mqtt_message(
    hass, mqtt_mock_entry_no_yaml_config, caplog
):
    """Test the setting of the location via MQTT."""
    await mqtt_mock_entry_no_yaml_config()
    async_fire_mqtt_message(
        hass,
        "homeassistant/device_tracker/bla/config",
        '{ "name": "test", "state_topic": "test-topic" }',
    )
    await hass.async_block_till_done()

    state = hass.states.get("device_tracker.test")

    assert state.state == STATE_UNKNOWN

    async_fire_mqtt_message(hass, "test-topic", "test-location")
    state = hass.states.get("device_tracker.test")
    assert state.state == "test-location"

async def test_setting_device_tracker_location_via_lat_lon_message(
    hass, mqtt_mock_entry_no_yaml_config, caplog
):
    """Test the setting of the latitude and longitude via MQTT."""
    await mqtt_mock_entry_no_yaml_config()
    async_fire_mqtt_message(
        hass,
        "homeassistant/device_tracker/bla/config",
        "{ "
        '"name": "test", '
        '"state_topic": "test-topic", '
        '"json_attributes_topic": "attributes-topic" '
        "}",
    )
    await hass.async_block_till_done()

    state = hass.states.get("device_tracker.test")

    assert state.state == STATE_UNKNOWN

    hass.config.latitude = 32.87336
    hass.config.longitude = -117.22743

    async_fire_mqtt_message(
        hass,
        "attributes-topic",
        '{"latitude":32.87336,"longitude": -117.22743, "gps_accuracy":1.5}',
    )
    state = hass.states.get("device_tracker.test")
    assert state.attributes["latitude"] == 32.87336
    assert state.attributes["longitude"] == -117.22743
    assert state.attributes["gps_accuracy"] == 1.5
    assert state.state == STATE_HOME

    async_fire_mqtt_message(
        hass,
        "attributes-topic",
        '{"latitude":50.1,"longitude": -2.1, "gps_accuracy":1.5}',
    )
    state = hass.states.get("device_tracker.test")
    assert state.attributes["latitude"] == 50.1
    assert state.attributes["longitude"] == -2.1
    assert state.attributes["gps_accuracy"] == 1.5
    assert state.state == STATE_NOT_HOME

    async_fire_mqtt_message(hass, "attributes-topic", '{"longitude": -117.22743}')
    state = hass.states.get("device_tracker.test")
    assert state.attributes["longitude"] == -117.22743
    assert state.state == STATE_UNKNOWN

    async_fire_mqtt_message(hass, "attributes-topic", '{"latitude":32.87336}')
    state = hass.states.get("device_tracker.test")
    assert state.attributes["latitude"] == 32.87336
    assert state.state == STATE_UNKNOWN

async def test_setting_blocked_attribute_via_mqtt_json_message(
    hass, mqtt_mock_entry_no_yaml_config
):
    """Test the setting of attribute via MQTT with JSON payload."""
    await help_test_setting_blocked_attribute_via_mqtt_json_message(
        hass,
        mqtt_mock_entry_no_yaml_config,
        device_tracker.DOMAIN,
        DEFAULT_CONFIG,
        None,
    )


text:
<?xml version="1.0" encoding="UTF-8"?>
<!--
  ~ Copyright 2016 Red Hat, Inc. and/or its affiliates.
  ~
  ~ Licensed under the Apache License, Version 2.0 (the "License");
  ~ you may not use this file except in compliance with the License.
  ~ You may obtain a copy of the License at
  ~
  ~       http://www.apache.org/licenses/LICENSE-2.0
  ~
  ~ Unless required by applicable law or agreed to in writing, software
  ~ distributed under the License is distributed on an "AS IS" BASIS,
  ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  ~ See the License for the specific language governing permissions and
  ~ limitations under the License.
  -->

<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xmlns="http://maven.apache.org/POM/4.0.0"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>

  <parent>
    <groupId>org.kie</groupId>
    <artifactId>business-central-theme</artifactId>
    <version>7.20.0-SNAPSHOT</version>
  </parent>

  <artifactId>business-central-theme-product</artifactId>
  <packaging>jar</packaging>

  <name>KIE Theme - Product</name>
  <description>KIE Theme - Product</description>

  <properties>
    <java.module.name>org.kie.wb.theme.product</java.module.name>
    <version.org.webjars.npm.rcue>3.18.1</version.org.webjars.npm.rcue>
  </properties>

  <dependencies>
    <dependency>
      <groupId>org.uberfire</groupId>
      <artifactId>uberfire-workbench-client-views-patternfly</artifactId>
      <scope>provided</scope>
    </dependency>
  </dependencies>

  <build>
    <plugins>
      <plugin>
        <groupId>org.apache.maven.plugins</groupId>
        <artifactId>maven-dependency-plugin</artifactId>
        <executions>
          <execution>
            <id>unpack-rcue</id>
            <phase>process-resources</phase>
            <goals>
              <goal>unpack</goal>
            </goals>
            <configuration>
              <artifactItems>
                <artifactItem>
                  <groupId>org.webjars.npm</groupId>
                  <artifactId>rcue</artifactId>
                  <version>${version.org.webjars.npm.rcue}</version>
                  <type>jar</type>
                  <overWrite>true</overWrite>
                  <outputDirectory>${project.build.directory}/rcue</outputDirectory>
                </artifactItem>
              </artifactItems>
              <overWriteReleases>false</overWriteReleases>
              <overWriteSnapshots>true</overWriteSnapshots>
            </configuration>
          </execution>
        </executions>
      </plugin>
      <plugin>
        <groupId>org.apache.maven.plugins</groupId>
        <artifactId>maven-resources-plugin</artifactId>
        <executions>
          <execution>
            <id>copy-rcue-resources</id>
            <phase>process-resources</phase>
            <goals>
              <goal>copy-resources</goal>
            </goals>
            <configuration>
              <outputDirectory>${project.build.outputDirectory}/org/kie/bc/public</outputDirectory>
              <resources>
                <resource>
                  <directory>${project.build.directory}/rcue/META-INF/resources/webjars/rcue/${version.org.webjars.npm.rcue}/dist</directory>
                  <includes>
                    <include>**</include>
                  </includes>
                </resource>
              </resources>
            </configuration>
          </execution>
        </executions>
      </plugin>
    </plugins>
  </build>

</project>


text:
/*
 * test : a simple system for introducing tests
 */

#ifndef HOBBES_TEST_SYSTEM_HPP_INCLUDED
#define HOBBES_TEST_SYSTEM_HPP_INCLUDED

#include <map>
#include <vector>
#include <string>
#include <set>
#include <stdexcept>
#include <sstream>
#include <hobbes/util/str.H>
#include <hobbes/util/stream.H>

#include <unistd.h>
#include <libgen.h>

#if defined(__APPLE__) && defined(__MACH__)
#  include <mach-o/dyld.h>
#elif __linux__
#  include <linux/limits.h>
#else
#  error "platform is not supported"
#endif

template<class Fn>
auto execPath(Fn && fn) -> void {

#if defined(__APPLE__) && defined(__MACH__)
#  if !defined(PROC_PIDPATHINFO_MAXSIZE) && !defined(PROC_PIDPATHINFO_MAXSIZE_HOBBES_FEATURED)
#    define PROC_PIDPATHINFO_MAXSIZE_HOBBES_FEATURED 2048
#  elif defined(PROC_PIDPATHINFO_MAXSIZE)
#    define PROC_PIDPATHINFO_MAXSIZE_HOBBES_FEATURED = PROC_PIDPATHINFO_MAXSIZE
#  endif
  std::string r(PROC_PIDPATHINFO_MAXSIZE_HOBBES_FEATURED, 0);
  uint32_t count = PROC_PIDPATHINFO_MAXSIZE_HOBBES_FEATURED;
  if (-1 != _NSGetExecutablePath(const_cast<char*>(r.data()), &count)) {
#elif __linux__
  std::string r(PATH_MAX, 0);
  if (-1 != ::readlink("/proc/self/exe", const_cast<char*>(r.data()), r.size())) {
#endif
    fn(std::string(dirname(const_cast<char*>(r.data()))));
  } else {
    throw std::runtime_error("failed to readlink(\"/proc/self/exe\") ...");
  }
}

struct Args final {
  Args() : report("test_report.json") {}

  std::set<std::string> tests;
  const char* report;
};

struct Result final {
  enum class Status {
    Pass,
    Fail,
    Skipped,
  };

  Result(const std::string& name) : testcase(name) {}

  void record(Status s, long d, const std::string& e = "") {
    status = s;
    duration = d;
    error = e;
  }

  std::string testcase;
  Status status {Status::Skipped};
  long duration {0};
  std::string error;
};

class TestCoord {
public:
  typedef void (*PTEST)();
  static TestCoord& instance();
  bool installTest(const std::string& group, const std::string& test, PTEST pf);
  std::set<std::string> testGroupNames() const;
  int runTestGroups(const Args&);

private:
  std::string toJSON();
  typedef std::vector<std::pair<std::string, PTEST>> Tests;
  typedef std::map<std::string, Tests> GroupedTests;
  typedef std::vector<Result> Results;
  typedef std::map<std::string, Results> GroupedResults;
  GroupedTests tests;
  GroupedResults results;
};

#define TEST(G,N) \
  void test_##G##_##N(); \
  bool install_##G##_##N = TestCoord::instance().installTest(#G, #N, &test_##G##_##N); \
  void test_##G##_##N()

#define FILEINFO(file, line) (std::string("(") + hobbes::str::rsplit(file, "/").second + ":" + std::to_string(line) + ")")

#define EXPECT_TRUE(p) \
  if (!(p)) { \
    std::ostringstream __errmsg; \
    __errmsg << "Expression false, expected true: " #p << " " << FILEINFO(__FILE__, __LINE__); \
    throw std::runtime_error(__errmsg.str()); \
  }
#define EXPECT_FALSE(p) \
  if ((p)) { \
    std::ostringstream __errmsg; \
    __errmsg << "Expression true, expected false: " #p << " " << FILEINFO(__FILE__, __LINE__); \
    throw std::runtime_error("Expression true, expected false: " #p); \
  }
#define EXPECT_EQ(p,x) \
  { \
    auto v = (p); \
    auto z = (x); \
    if (!(v == z)) { \
      std::ostringstream __errmsg; \
      __errmsg << "Expression '" #p "' == " << v << ", but expected " << z << " " << FILEINFO(__FILE__, __LINE__); \
      throw std::runtime_error(__errmsg.str()); \
    } \
  }
#define EXPECT_NEQ(p,x) \
  { \
    auto v = (p); \
    auto z = (x); \
    if (v == z) { \
      std::ostringstream __errmsg; \
      __errmsg << "Expression '" #p "' == " << v << ", but expected anything else" << " " << FILEINFO(__FILE__, __LINE__); \
      throw std::runtime_error(__errmsg.str()); \
    } \
  }
#define EXPECT_ALMOST_EQ(p,x,eps) \
  { \
    auto v    = (p); \
    auto z    = (x); \
    auto d    = v - z; \
    auto absd = (d < 0.0 ? -1.0 : 1.0) * d; \
    if (absd > eps) { \
      std::ostringstream __errmsg; \
      __errmsg << "Expression '" #p "' == " << v << ", but expected " << z << " " << FILEINFO(__FILE__, __LINE__); \
      throw std::runtime_error(__errmsg.str()); \
    } \
  }

#define EXPECT_EXCEPTION(p) \
  { \
    bool __pass = true; \
    try { \
      (p); \
      __pass = false; \
    } catch (...) { \
    } \
    if (!__pass) { \
      std::ostringstream __errmsg; \
      __errmsg << "Expression '" #p "' ran successfully but an exception was expected" << " " << FILEINFO(__FILE__, __LINE__); \
      throw std::runtime_error(__errmsg.str()); \
    } \
  }

#endif


text:
<?php
/**
 * @author Mark Scherer
 * @license http://www.opensource.org/licenses/mit-license.php The MIT License
 */
App::uses('EmailLib', 'Tools.Lib');
App::uses('AppShell', 'Console/Command');

class QueueEmailTask extends AppShell {

	/**
	 * List of default variables for EmailComponent
	 *
	 * @var array
	 */
	public $defaults = array(
		'to' => null,
		'from' => null,
	);

	public $timeout = 120;

	public $retries = 1;

	/**
	 * @var boolean
	 */
	public $autoUnserialize = true;

	public $Email;

	public function add() {
		$this->err('Queue Email Task cannot be added via Console.');
		$this->out('Please use createJob() on the QueuedTask Model to create a Proper Email Task.');
		$this->out('The Data Array should look something like this:');
		$this->out(var_export(array(
			'settings' => array(
				'to' => 'email@example.com',
				'subject' => 'Email Subject',
				'from' => 'system@example.com',
				'template' => 'sometemplate'
			),
			'vars' => array(
				'content' => 'hello world',
			)
		), true));
		$this->out('Alternativly, you can pass the whole EmailLib to directly use it.');
	}

	/**
	 * QueueEmailTask::run()
	 *
	 * @param mixed $data
	 * @return boolean Success
	 */
	public function run($data) {
		if (!isset($data['settings'])) {
			$this->err('Queue Email task called without settings data.');
			return false;
		}

		if (is_object($email = $data['settings']) && $email instanceof CakeEmail) {
			try {
				$transport = $email->transportClass();
				$config = $email->config();
				//echo returns($config);
				$transport->config($config);
				$result = $transport->send($email);

				if (!isset($config['log']) || !empty($config['logTrace']) && $config['logTrace'] === true) {
					$config['log'] = 'email_trace';
				} elseif (!empty($config['logTrace'])) {
					$config['log'] = $config['logTrace'];
				}
				if (isset($config['logTrace']) && !$config['logTrace']) {
					$config['log'] = false;
				}

				if (!empty($config['logTrace'])) {
					$this->_log($result, $config['log']);
				}
				return (bool)$result;
			} catch (Exception $e) {

				$error = $e->getMessage();
				$error .= ' (line ' . $e->getLine() . ' in ' . $e->getFile() . ')' . PHP_EOL . $e->getTraceAsString();
				CakeLog::write('email_error', $error);

				return false;
			}
		}

		$this->Email = new EmailLib();
		$settings = array_merge($this->defaults, $data['settings']);
		foreach ($settings as $method => $setting) {
			call_user_func_array(array($this->Email, $method), (array)$setting);
		}
		$message = null;
		if (!empty($data['vars'])) {
			if (isset($data['vars']['content'])) {
				$message = $data['vars']['content'];
			}
			$this->Email->viewVars($data['vars']);
		}
		return $this->Email->send($message);
	}

	protected function _log($contents, $log) {
		$config = array(
			'level' => LOG_DEBUG,
			'scope' => 'email'
		);
		if ($log !== true) {
			if (!is_array($log)) {
				$log = array('level' => $log);
			}
			$config = array_merge($config, $log);
		}
		CakeLog::write(
			$config['level'],
			PHP_EOL . $contents['headers'] . PHP_EOL . $contents['message'],
			$config['scope']
		);
	}

}


text:
using DogAgilityCompetition.Circe;
using DogAgilityCompetition.Controller.Engine.Storage;

namespace DogAgilityCompetition.Controller.Engine.Visualization.Changes;

/// <summary>
/// A visualization change that updates/clears class information.
/// </summary>
public sealed class ClassInfoUpdate : NullableVisualizationChange<CompetitionClassInfo>
{
    public static ClassInfoUpdate Hidden => new(null);

    public ClassInfoUpdate(CompetitionClassInfo? value)
        : base(value)
    {
    }

    public override void ApplyTo(IVisualizationActor actor)
    {
        Guard.NotNull(actor, nameof(actor));
        actor.SetClass(Value);
    }
}


text:
@echo off

echo --- Build tests.exe ---
dub build -b unittest -c unittests %*
if ERRORLEVEL 1 goto build_error

echo --- Run tests without coverage ---
build\tests.exe
if ERRORLEVEL 1 goto tests_failed

echo --- Run tests with coverage ---
if exist "%~dp0coverage" (
  rmdir "%~dp0coverage" /s /q
)

opencppcoverage --sources "%~dp0source" --modules %~dp0 --export_type "html:%~dp0coverage" build\tests.exe

echo --- Finished ---
exit /b 0

:build_error
echo --- BUILD FAILED ---
exit /b 1

:tests_failed
echo --- TESTS FAILED ---
exit /b 1


text:
<?php

class once extends core{
	function item_select(){//ok
		// used varibles
		$obj['errors'] =  array();
		$obj['error'] = 0 ;

		if($this->once_creator_check()){
			if($this->once_csrf_token_check($this->data['csrf_token'])){
				if($this->data['layer_id']>0){
					// Prepare statements to get all layers
					$stmt = $this->pdo->prepare("SELECT id, layer_id, plugin_id FROM edit_layers_cols WHERE id=:id");
					$stmt->bindParam(':id', $this->data['grid_id'], PDO::PARAM_INT);
					$stmt->execute();

					if($stmt->rowCount()){
						$wiersz=$stmt->fetch(PDO::FETCH_ASSOC);

						// Check if that plugin is rl used
						if($wiersz['plugin_id']==$this->data['plugin_id']){
							// Update source with default snippet code
							// Prepare statements to check if page snippet exist
							$stmt = $this->pdo->prepare("SELECT id FROM edit_snippets WHERE id=:id");
							$stmt->bindParam(':id', $this->data['id'], PDO::PARAM_INT);
							$stmt->execute();

							if($stmt->rowCount()){
								$row=$stmt->fetch(PDO::FETCH_ASSOC);

								// Copy snippet from folder and save to files.
								copy($this->data['root_path'].'/once/snippets/'.$row['id'].'/snippet.html',$this->data['root_path'].'/layers/layer_'.$wiersz['id'].'.php');

								// Make prefix to reconize layers
								$this->data['type']='layers';
								$this->data['path']='css';
								$this->data['file']='style.css';
								$this->data['source']=file_get_contents($this->data['root_path'].'/once/snippets/'.$row['id'].'/snippet.css');
								$this->data['id']=$this->data['grid_id'];
								$response=$this->once_save_source();

								$this->data['type']='layers';
								$this->data['path']='js';
								$this->data['file']='script.js';
								$this->data['source']=file_get_contents($this->data['root_path'].'/once/snippets/'.$row['id'].'/snippet.js');
								$this->data['id']=$this->data['grid_id'];
								$response=$this->once_save_source();
							}

							$obj['status']='ok';
						}else{
							$obj['errors'][]='Page snippet not already exists';
							$obj['error']++;
						}
					}else{
						$obj['errors'][]='Col not exist';
						$obj['error']++;
					}
				}else if($this->data['page_id']>0){
					// Prepare statements to get all layers
					$stmt = $this->pdo->prepare("SELECT id, page_id, plugin_id FROM edit_pages_cols WHERE id=:id");
					$stmt->bindParam(':id', $this->data['grid_id'], PDO::PARAM_INT);
					$stmt->execute();

					if($stmt->rowCount()){
						$wiersz=$stmt->fetch(PDO::FETCH_ASSOC);

						// Check if that plugin is rl used
						if($wiersz['plugin_id']==$this->data['plugin_id']){
							// Update source with default snippet code
							// Prepare statements to check if page snippet exist
							$stmt = $this->pdo->prepare("SELECT id FROM edit_snippets WHERE id=:id");
							$stmt->bindParam(':id', $this->data['id'], PDO::PARAM_INT);
							$stmt->execute();

							if($stmt->rowCount()){
								$row=$stmt->fetch(PDO::FETCH_ASSOC);

								// Copy snippet from folder and save to files.
								copy($this->data['root_path'].'/once/snippets/'.$row['id'].'/snippet.html',$this->data['root_path'].'/pages/page_'.$wiersz['page_id'].'_'.$wiersz['id'].'.php');

								// Make prefix to reconize pages
								$this->data['type']='pages';
								$this->data['path']='css';
								$this->data['file']='style.css';
								$this->data['source']=file_get_contents($this->data['root_path'].'/once/snippets/'.$row['id'].'/snippet.css');
								$this->data['id']=$wiersz['page_id'].'_'.$wiersz['id'];
								$response=$this->once_save_source();

								$this->data['type']='pages';
								$this->data['path']='js';
								$this->data['file']='script.js';
								$this->data['source']=file_get_contents($this->data['root_path'].'/once/snippets/'.$row['id'].'/snippet.js');
								$this->data['id']=$wiersz['page_id'].'_'.$wiersz['id'];
								$response=$this->once_save_source();
							}

							$obj['status']='ok';
						}else{
							$obj['errors'][]='Page snippet not already exists';
							$obj['error']++;
						}

						// Refresh code
						$this->gen_page($wiersz['page_id']);

						$obj['status']='ok';
					}else{
						$obj['errors'][]='Col not exist';
						$obj['error']++;
					}
				}else{
					$obj['errors'][]='Definitely something wrong';
					$obj['error']++;
				}
			}else{
				$obj['errors'][]='CSFR token invalid';
				$obj['error']++;
			}
		}else{
			$obj['errors'][]='You don\'t have permission';
			$obj['error']++;
		}

		// Return depends on type
		if($this->data['ajax']){
			// Print JSON object
			echo json_encode($obj);
		}else{
			// Return JSON object
			return $obj;
		}
	}

}
?>


text:
package net.digitalid.utility.validation.annotations.type;

import java.lang.annotation.Documented;
import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;

import javax.annotation.Nonnull;

import net.digitalid.utility.annotations.type.ThreadSafe;

/**
 * This annotation indicates that the objects of the annotated class are mutable.
 * It is not safe to share mutable objects between various instances and threads,
 * unless the annotated class is also {@link ThreadSafe thread-safe}.
 *
 * @see Immutable
 * @see ReadOnly
 * @see Stateless
 * @see Utility
 */
@Documented
@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
public @interface Mutable {

    /* -------------------------------------------------- Value -------------------------------------------------- */

    /**
     * Returns the type which provides read-only access to the annotated type (or the type of this annotation as a default because null is not allowed).
     */
    @Nonnull Class<?> value() default Mutable.class;

}


text:
using cdc::GuiWidget;

GuiWidget::GuiWidget()
{
}

void GuiWidget::setPosition(float x, float y)
{
	position.set(x, y);
}

void GuiWidget::setPosition(int x, int y)
{
	position.set(x, y);
}

const sf::Vector2f& GuiWidget::getPosition() const
{
	return position.get();
}