提交 786406bd 作者: Brian Tiger Chow

Merge pull request #588 from jbenet/misc/2015-01-16

Miscellaneous changes
...@@ -108,7 +108,6 @@ type bitswap struct { ...@@ -108,7 +108,6 @@ type bitswap struct {
// GetBlock attempts to retrieve a particular block from peers within the // GetBlock attempts to retrieve a particular block from peers within the
// deadline enforced by the context. // deadline enforced by the context.
func (bs *bitswap) GetBlock(parent context.Context, k u.Key) (*blocks.Block, error) { func (bs *bitswap) GetBlock(parent context.Context, k u.Key) (*blocks.Block, error) {
log := log.Prefix("bitswap(%s).GetBlock(%s)", bs.self, k)
// Any async work initiated by this function must end when this function // Any async work initiated by this function must end when this function
// returns. To ensure this, derive a new context. Note that it is okay to // returns. To ensure this, derive a new context. Note that it is okay to
...@@ -121,11 +120,9 @@ func (bs *bitswap) GetBlock(parent context.Context, k u.Key) (*blocks.Block, err ...@@ -121,11 +120,9 @@ func (bs *bitswap) GetBlock(parent context.Context, k u.Key) (*blocks.Block, err
ctx = eventlog.ContextWithLoggable(ctx, eventlog.Uuid("GetBlockRequest")) ctx = eventlog.ContextWithLoggable(ctx, eventlog.Uuid("GetBlockRequest"))
defer log.EventBegin(ctx, "GetBlockRequest", &k).Done() defer log.EventBegin(ctx, "GetBlockRequest", &k).Done()
log.Debugf("GetBlockRequestBegin")
defer func() { defer func() {
cancelFunc() cancelFunc()
log.Debugf("GetBlockRequestEnd")
}() }()
promise, err := bs.GetBlocks(ctx, []u.Key{k}) promise, err := bs.GetBlocks(ctx, []u.Key{k})
...@@ -150,7 +147,6 @@ func (bs *bitswap) GetBlock(parent context.Context, k u.Key) (*blocks.Block, err ...@@ -150,7 +147,6 @@ func (bs *bitswap) GetBlock(parent context.Context, k u.Key) (*blocks.Block, err
// resources, provide a context with a reasonably short deadline (ie. not one // resources, provide a context with a reasonably short deadline (ie. not one
// that lasts throughout the lifetime of the server) // that lasts throughout the lifetime of the server)
func (bs *bitswap) GetBlocks(ctx context.Context, keys []u.Key) (<-chan *blocks.Block, error) { func (bs *bitswap) GetBlocks(ctx context.Context, keys []u.Key) (<-chan *blocks.Block, error) {
// TODO log the request
promise := bs.notifications.Subscribe(ctx, keys...) promise := bs.notifications.Subscribe(ctx, keys...)
select { select {
...@@ -172,18 +168,6 @@ func (bs *bitswap) HasBlock(ctx context.Context, blk *blocks.Block) error { ...@@ -172,18 +168,6 @@ func (bs *bitswap) HasBlock(ctx context.Context, blk *blocks.Block) error {
return bs.network.Provide(ctx, blk.Key()) return bs.network.Provide(ctx, blk.Key())
} }
func (bs *bitswap) sendWantlistMsgToPeer(ctx context.Context, m bsmsg.BitSwapMessage, p peer.ID) error {
log := log.Prefix("bitswap(%s).bitswap.sendWantlistMsgToPeer(%d, %s)", bs.self, len(m.Wantlist()), p)
log.Debug("sending wantlist")
if err := bs.send(ctx, p, m); err != nil {
log.Errorf("send wantlist error: %s", err)
return err
}
log.Debugf("send wantlist success")
return nil
}
func (bs *bitswap) sendWantlistMsgToPeers(ctx context.Context, m bsmsg.BitSwapMessage, peers <-chan peer.ID) error { func (bs *bitswap) sendWantlistMsgToPeers(ctx context.Context, m bsmsg.BitSwapMessage, peers <-chan peer.ID) error {
if peers == nil { if peers == nil {
panic("Cant send wantlist to nil peerchan") panic("Cant send wantlist to nil peerchan")
...@@ -207,7 +191,9 @@ func (bs *bitswap) sendWantlistMsgToPeers(ctx context.Context, m bsmsg.BitSwapMe ...@@ -207,7 +191,9 @@ func (bs *bitswap) sendWantlistMsgToPeers(ctx context.Context, m bsmsg.BitSwapMe
wg.Add(1) wg.Add(1)
go func(p peer.ID) { go func(p peer.ID) {
defer wg.Done() defer wg.Done()
bs.sendWantlistMsgToPeer(ctx, m, p) if err := bs.send(ctx, p, m); err != nil {
log.Error(err) // TODO remove if too verbose
}
}(peerToQuery) }(peerToQuery)
} }
wg.Wait() wg.Wait()
...@@ -304,23 +290,19 @@ func (bs *bitswap) clientWorker(parent context.Context) { ...@@ -304,23 +290,19 @@ func (bs *bitswap) clientWorker(parent context.Context) {
case <-broadcastSignal: // resend unfulfilled wantlist keys case <-broadcastSignal: // resend unfulfilled wantlist keys
bs.sendWantlistToProviders(ctx) bs.sendWantlistToProviders(ctx)
broadcastSignal = time.After(rebroadcastDelay.Get()) broadcastSignal = time.After(rebroadcastDelay.Get())
case ks := <-bs.batchRequests: case keys := <-bs.batchRequests:
if len(ks) == 0 { if len(keys) == 0 {
log.Warning("Received batch request for zero blocks") log.Warning("Received batch request for zero blocks")
continue continue
} }
for i, k := range ks { for i, k := range keys {
bs.wantlist.Add(k, kMaxPriority-i) bs.wantlist.Add(k, kMaxPriority-i)
} }
// NB: send want list to providers for the first peer in this list. // NB: Optimization. Assumes that providers of key[0] are likely to
// the assumption is made that the providers of the first key in // be able to provide for all keys. This currently holds true in most
// the set are likely to have others as well. // every situation. Later, this assumption may not hold as true.
// This currently holds true in most every situation, since when
// pinning a file, you store and provide all blocks associated with
// it. Later, this assumption may not hold as true if we implement
// newer bitswap strategies.
child, _ := context.WithTimeout(ctx, providerRequestTimeout) child, _ := context.WithTimeout(ctx, providerRequestTimeout)
providers := bs.network.FindProvidersAsync(child, ks[0], maxProvidersPerRequest) providers := bs.network.FindProvidersAsync(child, keys[0], maxProvidersPerRequest)
err := bs.sendWantlistToPeers(ctx, providers) err := bs.sendWantlistToPeers(ctx, providers)
if err != nil { if err != nil {
log.Errorf("error sending wantlist: %s", err) log.Errorf("error sending wantlist: %s", err)
......
...@@ -160,6 +160,9 @@ func (e *Engine) Peers() []peer.ID { ...@@ -160,6 +160,9 @@ func (e *Engine) Peers() []peer.ID {
// MessageReceived performs book-keeping. Returns error if passed invalid // MessageReceived performs book-keeping. Returns error if passed invalid
// arguments. // arguments.
func (e *Engine) MessageReceived(p peer.ID, m bsmsg.BitSwapMessage) error { func (e *Engine) MessageReceived(p peer.ID, m bsmsg.BitSwapMessage) error {
e.lock.Lock()
defer e.lock.Unlock()
log := log.Prefix("bitswap.Engine.MessageReceived(%s)", p) log := log.Prefix("bitswap.Engine.MessageReceived(%s)", p)
log.Debugf("enter. %d entries %d blocks", len(m.Wantlist()), len(m.Blocks())) log.Debugf("enter. %d entries %d blocks", len(m.Wantlist()), len(m.Blocks()))
defer log.Debugf("exit") defer log.Debugf("exit")
...@@ -175,9 +178,6 @@ func (e *Engine) MessageReceived(p peer.ID, m bsmsg.BitSwapMessage) error { ...@@ -175,9 +178,6 @@ func (e *Engine) MessageReceived(p peer.ID, m bsmsg.BitSwapMessage) error {
} }
}() }()
e.lock.Lock()
defer e.lock.Unlock()
l := e.findOrCreate(p) l := e.findOrCreate(p)
if m.Full() { if m.Full() {
l.wantList = wl.New() l.wantList = wl.New()
......
...@@ -32,7 +32,7 @@ const ( ...@@ -32,7 +32,7 @@ const (
// DefaultConfigFile is the filename of the configuration file // DefaultConfigFile is the filename of the configuration file
DefaultConfigFile = "config" DefaultConfigFile = "config"
// EnvDir is the environment variable used to change the path root. // EnvDir is the environment variable used to change the path root.
EnvDir = "IPFS_DIR" EnvDir = "IPFS_PATH"
) )
// PathRoot returns the default configuration root directory // PathRoot returns the default configuration root directory
......
...@@ -67,7 +67,7 @@ test_wait_output_n_lines_60_sec() { ...@@ -67,7 +67,7 @@ test_wait_output_n_lines_60_sec() {
test_init_ipfs() { test_init_ipfs() {
test_expect_success "ipfs init succeeds" ' test_expect_success "ipfs init succeeds" '
export IPFS_DIR="$(pwd)/.go-ipfs" && export IPFS_PATH="$(pwd)/.go-ipfs" &&
ipfs init -b=1024 > /dev/null ipfs init -b=1024 > /dev/null
' '
......
...@@ -9,7 +9,7 @@ test_description="Test init command" ...@@ -9,7 +9,7 @@ test_description="Test init command"
. lib/test-lib.sh . lib/test-lib.sh
test_expect_success "ipfs init succeeds" ' test_expect_success "ipfs init succeeds" '
export IPFS_DIR="$(pwd)/.go-ipfs" && export IPFS_PATH="$(pwd)/.go-ipfs" &&
ipfs init >actual_init ipfs init >actual_init
' '
...@@ -35,7 +35,7 @@ test_expect_success "ipfs peer id looks good" ' ...@@ -35,7 +35,7 @@ test_expect_success "ipfs peer id looks good" '
test_expect_success "ipfs init output looks good" ' test_expect_success "ipfs init output looks good" '
STARTHASH="QmYpv2VEsxzTTXRYX3PjDg961cnJE3kY1YDXLycHGQ3zZB" && STARTHASH="QmYpv2VEsxzTTXRYX3PjDg961cnJE3kY1YDXLycHGQ3zZB" &&
echo "initializing ipfs node at $IPFS_DIR" >expected && echo "initializing ipfs node at $IPFS_PATH" >expected &&
echo "generating key pair...done" >>expected && echo "generating key pair...done" >>expected &&
echo "peer identity: $PEERID" >>expected && echo "peer identity: $PEERID" >>expected &&
printf "\\n%s\\n" "to get started, enter: ipfs cat $STARTHASH" >>expected && printf "\\n%s\\n" "to get started, enter: ipfs cat $STARTHASH" >>expected &&
......
...@@ -13,7 +13,7 @@ test_description="Test daemon command" ...@@ -13,7 +13,7 @@ test_description="Test daemon command"
# NOTE: this should remove bootstrap peers (needs a flag) # NOTE: this should remove bootstrap peers (needs a flag)
test_expect_success "ipfs daemon --init launches" ' test_expect_success "ipfs daemon --init launches" '
export IPFS_DIR="$(pwd)/.go-ipfs" && export IPFS_PATH="$(pwd)/.go-ipfs" &&
ipfs daemon --init 2>&1 >actual_init & ipfs daemon --init 2>&1 >actual_init &
' '
...@@ -35,7 +35,7 @@ test_expect_success "ipfs peer id looks good" ' ...@@ -35,7 +35,7 @@ test_expect_success "ipfs peer id looks good" '
# note this is almost the same as t0020-init.sh "ipfs init output looks good" # note this is almost the same as t0020-init.sh "ipfs init output looks good"
test_expect_success "ipfs daemon output looks good" ' test_expect_success "ipfs daemon output looks good" '
STARTHASH="QmYpv2VEsxzTTXRYX3PjDg961cnJE3kY1YDXLycHGQ3zZB" && STARTHASH="QmYpv2VEsxzTTXRYX3PjDg961cnJE3kY1YDXLycHGQ3zZB" &&
echo "initializing ipfs node at $IPFS_DIR" >expected && echo "initializing ipfs node at $IPFS_PATH" >expected &&
echo "generating key pair...done" >>expected && echo "generating key pair...done" >>expected &&
echo "peer identity: $PEERID" >>expected && echo "peer identity: $PEERID" >>expected &&
echo "\nto get started, enter: ipfs cat $STARTHASH" >>expected && echo "\nto get started, enter: ipfs cat $STARTHASH" >>expected &&
......
Markdown 格式
0%
您添加了 0 到此讨论。请谨慎行事。
请先完成此评论的编辑!
注册 或者 后发表评论