-
Notifications
You must be signed in to change notification settings - Fork 47
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #141 from dusk-network/kadcast
Kadcast functional implementation
- Loading branch information
Showing
12 changed files
with
1,122 additions
and
0 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,44 @@ | ||
| package main | ||
|
|
||
| import ( | ||
| log "github.com/sirupsen/logrus" | ||
|
|
||
| "github.com/dusk-network/dusk-blockchain/pkg/p2p/kadcast" | ||
| "github.com/dusk-network/dusk-blockchain/pkg/util/container/ring" | ||
| ) | ||
|
|
||
| func main() { | ||
| log.Infoln("Starting Kadcast Node!") | ||
| // Our node info. | ||
| var port uint16 = 25519 | ||
| ip := [4]byte{62, 57, 180, 247} | ||
| router := kadcast.MakeRouter(ip, port) | ||
| log.Infoln("Router was created Successfully.") | ||
|
|
||
| // Create buffer. | ||
| queue := ring.NewBuffer(500) | ||
|
|
||
| // Launch PacketProcessor rutine. | ||
| go kadcast.ProcessPacket(queue, &router) | ||
|
|
||
| // Launch a listener for our node. | ||
| go kadcast.StartUDPListener("udp", queue, router.MyPeerInfo) | ||
|
|
||
| // Create BootstrapNodes Peer structs | ||
| var port1 uint16 = 25519 | ||
| ip1 := [4]byte{157, 230, 219, 77} | ||
| boot1 := kadcast.MakePeer(ip1, port1) | ||
| var bootstrapNodes []kadcast.Peer | ||
| bootstrapNodes = append(bootstrapNodes[:], boot1) | ||
|
|
||
| // Start Bootstrapping process. | ||
| err := kadcast.InitBootstrap(&router, bootstrapNodes) | ||
| if err != nil { | ||
| log.Panic("Error during the Bootstrap Process. Job terminated.") | ||
| } | ||
|
|
||
| // Once the bootstrap succeeded, start the network discovery. | ||
| kadcast.StartNetworkDiscovery(&router) | ||
|
|
||
| select {} | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,99 @@ | ||
| package kadcast | ||
|
|
||
| // MaxBucketPeers represents the maximum | ||
| //number of peers that a `bucket` can hold. | ||
| var MaxBucketPeers uint8 = 25 | ||
|
|
||
| // bucket stores peer info of the peers that are at a certain | ||
| // distance range to the peer itself. | ||
| type bucket struct { | ||
| idLength uint8 | ||
| peerCount uint8 | ||
| totalPeersPassed uint64 | ||
| // Should always be less than `MaxBucketPeers` | ||
| entries []Peer | ||
| // This map keeps the order of arrivals for LRU | ||
| lru map[Peer]uint64 | ||
| // This map allows us to quickly see if a Peer is | ||
| // included on a entries set without iterating over | ||
| // it. | ||
| lruPresent map[Peer]bool | ||
| } | ||
|
|
||
| // Allocates space for a `bucket` and returns a instance | ||
| // of it with the specified `idLength`. | ||
| func makeBucket(idlen uint8) bucket { | ||
| return bucket{ | ||
| idLength: idlen, | ||
| totalPeersPassed: 0, | ||
| peerCount: 0, | ||
| entries: make([]Peer, 0, MaxBucketPeers), | ||
| lru: make(map[Peer]uint64), | ||
| lruPresent: make(map[Peer]bool), | ||
| } | ||
| } | ||
|
|
||
| // Finds the Least Recently Used Peer on the entries set | ||
| // of the `bucket` and returns it's index on the entries | ||
| // set and the `Peer` info that is hold on it. | ||
| func (b bucket) findLRUPeerIndex() (int, uint64) { | ||
| var val = b.totalPeersPassed | ||
| i := 0 | ||
| for index, p := range b.entries { | ||
| if b.lru[p] <= val { | ||
| val = b.lru[p] | ||
| i = index | ||
| } | ||
| } | ||
| return i, val | ||
| } | ||
|
|
||
| // Remove a `Peer` from the entries set without | ||
| // caring about the order. | ||
| // It also maps the `Peer` to false on the LRU map. | ||
| // The resulting slice of entries is then returned. | ||
| func (b *bucket) removePeerAtIndex(index int) []Peer { | ||
| // Remove peer from the lruPresent map. | ||
| b.lruPresent[b.entries[index]] = false | ||
|
|
||
| b.entries[index] = b.entries[len(b.entries)-1] | ||
| // We do not need to put s[i] at the end, as it will be discarded anyway | ||
| return b.entries[:len(b.entries)-1] | ||
| } | ||
|
|
||
| // Adds a `Peer` to the `bucket` entries list. | ||
| // It also increments the peerCount all according | ||
| // the LRU policy. | ||
| func (b *bucket) addPeer(peer Peer) { | ||
| // Check if the entries set can hold more peers. | ||
| if len(b.entries) < int(MaxBucketPeers) { | ||
| // Insert it into the set if not present | ||
| // on the current entries set. | ||
| if b.lruPresent[peer] == false { | ||
| b.entries = append(b.entries, peer) | ||
| b.peerCount++ | ||
| b.lruPresent[peer] = true | ||
| } | ||
| // Store recently used peer. | ||
| b.lru[peer] = b.totalPeersPassed | ||
| b.totalPeersPassed++ | ||
| return | ||
| } | ||
| // If the entries set is full, we perform | ||
| // LRU and remove a peer to include the new one. | ||
| // | ||
| // Check if peer is not already present into the | ||
| // entries set | ||
| if b.lruPresent[peer] == false { | ||
| // Search for the least recently used peer. | ||
| var index, _ = b.findLRUPeerIndex() | ||
| // Remove it from the entries set and from | ||
| // the lruPresent map. | ||
| b.entries = b.removePeerAtIndex(index) | ||
| // Add the new peer to the entries set. | ||
| b.entries = append(b.entries, peer) | ||
| b.lruPresent[peer] = true | ||
| b.totalPeersPassed++ | ||
| } | ||
| b.lru[peer] = b.totalPeersPassed | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,65 @@ | ||
| package kadcast | ||
|
|
||
| import ( | ||
| log "github.com/sirupsen/logrus" | ||
| "net" | ||
| "time" | ||
|
|
||
| "github.com/dusk-network/dusk-blockchain/pkg/util/container/ring" | ||
| ) | ||
|
|
||
| // StartUDPListener listens infinitely for UDP packet arrivals and | ||
| // executes it's processing inside a gorutine by sending | ||
| // the packets to the circularQueue. | ||
| func StartUDPListener(netw string, queue *ring.Buffer, MyPeerInfo Peer, ) { | ||
|
|
||
| lAddr := getLocalUDPAddress() | ||
| // Set listening port. | ||
| lAddr.Port = int(MyPeerInfo.port) | ||
| PacketConnCreation: | ||
| // listen to incoming udp packets | ||
| pc, err := net.ListenUDP(netw, &lAddr) | ||
| if err != nil { | ||
| log.Panic(err) | ||
| } | ||
| // Set initial deadline. | ||
| pc.SetDeadline(time.Now().Add(time.Minute)) | ||
|
|
||
| // Instanciate the buffer | ||
| buffer := make([]byte, 1024) | ||
| for { | ||
| // Read UDP packet. | ||
| byteNum, uAddr, err := pc.ReadFromUDP(buffer) | ||
|
|
||
| if err != nil { | ||
| log.WithError(err).Warn("Error on packet read") | ||
| pc.Close() | ||
| goto PacketConnCreation | ||
| } | ||
| // Set a new deadline for the connection. | ||
| pc.SetDeadline(time.Now().Add(5 * time.Minute)) | ||
| // Serialize the packet. | ||
| encodedPack := encodeRedPacket(uint16(byteNum), *uAddr, buffer[0:byteNum]) | ||
| // Send the packet to the Consumer putting it on the queue. | ||
| queue.Put(encodedPack) | ||
| } | ||
| } | ||
|
|
||
| // Gets the local address of the sender `Peer` and the UDPAddress of the | ||
| // reciever `Peer` and sends to it a UDP Packet with the payload inside. | ||
| func sendUDPPacket(netw string, addr net.UDPAddr, payload []byte) { | ||
| localAddr := getLocalUDPAddress() | ||
| conn, err := net.DialUDP(netw, &localAddr, &addr) | ||
| if err != nil { | ||
| log.WithError(err).Warn("Could not stablish a connection with the dest Peer.") | ||
| return | ||
| } | ||
| defer conn.Close() | ||
|
|
||
| // Simple write | ||
| _, err = conn.Write(payload) | ||
| if err != nil { | ||
| log.WithError(err).Warn("Error while writting to the filedescriptor.") | ||
| return | ||
| } | ||
| } |
Oops, something went wrong.