Bitswap

.gitignore

@@ -0,0 +1,4 @@
+*.swp
+.ipfsconfig
+*.out
+*.test

README.md

@@ -62,6 +62,15 @@ Guidelines:
 - if you'd like to work on ipfs part-time (20+ hrs/wk) or full-time (40+ hrs/wk), contact [@jbenet](https://github.com/jbenet)
 - have fun!
 
+## Todo
+
+Ipfs is still under heavy development, there is a lot to be done!
+
+- [ ] Finish Bitswap
+- [ ] Connect fuse interface to Blockservice
+- [ ] Write tests for bitswap
+- [ ] Come up with more TODO items
+
 ## Development Dependencies
 
 If you make changes to the protocol buffers, you will need to install the [protoc compiler](https://code.google.com/p/protobuf/downloads/list).

bitswap/bitswap.go

@@ -1,14 +1,16 @@
 package bitswap
 
 import (
+	"code.google.com/p/goprotobuf/proto"
 	blocks "github.com/jbenet/go-ipfs/blocks"
 	peer "github.com/jbenet/go-ipfs/peer"
+	routing "github.com/jbenet/go-ipfs/routing"
+	dht "github.com/jbenet/go-ipfs/routing/dht"
 	swarm "github.com/jbenet/go-ipfs/swarm"
 	u "github.com/jbenet/go-ipfs/util"
 
 	ds "github.com/jbenet/datastore.go"
 
-	"errors"
 	"time"
 )
 
@@ -27,12 +29,18 @@ type BitSwap struct {
 	peer *peer.Peer
 
 	// net holds the connections to all peers.
-	net swarm.Network
+	net     swarm.Network
+	meschan *swarm.Chan
 
 	// datastore is the local database
 	// Ledgers of known
 	datastore ds.Datastore
 
+	// routing interface for communication
+	routing *dht.IpfsDHT
+
+	listener *swarm.MesListener
+
 	// partners is a map of currently active bitswap relationships.
 	// The Ledger has the peer.ID, and the peer connection works through net.
 	// Ledgers of known relationships (active or inactive) stored in datastore.
@@ -44,27 +52,183 @@ type BitSwap struct {
 
 	// wantList is the set of keys we want values for. a map for fast lookups.
 	wantList KeySet
+
+	haltChan chan struct{}
 }
 
 // NewBitSwap creates a new BitSwap instance. It does not check its parameters.
-func NewBitSwap(p *peer.Peer, net swarm.Network, d ds.Datastore) *BitSwap {
-	return &BitSwap{
+func NewBitSwap(p *peer.Peer, net swarm.Network, d ds.Datastore, r routing.IpfsRouting) *BitSwap {
+	bs := &BitSwap{
 		peer:      p,
 		net:       net,
 		datastore: d,
 		partners:  LedgerMap{},
 		wantList:  KeySet{},
+		routing:   r.(*dht.IpfsDHT),
+		meschan:   net.GetChannel(swarm.PBWrapper_BITSWAP),
+		haltChan:  make(chan struct{}),
+		listener:  swarm.NewMesListener(),
 	}
+
+	go bs.handleMessages()
+	return bs
 }
 
 // GetBlock attempts to retrieve a particular block from peers, within timeout.
-func (s *BitSwap) GetBlock(k u.Key, timeout time.Time) (
+func (bs *BitSwap) GetBlock(k u.Key, timeout time.Duration) (
 	*blocks.Block, error) {
-	return nil, errors.New("not implemented")
+	begin := time.Now()
+	tleft := timeout - time.Now().Sub(begin)
+	provs_ch := bs.routing.FindProvidersAsync(k, 20, timeout)
+
+	valchan := make(chan []byte)
+	after := time.After(tleft)
+
+	// TODO: when the data is received, shut down this for loop
+	go func() {
+		for p := range provs_ch {
+			go func(pr *peer.Peer) {
+				ledger := bs.GetLedger(pr.Key())
+				blk, err := bs.getBlock(k, pr, tleft)
+				if err != nil {
+					u.PErr("getBlock returned: %v\n", err)
+					return
+				}
+				// NOTE: this credits everyone who sends us a block,
+				//       even if we dont use it
+				ledger.ReceivedBytes(uint64(len(blk)))
+				select {
+				case valchan <- blk:
+				default:
+				}
+			}(p)
+		}
+	}()
+
+	select {
+	case blkdata := <-valchan:
+		close(valchan)
+		return blocks.NewBlock(blkdata)
+	case <-after:
+		return nil, u.ErrTimeout
+	}
+}
+
+func (bs *BitSwap) getBlock(k u.Key, p *peer.Peer, timeout time.Duration) ([]byte, error) {
+	u.DOut("[%s] getBlock '%s' from [%s]\n", bs.peer.ID.Pretty(), k.Pretty(), p.ID.Pretty())
+	//
+	mes := new(PBMessage)
+	mes.Id = proto.Uint64(swarm.GenerateMessageID())
+	mes.Key = proto.String(string(k))
+	typ := PBMessage_GET_BLOCK
+	mes.Type = &typ
+	//
+
+	after := time.After(timeout)
+	resp := bs.listener.Listen(mes.GetId(), 1, timeout)
+	smes := swarm.NewMessage(p, mes)
+	bs.meschan.Outgoing <- smes
+
+	select {
+	case resp_mes := <-resp:
+		pmes := new(PBMessage)
+		err := proto.Unmarshal(resp_mes.Data, pmes)
+		if err != nil {
+			return nil, err
+		}
+		if pmes.GetSuccess() {
+			return pmes.GetValue(), nil
+		}
+		return nil, u.ErrNotFound
+	case <-after:
+		u.PErr("getBlock for '%s' timed out.\n", k)
+		return nil, u.ErrTimeout
+	}
 }
 
 // HaveBlock announces the existance of a block to BitSwap, potentially sending
 // it to peers (Partners) whose WantLists include it.
-func (s *BitSwap) HaveBlock(k u.Key) (*blocks.Block, error) {
-	return nil, errors.New("not implemented")
+func (bs *BitSwap) HaveBlock(k u.Key) error {
+	return bs.routing.Provide(k)
+}
+
+func (bs *BitSwap) handleMessages() {
+	for {
+		select {
+		case mes := <-bs.meschan.Incoming:
+			pmes := new(PBMessage)
+			err := proto.Unmarshal(mes.Data, pmes)
+			if err != nil {
+				u.PErr("%v\n", err)
+				continue
+			}
+			if pmes.GetResponse() {
+				bs.listener.Respond(pmes.GetId(), mes)
+				continue
+			}
+
+			switch pmes.GetType() {
+			case PBMessage_GET_BLOCK:
+				go bs.handleGetBlock(mes.Peer, pmes)
+			default:
+				u.PErr("Invalid message type.\n")
+			}
+		case <-bs.haltChan:
+			return
+		}
+	}
+}
+
+func (bs *BitSwap) handleGetBlock(p *peer.Peer, pmes *PBMessage) {
+	u.DOut("handleGetBlock.\n")
+	ledger := bs.GetLedger(p.Key())
+
+	u.DOut("finding [%s] in datastore.\n", u.Key(pmes.GetKey()).Pretty())
+	idata, err := bs.datastore.Get(ds.NewKey(pmes.GetKey()))
+	if err != nil {
+		u.PErr("handleGetBlock datastore returned: %v\n", err)
+		if err == ds.ErrNotFound {
+			return
+		}
+		return
+	}
+
+	u.DOut("found value!\n")
+	data, ok := idata.([]byte)
+	if !ok {
+		u.PErr("Failed casting data from datastore.")
+		return
+	}
+
+	if ledger.ShouldSend() {
+		u.DOut("Sending value back!\n")
+		resp := &Message{
+			Value:    data,
+			Response: true,
+			ID:       pmes.GetId(),
+			Type:     PBMessage_GET_BLOCK,
+			Success:  true,
+		}
+		bs.meschan.Outgoing <- swarm.NewMessage(p, resp.ToProtobuf())
+		ledger.SentBytes(uint64(len(data)))
+	} else {
+		u.DOut("Ledger decided not to send anything...\n")
+	}
+}
+
+func (bs *BitSwap) GetLedger(k u.Key) *Ledger {
+	l, ok := bs.partners[k]
+	if ok {
+		return l
+	}
+
+	l = new(Ledger)
+	l.Strategy = StandardStrategy
+	l.Partner = peer.ID(k)
+	bs.partners[k] = l
+	return l
+}
+
+func (bs *BitSwap) Halt() {
+	bs.haltChan <- struct{}{}
 }

bitswap/ledger.go

@@ -13,11 +13,8 @@ type Ledger struct {
 	// Partner is the ID of the remote Peer.
 	Partner peer.ID
 
-	// BytesSent counts the number of bytes the local peer sent to Partner
-	BytesSent uint64
-
-	// BytesReceived counts the number of bytes local peer received from Partner
-	BytesReceived uint64
+	// Accounting tracks bytes sent and recieved.
+	Accounting debtRatio
 
 	// FirstExchnage is the time of the first data exchange.
 	FirstExchange *time.Time
@@ -27,7 +24,21 @@ type Ledger struct {
 
 	// WantList is a (bounded, small) set of keys that Partner desires.
 	WantList KeySet
+
+	Strategy StrategyFunc
 }
 
 // LedgerMap lists Ledgers by their Partner key.
 type LedgerMap map[u.Key]*Ledger
+
+func (l *Ledger) ShouldSend() bool {
+	return l.Strategy(l.Accounting)
+}
+
+func (l *Ledger) SentBytes(n uint64) {
+	l.Accounting.BytesSent += n
+}
+
+func (l *Ledger) ReceivedBytes(n uint64) {
+	l.Accounting.BytesRecv += n
+}

bitswap/message.go

@@ -0,0 +1,32 @@
+package bitswap
+
+import (
+	"code.google.com/p/goprotobuf/proto"
+	u "github.com/jbenet/go-ipfs/util"
+)
+
+type Message struct {
+	Type     PBMessage_MessageType
+	ID       uint64
+	Response bool
+	Key      u.Key
+	Value    []byte
+	Success  bool
+}
+
+func (m *Message) ToProtobuf() *PBMessage {
+	pmes := new(PBMessage)
+	pmes.Id = &m.ID
+	pmes.Type = &m.Type
+	if m.Response {
+		pmes.Response = proto.Bool(true)
+	}
+
+	if m.Success {
+		pmes.Success = proto.Bool(true)
+	}
+
+	pmes.Key = proto.String(string(m.Key))
+	pmes.Value = m.Value
+	return pmes
+}