 /*++
*******************************************************************************
* Copyright (c) 1990-97 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:

   DsLoop.c

Abstract:

   This program is a win32 console application which is used to stress a 
   DStreams service provider driver.  It uses the DStreams DsIoctl interface
	to communicate with the driver.

	This app runs a LOOP stress.  It opens multiple streams to the LOOP device,
	starts up two threads per stream - one for tx and one for rx, and sends data
	as fast as possible.  Terminate stress by hitting CTRL-C or CTRL-BREAK.

	Configuration params *should* be stored/retrieved from the registry, but for
	now are compiled in as constants.

Author:

   Michael Zarns

Revision History:

--*/

#include <stdio.h>
#include <stdlib.h>
#include <windows.h>

#include "DsIoctl.h"


// Define COUNT to send a counted number of PutMsgs.  Debug printf's tell how many
// Getmsgs were received -- compare tx vs rx to verify no packets lost.
// #define COUNT ( 1 )

// Define HOUSTON to build dsloop for a houston adapter.  Otherwise it builds for
// Syncport/DFPrime.  Syncport uses SLO0 loopback device, houston uses LIQUOR or
// IWANLOOP.
//#define HOUSTON

// Define LIQUOR along with HOUSTON to use the liquor cabinet.  Otherwise it builds
// for IWANLOOP.
//#define LIQUOR

// Define INITIALIZE_DATA to enable the TxThread to initialize the data portion of the 
// putmsg packet.  Verification is *NOT* performed in the RxThread.
//#define INITIALIZE_DATA

// Define LOUD to help troubleshoot flowcontrol problems.  It causes chars to be
// printed out for every packet transmitted/received, and when timeouts occur.
//#define LOUD


///////////////////////////////////////////////////////////////////////////////
//
// Globals 
//
struct STRESS_DEVICE;

HANDLE ghShutdownEvent;					// Set by ControlHandler to indicate shutdown
struct STRESS_DEVICE * gaDevices;	// Array of stress device object
int gnDevices;								// Count of devices in the gaDevices array.


///////////////////////////////////////////////////////////////////////////////
//
//  Constants and Defines
//

// Count of Streams to use
#define NSTREAMS 8
// Count of PutMsgs per Stream
#define NPUTS 64
// Count of GetMsgs per Stream
#define NGETS 128
// Count of bytes per PutMsg buffer
#define NPUTBYTES 64
// Count of bytes per GetMsg buffer
#define NGETBYTES 64
// Count of bytes for control buffer
#define NCTRLBYTES 4

#ifdef HOUSTON

#ifdef LIQUOR
#define LOOP_DEVICE      "LIQUOR"
#define LOOP_MINOR_BEER  1
#define LOOP_MINOR_LUSH  2
#define LOOP_MINOR_VOMIT 3
#define USE_MINOR        TRUE
#define LOOP_MINOR       LOOP_MINOR_VOMIT
#else // IWANLOOP
#include "dxbtypes.h"
#ifndef MAXIOCBSZ
#define MAXIOCBSZ 1024
#endif
#include "iwan_control.h"
#include "iwan_proto.h"
#define LOOP_DEVICE      "IWANLOOP"
#define USE_MINOR        FALSE
#define LOOP_MINOR       0
#endif

#else 

#define LOOP_DEVICE      "SLO0"
#define USE_MINOR        FALSE
#define LOOP_MINOR       0

#endif

#ifdef LOUD
HANDLE hMutex;
#endif

// A MY_PUTMSG object is used for every PutMsg request
typedef struct MY_PUTMSG
{
	DS_PUTMSG DsPutMsg;
	PUCHAR pControl;
	PUCHAR pData;
	OVERLAPPED Ol;
	DWORD dwBytes;
} MY_PUTMSG, *PMY_PUTMSG;

// A MY_GETMSG object is used for every GetMsg request
typedef struct MY_GETMSG
{
	DS_GETMSG DsGetMsg;
	PUCHAR pControl;
	PUCHAR pData;
	OVERLAPPED Ol;
	DWORD dwBytes;
} MY_GETMSG, *PMY_GETMSG;

