Tricks of the Windows Gam…ming Gurus (2nd Edition)/Disc2.iso/msdn_vcb/samples/vc98/sdk/netds/winsock/iocomplt/socksrv.c

C/C++ Source or Header | 1997-10-05 | 19.1 KB | 756 lines

// THIS CODE AND INFORMATION IS PROVIDED "AS IS" WITHOUT WARRANTY OF
// ANY KIND, EITHER EXPRESSED OR IMPLIED, INCLUDING BUT NOT LIMITED TO
// THE IMPLIED WARRANTIES OF MERCHANTABILITY AND/OR FITNESS FOR A
// PARTICULAR PURPOSE.
//
// Copyright (C) 1993-1997  Microsoft Corporation.  All Rights Reserved.
//
//  MODULE:   socksrv.c
//
//  PURPOSE:  Handle and Benchmark client transactions.
//
//  FUNCTIONS:
//     main              - SockSrv entry point
//     SortTheBuffer          - Create a sorted copy of a buffer
//     CompleteBenchmark      - Time the client transactions
//     WorkerThread          - Process client requests
//     CreateWorkers          - Create worker threads
//     CreateNetConnections   - Establish client connections
//     ParseSwitch          - Process command line option
//     ShowUsage          - Display usage help
//     Random              - Generate a random # within a given range
//
//

#include "socksrv.h"

int _CRTAPI1
main (
        int argc,
        char *argv[],
        char *envp[]
)
{

   char chChar, *pchChar;
   DWORD lc;
   BOOL b;
   int WriteCount;

   //
   // try to get timing more accurate... Avoid context
   // switch that could occur when threads are released
   //

   SetThreadPriority (GetCurrentThread (), THREAD_PRIORITY_TIME_CRITICAL);

   //
   // Figure out how many processors we have to size the minimum
   // number of worker threads and concurrency
   //

   GetSystemInfo (&SystemInfo);

   fVerbose = FALSE;
   dwNumberOfClients = 1;
   dwNumberOfWorkers = 2 * SystemInfo.dwNumberOfProcessors;
   dwConcurrency = SystemInfo.dwNumberOfProcessors;
   fTcp = TRUE;
   dwWorkIndex = 4;

   while (--argc)
   {
      pchChar = *++argv;
      if (*pchChar == '/' || *pchChar == '-')
      {
         while (chChar = *++pchChar)
         {
            ParseSwitch (chChar, &argc, &argv);
         }
      }
   }

   //
   // If we are doing file I/O, then create a large file that clients
   // can randomly seek and read from
   //

   srand (1);

   //
   // Create a new file and make it the correct size
   //

   DeleteFile ("socksrv.dat");

   hFile = CreateFile (
                         "socksrv.dat",
                         GENERIC_READ | GENERIC_WRITE,
                         FILE_SHARE_READ | FILE_SHARE_WRITE,
                         NULL,
                         OPEN_ALWAYS,
   FILE_FLAG_DELETE_ON_CLOSE | FILE_ATTRIBUTE_NORMAL | FILE_FLAG_OVERLAPPED,
                         NULL
      );
   if (hFile == INVALID_HANDLE_VALUE)
   {
      fprintf (stderr, "SOCKSRV: Error opening file %d\n", GetLastError ());
      exit (1);
   }

   //
   // Initialize the file with random data
   //

   ClientData[0].Overlapped.Offset = 0;
   for (WriteCount = 0; WriteCount < NUMBER_OF_WRITES; WriteCount++)
   {

      for (lc = 0; lc < SIXTEEN_K; lc++)
      {
         InitialBuffer[lc] = (DWORD) rand ();
      }

      b = WriteFile (hFile, InitialBuffer, sizeof (InitialBuffer), &lc, &ClientData[0].Overlapped);

      if (!b && GetLastError () != ERROR_IO_PENDING)
      {
         fprintf (stderr, "SOCKSRV: Error in pre-write %d\n", GetLastError ());
         exit (1);
      }
      b = GetOverlappedResult (
                                 hFile,
                                 &ClientData[0].Overlapped,
                                 &lc,
                                 TRUE
         );
      if (!b || lc != sizeof (InitialBuffer))
      {
         fprintf (stderr, "SOCKSRV: Wait for pre-write failed %d\n", GetLastError ());
         exit (1);
      }
      ClientData[0].Overlapped.Offset += lc;
   }

   srand (1);
   fprintf (stdout, "SOCKSRV: %2d Clients %2d Workers Concurrency %d Using %s WorkIndex %d\n",
            dwNumberOfClients,
            dwNumberOfWorkers,
            dwConcurrency,
            fTcp ? "TCP" : "IPX",
            dwWorkIndex
      );

   if (!CreateNetConnections ())
   {
      exit (1);
   }

   if (!CreateWorkers ())
   {
      exit (1);
   }

   if (fVerbose)
   {
      fprintf (stdout, "Workers Created, Waiting for Clients to complete\n");
   }

   CompleteBenchmark ();

   exit (1);
   return 1;
}

