/*++
*******************************************************************************
* Copyright (c) 1999-2000 Digi International Inc., All Rights Reserved        *
*                                                                             *
* This software contains proprietary and confidential information of Digi     *
* International Inc.  By accepting transfer of this copy, Recipient agrees    *
* to retain this software in confidence, to prevent disclosure to others,     *
* and to make no use of this software other than that for which it was        *
* delivered.  This is an unpublished copyrighted work of Digi International   *
* Inc.  Except as permitted by federal law, 17 USC 117, copying is strictly   *
* prohibited.                                                                 *
*                                                                             *
* Restricted Rights Legend                                                    *
*                                                                             *
* Use, duplication, or disclosure by the Government is subject to             *
* restrictions set forth in sub-paragraph (c)(1)(ii) of The Rights in         *
* Technical Data and Computer Software clause at DFARS 252.227-7031 or        *
* subparagraphs (c)(1) and (2) of the Commercial Computer Software -          *
* Restricted Rights at 48 CFR 52.227-19, as applicable.                       *
*                                                                             *
* Digi International Inc. 11001 Bren Road East, Minnetonka, MN 55343          *
*                                                                             *
*******************************************************************************

Module Name:

   asynrx.c

Abstract:

   Implementation of the receive (rx) engine.  This engine is the lowest layer of 
   code which performs all dxb GetMsg requests.  The DStreams API is driven from above,
	which means that all incoming (received) messages will not be presented until we ask
	for them by sending down a GetMsg request.

	The rx engine is responsible for sending GetMsg requests.  If a requests fails due
	to resource limitations, the engine must reschedule itself and retry again later.   

Author:

   Michael Zarns

--*/
#include "digiasyn.h"

STATIC VOID
RxEngineGetMsgComplete( 
		IN HANDLE   hOwner,
		IN HANDLE   Context,
		IN PVOID    pControl,
		IN ULONG    cbControl,
		IN PVOID    pData,
		IN ULONG    cbData,
		IN NTSTATUS IoStatus )
/*++

Description

	Completion routine defined by Dsapi, called when a previous GetMsg request has 
	completed.  Pass message up to owner, then reschedule RxEngine to send down another
	GetMsg request.  Always try to keep nGetMsgMax requests outstanding.

	We never call DsGetMsg directly because calling back into Dxb recursively from the 
	GetMsg completion routine could blow the kernel	stack.  Always reschedule RxEngine
	timer and let it send the request down.

--*/
{
    PASYN_RXENGINE const pEngine = Context;
    PVOID pAllocatedBuffer = pData ? (PCHAR)pData - pEngine->cbDataHeader : NULL;

    DigiDump( DIGIRXENGINE, ( "Asyn!RxEngineGetMsgComplete: hOwner 0x%x, Context 0x%x, IoStatus 0x%x\n", hOwner, Context, IoStatus ) );

    // Decrement count of pending GetMsg requests, since we're now
    // completing one
    DigiAcquireSpinLock( &pEngine->Lock );
    ASSERT( pEngine->nGetMsgPending != 0 );
    pEngine->nGetMsgPending--;
    DigiReleaseSpinLock( &pEngine->Lock );
	
    // If error, just free the data and stop scheduling - future error
    // handling will tidy everything up.
    if ( IoStatus != STATUS_SUCCESS )
    {
        PASYN_PORT_EXT	pPortExt;

        if ( pAllocatedBuffer )
            RxEngineFreeData( pAllocatedBuffer );

        if ( IoStatus == STATUS_BUFFER_TOO_SMALL )
        {
            // One specific error case - bigger packet came up than we
            // expected.  Purge it.  In this case, we do want to
            // restart the engine
            DigiDump( DIGIWARN, ( "Asyn!RxEngine: rx packet too big, purging it!\n" ) );
            DsPurge( pEngine->hStream, DS_PURGE_GETMSG );
            goto RequeueRxEngine;
        }
        else if ( IoStatus != STATUS_CANCELLED )
        {
            // Don't warn if it was cancelled.  That's an expected
            // response when the stream is being closed.
            DigiDump( DIGIWARN, ( "Asyn!RxEngineGetMsgComplete: Warning, IoStatus 0x%x!\n", IoStatus ) );
        }
        else    // STATUS_CANCELLED, expected response, make sure RxEngine gets rescheduled
        {
            goto RequeueRxEngine;
        }

        // Stop further getmsg processing
        RxEnginePause(pEngine);

        // complete any outstanding IO with error status
        (*pEngine->pOnCleanup)(pEngine->hOwner,IoStatus);

        // and exit without rescheduling RxEngine
        goto Exit;
    }

    if (cbControl)
    {
        // Control bytes present means a control+data message
        pEngine->pOnCdata( hOwner, pControl, cbControl, pAllocatedBuffer, cbData );
    }
    else if (cbData)
    {
        // Otherwise data bytes present indicate a data message, but only 
        // if data is really present.
        pEngine->pOnData( hOwner, pAllocatedBuffer, cbData );
    }
    else if (pAllocatedBuffer)
    {
        // Don't forward empty messages up.  If empty and a buffer was allocated, 
        // free it.
        RxEngineFreeData(pAllocatedBuffer);
    }
 RequeueRxEngine:

    {
        LARGE_INTEGER Delay = {0};
        (void)KeSetTimer( &pEngine->Timer, Delay, &pEngine->Dpc );
    }
 Exit:
    ;
}


