// param points to the DrThread, already IncRef'd
DWORD WINAPI DrThread::ThreadEntryStatic(void * param)
{
    DrThread *pThread = (DrThread *)param;

    // We save a pointer to this thread object in thread-local storage. That way,
    // You can always find your current thread object with DrGetCurrentThread().
    //
    LogAssert(t_pThread.IsNull());
    t_pThread = pThread;  // save pointer to ourselves in thread local storage

    DrLogI( "Dryad thread starting. %s, pThread=%p", pThread->m_strDescription.GetString(), pThread);

    DWORD ret = pThread->ThreadEntry();

    DrLogI( "Dryad thread exiting. %s, pThread=%p, exitcode=%08x", pThread->m_strDescription.GetString(), pThread, ret);

    LogAssert(t_pThread == pThread);

    // no longer under our control...
    t_pThread = NULL;

    pThread->DecRef(); // The DrThread will be freed when noone is interested in it anymoe

    return ret;
}

StrTab::StrTab(std::istream &file, off_t start_offset, off_t end_offset, bool makeUnique)
{
	LogAssert(end_offset > start_offset);
	file.seekg(start_offset, file.beg);
	LogAssert(file);
	char c = '\0';
	std::map<std::string, int> unique_id;
	while (file && file.tellg() <= end_offset)
	{
		std::string s;
		file.get(c);
		while (c && file && file.tellg() <= end_offset)
		{
			s += c;
			file.get(c);
		}
		if (makeUnique)
		{
			if (unique_id.find(s) != unique_id.end())
			{
				std::stringstream ss;
				ss << s << " " << ++unique_id[s];
				s = ss.str();
				LogWarning("Munged string to make unique: \"%s\"", s.c_str());
			}
			else
				unique_id.emplace(s, 0);
		}
		readStrings.push_back(s);
	}
	if (c)
		LogError("Table didn't end with NULL");
}

void FontSymbol::RenderToBuffer(unsigned char *buffer, unsigned int textureWidth, unsigned int textureHeight)
{
    if (IsEmpty())
    {
        return;
    }

    InitTextureCoords(textureWidth, textureHeight);

    FT_Glyph_To_Bitmap(&this->ftGlyph, FT_RENDER_MODE_NORMAL, NULL, true);
    FT_BitmapGlyph bitmapGlyph = reinterpret_cast<FT_BitmapGlyph>(this->ftGlyph);

    LogAssert(bitmapGlyph->bitmap.width == this->width && "Dimensions don't match.");
    LogAssert(bitmapGlyph->bitmap.rows == this->height && "Dimensions don't match.");

    for (int y = 0; y < bitmapGlyph->bitmap.rows; ++y)
    {
        for (int x = 0; x < bitmapGlyph->bitmap.width; ++x)
        {
            int index = ((this->y + y) * textureWidth) + (this->x + x);
            LogAssert(index >= 0 && "Invalid index.");
            buffer[index] = bitmapGlyph->bitmap.buffer[y * bitmapGlyph->bitmap.width + x];
        }
    }
}

void RChannelFifoNBWriter::AcceptReturningUnit(RChannelFifoUnit* unit)
{
    bool wakeUpSupplierDrain = false;

    {
        AutoCriticalSection acs(&m_baseDR);

        LogAssert(m_outstandingUnits > 0);
        --m_outstandingUnits;

        if (m_supplierState == SS_Draining)
        {
            /* the supplier has requested a drain */
            if (m_outstandingUnits == 0)
            {
                m_supplierState = SS_Stopped;
                ++m_supplierEpoch;
                wakeUpSupplierDrain = true;
            }
        }
    }

    if (wakeUpSupplierDrain)
    {
        BOOL bRet = ::SetEvent(m_supplierDrainEvent);
        LogAssert(bRet != 0);
    }

    delete unit;
}

