unit Hub;

uses Classes, SysUtils, SyncObjs, {$IFDEF LOCK_FREE} PasMP, {$ENDIF}

{$IFDEF FPC} fgl {$ELSE}Generics.Collections, System.Rtti{$ENDIF},

GarbageCollector, GInterfaces;

THub = class(TCustomHub)

strict private

type

TTaskKind = (tkMethod, tkIntfMethod, tkProc);

{ TTask }

TTask = record

Kind: TTaskKind;

Method: TThreadMethod;

Intf: IInterface;

Proc: TTaskProc;

Arg: Pointer;

procedure Init(const Method: TThreadMethod); overload;

procedure Init(Intf: IInterface; const Method: TThreadMethod); overload;

procedure Init(const Proc: TTaskProc; Arg: Pointer); overload;

{$IFDEF FPC}

class operator = (const a,b : TTask): Boolean;

{$ENDIF}

end;

TQueue = {$IFDEF DCC}TList<TTask>{$ELSE}TFPGList<TTask>{$ENDIF};

procedure GetAddresses(const Task: TThreadMethod; out Obj: TObject;

out MethodAddr: Pointer); inline;

var

FQueue: TQueue;

FAccumQueue: TQueue;

FLoopExit: Boolean;

FName: string;

FLock: SyncObjs.TCriticalSection;

FLS: Pointer;

FGC: Pointer;

FIsSuspended: Boolean;

{$IFDEF LOCK_FREE}

FLockFreeQueue: TPasMPUnboundedQueue;

{$ENDIF}

procedure Clean;

procedure Swap(var A1, A2: TQueue);

protected

procedure Lock; inline;

procedure Unlock; inline;

function TryLock: Boolean; inline;

function ServeTaskQueue: Boolean; dynamic;

procedure AfterSwap; dynamic;

public

constructor Create;

destructor Destroy; override;

property Name: string read FName write FName;

property IsSuspended: Boolean read FIsSuspended write FIsSuspended;

function HLS(const Key: string): TObject; override;

procedure HLS(const Key: string; Value: TObject); override;

function GC(A: TObject): IGCObject; override;

procedure EnqueueTask(const Task: TThreadMethod); override;

procedure EnqueueTask(const Task: TTaskProc; Arg: Pointer); override;

procedure Pulse; override;

procedure LoopExit; override;

procedure Loop(const Cond: TExitCondition; Timeout: LongWord = INFINITE); override;

procedure Loop(const Cond: TExitConditionStatic; Timeout: LongWord = INFINITE); override;

procedure Switch; override;

end;

TSingleThreadHub = class(THub)

type

TMultiplexorEvent = (meError, meEvent, meSync, meIO, meWinMsg,

meTimer, meWaitTimeout);

TMultiplexorEvents = set of TMultiplexorEvent;

const

ALL_EVENTS: TMultiplexorEvents = [Low(TMultiplexorEvent)..High(TMultiplexorEvent)];

private

type

TTimeoutRoutine = function(Timeout: LongWord): Boolean of object;

var

FEvents: Pointer;

FReadFiles: Pointer;

FWriteFiles: Pointer;

FTimeouts: Pointer;

FWakeMainThread: TNotifyEvent;

FThreadId: LongWord;

FTimerFlag: Boolean;

FIOFlag: Boolean;

FIsDef: Boolean;

FProcessWinMsg: Boolean;

FSyncEvent: TEvent;

FServeRoutine: TTimeoutRoutine;

procedure SetTriggerFlag(const TimerFlag, IOFlag: Boolean);

procedure WakeUpThread(Sender: TObject);

procedure SetupEnviron;

procedure BeforeServe;inline;

procedure AfterServe; inline;

function ServeAsDefHub(Timeout: LongWord): Boolean;

function ServeHard(Timeout: LongWord): Boolean;

protected

function ServeMultiplexor(TimeOut: LongWord): TMultiplexorEvent; dynamic;

public

constructor Create;

destructor Destroy; override;

// IO files, sockets, com-ports, pipes, etc.

procedure Cancel(Fd: THandle); override;

function Write(Fd: THandle; Buf: Pointer; Len: LongWord; const Cb: TIOCallback; Data: Pointer; Offset: Int64 = -1): Boolean; override;

function WriteTo(Fd: THandle; Buf: Pointer; Len: LongWord; const Addr: TAnyAddress; const Cb: TIOCallback; Data: Pointer): Boolean; override;

