Dht: module { PATH: con "/dis/lib/dht.dis"; BIT8SZ: con 1; BIT16SZ: con 2; BIT32SZ: con 4; BIT64SZ: con 8; IOHDRSZ: con 70; # room for Tmsg/Rmsg header MAXFDATA: con 10000; # `reasonable' iounit (was 8192) MAXRPC: con IOHDRSZ+MAXFDATA; # usable default for fversion and iounit # The code below comes mostly from the Kademlia specs # see http://xlattice.sourceforge.net/components/protocol/kademlia/specs.html # or http://www.freezepage.com/1350050954SJTTPDWBEX # DHT constants ALPHA: con 2; B: con 32; BB: con B/8; # B in bytes K: con 5; EXPIRE_TIME: con 300; # 86410 REFRESH_TIME: con 60; # 3600 REPLICATE_TIME: con 3600; # 3600 REPUBLISH_TIME: con 60; # 86400 RANDOMNESS: con 1000; # 1000 WAIT_TIME: con 2000; # 2000 TKEEP_ALIVE: con 10000; # ? MAXRETRANSMIT: con 5; # DHT messages TPing, # 100 RPing, TStore, # 102 RStore, TFindValue, # 104 RFindValue, TFindNode, # 106 RFindNode, TAskRandezvous, # 108 RAskRandezvous, TInvitation, # 110 RInvitation, TObserve, # 112 RObserve, TUser, # 114 RUser, Tmax: con 100+iota; # DHT functions error codes EBucketFull, EAlreadyPresent, EMax: con 100+iota; # FindValue result codes FVNodes, FVValue, FVMax: con iota; # Store result codes SFail, SSuccess: con iota; # AskRandezvous result codes RFail, RSuccess: con iota; # Callbacks channel actions QAddCallback, QRemoveCallback, # Store channel actions (maybe are not used) QAddItem, QRemoveItem, QUpdateItem, # Contacts channel actions QAddContact, QRemoveContact: con iota; # Query for rmsg return values QTimeOut: con (iota * -1); Node: adt { id: Key; prvaddr: string; pubaddr: string; srvaddr: string; srvid: Key; text: fn(nil: self ref Node): string; eq: fn(n1: ref Node, n2: ref Node): int; }; StoreItem: adt { data: array of byte; lastupdate: int; publishtime: int; eq: fn(a, b: ref StoreItem): int; }; Stats: adt { senttmsgs: int; # the total number of sent Tmsgs sentrmsgs: int; # the total number of sent Rmsgs recvdtmsgs: int; # the total number of received Tmsgs recvdrmsgs: int; # the total number of received Rmsgs processerrors: int; # the number of incoming message processing errors senderrors: int; # the number of errors during sending startuptime: int; # the time of dht startup sentmsgsbytype: array of int; # the number of sends of every message recvmsgsbytype: array of int; # the number of receives of every message findvaluecalled: int; # the number of times dhtfindvalue was called findnodecalled: int; # the number of times dhtfindnode was called pingcalled: int; # the number of times ping was called storecalled: int; # the number of times store was called totalrtt: int; # the sum of all rtts of all received answers answersgot: int; # the total number of received answers expiredentries: int; # the number of expired store entries unanswerednodes: int; # the number of unanswered nodes during bucket overflows bucketoverflows: int; # the count of bucket overflows logentries: int; # the number of log entries emitted }; Event: adt { pick { UserMessageReceived => msg: ref Tmsg.User; NodeAdded or NodeRemoved => node: ref Node; StoreItemAdded or StoreItemExpired or StoreItemReplicated or StoreItemRepublished => key: Key; item: ref StoreItem; DhtStarted or DhtDestroyed => # no data QueryForRmsgTimeouted => target: ref Node; msg: ref Tmsg; retransmits: int; callroutine: string; RendezvousServerChanged => oldsrv: string; newsrv: string; PublicAddrChanged => oldaddr: string; newaddr: string; } }; # DHT message handlers Tmsg: adt { uid: Key; sender: Node; targetid: Key; pick { Ping => # no additional data Store => key: Key; value: ref StoreItem; FindNode => key: Key; FindValue => key: Key; AskRandezvous => oppid: Key; addr: string; Invitation => oppprvaddr: string; opppubaddr: string; oppid: Key; Observe => # no data User => data: array of byte; } read: fn(fd: ref Sys->FD, msize: int): ref Tmsg; unpack: fn(a: array of byte): (int, ref Tmsg); pack: fn(nil: self ref Tmsg): array of byte; packedsize: fn(nil: self ref Tmsg): int; text: fn(nil: self ref Tmsg): string; mtype: fn(nil: self ref Tmsg): int; }; Rmsg: adt { uid: Key; senderid: Key; targetid: Key; pick { Ping => # no additional data Store => result: int; FindNode => nodes: array of Node; FindValue => nodes: array of Node; value: list of ref StoreItem; # TODO: remove if unused AskRandezvous => result: int; Invitation => result: int; Observe => observedaddr: string; User => data: array of byte; } read: fn(fd: ref Sys->FD, msize: int): ref Rmsg; unpack: fn(a: array of byte): (int, ref Rmsg); pack: fn(nil: self ref Rmsg): array of byte; packedsize: fn(nil: self ref Rmsg): int; text: fn(nil: self ref Rmsg): string; mtype: fn(nil: self ref Rmsg): int; }; Bucket: adt { nodes: list of ref Node; minrange: Key; maxrange: Key; lastaccess: int; isinrange: fn(nil: self ref Bucket, id: Key): int; addnode: fn(nil: self ref Bucket, n: ref Node): int; getnodes: fn(nil: self ref Bucket, size: int): array of Node; removenode: fn(nil: self ref Bucket, id: Key); findnode: fn(nil: self ref Bucket, id: Key): ref Node; text: fn(nil: self ref Bucket, tabs: int): string; }; Contacts: adt { buckets: array of ref Bucket; local: cyclic ref Local; addcontact: fn(nil: self ref Contacts, n: ref Node); removecontact: fn(nil: self ref Contacts, id: Key); getnode: fn(nil: self ref Contacts, id: Key): ref Node; findclosenodes: fn(nil: self ref Contacts, id: Key): array of Node; touch: fn(nil: self ref Contacts, idx: int); findbucket: fn(nil: self ref Contacts, id: Key): int; split: fn(nil: self ref Contacts, idx: int); randomidinbucket: fn(nil: self ref Contacts, idx: int): Key; text: fn(nil: self ref Contacts, tabs: int): string; getpublicnodes: fn(nil: self ref Contacts): array of ref Node; }; Local: adt { node: Node; localaddr: (IPaddr, int); contacts: cyclic ref Contacts; # store consists of Key, data and last access time store: ref HashTable[list of ref StoreItem]; ourstore: ref HashTable[list of ref StoreItem]; # stats object, publically usable stats: ref Stats; eventlistener: chan of ref Event; # private data and methods callbacksch: chan of (int, string, chan of ref Rmsg); callbacks: ref HashTable[chan of ref Rmsg]; storech: chan of (string, ref StoreItem, ref StoreItem, ref HashTable[list of ref StoreItem]); contactsch: chan of (int, ref Node); timerpid, processpid, callbacksprocpid, contactsprocpid, storeprocpid: int; logfd: ref Sys->FD; conn: Sys->Connection; tchan: chan of (array of byte, int, int); rchan: chan of array of byte; serverlastseenalive: int; # public API dhtfindnode: fn(nil: self ref Local, id: Key, nodes: array of ref Node): ref Node; dhtfindvalue: fn(nil: self ref Local, id: Key): list of ref StoreItem; dhtstore: fn(nil: self ref Local, key: Key, data: array of byte); # returns the rtt, or -1 if node is not reachable # raises exception if node is not found (??) dhtping: fn(nil: self ref Local, id: Key): int; # sets the file descriptor for logs, if nil logging is turned off setlogfd: fn(nil: self ref Local, fd: ref Sys->FD); # do some periodic processing process: fn(nil: self ref Local); # find k closest nodes to the given one findkclosest: fn(nil: self ref Local, id: Key): array of ref Node; # process some message processrmsg: fn(nil: self ref Local, buf: array of byte); processtmsg: fn(nil: self ref Local, buf: array of byte, raddr: string); # actually sends a message to the specified address sendmsg: fn(nil: self ref Local, addr: string, data: array of byte, retransmits: int); # send the message and setup callback with given channel sendtmsg: fn(nil: self ref Local, n: ref Node, msg: ref Tmsg, retransmits: int): chan of ref Rmsg; # same as above, but without callbacks sendrmsg: fn(nil: self ref Local, prvaddr: string, pubaddr: string, msg: ref Rmsg); # log some data logevent: fn(nil: self ref Local, source: string, msg: string); # fire the event with some interval timer: fn(nil: self ref Local); # the things that would sync everything callbacksproc: fn(nil: self ref Local); contactsproc: fn(nil: self ref Local); storeproc: fn(nil: self ref Local); # finish all internal threads and close the server destroy: fn(nil: self ref Local); # NAT traversing related methods processrandezvousquery: fn(nil: self ref Local, m: ref Tmsg.AskRandezvous, askingnode: ref Node); askrandezvous: fn(nil: self ref Local, nodeaddr, srvaddr: string, nodeid, srvid: Key): int; queryforrmsg: fn(nil: self ref Local, node: ref Node, msg: ref Tmsg, retransmits: int, callroutine: string): (int, ref Rmsg); changeserver: fn(nil: self ref Local); # events handling postevent: fn(nil: self ref Local, event: ref Event); }; init: fn(); start: fn(localaddr: string, bootstrap: array of ref Node, id: Key, logfd: ref Sys->FD): ref Local; };