Windows에서 Coroutine 구현에 필요한 함수

Windows에는

LPVOID WINAPI CreateFiber(
  __in      SIZE_T dwStackSize,
  __in      LPFIBER_START_ROUTINE lpStartAddress,
  __in_opt  LPVOID lpParameter
);

계열의 함수가 존재한다.

MSDN의 CreateFiber()를 사용한 예제이다.

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

VOID
__stdcall
ReadFiberFunc(LPVOID lpParameter);

VOID
__stdcall
WriteFiberFunc(LPVOID lpParameter);

void DisplayFiberInfo(void);

typedef struct
{
   DWORD dwParameter;          // DWORD parameter to fiber (unused)
   DWORD dwFiberResultCode;    // GetLastError() result code
   HANDLE hFile;               // handle to operate on
   DWORD dwBytesProcessed;     // number of bytes processed
} FIBERDATASTRUCT, *PFIBERDATASTRUCT, *LPFIBERDATASTRUCT;

#define RTN_OK 0
#define RTN_USAGE 1
#define RTN_ERROR 13

#define BUFFER_SIZE 32768   // read/write buffer size
#define FIBER_COUNT 3       // max fibers (including primary)

#define PRIMARY_FIBER 0 // array index to primary fiber
#define READ_FIBER 1    // array index to read fiber
#define WRITE_FIBER 2   // array index to write fiber

LPVOID g_lpFiber[FIBER_COUNT];
LPBYTE g_lpBuffer;
DWORD g_dwBytesRead;

int __cdecl _tmain(int argc, TCHAR *argv[])
{
   LPFIBERDATASTRUCT fs;

   if (argc != 3)
   {
      printf("Usage: %s <SourceFile> <DestinationFile>\n", argv[0]);
      return RTN_USAGE;
   }

   //
   // Allocate storage for our fiber data structures
   //
   fs = (LPFIBERDATASTRUCT)HeapAlloc(
                              GetProcessHeap(), 0,
                              sizeof(FIBERDATASTRUCT) * FIBER_COUNT);

   if (fs == NULL)
   {
      printf("HeapAlloc error (%d)\n", GetLastError());
      return RTN_ERROR;
   }

   //
   // Allocate storage for the read/write buffer
   //
   g_lpBuffer = (LPBYTE)HeapAlloc(GetProcessHeap(), 0, BUFFER_SIZE);
   if (g_lpBuffer == NULL)
   {
      printf("HeapAlloc error (%d)\n", GetLastError());
      return RTN_ERROR;
   }

   //
   // Open the source file
   //
   fs[READ_FIBER].hFile = CreateFile(
                                    argv[1],
                                    GENERIC_READ,
                                    FILE_SHARE_READ,
                                    NULL,
                                    OPEN_EXISTING,
                                    FILE_FLAG_SEQUENTIAL_SCAN,
                                    NULL
                                    );

   if (fs[READ_FIBER].hFile == INVALID_HANDLE_VALUE)
   {
      printf("CreateFile error (%d)\n", GetLastError());
      return RTN_ERROR;
   }

   //
   // Open the destination file
   //
   fs[WRITE_FIBER].hFile = CreateFile(
                                     argv[2],
                                     GENERIC_WRITE,
                                     0,
                                     NULL,
                                     CREATE_NEW,
                                     FILE_FLAG_SEQUENTIAL_SCAN,
                                     NULL
                                     );

   if (fs[WRITE_FIBER].hFile == INVALID_HANDLE_VALUE)
   {
      printf("CreateFile error (%d)\n", GetLastError());
      return RTN_ERROR;
   }

   //
   // Convert thread to a fiber, to allow scheduling other fibers
   //
   g_lpFiber[PRIMARY_FIBER]=ConvertThreadToFiber(&fs[PRIMARY_FIBER]);

   if (g_lpFiber[PRIMARY_FIBER] == NULL)
   {
      printf("ConvertThreadToFiber error (%d)\n", GetLastError());
      return RTN_ERROR;
   }

   //
   // Initialize the primary fiber data structure.  We don't use
   // the primary fiber data structure for anything in this sample.
   //
   fs[PRIMARY_FIBER].dwParameter = 0;
   fs[PRIMARY_FIBER].dwFiberResultCode = 0;
   fs[PRIMARY_FIBER].hFile = INVALID_HANDLE_VALUE;

   //
   // Create the Read fiber
   //
   g_lpFiber[READ_FIBER]=CreateFiber(0,ReadFiberFunc,&fs[READ_FIBER]);

   if (g_lpFiber[READ_FIBER] == NULL)
   {
      printf("CreateFiber error (%d)\n", GetLastError());
      return RTN_ERROR;
   }

   fs[READ_FIBER].dwParameter = 0x12345678;

   //
   // Create the Write fiber
   //
   g_lpFiber[WRITE_FIBER]=CreateFiber(0,WriteFiberFunc,&fs[WRITE_FIBER]);

   if (g_lpFiber[WRITE_FIBER] == NULL)
   {
      printf("CreateFiber error (%d)\n", GetLastError());
      return RTN_ERROR;
   }

   fs[WRITE_FIBER].dwParameter = 0x54545454;

   //
   // Switch to the read fiber
   //
   SwitchToFiber(g_lpFiber[READ_FIBER]);

   //
   // We have been scheduled again. Display results from the
   // read/write fibers
   //
   printf("ReadFiber: result code is %lu, %lu bytes processed\n",
   fs[READ_FIBER].dwFiberResultCode, fs[READ_FIBER].dwBytesProcessed);

   printf("WriteFiber: result code is %lu, %lu bytes processedn",
   fs[WRITE_FIBER].dwFiberResultCode, fs[WRITE_FIBER].dwBytesProcessed);

   //
   // Delete the fibers
   //
   DeleteFiber(g_lpFiber[READ_FIBER]);
   DeleteFiber(g_lpFiber[WRITE_FIBER]);

   //
   // Close handles
   //
   CloseHandle(fs[READ_FIBER].hFile);
   CloseHandle(fs[WRITE_FIBER].hFile);

   //
   // Free allocated memory
   //
   HeapFree(GetProcessHeap(), 0, g_lpBuffer);
   HeapFree(GetProcessHeap(), 0, fs);

   return RTN_OK;
}

