Peernet-Archive/core
1
0
Fork 0
mirror of https://github.com/PeernetOfficial/core.git synced 2026-07-17 02:47:51 +01:00
Code Issues Packages Projects Releases Wiki Activity

Transfer: Change virtual connection from net.PacketConn to channels which is faster and less complex.

Browse Source 
Improve termination behavior and documentation of it.
This commit is contained in:
Kleissner
2021-10-24 06:27:23 +02:00
parent 153ca78ed7
commit f6bbd02a01
7 changed files with 78 additions and 117 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
Commands.go
View File

@@ -266,14 +266,14 @@ func (peer *PeerInfo) cmdTransfer(msg *protocol.MessageTransfer, connection *Con
case protocol.TransferControlNotAvailable:
if v, ok := msg.SequenceInfo.Data.(*virtualPacketConn); ok {
v.Terminate(false, 404)
v.Terminate(404)
return
}
case protocol.TransferControlTerminate:
if v, ok := msg.SequenceInfo.Data.(*virtualPacketConn); ok {
// Since an incoming terminate notice means the remote peer already terminated the connection, set sendNotice to false.
v.Terminate(false, 2)
v.Terminate(2)
return
}

17
Transfer UDT.go
View File

@@ -32,7 +32,7 @@ func (peer *PeerInfo) startFileTransferUDT(hash []byte, offset, limit uint64, se
networks.Sequences.RegisterSequenceBi(peer.PublicKey, sequenceNumber, virtualConnection, transferSequenceTimeout, virtualConnection.sequenceTerminate)
// start UDT sender
udtConn, err := udt.DialUDT(udt.DefaultConfig(), virtualConnection, true)
udtConn, err := udt.DialUDT(udt.DefaultConfig(), virtualConnection, virtualConnection.incomingData, virtualConnection.outgoingData, virtualConnection.terminateChan, true)
if err != nil {
return err
}
@@ -40,25 +40,24 @@ func (peer *PeerInfo) startFileTransferUDT(hash []byte, offset, limit uint64, se
_, err = UserWarehouse.ReadFile(hash, int64(offset), int64(limit), udtConn)
// close the UDT client and virtual connection in any case
udtConn.Close() // warning: This is currently blocking.
//virtualConnection.Terminate(false, 1)
udtConn.Close() // warning: This is currently blocking in case the other side does not call Close().
return err
}
// RequestFileTransferUDT creates a UDT server listening for incoming data transfer and requests a file transfer from a remote peer.
func (peer *PeerInfo) RequestFileTransferUDT(hash []byte, offset, limit uint64) (udtConn net.Conn, udtListener net.Listener, err error) {
func (peer *PeerInfo) RequestFileTransferUDT(hash []byte, offset, limit uint64) (udtConn net.Conn, err error) {
virtualConnection := newVirtualPacketConn(peer, 0, hash, offset, limit, true)
// new sequence
sequence := networks.Sequences.NewSequenceBi(peer.PublicKey, &peer.messageSequence, virtualConnection, transferSequenceTimeout, virtualConnection.sequenceTerminate)
if sequence == nil {
return nil, nil, errors.New("cannot acquire sequence")
return nil, errors.New("cannot acquire sequence")
}
virtualConnection.sequenceNumber = sequence.SequenceNumber
// start UDT receiver
udtListener = udt.ListenUDT(udt.DefaultConfig(), virtualConnection)
udtListener := udt.ListenUDT(udt.DefaultConfig(), virtualConnection, virtualConnection.incomingData, virtualConnection.outgoingData, virtualConnection.terminateChan)
// request file transfer
peer.sendTransfer(nil, protocol.TransferControlRequestStart, virtualConnection.transferProtocol, hash, offset, limit, virtualConnection.sequenceNumber)
@@ -67,8 +66,10 @@ func (peer *PeerInfo) RequestFileTransferUDT(hash []byte, offset, limit uint64)
udtConn, err = udtListener.Accept()
if err != nil {
udtListener.Close()
return nil, nil, err
return nil, err
}
return udtConn, udtListener, nil
// We do not close the UDT listener here. It should automatically close after udtConn is closed.
return udtConn, nil
}

102
Transfer Virtual Connection.go
View File

@@ -3,23 +3,19 @@ File Name: Transfer Virtual Connection.go
Copyright: 2021 Peernet s.r.o.
Author: Peter Kleissner
This file defines a virtual net.PacketConn which sends transfer messages and can be used to embed other transfer protocols.
This file defines a virtual connection between a transfer protocol and Peernet messages.
If either the downstream transfer protocol or upstream Peernet messages indicate termination, the virtual connection ceases to exist.
*/
package core
import (
"errors"
"io"
"net"
"sync"
"time"
"github.com/PeernetOfficial/core/protocol"
)
// virtualPacketConn is a virtual connection.
// The required functions are ReadFrom, WriteTo, Close, LocalAddr, SetDeadline, SetReadDeadline, SetWriteDeadline.
type virtualPacketConn struct {
peer *PeerInfo
@@ -37,6 +33,7 @@ type virtualPacketConn struct {
// data channel
incomingData chan []byte
outgoingData chan []byte
// internal data
closed bool
@@ -47,7 +44,7 @@ type virtualPacketConn struct {
// newVirtualPacketConn creates a new virtual connection (both incomign and outgoing).
func newVirtualPacketConn(peer *PeerInfo, protocol uint8, hash []byte, offset, limit uint64, isListen bool) (v *virtualPacketConn) {
return &virtualPacketConn{
v = &virtualPacketConn{
transferProtocol: protocol,
peer: peer,
hash: hash,
@@ -55,64 +52,44 @@ func newVirtualPacketConn(peer *PeerInfo, protocol uint8, hash []byte, offset, l
limit: limit,
isListenMode: isListen,
incomingData: make(chan []byte),
outgoingData: make(chan []byte),
terminateChan: make(chan struct{}),
}
}
// ReadFrom reads a packet from the connection, copying the payload into p. It returns the number of bytes copied into p and the return address that was on the packet.
func (v *virtualPacketConn) ReadFrom(p []byte) (n int, addr net.Addr, err error) {
if v.closed {
return 0, nil, errors.New("connection closed")
}
// The underlying transfer messages feature a message sequence timeout (via Sequences.RegisterSequenceBi and Sequences.NewSequenceBi).
// The sequence timeout triggers the virtualPacketConn.Terminate function which closes terminateChan.
// Therefore an additional timeout here would be redundant. Instead, a transfer-wide timeout, that closes terminateChan upon expiration, shall be used.
select {
case data := <-v.incomingData:
n = copy(p, data)
case <-v.terminateChan:
return n, nil, io.EOF
}
go v.writeForward()
return
}
// Write sends the data to the remote peer. It makes it possible to use virtualPacketConn as io.Writer.
// Note: If the packet size exceed the limit from protocol.EncodeTransfer, this function fails.
func (v *virtualPacketConn) Write(p []byte) (n int, err error) {
if v.IsTerminated() {
return 0, errors.New("connection closed")
}
// writeForward forwards outgoing messages
func (v *virtualPacketConn) writeForward() {
for {
select {
case data := <-v.outgoingData:
v.peer.sendTransfer(data, protocol.TransferControlActive, v.transferProtocol, v.hash, v.offset, v.limit, v.sequenceNumber)
// create a new packet
err = v.peer.sendTransfer(p, protocol.TransferControlActive, v.transferProtocol, v.hash, v.offset, v.limit, v.sequenceNumber)
if err == nil {
n = len(p)
case <-v.terminateChan:
return
}
}
return
}
// receiveData receives incoming data via an external message. Blocking until a read occurs or the connection is terminated!
func (v *virtualPacketConn) receiveData(data []byte) (err error) {
func (v *virtualPacketConn) receiveData(data []byte) {
if v.IsTerminated() {
return errors.New("connection closed")
return
}
// pass on the data
// pass the data on
select {
case v.incomingData <- data:
case <-v.terminateChan:
}
return
}
// Terminate closes the connection and optionally sends a termination message to the remote peer. Multiple termination signals have no effect.
// Reason: 404 = Remote peer does not store file, 1 = Local termination signal, 2 = Remote termination signal, 3 = Sequence invalidation or expiration
func (v *virtualPacketConn) Terminate(sendSignal bool, reason int) (err error) {
// Terminate closes the connection. Do not call this function manually. Use the underlying protocol's function to close the connection.
// Reason: 404 = Remote peer does not store file (upstream), 1 = Transfer Protocol indicated closing (downstream), 2 = Remote termination signal (upstream), 3 = Sequence invalidation or expiration (upstream)
func (v *virtualPacketConn) Terminate(reason int) (err error) {
v.Lock()
defer v.Unlock()
@@ -124,10 +101,6 @@ func (v *virtualPacketConn) Terminate(sendSignal bool, reason int) (err error) {
v.reason = reason
close(v.terminateChan)
if sendSignal {
err = v.peer.sendTransfer(nil, protocol.TransferControlTerminate, v.transferProtocol, v.hash, 0, 0, v.sequenceNumber)
}
return
}
@@ -138,35 +111,12 @@ func (v *virtualPacketConn) IsTerminated() bool {
// sequenceTerminate is a wrapper for sequenece termination (invalidation or expiration)
func (v *virtualPacketConn) sequenceTerminate() {
v.Terminate(false, 3)
v.Terminate(3)
}
// Close closes the connection.
// Close provides a Close function to be called by the underlying transfer protocol.
// Do not call the function manually; otherwise the underlying transfer protocol may not have time to send a termination message (and the remote peer would subsequently try to reconnect).
func (v *virtualPacketConn) Close() (err error) {
return v.Terminate(false, 1)
}
// WriteTo writes a packet with payload p to addr. WriteTo can be made to time out and return an Error after a fixed time limit.
func (v *virtualPacketConn) WriteTo(p []byte, addr net.Addr) (n int, err error) {
return v.Write(p)
}
// LocalAddr returns the local network address
func (v *virtualPacketConn) LocalAddr() (addr net.Addr) {
return
}
// SetDeadline sets the read and write deadlines associated with the connection. It is equivalent to calling both SetReadDeadline and SetWriteDeadline.
func (v *virtualPacketConn) SetDeadline(t time.Time) (err error) {
return nil
}
// SetReadDeadline sets the deadline for future ReadFrom calls and any currently-blocked ReadFrom call.
func (v *virtualPacketConn) SetReadDeadline(t time.Time) (err error) {
return nil
}
// SetWriteDeadline sets the deadline for future WriteTo calls and any currently-blocked WriteTo call.
func (v *virtualPacketConn) SetWriteDeadline(t time.Time) (err error) {
return nil
networks.Sequences.InvalidateSequence(v.peer.PublicKey, v.sequenceNumber, true)
return v.Terminate(1)
}

6
protocol/Sequence.go
View File

@@ -165,9 +165,9 @@ func (manager *SequenceManager) ValidateSequence(publicKey *btcec.PublicKey, seq
return sequence, sequence.expires.After(time.Now()), rtt
}
// InvalidateSequence invalidates the sequence number.
func (manager *SequenceManager) InvalidateSequence(publicKey *btcec.PublicKey, sequenceNumber uint32) {
key := sequence2Key(false, publicKey, sequenceNumber)
// InvalidateSequence invalidates the sequence number. It does not call invalidateFunc.
func (manager *SequenceManager) InvalidateSequence(publicKey *btcec.PublicKey, sequenceNumber uint32, bidirectional bool) {
key := sequence2Key(bidirectional, publicKey, sequenceNumber)
manager.Lock()
delete(manager.sequences, key)

5
udt/listener.go
View File

@@ -5,6 +5,7 @@ import (
"encoding/binary"
"errors"
"fmt"
"io"
"net"
"sync"
"time"
@@ -162,8 +163,8 @@ func (l *listener) readHandshake(m *multiplexer, hsPacket *packet.HandshakePacke
}
// ListenUDT listens for incoming UDT connections using the existing provided packet connection. It creates a UDT server.
func ListenUDT(config *Config, packetConn net.PacketConn) net.Listener {
m := newMultiplexer(packetConn, config.MaxPacketSize)
func ListenUDT(config *Config, closer io.Closer, incomingData <-chan []byte, outgoingData chan<- []byte, terminationSignal <-chan struct{}) net.Listener {
m := newMultiplexer(closer, config.MaxPacketSize, incomingData, outgoingData, terminationSignal)
l := &listener{
m: m,

56
udt/multiplexer.go
View File

@@ -4,8 +4,8 @@ package udt
import (
"fmt"
"io"
"math/rand"
"net"
"sync"
"github.com/PeernetOfficial/core/udt/packet"
@@ -13,20 +13,27 @@ import (
// A multiplexer is a single UDT socket over a single PacketConn.
type multiplexer struct {
conn net.PacketConn // the UDPConn from which we read/write
socket *udtSocket // Socket
socketID uint32 // Socket ID
listenSock *listener // the server socket listening to incoming connections, if there is one. Set by caller.
maxPacketSize uint // the Maximum Transmission Unit of packets sent from this address
pktOut chan packet.Packet // packets queued for immediate sending
sync.Mutex // Synchronized access to socket/listenSock
socket *udtSocket // Socket
socketID uint32 // Socket ID
listenSock *listener // the server socket listening to incoming connections, if there is one. Set by caller.
maxPacketSize uint // the Maximum Transmission Unit of packets sent from this address
pktOut chan packet.Packet // packets queued for immediate sending
sync.Mutex // Synchronized access to socket/listenSock
incomingData <-chan []byte // source to read packets from
outgoingData chan<- []byte // destination to send packets to
terminationSignal <-chan struct{} // external termination signal to watch
closer io.Closer // external closer to call in case the local socket/listener closes
}
func newMultiplexer(conn net.PacketConn, maxPacketSize uint) (m *multiplexer) {
// The closer is called when the socket/listener closes. The terminationSignal is an external (upstream) signal to watch for.
func newMultiplexer(closer io.Closer, maxPacketSize uint, incomingData <-chan []byte, outgoingData chan<- []byte, terminationSignal <-chan struct{}) (m *multiplexer) {
m = &multiplexer{
conn: conn,
maxPacketSize: maxPacketSize, // to be verified?!
pktOut: make(chan packet.Packet, 100), // todo: figure out how to size this
maxPacketSize: maxPacketSize,
pktOut: make(chan packet.Packet, 100),
closer: closer,
incomingData: incomingData,
outgoingData: outgoingData,
terminationSignal: terminationSignal,
}
go m.goRead()
@@ -46,7 +53,7 @@ func (m *multiplexer) unlistenUDT(l *listener) {
m.listenSock = nil
m.conn.Close()
m.closer.Close()
close(m.pktOut)
}
@@ -66,22 +73,23 @@ func (m *multiplexer) closeSocket(sockID uint32) {
m.socket = nil
m.conn.Close()
m.closer.Close()
close(m.pktOut)
}
// read runs in a goroutine and reads packets from conn using a buffer from the readBufferPool, or a new buffer.
func (m *multiplexer) goRead() {
buf := make([]byte, m.maxPacketSize)
for {
numBytes, _, err := m.conn.ReadFrom(buf)
if err != nil {
var buf []byte
select {
case buf = <-m.incomingData:
case <-m.terminationSignal:
return
}
p, err := packet.DecodePacket(buf[0:numBytes])
p, err := packet.DecodePacket(buf)
if err != nil {
fmt.Printf("Unable to decode packet: %s\n", err)
fmt.Printf("Error decoding UDT packet: %s\n", err)
return
}
@@ -111,16 +119,16 @@ func (m *multiplexer) goRead() {
func (m *multiplexer) goWrite() {
buf := make([]byte, m.maxPacketSize)
for pkt := range m.pktOut {
plen, err := pkt.WriteTo(buf)
plen, err := pkt.WriteTo(buf) // encode
if err != nil {
// TODO: handle write error
fmt.Printf("Unable to buffer out: %s\n", err.Error())
fmt.Printf("Error encoding UDT packet: %s\n", err.Error())
return
}
if _, err = m.conn.WriteTo(buf[0:plen], nil); err != nil {
// TODO: handle write error
fmt.Printf("Unable to write out: %s\n", err.Error())
select {
case m.outgoingData <- buf[0:plen]:
case <-m.terminationSignal:
return
}
}

5
udt/udt.go
View File

@@ -12,12 +12,13 @@ implemented:
*/
import (
"io"
"net"
)
// DialUDT establishes an outbound UDT connection using the existing provided packet connection. It creates a UDT client.
func DialUDT(config *Config, packetConn net.PacketConn, isStream bool) (net.Conn, error) {
m := newMultiplexer(packetConn, config.MaxPacketSize)
func DialUDT(config *Config, closer io.Closer, incomingData <-chan []byte, outgoingData chan<- []byte, terminationSignal <-chan struct{}, isStream bool) (net.Conn, error) {
m := newMultiplexer(closer, config.MaxPacketSize, incomingData, outgoingData, terminationSignal)
s := m.newSocket(config, false, !isStream)
err := s.startConnect()