4 uses MemStream
,NetAddr
,UnixType
;
13 stream
: tMemoryStream
;
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
;
45 USES SysUtils
,Sockets
,BaseUnix
49 {aim for most simple implementation, since could be extended anytime}
54 var pollArr
: packed array [tPollTop
] of tPollFd
;
56 cb
: tFDEventHandler
; {proc+object}
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
69 var ShedTop
: ^tSheduled
;
70 var ShedUU
: ^tSheduled
;
71 var LastShed
: UnixType
.timeval
;
72 var PollTimeout
:LongInt;
74 procedure SC(fn
:pointer; retval
:cint
);
76 if retval
< 0 then begin
77 raise eXception
.Create(Format('Socket error %d operation %P',[SocketError
,fn
]));
81 procedure s_SetupInet
;
82 var bind_addr
:tInetSockAddr
;
86 with bind_addr
do begin
88 oi
:=OptIndex('-port');
89 if oi
=0 then sin_port
:=htons(3511)
91 assert(OptParamCount(oi
)=1);
92 sin_port
:=htons(StrToInt(paramstr(oi
+1)));
94 sin_addr
.s_addr
:=0; {any}
95 s_inet
:=fpSocket(sin_family
,SOCK_DGRAM
,IPPROTO_UDP
);
97 turnon
:=IP_PMTUDISC_DO
;
98 SC(@fpsetsockopt
,fpsetsockopt(s_inet
, IPPROTO_IP
, IP_MTU_DISCOVER
, @turnon
, sizeof(turnon
)));
100 SC(@fpBind
,fpBind(s_inet
,@bind_addr
,sizeof(bind_addr
)));
101 with PollArr
[0] do begin
108 var Terminated
:boolean=false;
110 procedure SendMessage(const data
; len
:word; const rcpt
:tSockAddrL
);
112 {SC(@fpsendto,}fpsendto(s_inet
,@data
,len
,0,@rcpt
,sizeof(sockaddr_in
)){)};
114 procedure SendMessage(const data
; len
:word; const rcpt
:tNetAddr
);
118 SendMessage(data
,len
,sa
);
120 procedure SendMessage(const data
; len
:word; const rcpt
:tNetAddr
; channel
:word );
122 SendMessage(data
,len
,rcpt
);
123 {todo: optimization??}
126 procedure SignalHandler(sig
:cint
);CDecl;
129 if terminated
then raise eControlC
.Create('CtrlC DoubleTap') ;
131 writeln('Shutdown requested');
134 {index=iphash+opcode}
135 type tPeerTableBucket
=record
138 handler
:tObjMessageHandler
;
140 var PT
:array [0..255] of ^tPeerTableBucket
;
141 var PT_opcodes
: set of 1..36;
143 function FindPT(opcode
:byte; const addr
:tNetAddr
):Word; { $FFFF=fail}
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
;
155 function IsMsgHandled(OpCode
:byte; from
:tNetAddr
):boolean;
156 begin result
:=FindPT(opcode
,from
)<>$FFFF end;
158 procedure UnSetMsgHandler(const from
:tNetAddr
; opcode
:byte);
161 h
:=FindPT(opcode
,from
);
162 if h
=$FFFF then exit
;
165 {go reverse exit on null, hash them, match: move to H and stop}
167 while (i
<>h
)and assigned(PT
[i
]) do begin
168 if (PT
[i
]^.remote
.hash
+PT
[i
]^.opcode
)=h
then begin
173 if i
=0 then i
:=high(PT
) else dec(i
);
177 procedure SetMsgHandler(OpCode
:byte; from
:tNetAddr
; handler
:tObjMessageHandler
);
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
;
188 PT
[i
]^.opcode
:=OpCode
;
190 PT
[i
]^.handler
:=handler
;
191 Include(PT_opcodes
,opcode
);
194 {do not waste stack on statics}
195 var EventsCount
:integer;
196 var Buffer
:array [1..1024] of byte;
198 var From
:tSockAddrL
; {use larger struct so everything fits}
199 var FromLen
:LongWord
;
200 var curhnd
:tMessageHandler
;
201 var curhndo
:tObjMessageHandler
;
205 procedure DoSock(var p
:tPollFD
);
211 if (p
.revents
and pollIN
)=0 then exit
;
212 FromLen
:=sizeof(From
);
213 pkLen
:=fprecvfrom(p
.FD
,@Buffer
,sizeof(Buffer
),0,@from
,@fromlen
);
214 SC(@fprecvfrom
,pkLen
);
216 FromG
.FromSocket(from
);
217 Msg
.Source
:=@FromG
; {!thread}
219 Msg
.Data
:=@Buffer
; {!thread}
220 Msg
.stream
.Init(@Buffer
,pkLen
,sizeof(Buffer
));
221 Msg
.channel
:=0; {!multisocket}
222 if Buffer
[1]>=128 then curhnd
:=HiHnd
else if Buffer
[1]<=high(hnd
) then curhnd
:=hnd
[Buffer
[1]];
223 if Buffer
[1] in PT_opcodes
then ptidx
:=FindPT(Buffer
[1],FromG
) else ptidx
:=0;
224 if ptidx
>0 then curhndo
:=PT
[ptidx
]^.handler
;
230 var now
:UnixType
.timeval
absolute iNow
;
232 var olTop
:^tSheduled
;
238 ShedTop
:=nil; {unlink the current shed list}
239 fpgettimeofday(@Now
,nil);
240 delta
:=(Now
.tv_sec
-LastShed
.tv_sec
);
241 if delta
>6 then delta
:=5000 else delta
:=(delta
*1000)+((Now
.tv_usec
-LastShed
.tv_usec
) div 1000);
243 //writeln('DeltaTime: ',delta);
244 while assigned(cur
) do begin
245 if (cur
^.left
<=delta
)or(cur
^.left
=0) then begin
252 DEC(cur
^.left
,delta
);
253 //writeln('Left: ',cur^.left);
254 if pollTimeOut
>cur
^.left
then PollTimeOut
:=cur
^.left
;
259 pcur
^:=ShedTop
; {append newly added tasks to end of untriggererd list}
260 ShedTop
:=olTop
; {link in the untriggered tasks}
262 while assigned(cur
) do begin
263 if cur
^.left
<PollTimeout
then PollTimeout
:=cur
^.left
;
266 if pollTimeout
=0 then pollTimeOut
:=1;
272 while not terminated
do begin
275 EventsCount
:=fpPoll(@PollArr
[0],PollTop
,PollTimeout
);
277 if (eventscount
=-1)and terminated
then break
;
278 if eventscount
=-1 then break
; {fixme: print error}
279 if eventscount
=0 then continue
else begin
282 if assigned(curhndo
) then curhndo(msg
)
283 else if assigned(curhnd
) then curhnd(msg
)
284 else raise eXception
.Create('No handler for opcode '+IntToStr(Buffer
[1]));
287 for tp
:=1 to pollTop
do if PollArr
[tp
].revents
>0 then begin
288 PollHnd
[tp
].CB(PollArr
[tp
].rEvents
);
289 PollArr
[tp
].revents
:=0;
293 write('Loop broken [');
298 procedure SetMsgHandler(OpCode
:byte; handler
:tMessageHandler
);
299 begin assert(hnd
[OpCode
]=nil); hnd
[OpCode
]:=handler
; end;
300 procedure SetHiMsgHandler(handler
:tMessageHandler
);
301 begin Hihnd
:=handler
; end;
303 procedure WatchFD(fd
:tHandle
; h
:tFDEventHandler
);
306 if assigned(h
) then begin
307 PollHnd
[pollTop
].CB
:=h
;
308 PollArr
[pollTop
].fd
:=fd
;
309 PollArr
[pollTop
].events
:=POLLERR
or POLLHUP
or POLLIN
or POLLPRI
or
310 POLLRDBAND
or POLLRDNORM
;
311 PollArr
[pollTop
].revents
:=0;
312 //writeln('Add watch ',pollTop,' on ',fd,' to ',IntToHex(qword(@h),8));
314 end else for opt
:=0 to high(opt
) do if PollArr
[opt
].fd
=fd
then begin
315 if (pollTop
-1)>opt
then begin
316 PollArr
[opt
]:=PollArr
[pollTop
-1];
317 PollHnd
[opt
]:=PollHnd
[pollTop
-1];
320 PollArr
[pollTop
].fd
:=-1;
321 PollArr
[pollTop
].events
:=0;
322 PollArr
[pollTop
].revents
:=0;
327 procedure Shedule(timeout
{ms}: LongWord
; h
:tOnTimer
);
331 if Assigned(ShedUU
) then begin
333 ShedUU
:=ShedUU
^.next
;
334 end else New(ShedTop
);
335 ShedTop
^.Left
:=timeout
;
340 procedure UnShedule(h
:tOnTimer
);
346 while assigned(cur
) do begin
347 if cur
^.cb
=h
then begin
348 pcur
^:=cur
^.next
; {unlink from main list}
349 cur
^.next
:=ShedUU
; ShedUU
:=cur
; {link to unused}
358 function OptIndex(o
:string):word;
361 while result
>0 do begin
362 if o
=system
.paramstr(result
) then break
;
367 function OptParamCount(o
:word):word;
371 if o
>0 then for i
:=o
+1 to paramcount
do begin
372 if paramstr(i
)[1]<>'-' then inc(result
)
380 fpSignal(SigInt
,@SignalHandler
);
381 fpSignal(SigTerm
,@SignalHandler
);
382 for i
:=1 to high(hnd
) do hnd
[i
]:=nil;
383 for i
:=1 to high(PT
) do PT
[i
]:=nil;
385 pollTop
:=1; {1 for basic listen}
387 ShedUU
:=nil; {todo: allocate a few to improve paging}
388 fpgettimeofday(@LastShed
,nil);