VOID
__stdcall
ReadFiberFunc(
             LPVOID lpParameter
             )
{
   LPFIBERDATASTRUCT fds = (LPFIBERDATASTRUCT)lpParameter;

   //
   // If this fiber was passed NULL for fiber data, just return,
   // causing the current thread to exit
   //
   if (fds == NULL)
   {
      printf("Passed NULL fiber data; exiting current thread.\n");
      return;
   }

   //
   // Display some information pertaining to the current fiber
   //
   DisplayFiberInfo();

   fds->dwBytesProcessed = 0;

   while (1)
   {
      //
      // Read data from file specified in the READ_FIBER structure
      //
      if (!ReadFile(fds->hFile, g_lpBuffer, BUFFER_SIZE,
         &g_dwBytesRead, NULL))
      {
         break;
      }

      //
      // if we reached EOF, break
      //
      if (g_dwBytesRead == 0) break;

      //
      // Update number of bytes processed in the fiber data structure
      //
      fds->dwBytesProcessed += g_dwBytesRead;

      //
      // Switch to the write fiber
      //
      SwitchToFiber(g_lpFiber[WRITE_FIBER]);
   } // while

   //
   // Update the fiber result code
   //
   fds->dwFiberResultCode = GetLastError();

   //
   // Switch back to the primary fiber
   //
   SwitchToFiber(g_lpFiber[PRIMARY_FIBER]);
}

VOID
__stdcall
WriteFiberFunc(
              LPVOID lpParameter
              )
{
   LPFIBERDATASTRUCT fds = (LPFIBERDATASTRUCT)lpParameter;
   DWORD dwBytesWritten;

   //
   // If this fiber was passed NULL for fiber data, just return,
   // causing the current thread to exit
   //
   if (fds == NULL)
   {
      printf("Passed NULL fiber data; exiting current thread.\n");
      return;
   }

   //
   // Display some information pertaining to the current fiber
   //
   DisplayFiberInfo();

   //
   // Assume all writes succeeded.  If a write fails, the fiber
   // result code will be updated to reflect the reason for failure
   //
   fds->dwBytesProcessed = 0;
   fds->dwFiberResultCode = ERROR_SUCCESS;

   while (1)
   {
      //
      // Write data to the file specified in the WRITE_FIBER structure
      //
      if (!WriteFile(fds->hFile, g_lpBuffer, g_dwBytesRead,
         &dwBytesWritten, NULL))
      {
         //
         // If an error occurred writing, break
         //
         break;
      }

      //
      // Update number of bytes processed in the fiber data structure
      //
      fds->dwBytesProcessed += dwBytesWritten;

      //
      // Switch back to the read fiber
      //
      SwitchToFiber(g_lpFiber[READ_FIBER]);
   }  // while

   //
   // If an error occurred, update the fiber result code...
   //
   fds->dwFiberResultCode = GetLastError();

   //
   // ...and switch to the primary fiber
   //
   SwitchToFiber(g_lpFiber[PRIMARY_FIBER]);
}

void
DisplayFiberInfo(
                void
                )
{
   LPFIBERDATASTRUCT fds = (LPFIBERDATASTRUCT)GetFiberData();
   LPVOID lpCurrentFiber = GetCurrentFiber();

   //
   // Determine which fiber is executing, based on the fiber address
   //
   if (lpCurrentFiber == g_lpFiber[READ_FIBER])
      printf("Read fiber entered");
   else
   {
      if (lpCurrentFiber == g_lpFiber[WRITE_FIBER])
         printf("Write fiber entered");
      else
      {
         if (lpCurrentFiber == g_lpFiber[PRIMARY_FIBER])
            printf("Primary fiber entered");
         else
            printf("Unknown fiber entered");
      }
   }

   //
   // Display dwParameter from the current fiber data structure
   //
   printf(" (dwParameter is 0x%lx)\n", fds->dwParameter);

}

