ServerLoop.pas

   1 UNIT ServerLoop;
   2
   3 INTERFACE
   4 uses MemStream,NetAddr,UnixType,Sockets;
   5
   6 procedure Main;
   7 procedure RequestTerminate(c:byte);
   8
   9 {#Message handling#}
  10 type tSMsg=object
  11  Source: ^tNetAddr;
  12  Length: {Long}Word;
  13  Data: pointer;
  14  stream: tMemoryStream;
  15  channel: word;
  16  end;
  17 type tMessageHandler=procedure(msg:tSMsg);
  18 procedure SetMsgHandler(OpCode:byte; handler:tMessageHandler);
  19 procedure SetHiMsgHandler(handler:tMessageHandler);
  20
  21 function GetSocket(const rcpt:tNetAddr):tSocket;
  22 procedure SendMessage(const data; len:word; const rcpt:tNetAddr );
  23 {procedure SendReply(const data; len:word; const rcpt:tSMsg );}
  24 procedure SendMessage(const data; len:word; const rcpt:tNetAddr; channel:word );
  25
  26 {#Sheduling and watching#}
  27 type tFDEventHandler=procedure(ev:Word) of object;
  28 type tOnTimer=procedure of object;
  29 procedure WatchFD(fd:tHandle; h:tFDEventHandler);
  30 procedure WatchFDRW(fd:tHandle; h:tFDEventHandler);
  31 procedure Shedule(timeout{ms}: LongWord; h:tOnTimer);
  32 procedure UnShedule(h:tOnTimer);
  33  {note unshed will fail when called from OnTimer proc}
  34
  35 type tObjMessageHandler=procedure(msg:tSMsg) of object;
  36 {deliver message from peer to the object}
  37 procedure SetMsgHandler(OpCode:byte; from:tNetAddr; handler:tObjMessageHandler); overload;
  38 function IsMsgHandled(OpCode:byte; from:tNetAddr):boolean;
  39
  40 function OptIndex(o:string):word;
  41 function OptParamCount(o:word):word;
  42
  43 var OnTerminate:procedure;
  44 var VersionString:string[63];
  45 const VersionBrand='BrodNetD';
  46
  47 type tTimeVal=UnixType.timeval;
  48 type tMTime=DWORD;
  49 var iNow:tTimeVal;
  50 var mNow:tMTime; { miliseconds since start }
  51                   {overflows in hunderd hours }
  52 function GetMTime:tMTime;
  53 procedure SetThreadName(name:pchar);
  54 procedure SC(fn:pointer; retval:cint);
  55
  56 IMPLEMENTATION
  57
  58 USES SysUtils,BaseUnix
  59      ,Unix
  60      ,Linux
  61      ;
  62
  63 {aim for most simple implementation, since could be extended anytime}
  64
  65 var s_inet:tSocket;
  66
  67 type tPollTop=0..7;
  68 var pollArr: packed array [tPollTop] of tPollFd;
  69 type tFdHndDsc=record
  70  cb: tFDEventHandler; {proc+object}
  71  end;
  72 var pollHnd: array [tPollTop] of tFdHndDsc;
  73 var pollTop: tPollTop;
  74
  75 var hnd: array [1..36] of tMessageHandler;
  76 var HiHnd: tMessageHandler;
  77
  78 type tSheduled_ptr=^tSheduled; tSheduled=record
  79  left:LongWord;
  80  cb:tOnTimer;
  81  next:tSheduled_ptr;
  82  end;
  83 var ShedTop: ^tSheduled;
  84 var ShedUU: ^tSheduled;
  85 var LastShed: tMTime;
  86 var PollTimeout:LongInt;
  87
  88 procedure SC(fn:pointer; retval:cint);
  89  begin
  90   if retval < 0 then begin
  91    raise eXception.Create(Format('Socket error %d operation %P',[SocketError,fn]));
  92   end;
  93  end;
  94
  95 procedure s_SetupInet;
  96  var bind_addr:tInetSockAddr;
  97  var turnon:cint;
  98  var oi:word;
  99  begin
 100   with bind_addr do begin
 101    sin_family:=AF_INET;
 102    oi:=OptIndex('-port');
 103    if oi=0 then sin_port:=htons(3511)
 104    else begin
 105     assert(OptParamCount(oi)=1);
 106     sin_port:=htons(StrToInt(paramstr(oi+1)));
 107    end;
 108    sin_addr.s_addr:=0; {any}
 109    s_inet:=fpSocket(sin_family,SOCK_DGRAM,IPPROTO_UDP);
 110    SC(@fpSocket,s_inet);
 111    turnon:=IP_PMTUDISC_DO;
 112    SC(@fpsetsockopt,fpsetsockopt(s_inet, IPPROTO_IP, IP_MTU_DISCOVER, @turnon, sizeof(turnon)));
 113   end;
 114   SC(@fpBind,fpBind(s_inet,@bind_addr,sizeof(bind_addr)));
 115   with PollArr[0] do begin
 116    fd:=s_inet;
 117    events:=pollIN;
 118    revents:=0;
 119   end;
 120  end;
 121
 122 var Terminated:boolean=false;
 123
 124 function GetSocket(const rcpt:tNetAddr):tSocket;
 125  begin
 126  result:=s_inet;
 127 end;
 128 procedure SendMessage(const data; len:word; const rcpt:tSockAddrL );
 129  begin
 130  {SC(@fpsendto,}fpsendto(s_inet,@data,len,0,@rcpt,sizeof(sockaddr_in)){)};
 131 end;
 132 procedure SendMessage(const data; len:word; const rcpt:tNetAddr );
 133  var sa:tSockAddrL;
 134  begin
 135  rcpt.ToSocket(sa);
 136  SendMessage(data,len,sa);
 137 end;
 138 procedure SendMessage(const data; len:word; const rcpt:tNetAddr; channel:word );
 139  begin
 140  SendMessage(data,len,rcpt);
 141  {todo: optimization??}
 142 end;
 143
 144 procedure SignalHandler(sig:cint);CDecl;
 145  begin
 146   writeln;
 147   if terminated then raise eControlC.Create('CtrlC DoubleTap') ;
 148   Terminated:=true;
 149  end;
 150 procedure FatalSignalHandler(sig:cint);CDecl;
 151   begin
 152   raise eExternal.Create('Unexpected Signal '+IntToStr(sig)) ;
 153   Terminated:=true;
 154 end;
 155
 156 {index=iphash+opcode}
 157 type tPeerTableBucket=record
 158  opcode:byte;
 159  remote:tNetAddr;
 160  handler:tObjMessageHandler;
 161 end;
 162 var PT:array [0..255] of ^tPeerTableBucket;
 163 var PT_opcodes: set of 1..high(hnd);
 164
 165 function FindPT(opcode:byte; addr:tNetAddr):Word; { $FFFF=fail}
 166  var i,o:word;
 167  begin
 168  i:=(addr.hash+opcode) mod high(PT); {0..63}
 169  for o:=0 to high(PT) do begin
 170   result:=(i+o) mod high(PT);
 171   if not assigned(PT[result]) then break;
 172   if (PT[result]^.opcode=opcode) and (PT[result]^.remote=addr) then exit;
 173  end;
 174  result:=$FFFF;
 175 end;
 176
 177 function IsMsgHandled(OpCode:byte; from:tNetAddr):boolean;
 178  begin result:=FindPT(opcode,from)<>$FFFF end;
 179
 180 procedure UnSetMsgHandler(const from:tNetAddr; opcode:byte);
 181  var i,h:word;
 182  begin
 183  h:=FindPT(opcode,from);
 184  if h=$FFFF then exit;
 185  Dispose(PT[h]);
 186  PT[h]:=nil;
 187  {go reverse exit on null, hash them, match: move to H and stop}
 188  if h=0 then i:=high(PT) else i:=h-1;
 189  while (i<>h)and assigned(PT[i]) do begin
 190   if (PT[i]^.remote.hash+PT[i]^.opcode)=h then begin
 191    PT[h]:=PT[i];
 192    PT[i]:=nil;
 193    break;
 194   end;
 195   if i=0 then i:=high(PT) else dec(i);
 196  end;
 197 end;
 198
 199 procedure SetMsgHandler(OpCode:byte; from:tNetAddr; handler:tObjMessageHandler);
 200  var h,o,i:word;
 201  begin
 202  UnSetMsgHandler(from,opcode);
 203  if handler=nil then exit;
 204  h:=(from.hash+opcode) mod high(PT);
 205  for o:=0 to high(PT) do begin
 206   i:=(h+o) mod high(PT);
 207   if not assigned(PT[i]) then break;
 208  end;
 209  New(PT[i]);
 210  PT[i]^.opcode:=OpCode;
 211  PT[i]^.remote:=from;
 212  PT[i]^.handler:=handler;
 213  if opcode<=high(hnd) then Include(PT_opcodes,opcode);
 214 end;
 215
 216 {do not waste stack on statics}
 217  var EventsCount:integer;
 218  var Buffer:array [1..4096] of byte;
 219  var pkLen:LongWord;
 220  var From:tSockAddrL; {use larger struct so everything fits}
 221  var FromLen:LongWord;
 222  var FromG:tNetAddr;
 223  var curhnd:tMessageHandler;
 224  var curhndo:tObjMessageHandler;
 225  var Msg:tSMsg;
 226  var tp:tPollTop;
 227
 228 function DoSock(var p:tPollFD):boolean;
 229  var ptidx:word;
 230  begin
 231  curhnd:=nil;
 232  curhndo:=nil;
 233  result:=false;
 234  ptidx:=$FFFF;
 235  if (p.revents and pollIN)=0 then exit else  result:=true;
 236  FromLen:=sizeof(From);
 237  pkLen:=fprecvfrom(p.FD,@Buffer,sizeof(Buffer),0,@from,@fromlen);
 238  SC(@fprecvfrom,pkLen);
 239  p.revents:=0;
 240  FromG.FromSocket(from);
 241  Msg.Source:=@FromG; {!thread}
 242  Msg.Length:=pkLen;
 243  Msg.Data:=@Buffer; {!thread}
 244  Msg.stream.Init(@Buffer,pkLen,sizeof(Buffer));
 245  Msg.channel:=0; {!multisocket}
 246  if Buffer[1]>=128 then curhnd:=HiHnd else if Buffer[1]<=high(hnd) then curhnd:=hnd[Buffer[1]];
 247  if (Buffer[1]>high(hnd))or(Buffer[1] in PT_opcodes) then begin
 248   ptidx:=FindPT(Buffer[1],FromG);
 249   if ptidx<$FFFF then curhndo:=PT[ptidx]^.handler;
 250  end;
 251 end;
 252
 253 var GetMTimeOffsetSec:DWORD=0;
 254 function GetMTime:tMTime;
 255  {$IFDEF UNIX}
 256  var time:UnixType.timespec;
 257  var trans:QWORD;
 258  begin
 259  assert(clock_gettime(CLOCK_MONOTONIC,@time)=0);
 260  trans:=((time.tv_sec-GetMTimeOffsetSec)*1000)+(time.tv_nsec div 1000000);
 261  GetMTime:=trans and $FFFFFFFF;
 262  {$ELSE}{$ERROR Not Implemented on non unix}
 263  begin GetMTime:=0;
 264 {$ENDIF}end;
 265 procedure InitMTime; {$IFDEF UNIX}
 266  var time:UnixType.timespec;
 267  begin
 268  assert(clock_gettime(CLOCK_MONOTONIC,@time)=0);
 269  GetMTimeOffsetSec:=time.tv_sec;
 270  {$ELSE}{$ERROR Not Implemented on non unix}
 271  begin
 272 {$ENDIF}end;
 273
 274 {$IFDEF Linux}
 275 function prctl( option:cint; arg2,arg3,arg4,arg5:culong):cint; cdecl; external;
 276 const PR_SET_NAME=15;
 277 {$ENDIF}
 278 procedure SetThreadName(name:pchar);
 279 {$IFDEF Linux} begin prctl(PR_SET_NAME,culong(pchar(name)),0,0,0)
 280 {$ELSE}begin{$NOTE Custom thread mames not supported}
 281 {$ENDIF} end;
 282
 283 procedure ShedRun;
 284  var cur:^tSheduled;
 285  var pcur:^pointer;
 286  var delta:LongWord;
 287  var tasks:word;
 288  begin
 289  {Sheduling}
 290  {gmagic with delta-time, increment mNow, ...}
 291  mNow:=GetMTime;
 292  delta:=mNow-LastShed;
 293  LastShed:=mNow;
 294  //writeln('DeltaTime: ',delta);
 295  {first tick all tasks}
 296  tasks:=0;
 297  cur:=ShedTop;
 298  while assigned(cur) do begin
 299   if cur^.left<=delta then cur^.left:=0 else begin
 300    dec(cur^.left,delta);
 301    {also set next wake time}
 302    if cur^.left<PollTimeout then PollTimeout:=cur^.left;
 303   end;
 304   {count tasks here}
 305   inc(tasks);
 306   cur:=cur^.next;
 307  end;
 308  {correct floating-point glitch}
 309  if pollTimeout=0 then pollTimeOut:=1;
 310  {run first runnable task}
 311  pcur:=@ShedTop;
 312  cur:=pcur^;
 313  while assigned(cur) do begin
 314   if cur^.left=0 then begin
 315    {unlink}
 316    pcur^:=cur^.next;
 317    {link to unused}
 318    cur^.next:=ShedUU;
 319    ShedUU:=cur;
 320    {call}
 321    cur^.cb;
 322    {do rest later}
 323    pollTimeout:=0;
 324    break;
 325   end;
 326   pcur:=@cur^.next;
 327   cur:=cur^.next;
 328  end;
 329 end;
 330
 331 var ReExec:boolean=false;
 332 procedure Main;
 333  begin
 334  s_setupInet;
 335  while not terminated  do begin
 336   PollTimeout:=5000;
 337   ShedRun;
 338   EventsCount:=fpPoll(@PollArr[0],PollTop,PollTimeout);
 339   ShedRun;
 340   if (eventscount=-1)and terminated then break;
 341   if eventscount=-1 then break;  {fixme: print error}
 342   if eventscount=0 then continue else begin
 343    {INET socket}
 344    if DoSock(PollArr[0]) then
 345    if assigned(curhndo) then curhndo(msg)
 346    else if assigned(curhnd) then curhnd(msg)
 347    else {raise eXception.Create('}writeln('ServerLoop: No handler for opcode '+IntToStr(Buffer[1]));
 348    {INET6...}
 349    {Generic}
 350    for tp:=1 to pollTop do if PollArr[tp].revents>0 then begin
 351     PollHnd[tp].CB(PollArr[tp].rEvents);
 352     PollArr[tp].revents:=0;
 353    end;
 354   end;
 355  end;
 356  if assigned(onTerminate) then onTerminate;
 357  CloseSocket(s_inet);
 358  if ReExec then fpExecv(paramstr(0),argv);
 359 end;
 360
 361 procedure SetMsgHandler(OpCode:byte; handler:tMessageHandler);
 362 begin assert(hnd[OpCode]=nil); hnd[OpCode]:=handler; end;
 363 procedure SetHiMsgHandler(handler:tMessageHandler);
 364 begin Hihnd:=handler; end;
 365
 366 procedure WatchFD(fd:tHandle; h:tFDEventHandler; e:LongWord);
 367  var opt: tPollTop;
 368 begin
 369  if assigned(h) then begin
 370   PollHnd[pollTop].CB:=h;
 371   PollArr[pollTop].fd:=fd;
 372   PollArr[pollTop].events:=e;
 373   PollArr[pollTop].revents:=0;
 374   //writeln('Add watch ',pollTop,' on ',fd,' to ',IntToHex(qword(@h),8));
 375   Inc(PollTop);
 376  end else for opt:=0 to high(opt) do if PollArr[opt].fd=fd then begin
 377   if (pollTop-1)>opt then begin
 378    PollArr[opt]:=PollArr[pollTop-1];
 379    PollHnd[opt]:=PollHnd[pollTop-1];
 380   end;
 381   dec(pollTop);
 382   PollArr[pollTop].fd:=-1;
 383   PollArr[pollTop].events:=0;
 384   PollArr[pollTop].revents:=0;
 385   break;
 386  end;
 387 end;
 388 procedure WatchFD(fd:tHandle; h:tFDEventHandler);
 389   begin
 390   WatchFD(fd,h,POLLERR or POLLHUP or POLLIN or POLLPRI or
 391   POLLRDBAND or POLLRDNORM);
 392 end;
 393 procedure WatchFDRW(fd:tHandle; h:tFDEventHandler);
 394   begin
 395   WatchFD(fd,h,POLLERR or POLLHUP or POLLIN or POLLPRI or
 396   POLLRDBAND or POLLRDNORM or POLLOUT);
 397 end;
 398
 399 procedure Shedule(timeout{ms}: LongWord; h:tOnTimer);
 400  var old:^tSheduled;
 401  begin
 402  old:=ShedTop;
 403  if Assigned(ShedUU) then begin
 404   ShedTop:=ShedUU;
 405   ShedUU:=ShedUU^.next;
 406  end else New(ShedTop);
 407  ShedTop^.Left:=timeout;
 408  ShedTop^.CB:=h;
 409  ShedTop^.Next:=old;
 410 end;
 411
 412 procedure UnShedule(h:tOnTimer);
 413  var cur:^tSheduled;
 414  var pcur:^pointer;
 415  begin
 416  //if ShedTop=nil then AbstractError;
 417  pcur:=@ShedTop;
 418  cur:=pcur^;
 419  while assigned(cur) do begin
 420   if 0=CompareByte(cur^.cb,h,sizeof(h)) then begin
 421    pcur^:=cur^.next; {unlink from main list}
 422    cur^.next:=ShedUU; ShedUU:=cur; {link to unused}
 423    cur:=pcur^;
 424   end else begin
 425    pcur:=@cur^.next;
 426    cur:=pcur^;
 427   end;
 428  end;
 429 end;
 430
 431 var DoShowOpts:boolean=false;
 432 function OptIndex(o:string):word;
 433  begin
 434  if DoShowOpts then writeln('Option: ',o);
 435  result:=paramcount;
 436  while result>0 do begin
 437   if o=system.paramstr(result) then break;
 438   dec(result);
 439  end;
 440 end;
 441
 442 function OptParamCount(o:word):word;
 443  var i:word;
 444  begin
 445  result:=0;
 446  if o>0 then for i:=o+1 to paramcount do begin
 447   if paramstr(i)[1]<>'-' then inc(result)
 448   else break;
 449  end;
 450 end;
 451 procedure RequestTerminate(c:byte);
 452 begin Terminated:=true;
 453   if c=9 then ReExec:=true;
 454 end;
 455
 456 var i:byte;
 457 var nb:array [0..0] of byte;
 458 {$I gitver.inc}
 459 BEGIN
 460  VersionString:=GIT_VERSION+'-'+IntToStr(BUILD_VERSION);
 461  writeln('ServerLoop: ',VersionBrand,' ',VersionString);
 462  mNow:=0;
 463  Randomize;
 464  fpSignal(SigInt,@SignalHandler);
 465  fpSignal(SigTerm,@SignalHandler);
 466  fpSignal(SigPipe,baseunix.signalhandler(SIG_IGN));
 467  for i:=1 to high(hnd) do hnd[i]:=nil;
 468  for i:=1 to high(PT) do PT[i]:=nil;
 469  PT_opcodes:=[];
 470  pollTop:=1; {1 for basic listen}
 471  ShedTop:=nil;
 472  ShedUU:=nil; {todo: allocate a few to improve paging}
 473  InitMTime;
 474  LastShed:=GetMTime;
 475  if OptIndex('-h')>0 then DoShowOpts:=true;
 476  OnTerminate:=nil;
 477  Flush(OUTPUT);
 478  SetTextBuf(OUTPUT,nb);
 479 END.