function Read(Fd: THandle; Buf: Pointer; Len: LongWord; const Cb: TIOCallback; Data: Pointer; Offset: Int64 = -1): Boolean; override;

function ReadFrom(Fd: THandle; Buf: Pointer; Len: LongWord; const Addr: TAnyAddress; const Cb: TIOCallback; Data: Pointer): Boolean; override;

// IO timeouts and timers

function CreateTimer(const Cb: TInervalCallback; Data: Pointer; Interval: LongWord): THandle; override;

procedure DestroyTimer(Id: THandle); override;

function CreateTimeout(const Cb: TInervalCallback; Data: Pointer; Interval: LongWord): THandle; override;

procedure DestroyTimeout(Id: THandle); override;

// Event objects

function AddEvent(Ev: THandle; const Cb: TEventCallback; Data: Pointer): Boolean; override;

procedure RemEvent(Ev: THandle); override;

// services

function Serve(TimeOut: LongWord): Boolean; override;

function Wait(TimeOut: LongWord; const Events: TMultiplexorEvents=[];

const ProcessWinMsg: Boolean = True): TMultiplexorEvent;

procedure Pulse; override;

end;

function GetCurrentHub: TCustomHub;

function DefHub(ThreadID: LongWord = 0): TSingleThreadHub; overload;

function DefHub(Thread: TThread): TSingleThreadHub; overload;

HubInfrasctuctureEnable: Boolean;

IOEventTupleCounter: Integer;

IOFdTupleCounter: Integer;

IOTimeTupleCounter: Integer;

uses Math, GreenletsImpl, Greenlets, sock,

{$IFDEF MSWINDOWS}

{$IFDEF DCC}Winapi.Windows{$ELSE}windows{$ENDIF}

{$ELSE}

// TODO

{$ENDIF};

NANOSEC_PER_MSEC = 1000000;

TObjMethodStruct = packed record

Code: Pointer;

Data: TObject;

end;

THubMap = {$IFDEF DCC}TDictionary{$ELSE}TFPGMap{$ENDIF}<LongWord, TSingleThreadHub>;

TEventTuple = record

Event: THandle;

Cb: TEventCallback;

Data: Pointer;

Hub: THub;

procedure Trigger(const Aborted: Boolean);

end;

PEvents = ^TEvents;

TEvents = record

const

MAX_EV_COUNT = 1024;

var

SyncEvent: TEvent;

FBuf: array[0..MAX_EV_COUNT-1] of THandle;

FCb: array[0..MAX_EV_COUNT-1] of TEventCallback;

FData: array[0..MAX_EV_COUNT-1] of Pointer;

FHub: array[0..MAX_EV_COUNT-1] of THub;

FBufSize: Integer;

procedure Enqueue(Hnd: THandle; const Cb: TEventCallback;

Data: Pointer; Hub: THub);

procedure DequeueByIndex(Index: Integer); overload;

procedure Dequeue(Hnd: THandle); overload;

function Buf: Pointer; inline;

function TupleByIndex(Index: Integer): TEventTuple; inline;

function Find(Hnd: THandle; out Index: Integer): Boolean; inline;

function BufSize: LongWord; inline;

function IsInitialized: Boolean; inline;

procedure Initialize; inline;

procedure DeInitialize; inline;

end;

PFileTuple = ^TFileTuple;

TFileTuple = record

Fd: THandle;

ReadData: Pointer;

WriteData: Pointer;

ReadCb: TIOCallback;

WriteCb: TIOCallback;

Hub: THub;

Active: Boolean;

CleanTime: TTime;

{$IFDEF MSWINDOWS}

Overlap: TOverlapped;

{$ENDIF}

procedure RecalcCleanTime;

procedure Trigger(ErrorCode: Integer; Len: Integer; const Op: TIOOperation);

end;

PFiles = ^TFiles;

TFiles = record

const

TRASH_CLEAR_TIMEOUT_MINS = 30;

type

TMap = {$IFDEF DCC}TDictionary{$ELSE}TFPGMap{$ENDIF}<THandle, PFileTuple>;

FList = {$IFDEF DCC}TList{$ELSE}TFPGList{$ENDIF}<PFileTuple>;

var

FMap: TMap;

// in input-output operations callbacks can come to private

// descriptors. This is found for sockets

FTrash: FList;

FTrashLAstClean: TTime;