VOID
WINAPI
CompleteBenchmark (
                     VOID
)
{
   DWORD LowestTicks;
   DWORD HighestTicks;
   DWORD TotalTicks;
   DWORD TotalIterations;
   DWORD TotalBytesTransferred;
   DWORD i;

   StartTime = GetTickCount ();
   SetEvent (hBenchmarkStart);
   WaitForSingleObject (hBenchmarkComplete, INFINITE);
   EndTime = GetTickCount ();

   LowestTicks = ClientData[0].IoBuffer.u.IAmDone.TotalTicks;
   HighestTicks = ClientData[0].IoBuffer.u.IAmDone.TotalTicks;
   TotalTicks = EndTime - StartTime;
   TotalIterations = 0;
   TotalBytesTransferred = 0;

   for (i = 0; i < dwNumberOfClients; i++)
   {
      TotalIterations += ClientData[i].IoBuffer.u.IAmDone.TotalIterations;
      TotalBytesTransferred += ClientData[i].IoBuffer.u.IAmDone.TotalBytesTransferred;

      //
      // find fastest client
      //
      if (LowestTicks > ClientData[i].IoBuffer.u.IAmDone.TotalTicks)
      {
         LowestTicks = ClientData[i].IoBuffer.u.IAmDone.TotalTicks;
      }

      //
      // find slowest client
      //
      if (HighestTicks < ClientData[i].IoBuffer.u.IAmDone.TotalTicks)
      {
         HighestTicks = ClientData[i].IoBuffer.u.IAmDone.TotalTicks;
      }
   }

   fprintf (stdout, "\nSOCKSRV:TPS, %ld (Fastest Client %dms Slowest Client %dms)\n",
            (TotalIterations * 1000) / TotalTicks,
            LowestTicks,
            HighestTicks
      );

   if (fVerbose)
   {
      fprintf (stdout, "\n%ld bytes transferred in %ld iterations, time = %ld ms\n",
               TotalBytesTransferred,
               TotalIterations,
               TotalTicks
         );

      for (i = 0; i < dwNumberOfWorkers; i++)
      {
         fprintf (stdout, "Thread[%2d] %d Transactions %d Bytes\n",
                  i,
                  ThreadData[i].TotalTransactions,
                  ThreadData[i].TotalBytesTransferred
            );
      }
   }
}

