/*
 * Copyright (c) 2018-2020, NVIDIA CORPORATION.  All rights reserved.
 *
 * NVIDIA CORPORATION and its licensors retain all intellectual property
 * and proprietary rights in and to this software, related documentation
 * and any modifications thereto.  Any use, reproduction, disclosure or
 * distribution of this software and related documentation without an express
 * license agreement from NVIDIA CORPORATION is strictly prohibited.
 */

#include <iostream>
#include <chrono>
#include <pthread.h>
#include <unistd.h>

#include "CCompositeNvSci.hpp"
#include "CUtils.hpp"

#include "nvscisync.h"
#include "nvscistream.h"
#include "nvmedia_2d_nvscisync.h"
#include "nvmedia_image_nvscibuf.h"

using namespace std;

SIPLStatus CCompositeNvSci::Init(uint32_t uDispID,
                                 uint32_t uWinID,
                                 uint32_t uWinDepth,
                                 NvMediaRect *pRect,
                                 NvSciBufModule bufModule,
                                 NvSciSyncModule syncModule)
{
    // Check arguments
    if ((bufModule == nullptr) || (syncModule == nullptr)) {
        LOG_ERR("Compositor: Received unexpected nullptr\n");
        return NVSIPL_STATUS_BAD_ARGUMENT;
    }

    // Initialize state
    m_bRunning = false;

    // Initialize fence structures
    for (auto i = 0u; i < NUM_OF_INPUTS; i++) {
        for (auto j = 0u; j < NUM_PACKETS; j++) {
            m_streams[i].prefences[j] = NvSciSyncFenceInitializer;
            m_streams[i].postfences[j] = NvSciSyncFenceInitializer;
        }
    }

    // Create an NvMedia Device to be used to allocated images.
    m_pDevice.reset(NvMediaDeviceCreate());
    CHK_PTR_AND_RETURN(m_pDevice, "NvMediaDeviceCreate");

    // Create 2D for each input.
    for (auto row = 0U; row < NUM_OF_ROWS; row++) {
        m_p2DDevices[row].reset(NvMedia2DCreate(m_pDevice.get()));
        if (!m_p2DDevices[row]) {
            LOG_ERR("Compositor: NvMedia2DCreate failed for source:%u\n", row);
            return NVSIPL_STATUS_ERROR;
        }
    }

    // Initialize display
    auto status = InitDisplay(uDispID, uWinID, uWinDepth);
    CHK_STATUS_AND_RETURN(status, "CCompositeNvSci::InitDisplay");

    m_outputWidth = m_oDispInfo.width;
    m_outputHeight = m_oDispInfo.height;
    if (pRect != nullptr) {
        m_outputRect = *pRect;
        m_outputWidth = (pRect->x1 - pRect->x0);
        m_outputHeight = (pRect->y1 - pRect->y0);
        if (m_outputWidth == 0 || m_outputHeight == 0) {
            LOG_ERR("Compositor: Invalid output resolution specified\n");
            return NVSIPL_STATUS_BAD_ARGUMENT;
        }
        m_bRectSet = true;
    }

    // Create output buffer pool
    vector<NvMediaSurfAllocAttr> vAllocAttr;
    vAllocAttr.push_back( { NVM_SURF_ATTR_WIDTH, m_outputWidth });
    vAllocAttr.push_back( { NVM_SURF_ATTR_HEIGHT, m_outputHeight });
    vAllocAttr.push_back( { NVM_SURF_ATTR_CPU_ACCESS, NVM_SURF_ATTR_CPU_ACCESS_UNCACHED });
    vAllocAttr.push_back( { NVM_SURF_ATTR_ALLOC_TYPE, NVM_SURF_ATTR_ALLOC_ISOCHRONOUS });

    NVM_SURF_FMT_DEFINE_ATTR(compositeFormatAttrs);
    NVM_SURF_FMT_SET_ATTR_RGBA(compositeFormatAttrs, RGBA, UINT, 8, PL);
    auto surfType = NvMediaSurfaceFormatGetType(compositeFormatAttrs, NVM_SURF_FMT_ATTR_MAX);

    // Create output images and add them to the queue.
    for (auto i = 0u; i < NUM_OUT_BUFFERS; i++) {
        m_outputImages[i].reset(NvMediaImageCreateNew(m_pDevice.get(),
                                surfType,
                                vAllocAttr.data(),
                                vAllocAttr.size(),
                                0));
        CHK_PTR_AND_RETURN(m_outputImages[i], "NvMediaImageCreateNew");
        m_outputQueue.push(m_outputImages[i].get());
    }

    // Create a blank NvMediaImage to clear the frames during transitions
    m_pBlankFrame.reset(NvMediaImageCreateNew(m_pDevice.get(),
                                             surfType,
                                             vAllocAttr.data(),
                                             vAllocAttr.size(),
                                             0));
    CHK_PTR_AND_RETURN(m_pBlankFrame, "NvMediaImageCreateNew");

    // Memset the NvMediaImage (assumes the blank frame to be RGBA)
    uint8_t* buff = new (std::nothrow) uint8_t[m_outputWidth * m_outputHeight * 4];
    CHK_PTR_AND_RETURN(buff, "Compositor: buffer allocation");
    std::fill(buff, buff + m_outputWidth * m_outputHeight * 4, 0);
    uint8_t** buffers = &buff;
    auto buffPitches = static_cast<uint32_t>(m_outputWidth * 4);
    NvMediaImageSurfaceMap surfaceMap;
    NvMediaImageLock(m_pBlankFrame.get(), NVMEDIA_IMAGE_ACCESS_WRITE, &surfaceMap);
    auto nvmStatus = NvMediaImagePutBits(m_pBlankFrame.get(),
                                         nullptr,
                                         (void **)buffers,
                                         &buffPitches);
    if (nvmStatus != NVMEDIA_STATUS_OK) {
        LOG_ERR("Compositor: NvMediaImagePutBits failed\n");
    }
    NvMediaImageUnlock(m_pBlankFrame.get());
    delete [] buff;

    // Clear all output buffers
    ClearOutputBuffers();

    m_sciBufModule = bufModule;
    m_sciSyncModule = syncModule;

    return NVSIPL_STATUS_OK;
}