procedure TrashClean(const OnlyByTimeout: Boolean = True);

function IsInitialized: Boolean;

procedure Initialize;

procedure Deinitialize;

function Enqueue(Fd: THandle; const Op: TIOOperation;

const Cb: TIOCallback; Data: Pointer; Hub: THub): PFileTuple; overload;

procedure Dequeue(const Fd: THandle);

function Find(Fd: THandle): Boolean; overload;

function Find(Id: THandle; out Tup: PFileTuple): Boolean; overload;

end;

PTimeoutTuple = ^TTimeoutTuple;

TTimeoutTuple = record

Id: THandle;

Cb: TInervalCallback;

Data: Pointer;

Hub: THub;

procedure Trigger;

end;

TRawGreenletPImpl = class(TRawGreenletImpl);

PTimeouts = ^TTimeouts;

TTimeouts = record

type

TMap = {$IFDEF DCC}TDictionary{$ELSE}TFPGMap{$ENDIF}<THandle, PTimeoutTuple>;

var

FMap: TMap;

function IsInitialized: Boolean;

procedure Initialize;

procedure Deinitialize;

public

function Enqueue(Id: THandle; const Cb: TInervalCallback;

Data: Pointer; Hub: THub): PTimeoutTuple;

procedure Dequeue(Id: THandle);

function Find(Id: THandle): Boolean; overload;

end;

threadvar

CurrentHub: THub;

DefHubsLock: SyncObjs.TCriticalSection;

DefHubs: THubMap;

{$I iocpwin.inc}

WsaDataOnce: TWSADATA;

function GetCurrentProcessorNumber: DWORD; external kernel32 name 'GetCurrentProcessorNumber';

function GetCurrentHub: TCustomHub;

if Assigned(CurrentHub) then

Result := CurrentHub

else

Result := DefHub;

end;

function DefHub(ThreadID: LongWord): TSingleThreadHub;

ID: LongWord;

CintainsKey: Boolean;

{$IFNDEF DCC}

Index: Integer;

{$ENDIF}

if not Assigned(DefHubs) then

Exit(nil);

if ThreadID = 0 then

ID := TThread.CurrentThread.ThreadID

else

ID := ThreadID;

DefHubsLock.Acquire;

try

{$IFDEF DCC}

CintainsKey := DefHubs.ContainsKey(ID);

{$ELSE}

CintainsKey := DefHubs.Find(ID, Index);

{$ENDIF}

if CintainsKey then

Result := DefHubs[ID]

else begin

Result := TSingleThreadHub.Create;

Result.FIsDef := True;

Result.FThreadId := ID;

DefHubs.Add(ID, Result);

end;

finally

DefHubsLock.Release

end;

end;

function DefHub(Thread: TThread): TSingleThreadHub;

Result := DefHub(Thread.ThreadID)

end;

procedure STFileIOCompletionReadRoutine(dwErrorCode: DWORD;

dwNumberOfBytesTransfered:DWORD; lpOverlapped: POverlapped); stdcall;

Tuple: PFileTuple;

Tuple := PFileTuple(lpOverlapped.hEvent);

if not Tuple.Active then begin

Tuple.RecalcCleanTime;

Exit;

end;

TSingleThreadHub(Tuple.Hub).SetTriggerFlag(False, True);

Tuple.Trigger(dwErrorCode, dwNumberOfBytesTransfered, ioRead);

end;

procedure STFlaggedFileIOCompletionReadRoutine(dwErrorCode: DWORD;

dwNumberOfBytesTransfered:DWORD; lpOverlapped: POverlapped; Flags: DWORD); stdcall;

Tuple: PFileTuple;

Tuple := PFileTuple(lpOverlapped.hEvent);

if not Tuple.Active then begin

Tuple.RecalcCleanTime;

Exit;

end;

TSingleThreadHub(Tuple.Hub).SetTriggerFlag(False, True);

Tuple.Trigger(dwErrorCode, dwNumberOfBytesTransfered, ioRead);

end;

procedure STFileIOCompletionWriteRoutine(dwErrorCode: DWORD;

dwNumberOfBytesTransfered:DWORD; lpOverlapped: POverlapped); stdcall;

Tuple: PFileTuple;

Tuple := PFileTuple(lpOverlapped.hEvent);

if not Tuple.Active then begin

Tuple.RecalcCleanTime;

Exit;

end;

