Socket IO重叠模型(完成例程)

Server:

{*******************************************************}
{                                                       }
{       Overlap IO Server                               }
{       Creation Date 2010.03.18                        }
{       Created By: ming                                }
{                                                       }
{*******************************************************}
unit unitWorkThread;

interface

uses
  Windows, Messages, Forms, SysUtils, Classes, StdCtrls, unitWinsock2;

const
  WM_ACTION = WM_USER + 100;

type
  TAcceptThread = class(TThread)
  private
    FEvent: HWND;
    FMemo: TMemo;
    FLogMsg: string;
  protected
    procedure Execute;override;
  public
    procedure doLogMsg(const msg: String);
    procedure syncLogMsg;
    procedure exitThread;
    constructor Create(Memo: TMemo);
    destructor Destroy; override;
  end;
  //
  
const
  DATA_BUFFSIZE = 1024;
var
  gStartupFlag: Integer = -1;
  AcceptThread: TAcceptThread;
  //
  ListenPort: DWORD = 61000;
  ListenSocket: TSocket;
  ServerAddr: TSockAddr;
  ClientAddr: TSockAddr;
  AcceptSocket: array [0..WSA_MAXIMUM_WAIT_EVENTS-1] of TSocket;
  AcceptOverlapper: array [0..WSA_MAXIMUM_WAIT_EVENTS-1] of TOVERLAPPED;
  EventArray: array [0..WSA_MAXIMUM_WAIT_EVENTS-1] of WSAEVENT;
  DataBuf: array [0..DATA_BUFFSIZE-1] of TWSABUF;
  Buffer: array [0..DATA_BUFFSIZE-1,0..DATA_BUFFSIZE-1] of Char;
  RetMsg: array [0..DATA_BUFFSIZE-1] of Char;
  dwRecvBytes,dwSendBytes,dwCount,dwFlag: DWORD;
  dwEventTotal: Integer=0;
  //
  MainForm: HWND;

implementation

procedure SyncAddLog(const msg:string);
begin
  SendMessage(MainForm,WM_ACTION,WParam(PChar(msg)),0);
end;

function FmtErrMsg(const errMsg:string; const errCode:Integer=0):string;
begin
  Result := Format('ErrMsg:%s,ErrCode:%d',[errMsg,errCode]);
end;

procedure showErrMsg(const errMsg:string; const errCode:Integer=0);
var
  szMsg: string;
begin
  szMsg := Format('ErrMsg:%s,ErrCode:%d',[errMsg,errCode]);
  MessageBox(0,PChar(szMsg),'Error',0);
end;

procedure _ReleaseIO(const idx:Integer);
var
  i: Integer;
begin
  i := idx;
  if AcceptSocket[i] <> 0 then
    closesocket(AcceptSocket[i]);
  AcceptSocket[i] := 0;
  WSACloseEvent(EventArray[i]);
  AcceptOverlapper[i].hEvent := 0;
  EventArray[i] := 0;
  DataBuf[i].len := 0;
  DataBuf[i].buf := nil;
  InterlockedDecrement(dwEventTotal);
end;

procedure SendCompletionRoutine(
  error,BytesTransferred: DWORD;
  Overlapped: POverlapped;
  InFlags: DWORD); stdcall;
  function getIndex:Integer;
  var k: Integer;
  begin
    for k := 0 to WSA_MAXIMUM_WAIT_EVENTS-1 do
    begin
      if EventArray[k] = Overlapped^.hEvent then
        Break;
    end;
    Result := k;
  end;
var
  i: Integer;
begin
  i := getIndex;
//  if (error<>0) or (BytesTransferred=0) then
//    SyncAddLog(Format('%d Send Error,BytesTransferred=%d',[i,BytesTransferred]))
//  else
//    SyncAddLog(Format('%d Successful,BytesTransferred=%d',[i,BytesTransferred]));
  _ReleaseIO(i);
end;

procedure returnMsg(idx:integer; const msg: string);
var
  len,errCode: Integer;
  szText: string;