CreateFiber() 함수를 사용해, coroutine을 구현해 볼수 있지만...

여기서는 다른 방법으로 Coroutine을 구현하는 방법을 설명하도록 하겠다.

Windows에는 프로그램의 Context를 제어하기 위한 함수가 있다.

BOOL GetThreadContext(
  HANDLE hThread,  LPCONTEXT lpContext);

과

BOOL SetThreadContext(
  HANDLE hThread,  const CONTEXT* lpContext);

이 존재한다. 함수명만 봐서는 쓰레드에서 사용될것 같은 함수라고 다들 짐작을 하겠지만 실제로는 더 많은
처리를 할수 있다. == 여기에서 CONTEXT는 프로그램의 실행정보가 포함되어 있다(즉, REG값을 보관한다)

해당 함수를 통해 구현에 필요한 기본적인 처리를 해보자.

1. 현재 구동되고 있는 위치(쓰레드)의 CONTEXT를 저장할 필요가 있다.
 

ctx.ContextFlags = CONTEXT_FULL;
GetThreadContext( GetCurrentThread(), &ctx);

   이렇게 하여, 현재 프로세서(쓰레드) 위치의 REG정보를 저장해 둔다.

2. 구동에 필요한 CONTEXT를 생성한다.

#define PAGE_SIZE   16384
#define PAGE_ALIGN(x)  (((x) + (PAGE_SIZE - 1)) & ~(PAGE_SIZE - 1))

size_t mstack = PAGE_ALIGN(STACK_SIZE + 1) + sizeof(CONTEXT);
if (!(stack = (void *)GlobalAlloc( GPTR, mstack))) return NULL;

// STACK설정이 가능하도록 메모리 정렬을 한다.
u = (CONTEXT *)((DWORD_PTR)
        ((struct CONTEXT *)(((char *)stack) + mstack)) & ~(sizeof(DWORD_PTR) - 1));
(--u)->ContextFlags = CONTEXT_FULL;
if (GetThreadContext( GetCurrentThread(), u) == FALSE)
   ;
else {
    stack = ((char *)(u - 1)) - sizeof(__int64); {
     u->Eip = (DWORD)((DWORD_PTR)CALL_FUNC); // CALL_FUNC( void *PARAM)
     u->Esp = (DWORD)((DWORD_PTR)(((char *)stack) - sizeof(DWORD)));
     u->ContextFlags = CONTEXT_FULL;
     *((__int64 *)stack) = (__int64)PARAM;
     return u;
    }
} return NULL;

   여기에서 중요한 것은, STACK의 위치이다. 소스코드를 보면 알겠지만 스택의 위치를 메모리 할당 시작
   위치로 하지 않고, 마지막 위치로 설정하는 것을 알수 있을 것이다.

   이는 스텍의 구조를 생각해 보면 왜 그렇게 하는지 쉽게 이해할수 있다.

   참고로, x86_64에서는 EIP나 ESP대신 다른 REG에 메모리를 설정해야 할 필요가 있다.
   64bit로 아직 서버를 만들지 않아 해당 루틴을 수정하지는 않았다.

3. 이렇게 만들어진 CONTEXT간의 전환을 구현해 보자

 SELF_CTX.ContextFlags = CONTEXT_FULL;
 if (!GetThreadContext( GetCurrentThread(), SELF_CTX) ||
       !SetThreadContext( GetCurrentThread(), TARGET_CTX))
   FATAL( "context switch failed");

이러한 방식으로, 하여 Fiber와 같은 역활을 담당하는 Coroutine을 만들어 낼수 있다.

위에 설명한 부분은 필자가 설명을 하기 위해 실제 소스에서 일부 루틴을 가지고 설명한 것이고, 실제로 위의
방식으로 필자는 다양한 방식의 처리를 한다.

예를 들어,
   1. 일정 시간 이후 해당 함수로 진입 하도록 처리 - sleep()과 같은 처리
   2. 생성된 coroutine간의 스케쥴링 처리
      - 스케쥴링 처리에 특정 coroutine을 제외 하는 기능
   3. 쓰레드별 독립적인 coroutine 처리
   4. coroutine의 순환 반복 처리
      - 일반 함수와 같이 함수의 return을 통해 호출한 타 coroutine으로 값을 전달하는 기능
      - 설정된 coroutine을 재 호출 처리 하는 기능
   5. 물론, coroutine의 resume 및 suspend기능도 존재
   6. 현재 구동 중인 coroutine을 제거하는 기능
   7. coroutine별 독립 param 전달 처리

등, 수많은 기능을 구현하기 위해 Fiber를 사용하지 않고, 이렇게 구현을 하였으며 물론, 기존에 가지고 있던 LINUX버젼과의 API 호환성 문제를 해결하기 위해서도 필요했다.

다음에 더... 알아보도록 하겠다~