query.go 7.3 KB
Newer Older
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
1 2 3
package dht

import (
4 5
	"sync"

6
	key "github.com/ipfs/go-ipfs/blocks/key"
7 8 9 10 11 12 13 14
	notif "github.com/ipfs/go-ipfs/notifications"
	peer "github.com/ipfs/go-ipfs/p2p/peer"
	queue "github.com/ipfs/go-ipfs/p2p/peer/queue"
	"github.com/ipfs/go-ipfs/routing"
	eventlog "github.com/ipfs/go-ipfs/thirdparty/eventlog"
	u "github.com/ipfs/go-ipfs/util"
	pset "github.com/ipfs/go-ipfs/util/peerset"
	todoctr "github.com/ipfs/go-ipfs/util/todocounter"
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
15

16 17 18
	process "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/goprocess"
	ctxproc "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/goprocess/context"
	context "github.com/ipfs/go-ipfs/Godeps/_workspace/src/golang.org/x/net/context"
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
19 20
)

21
var maxQueryConcurrency = AlphaValue
22

Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
23
type dhtQuery struct {
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
24
	dht         *IpfsDHT
25
	key         key.Key   // the key we're querying for
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
26 27
	qfunc       queryFunc // the function to execute per peer
	concurrency int       // the concurrency parameter
28 29 30
}

type dhtQueryResult struct {
31 32 33 34
	value         []byte          // GetValue
	peer          peer.PeerInfo   // FindPeer
	providerPeers []peer.PeerInfo // GetProviders
	closerPeers   []peer.PeerInfo // *
35 36 37 38
	success       bool
}

// constructs query
39
func (dht *IpfsDHT) newQuery(k key.Key, f queryFunc) *dhtQuery {
40 41
	return &dhtQuery{
		key:         k,
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
42
		dht:         dht,
43 44 45
		qfunc:       f,
		concurrency: maxQueryConcurrency,
	}
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
46 47 48 49 50 51 52
}

// QueryFunc is a function that runs a particular query with a given peer.
// It returns either:
// - the value
// - a list of peers potentially better able to serve the query
// - an error
53
type queryFunc func(context.Context, peer.ID) (*dhtQueryResult, error)
54 55

// Run runs the query at hand. pass in a list of peers to use first.
56
func (q *dhtQuery) Run(ctx context.Context, peers []peer.ID) (*dhtQueryResult, error) {
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
57 58 59 60 61 62
	select {
	case <-ctx.Done():
		return nil, ctx.Err()
	default:
	}

63 64 65
	ctx, cancel := context.WithCancel(ctx)
	defer cancel()

66 67
	runner := newQueryRunner(q)
	return runner.Run(ctx, peers)
68 69 70
}

type dhtQueryRunner struct {
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
71 72 73 74
	query          *dhtQuery        // query to run
	peersSeen      *pset.PeerSet    // all peers queried. prevent querying same peer 2x
	peersToQuery   *queue.ChanQueue // peers remaining to be queried
	peersRemaining todoctr.Counter  // peersToQuery + currently processing
75

Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
76
	result *dhtQueryResult // query result
77
	errs   u.MultiErr      // result errors. maybe should be a map[peer.ID]error
78

Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
79 80
	rateLimit chan struct{} // processing semaphore
	log       eventlog.EventLogger
81

82
	proc process.Process
83 84 85
	sync.RWMutex
}

86 87 88
func newQueryRunner(q *dhtQuery) *dhtQueryRunner {
	proc := process.WithParent(process.Background())
	ctx := ctxproc.WithProcessClosing(context.Background(), proc)
89 90 91 92
	return &dhtQueryRunner{
		query:          q,
		peersToQuery:   queue.NewChanQueue(ctx, queue.NewXORDistancePQ(q.key)),
		peersRemaining: todoctr.NewSyncCounter(),
93
		peersSeen:      pset.New(),
94
		rateLimit:      make(chan struct{}, q.concurrency),
95
		proc:           proc,
96 97 98
	}
}

99
func (r *dhtQueryRunner) Run(ctx context.Context, peers []peer.ID) (*dhtQueryResult, error) {
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
100 101
	r.log = log

102 103 104 105
	if len(peers) == 0 {
		log.Warning("Running query with no peers!")
		return nil, nil
	}
106

107 108 109
	// setup concurrency rate limiting
	for i := 0; i < r.query.concurrency; i++ {
		r.rateLimit <- struct{}{}
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
110 111
	}

112 113
	// add all the peers we got first.
	for _, p := range peers {
114
		r.addPeerToQuery(p)
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
115 116
	}

Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
117
	// go do this thing.
118
	// do it as a child proc to make sure Run exits
119
	// ONLY AFTER spawn workers has exited.
120
	r.proc.Go(r.spawnWorkers)
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
121 122 123 124

	// so workers are working.

	// wait until they're done.
125
	err := routing.ErrNotFound
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
126

127 128 129
	// now, if the context finishes, close the proc.
	// we have to do it here because the logic before is setup, which
	// should run without closing the proc.
rht's avatar
rht committed
130
	ctxproc.CloseAfterContext(r.proc, ctx)
131

Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
132
	select {
133
	case <-r.peersRemaining.Done():
134
		r.proc.Close()
135 136 137
		r.RLock()
		defer r.RUnlock()

138 139 140 141 142 143
		err = routing.ErrNotFound

		// if every query to every peer failed, something must be very wrong.
		if len(r.errs) > 0 && len(r.errs) == r.peersSeen.Size() {
			log.Debugf("query errs: %s", r.errs)
			err = r.errs[0]
144 145
		}

146
	case <-r.proc.Closed():
147 148
		r.RLock()
		defer r.RUnlock()
149
		err = context.DeadlineExceeded
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
150
	}
151

Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
152 153
	if r.result != nil && r.result.success {
		return r.result, nil
154 155
	}

Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
156
	return nil, err
157 158
}

