test and finish FS
[brdnet.git] / ServerLoop.pas
blob3e4e58c9754614375ba16de98a610fc4c8d51b50
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);
35 function IsMsgHandled(OpCode:byte; from:tNetAddr):boolean;
37 function OptIndex(o:string):word;
38 function OptParamCount(o:word):word;
40 type tTimeVal=UnixType.timeval;
41 var iNow:tTimeVal;
43 IMPLEMENTATION
45 USES SysUtils,Sockets,BaseUnix
46 ,Unix
49 {aim for most simple implementation, since could be extended anytime}
51 var s_inet:tSocket;
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;
61 var hnd: array [1..36] of tMessageHandler;
62 var HiHnd: tMessageHandler;
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;
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;
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;
108 var Terminated:boolean=false;
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;
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;
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);
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;
155 function IsMsgHandled(OpCode:byte; from:tNetAddr):boolean;
156 begin result:=FindPT(opcode,from)<>$FFFF end;
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 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;
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;
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;
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;
231 procedure ShedRun;
232 var cur:^tSheduled;
233 var pcur:^pointer;
234 var now:UnixType.timeval absolute iNow;
235 var delta:LongWord;
236 var olTop:^tSheduled;
237 begin
238 {Sheduling}
239 {fixme: proste niak aby to šlo vymazať z callbacku}
240 olTop:=ShedTop;
241 pcur:=@olTop;
242 cur:=pcur^;
243 ShedTop:=nil; {unlink the current shed list}
244 fpgettimeofday(@Now,nil);
245 delta:=(Now.tv_sec-LastShed.tv_sec);
246 if delta>6 then delta:=5000 else delta:=(delta*1000)+((Now.tv_usec-LastShed.tv_usec) div 1000);
247 LastShed:=Now;
248 //writeln('DeltaTime: ',delta);
249 while assigned(cur) do begin
250 if (cur^.left<=delta)or(cur^.left=0) then begin
251 cur^.cb;
252 pcur^:=cur^.next;
253 cur^.next:=ShedUU;
254 ShedUU:=cur;
255 cur:=pcur^;
256 end else begin
257 DEC(cur^.left,delta);
258 //writeln('Left: ',cur^.left);
259 if pollTimeOut>cur^.left then PollTimeOut:=cur^.left;
260 pcur:=@cur^.next;
261 cur:=cur^.next;
262 end;
263 end;
264 pcur^:=ShedTop; {append newly added tasks to end of untriggererd list}
265 ShedTop:=olTop; {link in the untriggered tasks}
266 cur:=olTop;
267 while assigned(cur) do begin
268 if cur^.left<PollTimeout then PollTimeout:=cur^.left;
269 cur:=cur^.next;
270 end;
271 if pollTimeout=0 then pollTimeOut:=1;
272 end;
274 procedure Main;
275 begin
276 s_setupInet;
277 while not terminated do begin
278 PollTimeout:=5000;
279 ShedRun;
280 EventsCount:=fpPoll(@PollArr[0],PollTop,PollTimeout);
281 ShedRun;
282 if (eventscount=-1)and terminated then break;
283 if eventscount=-1 then break; {fixme: print error}
284 if eventscount=0 then continue else begin
285 {INET socket}
286 if DoSock(PollArr[0]) then
287 if assigned(curhndo) then curhndo(msg)
288 else if assigned(curhnd) then curhnd(msg)
289 else raise eXception.Create('No handler for opcode '+IntToStr(Buffer[1]));
290 {INET6...}
291 {Generic}
292 for tp:=1 to pollTop do if PollArr[tp].revents>0 then begin
293 PollHnd[tp].CB(PollArr[tp].rEvents);
294 PollArr[tp].revents:=0;
295 end;
296 end;
297 end;
298 write('Loop broken [');
299 CloseSocket(s_inet);
300 writeln(']');
301 end;
303 procedure SetMsgHandler(OpCode:byte; handler:tMessageHandler);
304 begin assert(hnd[OpCode]=nil); hnd[OpCode]:=handler; end;
305 procedure SetHiMsgHandler(handler:tMessageHandler);
306 begin Hihnd:=handler; end;
308 procedure WatchFD(fd:tHandle; h:tFDEventHandler);
309 var opt: tPollTop;
310 begin
311 if assigned(h) then begin
312 PollHnd[pollTop].CB:=h;
313 PollArr[pollTop].fd:=fd;
314 PollArr[pollTop].events:=POLLERR or POLLHUP or POLLIN or POLLPRI or
315 POLLRDBAND or POLLRDNORM;
316 PollArr[pollTop].revents:=0;
317 //writeln('Add watch ',pollTop,' on ',fd,' to ',IntToHex(qword(@h),8));
318 Inc(PollTop);
319 end else for opt:=0 to high(opt) do if PollArr[opt].fd=fd then begin
320 if (pollTop-1)>opt then begin
321 PollArr[opt]:=PollArr[pollTop-1];
322 PollHnd[opt]:=PollHnd[pollTop-1];
323 end;
324 dec(pollTop);
325 PollArr[pollTop].fd:=-1;
326 PollArr[pollTop].events:=0;
327 PollArr[pollTop].revents:=0;
328 break;
329 end;
330 end;
332 procedure Shedule(timeout{ms}: LongWord; h:tOnTimer);
333 var old:^tSheduled;
334 begin
335 old:=ShedTop;
336 if Assigned(ShedUU) then begin
337 ShedTop:=ShedUU;
338 ShedUU:=ShedUU^.next;
339 end else New(ShedTop);
340 ShedTop^.Left:=timeout;
341 ShedTop^.CB:=h;
342 ShedTop^.Next:=old;
343 end;
345 procedure UnShedule(h:tOnTimer);
346 var cur:^tSheduled;
347 var pcur:^pointer;
348 begin
349 pcur:=@ShedTop;
350 cur:=pcur^;
351 while assigned(cur) do begin
352 if cur^.cb=h then begin
353 pcur^:=cur^.next; {unlink from main list}
354 cur^.next:=ShedUU; ShedUU:=cur; {link to unused}
355 break;
356 end else begin
357 pcur:=@cur^.next;
358 cur:=pcur^;
359 end;
360 end;
361 end;
363 function OptIndex(o:string):word;
364 begin
365 result:=paramcount;
366 while result>0 do begin
367 if o=system.paramstr(result) then break;
368 dec(result);
369 end;
370 end;
372 function OptParamCount(o:word):word;
373 var i:word;
374 begin
375 result:=0;
376 if o>0 then for i:=o+1 to paramcount do begin
377 if paramstr(i)[1]<>'-' then inc(result)
378 else break;
379 end;
380 end;
382 var i:byte;
383 BEGIN
384 Randomize;
385 fpSignal(SigInt,@SignalHandler);
386 fpSignal(SigTerm,@SignalHandler);
387 for i:=1 to high(hnd) do hnd[i]:=nil;
388 for i:=1 to high(PT) do PT[i]:=nil;
389 PT_opcodes:=[];
390 pollTop:=1; {1 for basic listen}
391 ShedTop:=nil;
392 ShedUU:=nil; {todo: allocate a few to improve paging}
393 fpgettimeofday(@LastShed,nil);
394 END.