STATIC VOID
RxEngine(
	IN PKDPC Ignored1, 
	IN PVOID hEngine, 
	IN PVOID Ignored2, 
	IN PVOID Ignored3 )
/*++

Description:

	The worker bee routine which sends getmsg requests.

--*/
{
    PASYN_RXENGINE pEngine = hEngine;

    for (;;)
    {
        DigiAcquireSpinLock( &pEngine->Lock );
        if ( pEngine->fHalt  )
        {
            // RxEngine has been halted. Don't send any more GetMsg requests.     
            DigiReleaseSpinLock( &pEngine->Lock );
            if ( pEngine->nGetMsgPending == 0 )
                KeCancelTimer( &pEngine->Timer );
            break;
        }
      
        if ( pEngine->fPaused )	
        {      
            // We've been told to temporarily stop sending down GetMsg requests.
            // Note: Doesn't Purge outstanding requests, simply stops sending more.
            DigiReleaseSpinLock( &pEngine->Lock );
            break;
        }
      
        ASSERT( pEngine->nGetMsgPending <= pEngine->nGetMsgMax );
        if ( pEngine->nGetMsgPending == pEngine->nGetMsgMax )
        {
            // Already have maximum number of requests outstanding.  Go away.
            DigiReleaseSpinLock( &pEngine->Lock );
            break;
        }

        pEngine->nGetMsgPending++;
        DigiReleaseSpinLock( &pEngine->Lock );
		
        // Send down a GetMsg request
        {
            NTSTATUS Status = STATUS_UNSUCCESSFUL;
            PVOID pAllocatedBuffer = NULL;
            PCHAR pData = NULL;
         
            if ( pEngine->cbMaxData != 0 )
            {
         	// Allocate a data buffer if needed.
                pAllocatedBuffer = DigiAllocMem( NonPagedPool, pEngine->cbDataHeader+pEngine->cbMaxData );

                if ( pAllocatedBuffer == NULL )
                {
                    // Don't send getmsg request if alloc failed!            
                    DigiDump( DIGIWARN, ( "Asyn!RxEngine: Alloc failed for RxEngine %p Priority %x!\n", pEngine, pEngine->Priority ) );
                    goto SystemResourceError;
                }               
			
                // Tell dxb to start filling in data at offset cbDataHeader.  This
                // allows for a client-defined header padding, most likely used for
                // a LIST_ENTRY struct to queue received buffers.
           	pData = (PCHAR)pAllocatedBuffer+pEngine->cbDataHeader;
            }
         
            DigiDump( DIGIRXENGINE, ( "Asyn!RxEngine: pEngine 0x%x, sending GetMsg\n", pEngine ) );
         
            Status = DsGetMsgEx( pEngine->hStream, pEngine->cbMaxCtrl, pData, pEngine->cbMaxData, 
                                 pEngine, pEngine->Priority, RxEngineGetMsgComplete );
                              
        SystemResourceError:
            if ( NT_ERROR( Status ) )
            {
                // Out of system resources, bail out and retry later!
                if ( pAllocatedBuffer )
                    DigiFreeMem( pAllocatedBuffer );
                DigiAcquireSpinLock( &pEngine->Lock );
                pEngine->nGetMsgPending--;
                DigiReleaseSpinLock( &pEngine->Lock );
                {
                    LARGE_INTEGER Delay;
                    Delay.QuadPart = ONE_MILLISECOND * 1000;
                    (void)KeSetTimer( &pEngine->Timer, Delay, &pEngine->Dpc );
                    DigiDump( DIGIRXENGINE | DIGIWARN, ( "Asyn!RxEngine: DigiAllocMem or DsGetMsgEx returned with an error 0x%x\n\tKeSetTimer called with timer. Engine = %p Priority = %x\n", Status, pEngine, pEngine->Priority ) );
                }
                break;
            }
        }
    }
}