// A MY_IOCTL object is used for every Ioctl request
typedef struct MY_IOCTL
{
	DS_IOCTL DsIoctl;
	PUCHAR pData;
	OVERLAPPED Ol;
	DWORD dwBytes;
} MY_IOCTL, *PMY_IOCTL;

// A STREAM object holds all info for a single stream
typedef struct STREAM
{
	HANDLE hStream;
	HANDLE hFile;		// a copy of the filehandle in parent STRESS_DEVICE struct.
	HANDLE hTxThread;
	HANDLE hRxThread;
	MY_PUTMSG Puts[NPUTS];
	MY_GETMSG Gets[NGETS];
} STREAM, *PSTREAM;


// The STRESS_DEVICE object contains all info for a specific device, i.e. digisxb2, 
// on which we'll run stress.
typedef struct STRESS_DEVICE
{
	PCHAR pDeviceName;
	HANDLE hFile;
	STREAM Streams[NSTREAMS];
} STRESS_DEVICE, *PSTRESS_DEVICE;


// Use this to initialize the Overlapped struct
_inline VOID InitOl( LPOVERLAPPED pOl )
{
	ZeroMemory( pOl, sizeof( OVERLAPPED ) );
	pOl->hEvent = CreateEvent( NULL, TRUE, FALSE, NULL );
}


///////////////////////////////////////////////////////////////////////////////
//
// Prototypes
//

BOOL Startup( PSTRESS_DEVICE pDevice );
VOID Shutdown( void );
BOOL ControlHandler( DWORD CtrlType );
DWORD WINAPI TxThread( PVOID Context );
DWORD WINAPI RxThread( PVOID Context );


#define PrintUsage() \
   printf( "usage:\n\n" \
           "   DsLoop <device> [<device2> <device3> ...]\n\n" \
           "where <device> is a devicename, i.e. digisxb2\n" );

#if DBG
#define ASSERT( statement )                                    \
{																					\
	if ( !(statement) )														\
	{																				\
		printf( "\nASSERT Failed: " #statement  						\
		        " \n\tfile: %s, line %d\n", __FILE__, __LINE__ );	\
		DebugBreak();															\
	}																				\
}
#else
#define ASSERT( statement )  ;
#endif


///////////////////////////////////////////////////////////////////////////////
//
// Routines
// 

void _cdecl
main( int argc, char *argv[] )
{
	char *pDeviceName = NULL;
	int i;
	BOOL bSuccess;


	// This program requires at least 1 argument - the name of the device on which to 
	// run stress, i.e. digisxb2.  Multiple devices may be specified.
	if ( argc == 1 )
	{
		PrintUsage(  );
		return;
	}
	ASSERT( argc > 1 );

	// alloc global array of stress devices
	gnDevices = argc-1;
	gaDevices = GlobalAlloc( GPTR, (gnDevices)*sizeof( STRESS_DEVICE ) );
	if ( !gaDevices )
		return;

	// Set handler to catch CTRL-C which signals us to cleanup and exit
	ghShutdownEvent = CreateEvent( NULL, TRUE, FALSE, NULL );
	bSuccess = SetConsoleCtrlHandler( ControlHandler, TRUE );
	ASSERT( bSuccess );

	// Start stress on all devices
 	for ( i = 0; i < gnDevices; i++ )
	{
		pDeviceName = argv[i+1];	// argv[0] is the executable name
		ASSERT( pDeviceName );
		ASSERT( *pDeviceName );

		gaDevices[i].pDeviceName = pDeviceName;
		bSuccess = Startup( &gaDevices[i] );
		if ( !bSuccess )
		{
			// Disable CTRL-C, we're already cleaning up
			SetConsoleCtrlHandler( NULL, TRUE );
			SetEvent( ghShutdownEvent );
			break;
		}
	}

	WaitForSingleObject( ghShutdownEvent, INFINITE );
	Shutdown();
	ExitProcess( 0 );
}