TSingleThreadHub(Tuple.Hub).SetTriggerFlag(False, True);

Tuple.Trigger(dwErrorCode, dwNumberOfBytesTransfered, ioWrite);

end;

procedure STFlaggedFileIOCompletionWriteRoutine(dwErrorCode: DWORD;

dwNumberOfBytesTransfered:DWORD; lpOverlapped: POverlapped; Flags: DWORD); stdcall;

Tuple: PFileTuple;

Tuple := PFileTuple(lpOverlapped.hEvent);

if not Tuple.Active then begin

Tuple.RecalcCleanTime;

Exit;

end;

TSingleThreadHub(Tuple.Hub).SetTriggerFlag(False, True);

Tuple.Trigger(dwErrorCode, dwNumberOfBytesTransfered, ioWrite);

end;

procedure STTimerAPCProc(lpArgToCompletionRoutine: Pointer;

dwTimerLowValue: DWORD; dwTimerHighValue: DWORD); stdcall;

Tuple: PTimeoutTuple;

Tuple := lpArgToCompletionRoutine;

TSingleThreadHub(Tuple.Hub).SetTriggerFlag(True, False);

Tuple.Trigger;

end;

function GetEvents(Hub: TSingleThreadHub): PEvents; inline;

Result := PEvents(Hub.FEvents)

end;

function GetReadFiles(Hub: TSingleThreadHub): PFiles; inline;

Result := PFiles(Hub.FReadFiles)

end;

function GetWriteFiles(Hub: TSingleThreadHub): PFiles; inline;

Result := PFiles(Hub.FWriteFiles)

end;

function GetTimeouts(Hub: TSingleThreadHub): PTimeouts; inline;

Result := PTimeouts(Hub.FTimeouts)

end;

procedure THub.AfterSwap;

end;

function THub.ServeTaskQueue: Boolean;

I: Integer;

Arg: Pointer;

Proc: TTaskProc;

OldHub: THub;

Tsk: TTask;

procedure Process(var Tsk: TTask);

begin

case Tsk.Kind of

tkMethod: begin

Tsk.Method();

end;

tkIntfMethod: begin

if Assigned(Tsk.Intf) then begin

Tsk.Method();

Tsk.Intf := nil;

end;

end;

tkProc: begin

Proc := Tsk.Proc;

Arg := Tsk.Arg;

Proc(Arg)

end;

end;

end;

Result := False;

OldHub := CurrentHub;

try

CurrentHub := Self;

{$IFDEF MSWINDOWS}

if TThread.CurrentThread.ThreadID = MainThreadID then begin

Result := CheckSynchronize(0)

end;

{$ENDIF}

{$IFDEF LOCK_FREE}

try

while FLockFreeQueue.Dequeue(Tsk) do begin

Result := True;

Process(Tsk)

end;

finally

while FLockFreeQueue.Dequeue(Tsk) do ;

end;

{$ELSE}

// transaction commit to write and transfer data

// in transaction for reading - to reduce problems with race condition

Swap(FQueue, FAccumQueue);

Result := Result or (FQueue.Count > 0);

if FQueue.Count > 0 then

try

for I := 0 to FQueue.Count-1 do begin

Tsk := FQueue[I];

Process(Tsk);

end;

finally

// obligatory it is necessary to clean, differently at Abort or Exception

// on the iteration trace. queues will come up with "outdated" calls

FQueue.Clear;

end;

{$ENDIF}

finally

CurrentHub := OldHub;

end;

end;

procedure THub.Clean;

procedure CleanQueue(A: TQueue);

var

I: Integer;

begin

for I := 0 to A.Count-1 do

with A[I] do begin

case Kind of

tkIntfMethod: begin

//Intf._Release;

end;

tkMethod:;

tkProc: ;

end;

end;

A.Clear;

end;

Lock;

try

CleanQueue(FAccumQueue);

CleanQueue(FQueue);

finally

Unlock

end;

end;

procedure THub.Swap(var A1, A2: TQueue);

Tmp: TQueue;

Lock;

try

Tmp := A1;

A1 := A2;

A2 := Tmp;

finally

AfterSwap;

Unlock;

end;

end;

procedure THub.Lock;

{$IFDEF DCC}

TMonitor.Enter(FLock);

{$ELSE}

FLock.Acquire;

{$ENDIF}

end;

procedure THub.Unlock;

{$IFDEF DCC}