//
// Notify GM of completed vertex
//
void DVertexPnController::Terminate(DrError vertexState,
                                    UInt32 exitCode)
{
    LogAssert(vertexState != DryadError_VertexRunning);

    if (vertexState == DrError_OK)
    {
        LogAssert(exitCode == DrExitCode_StillActive);
    }

    //
    // take Critical section to update vertex state
    //
    {
        AutoCriticalSection acs(&m_baseCS);

        m_currentStatus->SetVertexState(vertexState);
    }

    //
    // Log vertex termination
    //
    DrLogI( "Terminating vertex. Vertex %u.%u exitCode %s vertexState %s",
        m_currentStatus->GetProcessStatus()->GetVertexId(),
        m_currentStatus->GetProcessStatus()->GetVertexInstanceVersion(),
        DREXITCODESTRING(exitCode), DRERRORSTRING(vertexState));

    //
    // Send status to GM
    //
    SendStatus(exitCode, true);
}

void DryadSubGraphVertex::MakeInputParser(UInt32 whichInput,
                                          RChannelItemParserRef* pParser)
{
    if (m_virtualInput == s_invalidVertex)
    {
        ReportError(DryadError_VertexInitialization,
                    "MakeInputParser called before initialization");
        return;
    }

    if (whichInput >= m_numberOfInputEdges)
    {
        ReportError(DryadError_VertexInitialization,
                    "MakeInputParser called with illegal port");
        return;
    }

    EdgeInfo* edge = m_inputEdge[whichInput];
    LogAssert(edge->GetSourceVertex() == m_virtualInput);
    LogAssert(edge->GetSourcePort() == whichInput);

    UInt32 inputVertex = edge->GetDestinationVertex();
    LogAssert(inputVertex < m_numberOfVertices);

    VertexInfo* input = &(m_vertex[inputVertex]);
    UInt32 inputPort = edge->GetDestinationPort();
    LogAssert(inputPort < input->GetInputPortCount());

    DryadVertexProgramBase* subProgram = input->GetVertexProgram();

    subProgram->MakeInputParser(inputPort, pParser);

    ReportError(subProgram->GetErrorCode(),
                subProgram->GetErrorMetaData());
}

void RChannelFifoWriterBase::SyncHandler::UseEvent(DryadHandleListEntry* event)
{
    LONG postIncrement = ::InterlockedIncrement(&m_usingEvent);
    LogAssert(postIncrement == 1);
    LogAssert(m_event == NULL);
    m_event = event;
}

 DryadConfigurationManager(
     const char* netLibName,
     const char* iniFileName,
     int argc,
     char* argv[])
 {
     m_strNetLibName = netLibName;
     m_strIniFileName = iniFileName;
     m_argc = argc;
     m_nOpts = 0;
     if (argc == 0) {
         m_argv = NULL;
     } else {
         LogAssert(argv != NULL);
         m_argv = new char *[argc];
         LogAssert(m_argv != NULL);
         for (int i = 0; i < argc; i++) {
             if (argv[i] == NULL) {
                 m_argv[i] = NULL;
             } else {
                 size_t len = strlen(argv[i])+1;
                 m_argv[i] = new char[len];
                 LogAssert(m_argv[i] != NULL);
                 memcpy(m_argv[i], argv[i], len);
             }
         }
     }
 }

/* called with RChannelReaderImpl::m_baseDR held */
void RChannelReaderImpl::
    AddHandlerToQueue(const char* caller,
                      RChannelProcessRequest* request,
                      ChannelProcessRequestList* requestList,
                      ChannelUnitList* returnUnitList)
{
    LogAssert(m_state == RS_Running);

    if (m_unitList.IsEmpty())
    {
//         DrLogD(
//             "queueing handler",
//             "caller: %s", caller);
        m_handlerList.InsertAsTail(m_handlerList.CastIn(request));
    }
    else
    {
//         DrLogD(
//             "adding process work request",
//             "caller: %s", caller);

        LogAssert(m_handlerList.IsEmpty());

        /* put this in the handler list since that's where
           TransferWaitingItems expects to find it */
        m_handlerList.InsertAsTail(m_handlerList.CastIn(request));

        TransferWaitingItems(caller,
                             requestList,
                             returnUnitList);
    }
}