BOOL 
SyncIoRequest( HANDLE hDevice, 
               ULONG  Ioctl, 
               PVOID  pInData, 
               ULONG  cbDataIn, 
               PVOID  pOutData, 
               ULONG  cbDataOut )
{
	OVERLAPPED Ol;
	DWORD dwBytes;
	BOOL bSuccess;

	ZeroMemory( &Ol, sizeof( Ol ) );
	Ol.hEvent = CreateEvent( NULL, TRUE, FALSE, NULL );
	if ( !Ol.hEvent )
		return( FALSE );

	bSuccess = DeviceIoControl( hDevice, Ioctl, 
	                            pInData, cbDataIn,
	                            pOutData, cbDataOut,
	                            &dwBytes, &Ol );
	if ( !bSuccess  && GetLastError() == ERROR_IO_PENDING )
		bSuccess = GetOverlappedResult( hDevice, &Ol, &dwBytes, TRUE );
	
	if ( !bSuccess )
		return( FALSE );
	return TRUE;
}


BOOL 
AsyncIoRequest( HANDLE hDevice, 
                ULONG Ioctl, 
                PVOID pInData, 
                ULONG cbDataIn, 
                PVOID pOutData, 
                ULONG cbDataOut,
                PDWORD pdwBytes,
                LPOVERLAPPED pOl )
{
	HANDLE hTemp;

	ASSERT( pOl->hEvent );
	hTemp = pOl->hEvent;
	ZeroMemory( pOl, sizeof( OVERLAPPED ) );
	pOl->hEvent = hTemp;
	ResetEvent( pOl->hEvent );
	*pdwBytes = 0;

	return( DeviceIoControl( hDevice, Ioctl, 
	                         pInData, cbDataIn,
	                         pOutData, cbDataOut,
	                         pdwBytes, pOl ) ); 
}


BOOL
AsyncPutMsg( HANDLE hFile, HANDLE hStream, PMY_PUTMSG pPut )
{
	BOOL bSuccess;
	bSuccess = AsyncIoRequest( hFile, (ULONG)IOCTL_DS_PUTMSG,
	                           &pPut->DsPutMsg, sizeof( DS_PUTMSG ),
	                           &pPut->DsPutMsg, sizeof( DS_PUTMSG ),
	                           &pPut->dwBytes, &pPut->Ol );
	if ( !bSuccess )
	{
		if ( GetLastError() != ERROR_IO_PENDING )
		{
			// Request failed, set event so we try again next time around
			OutputDebugString( "PutMsg request failed!\n" );
			SetEvent( pPut->Ol.hEvent );
		}
	}
	return( bSuccess );
}

BOOL 
SyncPutMsg( HANDLE hFile, HANDLE hStream, PMY_PUTMSG pPut )
{
	BOOL bSuccess;
	bSuccess = SyncIoRequest( hFile, (ULONG)IOCTL_DS_PUTMSG,
	                           &pPut->DsPutMsg, sizeof( DS_PUTMSG ),
	                           &pPut->DsPutMsg, sizeof( DS_PUTMSG ) );
	if ( !bSuccess )
	{
		if ( GetLastError() != ERROR_IO_PENDING )
		{
			// Request failed, set event so we try again next time around
			OutputDebugString( "PutMsg request failed!\n" );
		}
	}
	return( bSuccess );
}


BOOL
AsyncGetMsg( HANDLE hFile, HANDLE hStream, PMY_GETMSG pGet )
{
	BOOL bSuccess;
	bSuccess = AsyncIoRequest( hFile, (ULONG)IOCTL_DS_GETMSG,
	                           &pGet->DsGetMsg, sizeof( DS_GETMSG ),
	                           &pGet->DsGetMsg, sizeof( DS_GETMSG ),
	                           &pGet->dwBytes, &pGet->Ol );
	if ( !bSuccess )
	{
		if ( GetLastError() != ERROR_IO_PENDING )
		{
			// Request failed, set event so we try again next time around
			OutputDebugString( "GetMsg request failed!\n" );
			SetEvent( pGet->Ol.hEvent );
		}
	}
	return( bSuccess );
}