TMonitor.Exit(FLock);

{$ELSE}

FLock.Release;

{$ENDIF}

end;

function THub.TryLock: Boolean;

{$IFDEF DCC}

Result := TMonitor.TryEnter(FLock);

{$ELSE}

Result := FLock.TryEnter

{$ENDIF}

end;

constructor THub.Create;

inherited Create;

FQueue := TQueue.Create;

FAccumQueue := TQueue.Create;

FLock := SyncObjs.TCriticalSection.Create;

FLS := TLocalStorage.Create;

FGC := TGarbageCollector.Create;

GetJoiner(Self);

{$IFDEF LOCK_FREE}

FLockFreeQueue := TPasMPUnboundedQueue.Create(SizeOf(TTask));

{$ENDIF}

end;

destructor THub.Destroy;

Clean;

FQueue.Free;

FAccumQueue.Free;

TLocalStorage(FLS).Free;

TGarbageCollector(FGC).Free;

FLock.Free;

{$IFDEF LOCK_FREE}

FLockFreeQueue.Free;

{$ENDIF}

inherited;

end;

function THub.HLS(const Key: string): TObject;

Result := TLocalStorage(FLS).GetValue(Key);

end;

procedure THub.HLS(const Key: string; Value: TObject);

Obj: TObject;

if TLocalStorage(FLS).IsExists(Key) then begin

Obj := TLocalStorage(FLS).GetValue(Key);

Obj.Free;

TLocalStorage(FLS).UnsetValue(Key);

end;

if Assigned(Value) then

TLocalStorage(FLS).SetValue(Key, Value)

end;

function THub.GC(A: TObject): IGCObject;

G: TGarbageCollector;

G := TGarbageCollector(FGC);

Result := G.SetValue(A);

end;

procedure THub.EnqueueTask(const Task: TTaskProc; Arg: Pointer);

Tsk: TTask;

{$IFDEF LOCK_FREE}

Q: TPasMPUnboundedQueue;

{$ENDIF}

Tsk.Init(Task, Arg);

{$IFDEF LOCK_FREE}

FLockFreeQueue.Enqueue(Tsk);

{$ELSE}

Lock;

try

FAccumQueue.Add(Tsk);

finally

Unlock

end;

{$ENDIF}

Pulse;

end;

procedure THub.EnqueueTask(const Task: TThreadMethod);

Obj: TObject;

MethodAddr: Pointer;

Intf: IInterface;

Tsk: TTask;

{$IFDEF LOCK_FREE}

Q: TPasMPUnboundedQueue;

{$ENDIF}

GetAddresses(Task, Obj, MethodAddr);

{$IFDEF DEBUG}

Assert(not Obj.InheritsFrom(TRawGreenletImpl), 'Enqueue Greenlet methods only by Proxy');

{$ENDIF}

if Obj.GetInterface(IInterface, Intf) then

Tsk.Init(Intf, Task)

else

Tsk.Init(Task);

{$IFDEF LOCK_FREE}

if Assigned(Intf) then

Intf._AddRef;

FLockFreeQueue.Enqueue(Tsk);

{$ELSE}

Lock;

try

// write transaction

FAccumQueue.Add(Tsk);

finally

Unlock;

end;

{$ENDIF}

Pulse;

end;

procedure THub.GetAddresses(const Task: TThreadMethod; out Obj: TObject;

out MethodAddr: Pointer);

Struct: TObjMethodStruct absolute Task;

Obj := Struct.Data;

MethodAddr := Struct.Code;

end;

procedure THub.Loop(const Cond: TExitCondition; Timeout: LongWord);

Stop: TTime;

FLoopExit := False;

Stop := Now + TimeOut2Time(Timeout);

while not Cond() and (Now < Stop) and (not FLoopExit) do

Serve(Time2TimeOut(Stop - Now))

end;

procedure THub.Loop(const Cond: TExitConditionStatic; Timeout: LongWord);

Stop: TTime;

FLoopExit := False;

Stop := Now + TimeOut2Time(Timeout);

while not Cond() and (Now < Stop) and (not FLoopExit) do

Serve(Time2TimeOut(Stop - Now))

end;

procedure THub.Switch;

if Greenlets.GetCurrent <> nil then begin

TRawGreenletPImpl.Switch2RootContext

end;

end;

procedure THub.LoopExit;

FLoopExit := True;

Pulse;

end;

procedure THub.Pulse;

