Groups | Search | Server Info | Keyboard shortcuts | Login | Register [http] [https] [nntp] [nntps]

Groups > comp.programming.threads > #2248

Here is my new algorithm...

Cross-posted to 2 groups.

Show all headers | View raw

Here is the source code of my new algorithm:



{*************************************************************
*      Module: A very fast Concurrent FIFO queue
*      Version: 1.0
*      Author: Amine Moulay Ramdane
*     Company: Cyber-NT Communications
*
*       Email: aminer@videotron.ca
*     Website: http://pages.videotron.com/aminer/
*        Date: April 24, 2014
*    Last update: April 24, 2014
*
* Copyright © 2013 Amine Moulay Ramdane.All rights reserved
*
*************************************************************}

unit WQueue;


interface

{$IFDEF FPC}
{$ASMMODE intel}
{$ENDIF}


uses
{$IF defined(WIN32) or  defined(WIN64) }
windows,
{$IFEND}

sysutils,semacondvar;

{$I defines.inc}

const margin=1000; // limited to 1000 threads...
       Alignment = 64; // alignment, needs to be power of 2

type
{$IFDEF CPU64}
int = int64;
Long = uint64;
{$ENDIF CPU64}
{$IFDEF CPU32}
int = integer;
Long = longword;
{$ENDIF CPU32}


tNodeQueue = tObject;

typecache2  = array[0..13] of integer;
typecache3  = array[0..11] of integer;

type cell = record
     obj:long;
     flag:long;
     {$IFDEF CPU32}
     cache:typecache2;
     {$ENDIF CPU32}
     {$IFDEF CPU64}
     cache:typecache3;
     {$ENDIF CPU64}
     end;


MyArray = array[0..0] of cell;
PMyRecord1 = ^MyArray;


   typecache1  = array[0..15] of longword;

   TWQueue = class
   private
       tail:long;
       tmp1:typecache1;
       head: long;
       fMask : long;
       fSize : long;
       temp:long;
       Buffer1: pointer;
       FCount1: PMyRecord1;
       fwait:boolean;
       sema:TSemaMonitor;
       tab : array of tNodeQueue;
       procedure setobject(lp : long;const aobject : long);
       function getObject(lp : long):long;
       function getLength:long;
       function getSize:long;

   public
      {$IFDEF CPU64}
      constructor create(aPower : int =13;wait:boolean=true);  {allocate 
tab with size equal 2^aPower, for 20 size is equal 1048576}
      {$ENDIF CPU64}
      {$IFDEF CPU32}
      constructor create(aPower : int=13;wait:boolean=true);  {allocate 
tab with size equal 2^aPower, for 20 size is equal 1048576}
      {$ENDIF CPU32}


       destructor Destroy; override;
       function push(tm : long):boolean;
       function pop(var obj:long):boolean;
       property length : long read getLength;
       property count: long read getLength;
       property size : long read getSize;

    end;


implementation

{$IF defined(WIN32) or  defined(WIN64) }
function SwitchToThread: BOOL; stdcall; external kernel32 name 
'SwitchToThread';

{$IFEND}



function LockedIncLong(var Target: long): long;
asm
         {$IFDEF CPU32}
         // --> EAX Target
         // <-- EAX Result
         MOV     ECX, EAX
         MOV     EAX, 1
         //sfence
        LOCK XADD [ECX], EAX
         inc     eax
         {$ENDIF CPU32}
         {$IFDEF CPU64}
         // --> RCX Target
         // <-- EAX Result
         MOV     rax, 1
         //sfence
         LOCK XADD [rcx], rax
         INC     rax
         {$ENDIF CPU64}
end;

{$IF defined(CPU64) }
function LockedCompareExchange(CompareVal, NewVal: long; var Target: 
long): long; overload;
asm
mov rax, rcx
lock cmpxchg [r8], rdx
end;
{$IFEND}
{$IF defined(CPU32) }
function LockedCompareExchange(CompareVal, NewVal: long; var 
Target:long): long; overload;
asm
lock cmpxchg [ecx], edx
end;
{$IFEND}


function CAS(var Target:long;Comp ,Exch : long): boolean;
var ret:long;
begin

ret:=LockedCompareExchange(Comp,Exch,Target);
if ret=comp
  then result:=true
  else result:=false;

end; { CAS }

{$IFDEF CPU64}
constructor TWQueue.create(aPower : int=13;wait:boolean=true );
{$ENDIF CPU64}
{$IFDEF CPU32}
constructor TWQueue.create(aPower : int=13;wait:boolean=true );
{$ENDIF CPU32}
var i:int;

begin
fwait:=wait;
   if aPower < 10
     then
      begin
       writeln('Constructor argument must be greater or equal to 10');
        halt;
      end;
  if (aPower < 0) or (aPower > high(integer))
     then
      begin
       writeln('Constructor argument is incorrect');
        halt;
      end;

{$IFDEF CPU64}
fMask:=not($FFFFFFFFFFFFFFFF shl aPower);{$ENDIF CPU64}
{$IFDEF CPU32}
fMask:=not($FFFFFFFF shl aPower);
{$ENDIF CPU32}

   fSize:=(1 shl aPower) - margin;
   Buffer1 := AllocMem((SizeOf(cell)*(1 shl aPower)) + Alignment);
FCount1 := PMyRecord1((int(Buffer1) + Alignment - 1)
                            and not (Alignment - 1));

for i:=0 to (1 shl aPower)-1 do fcount1^[i].flag:=0;
   tail:=0;
   head:=0;
   temp:=0;
  sema:=TSemaMonitor.create(true,0,high(int));
end;

destructor  TWQueue.Destroy;

begin
sema.free;
freemem(buffer1);
inherited Destroy;
end;


procedure TWQueue.setObject(lp : long;const aobject : long);
begin
   fcount1^[lp and fMask].obj:=aObject;
end;

function TWQueue.getObject(lp : long):long;
begin
   result:=fcount1^[lp and fMask].obj;
end;


function TWQueue.push(tm : long):boolean;
var lastHead,newtemp:long;
i,j:integer;
begin

if getlength >= fsize
   then
       begin
           result:=false;
           exit;
       end;

newTemp:=LockedIncLong(head);
lastHead:=newtemp-1;

repeat
asm pause end;
until fcount1^[lastHead and fMask].flag=0;
setObject(lastHead,tm);
fcount1^[lastHead and fMask].flag:=1;
if fwait then sema.signal;
result:=true;
end;


function TWQueue.pop(var obj:long):boolean;

var lastTail,newtemp,newtemp1,newtemp2 : long;
begin

if fwait then sema.wait;

repeat

newtemp1:=tail;
newtemp2:=newtemp1+1;

if newtemp2<=head then
  else
   begin
    result:=false;
    exit;
   end;
if CAS(tail,newtemp1,newtemp2) then break;
sleep(0);
until false;

lastTail:=newtemp1;
repeat
asm pause end;
until fcount1^[lastTail and fMask].flag=1;
obj:=getObject(lastTail);
fcount1^[lastTail and fMask].flag:=0;
result:=true;

end;

function TWQueue.getLength:long;
var head1,tail1:long;
begin
head1:=head;
tail1:=tail;
   if head1 < tail1
        then result:= (High(long)-tail1)+(1+head1)
        else result:=(head1-tail1);
end;

function TWQueue.getSize:long;

begin
   result:=fSize;
end;

end.

end.

Back to comp.programming.threads | Previous | Next | Find similar | Unroll thread

Thread

Here is my new algorithm... aminer <aminer@toto.net> - 2014-04-24 18:27 -0700

csiph-web