begin
  szText := 'Msg Has Received.';
  len := Length(szText);
  ZeroMemory(@RetMsg,DATA_BUFFSIZE);
  CopyMemory(@RetMsg,@szText[1],Length(szText));
  DataBuf[idx].Buf := @RetMsg;
  DataBuf[idx].len := len;
  if (WSASend(AcceptSocket[idx],@DataBuf[idx],1,@dwSendBytes,0
    ,@AcceptOverlapper[idx],@SendCompletionRoutine))=SOCKET_ERROR then
  begin
    errCode := WSAGetLastError;
    if errCode <> WSA_IO_PENDING then
    begin
      showErrMsg('WSAGetLastError Error!',errCode);
    end;
  end;
end;

procedure RecvCompletionRoutine(
  error,BytesTransferred: DWORD;
  Overlapped: POverlapped;
  InFlags: DWORD); stdcall;
  function getIndex:Integer;
  var k: Integer;
  begin
    for k := 0 to WSA_MAXIMUM_WAIT_EVENTS-1 do
    begin
      if EventArray[k] = Overlapped^.hEvent then
        Break;
    end;
    Result := k;
  end;
var
  szText: string;
  i: Integer;
begin
  i := getIndex;
  if (error<>0) or (BytesTransferred=0) then
  begin
    SyncAddLog(Format('%d Recv Error,BytesTransferred=%d',[i,BytesTransferred]));
    _ReleaseIO(i);
  end
  else
  begin
    szText := StrPas(DataBuf[i].buf);
    SyncAddLog(szText);
    returnMsg(i,'');
  end;
end;

function StartUpSocket: Integer;
var
  wsaData: TWSAData;
  err: Integer;
begin
  Result := -1;
  err := WSAStartup(MakeWord(2,2),wsaData);
  if err <> 0 then
  begin
    showErrMsg('WSAStartup Error!');
    Exit;
  end;
  if (Lo(wsaData.wVersion)<>2) or (Hi(wsaData.wVersion)<>2) then
  begin
    showErrMsg('Socket Version Error!');
    Exit;
  end;
  Result := 0;
end;

function SocketListen:Integer;
var
  len: Integer;
begin
  Result := -1;
  if gStartupFlag <> 0 then Exit;

  ListenSocket := WSASocket(AF_INET,SOCK_STREAM,IPPROTO_TCP,nil,0,WSA_FLAG_OVERLAPPED);

  if ListenSocket = INVALID_SOCKET then
  begin
    showErrMsg('Create ListenSocket Error!');
    Exit;
  end;

  ServerAddr.sin_family := AF_INET;
  ServerAddr.sin_addr.S_addr := htonl(INADDR_ANY);
  ServerAddr.sin_port := htons(ListenPort);
  len := SizeOf(ServerAddr);
  
  if bind(ListenSocket,PSockaddr(@ServerAddr),len)=SOCKET_ERROR then
  begin
    showErrMsg('bind Error!',WSAGetLastError);
    Exit;
  end;
  if listen(ListenSocket,5)=SOCKET_ERROR then
  begin
    showErrMsg('listen Error!',WSAGetLastError);
    Exit;
  end;
  Result := 0;
end;

{ TAcceptThread }

constructor TAcceptThread.Create(Memo: TMemo);
begin
  inherited Create(False);
  FreeOnTerminate := True;
  FMemo := Memo;
  FEvent := CreateEvent(nil,False,False,nil);
end;

destructor TAcceptThread.Destroy;
begin
  CloseHandle(FEvent);
  inherited;
end;

procedure TAcceptThread.Execute;
var
  i,len,errCode: Integer;
  tempSocket: TSocket;
  function getIndex:Integer;
  var k: Integer;
  begin
    for k := 0 to WSA_MAXIMUM_WAIT_EVENTS-1 do
    begin
      if EventArray[k] = 0 then
        Break;
    end;
    Result := k;
  end;