SIPLStatus CCompositeNvSci::RegisterSource(uint32_t groupIndex,   // group
                                           uint32_t modIndex,     // row
                                           uint32_t outIndex,     // col
                                           bool isRaw,
                                           uint32_t &id,
                                           NvSciStreamBlock *consumer,
                                           NvSciStreamBlock **upstream,
                                           NvSciStreamBlock *queue,
                                           QueueType type)
{
    NvSciError sciErr;
    ConsumerStream *pStream;

    if ((groupIndex >= NUM_OF_GROUPS) or (modIndex >= NUM_OF_ROWS) or (outIndex >= NUM_OF_COLS)) {
        LOG_ERR("Compositor: RegisterSource: Invalid argument\n");
        return NVSIPL_STATUS_BAD_ARGUMENT;
    }

    if ((consumer == nullptr) || (upstream == nullptr)) {
        LOG_ERR("Compositor: RegisterSource: Received unexpected nullptr\n");
        return NVSIPL_STATUS_BAD_ARGUMENT;
    }

    // Compute global ID for the source
    id = groupIndex * (NUM_OF_COLS * NUM_OF_ROWS) + modIndex * NUM_OF_COLS + outIndex;

    // Add source to the group
    m_GroupInfos[groupIndex].push_back(id);

    pStream = &m_streams[id];

    switch(type) {
    case QueueType_Mailbox:
        sciErr = NvSciStreamMailboxQueueCreate(&pStream->queue);
        CHK_NVSCISTATUS_AND_RETURN(sciErr, "NvSciStreamMailboxQueueCreate");
        break;
    default:
    case QueueType_Fifo:
        sciErr = NvSciStreamFifoQueueCreate(&pStream->queue);
        CHK_NVSCISTATUS_AND_RETURN(sciErr, "NvSciStreamFifoQueueCreate");
        break;
    }

    sciErr = NvSciStreamConsumerCreate(pStream->queue, &pStream->consumer);
    CHK_NVSCISTATUS_AND_RETURN(sciErr, "NvSciStreamConsumerCreate");

    *consumer = pStream->consumer;
    *upstream = &pStream->upstream;
    *queue = pStream->queue;

    pStream->id = id;
    pStream->row = modIndex;
    pStream->group = groupIndex;
    m_usedStreams.push_back(pStream);

    LOG_INFO("Compositor: Registered output:%u from link:%u of quad: %u as id:%u\n", outIndex, modIndex, groupIndex, id);
    return NVSIPL_STATUS_OK;
}