BOOL 
SyncGetMsg( HANDLE hFile, HANDLE hStream, PMY_GETMSG pGet )
{
	BOOL bSuccess;
	bSuccess = SyncIoRequest( hFile, (ULONG)IOCTL_DS_GETMSG,
	                           &pGet->DsGetMsg, sizeof( DS_GETMSG ),
	                           &pGet->DsGetMsg, sizeof( DS_GETMSG ) );
	if ( !bSuccess )
	{
		if ( GetLastError() != ERROR_IO_PENDING )
		{
			// Request failed, set event so we try again next time around
			OutputDebugString( "GetMsg request failed!\n" );
		}
	}
	return( bSuccess );
}


BOOL 
SyncIoctl( HANDLE hFile, HANDLE hStream, PMY_IOCTL pIoctl )
{
	BOOL bSuccess;
	bSuccess = SyncIoRequest( hFile, (ULONG)IOCTL_DS_IOCTL,
	                           &pIoctl->DsIoctl, sizeof( DS_IOCTL ),
	                           &pIoctl->DsIoctl, sizeof( DS_IOCTL ) );
	if ( !bSuccess )
	{
		if ( GetLastError() != ERROR_IO_PENDING )
		{
			// Request failed, set event so we try again next time around
			OutputDebugString( "Ioctl request failed!\n" );
		}
	}
	else if ( pIoctl->DsIoctl.ReturnValue != 0 || pIoctl->DsIoctl.ErrorCode != 0  )
	{
		OutputDebugString( "Ioctl request failed!\n" );
		bSuccess = FALSE;
	}
	return( bSuccess );
}



