Skip to content

Optimize Symbol Fetch #208

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
j-rafique wants to merge 1 commit into from
Closed

Optimize Symbol Fetch #208

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
272 changes: 121 additions & 151 deletions p2p/kademlia/dht.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@ import (
"github.com/btcsuite/btcutil/base58"
"github.com/cenkalti/backoff/v4"
"github.com/cosmos/cosmos-sdk/crypto/keyring"
"golang.org/x/sync/semaphore"

"github.com/LumeraProtocol/lumera/x/lumeraid/securekeyx"
"github.com/LumeraProtocol/supernode/v2/pkg/errors"
Expand Down Expand Up @@ -672,11 +673,7 @@ func (s *DHT) BatchRetrieve(ctx context.Context, keys []string, required int32,
var foundLocalCount int32

hexKeys := make([]string, len(keys))
globalClosestContacts := make(map[string]*NodeList)
hashes := make([][]byte, len(keys))
knownNodes := make(map[string]*Node)
var knownMu sync.Mutex
var closestMu sync.RWMutex

defer func() {
resMap.Range(func(key, value interface{}) bool {
Expand All @@ -700,32 +697,67 @@ func (s *DHT) BatchRetrieve(ctx context.Context, keys []string, required int32,
}
}()

for _, key := range keys {
result[key] = nil
}

supernodeAddr, _ := s.getSupernodeAddress(ctx)
hostIP := parseSupernodeAddress(supernodeAddr)
self := &Node{ID: s.ht.self.ID, IP: hostIP, Port: s.ht.self.Port}
self.SetHashedID()

for i, key := range keys {
decoded := base58.Decode(key)
if len(decoded) != B/8 {
return nil, fmt.Errorf("invalid key: %v", key)
}
hashes[i] = decoded
hexKeys[i] = hex.EncodeToString(decoded)
result[key] = nil
}

foundLocalCount, err = s.fetchAndAddLocalKeys(ctx, hexKeys, &resMap, required)
if err != nil {
return nil, fmt.Errorf("fetch and add local keys: %v", err)
}
p2pmetrics.ReportFoundLocal(p2pmetrics.TaskIDFromContext(ctx), int(foundLocalCount))

if foundLocalCount >= required {
logtrace.Debug(ctx, "DHT BatchRetrieve satisfied from local storage", logtrace.Fields{
"txid": txID, "found_local": foundLocalCount, "required": required,
})
return result, nil
}

if len(localOnly) > 0 && localOnly[0] {
logtrace.Debug(ctx, "DHT BatchRetrieve local-only mode, insufficient keys", logtrace.Fields{
"txid": txID, "found_local": foundLocalCount, "required": required,
})
return result, fmt.Errorf("local-only: found %d, required %d", foundLocalCount, required)
}

supernodeAddr, addrErr := s.getSupernodeAddress(ctx)
if addrErr != nil {
logtrace.Warn(ctx, "Failed to get supernode address", logtrace.Fields{
logtrace.FieldModule: "dht",
logtrace.FieldError: addrErr.Error(),
})
}
hostIP := parseSupernodeAddress(supernodeAddr)
self := &Node{ID: s.ht.self.ID, IP: hostIP, Port: s.ht.self.Port}
self.SetHashedID()

knownNodes := make(map[string]*Node)
var knownMu sync.Mutex

for _, n := range s.ht.nodes() {
nn := &Node{ID: n.ID, IP: n.IP, Port: n.Port}
nn.SetHashedID()
knownNodes[string(nn.ID)] = nn
}

ignoreList := s.ignorelist.ToNodeList()

globalClosestContacts := make(map[string]*NodeList)
var closestMu sync.RWMutex

for i := range keys {
top6 := s.ht.closestContactsWithIncludingNode(Alpha, hashes[i], s.ignorelist.ToNodeList(), nil)
if _, found := resMap.Load(hexKeys[i]); found {
continue
}

top6 := s.ht.closestContactsWithIncludingNode(Alpha, hashes[i], ignoreList, nil)
closestMu.Lock()
globalClosestContacts[keys[i]] = top6
closestMu.Unlock()
Expand All @@ -734,22 +766,12 @@ func (s *DHT) BatchRetrieve(ctx context.Context, keys []string, required int32,

delete(knownNodes, string(self.ID))

foundLocalCount, err = s.fetchAndAddLocalKeys(ctx, hexKeys, &resMap, required)
if err != nil {
return nil, fmt.Errorf("fetch and add local keys: %v", err)
}
// Report how many were found locally, for event metrics
p2pmetrics.ReportFoundLocal(p2pmetrics.TaskIDFromContext(ctx), int(foundLocalCount))
if foundLocalCount >= required {
return result, nil
}

batchSize := batchRetrieveSize
var networkFound int32
totalBatches := int(math.Ceil(float64(required) / float64(batchSize)))
parallelBatches := int(math.Min(float64(totalBatches), float64(fetchSymbolsBatchConcurrency)))

semaphore := make(chan struct{}, parallelBatches)
sem := semaphore.NewWeighted(int64(parallelBatches))
var wg sync.WaitGroup
gctx, cancel := context.WithCancel(ctx)
defer cancel()
Expand All @@ -759,28 +781,40 @@ func (s *DHT) BatchRetrieve(ctx context.Context, keys []string, required int32,
if end > len(keys) {
end = len(keys)
}

if atomic.LoadInt32(&networkFound)+int32(foundLocalCount) >= int32(required) {
break
}

wg.Add(1)
semaphore <- struct{}{}
go s.processBatch(
gctx,
keys[start:end],
hexKeys[start:end],
semaphore, &wg,
globalClosestContacts,
&closestMu,
knownNodes, &knownMu,
&resMap,
required,
foundLocalCount,
&networkFound,
cancel,
txID,
writer,
)
go func(start, end int) {
defer wg.Done()

if err := sem.Acquire(gctx, 1); err != nil {
return
}
defer sem.Release(1)

if atomic.LoadInt32(&networkFound)+int32(foundLocalCount) >= int32(required) {
return
}

s.processBatch(
gctx,
keys[start:end],
hexKeys[start:end],
globalClosestContacts,
&closestMu,
knownNodes, &knownMu,
&resMap,
required,
foundLocalCount,
&networkFound,
cancel,
txID,
writer,
)
}(start, end)
}

wg.Wait()
Expand All @@ -798,8 +832,6 @@ func (s *DHT) processBatch(
ctx context.Context,
batchKeys []string,
batchHexKeys []string,
semaphore chan struct{},
wg *sync.WaitGroup,
globalClosestContacts map[string]*NodeList,
closestMu *sync.RWMutex,
knownNodes map[string]*Node,
Expand All @@ -812,95 +844,61 @@ func (s *DHT) processBatch(
txID string,
writer func(symbolID string, data []byte) error,
) {
defer wg.Done()
defer func() { <-semaphore }()
select {
case <-ctx.Done():
return
default:
}

for i := 0; i < maxIterations; i++ {
select {
case <-ctx.Done():
return
default:
}
fetchMap := make(map[string][]int)

// Build fetch map (read globalClosestContacts under RLock)
fetchMap := make(map[string][]int)
for i, key := range batchKeys {
closestMu.RLock()
nl := globalClosestContacts[key]
closestMu.RUnlock()
if nl == nil {
continue
}
for _, node := range nl.Nodes {
nodeID := string(node.ID)
fetchMap[nodeID] = append(fetchMap[nodeID], i)
}
}
closestMu.RLock()
localContacts := make(map[string]*NodeList, len(batchKeys))
for _, key := range batchKeys {
localContacts[key] = globalClosestContacts[key]
}
closestMu.RUnlock()

foundCount, newClosestContacts, batchErr := s.iterateBatchGetValues(
ctx, knownNodes, batchKeys, batchHexKeys, fetchMap, resMap, required, foundLocalCount+atomic.LoadInt32(networkFound), writer,
)
if batchErr != nil {
logtrace.Error(ctx, "Iterate batch get values failed", logtrace.Fields{
logtrace.FieldModule: "dht", "txid": txID, logtrace.FieldError: batchErr.Error(),
})
for idx, key := range batchKeys {
nl := localContacts[key]
if nl == nil {
continue
}

atomic.AddInt32(networkFound, int32(foundCount))
if atomic.LoadInt32(networkFound)+int32(foundLocalCount) >= int32(required) {
cancel()
break
for _, node := range nl.Nodes {
nodeID := string(node.ID)
fetchMap[nodeID] = append(fetchMap[nodeID], idx)
}
}

changed := false
for key, nodesList := range newClosestContacts {
if nodesList == nil || nodesList.Nodes == nil {
continue
}

closestMu.RLock()
curr := globalClosestContacts[key]
closestMu.RUnlock()
if curr == nil || curr.Nodes == nil {
logtrace.Warn(ctx, "Global contacts missing key during merge", logtrace.Fields{"key": key})
continue
}

if !haveAllNodes(nodesList.Nodes, curr.Nodes) {
changed = true
}

nodesList.AddNodes(curr.Nodes)
nodesList.Sort()
nodesList.TopN(Alpha)

s.addKnownNodesSafe(ctx, nodesList.Nodes, knownNodes, knownMu)

closestMu.Lock()
globalClosestContacts[key] = nodesList
closestMu.Unlock()
}
foundCount, batchErr := s.iterateBatchGetValues(
ctx, knownNodes, batchHexKeys, fetchMap, resMap, required, foundLocalCount+atomic.LoadInt32(networkFound), writer,
)
if batchErr != nil {
logtrace.Error(ctx, "Iterate batch get values failed", logtrace.Fields{
logtrace.FieldModule: "dht", "txid": txID, logtrace.FieldError: batchErr.Error(),
})
}

if !changed {
break
}
atomic.AddInt32(networkFound, int32(foundCount))
if atomic.LoadInt32(networkFound)+int32(foundLocalCount) >= int32(required) {
cancel()
}
}

func (s *DHT) iterateBatchGetValues(ctx context.Context, nodes map[string]*Node, keys []string, hexKeys []string, fetchMap map[string][]int,
resMap *sync.Map, req, alreadyFound int32, writer func(symbolID string, data []byte) error) (int, map[string]*NodeList, error) {
semaphore := make(chan struct{}, storeSameSymbolsBatchConcurrency) // Limit concurrency to 3
closestContacts := make(map[string]*NodeList)
func (s *DHT) iterateBatchGetValues(ctx context.Context, nodes map[string]*Node, hexKeys []string, fetchMap map[string][]int,
resMap *sync.Map, req, alreadyFound int32, writer func(symbolID string, data []byte) error) (int, error) {
sem := semaphore.NewWeighted(int64(storeSameSymbolsBatchConcurrency))
var wg sync.WaitGroup
contactsMap := make(map[string]map[string][]*Node)
var firstErr error
var mu sync.Mutex // To protect the firstErr
foundCount := int32(0)

gctx, cancel := context.WithCancel(ctx) // Create a cancellable context
defer cancel()
for nodeID, node := range nodes {
if _, ok := fetchMap[nodeID]; !ok {

for nodeID := range fetchMap {
node, ok := nodes[nodeID]
if !ok {
continue
}

Expand All @@ -912,18 +910,17 @@ func (s *DHT) iterateBatchGetValues(ctx context.Context, nodes map[string]*Node,
continue
}

contactsMap[nodeID] = make(map[string][]*Node)
wg.Add(1)
go func(node *Node, nodeID string) {
defer wg.Done()

select {
case <-ctx.Done():
if err := sem.Acquire(gctx, 1); err != nil {
return
case <-gctx.Done():
}
defer sem.Release(1)

if atomic.LoadInt32(&foundCount) >= int32(req-alreadyFound) {
return
case semaphore <- struct{}{}:
defer func() { <-semaphore }()
}

callStart := time.Now()
Expand Down Expand Up @@ -986,8 +983,6 @@ func (s *DHT) iterateBatchGetValues(ctx context.Context, nodes map[string]*Node,
break
}
}
} else {
contactsMap[nodeID][k] = v.Closest
}
}

Expand All @@ -1000,6 +995,7 @@ func (s *DHT) iterateBatchGetValues(ctx context.Context, nodes map[string]*Node,
Error: "",
DurationMS: time.Since(callStart).Milliseconds(),
})

}(node, nodeID)
}

Expand All @@ -1018,33 +1014,7 @@ func (s *DHT) iterateBatchGetValues(ctx context.Context, nodes map[string]*Node,
})
}

for _, closestNodes := range contactsMap {
for key, nodes := range closestNodes {
comparator, err := hex.DecodeString(key)
if err != nil {
logtrace.Error(ctx, "Failed to decode hex key in closestNodes.Range", logtrace.Fields{
logtrace.FieldModule: "dht",
"key": key,
logtrace.FieldError: err.Error(),
})
return 0, nil, err
}
bkey := base58.Encode(comparator)

if _, ok := closestContacts[bkey]; !ok {
closestContacts[bkey] = &NodeList{Nodes: nodes, Comparator: comparator}
} else {
closestContacts[bkey].AddNodes(nodes)
}
}
}

for key, nodes := range closestContacts {
nodes.Sort()
nodes.TopN(Alpha)
closestContacts[key] = nodes
}
return int(foundCount), closestContacts, firstErr
return int(foundCount), firstErr
}

func (s *DHT) doBatchGetValuesCall(ctx context.Context, node *Node, requestKeys map[string]KeyValWithClosest) (map[string]KeyValWithClosest, error) {
Expand Down
Loading
Loading