TC.pas

   1 unit TC;
   2 {TransmissionControll over UDP
   3 some dg larger
   4 if pass set payload to that
   5
   6 useful for file transfer, voip should only consult the current rate
   7 and detect congestion based on latency
   8
   9 opcodes:
  10  data=4
  11   mark:1;payload:XX
  12  data-no-report=8
  13  data-imm-ack=6
  14  cont=5
  15   mark:1;rate:Word4(shr 6)
  16  ack=7
  17   mark:1;len:Word2
  18 }
  19 INTERFACE
  20 uses MemStream,NetAddr;
  21
  22 type tTCSSe=record
  23  Rate:Real; {sending rate}
  24  Size:word; {datagram size}
  25  RateIF:single; {rate increase fraction}
  26  SizeIF:single; {size increase fraction}
  27  {new:=old+old*IF}
  28  end;
  29
  30 type tTCS=object {this is sender part}
  31  {in order methods should be set/called}
  32  procedure Init; {set defaults for vars}
  33  public
  34  remote:tNetAddr;
  35  Mark:byte;
  36  MarkStart:tDateTime; {when the mark was started}
  37  MarkData:LongWord; {how much data sent}
  38  txLastSize:Word;
  39  siMark:byte;
  40  siNow,siWait:boolean;
  41  isTimeout:word;
  42  Cur:tTCSSe; {current values}
  43  Limit:tTCSSe; {maximum alloved}
  44  Initial:tTCSSe; {after start/timeout}
  45  minRateIF:single; {used after rate decrease}
  46  CanSend: procedure of object;   {called when transmit possible}
  47  procedure Start; {start the transmission}
  48  function MaxSize(req:word):word;
  49  procedure WriteHeaders(var s:tMemoryStream); {add headers before the data}
  50  procedure Send(var s:tMemoryStream);
  51  private
  52  {timer callbacks}
  53  procedure TransmitDelay;
  54  procedure Timeout;
  55  procedure OnCont(rmark:byte;rrate:real);
  56  procedure OnAck(rmark:byte;rsize:word);
  57  procedure Done; {unregister all callbacks}
  58  end;
  59
  60 procedure RegTxer(var t:tTCS);
  61 procedure DelTxer(var t:tTCS);
  62
  63 IMPLEMENTATION
  64 uses ServerLoop,SysUtils;
  65
  66 var Txers:array [0..31] of ^tTCS;
  67
  68 procedure RegTxer(var t:tTCS);
  69  var tn:byte;
  70  begin
  71  for tn:=0 to high(TXERS) do if txers[tn]=nil then break;
  72  assert(not assigned(txers[tn]));
  73  txers[tn]:=@t;
  74 end;
  75
  76 procedure DelTxer(var t:tTCS);
  77  var tn:byte;
  78  begin
  79  tn:=0; while tn<=high(TXERS) do if txers[tn]=@t then break else inc(tn);
  80  assert(tn<=high(TXERS));
  81  t.Done;
  82  txers[tn]:=nil;
  83 end;
  84
  85 type tTCSp=^tTCS;
  86 function GetTxer(const cource:tNetAddr):tTCSp;
  87  var tn:byte;
  88  begin
  89  result:=nil;
  90  tn:=0; while tn<=high(TXERS) do if txers[tn]^.remote=cource then break else inc(tn);
  91  if tn<=high(TXERS) then result:=txers[tn];
  92 end;
  93
  94 procedure RecvCont(msg:ServerLoop.tSMsg);
  95  var t:^tTCS;
  96  var rmark:byte;
  97  var rrate:longword;
  98  var rsize:word absolute rrate;
  99  var opcode:byte;
 100  begin
 101  t:=GetTxer(msg.source^);
 102  if not assigned(t) then exit;
 103  opcode:=msg.stream.ReadByte; {skip opcode}
 104  rmark:=msg.stream.ReadByte;
 105  if opcode=5 {periodic} then begin
 106   rrate:=msg.stream.ReadWord(4);
 107   t^.OnCont(rmark,rrate);
 108  end else if opcode=7 {explicit} then begin
 109   rsize:=msg.stream.ReadWord(2);
 110   t^.OnAck(rmark,rsize);
 111  end;
 112 end;
 113
 114 procedure tTCS.Init;
 115  begin
 116  remote.clear;
 117  //SizeIncScarcity:=20; {inverse probability of size experiment}
 118  Limit.Rate:=2*1024*1024*1024; {2GB}
 119  Limit.Size:=4096;
 120  Limit.RateIF:=1;
 121  Limit.SizeIF:=2;
 122  Initial.Rate:={20*}1024;
 123  Initial.Size:=32+5;
 124  Initial.RateIF:=0.5;
 125  Initial.SizeIF:=2;
 126  minRateIF:=0.01;
 127  CanSend:=nil;
 128 end;
 129
 130 procedure tTCS.Start; {start the transmission}
 131  begin
 132  Assert(assigned(CanSend) ); Assert(not remote.isnil);
 133  Cur:=Initial;
 134  mark:=Random(256); MarkData:=0;
 135  siMark:=0;
 136  isTimeout:=0;
 137  Shedule(80,@TransmitDelay);
 138  Shedule(2000,@Timeout);
 139 end;
 140
 141 function tTCS.MaxSize(req:word):word;
 142  begin
 143  if siNow
 144  then result:=round(cur.Size*(1+cur.SizeIF))
 145  else result:=cur.Size;
 146  dec(result,2);
 147  if result>req then result:=req;
 148 end;
 149
 150 procedure tTCS.WriteHeaders(var s:tMemoryStream);
 151  begin
 152  if siNow then begin
 153   s.WriteByte(6);{opcode}
 154   s.WriteByte(siMark);
 155  end else if isTimeout=0 then begin
 156   s.WriteByte(4);{opcode}
 157   s.WriteByte(mark);
 158  end else begin
 159   s.WriteByte(6);{opcode}
 160   s.WriteByte(simark);
 161  end;
 162 end;
 163
 164 procedure tTCS.Send(var s:tMemoryStream);
 165  begin
 166  ServerLoop.SendMessage(s.base^,s.length,remote);
 167  if MarkData=0 then begin
 168   MarkStart:=Now;
 169   MarkData:=1;
 170  end else MarkData:=MarkData+s.length;
 171  txLastSize:=s.length;
 172  siNow:=false;
 173 end;
 174
 175 procedure tTCS.OnCont(rmark:byte;rrate:real);
 176  var rnow:tDateTime;
 177  var RateFill:single;
 178  var txRate:real;
 179  var rxRate:real;
 180  begin
 181  if (rmark=Mark) then begin
 182   rnow:=Now;
 183   rxRate:=(rrate*64); {B/s}
 184   txRate:=MarkData/((rnow-MarkStart)*SecsPerDay);
 185   RateFill:=rxRate/txRate;
 186   write('speed: ',(rxRate/1024):1:3,'kB/s (',(RateFill*100):3:1,'% of ',txRate/1024:1:3,'), ');
 187   UnShedule(@Timeout);
 188   Shedule(2000,@Timeout);
 189   if RateFill<0.85 then begin
 190    write('limit, ');
 191    cur.Rate:=rxrate;
 192    cur.RateIF:=minRateIF;
 193   end else
 194   if (txRate/cur.Rate)<0.7 then begin
 195     write('3hard, ');
 196   end else begin
 197    write('pass, ');
 198    cur.Rate:=txRate*(cur.RateIF+1);
 199    if cur.Rate>limit.Rate then cur.Rate:=Limit.Rate
 200    else cur.RateIF:=cur.RateIF*2;
 201    if cur.RateIF>limit.RateIF then cur.RateIF:=Limit.RateIF;
 202   end;
 203   repeat mark:=Random(256) until (mark<>rMark);
 204   MarkData:=0;
 205   writeln('-> ',(Cur.Rate/1024):1:4,'kB/s if=',cur.RateIF:6:4);
 206   if siWait then begin
 207    cur.SizeIF:=cur.SizeIF/2;
 208   end;
 209   siMark:=0;
 210 end end;
 211
 212 procedure tTCS.OnAck(rmark:byte;rsize:word);
 213  begin
 214  if rmark<>simark then exit;
 215  if isTimeout>0 then begin
 216    Shedule(80,@TransmitDelay);
 217    isTimeout:=0;
 218  end else
 219  if rsize>cur.size then begin
 220    writeln('size inc to ',rsize);
 221    cur.SizeIF:=((rSize/cur.Size)-1)*2;
 222    if cur.SizeIF>Limit.SizeIF then Cur.SizeIF:=Limit.SizeIF;
 223    if (rsize/cur.rate)<=0.3 then cur.size:=rSize; {use new size for all transmit}
 224  end;
 225  if rsize>=cur.size then siWait:=false;
 226 end;
 227
 228 procedure tTCS.Timeout;
 229  begin
 230  cur:=initial;
 231  mark:=Random(256); MarkData:=0;
 232  siMark:=0;
 233  Inc(isTimeout);
 234  Shedule(80,@TransmitDelay);
 235  Shedule(5000,@Timeout);
 236 end;
 237
 238 procedure tTCS.TransmitDelay;
 239  var txwait:real;
 240  var burst:word;
 241  begin
 242  txLastSize:=0;
 243  txwait:=0;
 244  burst:=0;
 245  if (siMark=0)and(cur.Size<limit.Size){and(random(10)=0)}and(istimeout=0) then begin
 246   siNow:=true;
 247   siWait:=true;
 248   siMark:=random(255)+1;
 249  end;
 250  repeat
 251   CanSend;
 252   if txLastSize=0 then exit;{pause}
 253   if (isTimeout>0) then exit;
 254   //txwait:=txwait+(txLastSize/cur.rate);
 255   txwait:=(MarkData/cur.Rate)-((Now-MarkStart)*SecsPerDay);
 256   inc(burst);
 257   siNow:=false;
 258  until (txwait>0.02)or(burst>200);
 259  if txwait<0.02 then txwait:=0.01;
 260  //writeln(txwait:1:3,burst);
 261  Shedule(round(txwait*1000),@TransmitDelay);
 262 end;
 263
 264 procedure tTCS.Done; {unregister all callbacks}
 265  begin
 266  UnShedule(@TransmitDelay);
 267  UnShedule(@Timeout);
 268 end;
 269
 270 BEGIN
 271  FillByte(txers,sizeof(txers),0); {make'em nil}
 272  SetMsgHandler(5,@RecvCont);
 273  SetMsgHandler(7,@RecvCont);
 274 END.