BOOL
WINAPI
CreateNetConnections (
                        void
)
{
   DWORD i;
   SOCKET listener;
   INT err;
   WSADATA WsaData;
   DWORD nbytes;
   BOOL b;

   err = WSAStartup (0x0101, &WsaData);
   if (err == SOCKET_ERROR)
   {
      fprintf (stdout, "WSAStartup Failed\n");
      return FALSE;
   }

   //
   // Open a socket to listen for incoming connections.
   //

   if (fTcp)
   {

      SOCKADDR_IN localAddr;

      listener = socket (AF_INET, SOCK_STREAM, 0);
      if (listener == INVALID_SOCKET)
      {
         fprintf (stdout, "Socket Create Failed\n");
         return FALSE;
      }

      //
      // Bind our server to the agreed upon port number.  See
      // commdef.h for the actual port number.
      //
      ZeroMemory (&localAddr, sizeof (localAddr));
      localAddr.sin_port = htons (SERVPORT);
      localAddr.sin_family = AF_INET;

      err = bind (listener, (PSOCKADDR) & localAddr, sizeof (localAddr));
      if (err == SOCKET_ERROR)
      {
         fprintf (stdout, "Socket Bind Failed\n");
         if (WSAGetLastError () == WSAEADDRINUSE)
            fprintf (stdout, "The port number may already be in use.\n");
         return FALSE;
      }

   }
   else
   {

      SOCKADDR_IPX localAddr;

      listener = socket (AF_IPX, SOCK_STREAM, NSPROTO_SPX);
      if (listener == INVALID_SOCKET)
      {
         fprintf (stdout, "Socket Create Failed\n");
         return FALSE;
      }

      ZeroMemory (&localAddr, sizeof (localAddr));
      localAddr.sa_socket = htons (7);
      localAddr.sa_family = AF_IPX;

      err = bind (listener, (PSOCKADDR) & localAddr, sizeof (localAddr));
      if (err == SOCKET_ERROR)
      {
         fprintf (stdout, "Socket Bind Failed\n");
         return FALSE;
      }
   }

   //
   // Prepare to accept client connections.  Allow up to 5 pending
   // connections.
   //
   err = listen (listener, 5);
   if (err == SOCKET_ERROR)
   {
      fprintf (stdout, "Socket Listen Failed\n");
      return FALSE;
   }


   //
   // Only Handle a single Queue
   //

   for (i = 0; i < dwNumberOfClients; i++)
   {

      //
      // Accept incoming connect requests and create wait events for each.
      //

      //
      // If we are doing file I/O, each client will need its own event
      // to use for async file I/O
      //

      if (hFile)
      {
         ClientData[i].hEvent = CreateEvent (NULL, TRUE, FALSE, NULL);
         if (!ClientData[i].hEvent)
         {
            fprintf (stdout, "Client Event Create Failed\n");
            return FALSE;
         }
      }

      ClientData[i].Socket = accept (listener, NULL, NULL);
      if (ClientData[i].Socket == INVALID_SOCKET)
      {
         fprintf (stdout, "Accept Failed\n");
         return FALSE;
      }

      if (fVerbose)
      {
         fprintf (stdout, "Client Connection Accepted\n");
      }

      //
      // note the 16 says how many concurrent cpu bound threads to allow thru
      // this should be tunable based on the requests. CPU bound requests will
      // really really honor this.
      //

      CompletionPort = CreateIoCompletionPort (
                                              (HANDLE) ClientData[i].Socket,
                                                 CompletionPort,
                                                 (DWORD) i,
                                                 dwConcurrency
         );

      if (!CompletionPort)
      {
         fprintf (stdout, "CompletionPort Create Failed\n");
         return FALSE;
      }

      ClientData[i].Flags = CLIENT_CONNECTED;

      //
      // Start off an asynchronous read on the socket.
      //

      ClientData[i].Overlapped.hEvent = NULL;

      b = ReadFile (
                      (HANDLE) ClientData[i].Socket,
                      &ClientData[i].IoBuffer,
                      sizeof (CLIENT_IO_BUFFER),
                      &nbytes,
                      &ClientData[i].Overlapped
         );

      if (!b && GetLastError () != ERROR_IO_PENDING)
      {
         fprintf (stdout, "ReadFile Failed\n");
         return FALSE;
      }
   }

   dwActiveClientCount = dwNumberOfClients;
   hBenchmarkComplete = CreateEvent (NULL, TRUE, FALSE, NULL);
   if (!hBenchmarkComplete)
   {
      fprintf (stdout, "Create Benchmark Complete Event Failed\n");
      return FALSE;
   }
   hBenchmarkStart = CreateEvent (NULL, TRUE, FALSE, NULL);
   if (!hBenchmarkStart)
   {
      fprintf (stdout, "Create Benchmark Start Event Failed\n");
      return FALSE;
   }

   return TRUE;
}