SIPLStatus CCompositeNvSci::DeleteImages(std::vector<NvMediaImage*>& images)
{
    auto bufferPoolSize = images.size();
    for (auto i = 0u; i < bufferPoolSize; i++) {
        if (images[i] == NULL) {
            LOG_ERR("Attempt to destroy null image\n");
            continue;
        }
        NvMediaImageDestroy(images[i]);
    }
    return NVSIPL_STATUS_OK;
}

SIPLStatus CCompositeNvSci::Deinit()
{
    // Clear all output buffers
    ClearOutputBuffers();

    for (ConsumerStream *pStream : m_usedStreams) {
        while (!pStream->vInputQueue.empty()) {
            auto index = pStream->vInputQueue.front();
            ReleasePacket(pStream, index);
            pStream->vInputQueue.pop();
        }
        if (pStream->consumerSyncObj != nullptr) {
            NvSciSyncObjFree(pStream->consumerSyncObj);
        }
        if (pStream->objFromProducer != nullptr) {
            NvSciSyncObjFree(pStream->objFromProducer);
        }
        for (NvSciBufObj bufObj : pStream->sciBufObjs) {
            NvSciBufObjFree(bufObj);
        }
        DeleteImages(pStream->imageList);

        std::vector<NvSciBufObj>().swap(pStream->sciBufObjs);
        std::vector<NvMediaImage*>().swap(pStream->imageList);
    }

    vector<ConsumerStream*>().swap(m_usedStreams);

    return NVSIPL_STATUS_OK;
}

bool CCompositeNvSci::CheckInputQueues(void)
{
    bool anyReady = false;
    const auto& vInput = m_GroupInfos[m_uGroupIndex];
    for (const auto& id : vInput) {
        if (!m_streams[id].vInputQueue.empty()) {
            anyReady = true;
            break;
        }
    }
    return anyReady;
}

