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 tTimeVal=UnixType.timeval;
  33 var iNow:tTimeVal;
  34
  35 IMPLEMENTATION
  36
  37 USES SysUtils,Sockets,BaseUnix
  38      ,Unix
  39      ;
  40
  41 {aim for most simple implementation, since could be extended anytime}
  42
  43 var s_inet:tSocket;
  44
  45 type tPollTop=0..7;
  46 var pollArr: packed array [tPollTop] of tPollFd;
  47 type tFdHndDsc=record
  48  cb: tFDEventHandler; {proc+object}
  49  end;
  50 var pollHnd: array [tPollTop] of tFdHndDsc;
  51 var pollTop: tPollTop;
  52
  53 var hnd: array [1..36] of tMessageHandler;
  54 var HiHnd: tMessageHandler;
  55
  56 type tSheduled_ptr=^tSheduled; tSheduled=record
  57  left:LongWord;
  58  cb:tOnTimer;
  59  next:tSheduled_ptr;
  60  end;
  61 var ShedTop: ^tSheduled;
  62 var ShedUU: ^tSheduled;
  63 var LastShed: UnixType.timeval;
  64 var PollTimeout:LongInt;
  65
  66 procedure SC(fn:pointer; retval:cint);
  67  begin
  68   if retval < 0 then begin
  69    raise eXception.Create(Format('Socket error %d operation %P',[SocketError,fn]));
  70   end;
  71  end;
  72
  73 procedure s_SetupInet;
  74  var bind_addr:tInetSockAddr;
  75  var turnon:cint;
  76  begin
  77   with bind_addr do begin
  78    family:=AF_INET;
  79    port:=htons(3511);
  80    addr:=0; {any}
  81    s_inet:=fpSocket(family,SOCK_DGRAM,IPPROTO_UDP);
  82    SC(@fpSocket,s_inet);
  83    turnon:=IP_PMTUDISC_DO;
  84    SC(@fpsetsockopt,fpsetsockopt(s_inet, IPPROTO_IP, IP_MTU_DISCOVER, @turnon, sizeof(turnon)));
  85   end;
  86   SC(@fpBind,fpBind(s_inet,@bind_addr,sizeof(bind_addr)));
  87   with PollArr[0] do begin
  88    fd:=s_inet;
  89    events:=pollIN;
  90    revents:=0;
  91   end;
  92  end;
  93
  94 var Terminated:boolean=false;
  95
  96 procedure SendMessage(const data; len:word; const rcpt:tSockAddrL );
  97  var rc:Integer;
  98  begin
  99  {SC(@fpsendto,}fpsendto(s_inet,@data,len,0,@rcpt,sizeof(sockaddr_in)){)};
 100 end;
 101 procedure SendMessage(const data; len:word; const rcpt:tNetAddr );
 102  var sa:tSockAddrL;
 103  begin
 104  rcpt.ToSocket(sa);
 105  SendMessage(data,len,sa);
 106 end;
 107 procedure SendMessage(const data; len:word; const rcpt:tNetAddr; channel:word );
 108  begin
 109  SendMessage(data,len,rcpt);
 110  {todo: optimization??}
 111 end;
 112
 113 procedure SignalHandler(sig:cint);CDecl;
 114  begin
 115   writeln;
 116   if terminated then raise eControlC.Create('CtrlC DoubleTap') ;
 117   Terminated:=true;
 118   writeln('Shutdown requested');
 119  end;
 120
 121 {do not waste stack on statics}
 122  var EventsCount:integer;
 123  var Buffer:array [1..1024] of byte;
 124  var pkLen:LongWord;
 125  var From:tSockAddrL; {use larger struct so everything fits}
 126  var FromLen:LongWord;
 127  var FromG:tNetAddr;
 128  var curhnd:tMessageHandler;
 129  var Msg:tSMsg;
 130  var tp:tPollTop;
 131
 132 procedure PrepareHandler;
 133  begin
 134  FromG.FromSocket(from);
 135  if Buffer[1]>128 then curhnd:=HiHnd else curhnd:=hnd[Buffer[1]];
 136  if not assigned(curhnd) then raise eXception.Create('No handler for opcode '+IntToStr(Buffer[1]));
 137  Msg.Source:=@FromG; {!thread}
 138  Msg.Length:=pkLen;
 139  Msg.Data:=@Buffer; {!thread}
 140  Msg.stream.Init(@Buffer,pkLen,sizeof(Buffer));
 141  Msg.channel:=0; {!multisocket}
 142 end;
 143
 144 procedure ShedRun;
 145  var cur:^tSheduled;
 146  var pcur:^pointer;
 147  var now:UnixType.timeval absolute iNow;
 148  var delta:LongWord;
 149  var olTop:^tSheduled;
 150  begin
 151  {Sheduling}
 152  olTop:=ShedTop;
 153  pcur:=@olTop;
 154  cur:=pcur^;
 155  ShedTop:=nil; {unlink the current shed list}
 156  fpgettimeofday(@Now,nil);
 157  delta:=(Now.tv_sec-LastShed.tv_sec);
 158  if delta>6 then delta:=5000 else delta:=(delta*1000)+((Now.tv_usec-LastShed.tv_usec) div 1000);
 159  LastShed:=Now;
 160  //writeln('DeltaTime: ',delta);
 161  while assigned(cur) do begin
 162   if (cur^.left<=delta)or(cur^.left=0) then begin
 163    cur^.cb;
 164    pcur^:=cur^.next;
 165    cur^.next:=ShedUU;
 166    ShedUU:=cur;
 167    cur:=pcur^;
 168   end else begin
 169    DEC(cur^.left,delta);
 170    //writeln('Left: ',cur^.left);
 171    if pollTimeOut>cur^.left then PollTimeOut:=cur^.left;
 172    pcur:=@cur^.next;
 173    cur:=cur^.next;
 174   end;
 175  end;
 176  pcur^:=ShedTop; {append newly added tasks to end of untriggererd list}
 177  ShedTop:=olTop; {link in the untriggered tasks}
 178  cur:=olTop;
 179  while assigned(cur) do begin
 180   if cur^.left<PollTimeout then PollTimeout:=cur^.left;
 181   cur:=cur^.next;
 182  end;
 183  if pollTimeout=0 then pollTimeOut:=1;
 184 end;
 185
 186 procedure Main;
 187  begin
 188  s_setupInet;
 189  while not terminated  do begin
 190   PollTimeout:=5000;
 191   ShedRun;
 192   EventsCount:=fpPoll(@PollArr[0],PollTop,PollTimeout);
 193   if (eventscount=-1)and terminated then break;
 194   if eventscount=-1 then break;  {fixme: print error}
 195   if eventscount=0 then continue else begin
 196    {INET socket}
 197    with PollArr[0] do begin
 198     if (revents and pollIN)>0 then begin
 199      FromLen:=sizeof(From);
 200      pkLen:=fprecvfrom(s_inet,@Buffer,sizeof(Buffer),0,@from,@fromlen);
 201      SC(@fprecvfrom,pkLen);
 202      PrepareHandler;
 203      curhnd(Msg);
 204      revents:=0;
 205     end;
 206    end;
 207    {INET6...}
 208    {Generic}
 209    for tp:=1 to pollTop do if PollArr[tp].revents>0 then begin
 210     PollHnd[tp].CB(PollArr[tp].rEvents);
 211     PollArr[tp].revents:=0;
 212    end;
 213   end;
 214  end;
 215  write('Loop broken [');
 216  CloseSocket(s_inet);
 217  writeln(']');
 218 end;
 219
 220 procedure SetMsgHandler(OpCode:byte; handler:tMessageHandler);
 221 begin assert(hnd[OpCode]=nil); hnd[OpCode]:=handler; end;
 222 procedure SetHiMsgHandler(handler:tMessageHandler);
 223 begin Hihnd:=handler; end;
 224
 225 procedure WatchFD(fd:tHandle; h:tFDEventHandler);
 226  var opt: tPollTop;
 227 begin
 228  if assigned(h) then begin
 229   PollHnd[pollTop].CB:=h;
 230   PollArr[pollTop].fd:=fd;
 231   PollArr[pollTop].events:=POLLERR or POLLHUP or POLLIN or POLLPRI or
 232   POLLRDBAND or POLLRDNORM;
 233   PollArr[pollTop].revents:=0;
 234   //writeln('Add watch ',pollTop,' on ',fd,' to ',IntToHex(qword(@h),8));
 235   Inc(PollTop);
 236  end else for opt:=0 to high(opt) do if PollArr[opt].fd=fd then begin
 237   if (pollTop-1)>opt then begin
 238    PollArr[opt]:=PollArr[pollTop-1];
 239    PollHnd[opt]:=PollHnd[pollTop-1];
 240   end;
 241   dec(pollTop);
 242   PollArr[pollTop].fd:=-1;
 243   PollArr[pollTop].events:=0;
 244   PollArr[pollTop].revents:=0;
 245   break;
 246  end;
 247 end;
 248
 249 procedure Shedule(timeout{ms}: LongWord; h:tOnTimer);
 250  var old:^tSheduled;
 251  begin
 252  old:=ShedTop;
 253  if Assigned(ShedUU) then begin
 254   ShedTop:=ShedUU;
 255   ShedUU:=ShedUU^.next;
 256  end else New(ShedTop);
 257  ShedTop^.Left:=timeout;
 258  ShedTop^.CB:=h;
 259  ShedTop^.Next:=old;
 260 end;
 261
 262 procedure UnShedule(h:tOnTimer);
 263  var cur:^tSheduled;
 264  var pcur:^pointer;
 265  begin
 266  pcur:=@ShedTop;
 267  cur:=pcur^;
 268  while assigned(cur) do begin
 269   if cur^.cb=h then begin
 270    pcur^:=cur^.next; {unlink from main list}
 271    cur^.next:=ShedUU; ShedUU:=cur; {link to unused}
 272    break;
 273   end else begin
 274    pcur:=@cur^.next;
 275    cur:=pcur^;
 276   end;
 277  end;
 278 end;
 279
 280 var i:byte;
 281 BEGIN
 282  Randomize;
 283  fpSignal(SigInt,@SignalHandler);
 284  fpSignal(SigTerm,@SignalHandler);
 285  for i:=1 to high(hnd) do hnd[i]:=nil;
 286  pollTop:=1; {1 for basic listen}
 287  ShedTop:=nil;
 288  ShedUU:=nil; {todo: allocate a few to improve paging}
 289  fpgettimeofday(@LastShed,nil);
 290 END.