3 implementation of custom dht, based on pastry and kademlia.
4 keyspace is divided into buckets of limited capacity
5 node belongs to bucket, where at least 'depth' bits match 'prefix'
8 TODO: weight nodes by IP-Address common prefix length.
11 {used by: messages, fileshare}
15 type tPID
=Store1
.tFID
;
23 procedure NodeBootstrap(const contact
:tNetAddr
);
24 procedure GetNextNode(var ibkt
:pointer; var ix
:byte; out peer
:tPeerPub
);
25 procedure InsertNode(const peer
:tPeerPub
);
28 uses ServerLoop
,MemStream
,opcode
;
31 tPeer
=object(tPeerPub
)
41 peer
: array [1..4] of tPeer
;
46 function MatchPrefix(const tp
:tFID
):boolean;
53 function PrefixLength(const a
,b
:tFID
):byte;
59 i
:=0; while(i
<=19) do begin
60 if a
[i
]<>b
[i
] then break
;
67 if (a
[i
] and m
)<>(b
[i
] and m
) then break
;
74 function tBucket
.MatchPrefix(const tp
:tFID
):boolean;
76 result
:=(depth
=0)or(PrefixLength(prefix
,tp
)>=depth
);
79 function FindBucket(const prefix
:tFID
):tBucket_ptr
;
84 while (cur
<>nil) and (result
=nil) do begin
85 if cur
^.MatchPrefix(prefix
) {first matching is deepest}
91 operator
=(const a
,b
:tFID
):boolean;
93 result
:=CompareWord(a
,b
,10)=0;
96 procedure SplitBucket(ob
:tBucket_ptr
);
97 procedure Toggle(var prefix
:tPID
; bit
:byte);
99 prefix
[bit
div 8]:= prefix
[bit
div 8] xor ($80 shr (bit
mod 8));
104 writeln('DHT: SplitBucket ',string(ob
^.prefix
),'/',ob
^.depth
);
105 {find pref to old bucket, in order to unlink}
106 if ob
=Table
then table
:=table
^.next
else begin
108 while assigned(nb
) and (nb
^.next
<>ob
) do nb
:=nb
^.next
;
109 assert(assigned(nb
),'old bucket not in table');
111 nb
^.next
:=nb
^.next
^.next
; nb
:=nil;
113 {increase depth of this bucket}
115 ob
^.ModifyTime
:=mNow
;
116 {create new bucket with toggled bit}
119 Toggle(nb
^.Prefix
,nb
^.depth
-1);
121 {clear nodes that do not belong in bucket}
122 for i
:=1 to high(tBucket
.peer
) do begin
123 if ob
^.peer
[i
].addr
.isNil
then continue
;
124 if ob
^.MatchPrefix(ob
^.peer
[i
].id
)
125 then nb
^.peer
[i
].addr
.clear
126 else ob
^.peer
[i
].addr
.clear
;
128 writeln('-> ',string(ob
^.prefix
),'/',ob
^.depth
);
129 for i
:=1 to high(tBucket
.peer
) do if not ob
^.peer
[i
].addr
.isnil
130 then writeln('-> -> ',string(ob
^.peer
[i
].id
));
131 writeln('-> ',string(nb
^.prefix
),'/',nb
^.depth
);
132 for i
:=1 to high(tBucket
.peer
) do if not nb
^.peer
[i
].addr
.isnil
133 then writeln('-> -> ',string(nb
^.peer
[i
].id
));
134 if table
=nil then table
:=nb
else begin
136 while assigned(ob
^.next
)and (ob
^.next
^.depth
>nb
^.depth
) do ob
:=ob
^.next
;
138 writeln('-> after /',ob
^.depth
);
140 Shedule(2000,@nb
^.Refresh
);
143 procedure UpdateNode(const id
:tFID
; const addr
:tNetAddr
);
148 if id
=MyID
then exit
;
151 if not assigned(bkt
) then begin
156 bkt
^.ModifyTime
:=mNow
;
159 for i
:=1 to high(bkt
^.peer
) do bkt
^.peer
[i
].addr
.Clear
;
160 Shedule(2000,@bkt
^.Refresh
);
163 for i
:=1 to high(bkt
^.peer
)
164 do if (fr
=0)and bkt
^.peer
[i
].addr
.isNil
then fr
:=i
165 //else if bkt^.peer[i].addr=addr then fr:=i
166 else if bkt
^.peer
[i
].id
=id
then begin
167 if bkt
^.peer
[i
].addr
<>addr
then continue
;
168 {found node in the bucket}
169 //writeln('DHT: UpdateNode ',string(id));
170 // ?? bkt^.ModifyTime:=mNow;
171 bkt
^.peer
[i
].LastMsgFrom
:=mNow
;
172 bkt
^.peer
[i
].ReqDelta
:=0;
173 exit
end else if (fr
=0) and (bkt
^.peer
[i
].ReqDelta
>=2)
174 then fr
:=i
{use non-responding as free};
176 if bkt
^.MatchPrefix(MyID
)
180 end; {the bucket is full!}
181 {drop new node and hope nodes in the bucket are good}
183 writeln('DHT: AddNode ',string(id
),' to ',string(bkt
^.prefix
),'/',bkt
^.depth
,'#',fr
);
184 bkt
^.ModifyTime
:=mNow
;
185 bkt
^.peer
[fr
].ID
:=ID
;
186 bkt
^.peer
[fr
].Addr
:=Addr
;
187 bkt
^.peer
[fr
].LastMsgFrom
:=mNow
;
188 bkt
^.peer
[fr
].LastResFrom
:=0;
189 bkt
^.peer
[fr
].ReqDelta
:=0;
193 procedure InsertNode(const peer
:tPeerPub
);
195 UpdateNode(peer
.id
,peer
.addr
);
198 procedure GetNextNode(var ibkt
:tBucket_ptr
; var ix
:byte; const id
:tPID
; maxrd
:word);
201 if not assigned(ibkt
) then exit
;
205 if ix
>high(tBucket
.peer
) then begin
208 if not assigned(bkt
) then break
;
210 until (not bkt
^.peer
[ix
].Addr
.isNil
)and(bkt
^.peer
[ix
].ReqDelta
<maxrd
);
214 procedure GetNextNode(var ibkt
:pointer; var ix
:byte; out peer
:tPeerPub
);
216 if ibkt
=nil then ibkt
:=Table
;
217 GetNextNode(ibkt
,ix
,MyID
,3);
219 then peer
:=tBucket(ibkt
^).peer
[ix
]
220 else peer
.addr
.clear
;
223 procedure RecvRequest(msg
:tSMsg
);
224 var s
:tMemoryStream
absolute msg
.stream
;
236 writeln('DHT: ',string(msg
.source
^),' Request for ',string(rID
^));
237 UpdateNode(hID
^,msg
.source
^);
238 {Select peers only from The bucket,
239 if it is broken, send none, but still Ack}
240 bkt
:=FindBucket(rID
^);
242 if assigned(bkt
) then begin
243 r
.WriteByte(opcode
.dhtSelect
);
245 r
.Write(msg
.Source
^,sizeof(tNetAddr
));
248 if (s
.RdBufLen
>0)and(s
.RdBufLen
<=8) then r
.Write(s
.RdBuf
^,s
.RdBufLen
);
249 for i
:=1 to high(tBucket
.peer
) do begin
250 if bkt
^.peer
[i
].addr
.isNil
then continue
;
251 if bkt
^.peer
[i
].addr
=msg
.source
^ then continue
;
252 if bkt
^.peer
[i
].ReqDelta
>1 then continue
;
253 writeln('-> Select to ',string(bkt
^.peer
[i
].addr
));
254 SendMessage(r
.base
^,r
.length
,bkt
^.peer
[i
].addr
);
259 else writeln('-> empty bucket');
260 r
.WriteByte(opcode
.dhtReqAck
);
262 writeln('-> ReqAck to ',string(msg
.Source
^));
263 SendMessage(r
.base
^,r
.length
,msg
.source
^);
264 FreeMem(r
.base
,r
.size
);
267 procedure SendRequest(const contact
:tNetAddr
; const forid
: tPID
; caps
:byte);
271 r
.WriteByte(opcode
.dhtRequest
);
272 r
.Write(MyID
,sizeof(tFID
));
273 r
.Write(ForID
,sizeof(tFID
));
275 SendMessage(r
.base
^,r
.length
,contact
);
276 FreeMem(r
.base
,r
.size
);
279 procedure RecvReqAck(msg
:tSMsg
);
280 var s
:tMemoryStream
absolute msg
.stream
;
285 writeln('DHT: ',string(msg
.source
^),' is ',string(hID
^),' (ReqAck)');
286 UpdateNode(hID
^,msg
.source
^);
289 procedure RecvWazzup(msg
:tSMsg
);
290 var s
:tMemoryStream
absolute msg
.stream
;
295 writeln('DHT: ',string(msg
.source
^),' is ',string(hID
^),' (Wazzup)');
296 UpdateNode(hID
^,msg
.source
^);
297 //UpdateSearch(hID^,msg.source^);
300 procedure NodeBootstrap(const contact
:tNetAddr
);
302 SendRequest(contact
,MyID
,0);
305 procedure RecvSelect(msg
:tSMsg
);
306 var s
:tMemoryStream
absolute msg
.stream
;
314 addr
:=s
.ReadPtr(sizeof(tNetAddr
));
316 //writeln('DHT: ',string(msg.source^),' Select for ',string(addr^));
317 if rID
^=MyID
then begin
318 //writeln('-> self');
321 r
.WriteByte(opcode
.dhtWazzup
);
323 //writeln('-> Wazzup to ',string(addr^));
324 SendMessage(r
.base
^,r
.length
,addr
^);
325 FreeMem(r
.base
,r
.size
);
328 procedure tBucket
.Refresh
;
333 procedure lSend(var peer
:tPeer
; const trg
:tPID
);
335 SendRequest(peer
.Addr
,trg
,0);
339 my
:=MatchPrefix(MyID
);
342 for i
:=1 to high(tBucket
.peer
) do
343 if (not peer
[i
].Addr
.isNil
) and (peer
[i
].ReqDelta
<4) then begin
344 if peer
[i
].ReqDelta
>0 then begin
345 {peer is not responding, but try once more}
346 writeln('DHT: Refresh (R',peer
[i
].ReqDelta
,') #',i
,' ',string(peer
[i
].addr
));
347 lSend(peer
[i
],prefix
);
350 else if (ol
=0) or (peer
[i
].LastMsgFrom
<peer
[ol
].LastMsgFrom
)
353 {now nudge the most quiet peer}
354 if (ol
>0) and (not rtr
) then begin
355 if not rtr
then writeln('DHT: Refresh (T',mNow
-peer
[ol
].LastMsgFrom
,') #',ol
,' ',string(peer
[ol
].addr
));
356 lSend(peer
[ol
],MyID
);
358 if (not rtr
)and(ol
=0) then begin
359 {no usable nodes in this bucket, try to recover from other buckets}
361 GetNextNode(rvb
,rv
,prefix
,desperate
);
362 if not assigned(rvb
) then begin
363 rv
:=0; rvb
:=Table
; {in extreme cases, try the whole table}
364 GetNextNode(rvb
,rv
,prefix
,desperate
);
366 if assigned(rvb
) then begin
367 writeln('DHT: Refresh (RV) #',rv
,' ',string(rvb
^.peer
[rv
].addr
));
368 lSend(rvb
^.peer
[rv
],prefix
);
369 end else inc(desperate
);
370 end else desperate
:=3;
372 then wait
:=18000+(depth
*600)
374 if rtr
then wait
:=wait
div 3;
375 Shedule(wait
,@Refresh
);
379 {to bootstrap: ping address to get ID and insert to bucket/il
380 ping may get lost: separate bootstrap unit :)
381 now jut Ass-U-Me wont get lost}
384 SetMsgHandler(opcode
.dhtRequest
,@recvRequest
);
385 SetMsgHandler(opcode
.dhtSelect
,@recvSelect
);
386 SetMsgHandler(opcode
.dhtReqAck
,@recvReqAck
);
387 SetMsgHandler(opcode
.dhtWazzup
,@recvWazzup
);