Download.pas

   1 unit Download;
   2 {manage downloads}
   3
   4 INTERFACE
   5 uses NetAddr,
   6      ServerLoop,opcode,MemStream
   7     ,Store1
   8     ,Chat
   9     ,Sha1
  10     ;
  11 {
  12  Same idea here as upmgr. Have an tAggr for each peer reporting speed and
  13  tJob for each file request saving to file and reqesting missed segments.
  14  The aggr should have limit of paralele jobs. Jobs will first be linked in
  15  tAggr queue and then started as slots become available.
  16
  17  After node restart, notify requester with a No-Source error. But should
  18  not be forgotten. More advanced DM could consult CHK or Category the file
  19  was found.
  20 }
  21 type
  22  pDownloadJob=^tDownloadJob;
  23  tDownloadJob=object
  24   total,done:LongWord;
  25   missc:LongWord;
  26   state:(stStop,stActive,stDone,stError,stLocalError);
  27   error:byte;
  28   error2:byte;
  29   fid:tFID;
  30   procedure Start;
  31   procedure Free;
  32   procedure Abort;
  33   protected
  34   refc:byte;
  35  end;
  36 function GetJob(const fid:tFID):pDownloadJob;
  37 function NewJob(const source:tNetAddr; const fid: tFID):pDownloadJob;
  38
  39 IMPLEMENTATION
  40 {TODO: cache for chats}
  41 type
  42  tAggr_ptr=^tAggr;
  43  tJob=object(tDownloadJob)
  44   aggr:tAggr_ptr;
  45   ix:byte;
  46   so:tStoreObjectInfo;
  47   ch:^tChat;
  48   active:boolean;
  49   procedure Init(const source:tNetAddr; const ifid: tFID);
  50   procedure MsgDATA(base,length:LongWord; data:pointer);
  51   procedure Start;
  52   procedure Free;
  53   procedure Abort;
  54   procedure Close;
  55   private
  56   HighestRequestBase, RemoteSkipTo :LongWord;
  57   procedure ReplyOPEN(msg:tSMsg; data:boolean);
  58   procedure ReplyLSEG(msg:tSMsg; data:boolean);
  59   procedure HandleFAIL(r:tMemoryStream);
  60   procedure HandleEPROTO(r:tMemoryStream);
  61   procedure ReplyDONE(msg:tSMsg; data:boolean);
  62   procedure MakeRequest;
  63  end;
  64 {$I DownloadTC.pas}
  65
  66 function GetJob(const fid:tFID):pDownloadJob;
  67  var a:^tAggr;
  68  var i:byte;
  69  var p:^pointer;
  70  begin
  71  p:=@AggrChain;
  72  a:=AggrChain;
  73  while assigned(a) do begin
  74   for i:=0 to high(tAggr.Jobs) do if assigned(a^.Jobs[i]) then begin
  75    if CompareWord(a^.Jobs[i],fid,10)=0 then begin
  76     GetJob:=a^.Jobs[i];
  77     Inc(a^.refs);
  78     assert(a^.Jobs[i]^.ix=i);
  79     assert(a^.Jobs[i]^.aggr=a);
  80     exit;
  81    end else break{for};
  82   end;
  83   p:=@a^.next; a:=p^;
  84  end;
  85  GetJob:=nil;
  86 end;
  87
  88 function NewJob(const source:tNetAddr; const fid: tFID):pDownloadJob;
  89  begin
  90  result:=GetJob(fid);
  91  if assigned(result) then exit;
  92  result:=GetMem(sizeof(tJob));
  93  tJob(pointer(result)^).init(source,fid);
  94 end;
  95
  96 procedure tJob.Init(const source:tNetAddr; const ifid: tFID);
  97  var dw:^tAggr;
  98  begin
  99  refc:=1;
 100  error:=0;
 101  error2:=0;
 102  fid:=ifid;
 103  active:=false;
 104  so.Open(fid);
 105  done:=0;
 106  missc:=0;
 107  if so.final then begin
 108   done:=total;
 109   state:=stDone;
 110   aggr:=nil;
 111   so.Close;
 112  exit end;
 113  state:=stStop;
 114  {todo: initialize Done}
 115  so.EnableWrite(fid);
 116  if so.rc<>0 then begin
 117   state:=stLocalError;
 118   error:=255;
 119   error2:=so.rc;
 120  exit end;
 121  dw:=GetAggr(source);
 122  if not assigned(aggr) then begin
 123   new(dw);
 124   dw^.Init(source);
 125   ix:=0;
 126  end else begin
 127   ix:=0; while (ix<high(tAggr.Jobs)) and (dw^.Jobs[ix]=nil) do inc(ix);
 128   assert(dw^.Jobs[ix]=nil);{todo}
 129  end;
 130  aggr:=dw;
 131  dw^.Jobs[ix]:=@self;
 132  inc(dw^.refs);
 133  if state=stStop then begin
 134   New(CH);
 135   ch^.Init(source);
 136  end;
 137  //assert( ((state=stStop)and(rlen>0))or((state=stDone)and(rlen=0)) );
 138 end;
 139
 140 procedure tJob.Start;
 141  var s:tMemoryStream;
 142  begin
 143  writeln('Download: job start');
 144  assert( state=stStop );
 145  state:=stActive;
 146  ch^.Callback:=@ReplyOPEN;
 147  ch^.streaminit(s,33);
 148  {service}s.WriteByte(opcode.upFileServer);
 149  {channel}s.WriteByte(ix);
 150  {opcode }s.WriteByte(opcode.upOPEN);
 151  {file   }s.Write(fid,20);
 152  {todo: request first segment}
 153  ch^.Send(s);
 154 end;
 155 procedure tJob.ReplyOPEN(msg:tSMsg; data:boolean);
 156  var r:tMemoryStream absolute msg.stream;
 157  var op:byte;
 158  var rsize:LongWord;
 159  var rfinal:byte;
 160  begin
 161  if not data then exit;
 162  op:=msg.stream.ReadByte;
 163  {valid responses: INFO FAIL EPROTO}
 164  if op=upFAIL then HandleFAIL(r)
 165  else if op=upINFO then begin
 166   rsize  :=r.ReadWord(4);
 167   rfinal :=r.readbyte;
 168   writeln('INFO size=',rsize,' final=',rfinal);
 169   self.total:=rsize;
 170   if so.length<>rsize then begin
 171    if so.length>0 then writeln('Download: warning: size mismatch!');
 172    so.SetFLength(rsize);
 173   end;
 174   MakeRequest;
 175  end
 176  else HandleEPROTO(r);
 177 end;
 178 {Strategy for fixing holes without waiting for DONE message:
 179  * stuff all small segments to LSEG
 180  * put large one at end
 181  * when datagrams from the large arrive, Repeat
 182 }
 183
 184 procedure tJob.MakeRequest;
 185  var s:tMemoryStream;
 186  var b,l,trl:LongWord;
 187  var cnt:byte;
 188  var mst:Pointer;
 189  const clim=83886080;
 190  begin
 191  write('Download: job MakeRequest');
 192  mst:=nil;
 193  trl:=0;
 194  cnt:=0;
 195  ch^.Callback:=@ReplyLSEG;
 196  ch^.streaminit(s,180);
 197  s.WriteByte(upLSEG);
 198  repeat
 199   {todo: skipto}
 200   so.GetMiss(b,l,mst);
 201   if l=0 then break;
 202   if (trl+l)>clim then l:=clim-trl;
 203   write(' ',b,'+',l);
 204   s.WriteWord(b,4);
 205   s.WriteWord(l,4);
 206   inc(trl,l);
 207   inc(cnt);
 208  until (s.WrBufLen<8)or(trl>=clim);
 209  writeln(' for ',trl,'B in ',cnt,' ',s.Length);
 210  if trl=0 then begin
 211   state:=stDone;
 212   writeln('Verifu!!!!!');
 213   so.VerifyAndReset;
 214   if not so.final then begin
 215    state:=stLocalError;
 216    error:=254;
 217    writeln('Verifu Faialed!!!!!');
 218   end;
 219   Aggr^.Stop(ix);
 220   Close;
 221  exit end;
 222  ch^.Send(s);
 223  if cnt>1 then HighestRequestBase:=b else HighestRequestBase:=$FFFFFFFF;
 224 end;
 225 procedure tJob.ReplyLSEG(msg:tSMsg; data:boolean);
 226  var r:tMemoryStream absolute msg.stream;
 227  var op:byte;
 228  var avail:LongWord;
 229  begin
 230  if not data then exit;
 231  op:=msg.stream.ReadByte;
 232  {valid responses: SEGOK UNAVL FAIL}
 233  if op=upFAIL then HandleFAIL(r)
 234  else if op=upUNAVL then begin
 235   avail:=msg.stream.ReadWord(4);
 236   writeln('Download: job ReplyLSEG: UNAVL avail=',avail);
 237   if avail=0 then begin
 238    state:=stLocalError;
 239    error:=253;
 240    Close;
 241   end else begin
 242    RemoteSkipTo:=avail;
 243    MakeRequest;
 244   end;
 245  end
 246  else if op=upSEGOK then begin
 247   avail:=msg.stream.ReadWord(4);
 248   MissC:=avail;
 249   writeln('Download: job ReplyLSEG: SEGOK avail=',avail);
 250   aggr^.Start(ix);
 251   ch^.Callback:=@ReplyDONE;
 252   ch^.Ack;
 253  end
 254  else if op=upDONE then begin
 255  end {ignore, done is sent async so it can hang in-flight a bit}
 256  else HandleEPROTO(r);
 257 end;
 258 procedure tJob.ReplyDONE(msg:tSMsg; data:boolean);
 259  var r:tMemoryStream absolute msg.stream;
 260  var op:byte;
 261  begin
 262  if not data then exit;
 263  op:=msg.stream.ReadByte;
 264  {valid responses: DONE}
 265  if op=upDONE then begin
 266   writeln('Download: ReplyDONE: DONE, miss=',MissC);
 267   MakeRequest;
 268  end else HandleEPROTO(r);
 269 end;
 270 procedure tJob.HandleFAIL(r:tMemoryStream);
 271  begin
 272  writeln('Download: FAIL');
 273  state:=stError;
 274  try
 275   error:=r.readByte;
 276   error2:=r.readByte;
 277  except end;
 278  Close;
 279 end;
 280 procedure tJob.HandleEPROTO(r:tMemoryStream);
 281  begin
 282  r.Seek(r.position-1);
 283  try error2:=r.ReadByte; except end;
 284  writeln('Download: EPROTO ',error2);
 285  state:=stLocalError;
 286  error:=252;
 287  Close;
 288 end;
 289
 290 procedure tJob.Close;
 291  var s:tMemoryStream;
 292  begin
 293  ch^.streaminit(s,2);
 294  {opcode }s.WriteByte(opcode.upClose);
 295  ch^.Send(s);
 296  ch^.Close;
 297  writeln('chat close');
 298  aggr^.Stop(ix);
 299 end;
 300 procedure tJob.Abort;
 301  begin
 302  assert(state=stActive);
 303  state:=stLocalError;
 304  error:=251;
 305  Close;
 306 end;
 307
 308 procedure tJob.MsgDATA(base,length:LongWord; data:pointer);
 309  begin
 310  so.WriteSeg(base,length,data);
 311  done:=done+length;
 312  if MissC<=length
 313  then MakeRequest
 314  else dec(MissC,length);
 315  if base>=HighestRequestBase then {TODO, last segment in list, MakeRequest};
 316 end;
 317
 318 procedure tJob.Free;
 319  begin
 320  Dec(refc);
 321  if refc=0 then begin
 322   writeln('Download: job closing');
 323   if state=stStop then Close
 324   else if state=stActive then Abort;
 325   {tu ja EAV, nieco s pointermi}
 326   if assigned(aggr) then begin
 327    aggr^.Jobs[ix]:=nil;
 328    dec(aggr^.refs);
 329    if aggr^.refs=0 then aggr^.Done;
 330   end;
 331   if state<>stDone then so.Close;
 332   FreeMem(@self,sizeof(tJob));
 333  end else writeln('not closing ',refc);
 334 end;
 335
 336 procedure tDownloadJob.Start;
 337  begin
 338  tJob(pointer(@self)^).Start;
 339 end;
 340 procedure tDownloadJob.Abort;
 341  begin
 342  tJob(pointer(@self)^).Abort;
 343 end;
 344 procedure tDownloadJob.Free;
 345  begin
 346  tJob(pointer(@self)^).Free;
 347 end;
 348 END.