swarm/network/stream: leveldb intervals store

swarm/network/stream/common_test.go

@@ -88,18 +88,22 @@ func newStreamerTester(t *testing.T) (*p2ptest.ProtocolTester, *Registry, *stora
 	if err != nil {
 		return nil, nil, nil, func() {}, err
 	}
-	teardown := func() {
+	removeDataDir := func() {
 		os.RemoveAll(datadir)
 	}
 
 	localStore, err := storage.NewTestLocalStoreForAddr(datadir, addr.Over())
 	if err != nil {
-		return nil, nil, nil, teardown, err
+		return nil, nil, nil, removeDataDir, err
 	}
 
 	db := storage.NewDBAPI(localStore)
 	delivery := NewDelivery(to, db)
 	streamer := NewRegistry(addr, delivery, localStore, intervals.NewMemStore(), defaultSkipCheck)
+	teardown := func() {
+		streamer.Close()
+		removeDataDir()
+	}
 	protocolTester := p2ptest.NewProtocolTester(t, network.NewNodeIDFromAddr(addr), 1, streamer.runProtocol)
 
 	err = waitForPeers(streamer, 1*time.Second, 1)

swarm/network/stream/intervals/dbstore.go

@@ -0,0 +1,78 @@
+// Copyright 2018 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
+
+package intervals
+
+import (
+	"github.com/syndtr/goleveldb/leveldb"
+)
+
+// DBStore uses LevelDB to store intervals.
+type DBStore struct {
+	db *leveldb.DB
+}
+
+// NewDBStore creates a new instance of DBStore.
+func NewDBStore(path string) (s *DBStore, err error) {
+	db, err := leveldb.OpenFile(path, nil)
+	if err != nil {
+		return nil, err
+	}
+	return &DBStore{
+		db: db,
+	}, nil
+}
+
+// Get retrieves Intervals for a specific key. If there is no Intervals
+// ErrNotFound is returned.
+func (s *DBStore) Get(key string) (i *Intervals, err error) {
+	k := []byte(key)
+	has, err := s.db.Has(k, nil)
+	if err != nil {
+		return nil, ErrNotFound
+	}
+	if !has {
+		return nil, ErrNotFound
+	}
+	data, err := s.db.Get(k, nil)
+	if err == leveldb.ErrNotFound {
+		err = ErrNotFound
+	}
+	i = &Intervals{}
+	if err = i.UnmarshalBinary(data); err != nil {
+		return nil, err
+	}
+	return i, err
+}
+
+// Put stores Intervals for a specific key.
+func (s *DBStore) Put(key string, i *Intervals) (err error) {
+	data, err := i.MarshalBinary()
+	if err != nil {
+		return err
+	}
+	return s.db.Put([]byte(key), data, nil)
+}
+
+// Delete removes Intervals stored under a specific key.
+func (s *DBStore) Delete(key string) (err error) {
+	return s.db.Delete([]byte(key), nil)
+}
+
+// Close releases the resources used by the underlying LevelDB.
+func (s *DBStore) Close() error {
+	return s.db.Close()
+}

swarm/network/stream/intervals/dbstore_test.go

@@ -0,0 +1,40 @@
+// Copyright 2018 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
+
+package intervals
+
+import (
+	"io/ioutil"
+	"os"
+	"testing"
+)
+
+// TestDBStore tests basic functionality of DBStore.
+func TestDBStore(t *testing.T) {
+	dir, err := ioutil.TempDir("", "intervals_test_db_store")
+	if err != nil {
+		panic(err)
+	}
+	defer os.RemoveAll(dir)
+
+	store, err := NewDBStore(dir)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer store.Close()
+
+	testStore(t, store)
+}

swarm/network/stream/intervals/intervals.go

@@ -17,7 +17,9 @@
 package intervals
 
 import (
+	"bytes"
 	"fmt"
+	"strconv"
 	"sync"
 )
 
@@ -152,3 +154,53 @@ func (i *Intervals) Last() (end uint64) {
 func (i *Intervals) String() string {
 	return fmt.Sprint(i.ranges)
 }
+
+// MarshalBinary encodes Intervals parameters into a semicolon separated list.
+// The first element in the list is base36-encoded start value. The following
+// elements are two base36-encoded value ranges separated by comma.
+func (i *Intervals) MarshalBinary() (data []byte, err error) {
+	d := make([][]byte, len(i.ranges)+1)
+	d[0] = []byte(strconv.FormatUint(i.start, 36))
+	for j := range i.ranges {
+		r := i.ranges[j]
+		d[j+1] = []byte(strconv.FormatUint(r[0], 36) + "," + strconv.FormatUint(r[1], 36))
+	}
+	return bytes.Join(d, []byte(";")), nil
+}
+
+// UnmarshalBinary decodes data according to the Intervals.MarshalBinary format.
+func (i *Intervals) UnmarshalBinary(data []byte) (err error) {
+	d := bytes.Split(data, []byte(";"))
+	l := len(d)
+	if l == 0 {
+		return nil
+	}
+	if l >= 1 {
+		i.start, err = strconv.ParseUint(string(d[0]), 36, 64)
+		if err != nil {
+			return err
+		}
+	}
+	if l == 1 {
+		return nil
+	}
+
+	i.ranges = make([][2]uint64, 0, l-1)
+	for j := 1; j < l; j++ {
+		r := bytes.SplitN(d[j], []byte(","), 2)
+		if len(r) < 2 {
+			return fmt.Errorf("range %d has less then 2 elements", j)
+		}
+		start, err := strconv.ParseUint(string(r[0]), 36, 64)
+		if err != nil {
+			return fmt.Errorf("parsing the first element in range %d: %v", j, err)
+		}
+		end, err := strconv.ParseUint(string(r[1]), 36, 64)
+		if err != nil {
+			return fmt.Errorf("parsing the second element in range %d: %v", j, err)
+		}
+		i.ranges = append(i.ranges, [2]uint64{start, end})
+	}
+
+	return nil
+}

swarm/network/stream/intervals/store.go

@@ -14,7 +14,6 @@
 // You should have received a copy of the GNU Lesser General Public License
 // along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
 
-// Package intervals TODO: implement LevelDB based Store.
 package intervals
 
 import (
@@ -33,6 +32,7 @@ type Store interface {
 	Get(key string) (i *Intervals, err error)
 	Put(key string, i *Intervals) (err error)
 	Delete(key string) (err error)
+	Close() error
 }
 
 // MemStore is the reference implementation of Store interface that is supposed
@@ -82,3 +82,8 @@ func (s *MemStore) Delete(key string) (err error) {
 	delete(s.db, key)
 	return nil
 }
+
+// Close doesnot do anything.
+func (s *MemStore) Close() error {
+	return nil
+}

swarm/network/stream/intervals/store_test.go

@@ -20,8 +20,11 @@ import "testing"
 
 // TestMemStore tests basic functionality of MemStore.
 func TestMemStore(t *testing.T) {
-	s := NewMemStore()
+	testStore(t, NewMemStore())
+}
 
+// testStore is a helper function to test various Store implementations.
+func testStore(t *testing.T, s Store) {
 	key1 := "key1"
 	i1 := NewIntervals(0)
 	i1.Add(10, 20)

swarm/network/stream/peer.go

@@ -240,6 +240,7 @@ func (p *Peer) setClientNolock(s Stream, from, to uint64) (c *client, err error)
 		Client:         is,
 		stream:         s,
 		priority:       cp.priority,
+		to:             to,
 		next:           next,
 		intervalsStore: p.streamer.intervalsStore,
 		intervalsKey:   intervalsKey,

swarm/network/stream/stream.go

@@ -190,6 +190,11 @@ func (r *Registry) PeerInfo(id discover.NodeID) interface{} {
 	return nil
 }
 
+func (r *Registry) Close() error {
+	r.store.Close()
+	return r.intervalsStore.Close()
+}
+
 func (r *Registry) getPeer(peerId discover.NodeID) *Peer {
 	r.peersMu.RLock()
 	defer r.peersMu.RUnlock()
@@ -286,6 +291,7 @@ type client struct {
 	stream    Stream
 	priority  uint8
 	sessionAt uint64
+	to        uint64
 	next      chan error
 
 	intervalsKey   string
@@ -348,7 +354,13 @@ func (c *client) batchDone(p *Peer, req *OfferedHashesMsg, hashes []byte) error
 		if err := c.AddInterval(tp.Takeover.Start, tp.Takeover.End); err != nil {
 			return err
 		}
-		return p.SendPriority(tp, c.priority)
+		if err := p.SendPriority(tp, c.priority); err != nil {
+			return err
+		}
+		if c.to > 0 && tp.Takeover.End >= c.to {
+			return p.streamer.Unsubscribe(p.Peer.ID(), req.Stream)
+		}
+		return nil
 	}
 	return nil
 }