end;

function TSingleThreadHub.AddEvent(Ev: THandle; const Cb: TEventCallback;

Data: Pointer): Boolean;

Index: Integer;

Tup: TEventTuple;

{$IFDEF MSWINDOWS}

Result := GetEvents(Self).BufSize < (MAXIMUM_WAIT_OBJECTS-1);

{$ELSE}

Result := True;

{$ENDIF}

if GetEvents(Self).Find(Ev, Index) then begin

Tup := GetEvents(Self).TupleByIndex(Index);

if (@Tup.Cb <> @Cb) or (Tup.Data <> Data) then

raise EHubError.CreateFmt('Handle %d already exists in demultiplexor queue', [Ev]);

end;

if Result then begin

GetEvents(Self).Enqueue(Ev, Cb, Data, Self);

Pulse;

end;

end;

procedure TSingleThreadHub.AfterServe;

if Assigned(FWakeMainThread) then

Classes.WakeMainThread := FWakeMainThread;

if not FIsDef then

FThreadId := 0;

end;

procedure TSingleThreadHub.BeforeServe;

if FIsDef then begin

if FThreadId <> TThread.CurrentThread.ThreadID then

raise EHubError.Create('Def Hub must be serving inside owner thread');

end

else begin

if FThreadId <> 0 then

raise EHubError.Create('Hub already serving by other thread');

FThreadId := TThread.CurrentThread.ThreadID;

end;

SetupEnviron;

if TThread.CurrentThread.ThreadID = MainThreadID then begin

FWakeMainThread := Classes.WakeMainThread;

Classes.WakeMainThread := Self.WakeUpThread;

end;

end;

procedure TSingleThreadHub.Cancel(Fd: THandle);

R, W: Boolean;

R := GetReadFiles(Self).Find(Fd);

W := GetWriteFiles(Self).Find(Fd);

if W or R then begin

{$IFDEF MSWINDOWS}

CancelIo(Fd);

{$ELSE}

{$ENDIF}

end;

if R then

GetReadFiles(Self).Dequeue(Fd);

if W then

GetWriteFiles(Self).Dequeue(Fd);

end;

constructor TSingleThreadHub.Create;

inherited Create;

SetupEnviron;

FServeRoutine := ServeHard;

end;

function TSingleThreadHub.CreateTimeout(const Cb: TInervalCallback;

Data: Pointer; Interval: LongWord): THandle;

DueTime: Int64;

TuplePtr: PTimeoutTuple;

{$IFDEF MSWINDOWS}

Result := CreateWaitableTimer(nil, False, '');

DueTime := Interval*(NANOSEC_PER_MSEC div 100);

DueTime := DueTime * -1;

TuplePtr := GetTimeouts(Self).Enqueue(Result, Cb, Data, Self);

if not SetWaitableTimer(Result, DueTime, 0, @STTimerAPCProc, TuplePtr, False) then begin

GetTimeouts(Self).Dequeue(Result);

end;

{$ELSE}

{$ENDIF}

end;

function TSingleThreadHub.CreateTimer(const Cb: TInervalCallback; Data: Pointer;

Interval: LongWord): THandle;

DueTime: Int64;

TuplePtr: PTimeoutTuple;

{$IFDEF MSWINDOWS}

Result := CreateWaitableTimer(nil, False, '');

DueTime := Interval*(NANOSEC_PER_MSEC div 100);

DueTime := DueTime * -1;

TuplePtr := GetTimeouts(Self).Enqueue(Result, Cb, Data, Self);

if not SetWaitableTimer(Result, DueTime, Interval, @STTimerAPCProc, TuplePtr, False) then begin

GetTimeouts(Self).Dequeue(Result);

end;

{$ELSE}

{$ENDIF}

end;

destructor TSingleThreadHub.Destroy;

ContainsKey: Boolean;

{$IFNDEF DCC}

Index: Integer;

{$ENDIF}

if Assigned(FEvents) then begin

GetEvents(Self).DeInitialize;

FreeMem(FEvents, SizeOf(TEvents));

end;

if Assigned(FReadFiles) then begin

GetReadFiles(Self).Deinitialize;

FreeMem(FReadFiles, SizeOf(TFiles));

end;

if Assigned(FWriteFiles) then begin

GetWriteFiles(Self).Deinitialize;

FreeMem(FWriteFiles, SizeOf(TFiles));

end;