BOOL 
Startup( 
	PSTRESS_DEVICE pDevice )
{
	CHAR FullDeviceName[64];
	HANDLE hFile;
	int i, j;

#ifdef LOUD
	hMutex = CreateMutex( NULL, FALSE, NULL );
	if ( hMutex == NULL )
		return FALSE;
#endif

	// Open the NT device
	strcpy( FullDeviceName, "\\\\.\\" );
	strcat( FullDeviceName, gaDevices->pDeviceName );
	printf( "Startup: opening device %s\n", FullDeviceName );
	hFile = CreateFile( FullDeviceName, GENERIC_READ | GENERIC_WRITE,  
	                    FILE_SHARE_READ | FILE_SHARE_WRITE, NULL, OPEN_EXISTING, 
	                    FILE_ATTRIBUTE_NORMAL | FILE_FLAG_OVERLAPPED, NULL );

	if ( hFile == INVALID_HANDLE_VALUE )
	{
		printf( "Startup: failure opening device\n");
		goto ExitError;
	}

	pDevice->hFile = hFile;

	// Open multiple streams to the Streams Loop device
	for ( i = 0; i < NSTREAMS; i++ )
	{
		DS_OPEN Open;
		BOOL bSuccess;
		DWORD Id;
		PSTREAM pStream;
		ULONG Minor = LOOP_MINOR;

#ifdef LOUD
		WaitForSingleObject( hMutex, 1000 );
		printf( "Startup: opening stream %d\n", i );
		ReleaseMutex( hMutex );
#endif

		ZeroMemory( &Open, sizeof( Open ) );
		strcpy( (PVOID)&Open.DeviceName, LOOP_DEVICE );
		Open.pUserData  = USE_MINOR ? &Minor       : NULL;
		Open.cbUserData = USE_MINOR ? sizeof Minor : 0;
		bSuccess = SyncIoRequest( pDevice->hFile, (ULONG)IOCTL_DS_OPEN,
		                          &Open, sizeof( Open ), &Open, sizeof( Open ) );
		if ( !bSuccess || Open.ErrorCode )
		{
			printf( "Startup: Open failed, bSuccess 0x%x, ErrorCode 0x%x!\n", bSuccess, Open.ErrorCode );
			goto ExitError;
		}

		pStream = &pDevice->Streams[i];
		pStream->hStream = Open.hStream;
		pStream->hFile = pDevice->hFile;

// Houston IWANLOOP specific code
#ifdef HOUSTON
#ifndef LIQUOR
#define SNID 1
		// Send SetSnid Ioctl
		{
			DWORD Error;
			MY_IOCTL Ioctl = {0};
			wan_snioc_t SetSnid = {0};
			SetSnid.w_type   = WAN_SNID;
			SetSnid.w_class  = ISDN_BCHAN;
			SetSnid.w_chanid = 1;			// One based?
			SetSnid.w_line   = 0;			// Zero based
			SetSnid.w_snid   = SNID;
			Ioctl.DsIoctl.hStream = pStream->hStream;
			Ioctl.DsIoctl.Command = W_SETSNID;
			Ioctl.DsIoctl.cbDataIn = Ioctl.DsIoctl.cbMaxDataOut = sizeof SetSnid;
			Ioctl.DsIoctl.pDataIn = Ioctl.DsIoctl.pDataOut = &SetSnid;

			bSuccess = SyncIoctl( pStream->hFile, pStream->hStream, &Ioctl );
			if ( !bSuccess )
			{
				printf( "SetSnid request failed\n" );
				goto ExitError;
			}
		}

		// Send wan registration
		{
			{
				MY_PUTMSG PutMsg = {0};
				struct wan_reg WanReg = {0};
				WanReg.wan_type = WAN_REG;
				WanReg.wan_snid = SNID;
				PutMsg.DsPutMsg.hStream = pStream->hStream;
				PutMsg.DsPutMsg.pControl = &WanReg;
				PutMsg.DsPutMsg.cbControl = sizeof WanReg;
				bSuccess = SyncPutMsg( pStream->hFile, pStream->hStream, &PutMsg );
				if ( !bSuccess )
				{
					printf( "Wan Registration PutMsg failed\n" );
					goto ExitError;
				}
			}	
			
			// Did wan registration succeed?
			{
				MY_GETMSG GetMsg = {0};
				struct wan_reg WanReg = {0};
				GetMsg.DsGetMsg.hStream = pStream->hStream;
				GetMsg.DsGetMsg.pControl = &WanReg;
				GetMsg.DsGetMsg.cbMaxControl = sizeof WanReg;
				bSuccess = SyncGetMsg( pStream->hFile, pStream->hStream, &GetMsg );
				if ( !bSuccess )
				{
					printf( "Wan Registration GetMsg request failed\n");
					goto ExitError;
				}
				
				if ( GetMsg.DsGetMsg.cbControl != sizeof WanReg || 
				     WanReg.wan_type != WAN_REG || WanReg.wan_status != WAN_SUCCESS )
				{
					printf( "Wan Registration failed, type 0x%x, status 0x%x\n", 
	                        WanReg.wan_type, WanReg.wan_status );
					goto ExitError;
				}
			}
		}

		// Send wc_connect
		{
			{
				MY_PUTMSG PutMsg = {0};
				struct wan_ctl WanCtl = {0};

				WanCtl.wan_type    = WAN_CTL;
				WanCtl.wan_command = WC_CONNECT;
				WanCtl.wan_remtype = WAN_TYPE_ASC;
				WanCtl.wan_remsize = '0';
				PutMsg.DsPutMsg.hStream = pStream->hStream;
				PutMsg.DsPutMsg.pControl = &WanCtl;
				PutMsg.DsPutMsg.cbControl = sizeof WanCtl;
				PutMsg.DsPutMsg.Priority = DS_PRIORITY_HIGH;
				bSuccess = SyncPutMsg( pStream->hFile, pStream->hStream, &PutMsg );
				if ( !bSuccess )
				{
					printf( "WC_CONNECT PutMsg failed\n");
					goto ExitError;
				}
			}	
			
			// Did WC_CONNECT succeed?
			{
				MY_GETMSG GetMsg = {0};
				struct wan_ctl WanCtl = {0};
				GetMsg.DsGetMsg.hStream = pStream->hStream;
				GetMsg.DsGetMsg.pControl = &WanCtl;
				GetMsg.DsGetMsg.cbMaxControl = sizeof WanCtl;
				bSuccess = SyncGetMsg( pStream->hFile, pStream->hStream, &GetMsg );
				if ( !bSuccess )
				{
					printf( "WC_CONNECT GetMsg request failed\n" );
					goto ExitError;
				}
				
				if ( GetMsg.DsGetMsg.cbControl != sizeof WanCtl || 
				     WanCtl.wan_command != WC_CONCNF || WanCtl.wan_status != WAN_SUCCESS )
				{
					printf( "WC_CONNECT GetMsg returned junk\n");
					goto ExitError;
				}
			}
		}
#undef SNID
#endif
#endif

		// init putmsgs requests for this stream
		for ( j = 0; j < NPUTS; j++ )
		{
			// Start Event state is signalled, so TxThread will immediately send a pkt
			pStream->Puts[j].Ol.hEvent = CreateEvent( NULL, TRUE, TRUE, NULL );
//			pStream->Puts[j].pControl = GlobalAlloc( GPTR, NCTRLBYTES );
			pStream->Puts[j].pData = GlobalAlloc( GPTR, NPUTBYTES );
			if ( !pStream->Puts[j].Ol.hEvent || 
//			     !pStream->Puts[j].pControl ||
			     !pStream->Puts[j].pData )
				return( FALSE );

			pStream->Puts[j].DsPutMsg.hStream = pStream->hStream;
//			pStream->Puts[j].DsPutMsg.pControlBuffer = pStream->Puts[j].pControl;
//			pStream->Puts[j].DsPutMsg.ControlLength = NCTRLBYTES;
			pStream->Puts[j].DsPutMsg.pData = pStream->Puts[j].pData;
			pStream->Puts[j].DsPutMsg.cbData = NPUTBYTES;
			
		}

		// init getmsg requests for this stream
		for ( j = 0; j < NGETS; j++ )
		{
			pStream->Gets[j].Ol.hEvent = CreateEvent( NULL, TRUE, TRUE, NULL );
			pStream->Gets[j].pControl = GlobalAlloc( GPTR, NCTRLBYTES );
			pStream->Gets[j].pData = GlobalAlloc( GPTR, NGETBYTES );
			if ( !pStream->Gets[j].Ol.hEvent || 
			     !pStream->Gets[j].pControl ||
			     !pStream->Gets[j].pData )
			{
				printf( "Failure on Get allocs\n" );
				goto ExitError;
			}

			pStream->Gets[j].DsGetMsg.hStream = pStream->hStream;
			pStream->Gets[j].DsGetMsg.pControl = pStream->Gets[j].pControl;
			pStream->Gets[j].DsGetMsg.cbMaxControl = NCTRLBYTES;
			pStream->Gets[j].DsGetMsg.pData = pStream->Gets[j].pData;
			pStream->Gets[j].DsGetMsg.cbMaxData = NGETBYTES;
		}

		// Create tx and rx threads
		pStream->hRxThread = CreateThread( NULL, 0, RxThread, pStream, 0, &Id );
		pStream->hTxThread = CreateThread( NULL, 0, TxThread, pStream, 0, &Id );
		if ( !pStream->hTxThread || !pStream->hRxThread )
		{
			printf( "Failure creating threads\n" );
			goto ExitError;
		}
	}
	
	return( TRUE );

ExitError:;
	printf( "Startup: Failure, aborting!\n" );
	return( FALSE );
}


