UploadThread.pas

   1 unit UploadThread;
   2 {coprocessor to Upload unit. Move I/O out of main thread}
   3 INTERFACE
   4 uses Store1,Sockets,NetAddr;
   5
   6 type tSegment=record
   7  base,len:LongWord;
   8 end;
   9 type tChannel=record
  10  s: array [1..24] of tSegment;
  11  seg:byte;
  12  weight:Word;
  13  wcur:Word;
  14  oi:tStoreObjectInfo;
  15  end;
  16 type tUploadThr=object
  17  thrid:tThreadID;
  18  crit:tRtlCriticalSection;
  19  socket:tSocket;
  20  remote:tSockAddrL;
  21  size1,size2:Word;
  22  mark:Byte;
  23  rate:Single;
  24  MarkTime:LongWord;{ms}
  25  MarkData:LongWord;
  26  chans:array [0..11] of ^tChannel;
  27  curc:byte;
  28  buffer:array [0..2047] of byte;
  29  stop:boolean; {the therad is stopped or stopping}
  30  wait:boolean; {the therad is waiting for data}
  31  waitc:byte;
  32
  33  procedure Main;
  34  procedure Init(source:tNetAddr);
  35  procedure Start;
  36  procedure Done;
  37 end;
  38
  39 IMPLEMENTATION
  40 uses MemStream,ServerLoop,SysUtils,opcode;
  41
  42 procedure tUploadThr.Init(source:tNetAddr);
  43  var i:integer;
  44  begin
  45  InitCriticalSection(crit);
  46  source.ToSocket(remote);
  47  socket:=GetSocket(source);
  48  MarkData:=0;
  49  MarkTime:=0;
  50  stop:=true;
  51  wait:=false;
  52  for i:=0 to high(chans) do chans[i]:=nil;
  53 end;
  54
  55 procedure tUploadThr.Main;
  56  var pch:byte;
  57  var s:tMemoryStream;
  58  var sz:Word;
  59  var txwait,delta:single;//msec
  60  var LastTime:tDateTime;//days
  61  var chan:^tChannel;
  62  var seg:^tSegment;
  63  begin
  64  txwait:=0;
  65  delta:=0;
  66  while not stop do begin
  67  EnterCriticalSection(crit);
  68  pch:=0;
  69  {find usable channel}
  70  while (chans[curc]=nil)or(chans[curc]^.wcur=0)or(chans[curc]^.seg=0) do begin
  71   if assigned(chans[curc])and(chans[curc]^.WCur=0) then chans[curc]^.WCur:=chans[curc]^.weight;
  72   inc(curc);
  73   inc(pch);
  74   if curc>high(chans) then curc:=0;
  75   if pch>(high(chans)+1) then begin wait:=true; break; end;
  76  end;
  77  if wait then begin
  78   LeaveCriticalSection(crit);
  79   if waitc>10
  80   then stop:=true
  81   else inc(waitc);
  82   sleep(200);
  83   continue;
  84  end;
  85  waitc:=0;
  86  LastTime:=SysUtils.Now;
  87  chan:=chans[curc];
  88  seg:=@chan^.s[chan^.seg];
  89  s.Init(@buffer,0,high(buffer));
  90  {prepare header}
  91  if size2>s.size then size2:=0;
  92  if size2=0 then begin
  93   sz:=size1; if size1>s.size then sz:=s.size;
  94   s.WriteByte(opcode.tcdata);
  95  end else begin
  96   sz:=size2; if sz>s.size then sz:=s.size;
  97   s.WriteByte(opcode.tcdataimm);
  98   size2:=0;
  99  end;
 100  Assert(seg^.len>0);
 101  s.WriteByte(mark);
 102  s.WriteByte(curc);
 103  s.WriteWord(seg^.base,4);
 104  dec(sz,s.length);
 105  if sz>seg^.Len then sz:=seg^.Len;
 106  assert(sz<=seg^.len);
 107  chan^.oi.ReadSeg(s.WrBuf,seg^.base,sz);
 108  Assert(chan^.oi.rc=0,'IO error reading segment');
 109  s.WrEnd(sz);
 110  assert((Seg^.Len-sz)>=0);
 111  Dec(Seg^.Len,sz);
 112  Dec(chan^.WCur);
 113  if Seg^.Len=0 then Dec(chan^.seg)
 114  else Inc(Seg^.Base,sz);
 115  LeaveCriticalSection(crit);
 116  fpSendTo(socket,s.base,s.length,0,@remote,sizeof(remote));
 117  txwait:=((MarkData/Rate)*1000)-(MarkTime);
 118  MarkData:=MarkData+s.length;
 119  if txWait>1000 then begin writeln('!!! txwait=',round(txWait)); txWait:=1000;end;
 120  if txWait>0 then Sleep(round(txWait));
 121  Delta:=Delta+((SysUtils.Now-LastTime)*MSecsPerDay);
 122  if Delta>5000 then Delta:=3000;
 123  if Delta<0 then Delta:=0;
 124  MarkTime:=MarkTime+trunc(Delta);
 125  Delta:=frac(Delta);
 126  end;
 127 end;
 128
 129 function thrfunc(p:pointer):PtrInt;
 130  begin
 131  tUploadThr(p^).Main;
 132  thrfunc:=9;
 133 end;
 134 procedure tUploadThr.Start;
 135  begin
 136  wait:=false;
 137  if not stop then exit;
 138  stop:=false;
 139  MarkData:=0;
 140  MarkTime:=0;
 141  thrid:=BeginThread(@ThrFunc,@self);
 142 end;
 143
 144 procedure tUploadThr.Done;
 145  begin
 146  if stop then exit;
 147  EnterCriticalSection(crit);
 148  stop:=true;
 149  LeaveCriticalSection(crit);
 150  WaitForThreadterminate(thrid,65535);
 151  DoneCriticalSection(crit);
 152 end;
 153 END.