path: root/swarm/network/forwarding.go

// Copyright 2016 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 network

import (
    "fmt"
    "math/rand"
    "time"

    "github.com/ethereum/go-ethereum/log"
    "github.com/ethereum/go-ethereum/swarm/storage"
)

const requesterCount = 3

/*
forwarder implements the CloudStore interface (use by storage.NetStore)
and serves as the cloud store backend orchestrating storage/retrieval/delivery
via the native bzz protocol
which uses an MSB logarithmic distance-based semi-permanent Kademlia table for
* recursive forwarding style routing for retrieval
* smart syncronisation
*/

type forwarder struct {
    hive *Hive
}

func NewForwarder(hive *Hive) *forwarder {
    return &forwarder{hive: hive}
}

// generate a unique id uint64
func generateId() uint64 {
    r := rand.New(rand.NewSource(time.Now().UnixNano()))
    return uint64(r.Int63())
}

var searchTimeout = 3 * time.Second

// forwarding logic
// logic propagating retrieve requests to peers given by the kademlia hive
func (self *forwarder) Retrieve(chunk *storage.Chunk) {
    peers := self.hive.getPeers(chunk.Key, 0)
    log.Trace(fmt.Sprintf("forwarder.Retrieve: %v - received %d peers from KΛÐΞMLIΛ...", chunk.Key.Log(), len(peers)))
OUT:
    for _, p := range peers {
        log.Trace(fmt.Sprintf("forwarder.Retrieve: sending retrieveRequest %v to peer [%v]", chunk.Key.Log(), p))
        for _, recipients := range chunk.Req.Requesters {
            for _, recipient := range recipients {
                req := recipient.(*retrieveRequestMsgData)
                if req.from.Addr() == p.Addr() {
                    continue OUT
                }
            }
        }
        req := &retrieveRequestMsgData{
            Key: chunk.Key,
            Id:  generateId(),
        }
        var err error
        if p.swap != nil {
            err = p.swap.Add(-1)
        }
        if err == nil {
            p.retrieve(req)
            break OUT
        }
        log.Warn(fmt.Sprintf("forwarder.Retrieve: unable to send retrieveRequest to peer [%v]: %v", chunk.Key.Log(), err))
    }
}

// requests to specific peers given by the kademlia hive
// except for peers that the store request came from (if any)
// delivery queueing taken care of by syncer
func (self *forwarder) Store(chunk *storage.Chunk) {
    var n int
    msg := &storeRequestMsgData{
        Key:   chunk.Key,
        SData: chunk.SData,
    }
    var source *peer
    if chunk.Source != nil {
        source = chunk.Source.(*peer)
    }
    for _, p := range self.hive.getPeers(chunk.Key, 0) {
        log.Trace(fmt.Sprintf("forwarder.Store: %v %v", p, chunk))

        if p.syncer != nil && (source == nil || p.Addr() != source.Addr()) {
            n++
            Deliver(p, msg, PropagateReq)
        }
    }
    log.Trace(fmt.Sprintf("forwarder.Store: sent to %v peers (chunk = %v)", n, chunk))
}

// once a chunk is found deliver it to its requesters unless timed out
func (self *forwarder) Deliver(chunk *storage.Chunk) {
    // iterate over request entries
    for id, requesters := range chunk.Req.Requesters {
        counter := requesterCount
        msg := &storeRequestMsgData{
            Key:   chunk.Key,
            SData: chunk.SData,
        }
        var n int
        var req *retrieveRequestMsgData
        // iterate over requesters with the same id
        for id, r := range requesters {
            req = r.(*retrieveRequestMsgData)
            if req.timeout == nil || req.timeout.After(time.Now()) {
                log.Trace(fmt.Sprintf("forwarder.Deliver: %v -> %v", req.Id, req.from))
                msg.Id = uint64(id)
                Deliver(req.from, msg, DeliverReq)
                n++
                counter--
                if counter <= 0 {
                    break
                }
            }
        }
        log.Trace(fmt.Sprintf("forwarder.Deliver: submit chunk %v (request id %v) for delivery to %v peers", chunk.Key.Log(), id, n))
    }
}

// initiate delivery of a chunk to a particular peer via syncer#addRequest
// depending on syncer mode and priority settings and sync request type
// this either goes via confirmation roundtrip or queued or pushed directly
func Deliver(p *peer, req interface{}, ty int) {
    p.syncer.addRequest(req, ty)
}

// push chunk over to peer
func Push(p *peer, key storage.Key, priority uint) {
    p.syncer.doDelivery(key, priority, p.syncer.quit)
}