TC Fix the suspend begaviour.
[brdnet.git] / ServerLoop.pas
blob0cca0f75fc0dd7b0147a40e22a652467f7a80798
1 UNIT ServerLoop;
3 INTERFACE
4 uses MemStream,NetAddr,UnixType;
6 procedure Main;
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);
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 );
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}
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);
36 type tTimeVal=UnixType.timeval;
37 var iNow:tTimeVal;
39 IMPLEMENTATION
41 USES SysUtils,Sockets,BaseUnix
42 ,Unix
45 {aim for most simple implementation, since could be extended anytime}
47 var s_inet:tSocket;
49 type tPollTop=0..7;
50 var pollArr: packed array [tPollTop] of tPollFd;
51 type tFdHndDsc=record
52 cb: tFDEventHandler; {proc+object}
53 end;
54 var pollHnd: array [tPollTop] of tFdHndDsc;
55 var pollTop: tPollTop;
57 var hnd: array [1..36] of tMessageHandler;
58 var HiHnd: tMessageHandler;
60 type tSheduled_ptr=^tSheduled; tSheduled=record
61 left:LongWord;
62 cb:tOnTimer;
63 next:tSheduled_ptr;
64 end;
65 var ShedTop: ^tSheduled;
66 var ShedUU: ^tSheduled;
67 var LastShed: UnixType.timeval;
68 var PollTimeout:LongInt;
70 procedure SC(fn:pointer; retval:cint);
71 begin
72 if retval < 0 then begin
73 raise eXception.Create(Format('Socket error %d operation %P',[SocketError,fn]));
74 end;
75 end;
77 procedure s_SetupInet;
78 var bind_addr:tInetSockAddr;
79 var turnon:cint;
80 begin
81 with bind_addr do begin
82 family:=AF_INET;
83 port:=htons(3511);
84 addr:=0; {any}
85 s_inet:=fpSocket(family,SOCK_DGRAM,IPPROTO_UDP);
86 SC(@fpSocket,s_inet);
87 turnon:=IP_PMTUDISC_DO;
88 SC(@fpsetsockopt,fpsetsockopt(s_inet, IPPROTO_IP, IP_MTU_DISCOVER, @turnon, sizeof(turnon)));
89 end;
90 SC(@fpBind,fpBind(s_inet,@bind_addr,sizeof(bind_addr)));
91 with PollArr[0] do begin
92 fd:=s_inet;
93 events:=pollIN;
94 revents:=0;
95 end;
96 end;
98 var Terminated:boolean=false;
100 procedure SendMessage(const data; len:word; const rcpt:tSockAddrL );
101 var rc:Integer;
102 begin
103 {SC(@fpsendto,}fpsendto(s_inet,@data,len,0,@rcpt,sizeof(sockaddr_in)){)};
104 end;
105 procedure SendMessage(const data; len:word; const rcpt:tNetAddr );
106 var sa:tSockAddrL;
107 begin
108 rcpt.ToSocket(sa);
109 SendMessage(data,len,sa);
110 end;
111 procedure SendMessage(const data; len:word; const rcpt:tNetAddr; channel:word );
112 begin
113 SendMessage(data,len,rcpt);
114 {todo: optimization??}
115 end;
117 procedure SignalHandler(sig:cint);CDecl;
118 begin
119 writeln;
120 if terminated then raise eControlC.Create('CtrlC DoubleTap') ;
121 Terminated:=true;
122 writeln('Shutdown requested');
123 end;
125 {index=iphash+opcode}
126 type tPeerTableBucket=record
127 opcode:byte;
128 remote:tNetAddr;
129 handler:tObjMessageHandler;
130 end;
131 var PT:array [0..255] of ^tPeerTableBucket;
132 var PT_opcodes: set of 1..36;
134 function FindPT(opcode:byte; const addr:tNetAddr):Word; { $FFFF=fail}
135 var i,o:word;
136 begin
137 i:=(addr.hash+opcode) mod high(PT); {0..63}
138 for o:=0 to high(PT) do begin
139 result:=(i+o) mod high(PT);
140 if not assigned(PT[result]) then break;
141 if (PT[result]^.opcode=opcode) and (PT[result]^.remote=addr) then exit;
142 end;
143 result:=$FFFF;
144 end;
146 procedure UnSetMsgHandler(const from:tNetAddr; opcode:byte);
147 var i,h:word;
148 begin
149 h:=FindPT(opcode,from);
150 if h=$FFFF then exit;
151 Dispose(PT[h]);
152 PT[h]:=nil;
153 {go reverse exit on null, hash them, match: move to H and stop}
154 i:=h-1;
155 while (i<>h)and assigned(PT[i]) do begin
156 if (PT[i]^.remote.hash+PT[i]^.opcode)=h then begin
157 PT[h]:=PT[i];
158 PT[i]:=nil;
159 break;
160 end;
161 if i=0 then i:=high(PT) else dec(i);
162 end;
163 end;
165 procedure SetMsgHandler(OpCode:byte; from:tNetAddr; handler:tObjMessageHandler);
166 var h,o,i:word;
167 begin
168 UnSetMsgHandler(from,opcode);
169 if handler=nil then exit;
170 h:=(from.hash+opcode) mod high(PT);
171 for o:=0 to high(PT) do begin
172 i:=(h+o) mod high(PT);
173 if not assigned(PT[i]) then break;
174 end;
175 New(PT[i]);
176 PT[i]^.opcode:=OpCode;
177 PT[i]^.remote:=from;
178 PT[i]^.handler:=handler;
179 Include(PT_opcodes,opcode);
180 end;
182 {do not waste stack on statics}
183 var EventsCount:integer;
184 var Buffer:array [1..1024] of byte;
185 var pkLen:LongWord;
186 var From:tSockAddrL; {use larger struct so everything fits}
187 var FromLen:LongWord;
188 var curhnd:tMessageHandler;
189 var curhndo:tObjMessageHandler;
190 var Msg:tSMsg;
191 var tp:tPollTop;
193 procedure DoSock(var p:tPollFD);
194 var FromG:tNetAddr;
195 var ptidx:byte;
196 begin
197 curhnd:=nil;
198 curhndo:=nil;
199 if (p.revents and pollIN)=0 then exit;
200 FromLen:=sizeof(From);
201 pkLen:=fprecvfrom(p.FD,@Buffer,sizeof(Buffer),0,@from,@fromlen);
202 SC(@fprecvfrom,pkLen);
203 p.revents:=0;
204 FromG.FromSocket(from);
205 Msg.Source:=@FromG; {!thread}
206 Msg.Length:=pkLen;
207 Msg.Data:=@Buffer; {!thread}
208 Msg.stream.Init(@Buffer,pkLen,sizeof(Buffer));
209 Msg.channel:=0; {!multisocket}
210 if Buffer[1]>128 then curhnd:=HiHnd else if Buffer[1]<=high(hnd) then curhnd:=hnd[Buffer[1]];
211 if Buffer[1] in PT_opcodes then ptidx:=FindPT(Buffer[1],FromG) else ptidx:=0;
212 if ptidx>0 then curhndo:=PT[ptidx]^.handler;
213 end;
215 procedure ShedRun;
216 var cur:^tSheduled;
217 var pcur:^pointer;
218 var now:UnixType.timeval absolute iNow;
219 var delta:LongWord;
220 var olTop:^tSheduled;
221 begin
222 {Sheduling}
223 olTop:=ShedTop;
224 pcur:=@olTop;
225 cur:=pcur^;
226 ShedTop:=nil; {unlink the current shed list}
227 fpgettimeofday(@Now,nil);
228 delta:=(Now.tv_sec-LastShed.tv_sec);
229 if delta>6 then delta:=5000 else delta:=(delta*1000)+((Now.tv_usec-LastShed.tv_usec) div 1000);
230 LastShed:=Now;
231 //writeln('DeltaTime: ',delta);
232 while assigned(cur) do begin
233 if (cur^.left<=delta)or(cur^.left=0) then begin
234 cur^.cb;
235 pcur^:=cur^.next;
236 cur^.next:=ShedUU;
237 ShedUU:=cur;
238 cur:=pcur^;
239 end else begin
240 DEC(cur^.left,delta);
241 //writeln('Left: ',cur^.left);
242 if pollTimeOut>cur^.left then PollTimeOut:=cur^.left;
243 pcur:=@cur^.next;
244 cur:=cur^.next;
245 end;
246 end;
247 pcur^:=ShedTop; {append newly added tasks to end of untriggererd list}
248 ShedTop:=olTop; {link in the untriggered tasks}
249 cur:=olTop;
250 while assigned(cur) do begin
251 if cur^.left<PollTimeout then PollTimeout:=cur^.left;
252 cur:=cur^.next;
253 end;
254 if pollTimeout=0 then pollTimeOut:=1;
255 end;
257 procedure Main;
258 begin
259 s_setupInet;
260 while not terminated do begin
261 PollTimeout:=5000;
262 ShedRun;
263 EventsCount:=fpPoll(@PollArr[0],PollTop,PollTimeout);
264 if (eventscount=-1)and terminated then break;
265 if eventscount=-1 then break; {fixme: print error}
266 if eventscount=0 then continue else begin
267 {INET socket}
268 DoSock(PollArr[0]);
269 if assigned(curhndo) then curhndo(msg)
270 else if assigned(curhnd) then curhnd(msg)
271 else raise eXception.Create('No handler for opcode '+IntToStr(Buffer[1]));
272 {INET6...}
273 {Generic}
274 for tp:=1 to pollTop do if PollArr[tp].revents>0 then begin
275 PollHnd[tp].CB(PollArr[tp].rEvents);
276 PollArr[tp].revents:=0;
277 end;
278 end;
279 end;
280 write('Loop broken [');
281 CloseSocket(s_inet);
282 writeln(']');
283 end;
285 procedure SetMsgHandler(OpCode:byte; handler:tMessageHandler);
286 begin assert(hnd[OpCode]=nil); hnd[OpCode]:=handler; end;
287 procedure SetHiMsgHandler(handler:tMessageHandler);
288 begin Hihnd:=handler; end;
290 procedure WatchFD(fd:tHandle; h:tFDEventHandler);
291 var opt: tPollTop;
292 begin
293 if assigned(h) then begin
294 PollHnd[pollTop].CB:=h;
295 PollArr[pollTop].fd:=fd;
296 PollArr[pollTop].events:=POLLERR or POLLHUP or POLLIN or POLLPRI or
297 POLLRDBAND or POLLRDNORM;
298 PollArr[pollTop].revents:=0;
299 //writeln('Add watch ',pollTop,' on ',fd,' to ',IntToHex(qword(@h),8));
300 Inc(PollTop);
301 end else for opt:=0 to high(opt) do if PollArr[opt].fd=fd then begin
302 if (pollTop-1)>opt then begin
303 PollArr[opt]:=PollArr[pollTop-1];
304 PollHnd[opt]:=PollHnd[pollTop-1];
305 end;
306 dec(pollTop);
307 PollArr[pollTop].fd:=-1;
308 PollArr[pollTop].events:=0;
309 PollArr[pollTop].revents:=0;
310 break;
311 end;
312 end;
314 procedure Shedule(timeout{ms}: LongWord; h:tOnTimer);
315 var old:^tSheduled;
316 begin
317 old:=ShedTop;
318 if Assigned(ShedUU) then begin
319 ShedTop:=ShedUU;
320 ShedUU:=ShedUU^.next;
321 end else New(ShedTop);
322 ShedTop^.Left:=timeout;
323 ShedTop^.CB:=h;
324 ShedTop^.Next:=old;
325 end;
327 procedure UnShedule(h:tOnTimer);
328 var cur:^tSheduled;
329 var pcur:^pointer;
330 begin
331 pcur:=@ShedTop;
332 cur:=pcur^;
333 while assigned(cur) do begin
334 if cur^.cb=h then begin
335 pcur^:=cur^.next; {unlink from main list}
336 cur^.next:=ShedUU; ShedUU:=cur; {link to unused}
337 break;
338 end else begin
339 pcur:=@cur^.next;
340 cur:=pcur^;
341 end;
342 end;
343 end;
345 var i:byte;
346 BEGIN
347 Randomize;
348 fpSignal(SigInt,@SignalHandler);
349 fpSignal(SigTerm,@SignalHandler);
350 for i:=1 to high(hnd) do hnd[i]:=nil;
351 for i:=1 to high(PT) do PT[i]:=nil;
352 PT_opcodes:=[];
353 pollTop:=1; {1 for basic listen}
354 ShedTop:=nil;
355 ShedUU:=nil; {todo: allocate a few to improve paging}
356 fpgettimeofday(@LastShed,nil);
357 END.