void CCompositeNvSci::ThreadFunc(void)
{
    pthread_setname_np(pthread_self(), "CCompositeNvSci");

    auto err = SetupStreaming();
    if(err != NVSIPL_STATUS_OK) {
       LOG_ERR("Compositor: failed to set up NvSci streaming\n");
       return;
    }

    m_bRunning = true;

    uint32_t sleepTimeMs = uint32_t(1000 / (m_oDispInfo.refreshRate));

    while (m_bRunning) {
        // Check if there is a pending group change.
        if (m_uNewGroupIndex != m_uGroupIndex) {
            // Clear all output buffers
            ClearOutputBuffers();

            // Release buffers from input queues for current group
            const auto &vInput = m_GroupInfos[m_uGroupIndex];
            for (const auto& id : vInput) {
                ConsumerStream *pStream = &m_streams[id];
                while(!pStream->vInputQueue.empty()) {
                    auto index = pStream->vInputQueue.front();
                    ReleasePacket(pStream, index);
                    pStream->vInputQueue.pop();
                }
            }
            // Update current group index
            m_uGroupIndex = m_uNewGroupIndex;
        }

        auto status = CollectEvents();
        if(status != NVSIPL_STATUS_OK) {
           LOG_ERR("Compositor: CollectEvents failed\n");
           return;
        }

        // Check for input readiness
        bool anyReady = CheckInputQueues();
        if (!anyReady) {
            LOG_INFO("Compositor does not have any inputs available yet\n");

            // Sleep for refresh rate
            std::this_thread::sleep_for(std::chrono::milliseconds(sleepTimeMs));
            continue;
        }

        // Get output image
        if (m_outputQueue.empty()) {
            LOG_ERR("Compositor output queue empty\n");
            return;
        }

        NvMediaImage* pDstImg = m_outputQueue.front();
        m_outputQueue.pop();

        // Get full buffer from input queue and composite on to output image
        const auto& vInput = m_GroupInfos[m_uGroupIndex];
        for (const auto&id : vInput) {
            ConsumerStream *pStream = &m_streams[id];

            if (pStream->vInputQueue.empty()) {
                LOG_INFO("Compositor: Input queue for source %u is empty\n", id);
                continue;
            }
            auto index = pStream->vInputQueue.front();
            pStream->vInputQueue.pop();

            auto pSrcImg = pStream->imageList[index];
            NvMediaRect& dstRect = m_oInputRects[id];

            auto err = PreSync(pStream, index);
            if(err != NVSIPL_STATUS_OK) {
                 ReleasePacket(pStream, index);
                 return;
            }

            // Convert RAW to RGBA if necessary
            bool isRaw = false;
            auto utilStatus = CUtils::IsRawImage(pSrcImg, isRaw);
            if (utilStatus != NVSIPL_STATUS_OK) {
                LOG_ERR("Compositor: CUtils::IsRawImage failed for source %u\n", id);
                ReleasePacket(pStream, index);
                return;
            }
            if (isRaw) {
                auto status = CUtils::ConvertRawToRgba(pSrcImg,
                                                       m_rawImageBufs[id].get(),
                                                       m_scratchRGBAImages[id].get(),
                                                       m_rgbaImageBufs[id].get());
                if (status != NVSIPL_STATUS_OK) {
                    LOG_ERR("Compositor: CUtils::IsRawConvertRawToRgbaImage failed for source %u\n", id);
                    ReleasePacket(pStream, index);
                    return;
                }
                pSrcImg = m_scratchRGBAImages[id].get();
            }

            // Blit
            auto nvmStatus = NvMedia2DBlitEx(m_p2DDevices[pStream->row].get(),
                                             pDstImg,
                                             &dstRect,
                                             pSrcImg,
                                             NULL,
                                             NULL,
                                             NULL);
            if (nvmStatus != NVMEDIA_STATUS_OK) {
                LOG_ERR("Compositor: NvMedia2DBlitEx failed for source %u\n", id);
                ReleasePacket(pStream, index);
                return;
            }

            err = PostSync(pStream, index);
            if(err != NVSIPL_STATUS_OK) {
               ReleasePacket(pStream, index);
               return;
            }

            // Release the input buffer
            ReleasePacket(pStream, index);
        }

        // Send composited output image to display
        NvMediaImage *releaseList[2] = { NULL }; // IDPFlip needs at least two elements
        NvMediaRect* pRect = nullptr;
        if (m_bRectSet) {
            pRect = &m_outputRect;
        }
        auto nvmStatus = NvMediaIDPFlip(m_pNvmIDP.get(), pDstImg, NULL, pRect, releaseList, NULL);
        if (nvmStatus != NVMEDIA_STATUS_OK) {
            LOG_ERR("Compositor: NvMediaIDPFlip failed!. nvmStatus: %u\n", nvmStatus);
            return;
        }
        auto pPrevImage = releaseList[0];
        if (pPrevImage != NULL) {
            // Put the previous image back on the queue.
            m_outputQueue.push(pPrevImage);
        }
    } // while ()

    err = CloseStreaming();
    if (err != NVSIPL_STATUS_OK) {
        LOG_ERR("Compositor: failed to close NvSci streaming\n");
        return;
    }

    return;
}

SIPLStatus CCompositeNvSci::CollectEvents(void)
{
    NvSciStreamEvent event;
    NvSciStreamCookie cookie;
    NvSciSyncFence fence = NvSciSyncFenceInitializer;

    for(ConsumerStream *pStream : m_usedStreams) {
        auto sciErr = NvSciStreamBlockEventQuery(pStream->consumer, 0, &event);
        if(sciErr == NvSciError_Timeout) {
            continue;   // nothing ready on this stream
        }
        CHK_NVSCISTATUS_AND_RETURN(sciErr, "Consumer PacketReady NvSciStreamBlockEventQuery");
        if(event.type != NvSciStreamEventType_PacketReady) {
             LOG_ERR("Compositor: did not receive expected PacketReady event\n");
             return NVSIPL_STATUS_ERROR;
        }

        sciErr = NvSciStreamConsumerPacketAcquire(pStream->consumer, &cookie, &fence);
        CHK_NVSCISTATUS_AND_RETURN(sciErr, "NvSciStreamConsumerPacketAcquire");
        // Index is one less than cookie, it was incremented on initialization since zero is invalid
        auto index = cookie - 1;

        if (pStream->group == m_uGroupIndex) {
            sciErr = NvSciSyncFenceDup(&fence, &pStream->prefences[index]);
            CHK_NVSCISTATUS_AND_RETURN(sciErr, "NvSciSyncFenceDup");
            pStream->vInputQueue.push(index);
        } else {
            pStream->postfences[index] = NvSciSyncFenceInitializer;
            ReleasePacket(pStream, index);
        }
        NvSciSyncFenceClear(&fence);
    }

    return NVSIPL_STATUS_OK;
}