begin
  inherited;
  len := SizeOf(ClientAddr);
  if SocketListen <> 0 then Exit;
  while not Terminated do
  begin
    if WaitForSingleObject(FEvent,100)=WAIT_OBJECT_0 then
      Break;
    
    tempSocket := accept(ListenSocket,PSockaddr(@ClientAddr),len);
    if tempSocket = INVALID_SOCKET then Continue;
    i := getIndex;
    AcceptSocket[i] := tempSocket;

    EventArray[i] := WSACreateEvent;
    ZeroMemory(@AcceptOverlapper[i],SizeOf(OVERLAPPED));
    AcceptOverlapper[i].hEvent := EventArray[i];
    ZeroMemory(@Buffer[i],DATA_BUFFSIZE);
    DataBuf[i].len := DATA_BUFFSIZE;
    DataBuf[i].buf := @Buffer[i];
    InterlockedIncrement(dwEventTotal);
    if (WSARecv(AcceptSocket[i],@DataBuf[i],1,@dwRecvBytes,@dwFlag
      ,@AcceptOverlapper[i],@RecvCompletionRoutine))=SOCKET_ERROR then
    begin
      errCode := WSAGetLastError;
      if errCode <> WSA_IO_PENDING then
      begin
        //showErrMsg('WSAGetLastError Error!',errCode);
        doLogMsg(FmtErrMsg('WSAGetLastError Error!',errCode));
        _ReleaseIO(i);
        Continue;
      end;
    end;
    //if WSAWaitForMultipleEvents(1,@EventArray,FALSE,WSA_INFINITE,True)=WAIT_IO_COMPLETION then
    if SleepEx(2000,True)=WAIT_IO_COMPLETION then
    begin
      doLogMsg('WAIT_IO_COMPLETION');
    end else
    begin
      _ReleaseIO(i);
      doLogMsg('Not WAIT_IO_COMPLETION');
    end;
  end;
end;

procedure TAcceptThread.exitThread;
begin
  if ListenSocket <> INVALID_SOCKET then
  begin
    shutdown(ListenSocket,SD_BOTH);
    closesocket(ListenSocket);
  end;
  SetEvent(FEvent);
end;

procedure TAcceptThread.doLogMsg(const msg: String);
begin
  FLogMsg := msg;
  Synchronize(syncLogMsg);
end;

procedure TAcceptThread.syncLogMsg;
begin
  FMemo.Lines.Add(FLogMsg);
end;

initialization
  gStartupFlag := StartupSocket;
finalization
  if gStartupFlag = 0 then
    WSACleanup;

end.

//Main form
unit unitMain;

interface

uses
  Windows, Messages, SysUtils, Variants, Classes, Graphics, Controls, Forms,
  Dialogs, StdCtrls, unitWorkThread, ExtCtrls;

type
  TForm1 = class(TForm)
    mmoLog: TMemo;
    Button1: TButton;
    btnSetPort: TButton;
    lbledtPort: TLabeledEdit;
    procedure FormCreate(Sender: TObject);
    procedure mmoLogDblClick(Sender: TObject);
    procedure btnSetPortClick(Sender: TObject);
    procedure FormDestroy(Sender: TObject);
  private
    { Private declarations }
    procedure onMyAction(var msg: TMessage);message WM_ACTION;
  public
    { Public declarations }
  end;

var
  Form1: TForm1;

implementation

{$R *.dfm}

procedure TForm1.btnSetPortClick(Sender: TObject);
begin
  ListenPort := StrToInt(lbledtPort.Text);
end;

procedure TForm1.FormCreate(Sender: TObject);
begin
  if gStartupFlag = 0 then
  begin
    AcceptThread := TAcceptThread.Create(mmoLog);
    MainForm := Self.Handle;
  end;
end;

procedure TForm1.FormDestroy(Sender: TObject);
begin
  if Assigned(AcceptThread) then  
    AcceptThread.exitThread;
end;

procedure TForm1.mmoLogDblClick(Sender: TObject);
begin
  TMemo(Sender).Clear;
end;

procedure TForm1.onMyAction(var msg: TMessage);
begin
  mmoLog.Lines.Add(StrPas(PChar(msg.WParam)));
end;

end.

Client:

{*******************************************************}
{                                                       }
{       Overlap IO Client                               }
{       Creation Date 2010.03.18                        }
{       Created By: ming                                }
{                                                       }
{*******************************************************}
unit unitWorkThread;

interface

uses
  Windows, Messages, SysUtils, Classes, StdCtrls, unitWinsock2;

const
  WM_ACTION = WM_USER + 100;
  DATA_BUFFSIZE = 1024;

