Chat.pas

   1 unit Chat;
   2 {
   3  Implement two-way realiable acked lock-step protocol
   4 }
   5 INTERFACE
   6 uses NetAddr,ServerLoop,MemStream;
   7
   8 type tChat=object
   9  remote:tNetAddr;
  10  opcode:byte;
  11  txSeq:Word;
  12  rxSeq:Word;
  13  rxAcked:boolean;
  14  closed:boolean;
  15  RTT:LongWord;{in ms}
  16  callback: procedure(msg:tSMsg; data:boolean) of object; {client must maintain active chats}
  17  TMhook  : procedure(willwait:LongWord      ) of object;
  18  DisposeHook: procedure of object; {called instead of freeing self}
  19  procedure Init(const iremote:tNetAddr);
  20  procedure AddHeaders(var s:tMemoryStream);
  21  procedure StreamInit(var s:tMemoryStream; l:word);
  22  procedure Send(s:tMemoryStream);
  23   {the stream can be invalidated, but the buffer must not be modified or freed}
  24  procedure Ack;
  25  procedure Close;
  26  private
  27  txPk:pointer; txLen:word; {last sent, not acked msg}
  28  txTime:tDateTime;
  29  procedure InitFrom(const iremote:tNetAddr; iopcode:byte);
  30  procedure Done;
  31  procedure Resend;
  32  procedure OnReply(msg:tSMsg);
  33 end;
  34
  35 type tChatHandler=procedure(var nchat:tChat; msg:tSMsg);
  36 procedure SetChatHandler(initcode:byte; handler:tChatHandler);
  37
  38 { download manager create FileRequest
  39  File Request open chat session to server
  40  upmgr accepts chat and send reply
  41  FileRequest acks, chat is then closed after TimeWait
  42  upmgr starts TC transfer
  43  transfer finished, upmgr send new Chat to FileRequest
  44  FileRequest acks, chat is closed on both ends
  45  FileRequest can open new chat if blocks are missing
  46
  47  => chat msgs must be created with New, disposed by Chat
  48  => there is TimeWait, no references are to the Chat, except Sheduler, it Disposes itself.
  49 }
  50
  51 { Chats are the HiMsg. Use hash table from ServerLoop, works for HiMsg too. }
  52
  53 IMPLEMENTATION
  54 uses SysUtils;
  55 procedure tChat.Init(const iremote:tNetAddr);
  56  begin
  57  remote:=iremote;
  58  opcode:=128+Random(128);
  59  while ServerLoop.IsMsgHandled(opcode,remote) do inc(opcode);
  60  InitFrom(remote,opcode);
  61 end;
  62 procedure tChat.InitFrom(const iremote:tNetAddr; iopcode:byte);
  63  begin
  64  remote:=iremote;
  65  opcode:=iopcode;
  66  SetMsgHandler(opcode,remote,@OnReply);
  67  txSeq:=0;
  68  rxSeq:=0;
  69  rxAcked:=true; {to not ack pk 0}
  70  closed:=false;
  71  txPk:=nil;
  72  txLen:=0;
  73  callback:=nil;
  74  TMhook:=nil;
  75  DisposeHook:=nil;
  76  RTT:=200; {a default for timeouts}
  77  txTime:=0;
  78 end;
  79 {struct
  80  opcode:byte
  81  seq:2
  82  ack_seq:2
  83  data:xx
  84 }
  85
  86 procedure tCHat.AddHeaders(var s:tMemoryStream);
  87  begin s.skip(5) end;
  88 procedure tChat.StreamInit(var s:tMemoryStream; l:word);
  89  begin
  90  s.Init(GetMem(l+5),0,l+5);
  91  AddHeaders(s);
  92 end;
  93
  94 procedure tChat.Send(s:tMemoryStream);
  95  begin
  96  if txLen>0 then begin
  97   FreeMem(txPk,txLen);
  98   UnShedule(@Resend);
  99  end;
 100  //assert(assigned(callback));
 101  Inc(txSeq);
 102  s.Seek(0);
 103  s.WriteByte(opcode);
 104  s.WriteWord(txSeq,2);
 105  if not rxAcked then begin
 106   s.WriteWord(rxSeq,2);
 107   rxAcked:=true;
 108  end else s.WriteWord(0,2);
 109  txPk:=s.base;
 110  txLen:=s.Length;
 111  ServerLoop.SendMessage(txPk^,txLen,remote);
 112  ServerLoop.Shedule(RTT*2,@Resend);
 113  txTime:=Now;
 114 end;
 115
 116 procedure tChat.Ack;
 117  var s:tMemoryStream;
 118  begin
 119  if not rxAcked then begin
 120   s.Init(GetMem(5),0,5);
 121   s.WriteByte(opcode);
 122   s.WriteWord(0,2);
 123   s.WriteWord(rxSeq,2);
 124   ServerLoop.SendMessage(s.base^,s.length,remote);
 125   FreeMem(s.base,s.length);
 126   rxAcked:=true;
 127  end;
 128 end;
 129
 130 procedure tChat.Close;
 131  begin
 132  assert(not closed);
 133  Ack;
 134  closed:=true;
 135  callback:=nil; {avoid calling}
 136  tmhook:=nil;
 137  //writeln('Chat: closing');
 138  if txLen=0 {no packets in flight} then begin
 139   Shedule(15000{todo},@Done); {wait for something lost}
 140  end;
 141 end;
 142
 143 procedure tChat.Done;
 144  begin
 145  if txLen>0 then FreeMem(txPk,txLen);
 146  SetMsgHandler(opcode,remote,nil);
 147  if assigned(DisposeHook) then DisposeHook
 148  else FreeMem(@self,sizeof(self));
 149  //writeln('Chat: closed');
 150 end;
 151
 152 procedure tChat.Resend;
 153  {timeout waiting for ack}
 154  begin
 155  {resend and reshedule}
 156  if txLen=0 then exit;
 157  txTime:=0;
 158  if RTT<1 then RTT:=2;
 159  RTT:=RTT*2;
 160  if assigned(TMhook) and (not closed) then begin
 161   TMhook(RTT);
 162   if closed then begin
 163    Done; {if hook decided to close then abort}
 164    exit;
 165   end;
 166  end;
 167  if closed and (RTT<400) then RTT:=400;
 168  if (RTT>=5000) and closed then begin
 169   Done {give up}
 170  end else begin
 171   {finally resend the msg}
 172   //writeln('Chat: retry');
 173   ServerLoop.SendMessage(txPk^,txLen,remote);
 174   ServerLoop.Shedule(RTT,@Resend);
 175  end;
 176 end;
 177
 178 procedure tChat.OnReply(msg:tSMsg);
 179  var seq,aseq:Word;
 180  var s:tMemoryStream;
 181  begin
 182  msg.stream.skip(1{opcode});
 183  seq:=msg.stream.ReadWord(2);
 184  aseq:=msg.stream.ReadWord(2);
 185  if aseq>0 then {ack of our msg} begin
 186   if (aseq=txSeq)and(txLen>0) {it is current} then begin
 187    if txTime>0 then RTT:=Round((Now-txTime)*MsecsPerDay);
 188    FreeMem(txPk,txLen);
 189    UnShedule(@Resend);
 190    if Closed then Shedule(5,@Done);
 191    TxLen:=0;
 192    txPk:=nil;
 193    if assigned(callback) then callback(msg,false);
 194   end else {write(' old-ack')it is ack of old data, do nothing};
 195  end;
 196  if seq>0 then {some data} begin
 197   if seq<=rxSeq then {remote didnt get our ack} begin
 198    s.Init(GetMem(5),0,5);
 199    s.WriteByte(opcode);
 200    s.WriteWord(0,2);
 201    s.WriteWord(rxSeq,2);
 202    ServerLoop.SendMessage(s.base^,s.length,remote);
 203    FreeMem(s.base,s.length);
 204    if seq=rxSeq then rxacked:=true;
 205   end else begin
 206    {some useful data!}
 207    rxSeq:=seq;
 208    rxAcked:=false;
 209    if assigned(callback) then callback(msg,true);
 210   end;
 211  end;
 212 end;
 213
 214 var ChatHandlers: array [1..32] of tChatHandler;
 215
 216 procedure SetChatHandler(initcode:byte; handler:tChatHandler);
 217  begin
 218  assert(ChatHandlers[initcode]=nil);
 219  ChatHandlers[initcode]:=handler;
 220 end;
 221
 222 procedure OnHiMsg(msg:tSMsg);
 223  {new chat was received!}
 224  var opcode:byte;
 225  var seq,aseq:word;
 226  var hnd:tChatHandler;
 227  var nchat:^tChat;
 228  var ix:byte;
 229  begin
 230  opcode:=msg.stream.ReadByte;
 231  assert(not IsMsgHandled(opcode,msg.source^));
 232  seq:=msg.stream.ReadWord(2);
 233  aseq:=msg.stream.ReadWord(2);
 234  if (seq<>1)or(aseq>0) then exit; {invalid initial state}
 235  ix:=msg.stream.ReadByte;
 236  if (ix<1)or(ix>high(ChatHandlers)) then exit;
 237  hnd:=ChatHandlers[ix];
 238  if not assigned(hnd) then raise eXception.Create('No handler for initcode '+IntToStr(ix));
 239  msg.stream.seek(msg.stream.position-1);{unskip the initcode}
 240  nchat:=GetMem(sizeof(tChat));
 241  nchat^.InitFrom(msg.Source^,opcode);
 242  nchat^.rxacked:=false;
 243  nchat^.rxSeq:=1;
 244  hnd(nchat^,msg);
 245 end;
 246
 247 BEGIN
 248  FillChar(ChatHandlers,sizeof(chathandlers),0);
 249  ServerLoop.SetHiMsgHandler(@OnHiMsg);
 250 END.