package dht import ( "context" "encoding/json" "sync" "github.com/google/uuid" kbucket "github.com/libp2p/go-libp2p-kbucket" "github.com/libp2p/go-libp2p/core/peer" ) // KeyKadID contains the Kademlia key in string and binary form. type KeyKadID struct { Key string Kad kbucket.ID } // NewKeyKadID creates a KeyKadID from a string Kademlia ID. func NewKeyKadID(k string) *KeyKadID { return &KeyKadID{ Key: k, Kad: kbucket.ConvertKey(k), } } // PeerKadID contains a libp2p Peer ID and a binary Kademlia ID. type PeerKadID struct { Peer peer.ID Kad kbucket.ID } // NewPeerKadID creates a PeerKadID from a libp2p Peer ID. func NewPeerKadID(p peer.ID) *PeerKadID { return &PeerKadID{ Peer: p, Kad: kbucket.ConvertPeerID(p), } } // NewPeerKadIDSlice creates a slice of PeerKadID from the passed slice of libp2p Peer IDs. func NewPeerKadIDSlice(p []peer.ID) []*PeerKadID { r := make([]*PeerKadID, len(p)) for i := range p { r[i] = NewPeerKadID(p[i]) } return r } // OptPeerKadID returns a pointer to a PeerKadID or nil if the passed Peer ID is it's default value. func OptPeerKadID(p peer.ID) *PeerKadID { if p == "" { return nil } return NewPeerKadID(p) } // NewLookupEvent creates a LookupEvent automatically converting the node // libp2p Peer ID to a PeerKadID and the string Kademlia key to a KeyKadID. func NewLookupEvent( node peer.ID, id uuid.UUID, key string, request *LookupUpdateEvent, response *LookupUpdateEvent, terminate *LookupTerminateEvent, ) *LookupEvent { return &LookupEvent{ Node: NewPeerKadID(node), ID: id, Key: NewKeyKadID(key), Request: request, Response: response, Terminate: terminate, } } // LookupEvent is emitted for every notable event that happens during a DHT lookup. // LookupEvent supports JSON marshalling because all of its fields do, recursively. type LookupEvent struct { // Node is the ID of the node performing the lookup. Node *PeerKadID // ID is a unique identifier for the lookup instance. ID uuid.UUID // Key is the Kademlia key used as a lookup target. Key *KeyKadID // Request, if not nil, describes a state update event, associated with an outgoing query request. Request *LookupUpdateEvent // Response, if not nil, describes a state update event, associated with an outgoing query response. Response *LookupUpdateEvent // Terminate, if not nil, describe a termination event. Terminate *LookupTerminateEvent } // NewLookupUpdateEvent creates a new lookup update event, automatically converting the passed peer IDs to peer Kad IDs. func NewLookupUpdateEvent( cause peer.ID, source peer.ID, heard []peer.ID, waiting []peer.ID, queried []peer.ID, unreachable []peer.ID, ) *LookupUpdateEvent { return &LookupUpdateEvent{ Cause: OptPeerKadID(cause), Source: OptPeerKadID(source), Heard: NewPeerKadIDSlice(heard), Waiting: NewPeerKadIDSlice(waiting), Queried: NewPeerKadIDSlice(queried), Unreachable: NewPeerKadIDSlice(unreachable), } } // LookupUpdateEvent describes a lookup state update event. type LookupUpdateEvent struct { // Cause is the peer whose response (or lack of response) caused the update event. // If Cause is nil, this is the first update event in the lookup, caused by the seeding. Cause *PeerKadID // Source is the peer who informed us about the peer IDs in this update (below). Source *PeerKadID // Heard is a set of peers whose state in the lookup's peerset is being set to "heard". Heard []*PeerKadID // Waiting is a set of peers whose state in the lookup's peerset is being set to "waiting". Waiting []*PeerKadID // Queried is a set of peers whose state in the lookup's peerset is being set to "queried". Queried []*PeerKadID // Unreachable is a set of peers whose state in the lookup's peerset is being set to "unreachable". Unreachable []*PeerKadID } // LookupTerminateEvent describes a lookup termination event. type LookupTerminateEvent struct { // Reason is the reason for lookup termination. Reason LookupTerminationReason } // NewLookupTerminateEvent creates a new lookup termination event with a given reason. func NewLookupTerminateEvent(reason LookupTerminationReason) *LookupTerminateEvent { return &LookupTerminateEvent{Reason: reason} } // LookupTerminationReason captures reasons for terminating a lookup. type LookupTerminationReason int // MarshalJSON returns the JSON encoding of the passed lookup termination reason. func (r LookupTerminationReason) MarshalJSON() ([]byte, error) { return json.Marshal(r.String()) } func (r LookupTerminationReason) String() string { switch r { case LookupStopped: return "stopped" case LookupCancelled: return "cancelled" case LookupStarvation: return "starvation" case LookupCompleted: return "completed" } panic("unreachable") } const ( // LookupStopped indicates that the lookup was aborted by the user's stopFn. LookupStopped LookupTerminationReason = iota // LookupCancelled indicates that the lookup was aborted by the context. LookupCancelled // LookupStarvation indicates that the lookup terminated due to lack of unqueried peers. LookupStarvation // LookupCompleted indicates that the lookup terminated successfully, reaching the Kademlia end condition. LookupCompleted ) type routingLookupKey struct{} // TODO: lookupEventChannel copies the implementation of eventChanel. // The two should be refactored to use a common event channel implementation. // A common implementation needs to rethink the signature of RegisterForEvents, // because returning a typed channel cannot be made polymorphic without creating // additional "adapter" channels. This will be easier to handle when Go // introduces generics. type lookupEventChannel struct { mu sync.Mutex ctx context.Context ch chan<- *LookupEvent } // waitThenClose is spawned in a goroutine when the channel is registered. This // safely cleans up the channel when the context has been canceled. func (e *lookupEventChannel) waitThenClose() { <-e.ctx.Done() e.mu.Lock() close(e.ch) // 1. Signals that we're done. // 2. Frees memory (in case we end up hanging on to this for a while). e.ch = nil e.mu.Unlock() } // send sends an event on the event channel, aborting if either the passed or // the internal context expire. func (e *lookupEventChannel) send(ctx context.Context, ev *LookupEvent) { e.mu.Lock() // Closed. if e.ch == nil { e.mu.Unlock() return } // in case the passed context is unrelated, wait on both. select { case e.ch <- ev: case <-e.ctx.Done(): case <-ctx.Done(): } e.mu.Unlock() } // RegisterForLookupEvents registers a lookup event channel with the given context. // The returned context can be passed to DHT queries to receive lookup events on // the returned channels. // // The passed context MUST be canceled when the caller is no longer interested // in query events. func RegisterForLookupEvents(ctx context.Context) (context.Context, <-chan *LookupEvent) { ch := make(chan *LookupEvent, LookupEventBufferSize) ech := &lookupEventChannel{ch: ch, ctx: ctx} go ech.waitThenClose() return context.WithValue(ctx, routingLookupKey{}, ech), ch } // LookupEventBufferSize is the number of events to buffer. var LookupEventBufferSize = 16 // PublishLookupEvent publishes a query event to the query event channel // associated with the given context, if any. func PublishLookupEvent(ctx context.Context, ev *LookupEvent) { ich := ctx.Value(routingLookupKey{}) if ich == nil { return } // We *want* to panic here. ech := ich.(*lookupEventChannel) ech.send(ctx, ev) }