VOID
Shutdown()
{
	int i, j, k;
	PSTRESS_DEVICE pDevice;
	PSTREAM pStream;

	// Shutdown all stress devices
	for ( i = 0; i < gnDevices; i++ )
	{
		pDevice = &gaDevices[i];
	
		// Shutdown all streams on this device
		for ( j = 0; j < NSTREAMS; j++ )
		{
			pStream = &pDevice->Streams[j];

			// Wait for threads to exit
			if ( pStream->hTxThread )
			{
				OutputDebugString( "Waiting for TxThread\n" );
				WaitForSingleObject( pStream->hTxThread, INFINITE );
				CloseHandle( pStream->hTxThread );
				OutputDebugString( "TxThread wait completed\n" );
			}
			if ( pStream->hRxThread )
			{
				OutputDebugString( "Waiting for RxThread\n" );
				WaitForSingleObject( pStream->hRxThread, INFINITE );
				CloseHandle( pStream->hRxThread );
				OutputDebugString( "RxThread wait completed\n" );
			}

			// Deallocate buffers
			for ( k = 0; k < NPUTS; k++ )
			{
				if ( pStream->Puts[k].pData )
					GlobalFree( pStream->Puts[k].pData );
				if ( pStream->Puts[k].pControl )
					GlobalFree( pStream->Puts[k].pControl );
			}
			for ( k = 0; k < NGETS; k++ )
			{
				if ( pStream->Gets[k].pData )
					GlobalFree( pStream->Gets[k].pData );
				if ( pStream->Gets[k].pControl )
					GlobalFree( pStream->Gets[k].pControl );
			}

			// Insert silly delay for Dense adapters that don't like closes too fast.
			Sleep( 200 );

			// Close the stream
			if ( pStream->hStream )
			{
				DS_CLOSE Close;
				OutputDebugString( "Closing stream..." );
				Close.hStream = pStream->hStream;
				SyncIoRequest( pDevice->hFile, (ULONG)IOCTL_DS_CLOSE, 
				               &Close, sizeof( Close ),
				               &Close, sizeof( Close ) );
				OutputDebugString( "Closed\n" );
			}
		}
		if ( pDevice->hFile )
		{
			OutputDebugString( "Closing Device\n" );
			CloseHandle( pDevice->hFile );
		}
	}
}


