ServerLoop.pas

   1 UNIT ServerLoop;
   2
   3 INTERFACE
   4 uses MemStream,NetAddr,UnixType;
   5
   6 procedure Main;
   7
   8 {#Message handling#}
   9 type tSMsg=object
  10  Source: ^tNetAddr;
  11  Length: {Long}Word;
  12  Data: pointer;
  13  stream: tMemoryStream;
  14  channel: word;
  15  end;
  16 type tMessageHandler=procedure(msg:tSMsg);
  17 procedure SetMsgHandler(OpCode:byte; handler:tMessageHandler);
  18 procedure SetHiMsgHandler(handler:tMessageHandler);
  19
  20 procedure SendMessage(const data; len:word; const rcpt:tNetAddr );
  21 {procedure SendReply(const data; len:word; const rcpt:tSMsg );}
  22 procedure SendMessage(const data; len:word; const rcpt:tNetAddr; channel:word );
  23
  24 {#Sheduling and watching#}
  25 type tFDEventHandler=procedure(ev:Word) of object;
  26 type tOnTimer=procedure of object;
  27 procedure WatchFD(fd:tHandle; h:tFDEventHandler);
  28 procedure Shedule(timeout{ms}: LongWord; h:tOnTimer);
  29 procedure UnShedule(h:tOnTimer);
  30  {note unshed will fail when called from OnTimer proc}
  31
  32 type tObjMessageHandler=procedure(msg:tSMsg) of object;
  33 {deliver message from peer to the object}
  34 procedure SetMsgHandler(OpCode:byte; from:tNetAddr; handler:tObjMessageHandler); overload;
  35 function IsMsgHandled(OpCode:byte; from:tNetAddr):boolean;
  36
  37 function OptIndex(o:string):word;
  38 function OptParamCount(o:word):word;
  39
  40 type tTimeVal=UnixType.timeval;
  41 var iNow:tTimeVal;
  42
  43 IMPLEMENTATION
  44
  45 USES SysUtils,Sockets,BaseUnix
  46      ,Unix
  47      ;
  48
  49 {aim for most simple implementation, since could be extended anytime}
  50
  51 var s_inet:tSocket;
  52
  53 type tPollTop=0..7;
  54 var pollArr: packed array [tPollTop] of tPollFd;
  55 type tFdHndDsc=record
  56  cb: tFDEventHandler; {proc+object}
  57  end;
  58 var pollHnd: array [tPollTop] of tFdHndDsc;
  59 var pollTop: tPollTop;
  60
  61 var hnd: array [1..36] of tMessageHandler;
  62 var HiHnd: tMessageHandler;
  63
  64 type tSheduled_ptr=^tSheduled; tSheduled=record
  65  left:LongWord;
  66  cb:tOnTimer;
  67  next:tSheduled_ptr;
  68  end;
  69 var ShedTop: ^tSheduled;
  70 var ShedUU: ^tSheduled;
  71 var LastShed: UnixType.timeval;
  72 var PollTimeout:LongInt;
  73
  74 procedure SC(fn:pointer; retval:cint);
  75  begin
  76   if retval < 0 then begin
  77    raise eXception.Create(Format('Socket error %d operation %P',[SocketError,fn]));
  78   end;
  79  end;
  80
  81 procedure s_SetupInet;
  82  var bind_addr:tInetSockAddr;
  83  var turnon:cint;
  84  var oi:word;
  85  begin
  86   with bind_addr do begin
  87    sin_family:=AF_INET;
  88    oi:=OptIndex('-port');
  89    if oi=0 then sin_port:=htons(3511)
  90    else begin
  91     assert(OptParamCount(oi)=1);
  92     sin_port:=htons(StrToInt(paramstr(oi+1)));
  93    end;
  94    sin_addr.s_addr:=0; {any}
  95    s_inet:=fpSocket(sin_family,SOCK_DGRAM,IPPROTO_UDP);
  96    SC(@fpSocket,s_inet);
  97    turnon:=IP_PMTUDISC_DO;
  98    SC(@fpsetsockopt,fpsetsockopt(s_inet, IPPROTO_IP, IP_MTU_DISCOVER, @turnon, sizeof(turnon)));
  99   end;
 100   SC(@fpBind,fpBind(s_inet,@bind_addr,sizeof(bind_addr)));
 101   with PollArr[0] do begin
 102    fd:=s_inet;
 103    events:=pollIN;
 104    revents:=0;
 105   end;
 106  end;
 107
 108 var Terminated:boolean=false;
 109
 110 procedure SendMessage(const data; len:word; const rcpt:tSockAddrL );
 111  begin
 112  {SC(@fpsendto,}fpsendto(s_inet,@data,len,0,@rcpt,sizeof(sockaddr_in)){)};
 113 end;
 114 procedure SendMessage(const data; len:word; const rcpt:tNetAddr );
 115  var sa:tSockAddrL;
 116  begin
 117  rcpt.ToSocket(sa);
 118  SendMessage(data,len,sa);
 119 end;
 120 procedure SendMessage(const data; len:word; const rcpt:tNetAddr; channel:word );
 121  begin
 122  SendMessage(data,len,rcpt);
 123  {todo: optimization??}
 124 end;
 125
 126 procedure SignalHandler(sig:cint);CDecl;
 127  begin
 128   writeln;
 129   if terminated then raise eControlC.Create('CtrlC DoubleTap') ;
 130   Terminated:=true;
 131   writeln('Shutdown requested');
 132  end;
 133
 134 {index=iphash+opcode}
 135 type tPeerTableBucket=record
 136  opcode:byte;
 137  remote:tNetAddr;
 138  handler:tObjMessageHandler;
 139 end;
 140 var PT:array [0..255] of ^tPeerTableBucket;
 141 var PT_opcodes: set of 1..high(hnd);
 142
 143 function FindPT(opcode:byte; addr:tNetAddr):Word; { $FFFF=fail}
 144  var i,o:word;
 145  begin
 146  i:=(addr.hash+opcode) mod high(PT); {0..63}
 147  for o:=0 to high(PT) do begin
 148   result:=(i+o) mod high(PT);
 149   if not assigned(PT[result]) then break;
 150   if (PT[result]^.opcode=opcode) and (PT[result]^.remote=addr) then exit;
 151  end;
 152  result:=$FFFF;
 153 end;
 154
 155 function IsMsgHandled(OpCode:byte; from:tNetAddr):boolean;
 156  begin result:=FindPT(opcode,from)<>$FFFF end;
 157
 158 procedure UnSetMsgHandler(const from:tNetAddr; opcode:byte);
 159  var i,h:word;
 160  begin
 161  h:=FindPT(opcode,from);
 162  if h=$FFFF then exit;
 163  Dispose(PT[h]);
 164  PT[h]:=nil;
 165  {go reverse exit on null, hash them, match: move to H and stop}
 166  if h=0 then i:=high(PT) else i:=h-1;
 167  while (i<>h)and assigned(PT[i]) do begin
 168   if (PT[i]^.remote.hash+PT[i]^.opcode)=h then begin
 169    PT[h]:=PT[i];
 170    PT[i]:=nil;
 171    break;
 172   end;
 173   if i=0 then i:=high(PT) else dec(i);
 174  end;
 175 end;
 176
 177 procedure SetMsgHandler(OpCode:byte; from:tNetAddr; handler:tObjMessageHandler);
 178  var h,o,i:word;
 179  begin
 180  UnSetMsgHandler(from,opcode);
 181  if handler=nil then exit;
 182  h:=(from.hash+opcode) mod high(PT);
 183  for o:=0 to high(PT) do begin
 184   i:=(h+o) mod high(PT);
 185   if not assigned(PT[i]) then break;
 186  end;
 187  New(PT[i]);
 188  PT[i]^.opcode:=OpCode;
 189  PT[i]^.remote:=from;
 190  PT[i]^.handler:=handler;
 191  if opcode<=high(hnd) then Include(PT_opcodes,opcode);
 192 end;
 193
 194 {do not waste stack on statics}
 195  var EventsCount:integer;
 196  var Buffer:array [1..1024] of byte;
 197  var pkLen:LongWord;
 198  var From:tSockAddrL; {use larger struct so everything fits}
 199  var FromLen:LongWord;
 200  var FromG:tNetAddr;
 201  var curhnd:tMessageHandler;
 202  var curhndo:tObjMessageHandler;
 203  var Msg:tSMsg;
 204  var tp:tPollTop;
 205
 206 function DoSock(var p:tPollFD):boolean;
 207  var ptidx:word;
 208  begin
 209  curhnd:=nil;
 210  curhndo:=nil;
 211  result:=false;
 212  ptidx:=$FFFF;
 213  if (p.revents and pollIN)=0 then exit else  result:=true;
 214  FromLen:=sizeof(From);
 215  pkLen:=fprecvfrom(p.FD,@Buffer,sizeof(Buffer),0,@from,@fromlen);
 216  SC(@fprecvfrom,pkLen);
 217  p.revents:=0;
 218  FromG.FromSocket(from);
 219  Msg.Source:=@FromG; {!thread}
 220  Msg.Length:=pkLen;
 221  Msg.Data:=@Buffer; {!thread}
 222  Msg.stream.Init(@Buffer,pkLen,sizeof(Buffer));
 223  Msg.channel:=0; {!multisocket}
 224  if Buffer[1]>=128 then curhnd:=HiHnd else if Buffer[1]<=high(hnd) then curhnd:=hnd[Buffer[1]];
 225  if (Buffer[1]>high(hnd))or(Buffer[1] in PT_opcodes) then begin
 226   ptidx:=FindPT(Buffer[1],FromG);
 227   if ptidx<$FFFF then curhndo:=PT[ptidx]^.handler;
 228  end;
 229 end;
 230
 231 procedure ShedRun;
 232  var cur:^tSheduled;
 233  var pcur:^pointer;
 234  var now:UnixType.timeval absolute iNow;
 235  var delta:LongWord;
 236  var tasks:word;
 237  begin
 238  {Sheduling}
 239  {fixme: proste niak aby to šlo vymazať z callbacku}
 240  {prejdi ich od zaciatku,
 241   po spusteni jednej, chod zas na zaciatok
 242  }
 243  pcur:=@ShedTop;
 244  cur:=pcur^;
 245  fpgettimeofday(@Now,nil);
 246  delta:=(Now.tv_sec-LastShed.tv_sec);
 247  if delta>6 then delta:=5000 else delta:=(delta*1000)+((Now.tv_usec-LastShed.tv_usec) div 1000);
 248  LastShed:=Now;
 249  //writeln('DeltaTime: ',delta);
 250  while assigned(cur) do begin
 251   if (cur^.left<=delta)or(cur^.left=0) then begin
 252    {unlink}
 253    pcur^:=cur^.next;
 254    {link to unused}
 255    cur^.next:=ShedUU;
 256    ShedUU:=cur;
 257    {call}
 258    cur^.cb;
 259    {go to beginning}
 260    break;
 261    pcur:=@ShedTop;
 262    cur:=pcur^;
 263   end else begin
 264    DEC(cur^.left,delta);
 265    //writeln('Left: ',cur^.left);
 266    if pollTimeOut>cur^.left then PollTimeOut:=cur^.left;
 267    pcur:=@cur^.next;
 268    cur:=cur^.next;
 269   end;
 270  end;
 271  cur:=ShedTop;
 272  tasks:=0;
 273  while assigned(cur) do begin
 274   if cur^.left<PollTimeout then PollTimeout:=cur^.left;
 275   cur:=cur^.next;
 276   inc(tasks);
 277  end;
 278  if pollTimeout=0 then pollTimeOut:=1;
 279  //if delta >4990 then writeln('ServerLoop: tasks=',tasks);
 280 end;
 281
 282 procedure Main;
 283  begin
 284  s_setupInet;
 285  while not terminated  do begin
 286   PollTimeout:=5000;
 287   ShedRun;
 288   EventsCount:=fpPoll(@PollArr[0],PollTop,PollTimeout);
 289   ShedRun;
 290   if (eventscount=-1)and terminated then break;
 291   if eventscount=-1 then break;  {fixme: print error}
 292   if eventscount=0 then continue else begin
 293    {INET socket}
 294    if DoSock(PollArr[0]) then
 295    if assigned(curhndo) then curhndo(msg)
 296    else if assigned(curhnd) then curhnd(msg)
 297    else raise eXception.Create('No handler for opcode '+IntToStr(Buffer[1]));
 298    {INET6...}
 299    {Generic}
 300    for tp:=1 to pollTop do if PollArr[tp].revents>0 then begin
 301     PollHnd[tp].CB(PollArr[tp].rEvents);
 302     PollArr[tp].revents:=0;
 303    end;
 304   end;
 305  end;
 306  write('Loop broken [');
 307  CloseSocket(s_inet);
 308  writeln(']');
 309 end;
 310
 311 procedure SetMsgHandler(OpCode:byte; handler:tMessageHandler);
 312 begin assert(hnd[OpCode]=nil); hnd[OpCode]:=handler; end;
 313 procedure SetHiMsgHandler(handler:tMessageHandler);
 314 begin Hihnd:=handler; end;
 315
 316 procedure WatchFD(fd:tHandle; h:tFDEventHandler);
 317  var opt: tPollTop;
 318 begin
 319  if assigned(h) then begin
 320   PollHnd[pollTop].CB:=h;
 321   PollArr[pollTop].fd:=fd;
 322   PollArr[pollTop].events:=POLLERR or POLLHUP or POLLIN or POLLPRI or
 323   POLLRDBAND or POLLRDNORM;
 324   PollArr[pollTop].revents:=0;
 325   //writeln('Add watch ',pollTop,' on ',fd,' to ',IntToHex(qword(@h),8));
 326   Inc(PollTop);
 327  end else for opt:=0 to high(opt) do if PollArr[opt].fd=fd then begin
 328   if (pollTop-1)>opt then begin
 329    PollArr[opt]:=PollArr[pollTop-1];
 330    PollHnd[opt]:=PollHnd[pollTop-1];
 331   end;
 332   dec(pollTop);
 333   PollArr[pollTop].fd:=-1;
 334   PollArr[pollTop].events:=0;
 335   PollArr[pollTop].revents:=0;
 336   break;
 337  end;
 338 end;
 339
 340 procedure Shedule(timeout{ms}: LongWord; h:tOnTimer);
 341  var old:^tSheduled;
 342  begin
 343  old:=ShedTop;
 344  if Assigned(ShedUU) then begin
 345   ShedTop:=ShedUU;
 346   ShedUU:=ShedUU^.next;
 347  end else New(ShedTop);
 348  ShedTop^.Left:=timeout;
 349  ShedTop^.CB:=h;
 350  ShedTop^.Next:=old;
 351 end;
 352
 353 procedure UnShedule(h:tOnTimer);
 354  var cur:^tSheduled;
 355  var pcur:^pointer;
 356  begin
 357  //if ShedTop=nil then AbstractError;
 358  pcur:=@ShedTop;
 359  cur:=pcur^;
 360  while assigned(cur) do begin
 361   if 0=CompareByte(cur^.cb,h,sizeof(h)) then begin
 362    pcur^:=cur^.next; {unlink from main list}
 363    cur^.next:=ShedUU; ShedUU:=cur; {link to unused}
 364    cur:=pcur^;
 365   end else begin
 366    pcur:=@cur^.next;
 367    cur:=pcur^;
 368   end;
 369  end;
 370 end;
 371
 372 function OptIndex(o:string):word;
 373  begin
 374  result:=paramcount;
 375  while result>0 do begin
 376   if o=system.paramstr(result) then break;
 377   dec(result);
 378  end;
 379 end;
 380
 381 function OptParamCount(o:word):word;
 382  var i:word;
 383  begin
 384  result:=0;
 385  if o>0 then for i:=o+1 to paramcount do begin
 386   if paramstr(i)[1]<>'-' then inc(result)
 387   else break;
 388  end;
 389 end;
 390
 391 var i:byte;
 392 BEGIN
 393  Randomize;
 394  fpSignal(SigInt,@SignalHandler);
 395  fpSignal(SigTerm,@SignalHandler);
 396  for i:=1 to high(hnd) do hnd[i]:=nil;
 397  for i:=1 to high(PT) do PT[i]:=nil;
 398  PT_opcodes:=[];
 399  pollTop:=1; {1 for basic listen}
 400  ShedTop:=nil;
 401  ShedUU:=nil; {todo: allocate a few to improve paging}
 402  fpgettimeofday(@LastShed,nil);
 403 END.