void RChannelReaderImpl::
    ProcessItemArrayRequest(RChannelProcessRequest* request)
{
    RChannelItemArray* itemArray = request->GetItemArray();
    RChannelItemArrayReaderHandler* handler = request->GetHandler();
    LogAssert(handler != NULL);
    LogAssert(itemArray != NULL);

    void* cancelCookie = request->GetCookie();

    handler->ProcessItemArray(itemArray);

    {
        AutoCriticalSection acs(&m_baseDR);

        LogAssert(m_state != RS_Stopped);

        /* now that the request has been processed remove it from our
           accounting */
        RemoveFromCancelMap(request, cancelCookie);
        /* if it was the last request with this cookie and somebody
           was blocked cancelling the cookie, wake them up */
        MaybeTriggerCancelEvent(cancelCookie);

        /* if it was the last request in the system and we are
           draining, wake up the drain thread */
        if (m_state == RS_Stopping && m_cookieMap.empty())
        {
            LogAssert(m_eventMap.empty());
            BOOL bRet = ::SetEvent(m_drainEvent);
            LogAssert(bRet != 0);
        }
    }
}

bool RChannelReader::FetchNextItem(RChannelItemRef* pOutItem,
                                   DrTimeInterval timeOut)
{
    RChannelItemArrayRef itemArray;
    bool delivered = FetchNextItemArray(1, &itemArray, timeOut);
    LogAssert(itemArray != NULL);
    if (delivered)
    {
        if (itemArray->GetNumberOfItems() == 1)
        {
            RChannelItemRef* a = itemArray->GetItemArray();
            pOutItem->TransferFrom(a[0]);
        }
        else
        {
            LogAssert(itemArray->GetNumberOfItems() == 0);
            *pOutItem = NULL;
        }
    }
    else
    {
        LogAssert(itemArray->GetNumberOfItems() == 0);
        *pOutItem = NULL;
    }

    return delivered;
}

void DryadSubGraphVertex::EdgeInfo::
    MakeFifo(UInt32 fifoLength, WorkQueue* workQueue)
{
    LogAssert(m_reader == NULL);
    LogAssert(m_writer == NULL);

    UInt32 uniquifier = RChannelFactory::GetUniqueFifoId();

    DrStr64 fifoName;
    fifoName.SetF("fifo://%u/internal-%u-%u.%u--%u.%u",
                  fifoLength, uniquifier,
                  m_sourceVertex, m_sourcePort,
                  m_destinationVertex, m_destinationPort);

    DVErrorReporter errorReporter;
    RChannelFactory::OpenReader(fifoName, NULL, NULL, 1, NULL, 0, 0, workQueue,
                                &errorReporter, &m_reader, NULL);
    LogAssert(errorReporter.NoError());
    RChannelFactory::OpenWriter(fifoName, NULL, NULL, 1, NULL, 0, NULL,
                                &errorReporter, &m_writer);
    LogAssert(errorReporter.NoError());

    m_reader->GetReader()->Start(NULL);
    m_writer->GetWriter()->Start();
}

void DryadSubGraphVertex::VertexCompleted(DrError status,
                                          DryadMetaData* errorData)
{
    bool finished = false;

    {
        AutoCriticalSection acs(&m_baseCS);

        LogAssert(GetErrorCode() != DryadError_VertexCompleted);
        if (status == DryadError_VertexCompleted)
        {
            status = DrError_OK;
        }

        if (NoError())
        {
            ReportError(status, errorData);
        }

        LogAssert(m_outstandingVertices > 0);
        --m_outstandingVertices;

        if (m_outstandingVertices == 0)
        {
            finished = true;
        }
    }

    if (finished)
    {
        /* the real status will be returned in AsyncPostCompletion
           after we have exited all handlers. */
        m_handler->ProgramCompleted();
    }
}

void RChannelFifoNBWriter::
    StartSupplier(RChannelBufferPrefetchInfo* prefetchCookie)
{
    ChannelUnitList unblockedList;

    {
        AutoCriticalSection acs(&m_baseDR);

        LogAssert(m_supplierState == SS_Stopped);

        LogAssert(m_writerState != WS_Closed);
        LogAssert(m_outstandingUnits == 0);

        BOOL bRet = ::ResetEvent(m_supplierDrainEvent);
        LogAssert(bRet != 0);

        m_supplierState = SS_Running;

        while (m_blockedList.IsEmpty() == false)
        {
            ++m_outstandingUnits;

            RChannelFifoUnit* blockedUnit =
                m_blockedList.CastOut(m_blockedList.GetHead());

            unblockedList.TransitionToTail(unblockedList.
                                           CastIn(blockedUnit));
        }

        m_sendLatch.AcceptList(&unblockedList);
    }

    SendUnits(&unblockedList);
}