VOID
RxEngineStop( 
	IN HANDLE hEngine )
/*++

Description:

	Stops the engine.

--*/
{
	PASYN_RXENGINE pEngine = hEngine;
	DigiDump( DIGIRXENGINE, ( "Asyn!RxEngineStop: pEngine 0x%x, hOwner 0x%x\n", pEngine, pEngine->hOwner ) );

	if ( pEngine )
	{
		DigiAcquireSpinLock( &pEngine->Lock );
		pEngine->fHalt = TRUE;
		DigiReleaseSpinLock( &pEngine->Lock );

        if (!KeCancelTimer(&pEngine->Timer) && KeGetCurrentIrql() < DISPATCH_LEVEL)
        {
            // RxEngine is currently running - lets back off for 20ms and let it finish
    
            LARGE_INTEGER Delay;
            Delay.QuadPart = ONE_MILLISECOND * 20;
            KeDelayExecutionThread(KernelMode,FALSE,&Delay);

        }

		(void)DsPurge( pEngine->hStream, DS_PURGE_GETMSG );
	}
}


HANDLE 
RxEngineStart(
	IN HANDLE                       hStream, 
	IN HANDLE                       hOwner, 
	IN PASYN_RXENGINE_DATA_HANDLER  pOnData,
	IN PASYN_RXENGINE_CDATA_HANDLER pOnCdata,
        IN PASYN_RXENGINE_CLEANUP_HANDLER pOnCleanup,
	IN ULONG                        cbMaxControl, 
	IN ULONG                        cbMaxData,
	IN ULONG                        cbDataHeader,
	IN ULONG                        nGetMsgRequests,
	IN DS_PRIORITY                  Priority )