BOOL 
ControlHandler( 
	DWORD CtrlType )
{
	switch ( CtrlType )
	{
		case CTRL_C_EVENT:
		case CTRL_BREAK_EVENT:
		case CTRL_CLOSE_EVENT:
		case CTRL_LOGOFF_EVENT:
		case CTRL_SHUTDOWN_EVENT:
			SetEvent( ghShutdownEvent );
			return( TRUE );

		default:
			ASSERT( "*** ControlHandler: Unrecognized CtrlType! ***\n" && 0 );
			return( FALSE );
	}
}

DWORD WINAPI 
TxThread( PVOID Context )
{
	DWORD ShutdownResult;
	PSTREAM pStream = Context;
	int Index = 0;
#ifdef COUNT
	int Count = COUNT;
#endif

	while ( ( ShutdownResult = WaitForSingleObject( ghShutdownEvent, 0 ) ) == WAIT_TIMEOUT )
	{
		DWORD dwBytes;
		if ( WaitForSingleObject( pStream->Puts[Index].Ol.hEvent, 100 ) == WAIT_OBJECT_0 && 
		     GetOverlappedResult( pStream->hFile, &pStream->Puts[Index].Ol, &dwBytes, FALSE ) )
		{
#ifdef COUNT
			if ( Count-- == 0 )
			{
				Count = 0;	// Once we've reached 0, stay at 0.
				continue;
			}
#endif

#ifdef INITIALIZE_DATA
			{
				ULONG n;
				for ( n=0; n < NPUTBYTES; n++ )
					pStream->Puts[Index].pData[n] = (UCHAR)( n % 256 );
			}
#endif
			// PutMsg request has completed, send another request immediately!
			AsyncPutMsg( pStream->hFile, pStream->hStream, &pStream->Puts[Index] );
			Index = (Index+1) % NPUTS;	// Next PutMsg
#ifdef LOUD
			WaitForSingleObject( hMutex, 1000 );
			printf( "(" );
			ReleaseMutex( hMutex );
#endif
		}
		else
		{
			// Wait timed out, do nothing.
			// We'll check for Shutdown, then wait again for this PutMsg.
#ifdef LOUD
			WaitForSingleObject( hMutex, 1000 );
			printf(",");
			ReleaseMutex( hMutex );
#endif
		}
	}

	// Shutting down.  Cancel all outstanding requests, wait completion and exit.
	{
		DS_PURGE Purge;
		Purge.hStream = pStream->hStream;
		Purge.Flags = DS_PURGE_PUTMSG;
		SyncIoRequest( pStream->hFile, (ULONG)IOCTL_DS_PURGE, 
		               &Purge, sizeof( Purge ), 
		               &Purge, sizeof( Purge ) );
	}

	for ( Index = 0; Index < NPUTS; Index++ )
	{
		// Wait for each request to complete
		char dbgstr[128];
		sprintf( dbgstr, "Waiting for pStream 0x%p, Puts[%d]\n", pStream, Index );
		OutputDebugString( dbgstr );
		WaitForSingleObject( pStream->Puts[Index].Ol.hEvent, INFINITE );
	}
	OutputDebugString( "Exiting TxThread\n" );
	ExitThread( 0 );
	return( 0 );	// Never executed, prevent compiler warning: no return value.
}