/* called with RChannelReaderImpl::m_baseDR held */
void RChannelReaderImpl::AddUnitToQueue(const char* caller,
                                        RChannelUnit* unit,
                                        ChannelProcessRequestList* requestList,
                                        ChannelUnitList* returnUnitList)
{
    LogAssert(m_state == RS_Running || m_state == RS_InterruptingSupplier);

    if (unit->GetType() == RChannelUnit_Item)
    {
        /* make sure we haven't already sent out a termination item for
           processing */
        LogAssert(m_writerTerminationItem == NULL &&
                  m_readerTerminationItem == NULL);

        if (m_handlerList.IsEmpty())
        {
//             DrLogD(
//                 "queueing item list",
//                 "caller: %s", caller);

            m_unitList.InsertAsTail(m_unitList.CastIn(unit));
            unit = NULL;
        }
        else
        {
//             DrLogD(
//                 "adding process work request",
//                 "caller: %s", caller);

            LogAssert(m_unitList.IsEmpty());

            /* put this in the unit list since that's where
               TransferWaitingItems expects to find it */
            m_unitList.InsertAsTail(m_unitList.CastIn(unit));

            TransferWaitingItems(caller,
                                 requestList,
                                 returnUnitList);
        }
    }
    else
    {
        LogAssert(unit->GetType() == RChannelUnit_BufferBoundary);
        if (m_unitList.IsEmpty())
        {
            returnUnitList->InsertAsTail(returnUnitList->CastIn(unit));
//             DrLogD(
//                 "adding buffer boundary dispatch request",
//                 "caller: %s", caller);
        }
        else
        {
            m_unitList.InsertAsTail(m_unitList.CastIn(unit));
//             DrLogD(
//                 "queueing buffer boundary dispatch request",
//                 "caller: %s", caller);
        }
    }
}

//
// Gets command line arguments
//
void DrGetUtf8CommandArgs(int argc, wchar_t *wargv[], char ***pargv)
{
    DrStr128 strArg;
    char **newArgv = NULL;

    //
    // increment count to account for tailing NULL
    //
    ++argc;

    //
    // If there are any command line args
    //
    if (argc > 0) 
    {
        //
        // Create argument storage and verify that it's correctly created
        //
        newArgv = new char *[(size_t)argc];
        LogAssert(newArgv != NULL);

        //
        // Foreach argument
        //
        for (int i = 0; i < argc; i++) 
        {
            if (wargv[i] == NULL) 
            {
                //
                // If NULL argument, store NULL
                //
                newArgv[i] = NULL;
            } 
            else 
            {
                // 
                // Get argument value and verify that it's a valid string
                //
                strArg.Set( wargv[i] );
                LogAssert(strArg.GetString() != NULL);

                //
                // Copy argument value into list 
                //
                newArgv[i] = new char[strArg.GetLength() + 1];
                LogAssert(newArgv[i] != NULL);
                memcpy(newArgv[i], strArg.GetString(), strArg.GetLength() + 1);
            }
        }
    }

    //
    // Store argument list (can be null if no arguments)
    //
    *pargv = newArgv;
}