type
  //
  TClientThread = class(TThread)
  private
    FMemo: TMemo;
    FEvent: HWND;
    FClientID: Integer;
    FLogMsg: String;
    //
    FRemoteIP:string;
    FRemotePort:DWORD;
    //
    FClientSocket: TSocket;
    FServerAddr: TSockAddrIn;
    FOverlapper: TOverlapped;
    FDataBuf: TWSABUF;
    FEventArray: array [0..1] of WSAEVENT;
    FBuf: array [0..DATA_BUFFSIZE-1] of AnsiChar;
    FTransBytes,FTransFlag: DWORD;
    function ConnectServer:Integer;
    procedure doLogMsg(const msg: String);
    procedure syncLogMsg;
  protected
    procedure Execute;override;
  public
    procedure _SetEvent;
    function SendMsg(const msg:string=''):Integer;
    function RecvMsg(const msg:string=''):Integer;
    constructor Create(Memo: TMemo; ID:Integer; const IP:string; port:DWORD);
    destructor Destroy; override;
  end;

const
  K_ClientCount = 100;
var
  MainFormHandle: HWND=0;
  gStartupFlag: Integer = -1;
  ClientThread: TClientThread;
  MsgArr: array [1..K_ClientCount] of string;

implementation

procedure showErrMsg(const errMsg:string; const errCode:Integer=0);
var
  szMsg: string;
begin
  szMsg := Format('ErrMsg:%s,ErrCode:%d',[errMsg,errCode]);
  MessageBox(0,PChar(szMsg),'Error',0);
end;

function StartupSocket: Integer;
var
  wsaData: TWSAData;
  err: Integer;
begin
  Result := -1;
  err := WSAStartup(MakeWord(2,2),wsaData);
  if err <> 0 then
  begin
    showErrMsg('WSAStartup Error!');
    Exit;
  end;
  if (Lo(wsaData.wVersion)<>2) or (Hi(wsaData.wVersion)<>2) then
  begin
    showErrMsg('Socket Version Error!');
    Exit;
  end;
  Result := 0;
end;

{ TClientThread }

function TClientThread.ConnectServer: Integer;
var
  len: Integer;
begin
  Result := -1;
  FClientSocket := WSASocket(AF_INET,SOCK_STREAM,IPPROTO_TCP,nil,0,WSA_FLAG_OVERLAPPED);
  if FClientSocket = INVALID_SOCKET then
    Exit;
  FServerAddr.sin_family := AF_INET;
  FServerAddr.sin_addr.S_addr := inet_addr(PAnsiChar(FRemoteIP));
  FServerAddr.sin_port := htons(FRemotePort);
  len := SizeOf(TSockAddrIn);
  if connect(FClientSocket,PSockAddr(@FServerAddr),len)=SOCKET_ERROR then
    Exit;
  FEventArray[0] := WSACreateEvent;
  FOverlapper.hEvent := FEventArray[0];
  Result := 0;
end;

constructor TClientThread.Create(Memo:TMemo; ID:Integer; const IP:string; port:DWORD);
begin
  inherited Create(True);
  FreeOnTerminate := True;
  FClientID := ID;
  FMemo := Memo;
  FRemoteIP := IP;
  FRemotePort := port;
  FEvent := CreateEvent(nil,False,False,nil);
  if ConnectServer = 0 then
    Resume;
end;

destructor TClientThread.Destroy;
begin
  shutdown(FClientSocket,0);
  closesocket(FClientSocket);
  if FEvent > 0 then
    CloseHandle(FEvent);
  WSACloseEvent(FOverlapper.hEvent);
  inherited;
end;

procedure TClientThread.syncLogMsg;
begin
  FMemo.Lines.Add(FLogMsg);
end;

procedure TClientThread.doLogMsg(const msg: String);
begin
  FLogMsg := msg;
  Synchronize(syncLogMsg);
end;

procedure TClientThread.Execute;
var
  dwFlag,dwIndex,dwBytesTransferred: DWORD;
  szText: string;
begin
  inherited;
  if SendMsg('')=0 then Exit;
  RecvMsg('');
  while not Terminated do
  begin
    dwIndex := WSAWaitForMultipleEvents(1,@FOverlapper.hEvent,FALSE,1000,FALSE);
    if (dwIndex=WSA_WAIT_FAILED) or (dwIndex=WSA_WAIT_TIMEOUT) then
    begin
      Continue;
    end;
    dwIndex := dwIndex - WSA_WAIT_EVENT_0;
    WSAResetEvent(FEventArray[dwIndex]);
    WSAGetOverlappedResult(FClientSocket,@FOverlapper,@dwBytesTransferred,FALSE,@dwFlag);
    //Break;
    if dwBytesTransferred=0 then
    begin
      MsgArr[FClientID] := Format('%d Error,dwBytesTransferred=0.',[FClientID]);
      //doLogMsg(Format('%d Error,dwBytesTransferred=0.',[FClientID]));
    end
    else
    begin
      szText := StrPas(FDataBuf.buf);
      MsgArr[FClientID] := Format('%d Msg: %s',[FClientID,szText]);
      //doLogMsg(Format('%d Msg: %s',[FClientID,szText]));
    end;
    Break;
  end;
