From 94bf8de5474359fe4f294475864712262fbe02c7 Mon Sep 17 00:00:00 2001 From: Zsolt Felfoldi Date: Wed, 27 Nov 2019 18:47:17 +0100 Subject: [PATCH] les: lespay API --- les/api.go | 24 +++++++++++++++++++----- les/client_handler.go | 13 +++++++++++++ les/clientpool.go | 26 ++++++++++++++------------ les/server_handler.go | 20 ++++++++++++++++---- les/tokensale.go | 8 ++++++-- p2p/discv5/net.go | 20 +++++++++++++++++--- p2p/enode/node.go | 6 +++--- 7 files changed, 88 insertions(+), 29 deletions(-) diff --git a/les/api.go b/les/api.go index f17f8ed943..43cafacb11 100644 --- a/les/api.go +++ b/les/api.go @@ -20,6 +20,7 @@ import ( "context" "errors" "fmt" + "reflect" "time" "github.com/ethereum/go-ethereum/common/hexutil" @@ -376,10 +377,13 @@ func (api *PrivateLespayAPI) makeCall(ctx context.Context, remote bool, nodeStr freeID string peer *peer node *enode.Node + err error ) if nodeStr != "" { - if peer = api.peerSet.Peer(nodeStr); peer != nil { - id = peer.ID() + if id, err = enode.ParseID(nodeStr); err == nil { + if peer = api.peerSet.Peer(peerIdToString(id)); peer == nil { + return nil, errors.New("peer not connected") + } freeID = peer.freeClientId() } else { var err error @@ -416,18 +420,25 @@ func (api *PrivateLespayAPI) makeCall(ctx context.Context, remote bool, nodeStr return nil, errors.New("UDP DHT not available") } cancelFn = api.dht.SendTalkRequest(node, "lespay", [][]byte{cmd}, func(payload interface{}) bool { - if replies, ok := payload.([][]byte); ok && len(replies) == 1 { - reply = replies[0] + fmt.Println("dht delivered", payload, reflect.TypeOf(payload)) + if replies, ok := payload.([]interface{}); ok && len(replies) == 1 { + reply, ok = replies[0].([]byte) } close(delivered) return reply != nil }) } select { + case <-time.After(time.Second * 5): + cancelFn() + return nil, errors.New("timeout") case <-ctx.Done(): cancelFn() return nil, ctx.Err() case <-delivered: + if len(reply) == 0 { + return nil, errors.New("unknown command") + } return reply, nil } } else { @@ -444,10 +455,13 @@ func (api *PrivateLespayAPI) Connection(ctx context.Context, remote bool, node s params := tsConnectionParams{requestedCapacity, stayConnected, paymentModule, setCap} enc, _ := rlp.EncodeToBytes(¶ms) var resEnc []byte - resEnc, err = api.makeCall(ctx, remote, node, enc) + fmt.Println("makeCall", remote, node, enc) + resEnc, err = api.makeCall(ctx, remote, node, append([]byte{tsConnection}, enc...)) if err != nil { + fmt.Println("makeCall err", err) return } err = rlp.DecodeBytes(resEnc, &results) + fmt.Println("decode err", err) return } diff --git a/les/client_handler.go b/les/client_handler.go index 20d19363c2..37f5e02ae5 100644 --- a/les/client_handler.go +++ b/les/client_handler.go @@ -17,6 +17,7 @@ package les import ( + "fmt" "math/big" "sync" "time" @@ -316,19 +317,24 @@ func (h *clientHandler) handleMsg(p *peer) error { p.freezeServer(false) p.Log().Debug("Service resumed") case LespayReplyMsg: + fmt.Println("LespayReply received") p.Log().Trace("Received tx status response") var resp struct { ReqID uint64 Replies [][]byte } if err := msg.Decode(&resp); err != nil { + fmt.Println("LespayReply decode err", err) return errResp(ErrDecode, "msg %v: %v", msg, err) } + fmt.Println("LespayReply decoded", resp) h.lespayReplyLock.Lock() if handler := h.lespayReplyHandlers[resp.ReqID]; handler != nil { + fmt.Println("handler found") delete(h.lespayReplyHandlers, resp.ReqID) responseError = !handler(resp.Replies) } else { + fmt.Println("handler not found") responseError = true } h.lespayReplyLock.Unlock() @@ -354,9 +360,16 @@ func (h *clientHandler) handleMsg(p *peer) error { func (h *clientHandler) makeLespayCall(p *peer, cmds [][]byte, handler func([][]byte) bool) func() bool { reqID := genReqID() + h.lespayReplyLock.Lock() + h.lespayReplyHandlers[reqID] = handler + h.lespayReplyLock.Unlock() if p.SendLespay(reqID, cmds) != nil { + h.lespayReplyLock.Lock() + delete(h.lespayReplyHandlers, reqID) + h.lespayReplyLock.Unlock() return nil } + fmt.Println("Lespay sent") return func() bool { h.lespayReplyLock.Lock() cancel := h.lespayReplyHandlers[reqID] != nil diff --git a/les/clientpool.go b/les/clientpool.go index 390ba36315..f132e0298c 100644 --- a/les/clientpool.go +++ b/les/clientpool.go @@ -594,19 +594,21 @@ func (f *clientPool) setCapacity(id enode.ID, freeID string, capacity uint64, mi if missing != 0 { return missing, capacity, errNoPriority } - if setCap && c == nil { - return missing, capacity, fmt.Errorf("client %064x is not connected", c.id[:]) - } // capacity update is possible - f.connectedCap += capacity - c.capacity - f.updateFullRatio() - f.priorityConnected += capacity - c.capacity - c.capacity = capacity - c.balanceTracker.setCapacity(capacity) - f.connectedQueue.Update(c.queueIndex) - totalConnectedGauge.Update(int64(f.connectedCap)) - updatePriceFactors(&c.balanceTracker, c.posFactors, c.negFactors, c.capacity) - c.peer.updateCapacity(c.capacity) + if setCap { + if c == nil { + return 0, capacity, fmt.Errorf("client %064x is not connected", c.id[:]) + } + f.connectedCap += capacity - c.capacity + f.updateFullRatio() + f.priorityConnected += capacity - c.capacity + c.capacity = capacity + c.balanceTracker.setCapacity(capacity) + f.connectedQueue.Update(c.queueIndex) + totalConnectedGauge.Update(int64(f.connectedCap)) + updatePriceFactors(&c.balanceTracker, c.posFactors, c.negFactors, c.capacity) + c.peer.updateCapacity(c.capacity) + } return 0, capacity, nil } diff --git a/les/server_handler.go b/les/server_handler.go index 954cbae2d9..106321f54d 100644 --- a/les/server_handler.go +++ b/les/server_handler.go @@ -20,7 +20,9 @@ import ( "encoding/binary" "encoding/json" "errors" + "fmt" "net" + "reflect" "sync" "sync/atomic" "time" @@ -130,7 +132,7 @@ func (h *serverHandler) handle(p *peer) error { } // Reject light clients if server is not synced. if !h.synced() { - return p2p.DiscRequested + //return p2p.DiscRequested } defer p.fcClient.Disconnect() @@ -982,11 +984,21 @@ func (h *serverHandler) broadcastHeaders() { } func (h *serverHandler) talkRequestHandler(id enode.ID, addr *net.UDPAddr, payload interface{}) (interface{}, bool) { - cmds, ok := payload.([][]byte) + fmt.Println("talkRequestHandler", id, addr, payload, reflect.TypeOf(payload)) + c, ok := payload.([]interface{}) if !ok { return nil, false } + cmds := make([][]byte, len(c)) + for i, c := range c { + cmds[i], ok = c.([]byte) + if !ok { + fmt.Println("type err", reflect.TypeOf(c)) + return nil, false + } + } + fmt.Println("ok", ok) results := h.server.tokenSale.runCommands(cmds, id, addr.IP.String()) - res, _ := rlp.EncodeToBytes(&results) - return res, true + fmt.Println("results", results) + return results, true } diff --git a/les/tokensale.go b/les/tokensale.go index 9d785fe41e..07b9be2894 100644 --- a/les/tokensale.go +++ b/les/tokensale.go @@ -367,7 +367,9 @@ func (t *tokenSale) runCommand(cmd []byte, id enode.ID, freeID string) []byte { ) if err := rlp.DecodeBytes(cmd[1:], ¶ms); err == nil { results.PcValue, results.PcBalance, err = t.deposit(id, params.PaymentModule, params.ProofOfPayment) - results.Err = err.Error() + if err != nil { + results.Err = err.Error() + } res, _ = rlp.EncodeToBytes(&results) } case tsBuyTokens: @@ -388,7 +390,9 @@ func (t *tokenSale) runCommand(cmd []byte, id enode.ID, freeID string) []byte { if err := rlp.DecodeBytes(cmd[1:], ¶ms); err == nil { results.AvailableCapacity, results.TokenBalance, results.TokensMissing, results.PcBalance, results.PcMissing, results.PaymentRequired, err = t.connection(id, freeID, params.RequestedCapacity, time.Duration(params.StayConnected)*time.Second, params.PaymentModule, params.SetCap) - results.Err = err.Error() + if err != nil { + results.Err = err.Error() + } res, _ = rlp.EncodeToBytes(&results) } } diff --git a/p2p/discv5/net.go b/p2p/discv5/net.go index 370a7237fc..0e61e71094 100644 --- a/p2p/discv5/net.go +++ b/p2p/discv5/net.go @@ -432,9 +432,11 @@ loop: // Ingress packet handling. case pkt := <-net.read: - //fmt.Println("read", pkt.ev) log.Trace("<-net.read") n := net.internNode(&pkt) + if pkt.ev == talkRequestPacket { + fmt.Println("read trp", n.state, pkt) + } prestate := n.state status := "ok" if err := net.handle(n, pkt.ev, &pkt); err != nil { @@ -722,11 +724,17 @@ func (net *Network) refresh(done chan<- struct{}) { func (net *Network) internNode(pkt *ingressPacket) *Node { if n := net.nodes[pkt.remoteID]; n != nil { + if pkt.ev == talkRequestPacket { + fmt.Println("node exists") + } n.IP = pkt.remoteAddr.IP n.UDP = uint16(pkt.remoteAddr.Port) n.TCP = uint16(pkt.remoteAddr.Port) return n } + if pkt.ev == talkRequestPacket { + fmt.Println("node created") + } n := NewNode(pkt.remoteID, pkt.remoteAddr.IP, uint16(pkt.remoteAddr.Port), uint16(pkt.remoteAddr.Port)) n.state = unknown net.nodes[pkt.remoteID] = n @@ -1061,7 +1069,7 @@ func (net *Network) handle(n *Node, ev nodeEvent, pkt *ingressPacket) error { //fmt.Println("handle", n.addr().String(), n.state, ev) if pkt != nil { if err := net.checkPacket(n, ev, pkt); err != nil { - //fmt.Println("check err:", err) + //fmt.Println("check err:", err, pkt) return err } // Start the background expiration goroutine after the first @@ -1076,6 +1084,7 @@ func (net *Network) handle(n *Node, ev nodeEvent, pkt *ingressPacket) error { if n.state == nil { n.state = unknown //??? } + //fmt.Println("old state:", n.state) next, err := n.state.handle(net, n, ev, pkt) net.transition(n, next) //fmt.Println("new state:", n.state) @@ -1241,8 +1250,10 @@ func (net *Network) handleQueryEvent(n *Node, ev nodeEvent, pkt *ingressPacket) net.talkRequestSubLock.RLock() subFn := net.talkRequestSubs[string(p.TalkID)] net.talkRequestSubLock.RUnlock() + fmt.Println("trp", p) if subFn != nil { resp, ok := subFn(enode.ID(n.sha), n.addr(), p.Payload) + fmt.Println("subFn", ok) if ok { net.conn.send(n, talkResponsePacket, talkResponse{ReplyTok: pkt.hash, Payload: resp}) } else { @@ -1367,7 +1378,7 @@ func (q *talkQuery) start(net *Network) bool { q.key = string(q.remote.sha[:]) + string(hash[:]) net.talkResponseSubs[q.key] = q.handler net.talkResponseSubLock.Unlock() - fmt.Println("sent") + fmt.Println("sent", q.remote.state) return true } // If the node is not known yet, it won't accept queries. @@ -1397,6 +1408,9 @@ func (net *Network) SendTalkRequest(to *enode.Node, talkID string, payload inter net.nodes[nodeID] = node } q := &talkQuery{remote: node, talkID: talkID, payload: payload, handler: handler} + if node.state == nil || !node.state.canQuery { + net.ping(node, node.addr()) + } select { case net.queryReq <- q: case <-net.closed: diff --git a/p2p/enode/node.go b/p2p/enode/node.go index 9eb2544ffe..3f6cda6d4a 100644 --- a/p2p/enode/node.go +++ b/p2p/enode/node.go @@ -217,7 +217,7 @@ func (n ID) MarshalText() ([]byte, error) { // UnmarshalText implements the encoding.TextUnmarshaler interface. func (n *ID) UnmarshalText(text []byte) error { - id, err := parseID(string(text)) + id, err := ParseID(string(text)) if err != nil { return err } @@ -229,14 +229,14 @@ func (n *ID) UnmarshalText(text []byte) error { // The string may be prefixed with 0x. // It panics if the string is not a valid ID. func HexID(in string) ID { - id, err := parseID(in) + id, err := ParseID(in) if err != nil { panic(err) } return id } -func parseID(in string) (ID, error) { +func ParseID(in string) (ID, error) { var id ID b, err := hex.DecodeString(strings.TrimPrefix(in, "0x")) if err != nil {