BOOL
WINAPI
CreateWorkers (
                 void
)
{
   DWORD ThreadId;
   HANDLE ThreadHandle;
   DWORD i;

   for (i = 0; i < dwNumberOfWorkers; i++)
   {

      ThreadHandle = CreateThread (
                                     NULL,
                                     0,
                                     WorkerThread,
                                     &ThreadData[i],
                                     0,
                                     &ThreadId
         );
      if (!ThreadHandle)
      {
         fprintf (stdout, "Create Worker Thread Failed\n");
         return FALSE;
      }

      CloseHandle (ThreadHandle);
   }

   return TRUE;
}

DWORD
WINAPI
WorkerThread (
                LPVOID WorkContext
)
{
   PPER_THREAD_DATA Me;
   DWORD WorkIndex;
   INT err;
   DWORD ResponseLength;
   BOOL b;
   LPOVERLAPPED lpo;
   DWORD nbytes;
   PPER_CLIENT_DATA CurrentClient;
   CHAR ReadBuffer[CLIENT_OUTBOUND_BUFFER_MAX];

   WaitForSingleObject (hBenchmarkStart, INFINITE);

   Me = (PPER_THREAD_DATA) WorkContext;

   for (;;)
   {

      b = GetQueuedCompletionStatus (
                                       CompletionPort,
                                       &nbytes,
                                       &WorkIndex,
                                       &lpo,
                                       INFINITE
         );

      if (b || lpo)
      {


         if (b)
         {

            CurrentClient = &ClientData[WorkIndex];

            switch (CurrentClient->IoBuffer.MessageType)
            {
            case CLIENT_IO_MT_RETURN_DATA:

               //
               // Determine how long a response was desired by the client.
               //

               ResponseLength = CurrentClient->IoBuffer.u.ReturnData.ByteCount;
               if (ResponseLength > CLIENT_OUTBOUND_BUFFER_MAX)
               {
                  ResponseLength = CLIENT_OUTBOUND_BUFFER_MAX;
               }

               //
               // If we are running in I/O mode, do a random read
               // Otherwise, use fill memory to supply the data
               //
               if (hFile)
               {


                  CurrentClient->Overlapped.Offset = Random (FILE_SIZE / CLIENT_OUTBOUND_BUFFER_MAX) * CLIENT_OUTBOUND_BUFFER_MAX;
                  CurrentClient->Overlapped.hEvent = CurrentClient->hEvent;
                  b = ReadFile (
                                  hFile,
                                  ReadBuffer,
                                  ResponseLength,
                                  &nbytes,
                                  &CurrentClient->Overlapped
                     );
                  if (!b && GetLastError () != ERROR_IO_PENDING)
                  {
                     fprintf (stderr, "SOCKSRV: Error in client read %d\n", GetLastError ());
                     exit (1);
                  }
                  b = GetOverlappedResult (
                                             hFile,
                                             &CurrentClient->Overlapped,
                                             &nbytes,
                                             TRUE
                     );
                  if (!b)
                  {
                     fprintf (stderr, "SOCKSRV: Wait for pre-write failed %d\n", GetLastError ());
                     exit (1);
                  }
                  CurrentClient->Overlapped.hEvent = NULL;
               }
               else
               {
                  FillMemory (CurrentClient->OutboundBuffer, ResponseLength, 0xfe);
               }

               //
               // Simulate a small compute bound workload
               //
               SortTheBuffer ((LPDWORD) CurrentClient->OutboundBuffer, (LPDWORD) ReadBuffer, nbytes >> 2);

               //
               // Send a response and post another asynchronous read on the
               // socket.
               //

               err = send (CurrentClient->Socket,
                           CurrentClient->OutboundBuffer,
                           ResponseLength,
                           0
                  );

               if (err == SOCKET_ERROR)
               {
                  fprintf (stdout, "Send Failed\n");
                  exit (1);
               }

               Me->TotalTransactions++;
               Me->TotalBytesTransferred += ResponseLength;

               //
               // reprime this client by posting another asynch read
               //

               b = ReadFile (
                               (HANDLE) CurrentClient->Socket,
                               &CurrentClient->IoBuffer,
                               sizeof (CLIENT_IO_BUFFER),
                               &nbytes,
                               &CurrentClient->Overlapped
                  );

               if (!b && GetLastError () != ERROR_IO_PENDING)
               {
                  fprintf (stdout, "ReadFile Failed\n");
                  exit (1);
               }
               break;

            case CLIENT_IO_MT_I_AM_DONE:
               CurrentClient->Flags |= CLIENT_DONE;
               if (fVerbose)
               {
                  fprintf (stdout, "Client Has Completed\n");
               }
               if (!InterlockedDecrement (&dwActiveClientCount))
               {
                  SetEvent (hBenchmarkComplete);
               }
               break;

            default:
               fprintf (stdout, "Invalid MessageType %x\n", CurrentClient->IoBuffer.MessageType);
               exit (1);
            }
         }
      }
      else
      {
         fprintf (stdout, "WorkThread Wait Failed\n");
         exit (1);
      }
   }
   return 1;
}