DWORD WINAPI 
RxThread( PVOID Context )
{
	DWORD ShutdownResult;
	PSTREAM pStream = Context;
	DS_PURGE Purge;
	int Index = 0;
	int Total = 0;
#ifdef COUNT
	int Count = COUNT;
#endif
	ULONG StartTime = GetTickCount();

	while ( ( ShutdownResult = WaitForSingleObject( ghShutdownEvent, 0 ) ) == 
	        WAIT_TIMEOUT )
	{
		DWORD dwBytes;
		if ( WaitForSingleObject( pStream->Gets[Index].Ol.hEvent, 100 ) == WAIT_OBJECT_0 && 
		     GetOverlappedResult( pStream->hFile, &pStream->Gets[Index].Ol, &dwBytes, FALSE ) )
		{
			// Request has completed, send another request immediately!
#ifdef COUNT
			if ( Count-- == 0 )
				SetEvent( ghShutdownEvent );
#endif

#if 0
			if ( pStream->Gets[Index].DsGetMsg.cbData != NPUTBYTES )
			{
				char dbgbuf[128];
				sprintf( dbgbuf, "Error, GetMsg returned 0x%x bytes, expected 0x%x!\n", pStream->Gets[Index].DsGetMsg.cbData, NPUTBYTES );
				OutputDebugString( dbgbuf );
			}
#endif
			AsyncGetMsg( pStream->hFile, pStream->hStream, &pStream->Gets[Index] );
			Index = (Index+1) % NGETS;	// Next GetMsg
			if ( ++Total % 1024 == 0 )
			{
				char dbgbuf[128];
				sprintf( dbgbuf, "Stream 0x%p PPS %d, Total %d\n", pStream, 
				         Total * 1024 / (GetTickCount()-StartTime), Total );
				OutputDebugString( dbgbuf );
			}
#ifdef LOUD
			WaitForSingleObject( hMutex, 1000 );
			printf( ")" );
			ReleaseMutex( hMutex );
#endif
		}
		else
		{
			// Wait timed out, do nothing.
			// We'll check for Shutdown, then wait again for this GetMsg to complete.
#ifdef LOUD
			WaitForSingleObject( hMutex, 1000 );
			printf( "." );
			ReleaseMutex( hMutex );
#endif
		}
	}

	// Shutting down.  Cancel all outstanding requests, wait completion and exit.
#ifdef COUNT
	{
		char dbgbuf[128];
		sprintf( dbgbuf, "Stream 0x%08x Total %d\n", pStream, Total );
		OutputDebugString( dbgbuf );
	}
#endif
//	ASSERT( ShutdownResult == WAIT_OBJECT_0 );
	Purge.hStream = pStream->hStream;
	Purge.Flags = DS_PURGE_GETMSG;
	SyncIoRequest( pStream->hFile, (ULONG)IOCTL_DS_PURGE, 
	               &Purge, sizeof( Purge ), 
	               &Purge, sizeof( Purge ) );
	for ( Index = 0; Index < NGETS; Index++ )
	{
		// Wait for each request to complete
		char dbgstr[128];
		sprintf( dbgstr, "Waiting for pStream 0x%p, Gets[%d]\n", pStream, Index );
		OutputDebugString( dbgstr );
		WaitForSingleObject( pStream->Gets[Index].Ol.hEvent, INFINITE );
	}
	OutputDebugString( "Exiting RxThread\n" );
	ExitThread( 0 );
	return( 0 );	// Never executed, prevent compiler warning: no return value.
}
