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