159
func (r *dhtQueryRunner) addPeerToQuery(next peer.ID) {
160
	// if new peer is ourselves...
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
161
	if next == r.query.dht.self {
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
162
		r.log.Debug("addPeerToQuery skip self")
163 164 165
		return
	}

166
	if !r.peersSeen.TryAdd(next) {
167 168 169 170 171 172
		return
	}

	r.peersRemaining.Increment(1)
	select {
	case r.peersToQuery.EnqChan <- next:
173
	case <-r.proc.Closing():
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
174
	}
175 176
}

177
func (r *dhtQueryRunner) spawnWorkers(proc process.Process) {
178
	for {
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
179

180 181 182 183
		select {
		case <-r.peersRemaining.Done():
			return

184
		case <-r.proc.Closing():
185 186
			return

Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
187 188 189 190
		case p, more := <-r.peersToQuery.DeqChan:
			if !more {
				return // channel closed.
			}
191 192 193

			// do it as a child func to make sure Run exits
			// ONLY AFTER spawn workers has exited.
194 195
			proc.Go(func(proc process.Process) {
				r.queryPeer(proc, p)
196
			})
197 198 199
		}
	}
}
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
200

201
func (r *dhtQueryRunner) queryPeer(proc process.Process, p peer.ID) {
202 203 204
	// make sure we rate limit concurrency.
	select {
	case <-r.rateLimit:
205
	case <-proc.Closing():
206 207 208 209
		r.peersRemaining.Decrement(1)
		return
	}

Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
210 211
	// ok let's do this!

212 213 214
	// create a context from our proc.
	ctx := ctxproc.WithProcessClosing(context.Background(), proc)

Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
215 216 217 218 219 220 221 222
	// make sure we do this when we exit
	defer func() {
		// signal we're done proccessing peer p
		r.peersRemaining.Decrement(1)
		r.rateLimit <- struct{}{}
	}()

	// make sure we're connected to the peer.
223
	// FIXME abstract away into the network layer
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
224
	if conns := r.query.dht.host.Network().ConnsToPeer(p); len(conns) == 0 {
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
225
		log.Infof("not connected. dialing.")
226 227 228
		// while we dial, we do not take up a rate limit. this is to allow
		// forward progress during potentially very high latency dials.
		r.rateLimit <- struct{}{}
229

Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
230
		pi := peer.PeerInfo{ID: p}
231 232

		if err := r.query.dht.host.Connect(ctx, pi); err != nil {
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
233
			log.Debugf("Error connecting: %s", err)
234

235
			notif.PublishQueryEvent(ctx, &notif.QueryEvent{
236 237 238 239
				Type:  notif.QueryError,
				Extra: err.Error(),
			})

240 241 242
			r.Lock()
			r.errs = append(r.errs, err)
			r.Unlock()
243
			<-r.rateLimit // need to grab it again, as we deferred.
244 245
			return
		}
246
		<-r.rateLimit // need to grab it again, as we deferred.
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
247
		log.Debugf("connected. dial success.")
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
248
	}
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
249

250
	// finally, run the query against this peer
251
	res, err := r.query.qfunc(ctx, p)
252 253

	if err != nil {
254
		log.Debugf("ERROR worker for: %v %v", p, err)
255 256 257 258 259
		r.Lock()
		r.errs = append(r.errs, err)
		r.Unlock()

	} else if res.success {
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
260
		log.Debugf("SUCCESS worker for: %v %s", p, res)
261 262 263
		r.Lock()
		r.result = res
		r.Unlock()
264
		go r.proc.Close() // signal to everyone that we're done.
265
		// must be async, as we're one of the children, and Close blocks.
266

267 268
	} else if len(res.closerPeers) > 0 {
		log.Debugf("PEERS CLOSER -- worker for: %v (%d closer peers)", p, len(res.closerPeers))
269
		for _, next := range res.closerPeers {
270 271 272 273 274
			if next.ID == r.query.dht.self { // dont add self.
				log.Debugf("PEERS CLOSER -- worker for: %v found self", p)
				continue
			}

275
			// add their addresses to the dialer's peerstore
276
			r.query.dht.peerstore.AddAddrs(next.ID, next.Addrs, peer.TempAddrTTL)
277
			r.addPeerToQuery(next.ID)
278
			log.Debugf("PEERS CLOSER -- worker for: %v added %v (%v)", p, next.ID, next.Addrs)
279
		}
280 281
	} else {
		log.Debugf("QUERY worker for: %v - not found, and no closer peers.", p)
Juan Batiz-Benet's avatar
Juan Batiz-Benet committed
282 283
	}
}