Skip to content

Commit

Permalink
fix
Browse files Browse the repository at this point in the history
Signed-off-by: kpango <i.can.feel.gravity@gmail.com>
  • Loading branch information
kpango committed Oct 13, 2020
1 parent e887e94 commit 0060b50
Showing 1 changed file with 7 additions and 24 deletions.
31 changes: 7 additions & 24 deletions pkg/gateway/vald/handler/grpc/handler.go
Original file line number Diff line number Diff line change
Expand Up @@ -130,7 +130,8 @@ func (s *server) search(ctx context.Context, cfg *payload.Search_Config,
span.End()
}
}()
maxDist := uint32(math.MaxUint32)
var maxDist uint32
atomic.StoreUint32(&maxDist, math.Float32bits(math.MaxFloat32))
num := int(cfg.GetNum())
res = new(payload.Search_Response)
res.Results = make([]*payload.Object_Distance, 0, s.gateway.GetAgentCount(ctx)*num)
Expand All @@ -147,8 +148,6 @@ func (s *server) search(ctx context.Context, cfg *payload.Search_Config,

eg.Go(safety.RecoverFunc(func() error {
defer cancel()
// visited := make(map[string]bool, len(res.Results))
// mu := sync.RWMutex{}
visited := new(sync.Map)
return s.gateway.BroadCast(ectx, func(ctx context.Context, target string, ac agent.AgentClient, copts ...grpc.CallOption) error {
r, err := f(ctx, ac, copts...)
Expand All @@ -157,25 +156,16 @@ func (s *server) search(ctx context.Context, cfg *payload.Search_Config,
return nil
}
for _, dist := range r.GetResults() {
if dist.GetDistance() > math.Float32frombits(atomic.LoadUint32(&maxDist)) {
if dist.GetDistance() >= math.Float32frombits(atomic.LoadUint32(&maxDist)) {
return nil
}
id := dist.GetId()
// mu.RLock()
// already := visited[id]
// mu.RUnlock()
_, already := visited.LoadOrStore(id, struct{}{})
if !already {
// mu.Lock()
// visited[id] = true
// mu.Unlock()
if _, already := visited.LoadOrStore(dist.GetId(), struct{}{}); !already {
dch <- dist
}
}
return nil
})
}))
// vl := make(map[string]struct{}, cap(res.GetResults()))
for {
select {
case <-ectx.Done():
Expand Down Expand Up @@ -217,17 +207,9 @@ func (s *server) search(ctx context.Context, cfg *payload.Search_Config,
return res, nil
case dist := <-dch:
pos := len(res.GetResults())
if pos >= num {
if dist.GetDistance() < math.Float32frombits(atomic.LoadUint32(&maxDist)) {
atomic.StoreUint32(&maxDist, math.Float32bits(dist.GetDistance()))
} else {
continue
}
if pos >= num && dist.GetDistance() >= math.Float32frombits(atomic.LoadUint32(&maxDist)) {
continue
}
// if _, ok := vl[dist.GetId()]; ok {
// continue
// }
// vl[dist.GetId()] = struct{}{}
switch pos {
case 0:
res.Results = append(res.Results, dist)
Expand Down Expand Up @@ -255,6 +237,7 @@ func (s *server) search(ctx context.Context, cfg *payload.Search_Config,
}
if len(res.GetResults()) > num && num != 0 {
res.Results = res.GetResults()[:num]
atomic.StoreUint32(&maxDist, math.Float32bits(res.GetResults()[len(res.GetResults())].GetDistance()))
}
}
}
Expand Down

0 comments on commit 0060b50

Please sign in to comment.