swarm, p2p/protocols: Stream accounting (#18337)

* swarm: completed 1st phase of swap accounting

* swarm, p2p/protocols: added stream pricing

* swarm/network/stream: gofmt simplify stream.go

* swarm: fixed review comments

* swarm: used snapshots for swap tests

* swarm: custom retrieve for swap (less cascaded requests at any one time)

* swarm: addressed PR comments

* swarm: log output formatting

* swarm: removed parallelism in swap tests

* swarm: swap tests simplification

* swarm: removed swap_test.go

* swarm/network/stream: added prefix space for comments

* swarm/network/stream: unit test for prices

* swarm/network/stream: don't hardcode price

* swarm/network/stream: fixed invalid price check

p2p/protocols/accounting.go

@@ -42,6 +42,8 @@ var (
 	mPeerDrops metrics.Counter
 	// how many times local node overdrafted and dropped
 	mSelfDrops metrics.Counter
+
+	MetricsRegistry metrics.Registry
 )
 
 // Prices defines how prices are being passed on to the accounting instance
@@ -59,7 +61,7 @@ const (
 
 // Price represents the costs of a message
 type Price struct {
-	Value   uint64 //
+	Value   uint64
 	PerByte bool // True if the price is per byte or for unit
 	Payer   Payer
 }
@@ -114,21 +116,20 @@ func NewAccounting(balance Balance, po Prices) *Accounting {
 // at the passed interval writes the metrics to a LevelDB
 func SetupAccountingMetrics(reportInterval time.Duration, path string) *AccountingMetrics {
 	// create an empty registry
-	registry := metrics.NewRegistry()
+	MetricsRegistry = metrics.NewRegistry()
 	// instantiate the metrics
-	mBalanceCredit = metrics.NewRegisteredCounterForced("account.balance.credit", registry)
+	mBalanceCredit = metrics.NewRegisteredCounterForced("account.balance.credit", MetricsRegistry)
-	mBalanceDebit = metrics.NewRegisteredCounterForced("account.balance.debit", registry)
+	mBalanceDebit = metrics.NewRegisteredCounterForced("account.balance.debit", MetricsRegistry)
-	mBytesCredit = metrics.NewRegisteredCounterForced("account.bytes.credit", registry)
+	mBytesCredit = metrics.NewRegisteredCounterForced("account.bytes.credit", MetricsRegistry)
-	mBytesDebit = metrics.NewRegisteredCounterForced("account.bytes.debit", registry)
+	mBytesDebit = metrics.NewRegisteredCounterForced("account.bytes.debit", MetricsRegistry)
-	mMsgCredit = metrics.NewRegisteredCounterForced("account.msg.credit", registry)
+	mMsgCredit = metrics.NewRegisteredCounterForced("account.msg.credit", MetricsRegistry)
-	mMsgDebit = metrics.NewRegisteredCounterForced("account.msg.debit", registry)
+	mMsgDebit = metrics.NewRegisteredCounterForced("account.msg.debit", MetricsRegistry)
-	mPeerDrops = metrics.NewRegisteredCounterForced("account.peerdrops", registry)
+	mPeerDrops = metrics.NewRegisteredCounterForced("account.peerdrops", MetricsRegistry)
-	mSelfDrops = metrics.NewRegisteredCounterForced("account.selfdrops", registry)
+	mSelfDrops = metrics.NewRegisteredCounterForced("account.selfdrops", MetricsRegistry)
 	// create the DB and start persisting
-	return NewAccountingMetrics(registry, reportInterval, path)
+	return NewAccountingMetrics(MetricsRegistry, reportInterval, path)
 }
 
-//Implement Hook.Send
 // Send takes a peer, a size and a msg and
 //   - calculates the cost for the local node sending a msg of size to peer using the Prices interface
 //   - credits/debits local node using balance interface
@@ -148,7 +149,6 @@ func (ah *Accounting) Send(peer *Peer, size uint32, msg interface{}) error {
 	return err
 }
 
-//Implement Hook.Receive
 // Receive takes a peer, a size and a msg and
 //   - calculates the cost for the local node receiving a msg of size from peer using the Prices interface
 //   - credits/debits local node using balance interface

swarm/network/stream/stream.go

@@ -88,9 +88,9 @@ type Registry struct {
 	intervalsStore state.Store
 	autoRetrieval  bool // automatically subscribe to retrieve request stream
 	maxPeerServers int
-	spec           *protocols.Spec   //this protocol's spec
 	balance        protocols.Balance // implements protocols.Balance, for accounting
 	prices         protocols.Prices  // implements protocols.Prices, provides prices to accounting
+	spec           *protocols.Spec   // this protocol's spec
 }
 
 // RegistryOptions holds optional values for NewRegistry constructor.
@@ -110,7 +110,7 @@ func NewRegistry(localID enode.ID, delivery *Delivery, syncChunkStore storage.Sy
 	if options.SyncUpdateDelay <= 0 {
 		options.SyncUpdateDelay = 15 * time.Second
 	}
-	//check if retriaval has been disabled
+	// check if retrieval has been disabled
 	retrieval := options.Retrieval != RetrievalDisabled
 
 	streamer := &Registry{
@@ -235,10 +235,14 @@ func NewRegistry(localID enode.ID, delivery *Delivery, syncChunkStore storage.Sy
 	return streamer
 }
 
+// This is an accounted protocol, therefore we need to provide a pricing Hook to the spec
+// For simulations to be able to run multiple nodes and not override the hook's balance,
 // we need to construct a spec instance per node instance
 func (r *Registry) setupSpec() {
 	// first create the "bare" spec
 	r.createSpec()
+	// now create the pricing object
+	r.createPriceOracle()
 	// if balance is nil, this node has been started without swap support (swapEnabled flag is false)
 	if r.balance != nil && !reflect.ValueOf(r.balance).IsNil() {
 		// swap is enabled, so setup the hook
@@ -756,6 +760,52 @@ func (r *Registry) createSpec() {
 	r.spec = spec
 }
 
+// An accountable message needs some meta information attached to it
+// in order to evaluate the correct price
+type StreamerPrices struct {
+	priceMatrix map[reflect.Type]*protocols.Price
+	registry    *Registry
+}
+
+// Price implements the accounting interface and returns the price for a specific message
+func (sp *StreamerPrices) Price(msg interface{}) *protocols.Price {
+	t := reflect.TypeOf(msg).Elem()
+	return sp.priceMatrix[t]
+}
+
+// Instead of hardcoding the price, get it
+// through a function - it could be quite complex in the future
+func (sp *StreamerPrices) getRetrieveRequestMsgPrice() uint64 {
+	return uint64(1)
+}
+
+// Instead of hardcoding the price, get it
+// through a function - it could be quite complex in the future
+func (sp *StreamerPrices) getChunkDeliveryMsgRetrievalPrice() uint64 {
+	return uint64(1)
+}
+
+// createPriceOracle sets up a matrix which can be queried to get
+// the price for a message via the Price method
+func (r *Registry) createPriceOracle() {
+	sp := &StreamerPrices{
+		registry: r,
+	}
+	sp.priceMatrix = map[reflect.Type]*protocols.Price{
+		reflect.TypeOf(ChunkDeliveryMsgRetrieval{}): {
+			Value:   sp.getChunkDeliveryMsgRetrievalPrice(), // arbitrary price for now
+			PerByte: true,
+			Payer:   protocols.Receiver,
+		},
+		reflect.TypeOf(RetrieveRequestMsg{}): {
+			Value:   sp.getRetrieveRequestMsgPrice(), // arbitrary price for now
+			PerByte: false,
+			Payer:   protocols.Sender,
+		},
+	}
+	r.prices = sp
+}
+
 func (r *Registry) Protocols() []p2p.Protocol {
 	return []p2p.Protocol{
 		{

swarm/network/stream/streamer_test.go

@@ -921,3 +921,34 @@ func TestMaxPeerServersWithoutUnsubscribe(t *testing.T) {
 		}
 	}
 }
+
+//TestHasPriceImplementation is to check that the Registry has a
+//`Price` interface implementation
+func TestHasPriceImplementation(t *testing.T) {
+	_, r, _, teardown, err := newStreamerTester(t, &RegistryOptions{
+		Retrieval: RetrievalDisabled,
+		Syncing:   SyncingDisabled,
+	})
+	defer teardown()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if r.prices == nil {
+		t.Fatal("No prices implementation available for the stream protocol")
+	}
+
+	pricesInstance, ok := r.prices.(*StreamerPrices)
+	if !ok {
+		t.Fatal("`Registry` does not have the expected Prices instance")
+	}
+	price := pricesInstance.Price(&ChunkDeliveryMsgRetrieval{})
+	if price == nil || price.Value == 0 || price.Value != pricesInstance.getChunkDeliveryMsgRetrievalPrice() {
+		t.Fatal("No prices set for chunk delivery msg")
+	}
+
+	price = pricesInstance.Price(&RetrieveRequestMsg{})
+	if price == nil || price.Value == 0 || price.Value != pricesInstance.getRetrieveRequestMsgPrice() {
+		t.Fatal("No prices set for chunk delivery msg")
+	}
+}