Download.pas

   1 unit Download;
   2 {manage downloads}
   3
   4 INTERFACE
   5 uses NetAddr,
   6      ServerLoop,opcode,MemStream;
   7 {
   8  Same idea here as upmgr. Have an tAggr for each peer reporting speed and
   9  tJob for each file request saving to file and reqesting missed segments.
  10  The aggr should have limit of paralele jobs. Jobs will first be linked in
  11  tAggr queue and then started as slots become available.
  12
  13  After node restart, notify requester with a No-Source error. But should
  14  not be forgotten. More advanced DM could consult CHK or Category the file
  15  was found.
  16 }
  17 type
  18  tJob_ptr=pointer;//^tJob;
  19  tAggr=object
  20   Rate:Real;
  21   ByteCnt:LongWord;
  22   DgrCnt:LongWord;
  23   CurMark,PrvMark:byte;
  24   StartT:tMTime;
  25   Jobs: array [0..15] of tJob_ptr;
  26   refs:byte;
  27   ChanOfs:byte;
  28   DgrCntCheck:LongWord;
  29   remote:tNetAddr;
  30   procedure Init(const src:tNetAddr);
  31   procedure MsgDATA(sz:Word;mark:byte);
  32   procedure MsgIMME(sz:Word;mark:byte);
  33   procedure Recv(msg:tSMsg);
  34   procedure Periodic;
  35   procedure Done;
  36  end;
  37 IMPLEMENTATION
  38
  39 procedure tAggr.Init(const src:tNetAddr);
  40  begin
  41  Rate:=0;
  42  ByteCnt:=0;
  43  DgrCnt:=0;
  44  CurMark:=0;PrvMark:=0;
  45  StartT:=mNow;
  46  refs:=high(Jobs); while refs>0 do begin Jobs[refs]:=nil; dec(refs) end;
  47  ChanOfs:=Random(255-high(Jobs));
  48  DgrCntCheck:=0;
  49  Shedule(5000,@Periodic);
  50  remote:=src;
  51  SetMsgHandler(opcode.tcdata,src,@Recv);
  52  SetMsgHandler(opcode.tcdataimm,src,@Recv);
  53 end;
  54
  55 procedure tAggr.Recv(msg:tSMsg);
  56  var op:byte;
  57  var mark:byte;
  58  var chan:byte;
  59  begin
  60  op:=msg.stream.readbyte;
  61  mark:=msg.stream.readbyte;
  62  if op=opcode.tcdataimm then MsgIMME(msg.length,mark);
  63  MsgDATA(msg.length,mark);
  64  chan:=msg.stream.readbyte;
  65  //delegate to others todo
  66 end;
  67
  68 procedure tAggr.MsgIMME(sz:Word; mark:byte);
  69  var r:tMemoryStream;
  70  var buf:array [1..4] of byte;
  71  begin
  72  r.Init(@buf,0,sizeof(buf));
  73  r.WriteByte(opcode.tceack);
  74  r.WriteByte(mark);
  75  r.WriteWord(sz,2);
  76  SendMessage(r.base^,r.length,remote);
  77 end;
  78
  79 procedure tAggr.MsgDATA(sz:Word; mark:byte);
  80  var r:tMemoryStream;
  81  var rateb: DWord; {BytesPerSecond shr 6 (=64)}
  82  var buf:array [1..6] of byte;
  83  begin
  84  if mark<>PrvMark then begin
  85   if mark<>CurMark then begin
  86    PrvMark:=CurMark;
  87    CurMark:=mark;
  88    StartT:=mNow;
  89    ByteCnt:=1;
  90    DgrCnt:=1;
  91   end else begin Inc(ByteCnt,sz); Inc(DgrCnt) end;
  92   inc(DgrCntCheck);
  93  end;
  94  if DgrCnt<8 then exit;
  95  if (mnow-Startt)<400 then exit;
  96  rate:=(ByteCnt/(mNow-StartT))*1000;
  97  writeln('Rate: ',(rate/1024):7:1);
  98  rateb:=round(rate/64);
  99  StartT:=mNow;
 100  ByteCnt:=1;
 101  r.Init(@buf,0,sizeof(buf));
 102  r.WriteByte(opcode.tccont);
 103  r.WriteByte(mark);
 104  r.WriteWord(rateb,4);
 105  SendMessage(r.base^,r.length,remote);
 106 end;
 107
 108 procedure tAggr.Periodic;
 109  begin
 110  if DgrCntCheck>1 then begin
 111   DgrCntCheck:=0;
 112   Shedule(5000,@Periodic);
 113  exit end;
 114  writeln('Periodic check failed');
 115  //todo do
 116 end;
 117
 118 procedure tAggr.Done;
 119  begin
 120  UnShedule(@Periodic);
 121 end;
 122
 123 END.