/*++

Description:

	Starts the engine, which immediately starts sending GetMsg requests.  Requires a 
	previously opened dxb stream.

Parameters

	hStream
		Previously opened dxb stream (opened via DsOpen or DsOpenEx).

	hOwner
		Handle passed back to the OnData callback routines when  messages arrive.

	pOnData
		Callback routine for data only messages

	pOnCdata
		Callback routine for control+data messages

        pOnCleanup
                Callback routine to handle fatal errors from below

	cbMaxControl
		Size of Control buffer to send down to dxb in the GetMsg request.  Specifying 0 
		causes no control buffer to be sent.

	cbMaxData
		Size of Data buffer to send down in the GetMsg request.  Specifying 0 causes no
		data buffer to be sent down.

	cbDataHeader
		Number of bytes of header padding to alloc in front of data buffer.  Allows
		caller to reserve space inside the data buffer for things like LIST_ENTRYs.
		This padding is added to cbMaxData when allocating data buffers.  It is
		never used on control buffers.

	nGetMsgRequests
		Max number of simultaneous GetMsg requests to pend with Dxb.

	Priority
		DStreams priority at which to send the GetMsg requests.  See dsioctl.h.

Returns:

	NULL if unsuccessful, or valid Handle for the RxEngine, to be used by caller for 
	Pausing, Unpausing and stopping this engine.

--*/
{
    PASYN_RXENGINE pEngine = NULL;
    PASYN_PORT_EXT pPortExt = hOwner;
    DigiDump( DIGIRXENGINE, ( "Asyn!RxEngineStart: pEngine 0x%x, hOwner 0x%x\n", pEngine, hOwner ) );
	   
    if ( pPortExt->Type == ASYN_TYPE_PORT_EXT && Priority <= DS_PRIORITY_MAX )
    {   
        pEngine = &pPortExt->RxEngine[Priority];
      
        if ( !pEngine->fInitialized )
        {      
            DigiDump( DIGIRXENGINE|DIGIINIT, ( "Asyn!RxEngineStart: First time initialize for pEngine 0x%x, hOwner 0x%x\n", pEngine, hOwner ) );
      
            RtlZeroMemory( pEngine, sizeof *pEngine );
            DigiInitializeSpinLock( &pEngine->Lock );
            KeInitializeDpc( &pEngine->Dpc, RxEngine, pEngine );
            KeInitializeTimer( &pEngine->Timer );
            pEngine->fInitialized = TRUE;
        }      
   
        pEngine->hStream      = hStream;
        pEngine->hOwner       = hOwner;
        pEngine->pOnData      = pOnData;
        pEngine->pOnCdata     = pOnCdata;
        pEngine->pOnCleanup   = pOnCleanup;
        pEngine->cbMaxCtrl    = cbMaxControl;
        pEngine->cbMaxData    = cbMaxData;
        pEngine->cbDataHeader = cbDataHeader;
        pEngine->fHalt        = FALSE;
        pEngine->fPaused      = FALSE;
        pEngine->nGetMsgMax   = nGetMsgRequests;
        pEngine->Priority     = Priority;

        KeInitializeEvent( &pEngine->Stopped, SynchronizationEvent, FALSE );

        // Start the engine
        RxEngine( NULL, pEngine, NULL, NULL );
    }

    return pEngine;
}


VOID
RxEnginePause( 
	IN HANDLE hEngine )
/*++

Description:

	Tell the engine to temporarily stop sending down GetMsg requests.  Does not purge
	outstanding GetMsg requests.

--*/
{
    PASYN_RXENGINE pEngine = hEngine;

    if (pEngine) {
        // Note: No need to acquire spinlock since we're the only one writing this value 
        // and we're not trying to synchronize anything with the reader, RxEngine().
        pEngine->fPaused = TRUE;
    }
}


VOID
RxEngineUnpause( 
	IN HANDLE hEngine )
/*++

Description:

	Tell the engine to restart sending down GetMsg requests after a pause.  

--*/
{
    PASYN_RXENGINE pEngine = hEngine;

    if (pEngine) {
        // Note: No need to acquire spinlock since we're the only one writing this value 
        // and we're not trying to synchronize anything with the reader, RxEngine().
        pEngine->fPaused = FALSE;
        
        {
            LARGE_INTEGER Delay = {0};
            (void)KeSetTimer( &pEngine->Timer, Delay, &pEngine->Dpc );
        }
    }
}


VOID
RxEngineFreeData(
	IN PVOID pData )
/*++

Description:

	Routine used to free data packets allocated by the RxEngine.  	
	RxEngineGetMsgComplete passes the pData buffer into the OnData handlers.  
	Handlers are responsible for freeing the buffer by calling RxEngineFreeData.

--*/
{
    DigiFreeMem( pData );
}
