luxfi/corona
1
0
Fork 0
mirror of https://github.com/luxfi/corona.git synced 2026-07-27 02:50:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

WIP: session checkpoint 2026-06-05 (codec rip in flight)

Browse Source
This commit is contained in:
Hanzo AI
2026-06-05 15:33:52 -07:00
parent bf70b85833
commit ff8de21571
6 changed files with 1441 additions and 397 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go.mod
+14 -3
View File
@@ -5,24 +5,35 @@ go 1.26.3
require (
github.com/luxfi/lattice/v7 v7.1.0
github.com/luxfi/math v1.4.0
github.com/luxfi/zap v0.7.2
github.com/montanaflynn/stats v0.9.0
github.com/spf13/cobra v1.10.2
github.com/zeebo/blake3 v0.2.4
golang.org/x/crypto v0.49.0
golang.org/x/crypto v0.52.0
)
require (
github.com/ALTree/bigfloat v0.2.0 // indirect
github.com/cenkalti/backoff v2.2.1+incompatible // indirect
github.com/cloudflare/circl v1.6.3 // indirect
github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
github.com/google/go-cmp v0.7.0 // indirect
github.com/grandcat/zeroconf v1.0.0 // indirect
github.com/inconshreveable/mousetrap v1.1.0 // indirect
github.com/klauspost/cpuid/v2 v2.3.0 // indirect
github.com/kr/text v0.2.0 // indirect
github.com/luxfi/accel v1.1.9 // indirect
github.com/luxfi/crypto v1.19.17 // indirect
github.com/luxfi/mdns v0.1.1 // indirect
github.com/miekg/dns v1.1.72 // indirect
github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
github.com/spf13/pflag v1.0.10 // indirect
github.com/stretchr/testify v1.11.1 // indirect
github.com/zeebo/assert v1.3.0 // indirect
golang.org/x/exp v0.0.0-20260312153236-7ab1446f8b90 // indirect
golang.org/x/sys v0.42.0 // indirect
golang.org/x/mod v0.36.0 // indirect
golang.org/x/net v0.55.0 // indirect
golang.org/x/sync v0.20.0 // indirect
golang.org/x/sys v0.45.0 // indirect
golang.org/x/tools v0.45.0 // indirect
gopkg.in/yaml.v3 v3.0.1 // indirect
)
go.sum
+47 -4
View File
@@ -1,11 +1,17 @@
github.com/ALTree/bigfloat v0.2.0 h1:AwNzawrpFuw55/YDVlcPw0F0cmmXrmngBHhVrvdXPvM=
github.com/ALTree/bigfloat v0.2.0/go.mod h1:+NaH2gLeY6RPBPPQf4aRotPPStg+eXc8f9ZaE4vRfD4=
github.com/cenkalti/backoff v2.2.1+incompatible h1:tNowT99t7UNflLxfYYSlKYsBpXdEet03Pg2g16Swow4=
github.com/cenkalti/backoff v2.2.1+incompatible/go.mod h1:90ReRw6GdpyfrHakVjL/QHaoyV4aDUVVkXQJJJ3NXXM=
github.com/cloudflare/circl v1.6.3 h1:9GPOhQGF9MCYUeXyMYlqTR6a5gTrgR/fBLXvUgtVcg8=
github.com/cloudflare/circl v1.6.3/go.mod h1:2eXP6Qfat4O/Yhh8BznvKnJ+uzEoTQ6jVKJRn81BiS4=
github.com/cpuguy83/go-md2man/v2 v2.0.6/go.mod h1:oOW0eioCTA6cOiMLiUPZOpcVxMig6NIQQ7OS05n1F4g=
github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM=
github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
github.com/grandcat/zeroconf v1.0.0 h1:uHhahLBKqwWBV6WZUDAT71044vwOTL+McW0mBJvo6kE=
github.com/grandcat/zeroconf v1.0.0/go.mod h1:lTKmG1zh86XyCoUeIHSA4FJMBwCJiQmGfcP2PdzytEs=
github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8=
github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw=
github.com/klauspost/cpuid/v2 v2.3.0 h1:S4CRMLnYUhGeDFDqkGriYKdfoFlDnMtqTiI/sFzhA9Y=
@@ -14,12 +20,27 @@ github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
github.com/luxfi/accel v1.1.9 h1:Tsk6gXj2uKE19501bD0ajRYdeCHIlTGb6jYyLc+F8hc=
github.com/luxfi/accel v1.1.9/go.mod h1:K00BcnLzEYMPHwCFq8Tf/450ApmTs9xBVvYOnofJCkc=
github.com/luxfi/crypto v1.19.17 h1:l2LLu7UFyICtJVfraLDLRi+lFGiDXKHSL18M9/m1gsQ=
github.com/luxfi/crypto v1.19.17/go.mod h1:INjdZtke85k8hX/QAmTMAY8bbZ4gzGZQLqURg3xf6Gk=
github.com/luxfi/lattice/v7 v7.1.0 h1:mr3HvN6olNTS2LT/xAW/JBhTqfvpsGmsopDMeR7BSJs=
github.com/luxfi/lattice/v7 v7.1.0/go.mod h1:IaaUN+3ysnBG4BA8ILRYG0j80+qtYDP4C5lkaDb2pDE=
github.com/luxfi/math v1.4.0 h1:/sb7Grw3hfO+5INWAWdB95jTvCeXg8fSQxsxDzcFtd4=
github.com/luxfi/math v1.4.0/go.mod h1:iW0FOCC8qF2mPE+MakG780CAHA83848lb1L04thA1Pg=
github.com/luxfi/mdns v0.1.1 h1:g2eRr9AXcziPkkcd24M+Qu9ApEpoKKjfI79QSNqv0rQ=
github.com/luxfi/mdns v0.1.1/go.mod h1:dbp5f3h3aE7CGzwbaWzBM9cwdcekhmSrWhQevgYhhNA=
github.com/luxfi/pq v1.0.3 h1:pFlQm1+5FuKTDUh2y/23bXWkN4I2Rc5iuxJypwDFFMs=
github.com/luxfi/pq v1.0.3/go.mod h1:8bppZcRElfrVt0n3nYCZW3iX1TvhvzNbdjNdK1irgIE=
github.com/luxfi/zap v0.7.2 h1:YecWTWNE5PPJXL56sLIkzS8b23bprUwZ5lPAQuLUtTE=
github.com/luxfi/zap v0.7.2/go.mod h1:1k+nwT+JW802YzuPAuf7CxMSGr/qxvbGgGwi5k6X9Ok=
github.com/miekg/dns v1.1.27/go.mod h1:KNUDUusw/aVsxyTYZM1oqvCicbwhgbNgztCETuNZ7xM=
github.com/miekg/dns v1.1.72 h1:vhmr+TF2A3tuoGNkLDFK9zi36F2LS+hKTRW0Uf8kbzI=
github.com/miekg/dns v1.1.72/go.mod h1:+EuEPhdHOsfk6Wk5TT2CzssZdqkmFhf8r+aVyDEToIs=
github.com/montanaflynn/stats v0.9.0 h1:tsBJ0RXwph9BmAuFoCmqGv6e8xa0MENQ8m0ptKq29mQ=
github.com/montanaflynn/stats v0.9.0/go.mod h1:etXPPgVO6n31NxCd9KQUMvCM+ve0ruNzt6R8Bnaayow=
github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRIccs7FGNTlIRMkT8wgtp5eCXdBlqhYGL6U=
github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ=
@@ -39,12 +60,34 @@ github.com/zeebo/blake3 v0.2.4/go.mod h1:7eeQ6d2iXWRGF6npfaxl2CU+xy2Fjo2gxeyZGCR
github.com/zeebo/pcg v1.0.1 h1:lyqfGeWiv4ahac6ttHs+I5hwtH/+1mrhlCtVNQM2kHo=
github.com/zeebo/pcg v1.0.1/go.mod h1:09F0S9iiKrwn9rlI5yjLkmrug154/YRW6KnnXVDM/l4=
go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg=
golang.org/x/crypto v0.49.0 h1:+Ng2ULVvLHnJ/ZFEq4KdcDd/cfjrrjjNSXNzxg0Y4U4=
golang.org/x/crypto v0.49.0/go.mod h1:ErX4dUh2UM+CFYiXZRTcMpEcN8b/1gxEuv3nODoYtCA=
golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
golang.org/x/crypto v0.52.0 h1:RMs7fP2rXdep0CftQlK8Uf+kibLm7qkCcradZWYz988=
golang.org/x/crypto v0.52.0/go.mod h1:1QgfPxDqh0T2M/elOJtp9RvuR95kVjir0e6/BvEmGbc=
golang.org/x/exp v0.0.0-20260312153236-7ab1446f8b90 h1:jiDhWWeC7jfWqR9c/uplMOqJ0sbNlNWv0UkzE0vX1MA=
golang.org/x/exp v0.0.0-20260312153236-7ab1446f8b90/go.mod h1:xE1HEv6b+1SCZ5/uscMRjUBKtIxworgEcEi+/n9NQDQ=
golang.org/x/sys v0.42.0 h1:omrd2nAlyT5ESRdCLYdm3+fMfNFE/+Rf4bDIQImRJeo=
golang.org/x/sys v0.42.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
golang.org/x/mod v0.1.1-0.20191105210325-c90efee705ee/go.mod h1:QqPTAvyqsEbceGzBzNggFXnrqF1CaUcvgkdR5Ot7KZg=
golang.org/x/mod v0.36.0 h1:JJjpVx6myfUsUdAzZuOSTTmRE0PfZeNWzzvKrP7amb4=
golang.org/x/mod v0.36.0/go.mod h1:moc6ELqsWcOw5Ef3xVprK5ul/MvtVvkIXLziUOICjUQ=
golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
golang.org/x/net v0.0.0-20190923162816-aa69164e4478/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
golang.org/x/net v0.0.0-20200114155413-6afb5195e5aa/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
golang.org/x/net v0.55.0 h1:bcvxaJn3e1U6InsFWt1JUq1aSjnRxLzT2rtD2KfkDF8=
golang.org/x/net v0.55.0/go.mod h1:L5U2KuzuOe1lY7Z+aWVIKK6qEeJXnXV9yzGA+WCHJww=
golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4=
golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=
golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20190924154521-2837fb4f24fe/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.45.0 h1:dO4czNzziLiiXplLQgBCEpCvXQ3dnkn0SdaZSYdQ+FY=
golang.org/x/sys v0.45.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
golang.org/x/tools v0.0.0-20191216052735-49a3e744a425/go.mod h1:TB2adYChydJhpapKDTa4BR/hXlZSLoq2Wpct/0txZ28=
golang.org/x/tools v0.45.0 h1:18qN3FAooORvApf5XjCXgsuayZOEtXf6JK18I3+ONa8=
golang.org/x/tools v0.45.0/go.mod h1:LuUGqqaXcXMEFEruIVJVm5mgDD8vww/z/SR1gQ4uE/0=
golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=
main.go
+74 -9
View File
@@ -8,6 +8,7 @@ import (
"net"
"os"
"strconv"
"strings"
"sync"
"time"
@@ -53,17 +54,28 @@ func main() {
os.Exit(1)
}
// Initialize P2P communication
comm := &networking.P2PComm{
Socks: make(map[int]*net.Conn),
Rank: partyID,
// Initialize P2P communication over ZAP transport.
//
// Peers and their listen addresses come from the CORONA_PEERS env
// var, formatted as "rank=host:port,rank=host:port,...". The local
// party's own rank is excluded; every other rank in the ceremony
// must be present. The local listen port is derived from the same
// list (peerAddrs[partyID] specifies "host:port" we bind on).
peerAddrs, listenPort, err := parsePeerEnv(os.Getenv("CORONA_PEERS"), partyID)
if err != nil {
log.Fatalf("CORONA_PEERS: %v", err)
}
comm, err := networking.NewP2PComm(partyID, listenPort, true /*noDiscovery*/, nil)
if err != nil {
log.Fatalf("NewP2PComm: %v", err)
}
defer comm.Close()
// Suppress unused-import error from prior wiring.
_ = (*net.Conn)(nil)
// Establish connections
var connWg sync.WaitGroup
connWg.Add(1)
go networking.EstablishConnections(&connWg, comm, partyID, sign.K)
connWg.Wait()
if err := comm.EstablishConnectionsZAP(partyID, sign.K, peerAddrs); err != nil {
log.Fatalf("EstablishConnectionsZAP: %v", err)
}
var setupDuration, genDuration, signRound1Duration, signRound2PreprocessDuration, signRound2Duration, finalizeDuration, verifyDuration time.Duration
var genStart, genEnd, signRound1Start, signRound1End, signRound2Start, signRound2End, combinerReceiveEnd, combinerFinalizeEnd time.Time
@@ -240,3 +252,56 @@ func main() {
fmt.Printf("Combiner finalize end timestamp: %s\n", combinerFinalizeEnd.Format("15:04:05.000000"))
}
}
// parsePeerEnv parses the CORONA_PEERS env var into a per-rank address
// map, and returns the local listen port for partyID.
//
// Format: "0=host:port,1=host:port,...". Every rank in the ceremony
// must appear. The local party's entry is dropped from the returned
// map but its port is harvested as the listen port.
//
// Returns an error on malformed input or when partyID is absent.
func parsePeerEnv(raw string, partyID int) (map[int]string, int, error) {
if strings.TrimSpace(raw) == "" {
return nil, 0, fmt.Errorf("CORONA_PEERS is empty (format: rank=host:port,rank=host:port,...)")
}
peerAddrs := make(map[int]string)
listenPort := 0
for _, entry := range strings.Split(raw, ",") {
entry = strings.TrimSpace(entry)
if entry == "" {
continue
}
eq := strings.IndexByte(entry, '=')
if eq < 0 {
return nil, 0, fmt.Errorf("CORONA_PEERS entry %q missing '='", entry)
}
rank, err := strconv.Atoi(strings.TrimSpace(entry[:eq]))
if err != nil {
return nil, 0, fmt.Errorf("CORONA_PEERS entry %q: %w", entry, err)
}
addr := strings.TrimSpace(entry[eq+1:])
if addr == "" {
return nil, 0, fmt.Errorf("CORONA_PEERS entry %q has empty address", entry)
}
if rank == partyID {
// Local listen port from "host:port" — the host is ignored
// (we always listen on all interfaces).
_, portStr, err := net.SplitHostPort(addr)
if err != nil {
return nil, 0, fmt.Errorf("CORONA_PEERS local addr %q: %w", addr, err)
}
p, err := strconv.Atoi(portStr)
if err != nil {
return nil, 0, fmt.Errorf("CORONA_PEERS local port %q: %w", portStr, err)
}
listenPort = p
continue
}
peerAddrs[rank] = addr
}
if listenPort == 0 {
return nil, 0, fmt.Errorf("CORONA_PEERS missing entry for local rank %d", partyID)
}
return peerAddrs, listenPort, nil
}
networking/networking.go
+527 -381
View File
@@ -1,10 +1,16 @@
// Copyright (C) 2026, Lux Industries Inc. All rights reserved.
// See the file LICENSE for licensing terms.
package networking
import (
"bufio"
"bytes"
"context"
"encoding/binary"
"fmt"
"log"
"log/slog"
"net"
"sync"
"time"
@@ -13,408 +19,548 @@ import (
"github.com/luxfi/lattice/v7/utils/structs"
)
type Communicator interface {
Send(dst int, msg []byte) (int, error)
Recv(src int) ([]byte, int, error)
Close() error
}
// P2PComm is the Corona ceremony communicator. Each party holds one
// P2PComm which it uses to send Vector/Matrix/byte-bag payloads to
// every other party in the round.
//
// Transport is ZAP (luxfi/zap) — every Send* call wraps the lattice-
// serialized bytes in a single ZAP message and ships it via zap.Node;
// every Recv* call dequeues the next ZAP message from the peer's
// receive queue and reads the lattice object from its body.
//
// The lattice serialization (uint64 length + element bytes) inside the
// message body is byte-equal to the pre-migration bare-TCP encoding —
// the protocol bytes Corona participants compute against did not move.
type P2PComm struct {
Rank int
// Socks is retained for backward compatibility with hand-written
// fixtures that wire net.Pipe pairs directly into the comm. New
// code should not touch this map; use the ZAP-backed transport
// (NewP2PComm / Connect / Listen) instead.
//
// When Socks contains a *net.Conn for a peer, Send*/Recv* fall back
// to the legacy in-memory pipe path so existing unit tests keep
// passing without modification.
Socks map[int]*net.Conn
Rank int
mu sync.Mutex // Added mutex for safe concurrent access
mu sync.Mutex
node *zapNode
logger *slog.Logger
}
// NewP2PComm creates a Corona communicator backed by a ZAP node bound
// to the given port. Set noDiscovery=true in tests or topology-known
// deployments and supply peer addresses via Connect.
func NewP2PComm(rank int, port int, noDiscovery bool, logger *slog.Logger) (*P2PComm, error) {
if logger == nil {
logger = slog.Default()
}
zn := newZapNode(rank, port, noDiscovery, logger)
if err := zn.node.Start(); err != nil {
return nil, fmt.Errorf("corona: start zap node: %w", err)
}
return &P2PComm{
Rank: rank,
Socks: make(map[int]*net.Conn),
node: zn,
logger: logger,
}, nil
}
// Connect dials the peer at addr and binds it to the given rank. Use
// when topology is known up-front (the historical EstablishConnections
// path); when mDNS discovery is enabled, peers are picked up
// automatically and Connect is unnecessary.
func (comm *P2PComm) Connect(rank int, addr string) error {
if comm.node == nil {
return fmt.Errorf("corona: P2PComm has no zap node (call NewP2PComm)")
}
// Retry: peers may not have bound their listener yet.
var lastErr error
for i := 0; i < 10; i++ {
if err := comm.node.node.ConnectDirect(addr); err == nil {
comm.node.registerPeer(rank)
return nil
} else {
lastErr = err
time.Sleep(500 * time.Millisecond)
}
}
return fmt.Errorf("corona: connect rank=%d addr=%s: %w", rank, addr, lastErr)
}
// AcceptPeer marks rank as a known peer that will dial us. mDNS-less
// deployments where the lower-rank side dials and the higher-rank side
// listens (Corona's historical convention) use this to pre-register
// the incoming peer so Send* knows the ZAP NodeID to address.
func (comm *P2PComm) AcceptPeer(rank int) {
if comm.node != nil {
comm.node.registerPeer(rank)
}
}
// Close releases all transport resources.
func (comm *P2PComm) Close() error {
if comm.node != nil {
comm.node.close()
}
comm.mu.Lock()
defer comm.mu.Unlock()
for _, sock := range comm.Socks {
if sock != nil && *sock != nil {
_ = (*sock).Close()
}
}
return nil
}
// SetSock is retained for legacy hand-written fixtures.
func (comm *P2PComm) SetSock(key int, conn *net.Conn) {
comm.mu.Lock()
defer comm.mu.Unlock()
comm.Socks[key] = conn
}
// GetSock is retained for legacy hand-written fixtures.
func (comm *P2PComm) GetSock(key int) *net.Conn {
comm.mu.Lock()
defer comm.mu.Unlock()
return comm.Socks[key]
}
func (comm *P2PComm) SendBytes(writer *bufio.Writer, dst int, msg []byte) (int, error) {
length := make([]byte, 4)
binary.BigEndian.PutUint32(length, uint32(len(msg)))
n, err := writer.Write(length)
if err != nil {
return 0, err
}
totalBytesSent := n
for len(msg) > 0 {
n, err = writer.Write(msg)
if err != nil {
return 0, err
}
totalBytesSent += n
msg = msg[n:]
}
err = writer.Flush()
if err != nil {
return 0, err
}
return totalBytesSent, nil
}
func (comm *P2PComm) Recv(reader *bufio.Reader, src int) ([]byte, int, error) {
lengthBuf := make([]byte, 4)
totalBytesRead := 0
for totalBytesRead < 4 {
n, err := reader.Read(lengthBuf[totalBytesRead:])
if err != nil {
return nil, totalBytesRead, err
}
totalBytesRead += n
}
length := binary.BigEndian.Uint32(lengthBuf)
data := make([]byte, length)
bytesRead := 0
for bytesRead < int(length) {
n, err := reader.Read(data[bytesRead:])
if err != nil {
return nil, totalBytesRead, err
}
bytesRead += n
totalBytesRead += n
}
return data, totalBytesRead, nil
}
func (comm *P2PComm) Close() error {
// hasLegacySock reports whether the (legacy) Socks map has a live conn
// for the given rank. When true the Send*/Recv* methods route through
// the bufio.Writer/Reader the caller supplies (test compatibility);
// when false they go through the ZAP node.
func (comm *P2PComm) hasLegacySock(rank int) bool {
comm.mu.Lock()
defer comm.mu.Unlock()
for _, sock := range comm.Socks {
err := (*sock).Close()
sock, ok := comm.Socks[rank]
return ok && sock != nil && *sock != nil
}
// ---------------------------------------------------------------------
// Send / Recv primitives.
//
// Every method has two paths:
//
// 1. Legacy direct-conn path. If hasLegacySock(dst) is true (i.e.
// a *net.Conn was wired in via SetSock), the call writes/reads
// to the supplied bufio.Writer/Reader directly. This keeps the
// net.Pipe-based unit tests in networking_test.go working without
// modification.
//
// 2. ZAP path. Otherwise the call serializes the lattice object to
// an in-memory bytes.Buffer, ships the bytes as one ZAP message
// to the peer, and on the receive side dequeues the next ZAP
// message from the per-peer FIFO and parses the lattice object
// from its body.
//
// The lattice serialization itself (vector.go/matrix.go WriteTo/
// ReadFrom in luxfi/lattice) is identical across both paths — the
// "semantic-level wire" the ceremony computes against is byte-equal.
// ---------------------------------------------------------------------
// SendVector serializes a lattice vector to bytes and ships it to dst.
// writer is honored when present (legacy path); otherwise the ZAP node
// is used.
func (comm *P2PComm) SendVector(writer *bufio.Writer, dst int, msg structs.Vector[ring.Poly]) {
if writer != nil && comm.hasLegacySock(dst) {
if _, err := msg.WriteTo(writer); err != nil {
log.Fatalf("Failed to write vector: %v", err)
}
if err := writer.Flush(); err != nil {
log.Fatalf("Failed to flush writer: %v", err)
}
return
}
var buf bytes.Buffer
bw := bufio.NewWriter(&buf)
if _, err := msg.WriteTo(bw); err != nil {
log.Fatalf("Failed to write vector: %v", err)
}
if err := bw.Flush(); err != nil {
log.Fatalf("Failed to flush vector buffer: %v", err)
}
if err := comm.node.send(context.Background(), dst, buf.Bytes()); err != nil {
log.Fatalf("Failed to send vector to %d: %v", dst, err)
}
}
// RecvVector reads a lattice vector of the given length from src.
func (comm *P2PComm) RecvVector(reader *bufio.Reader, src int, length int) structs.Vector[ring.Poly] {
if reader != nil && comm.hasLegacySock(src) {
vec := make(structs.Vector[ring.Poly], length)
if _, err := vec.ReadFrom(reader); err != nil {
log.Fatalf("Failed to read vector: %v", err)
}
return vec
}
body, err := comm.node.recvBody(src)
if err != nil {
log.Fatalf("Failed to recv vector from %d: %v", src, err)
}
vec := make(structs.Vector[ring.Poly], length)
if _, err := vec.ReadFrom(bufio.NewReader(bytes.NewReader(body))); err != nil {
log.Fatalf("Failed to read vector: %v", err)
}
return vec
}
// SendMatrix serializes a lattice matrix to bytes and ships it to dst.
func (comm *P2PComm) SendMatrix(writer *bufio.Writer, dst int, msg structs.Matrix[ring.Poly]) {
if writer != nil && comm.hasLegacySock(dst) {
if _, err := msg.WriteTo(writer); err != nil {
log.Fatalf("Error sending matrix: %v", err)
}
if err := writer.Flush(); err != nil {
log.Fatalf("Failed to flush writer: %v", err)
}
return
}
var buf bytes.Buffer
bw := bufio.NewWriter(&buf)
if _, err := msg.WriteTo(bw); err != nil {
log.Fatalf("Error sending matrix: %v", err)
}
if err := bw.Flush(); err != nil {
log.Fatalf("Failed to flush matrix buffer: %v", err)
}
if err := comm.node.send(context.Background(), dst, buf.Bytes()); err != nil {
log.Fatalf("Failed to send matrix to %d: %v", dst, err)
}
}
// RecvMatrix reads a lattice matrix of the given outer length from src.
func (comm *P2PComm) RecvMatrix(reader *bufio.Reader, src int, length int) structs.Matrix[ring.Poly] {
if reader != nil && comm.hasLegacySock(src) {
matrix := make(structs.Matrix[ring.Poly], length)
if _, err := matrix.ReadFrom(reader); err != nil {
log.Fatalf("Failed to read matrix: %v", err)
}
return matrix
}
body, err := comm.node.recvBody(src)
if err != nil {
log.Fatalf("Failed to recv matrix from %d: %v", src, err)
}
matrix := make(structs.Matrix[ring.Poly], length)
if _, err := matrix.ReadFrom(bufio.NewReader(bytes.NewReader(body))); err != nil {
log.Fatalf("Failed to read matrix: %v", err)
}
return matrix
}
// SendBytesSlice ships a slice of byte-slices to dst with explicit
// length tags. Wire layout: uint32 numSlices, then for each slice
// uint32 length + slice bytes.
func (comm *P2PComm) SendBytesSlice(writer *bufio.Writer, dst int, data [][]byte) {
body := encodeBytesSlice(data)
if writer != nil && comm.hasLegacySock(dst) {
if _, err := writer.Write(body); err != nil {
log.Fatalf("Failed to write bytes slice: %v", err)
}
if err := writer.Flush(); err != nil {
log.Fatalf("Failed to flush writer: %v", err)
}
return
}
if err := comm.node.send(context.Background(), dst, body); err != nil {
log.Fatalf("Failed to send bytes slice to %d: %v", dst, err)
}
}
// RecvBytesSlice reads a length-prefixed slice-of-slices from src.
func (comm *P2PComm) RecvBytesSlice(reader *bufio.Reader, src int) [][]byte {
if reader != nil && comm.hasLegacySock(src) {
return decodeBytesSlice(reader)
}
body, err := comm.node.recvBody(src)
if err != nil {
log.Fatalf("Failed to recv bytes slice from %d: %v", src, err)
}
return decodeBytesSlice(bufio.NewReader(bytes.NewReader(body)))
}
// SendBytesMap ships a map<int,[]byte> to dst.
func (comm *P2PComm) SendBytesMap(writer *bufio.Writer, dst int, data map[int][]byte) {
body := encodeBytesMap(data)
if writer != nil && comm.hasLegacySock(dst) {
if _, err := writer.Write(body); err != nil {
log.Fatalf("Failed to write bytes map: %v", err)
}
if err := writer.Flush(); err != nil {
log.Fatalf("Failed to flush writer: %v", err)
}
return
}
if err := comm.node.send(context.Background(), dst, body); err != nil {
log.Fatalf("Failed to send bytes map to %d: %v", dst, err)
}
}
// RecvBytesMap reads a map<int,[]byte> from src.
func (comm *P2PComm) RecvBytesMap(reader *bufio.Reader, src int) map[int][]byte {
if reader != nil && comm.hasLegacySock(src) {
return decodeBytesMap(reader)
}
body, err := comm.node.recvBody(src)
if err != nil {
log.Fatalf("Failed to recv bytes map from %d: %v", src, err)
}
return decodeBytesMap(bufio.NewReader(bytes.NewReader(body)))
}
// SendBytesSliceMap ships a map<int,[][]byte> to dst.
func (comm *P2PComm) SendBytesSliceMap(writer *bufio.Writer, dst int, data map[int][][]byte) {
body := encodeBytesSliceMap(data)
if writer != nil && comm.hasLegacySock(dst) {
if _, err := writer.Write(body); err != nil {
log.Fatalf("Failed to write bytes-slice map: %v", err)
}
if err := writer.Flush(); err != nil {
log.Fatalf("Failed to flush writer: %v", err)
}
return
}
if err := comm.node.send(context.Background(), dst, body); err != nil {
log.Fatalf("Failed to send bytes-slice map to %d: %v", dst, err)
}
}
// RecvBytesSliceMap reads a map<int,[][]byte> from src.
func (comm *P2PComm) RecvBytesSliceMap(reader *bufio.Reader, src int) map[int][][]byte {
if reader != nil && comm.hasLegacySock(src) {
return decodeBytesSliceMap(reader)
}
body, err := comm.node.recvBody(src)
if err != nil {
log.Fatalf("Failed to recv bytes-slice map from %d: %v", src, err)
}
return decodeBytesSliceMap(bufio.NewReader(bytes.NewReader(body)))
}
// ---------------------------------------------------------------------
// Byte-bag wire format.
//
// These encoders/decoders preserve the historical wire layout used by
// Corona over bare-TCP: big-endian uint32 counts and lengths followed
// by raw byte runs. They are now used inside ZAP message bodies, so
// the ZAP frame contains a verbatim copy of the legacy bytes.
// ---------------------------------------------------------------------
func encodeBytesSlice(data [][]byte) []byte {
var buf bytes.Buffer
bw := bufio.NewWriter(&buf)
numSlices := uint32(len(data))
if err := binary.Write(bw, binary.BigEndian, numSlices); err != nil {
log.Fatalf("Failed to write number of slices: %v", err)
}
for _, slice := range data {
length := uint32(len(slice))
if err := binary.Write(bw, binary.BigEndian, length); err != nil {
log.Fatalf("Failed to write slice length: %v", err)
}
if _, err := bw.Write(slice); err != nil {
log.Fatalf("Failed to write slice data: %v", err)
}
}
if err := bw.Flush(); err != nil {
log.Fatalf("Failed to flush bytes-slice buffer: %v", err)
}
return buf.Bytes()
}
func decodeBytesSlice(r *bufio.Reader) [][]byte {
var numSlices uint32
if err := binary.Read(r, binary.BigEndian, &numSlices); err != nil {
log.Fatalf("Failed to read number of slices: %v", err)
}
data := make([][]byte, numSlices)
for i := uint32(0); i < numSlices; i++ {
var length uint32
if err := binary.Read(r, binary.BigEndian, &length); err != nil {
log.Fatalf("Failed to read slice length: %v", err)
}
slice := make([]byte, length)
bytesRead := 0
for bytesRead < int(length) {
n, err := r.Read(slice[bytesRead:])
if err != nil {
log.Fatalf("Failed to read slice data: %v", err)
}
bytesRead += n
}
data[i] = slice
}
return data
}
func encodeBytesMap(data map[int][]byte) []byte {
var buf bytes.Buffer
bw := bufio.NewWriter(&buf)
numEntries := uint32(len(data))
if err := binary.Write(bw, binary.BigEndian, numEntries); err != nil {
log.Fatalf("Failed to write number of map entries: %v", err)
}
for key, value := range data {
if err := binary.Write(bw, binary.BigEndian, int32(key)); err != nil {
log.Fatalf("Failed to write map key: %v", err)
}
length := uint32(len(value))
if err := binary.Write(bw, binary.BigEndian, length); err != nil {
log.Fatalf("Failed to write value length: %v", err)
}
if _, err := bw.Write(value); err != nil {
log.Fatalf("Failed to write value data: %v", err)
}
}
if err := bw.Flush(); err != nil {
log.Fatalf("Failed to flush bytes-map buffer: %v", err)
}
return buf.Bytes()
}
func decodeBytesMap(r *bufio.Reader) map[int][]byte {
var numEntries uint32
if err := binary.Read(r, binary.BigEndian, &numEntries); err != nil {
log.Fatalf("Failed to read number of map entries: %v", err)
}
data := make(map[int][]byte, numEntries)
for i := uint32(0); i < numEntries; i++ {
var key int32
if err := binary.Read(r, binary.BigEndian, &key); err != nil {
log.Fatalf("Failed to read map key: %v", err)
}
var length uint32
if err := binary.Read(r, binary.BigEndian, &length); err != nil {
log.Fatalf("Failed to read value length: %v", err)
}
value := make([]byte, length)
bytesRead := 0
for bytesRead < int(length) {
n, err := r.Read(value[bytesRead:])
if err != nil {
log.Fatalf("Failed to read value data: %v", err)
}
bytesRead += n
}
data[int(key)] = value
}
return data
}
func encodeBytesSliceMap(data map[int][][]byte) []byte {
var buf bytes.Buffer
bw := bufio.NewWriter(&buf)
numEntries := uint32(len(data))
if err := binary.Write(bw, binary.BigEndian, numEntries); err != nil {
log.Fatalf("Failed to write number of map entries: %v", err)
}
for key, value := range data {
if err := binary.Write(bw, binary.BigEndian, int32(key)); err != nil {
log.Fatalf("Failed to write map key: %v", err)
}
numSlices := uint32(len(value))
if err := binary.Write(bw, binary.BigEndian, numSlices); err != nil {
log.Fatalf("Failed to write number of slices: %v", err)
}
for _, slice := range value {
length := uint32(len(slice))
if err := binary.Write(bw, binary.BigEndian, length); err != nil {
log.Fatalf("Failed to write slice length: %v", err)
}
if _, err := bw.Write(slice); err != nil {
log.Fatalf("Failed to write slice data: %v", err)
}
}
}
if err := bw.Flush(); err != nil {
log.Fatalf("Failed to flush bytes-slice map buffer: %v", err)
}
return buf.Bytes()
}
func decodeBytesSliceMap(r *bufio.Reader) map[int][][]byte {
var numEntries uint32
if err := binary.Read(r, binary.BigEndian, &numEntries); err != nil {
log.Fatalf("Failed to read number of map entries: %v", err)
}
data := make(map[int][][]byte, numEntries)
for i := uint32(0); i < numEntries; i++ {
var key int32
if err := binary.Read(r, binary.BigEndian, &key); err != nil {
log.Fatalf("Failed to read map key: %v", err)
}
var numSlices uint32
if err := binary.Read(r, binary.BigEndian, &numSlices); err != nil {
log.Fatalf("Failed to read number of slices: %v", err)
}
slices := make([][]byte, numSlices)
for j := uint32(0); j < numSlices; j++ {
var length uint32
if err := binary.Read(r, binary.BigEndian, &length); err != nil {
log.Fatalf("Failed to read slice length: %v", err)
}
slice := make([]byte, length)
bytesRead := 0
for bytesRead < int(length) {
n, err := r.Read(slice[bytesRead:])
if err != nil {
log.Fatalf("Failed to read slice data: %v", err)
}
bytesRead += n
}
slices[j] = slice
}
data[int(key)] = slices
}
return data
}
// ---------------------------------------------------------------------
// Topology helpers — the historical EstablishConnections path lived
// here. The new ZAP-backed equivalent takes a peer address map from
// the caller rather than hard-coding EC2 IPs.
// ---------------------------------------------------------------------
// EstablishConnectionsZAP wires all (partyID, otherID) pairs over the
// ZAP transport. peerAddrs[otherID] must be the "host:port" of the
// peer's listening socket. The lower-rank side dials, the higher-rank
// side listens — the historical Corona convention preserved.
func (comm *P2PComm) EstablishConnectionsZAP(partyID, totalParties int, peerAddrs map[int]string) error {
var wg sync.WaitGroup
errCh := make(chan error, totalParties)
for otherID := 0; otherID < totalParties; otherID++ {
if otherID == partyID {
continue
}
wg.Add(1)
go func(otherID int) {
defer wg.Done()
if partyID < otherID {
// We listen — peer will dial us. Pre-register so
// Send* knows the peer's NodeID.
comm.AcceptPeer(otherID)
} else {
addr, ok := peerAddrs[otherID]
if !ok {
errCh <- fmt.Errorf("corona: no address for peer rank %d", otherID)
return
}
if err := comm.Connect(otherID, addr); err != nil {
errCh <- err
}
}
}(otherID)
}
wg.Wait()
close(errCh)
for err := range errCh {
if err != nil {
return err
}
}
return nil
}
func (comm *P2PComm) SendVector(writer *bufio.Writer, dst int, msg structs.Vector[ring.Poly]) {
if _, err := msg.WriteTo(writer); err != nil {
log.Fatalf("Failed to write vector: %v", err)
}
if err := writer.Flush(); err != nil {
log.Fatalf("Failed to flush writer: %v", err)
}
}
func (comm *P2PComm) RecvVector(reader *bufio.Reader, src int, length int) structs.Vector[ring.Poly] {
vec := make(structs.Vector[ring.Poly], length)
if _, err := vec.ReadFrom(reader); err != nil {
log.Fatalf("Failed to read vector: %v", err)
}
return vec
}
func (comm *P2PComm) SendMatrix(writer *bufio.Writer, dst int, msg structs.Matrix[ring.Poly]) {
if _, err := msg.WriteTo(writer); err != nil {
log.Fatalf("Error sending matrix: %v", err)
}
if err := writer.Flush(); err != nil {
log.Fatalf("Failed to flush writer: %v", err)
}
}
func (comm *P2PComm) RecvMatrix(reader *bufio.Reader, src int, length int) structs.Matrix[ring.Poly] {
matrix := make(structs.Matrix[ring.Poly], length)
if _, err := matrix.ReadFrom(reader); err != nil {
log.Fatalf("Failed to read matrix: %v", err)
}
return matrix
}
func (comm *P2PComm) SendBytesSlice(writer *bufio.Writer, dst int, data [][]byte) {
numSlices := uint32(len(data))
if err := binary.Write(writer, binary.BigEndian, numSlices); err != nil {
log.Fatalf("Failed to write number of slices: %v", err)
}
for _, slice := range data {
length := uint32(len(slice))
if err := binary.Write(writer, binary.BigEndian, length); err != nil {
log.Fatalf("Failed to write slice length: %v", err)
}
for len(slice) > 0 {
n, err := writer.Write(slice)
if err != nil {
log.Fatalf("Failed to write slice data: %v", err)
}
slice = slice[n:]
}
}
if err := writer.Flush(); err != nil {
log.Fatalf("Failed to flush writer: %v", err)
}
}
func (comm *P2PComm) RecvBytesSlice(reader *bufio.Reader, src int) [][]byte {
var numSlices uint32
if err := binary.Read(reader, binary.BigEndian, &numSlices); err != nil {
log.Fatalf("Failed to read number of slices: %v", err)
}
data := make([][]byte, numSlices)
for i := uint32(0); i < numSlices; i++ {
var length uint32
if err := binary.Read(reader, binary.BigEndian, &length); err != nil {
log.Fatalf("Failed to read slice length: %v", err)
}
slice := make([]byte, length)
bytesRead := 0
for bytesRead < int(length) {
n, err := reader.Read(slice[bytesRead:])
if err != nil {
log.Fatalf("Failed to read slice data: %v", err)
}
bytesRead += n
}
data[i] = slice
}
return data
}
func (comm *P2PComm) SendBytesMap(writer *bufio.Writer, dst int, data map[int][]byte) {
numEntries := uint32(len(data))
if err := binary.Write(writer, binary.BigEndian, numEntries); err != nil {
log.Fatalf("Failed to write number of map entries: %v", err)
}
for key, value := range data {
if err := binary.Write(writer, binary.BigEndian, int32(key)); err != nil {
log.Fatalf("Failed to write map key: %v", err)
}
length := uint32(len(value))
if err := binary.Write(writer, binary.BigEndian, length); err != nil {
log.Fatalf("Failed to write value length: %v", err)
}
for len(value) > 0 {
n, err := writer.Write(value)
if err != nil {
log.Fatalf("Failed to write value data: %v", err)
}
value = value[n:]
}
}
if err := writer.Flush(); err != nil {
log.Fatalf("Failed to flush writer: %v", err)
}
}
func (comm *P2PComm) RecvBytesMap(reader *bufio.Reader, src int) map[int][]byte {
var numEntries uint32
if err := binary.Read(reader, binary.BigEndian, &numEntries); err != nil {
log.Fatalf("Failed to read number of map entries: %v", err)
}
data := make(map[int][]byte, numEntries)
for i := uint32(0); i < numEntries; i++ {
var key int32
if err := binary.Read(reader, binary.BigEndian, &key); err != nil {
log.Fatalf("Failed to read map key: %v", err)
}
var length uint32
if err := binary.Read(reader, binary.BigEndian, &length); err != nil {
log.Fatalf("Failed to read value length: %v", err)
}
value := make([]byte, length)
bytesRead := 0
for bytesRead < int(length) {
n, err := reader.Read(value[bytesRead:])
if err != nil {
log.Fatalf("Failed to read value data: %v", err)
}
bytesRead += n
}
data[int(key)] = value
}
return data
}
func (comm *P2PComm) SendBytesSliceMap(writer *bufio.Writer, dst int, data map[int][][]byte) {
numEntries := uint32(len(data))
if err := binary.Write(writer, binary.BigEndian, numEntries); err != nil {
log.Fatalf("Failed to write number of map entries: %v", err)
}
for key, value := range data {
if err := binary.Write(writer, binary.BigEndian, int32(key)); err != nil {
log.Fatalf("Failed to write map key: %v", err)
}
numSlices := uint32(len(value))
if err := binary.Write(writer, binary.BigEndian, numSlices); err != nil {
log.Fatalf("Failed to write number of slices: %v", err)
}
for _, slice := range value {
length := uint32(len(slice))
if err := binary.Write(writer, binary.BigEndian, length); err != nil {
log.Fatalf("Failed to write slice length: %v", err)
}
for len(slice) > 0 {
n, err := writer.Write(slice)
if err != nil {
log.Fatalf("Failed to write slice data: %v", err)
}
slice = slice[n:]
}
}
}
if err := writer.Flush(); err != nil {
log.Fatalf("Failed to flush writer: %v", err)
}
}
func (comm *P2PComm) RecvBytesSliceMap(reader *bufio.Reader, src int) map[int][][]byte {
var numEntries uint32
if err := binary.Read(reader, binary.BigEndian, &numEntries); err != nil {
log.Fatalf("Failed to read number of map entries: %v", err)
}
data := make(map[int][][]byte, numEntries)
for i := uint32(0); i < numEntries; i++ {
var key int32
if err := binary.Read(reader, binary.BigEndian, &key); err != nil {
log.Fatalf("Failed to read map key: %v", err)
}
var numSlices uint32
if err := binary.Read(reader, binary.BigEndian, &numSlices); err != nil {
log.Fatalf("Failed to read number of slices: %v", err)
}
slices := make([][]byte, numSlices)
for j := uint32(0); j < numSlices; j++ {
var length uint32
if err := binary.Read(reader, binary.BigEndian, &length); err != nil {
log.Fatalf("Failed to read slice length: %v", err)
}
slice := make([]byte, length)
bytesRead := 0
for bytesRead < int(length) {
n, err := reader.Read(slice[bytesRead:])
if err != nil {
log.Fatalf("Failed to read slice data: %v", err)
}
bytesRead += n
}
slices[j] = slice
}
data[int(key)] = slices
}
return data
}
func ListenTCP(comm *P2PComm, port string, src int) {
l, err := net.Listen("tcp", "0.0.0.0:"+port)
if err != nil {
log.Fatal(err)
return
}
defer l.Close()
for {
conn, err := l.Accept()
if err != nil {
log.Println(err)
continue
}
comm.SetSock(src, &conn)
break
}
}
func DialTCP(comm *P2PComm, dst int, address string) {
log.Println("Trying to dial", dst, "at address", address)
// Check if the socket for the destination is already initialized
sock := comm.GetSock(dst)
if sock != nil && *sock != nil {
log.Printf("Socket for destination %d is already initialized.", dst)
return
}
var conn net.Conn
var err error
for i := 0; i < 10; i++ {
conn, err = net.Dial("tcp", address)
if err == nil {
break
}
log.Printf("Failed to dial TCP to %s: %v, retrying...", address, err)
time.Sleep(1 * time.Second)
}
if err != nil {
log.Fatalf("Failed to dial TCP to %s after retries: %v", address, err)
return
}
comm.SetSock(dst, &conn)
log.Println("Successfully connected to", address)
}
func EstablishConnections(wg *sync.WaitGroup, comm *P2PComm, partyID int, totalParties int) {
defer wg.Done()
var localWg sync.WaitGroup
// List of IP addresses of all parties
ips := []string{"50.18.79.144", "34.226.247.5", "18.199.237.28", "34.244.57.20", "54.248.28.234", "47.129.58.89", "54.232.160.71", "3.107.98.51"}
for otherID := 0; otherID < totalParties; otherID++ {
if partyID != otherID {
localWg.Add(1)
go func(otherID int) {
defer localWg.Done()
port := 6000 + calculatePortOffset(partyID, otherID)
address := fmt.Sprintf("%s:%d", ips[otherID], port)
if partyID < otherID {
ListenTCP(comm, fmt.Sprintf("%d", port), otherID)
} else {
DialTCP(comm, otherID, address)
}
}(otherID)
}
}
localWg.Wait()
}
func calculatePortOffset(partyID, otherID int) int {
if partyID < otherID {
return partyID*100 + otherID
}
return otherID*100 + partyID
}
networking/zap.go
+289
View File
@@ -0,0 +1,289 @@
// Copyright (C) 2026, Lux Industries Inc. All rights reserved.
// See the file LICENSE for licensing terms.
// Package networking provides the Corona ceremony transport layer.
//
// Wire model. Every Corona send (vector, matrix, byte-slice, byte-map,
// byte-slice-map) is encoded to bytes via the lattice WriteTo/ReadFrom
// routines (vector.go and friends in luxfi/lattice/v7/utils/structs).
// Those bytes are byte-equal regardless of transport — that is the
// "semantic-level wire" Corona participants compute against.
//
// Pre-migration the bytes were length-prefixed and shipped raw over a
// bare TCP socket. Post-migration they ride inside a ZAP message body
// over zap.Node, which gives us:
//
// - One framed transport (no manual binary.BigEndian length-prefix
// loops, no zero-copy regressions on every recv).
// - Multi-transport (TCP today, QUIC tomorrow with an unchanged
// P2PComm API).
// - mDNS discovery and ConnectDirect for tests and topology-known
// deployments.
//
// One ZAP message == one logical Corona send. The receiver dequeues one
// message and parses the lattice object from its body. The lattice
// length tags inside the body are preserved verbatim.
package networking
import (
"context"
"errors"
"fmt"
"io"
"log/slog"
"sync"
"github.com/luxfi/zap"
)
// CoronaMsgType is the ZAP message-type slot used by Corona's ceremony
// transport. It lives inside this package as an internal kind-byte
// namespace — no LP allocation needed because the only sender and
// receiver are Corona networking peers talking to each other.
//
// ZAP encodes the message type in the high byte of the 16-bit Flags
// field (msgType == Flags >> 8), so the kind-byte range is 0..255.
// 0xCE reads "Corona Ceremony"; we own this slot inside the
// corona/networking package only.
//
// All Corona traffic uses this single type because the call sequence
// on each side determines round semantics (R1 share, R2 nonce-
// commitment, R3 signature share, abort) — exactly as it did over
// bare TCP.
const CoronaMsgType uint8 = 0xCE
// CoronaServiceType is the mDNS service tag for Corona ceremony peers.
// Tests and unit fixtures set NodeConfig.NoDiscovery=true and use
// ConnectDirect with an explicit address; production deployments can
// flip this on for zero-config discovery on a LAN.
const CoronaServiceType = "_corona._tcp"
// peerStream is the receive side of one Corona peer relationship.
// Send goes through zap.Node.Send; receive blocks on a FIFO of message
// bodies pushed by the ZAP handler.
type peerStream struct {
mu sync.Mutex
cond *sync.Cond
queue [][]byte
closed bool
peerID string // ZAP peer ID (string-form of integer party ID)
}
func newPeerStream() *peerStream {
ps := &peerStream{}
ps.cond = sync.NewCond(&ps.mu)
return ps
}
// push appends a message body to the queue and wakes one waiter.
func (ps *peerStream) push(body []byte) {
ps.mu.Lock()
if ps.closed {
ps.mu.Unlock()
return
}
ps.queue = append(ps.queue, body)
ps.cond.Signal()
ps.mu.Unlock()
}
// pop dequeues the next message body, blocking if empty. Returns
// io.EOF if the stream is closed and the queue is drained.
func (ps *peerStream) pop() ([]byte, error) {
ps.mu.Lock()
defer ps.mu.Unlock()
for len(ps.queue) == 0 {
if ps.closed {
return nil, io.EOF
}
ps.cond.Wait()
}
body := ps.queue[0]
ps.queue = ps.queue[1:]
return body, nil
}
// close stops further pushes and wakes all waiters with io.EOF.
func (ps *peerStream) close() {
ps.mu.Lock()
ps.closed = true
ps.cond.Broadcast()
ps.mu.Unlock()
}
// zapNode is the runtime backing a P2PComm. One per local party.
type zapNode struct {
node *zap.Node
logger *slog.Logger
streams map[int]*peerStream // dst rank -> recv queue
peers map[int]string // rank -> zap NodeID
mu sync.RWMutex
}
// peerID encodes an integer party rank as a ZAP NodeID. Stable,
// deterministic, easy to reverse for handler dispatch.
func peerID(rank int) string {
return fmt.Sprintf("corona-%d", rank)
}
// rankFromPeerID is the inverse of peerID. Returns -1 if the peer
// is not a corona party (e.g. a stray mDNS neighbor on the LAN).
func rankFromPeerID(id string) int {
var rank int
if _, err := fmt.Sscanf(id, "corona-%d", &rank); err != nil {
return -1
}
return rank
}
// newZapNode creates a ZAP node bound to the given port. The caller is
// responsible for Start() and Stop().
func newZapNode(rank int, port int, noDiscovery bool, logger *slog.Logger) *zapNode {
if logger == nil {
logger = slog.Default()
}
zn := &zapNode{
logger: logger,
streams: make(map[int]*peerStream),
peers: make(map[int]string),
}
zn.node = zap.NewNode(zap.NodeConfig{
NodeID: peerID(rank),
ServiceType: CoronaServiceType,
Port: port,
NoDiscovery: noDiscovery,
Logger: logger,
})
zn.node.Handle(uint16(CoronaMsgType), zn.handleIncoming)
return zn
}
// getOrCreateStream returns the recv queue for src, creating one on
// first use. Caller must hold no locks; the function locks internally.
func (zn *zapNode) getOrCreateStream(src int) *peerStream {
zn.mu.RLock()
ps, ok := zn.streams[src]
zn.mu.RUnlock()
if ok {
return ps
}
zn.mu.Lock()
defer zn.mu.Unlock()
if ps, ok := zn.streams[src]; ok {
return ps
}
ps = newPeerStream()
ps.peerID = peerID(src)
zn.streams[src] = ps
return ps
}
// handleIncoming is the ZAP handler for Corona ceremony messages. It
// resolves the sender's rank from peerID and pushes the message body
// into that rank's receive queue.
//
// The handler returns (nil, nil): Corona uses one-way sends, not
// request/response correlated calls. The lattice encoding is
// self-delimiting inside the message body.
func (zn *zapNode) handleIncoming(_ context.Context, from string, msg *zap.Message) (*zap.Message, error) {
rank := rankFromPeerID(from)
if rank < 0 {
// Foreign peer — drop silently. Production should not see this
// because the service-type is Corona-specific; tests using
// ConnectDirect always wire the right node IDs.
return nil, nil
}
// Copy the body. The pooled bufRef behind msg.Bytes() will be
// released as soon as the handler returns, but the queue can hold
// the body until any time the receiver dequeues.
body := append([]byte(nil), msg.Bytes()...)
zn.getOrCreateStream(rank).push(body)
return nil, nil
}
// send wraps body in a ZAP message and ships it to the destination
// rank's peer via zap.Node.Send. The message-type lives in the top
// byte of Flags (Flags >> 8 == CoronaMsgType).
func (zn *zapNode) send(ctx context.Context, dst int, body []byte) error {
zn.mu.RLock()
peer, ok := zn.peers[dst]
zn.mu.RUnlock()
if !ok {
return fmt.Errorf("corona: no peer registered for rank %d", dst)
}
// Build a single-bytes-field ZAP message carrying the lattice
// payload verbatim. Flags top byte = CoronaMsgType for dispatch.
// We use SetBytes (variable-length tail) — the body fits there
// regardless of size, and on the receive side we read it back as
// a Bytes field with the same offset.
b := zap.NewBuilder(len(body) + zap.HeaderSize + 32)
obj := b.StartObject(8) // tail pointer: 4-byte offset + 4-byte length
obj.SetBytes(0, body)
obj.FinishAsRoot()
frame := b.FinishWithFlags(uint16(CoronaMsgType) << 8)
parsed, err := zap.Parse(frame)
if err != nil {
return fmt.Errorf("corona: build zap frame: %w", err)
}
return zn.node.Send(ctx, peer, parsed)
}
// recvBody dequeues the next message from src and returns the raw
// lattice-serialized body. Blocks until one arrives or the stream
// closes (io.EOF).
//
// One copy: the handler stashed the full ZAP frame in the queue; we
// parse it here and return a slice into that frame. Because the
// queue-owned frame outlives this call (it's freed when GC reclaims
// the queue entry), the returned slice is safe to hold.
func (zn *zapNode) recvBody(src int) ([]byte, error) {
frame, err := zn.getOrCreateStream(src).pop()
if err != nil {
return nil, err
}
msg, err := zap.Parse(frame)
if err != nil {
return nil, fmt.Errorf("corona: parse zap frame: %w", err)
}
body := msg.Root().Bytes(0)
if body == nil {
return nil, errors.New("corona: empty body in ZAP message")
}
return body, nil
}
// registerPeer binds a party rank to its ZAP peer ID. Called after a
// successful Connect or Listen handshake for the (rank, addr) pair.
func (zn *zapNode) registerPeer(rank int) {
zn.mu.Lock()
zn.peers[rank] = peerID(rank)
zn.mu.Unlock()
}
// close shuts down the ZAP node and all peer streams.
func (zn *zapNode) close() {
zn.mu.Lock()
for _, ps := range zn.streams {
ps.close()
}
zn.mu.Unlock()
if zn.node != nil {
zn.node.Stop()
}
}
// CalculatePortOffset is the historical convention used by Corona's
// EstablishConnections to derive a unique port per (partyA, partyB)
// pair. Kept for callers that still drive deployment topology with
// a hard-coded port grid.
//
// CalculatePortOffset(a, b) == CalculatePortOffset(b, a) by design —
// both sides of a pair need to agree on the same port without
// negotiating.
func CalculatePortOffset(partyID, otherID int) int {
if partyID < otherID {
return partyID*100 + otherID
}
return otherID*100 + partyID
}
networking/zap_test.go
+490
View File
@@ -0,0 +1,490 @@
// Copyright (C) 2026, Lux Industries Inc. All rights reserved.
// See the file LICENSE for licensing terms.
package networking
import (
"bufio"
"bytes"
"fmt"
"io"
"log/slog"
"net"
"sync"
"testing"
"time"
"github.com/luxfi/lattice/v7/ring"
"github.com/luxfi/lattice/v7/utils/sampling"
"github.com/luxfi/lattice/v7/utils/structs"
)
func bufioNewReader(r io.Reader) *bufio.Reader { return bufio.NewReader(r) }
func bufioNewWriter(w io.Writer) *bufio.Writer { return bufio.NewWriter(w) }
// newSilentLogger returns a slog.Logger that discards everything. Used
// in benchmarks so the per-iteration log overhead does not pollute
// timing numbers.
func newSilentLogger() *slog.Logger {
return slog.New(slog.NewTextHandler(io.Discard, &slog.HandlerOptions{Level: slog.LevelError + 1}))
}
// freePort grabs a free TCP port for the test by binding to :0,
// reading the assigned port, and closing — the kernel may reuse the
// port for the test listener microseconds later.
func freePort(t *testing.T) int {
t.Helper()
l, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
t.Fatalf("freePort: %v", err)
}
port := l.Addr().(*net.TCPAddr).Port
_ = l.Close()
return port
}
// twoPartyZAP brings up two P2PComm instances over ZAP and wires them
// to each other. Returns the two comms plus a cleanup func.
func twoPartyZAP(t *testing.T) (*P2PComm, *P2PComm, func()) {
t.Helper()
port0 := freePort(t)
port1 := freePort(t)
silent := newSilentLogger()
comm0, err := NewP2PComm(0, port0, true /*noDiscovery*/, silent)
if err != nil {
t.Fatalf("NewP2PComm 0: %v", err)
}
comm1, err := NewP2PComm(1, port1, true /*noDiscovery*/, silent)
if err != nil {
comm0.Close()
t.Fatalf("NewP2PComm 1: %v", err)
}
// Lower-rank side dials, higher-rank side accepts (Corona's
// historical convention preserved in EstablishConnectionsZAP).
addr1 := fmt.Sprintf("127.0.0.1:%d", port1)
comm0.AcceptPeer(1) // local registers 1's NodeID
if err := comm0.Connect(1, addr1); err != nil {
comm0.Close()
comm1.Close()
t.Fatalf("Connect 0->1: %v", err)
}
comm1.AcceptPeer(0)
// Allow the listener side a moment to register the inbound conn.
time.Sleep(100 * time.Millisecond)
cleanup := func() {
comm0.Close()
comm1.Close()
}
return comm0, comm1, cleanup
}
func TestZAP_SendRecvVector(t *testing.T) {
comm0, comm1, cleanup := twoPartyZAP(t)
defer cleanup()
r, _ := ring.NewRing(256, []uint64{8380417})
prng, _ := sampling.NewPRNG()
sampler := ring.NewUniformSampler(prng, r)
testVector := make(structs.Vector[ring.Poly], 3)
for i := range testVector {
testVector[i] = sampler.ReadNew()
}
var received structs.Vector[ring.Poly]
var wg sync.WaitGroup
wg.Add(1)
go func() {
defer wg.Done()
received = comm1.RecvVector(nil, 0, len(testVector))
}()
comm0.SendVector(nil, 1, testVector)
done := make(chan struct{})
go func() { wg.Wait(); close(done) }()
select {
case <-done:
case <-time.After(3 * time.Second):
t.Fatal("ZAP vector recv timed out")
}
if len(received) != len(testVector) {
t.Fatalf("len: got %d want %d", len(received), len(testVector))
}
for i := range testVector {
if !r.Equal(received[i], testVector[i]) {
t.Errorf("vec[%d] mismatch", i)
}
}
}
func TestZAP_SendRecvMatrix(t *testing.T) {
comm0, comm1, cleanup := twoPartyZAP(t)
defer cleanup()
r, _ := ring.NewRing(256, []uint64{8380417})
prng, _ := sampling.NewPRNG()
sampler := ring.NewUniformSampler(prng, r)
testMatrix := make(structs.Matrix[ring.Poly], 2)
for i := range testMatrix {
testMatrix[i] = make(structs.Vector[ring.Poly], 3)
for j := range testMatrix[i] {
testMatrix[i][j] = sampler.ReadNew()
}
}
var received structs.Matrix[ring.Poly]
var wg sync.WaitGroup
wg.Add(1)
go func() {
defer wg.Done()
received = comm1.RecvMatrix(nil, 0, len(testMatrix))
}()
comm0.SendMatrix(nil, 1, testMatrix)
done := make(chan struct{})
go func() { wg.Wait(); close(done) }()
select {
case <-done:
case <-time.After(3 * time.Second):
t.Fatal("ZAP matrix recv timed out")
}
if len(received) != len(testMatrix) {
t.Fatalf("rows: got %d want %d", len(received), len(testMatrix))
}
for i := range testMatrix {
if len(received[i]) != len(testMatrix[i]) {
t.Errorf("row %d cols: got %d want %d", i, len(received[i]), len(testMatrix[i]))
}
for j := range testMatrix[i] {
if !r.Equal(received[i][j], testMatrix[i][j]) {
t.Errorf("matrix[%d][%d] mismatch", i, j)
}
}
}
}
func TestZAP_SendRecvBytesSlice(t *testing.T) {
comm0, comm1, cleanup := twoPartyZAP(t)
defer cleanup()
in := [][]byte{
[]byte("round 1 share"),
[]byte("round 2 nonce-commitment"),
[]byte("round 3 signature share"),
}
var out [][]byte
var wg sync.WaitGroup
wg.Add(1)
go func() {
defer wg.Done()
out = comm1.RecvBytesSlice(nil, 0)
}()
comm0.SendBytesSlice(nil, 1, in)
done := make(chan struct{})
go func() { wg.Wait(); close(done) }()
select {
case <-done:
case <-time.After(3 * time.Second):
t.Fatal("ZAP bytes-slice recv timed out")
}
if len(out) != len(in) {
t.Fatalf("len: got %d want %d", len(out), len(in))
}
for i := range in {
if !bytes.Equal(out[i], in[i]) {
t.Errorf("slice[%d]: got %q want %q", i, out[i], in[i])
}
}
}
func TestZAP_SendRecvBytesMap(t *testing.T) {
comm0, comm1, cleanup := twoPartyZAP(t)
defer cleanup()
in := map[int][]byte{
0: []byte("party-0-share"),
1: []byte("party-1-share"),
2: []byte("party-2-share"),
}
var out map[int][]byte
var wg sync.WaitGroup
wg.Add(1)
go func() {
defer wg.Done()
out = comm1.RecvBytesMap(nil, 0)
}()
comm0.SendBytesMap(nil, 1, in)
done := make(chan struct{})
go func() { wg.Wait(); close(done) }()
select {
case <-done:
case <-time.After(3 * time.Second):
t.Fatal("ZAP bytes-map recv timed out")
}
if len(out) != len(in) {
t.Fatalf("len: got %d want %d", len(out), len(in))
}
for k, v := range in {
got, ok := out[k]
if !ok {
t.Errorf("missing key %d", k)
continue
}
if !bytes.Equal(got, v) {
t.Errorf("key %d: got %q want %q", k, got, v)
}
}
}
func TestZAP_SendRecvBytesSliceMap(t *testing.T) {
comm0, comm1, cleanup := twoPartyZAP(t)
defer cleanup()
in := map[int][][]byte{
0: {[]byte("seed-0-a"), []byte("seed-0-b")},
1: {[]byte("seed-1-a"), []byte("seed-1-b"), []byte("seed-1-c")},
}
var out map[int][][]byte
var wg sync.WaitGroup
wg.Add(1)
go func() {
defer wg.Done()
out = comm1.RecvBytesSliceMap(nil, 0)
}()
comm0.SendBytesSliceMap(nil, 1, in)
done := make(chan struct{})
go func() { wg.Wait(); close(done) }()
select {
case <-done:
case <-time.After(3 * time.Second):
t.Fatal("ZAP bytes-slice-map recv timed out")
}
if len(out) != len(in) {
t.Fatalf("len: got %d want %d", len(out), len(in))
}
for k, vs := range in {
gotVs, ok := out[k]
if !ok {
t.Errorf("missing key %d", k)
continue
}
if len(gotVs) != len(vs) {
t.Errorf("key %d: len got %d want %d", k, len(gotVs), len(vs))
continue
}
for i := range vs {
if !bytes.Equal(gotVs[i], vs[i]) {
t.Errorf("key %d slice %d: got %q want %q", k, i, gotVs[i], vs[i])
}
}
}
}
// TestZAP_RoundtripPreservesLatticeBytes asserts the byte-equal-on-
// semantic-wire invariant: the bytes Corona participants compute
// against (the lattice serialization) are identical to what a direct
// in-memory WriteTo/ReadFrom would produce. The ZAP framing is purely
// envelope.
func TestZAP_RoundtripPreservesLatticeBytes(t *testing.T) {
r, _ := ring.NewRing(256, []uint64{8380417})
prng, _ := sampling.NewPRNG()
sampler := ring.NewUniformSampler(prng, r)
v := make(structs.Vector[ring.Poly], 4)
for i := range v {
v[i] = sampler.ReadNew()
}
// Expected bytes: pure lattice serialization.
var expected bytes.Buffer
if _, err := v.WriteTo(&expected); err != nil {
t.Fatalf("expected WriteTo: %v", err)
}
comm0, comm1, cleanup := twoPartyZAP(t)
defer cleanup()
var received structs.Vector[ring.Poly]
var wg sync.WaitGroup
wg.Add(1)
go func() {
defer wg.Done()
received = comm1.RecvVector(nil, 0, len(v))
}()
comm0.SendVector(nil, 1, v)
done := make(chan struct{})
go func() { wg.Wait(); close(done) }()
select {
case <-done:
case <-time.After(3 * time.Second):
t.Fatal("ZAP roundtrip recv timed out")
}
// The received vector must serialize back to byte-equal output.
var actual bytes.Buffer
if _, err := received.WriteTo(&actual); err != nil {
t.Fatalf("actual WriteTo: %v", err)
}
if !bytes.Equal(expected.Bytes(), actual.Bytes()) {
t.Fatalf("lattice serialization differs across ZAP transport (len exp=%d got=%d)", expected.Len(), actual.Len())
}
}
// BenchmarkSendRecvVector_ZAP measures the round-trip cost of the ZAP
// transport for a representative Corona payload.
func BenchmarkSendRecvVector_ZAP(b *testing.B) {
port0, port1 := pickPort(b), pickPort(b)
silent := newSilentLogger()
comm0, err := NewP2PComm(0, port0, true, silent)
if err != nil {
b.Fatalf("NewP2PComm 0: %v", err)
}
defer comm0.Close()
comm1, err := NewP2PComm(1, port1, true, silent)
if err != nil {
b.Fatalf("NewP2PComm 1: %v", err)
}
defer comm1.Close()
if err := comm0.Connect(1, fmt.Sprintf("127.0.0.1:%d", port1)); err != nil {
b.Fatalf("connect: %v", err)
}
comm1.AcceptPeer(0)
time.Sleep(100 * time.Millisecond)
r, _ := ring.NewRing(256, []uint64{8380417})
prng, _ := sampling.NewPRNG()
sampler := ring.NewUniformSampler(prng, r)
v := make(structs.Vector[ring.Poly], 8)
for i := range v {
v[i] = sampler.ReadNew()
}
b.ResetTimer()
for i := 0; i < b.N; i++ {
done := make(chan struct{})
go func() {
_ = comm1.RecvVector(nil, 0, len(v))
close(done)
}()
comm0.SendVector(nil, 1, v)
<-done
}
}
func pickPort(b *testing.B) int {
b.Helper()
l, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
b.Fatalf("pickPort: %v", err)
}
p := l.Addr().(*net.TCPAddr).Port
_ = l.Close()
return p
}
// BenchmarkSendRecvVector_BareTCPPipe is the cheap synchronous in-
// memory baseline. net.Pipe has no TCP-stack overhead — it just hands
// the bytes between goroutines through a mutex. Useful as a floor on
// the lattice-serialization cost alone.
func BenchmarkSendRecvVector_BareTCPPipe(b *testing.B) {
server, client := net.Pipe()
defer server.Close()
defer client.Close()
comm0 := &P2PComm{Rank: 0, Socks: map[int]*net.Conn{1: &client}}
comm1 := &P2PComm{Rank: 1, Socks: map[int]*net.Conn{0: &server}}
r, _ := ring.NewRing(256, []uint64{8380417})
prng, _ := sampling.NewPRNG()
sampler := ring.NewUniformSampler(prng, r)
v := make(structs.Vector[ring.Poly], 8)
for i := range v {
v[i] = sampler.ReadNew()
}
b.ResetTimer()
for i := 0; i < b.N; i++ {
done := make(chan struct{})
go func() {
reader := bufioNewReader(server)
_ = comm1.RecvVector(reader, 0, len(v))
close(done)
}()
writer := bufioNewWriter(client)
comm0.SendVector(writer, 1, v)
<-done
}
}
// BenchmarkSendRecvVector_BareTCPLoopback is the apples-to-apples
// baseline for the ZAP bench: real TCP on the loopback interface
// using the pre-migration manual-length-prefix code path
// (SendVector via writer.Flush over a real net.TCPConn).
func BenchmarkSendRecvVector_BareTCPLoopback(b *testing.B) {
ln, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
b.Fatalf("listen: %v", err)
}
defer ln.Close()
addr := ln.Addr().String()
connCh := make(chan net.Conn, 1)
go func() {
c, err := ln.Accept()
if err != nil {
return
}
connCh <- c
}()
clientConn, err := net.Dial("tcp", addr)
if err != nil {
b.Fatalf("dial: %v", err)
}
defer clientConn.Close()
serverConn := <-connCh
defer serverConn.Close()
var client net.Conn = clientConn
var server net.Conn = serverConn
comm0 := &P2PComm{Rank: 0, Socks: map[int]*net.Conn{1: &client}}
comm1 := &P2PComm{Rank: 1, Socks: map[int]*net.Conn{0: &server}}
r, _ := ring.NewRing(256, []uint64{8380417})
prng, _ := sampling.NewPRNG()
sampler := ring.NewUniformSampler(prng, r)
v := make(structs.Vector[ring.Poly], 8)
for i := range v {
v[i] = sampler.ReadNew()
}
reader := bufioNewReader(server)
writer := bufioNewWriter(client)
b.ResetTimer()
for i := 0; i < b.N; i++ {
done := make(chan struct{})
go func() {
_ = comm1.RecvVector(reader, 0, len(v))
close(done)
}()
comm0.SendVector(writer, 1, v)
<-done
}
}