VOID
WINAPI
ShowUsage (
             VOID
)
{
   fputs ("usage: SOCKSRV [switches]\n"
          "               [-?] show this message\n"
          "               [-v] verbose output\n"
          "               [-t number-of-threads] specify the number of worker threads\n"
     "               [-c number-of-clients] specify the number of clients\n"
          "               [-p concurrency-value] specify the concurrency\n"
          "               [-i ] use IPX instead of TCP\n"
          "               [-w work-index] specify how much compute to do\n"
          ,stderr);

   exit (1);
}


VOID
WINAPI
ParseSwitch (
               CHAR chSwitch,
               int *pArgc,
               char **pArgv[]
)
{

   switch (toupper (chSwitch))
   {

   case '?':
      ShowUsage ();
      break;

   case 'T':
      if (!--(*pArgc))
      {
         ShowUsage ();
      }
      (*pArgv)++;
      dwNumberOfWorkers = strtoul (*(*pArgv), NULL, 10);
      if (dwNumberOfWorkers > MAXIMUM_NUMBER_OF_WORKERS)
      {
         dwNumberOfWorkers = MAXIMUM_NUMBER_OF_WORKERS;
      }
      break;

   case 'C':
      if (!--(*pArgc))
      {
         ShowUsage ();
      }
      (*pArgv)++;
      dwNumberOfClients = strtoul (*(*pArgv), NULL, 10);
      if (dwNumberOfClients > MAXIMUM_NUMBER_OF_CLIENTS)
      {
         dwNumberOfClients = MAXIMUM_NUMBER_OF_CLIENTS;
      }
      break;

   case 'P':
      if (!--(*pArgc))
      {
         ShowUsage ();
      }
      (*pArgv)++;
      dwConcurrency = strtoul (*(*pArgv), NULL, 10);
      break;

   case 'W':
      if (!--(*pArgc))
      {
         ShowUsage ();
      }
      (*pArgv)++;
      dwWorkIndex = strtoul (*(*pArgv), NULL, 10);
      break;

   case 'V':
      fVerbose = TRUE;
      break;

   case 'I':
      fTcp = FALSE;
      break;

   default:
      fprintf (stderr, "SOCKSRV: Invalid switch - /%c\n", chSwitch);
      ShowUsage ();
      break;

   }
}

DWORD
WINAPI
Random (
          DWORD nMaxValue
)
{
   return (((2 * rand () * nMaxValue + RAND_MAX) / RAND_MAX - 1) / 2);
}

int _CRTAPI1
DwordComp (const void *e1, const void *e2)
{
   PULONG p1;
   PULONG p2;

   p1 = (PULONG) e1;
   p2 = (PULONG) e2;

   return (*p1 - *p2);
}

VOID
WINAPI
SortTheBuffer (
                 LPDWORD Destination,
                 LPDWORD Source,
                 int DwordCount
)

{
   DWORD i;

   for (i = 0; i < 2 * dwWorkIndex; i++)
   {
      CopyMemory (Destination, Source, DwordCount << 2);
      qsort ((void *) Destination, (size_t) DwordCount, (size_t) sizeof (DWORD), DwordComp);
   }
}