end;

function TClientThread.RecvMsg(const msg: string):Integer;
begin
  ZeroMemory(@FBuf,DATA_BUFFSIZE);
  FDataBuf.len := DATA_BUFFSIZE;
  FDataBuf.buf := @FBuf;
  Result := WSARecv(FClientSocket,@FDataBuf,1,@FTransBytes,@FTransFlag,@FOverlapper,nil);
end;

function TClientThread.SendMsg(const msg: string):Integer;
var
  len: Integer;
  szText: AnsiString;
  buf: array [0..100-1] of AnsiChar;
  dwBytes,dwFlag,dwBytesTransferred: DWORD;
  SendOverlapper: TOverlapped;
begin
  ZeroMemory(@SendOverlapper,SizeOf(TOverlapped));
  SendOverlapper.hEvent := WSACreateEvent;
  FillChar(buf,100,0);
  szText := 'Test Message.';
  szText := Format('%d Msg: %s',[FClientID,szText]);
  len := Length(szText);
  CopyMemory(@buf,@szText[1],len);
  FDataBuf.len := len;
  FDataBuf.buf := @buf;
  Result := WSASend(FClientSocket,@FDataBuf,1,@dwBytes,0,@SendOverlapper,nil);
  if Result <> SOCKET_ERROR then
  begin
    WSAGetOverlappedResult(FClientSocket,@SendOverlapper,@dwBytesTransferred,FALSE,@dwFlag);
    Result := dwBytesTransferred;
  end;
  WSACloseEvent(SendOverlapper.hEvent);
end;

procedure TClientThread._SetEvent;
begin

end;

initialization
  gStartupFlag := StartupSocket;
finalization
  if gStartupFlag = 0 then
    WSACleanup;

end.

//Main form
unit Unit1;

interface

uses
  Windows, Messages, SysUtils, Variants, Classes, Graphics, Controls, Forms,
  Dialogs, StdCtrls, unitWorkThread;

type
  TForm1 = class(TForm)
    Button1: TButton;
    Memo1: TMemo;
    Button2: TButton;
    cbbIP: TComboBox;
    edtPort: TEdit;
    procedure Button1Click(Sender: TObject);
    procedure Button2Click(Sender: TObject);
    procedure Memo1DblClick(Sender: TObject);
  private
    { Private declarations }
  public
    { Public declarations }
    procedure ShowMsg;
    procedure On_WM_Action(var msg:TMessage);message WM_ACTION;
  end;

var
  Form1: TForm1;

implementation

uses unitWinSock2;

{$R *.dfm}

procedure TForm1.Button1Click(Sender: TObject);
var
  i: Integer;
  port: DWORD;
  ClientArr: array [1..K_ClientCount] of TClientThread;
begin
  MainFormHandle := Self.Handle;
  port := StrToInt(edtPort.Text);
  if unitWorkThread.gStartupFlag = 0 then
  for i := 1 to K_ClientCount do
  begin
    ClientArr[i] := TClientThread.Create(Memo1,i,cbbIP.Text,port);
    Sleep(10);
  end;
  WaitForMultipleObjects(K_ClientCount,@ClientArr,True,480000);
  Memo1.Lines.Add('Execute completed------');
end;

procedure TForm1.Button2Click(Sender: TObject);
begin
  ShowMsg;
end;

procedure TForm1.ShowMsg;
var
  i: Integer;
begin
  for i := 1 to K_ClientCount do
  begin
    Memo1.Lines.Add(MsgArr[i]);
  end;
end;

procedure TForm1.On_WM_Action(var msg: TMessage);
begin
  case msg.LParam of
    1: ShowMsg;
  end;
end;

procedure TForm1.Memo1DblClick(Sender: TObject);
begin
  TMemo(Sender).Clear;
end;

end.