if Assigned(FTimeouts) then begin

GetTimeouts(Self).Deinitialize;

FreeMem(FTimeouts, SizeOf(TTimeouts));

end;

if FIsDef then begin

DefHubsLock.Acquire;

try

{$IFDEF DCC}

ContainsKey := DefHubs.ContainsKey(FThreadId);

{$ELSE}

ContainsKey := DefHubs.Find(FThreadId, Index);

{$ENDIF}

if ContainsKey then

DefHubs.Remove(FThreadId);

finally

DefHubsLock.Release

end;

end;

TRawGreenletPImpl.ClearContexts;

CurrentHub := nil;

inherited;

end;

procedure TSingleThreadHub.DestroyTimeout(Id: THandle);

{$IFDEF MSWINDOWS}

if GetTimeouts(Self).Find(Id) then begin

CancelWaitableTimer(Id);

GetTimeouts(Self).Dequeue(Id);

end

{$ELSE}

{$ENDIF}

end;

procedure TSingleThreadHub.DestroyTimer(Id: THandle);

{$IFDEF MSWINDOWS}

if GetTimeouts(Self).Find(Id) then begin

CancelWaitableTimer(Id);

GetTimeouts(Self).Dequeue(Id);

end;

{$ELSE}

{$ENDIF}

end;

procedure TSingleThreadHub.Pulse;

FSyncEvent.SetEvent;

end;

function TSingleThreadHub.ReadFrom(Fd: THandle; Buf: Pointer; Len: LongWord; const Addr: TAnyAddress; const Cb: TIOCallback; Data: Pointer): Boolean;

TuplePtr: PFileTuple;

Flags: DWORD;

Buf_: WSABUF;

RetValue: LongWord;

TuplePtr := GetReadFiles(Self).Enqueue(Fd, ioRead, Cb, Data, Self);

Flags := MSG_PARTIAL;

Buf_.len := Len;

Buf_.buf := Buf;

WSARecvFrom(Fd, @Buf_, 1, nil, Flags, Addr.AddrPtr,

@Addr.AddrLen, @TuplePtr.Overlap, @STFlaggedFileIOCompletionReadRoutine);

RetValue := WSAGetLastError;

Result := RetValue = WSA_IO_PENDING;

if not Result then begin

GetReadFiles(Self).Dequeue(Fd);

end;

end;

function TSingleThreadHub.Read(Fd: THandle; Buf: Pointer; Len: LongWord;

const Cb: TIOCallback; Data: Pointer; Offset: Int64): Boolean;

TuplePtr: PFileTuple;

TuplePtr := GetReadFiles(Self).Enqueue(Fd, ioRead, Cb, Data, Self);

{$IFDEF MSWINDOWS}

if Offset <> -1 then begin

TuplePtr^.Overlap.Offset := Offset and $FFFFFFFF;

TuplePtr^.Overlap.OffsetHigh := Offset shr 32;

end;

Result := ReadFileEx(Fd, Buf, Len, @TuplePtr^.Overlap, @STFileIOCompletionReadRoutine);

if not Result then

GetReadFiles(Self).Dequeue(Fd);

{$IFDEF DEBUG}

//raise EHubError.CreateFmt('Error Message: %s', [SysErrorMessage(GetLastError)]);

{$ENDIF}

{$ELSE}

{$ENDIF}

end;

procedure TSingleThreadHub.RemEvent(Ev: THandle);

GetEvents(Self).Dequeue(Ev);

end;

function TSingleThreadHub.ServeAsDefHub(Timeout: LongWord): Boolean;

Stop: TTime;

MxEvent: TMultiplexorEvent;

Stop := Now + TimeOut2Time(TimeOut);

repeat

if ServeTaskQueue then

Exit(True)

else begin

FProcessWinMsg := True;

MxEvent := ServeMultiplexor(Time2TimeOut(Stop - Now));

Result := MxEvent <> meWaitTimeout;

Result := Result or ServeTaskQueue;

end;

until (Now >= Stop) or Result;

end;

function TSingleThreadHub.ServeHard(Timeout: LongWord): Boolean;

BeforeServe;

try

Result := ServeAsDefHub(Timeout);

finally

AfterServe

end;

if FIsDef then

FServeRoutine := ServeAsDefHub

end;

function TSingleThreadHub.Serve(TimeOut: LongWord): Boolean;

Result := FServeRoutine(TimeOut);