void RChannelReaderImpl::Start(RChannelBufferPrefetchInfo* prefetchCookie)
{
    bool startSupplier = false;

    {
        AutoCriticalSection acs(&m_baseDR);

        LogAssert(m_supplier != NULL);
        LogAssert(m_state == RS_Stopped);
        LogAssert(m_cookieMap.empty());
        LogAssert(m_eventMap.empty());
        LogAssert(m_unitList.IsEmpty());
        LogAssert(m_handlerList.IsEmpty());
        LogAssert(m_interruptHandler == NULL);
        LogAssert(m_startedSupplier == false);
        m_sendLatch.Start();
        m_unitLatch.Start();

        m_state = RS_Running;
        m_readerTerminationItem = NULL;
        m_writerTerminationItem = NULL;
        m_numberOfSubItemsRead = 0;
        m_dataSizeRead = 0;
        m_prefetchCookie = prefetchCookie;

        if (m_lazyStart == false)
        {
            /* when we exit the lock, we will start the
               supplier. After the supplier has been started we'll set
               m_startedSupplierEvent to prevent a race condition in
               Interrupt */
            startSupplier = true;
            BOOL bRet = ::ResetEvent(m_startedSupplierEvent);
            LogAssert(bRet != 0);
            m_startedSupplier = true;
        }
    }

    if (startSupplier)
    {
        //todo: remove comments if not logging
//         DrLogI( "Starting Supplier");
        m_supplier->StartSupplier(m_prefetchCookie);
//         DrLogI( "Started Supplier");
        BOOL bRet = ::SetEvent(m_startedSupplierEvent);
        LogAssert(bRet != 0);
    }
}    

DrJobTicket* CreateGlobalJob()
{    
    DrError hr = S_OK;
    DrServiceDescriptor sd;
    DrRef<DrJobTicket> jobTicket = g_pDryadConfig->GetDefaultJobTicket();    
    
    sd.Set("xcps", g_pDryadConfig->GetDefaultClusterName(), NULL, "rd.RDRBasic.XComputeProcessScheduler_0");        
    
    DrRef<XcPsCreateJobRequest> msg;
    msg.Attach(new XcPsCreateJobRequest());
    msg->SetCreateJobTicket(jobTicket);    
    XcJobConstraint& constraint = msg->CreateJobConstraint();
    constraint.SetMaxConcurrentProcesses(999);
    constraint.SetMaxExecutionTime(DrTimeInterval_Hour);
    g_pDrClient->SendTo(msg, sd);    
    msg->WaitForResponse( &hr );
    if (hr != DrError_OK)
    {
        DrLogE( "DryadConfigurationManager",
            "CreateJob failied, error=%s",
            DRERRORSTRING(hr));
        LogAssert(false);
    }
    return jobTicket.Detach();    
}

/* called with m_baseDR held */
bool RChannelFifoWriter::ReWriteBlockedListForEarlyReturn(RChannelItemType
                                                          drainType)
{
    /* rewrite any blocked items to have the termination code correct
       before sending them back to the writer after a reader
       Drain. Also, discard the items, since the reader is never going
       to see them. */
    bool wakeUpWriterDrain = false;
    DrBListEntry* listEntry = m_blockedList.GetHead();
    while (listEntry != NULL)
    {
        RChannelFifoUnit* unit = m_blockedList.CastOut(listEntry);
        listEntry = m_blockedList.GetNext(listEntry);
        unit->SetStatusCode(drainType);
        unit->DiscardItems();
        if (unit == m_terminationUnit)
        {
            LogAssert(listEntry == NULL);
            m_terminationUnit = NULL;
            if (m_writerState == WS_Draining)
            {
                wakeUpWriterDrain = true;
            }
        }
    }

    return wakeUpWriterDrain;
}

DrError RChannelReader::GetTerminationStatus(DryadMetaDataRef* pErrorData)
{
    DrError status;

    RChannelItemRef writerTermination;
    RChannelItemRef readerTermination;
    GetTerminationItems(&writerTermination,
                        &readerTermination);

    LogAssert(readerTermination.Ptr() != NULL);
    if (writerTermination.Ptr() != NULL &&
        writerTermination->GetType() != RChannelItem_EndOfStream)
    {
        status = writerTermination->GetErrorFromItem();
        *pErrorData = writerTermination->GetMetaData();
    }
    else if (readerTermination->GetType() == RChannelItem_EndOfStream)
    {
        status = DrError_EndOfStream;
        *pErrorData = readerTermination->GetMetaData();
    }
    else
    {
        status = DryadError_ProcessingInterrupted;
        *pErrorData = readerTermination->GetMetaData();
    }

    return status;
}