SIPLStatus CCompositeNvSci::PreSync(ConsumerStream *pStream, uint32_t index)
{
    auto nvmStatus = NvMedia2DInsertPreNvSciSyncFence(m_p2DDevices[pStream->row].get(),
                                                      &pStream->prefences[index]);
    CHK_NVMSTATUS_AND_RETURN(nvmStatus, "Consumer NvMedia2DInsertPreNvSciSyncFence");

    NvSciSyncFenceClear(&pStream->prefences[index]);

    nvmStatus = NvMedia2DSetNvSciSyncObjforEOF(m_p2DDevices[pStream->row].get(), pStream->consumerSyncObj);
    CHK_NVMSTATUS_AND_RETURN(nvmStatus, "Consumer NvMedia2DSetNvSciSyncObjforEOF");

    return NVSIPL_STATUS_OK;
}

SIPLStatus CCompositeNvSci::PostSync(ConsumerStream *pStream, uint32_t index)
{
    auto nvmStatus = NvMedia2DGetEOFNvSciSyncFence(m_p2DDevices[pStream->row].get(),
                                                   pStream->consumerSyncObj,
                                                   &pStream->postfences[index]);
    CHK_NVMSTATUS_AND_RETURN(nvmStatus, "Consumer NvMedia2DGetEOFNvSciSyncFence");

    return NVSIPL_STATUS_OK;
}

SIPLStatus CCompositeNvSci::ReleasePacket(ConsumerStream *pStream, uint32_t index)
{
    auto sciErr = NvSciStreamConsumerPacketRelease(pStream->consumer,
                                                   pStream->packets[index],
                                                   &pStream->postfences[index]);
    CHK_NVSCISTATUS_AND_RETURN(sciErr, "NvSciStreamConsumerPacketRelease");

    NvSciSyncFenceClear(&pStream->postfences[index]);

    return NVSIPL_STATUS_OK;
}


SIPLStatus CCompositeNvSci::SetupStreaming(void)
{
    NvSciSyncAttrList signalerAttrList, waiterAttrList;
    NvSciBufAttrList bufAttrList;
    NvSciStreamEvent event;

    // Get waiter and signaler NvSciSync attributes from 2D
    auto sciErr = NvSciSyncAttrListCreate(m_sciSyncModule, &signalerAttrList);
    CHK_NVSCISTATUS_AND_RETURN(sciErr, "Consumer signaler NvSciSyncAttrListCreate");

    sciErr = NvSciSyncAttrListCreate(m_sciSyncModule, &waiterAttrList);
    CHK_NVSCISTATUS_AND_RETURN(sciErr, "Consumer waiter NvSciSyncAttrListCreate");

    auto nvmStatus = NvMedia2DFillNvSciSyncAttrList(m_p2DDevices[0].get(), signalerAttrList, NVMEDIA_SIGNALER);
    CHK_NVMSTATUS_AND_RETURN(nvmStatus, "Consumer signaler NvMedia2DFillNvSciSyncAttrList");

    nvmStatus = NvMedia2DFillNvSciSyncAttrList(m_p2DDevices[0].get(), waiterAttrList, NVMEDIA_WAITER);
    CHK_NVMSTATUS_AND_RETURN(nvmStatus, "Consumer waiter NvMedia2DFillNvSciSyncAttrList");

    // Send the waiter attributes to the producer
    for(ConsumerStream *pStream : m_usedStreams) {
        sciErr = NvSciStreamBlockSyncRequirements(pStream->consumer, false, waiterAttrList);
        CHK_NVSCISTATUS_AND_RETURN(sciErr, "Consumer NvSciStreamBlockSyncRequirements");
    }

    // Get SciSync attributes from producer, reconcile with our own signaler attributes,
    // create consumer sync objs with the reconciled attributes, register objs with 2D and
    // send to producer.
    for(ConsumerStream *pStream : m_usedStreams) {
       NvSciSyncAttrList unreconciledList[2];
       NvSciSyncAttrList reconciledList = NULL;
       NvSciSyncAttrList newConflictList = NULL;

       auto sciErr = NvSciStreamBlockEventQuery(pStream->consumer, STREAMING_TIMEOUT, &event);
       CHK_NVSCISTATUS_AND_RETURN(sciErr, "Consumer SyncAttr NvSciStreamBlockEventQuery");

       if(event.type != NvSciStreamEventType_SyncAttr) {
           LOG_ERR("Compositor: did not receive expected event type\n");
           return NVSIPL_STATUS_ERROR;
       }

       unreconciledList[0] = signalerAttrList;
       unreconciledList[1] = event.syncAttrList;

       sciErr = NvSciSyncAttrListReconcile(unreconciledList, 2, &reconciledList, &newConflictList);
       CHK_NVSCISTATUS_AND_RETURN(sciErr, "Consumer NvSciSyncAttrListReconcile");

       sciErr = NvSciSyncObjAlloc(reconciledList, &pStream->consumerSyncObj);
       CHK_NVSCISTATUS_AND_RETURN(sciErr, "Consumer NvSciSyncObjAlloc");

       NvSciSyncAttrListFree(reconciledList);
       NvSciSyncAttrListFree(event.syncAttrList);
       if(newConflictList != nullptr) {
          NvSciSyncAttrListFree(newConflictList);
       }

       auto nvmStatus = NvMedia2DRegisterNvSciSyncObj(m_p2DDevices[pStream->row].get(),
                                                      NVMEDIA_EOFSYNCOBJ,
                                                      pStream->consumerSyncObj);
       CHK_NVMSTATUS_AND_RETURN(nvmStatus, "NvMedia2DRegisterNvSciSyncObj for EOF");

       sciErr = NvSciStreamBlockSyncObjCount(pStream->consumer, 1);
       CHK_NVSCISTATUS_AND_RETURN(sciErr, "Consumer NvSciStreamBlockSyncObjCount");

       sciErr = NvSciStreamBlockSyncObject(pStream->consumer, 0, pStream->consumerSyncObj);
       CHK_NVSCISTATUS_AND_RETURN(sciErr, "Consumer NvSciStreamBlockSyncObject");
    }

    NvSciSyncAttrListFree(waiterAttrList);
    NvSciSyncAttrListFree(signalerAttrList);

    // Create SciBuf attributes.
    sciErr = NvSciBufAttrListCreate(m_sciBufModule, &bufAttrList);
    CHK_NVSCISTATUS_AND_RETURN(sciErr, "Consumer NvSciBufAttrListCreate");

    NvSciBufType bufType = NvSciBufType_Image;
    bool cpuAccess = true;
    NvSciBufAttrValAccessPerm accessPerm = NvSciBufAccessPerm_Readonly;
    NvSciBufAttrKeyValuePair attrKvp[] = {
                                            { NvSciBufGeneralAttrKey_Types,
                                              &bufType,
                                              sizeof(bufType)
                                            },
                                            { NvSciBufGeneralAttrKey_NeedCpuAccess,
                                              &cpuAccess,
                                              sizeof(cpuAccess)
                                            },
                                            { NvSciBufGeneralAttrKey_RequiredPerm,
                                              &accessPerm,
                                              sizeof(accessPerm)
                                            }
                                         };

    sciErr = NvSciBufAttrListSetAttrs(bufAttrList, attrKvp, 3);
    CHK_NVSCISTATUS_AND_RETURN(sciErr, "Consumer NvSciBufAttrListSetAttrs");

    // Get sync objs from producer and register with 2D.
    // Send SciBuf attributes to pool
    for(ConsumerStream *pStream : m_usedStreams) {
       auto sciErr = NvSciStreamBlockEventQuery(pStream->consumer, STREAMING_TIMEOUT, &event);
       CHK_NVSCISTATUS_AND_RETURN(sciErr, "Consumer SyncCount NvSciStreamBlockEventQuery");
       if(event.type != NvSciStreamEventType_SyncCount) {
           LOG_ERR("Compositor: did not receive expected SyncCount event type\n");
           return NVSIPL_STATUS_ERROR;
       }
       if(event.count != 1U) {
           LOG_ERR("Compositor: did not receive a single sync object as expected\n");
           return NVSIPL_STATUS_ERROR;
       }

       sciErr = NvSciStreamBlockEventQuery(pStream->consumer, STREAMING_TIMEOUT, &event);
       CHK_NVSCISTATUS_AND_RETURN(sciErr, "Consumer SyncDesc NvSciStreamBlockEventQuery");
       if(event.type != NvSciStreamEventType_SyncDesc) {
           LOG_ERR("Compositor: did not receive expected SyncDesc event type\n");
           return NVSIPL_STATUS_ERROR;
       }
       pStream->objFromProducer = event.syncObj;

       auto nvmStatus = NvMedia2DRegisterNvSciSyncObj(m_p2DDevices[pStream->row].get(),
                                                      NVMEDIA_PRESYNCOBJ,
                                                      pStream->objFromProducer);
       CHK_NVMSTATUS_AND_RETURN(nvmStatus, "Consumer NvMedia2DRegisterNvSciSyncObj");

       sciErr = NvSciStreamBlockPacketElementCount(pStream->consumer, 1);
       CHK_NVSCISTATUS_AND_RETURN(sciErr, "Consumer NvSciStreamBlockPacketElementCount");

       sciErr = NvSciStreamBlockPacketAttr(pStream->consumer, 0, 0,
                                           NvSciStreamElementMode_Asynchronous,
                                           bufAttrList);
       CHK_NVSCISTATUS_AND_RETURN(sciErr, "Consumer NvSciStreamBlockPacketAttr");
    }

    // Get element count and packet attributes from pool (discard because we don't allocate buffers).
    // Get SciBufObjs from pool, acknowledge.
    // Create Images from SciBufObjs.
    for(ConsumerStream *pStream : m_usedStreams) {
        auto sciErr = NvSciStreamBlockEventQuery(pStream->consumer, STREAMING_TIMEOUT, &event);
        CHK_NVSCISTATUS_AND_RETURN(sciErr, "Consumer PacketElementCount NvSciStreamBlockEventQuery");
        if(event.type != NvSciStreamEventType_PacketElementCount) {
            LOG_ERR("Compositor: did not receive PacketElementCount event as expected\n");
            return NVSIPL_STATUS_ERROR;
        }
        if(event.count != 1U) {
            LOG_ERR("Compositor: did not receive PacketElementCount==1 expected");
            return NVSIPL_STATUS_ERROR;
        }

        sciErr = NvSciStreamBlockEventQuery(pStream->consumer, STREAMING_TIMEOUT, &event);
        CHK_NVSCISTATUS_AND_RETURN(sciErr, "Consumer PacketAttr NvSciStreamBlockEventQuery");
        if(event.type != NvSciStreamEventType_PacketAttr) {
            LOG_ERR("Compositor: did not receive PacketAttr event as expected\n");
            return NVSIPL_STATUS_ERROR;
        }
        NvSciBufAttrListFree(event.bufAttrList);

        pStream->sciBufObjs.resize(NUM_PACKETS);
        pStream->imageList.resize(NUM_PACKETS);

        // This delay is needed to solve initialization failures
        // TODO: Implement a fix that doesn't rely on specific timing
        usleep(90000);

        for(auto i = 0u; i < NUM_PACKETS; i++) {
            auto sciErr = NvSciStreamBlockEventQuery(pStream->consumer, STREAMING_TIMEOUT, &event);
            CHK_NVSCISTATUS_AND_RETURN(sciErr, "Consumer PacketCreate NvSciStreamBlockEventQuery");
            if(event.type != NvSciStreamEventType_PacketCreate) {
                LOG_ERR("Compositor: did not receive expected PacketCreate event type\n");
                return NVSIPL_STATUS_ERROR;
            }

            pStream->packets[i] = event.packetHandle;

            NvSciStreamCookie cookie = i+1;

            sciErr = NvSciStreamBlockPacketAccept(pStream->consumer,
                                                  pStream->packets[i],
                                                  cookie, NvSciError_Success);
            CHK_NVSCISTATUS_AND_RETURN(sciErr, "Consumer NvSciStreamBlockPacketAccept");

            sciErr = NvSciStreamBlockEventQuery(pStream->consumer, STREAMING_TIMEOUT, &event);
            CHK_NVSCISTATUS_AND_RETURN(sciErr, "Consumer PacketElement NvSciStreamBlockEventQuery");
            if(event.type != NvSciStreamEventType_PacketElement) {
                LOG_ERR("Compositor: did not receive expected PacketElement event type\n");
                return NVSIPL_STATUS_ERROR;
            }

            sciErr = NvSciBufObjDup(event.bufObj, &pStream->sciBufObjs[i]);
            CHK_NVSCISTATUS_AND_RETURN(sciErr, "Consumer NvSciBufObjDup");
            NvSciBufObjFree(event.bufObj);

            sciErr = NvSciStreamBlockElementAccept(pStream->consumer,
                                                   pStream->packets[i],
                                                   event.index,
                                                   NvSciError_Success);
            CHK_NVSCISTATUS_AND_RETURN(sciErr, "Consumer NvSciStreamBlockElementAccept");

            auto nvmStatus = NvMediaImageCreateFromNvSciBuf(m_pDevice.get(),
                                                            pStream->sciBufObjs[i],
                                                            &pStream->imageList[i]);
            CHK_NVMSTATUS_AND_RETURN(nvmStatus, "Consumer NvMediaImageCreateFromNvSciBuf");

            bool isRaw = false;
            auto status = CUtils::IsRawImage(pStream->imageList[i], isRaw);
            CHK_STATUS_AND_RETURN(status, "CUtils::IsRawImage");
            if (isRaw) {
                if (m_scratchRGBAImages[pStream->id] == nullptr) {
                    // Allocate scratch RGBA image for raw to RGBA conversion
                    m_scratchRGBAImages[pStream->id].reset(CUtils::CreateRgbaImage(m_pDevice.get(),
                                                                                   pStream->imageList[i]));
                    CHK_PTR_AND_RETURN(m_scratchRGBAImages[pStream->id], "CUtils::CreateRgbaImage");
                }
                if (m_rawImageBufs[pStream->id] == nullptr) {
                    // Allocate CPU buffer for raw image
                    m_rawImageBufs[pStream->id].reset(CUtils::CreateRawBuffer(pStream->imageList[i]));
                    CHK_PTR_AND_RETURN(m_rawImageBufs[pStream->id], "CUtils::CreateRawBuffer");
                }
                if (m_rgbaImageBufs[pStream->id] == nullptr) {
                    // Allocate CPU buffer for RGBA image
                    m_rgbaImageBufs[pStream->id].reset(CUtils::CreateRgbaBuffer(m_scratchRGBAImages[pStream->id].get()));
                    CHK_PTR_AND_RETURN(m_rgbaImageBufs[pStream->id], "CUtils::CreateRgbaBuffer");
                }
            }
        }
    }

    NvSciBufAttrListFree(bufAttrList);

    return NVSIPL_STATUS_OK;
}

SIPLStatus CCompositeNvSci::CloseStreaming(void)
{
    for (ConsumerStream *pStream : m_usedStreams) {
        auto nvmStatus = NvMedia2DUnregisterNvSciSyncObj(m_p2DDevices[pStream->row].get(),
                                                         pStream->consumerSyncObj);
        CHK_NVMSTATUS_AND_RETURN(nvmStatus, "Consumer NvMedia2DUnregisterNvSciSyncObj");
        nvmStatus = NvMedia2DUnregisterNvSciSyncObj(m_p2DDevices[pStream->row].get(),
                                                    pStream->objFromProducer);
        CHK_NVMSTATUS_AND_RETURN(nvmStatus, "Consumer NvMedia2DUnregisterNvSciSyncObj");
    }

    return NVSIPL_STATUS_OK;
}