From ae888c8edce2bbc1a68610b5196305021c3673e2 Mon Sep 17 00:00:00 2001 From: Sanjit Bhat Date: Sun, 20 Oct 2024 14:28:47 -0400 Subject: [PATCH 1/8] rename timeseries to hist --- kt/history.go | 21 +++++++++++++++++++++ kt/test.go | 8 ++++---- kt/timeseries.go | 21 --------------------- 3 files changed, 25 insertions(+), 25 deletions(-) create mode 100644 kt/history.go delete mode 100644 kt/timeseries.go diff --git a/kt/history.go b/kt/history.go new file mode 100644 index 0000000..d8a7998 --- /dev/null +++ b/kt/history.go @@ -0,0 +1,21 @@ +package kt + +type HistEntry struct { + Epoch uint64 + HistVal []byte +} + +// GetHist searches hist at the epoch and rets the latest val, or false +// if there's no registered val. +func GetHist(o []*HistEntry, epoch uint64) (bool, []byte) { + var isReg bool + var val []byte + // entries inv: ordered by epoch field. + for _, e := range o { + if e.Epoch <= epoch { + isReg = true + val = e.HistVal + } + } + return isReg, val +} diff --git a/kt/test.go b/kt/test.go index 3cecc47..62a9a21 100644 --- a/kt/test.go +++ b/kt/test.go @@ -68,7 +68,7 @@ func testAll(servAddr, adtr0Addr, adtr1Addr uint64) { primitive.Sleep(1000_000_000) selfMonEp, err0 := aliceCli.SelfMon() primitive.Assume(!err0.err) - // could also state this as bob.epoch <= last epoch in TS. + // could also state this as bob.epoch <= last epoch in history. primitive.Assume(bob.epoch <= selfMonEp) err1 := aliceCli.Audit(adtr0Addr, adtr0Pk) primitive.Assume(!err1.err) @@ -80,7 +80,7 @@ func testAll(servAddr, adtr0Addr, adtr1Addr uint64) { primitive.Assume(!err4.err) // final check. bob got the right key. - isReg, aliceKey := GetTimeSeries(alice.pks, bob.epoch) + isReg, aliceKey := GetHist(alice.hist, bob.epoch) primitive.Assert(isReg == bob.isReg) if isReg { primitive.Assert(std.BytesEqual(aliceKey, bob.alicePk)) @@ -88,7 +88,7 @@ func testAll(servAddr, adtr0Addr, adtr1Addr uint64) { } type alice struct { - pks []*TimeSeriesEntry + hist []*HistEntry } func (a *alice) run(cli *Client) { @@ -97,7 +97,7 @@ func (a *alice) run(cli *Client) { pk := []byte{i} epoch, err0 := cli.Put(pk) primitive.Assume(!err0.err) - a.pks = append(a.pks, &TimeSeriesEntry{Epoch: epoch, TSVal: pk}) + a.hist = append(a.hist, &HistEntry{Epoch: epoch, HistVal: pk}) } } diff --git a/kt/timeseries.go b/kt/timeseries.go deleted file mode 100644 index d6fa312..0000000 --- a/kt/timeseries.go +++ /dev/null @@ -1,21 +0,0 @@ -package kt - -// TODO: rename to history. -type TimeSeriesEntry struct { - Epoch uint64 - TSVal []byte -} - -// GetTimeSeries rets whether a val is registered at the time and, if so, the val. -func GetTimeSeries(o []*TimeSeriesEntry, epoch uint64) (bool, []byte) { - var isReg bool - var val []byte - // entries inv: ordered by epoch field. - for _, e := range o { - if e.Epoch <= epoch { - isReg = true - val = e.TSVal - } - } - return isReg, val -} From 046bea672bb1d3e069b7d01e8193025a221fc8fe Mon Sep 17 00:00:00 2001 From: Sanjit Bhat Date: Sun, 20 Oct 2024 15:12:31 -0400 Subject: [PATCH 2/8] generalize setup for full test --- kt/kt_test.go | 7 +++-- kt/test.go | 71 ++++++++++++++++++++------------------------------- 2 files changed, 30 insertions(+), 48 deletions(-) diff --git a/kt/kt_test.go b/kt/kt_test.go index f1c3aaa..912ef06 100644 --- a/kt/kt_test.go +++ b/kt/kt_test.go @@ -6,10 +6,9 @@ import ( ) func TestAll(t *testing.T) { - serverAddr := makeUniqueAddr() - adtr0Addr := makeUniqueAddr() - adtr1Addr := makeUniqueAddr() - testAll(serverAddr, adtr0Addr, adtr1Addr) + servAddr := makeUniqueAddr() + adtrAddrs := []uint64{makeUniqueAddr(), makeUniqueAddr()} + testAllFull(servAddr, adtrAddrs) } func TestBasic(t *testing.T) { diff --git a/kt/test.go b/kt/test.go index 62a9a21..0389ba2 100644 --- a/kt/test.go +++ b/kt/test.go @@ -10,7 +10,6 @@ import ( "github.com/goose-lang/primitive" "github.com/goose-lang/std" "github.com/mit-pdos/pav/advrpc" - "github.com/mit-pdos/pav/cryptoffi" "sync" ) @@ -20,33 +19,25 @@ const ( charlieUid uint64 = 2 ) -func testAll(servAddr, adtr0Addr, adtr1Addr uint64) { - // start server and auditors. - serv, servSigPk, servVrfPk := newServer() - servRpc := newRpcServer(serv) - servRpc.Serve(servAddr) - adtr0, adtr0Pk := newAuditor(servSigPk) - adtr0Rpc := newRpcAuditor(adtr0) - adtr0Rpc.Serve(adtr0Addr) - adtr1, adtr1Pk := newAuditor(servSigPk) - adtr1Rpc := newRpcAuditor(adtr1) - adtr1Rpc.Serve(adtr1Addr) - primitive.Sleep(1_000_000) +func testAllFull(servAddr uint64, adtrAddrs []uint64) { + testAll(setup(servAddr, adtrAddrs)) +} +func testAll(setup *setupParams) { // run background threads. go func() { - charlie := newClient(charlieUid, servAddr, servSigPk, servVrfPk) - chaos(charlie, adtr0Addr, adtr1Addr, adtr0Pk, adtr1Pk) + charlie := newClient(charlieUid, setup.servAddr, setup.servSigPk, setup.servVrfPk) + chaos(charlie) }() go func() { - syncAdtr(servAddr, adtr0Addr, adtr1Addr) + syncAdtrs(setup.servAddr, setup.adtrAddrs) }() // run alice and bob. alice := &alice{} aliceMu := new(sync.Mutex) aliceMu.Lock() - aliceCli := newClient(aliceUid, servAddr, servSigPk, servVrfPk) + aliceCli := newClient(aliceUid, setup.servAddr, setup.servSigPk, setup.servVrfPk) go func() { alice.run(aliceCli) aliceMu.Unlock() @@ -54,7 +45,7 @@ func testAll(servAddr, adtr0Addr, adtr1Addr uint64) { bob := &bob{} bobMu := new(sync.Mutex) bobMu.Lock() - bobCli := newClient(bobUid, servAddr, servSigPk, servVrfPk) + bobCli := newClient(bobUid, setup.servAddr, setup.servSigPk, setup.servVrfPk) go func() { bob.run(bobCli) bobMu.Unlock() @@ -64,20 +55,18 @@ func testAll(servAddr, adtr0Addr, adtr1Addr uint64) { aliceMu.Lock() bobMu.Lock() - // alice SelfMon + Audit. bob Audit. ordering irrelevant across clients. - primitive.Sleep(1000_000_000) + // alice self monitor. in real world, she'll come on-line at times and do this. selfMonEp, err0 := aliceCli.SelfMon() primitive.Assume(!err0.err) - // could also state this as bob.epoch <= last epoch in history. + // this last self monitor will be our history bound. primitive.Assume(bob.epoch <= selfMonEp) - err1 := aliceCli.Audit(adtr0Addr, adtr0Pk) - primitive.Assume(!err1.err) - err2 := aliceCli.Audit(adtr1Addr, adtr1Pk) - primitive.Assume(!err2.err) - err3 := bobCli.Audit(adtr0Addr, adtr0Pk) - primitive.Assume(!err3.err) - err4 := bobCli.Audit(adtr1Addr, adtr1Pk) - primitive.Assume(!err4.err) + + // wait for auditors to catch all updates. + primitive.Sleep(1000_000_000) + + // alice and bob Audit. ordering irrelevant across clients. + doAudits(aliceCli, setup.adtrAddrs, setup.adtrPks) + doAudits(bobCli, setup.adtrAddrs, setup.adtrPks) // final check. bob got the right key. isReg, aliceKey := GetHist(alice.hist, bob.epoch) @@ -92,9 +81,9 @@ type alice struct { } func (a *alice) run(cli *Client) { - for i := byte(0); i < byte(20); i++ { + for i := uint64(0); i < uint64(20); i++ { primitive.Sleep(50_000_000) - pk := []byte{i} + pk := []byte{byte(i)} epoch, err0 := cli.Put(pk) primitive.Assume(!err0.err) a.hist = append(a.hist, &HistEntry{Epoch: epoch, HistVal: pk}) @@ -116,8 +105,8 @@ func (b *bob) run(cli *Client) { b.alicePk = pk } -// chaos from Charlie running all the ops. -func chaos(charlie *Client, adtr0Addr, adtr1Addr uint64, adtr0Pk, adtr1Pk cryptoffi.PublicKey) { +// chaos from Charlie running ops. +func chaos(charlie *Client) { for { primitive.Sleep(40_000_000) pk := []byte{2} @@ -127,26 +116,20 @@ func chaos(charlie *Client, adtr0Addr, adtr1Addr uint64, adtr0Pk, adtr1Pk crypto primitive.Assume(!err1.err) _, err2 := charlie.SelfMon() primitive.Assume(!err2.err) - charlie.Audit(adtr0Addr, adtr0Pk) - charlie.Audit(adtr1Addr, adtr1Pk) } } -func syncAdtr(servAddr, adtr0Addr, adtr1Addr uint64) { +func syncAdtrs(servAddr uint64, adtrAddrs []uint64) { servCli := advrpc.Dial(servAddr) - adtr0Cli := advrpc.Dial(adtr0Addr) - adtr1Cli := advrpc.Dial(adtr1Addr) + adtrs := mkRpcClients(adtrAddrs) var epoch uint64 for { primitive.Sleep(1_000_000) - upd, err0 := callServAudit(servCli, epoch) - if err0 { + upd, err := callServAudit(servCli, epoch) + if err { continue } - err1 := callAdtrUpdate(adtr0Cli, upd) - primitive.Assume(!err1) - err2 := callAdtrUpdate(adtr1Cli, upd) - primitive.Assume(!err2) + updAdtrs(upd, adtrs) epoch++ } } From 95f5b33397fd1c663e710925ce0f673abe970b22 Mon Sep 17 00:00:00 2001 From: Sanjit Bhat Date: Sun, 20 Oct 2024 16:46:15 -0400 Subject: [PATCH 3/8] put locked state inside struct, more intuitive --- kt/test.go | 45 ++++++++++++++++++++++++--------------------- 1 file changed, 24 insertions(+), 21 deletions(-) diff --git a/kt/test.go b/kt/test.go index 0389ba2..de0b688 100644 --- a/kt/test.go +++ b/kt/test.go @@ -33,30 +33,27 @@ func testAll(setup *setupParams) { syncAdtrs(setup.servAddr, setup.adtrAddrs) }() - // run alice and bob. - alice := &alice{} - aliceMu := new(sync.Mutex) - aliceMu.Lock() + // alice does a bunch of puts. aliceCli := newClient(aliceUid, setup.servAddr, setup.servSigPk, setup.servVrfPk) + alice := &alice{mu: new(sync.Mutex), cli: aliceCli, hist: nil} + alice.mu.Lock() go func() { - alice.run(aliceCli) - aliceMu.Unlock() + alice.run() }() - bob := &bob{} - bobMu := new(sync.Mutex) - bobMu.Lock() + // bob does a get. bobCli := newClient(bobUid, setup.servAddr, setup.servSigPk, setup.servVrfPk) + bob := &bob{mu: new(sync.Mutex), cli: bobCli, epoch: 0, isReg: false, alicePk: nil} + bob.mu.Lock() go func() { - bob.run(bobCli) - bobMu.Unlock() + bob.run() }() // wait for alice and bob to finish. - aliceMu.Lock() - bobMu.Lock() + alice.mu.Lock() + bob.mu.Lock() // alice self monitor. in real world, she'll come on-line at times and do this. - selfMonEp, err0 := aliceCli.SelfMon() + selfMonEp, err0 := alice.cli.SelfMon() primitive.Assume(!err0.err) // this last self monitor will be our history bound. primitive.Assume(bob.epoch <= selfMonEp) @@ -64,9 +61,9 @@ func testAll(setup *setupParams) { // wait for auditors to catch all updates. primitive.Sleep(1000_000_000) - // alice and bob Audit. ordering irrelevant across clients. - doAudits(aliceCli, setup.adtrAddrs, setup.adtrPks) - doAudits(bobCli, setup.adtrAddrs, setup.adtrPks) + // alice and bob audit. ordering irrelevant across clients. + doAudits(alice.cli, setup.adtrAddrs, setup.adtrPks) + doAudits(bob.cli, setup.adtrAddrs, setup.adtrPks) // final check. bob got the right key. isReg, aliceKey := GetHist(alice.hist, bob.epoch) @@ -77,32 +74,38 @@ func testAll(setup *setupParams) { } type alice struct { + mu *sync.Mutex + cli *Client hist []*HistEntry } -func (a *alice) run(cli *Client) { +func (a *alice) run() { for i := uint64(0); i < uint64(20); i++ { primitive.Sleep(50_000_000) pk := []byte{byte(i)} - epoch, err0 := cli.Put(pk) + epoch, err0 := a.cli.Put(pk) primitive.Assume(!err0.err) a.hist = append(a.hist, &HistEntry{Epoch: epoch, HistVal: pk}) } + a.mu.Unlock() } type bob struct { + mu *sync.Mutex + cli *Client epoch uint64 isReg bool alicePk []byte } -func (b *bob) run(cli *Client) { +func (b *bob) run() { primitive.Sleep(550_000_000) - isReg, pk, epoch, err0 := cli.Get(aliceUid) + isReg, pk, epoch, err0 := b.cli.Get(aliceUid) primitive.Assume(!err0.err) b.epoch = epoch b.isReg = isReg b.alicePk = pk + b.mu.Unlock() } // chaos from Charlie running ops. From e01df9f9cd77f50447b0ab21653c59b985e44a31 Mon Sep 17 00:00:00 2001 From: Sanjit Bhat Date: Tue, 22 Oct 2024 10:19:49 -0400 Subject: [PATCH 4/8] mu init change --- kt/test.go | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/kt/test.go b/kt/test.go index de0b688..992a085 100644 --- a/kt/test.go +++ b/kt/test.go @@ -35,14 +35,17 @@ func testAll(setup *setupParams) { // alice does a bunch of puts. aliceCli := newClient(aliceUid, setup.servAddr, setup.servSigPk, setup.servVrfPk) - alice := &alice{mu: new(sync.Mutex), cli: aliceCli, hist: nil} + alice := &alice{cli: aliceCli} + // TODO: if this works, change the other ones as well. + alice.mu = new(sync.Mutex) alice.mu.Lock() go func() { alice.run() }() // bob does a get. bobCli := newClient(bobUid, setup.servAddr, setup.servSigPk, setup.servVrfPk) - bob := &bob{mu: new(sync.Mutex), cli: bobCli, epoch: 0, isReg: false, alicePk: nil} + bob := &bob{cli: bobCli} + bob.mu = new(sync.Mutex) bob.mu.Lock() go func() { bob.run() From db5817a4bba449af66fcef9e387792144134c2f8 Mon Sep 17 00:00:00 2001 From: Sanjit Bhat Date: Tue, 22 Oct 2024 16:17:13 -0400 Subject: [PATCH 5/8] too many test case failures with background auditor syncing. move to synchronous setup --- kt/basictest.go | 6 +++--- kt/test.go | 53 ++++++++++--------------------------------------- 2 files changed, 14 insertions(+), 45 deletions(-) diff --git a/kt/basictest.go b/kt/basictest.go index a3a8382..159eb65 100644 --- a/kt/basictest.go +++ b/kt/basictest.go @@ -52,8 +52,8 @@ func testBasic(setup *setupParams) { primitive.Assume(!err2) adtrs := mkRpcClients(setup.adtrAddrs) - updAdtrs(upd0, adtrs) - updAdtrs(upd1, adtrs) + updAdtrsOnce(upd0, adtrs) + updAdtrsOnce(upd1, adtrs) // bob get. bob := newClient(bobUid, setup.servAddr, setup.servSigPk, setup.servVrfPk) @@ -80,7 +80,7 @@ func mkRpcClients(addrs []uint64) []*advrpc.Client { return c } -func updAdtrs(upd *UpdateProof, adtrs []*advrpc.Client) { +func updAdtrsOnce(upd *UpdateProof, adtrs []*advrpc.Client) { for _, cli := range adtrs { err := callAdtrUpdate(cli, upd) primitive.Assume(!err) diff --git a/kt/test.go b/kt/test.go index 992a085..358af27 100644 --- a/kt/test.go +++ b/kt/test.go @@ -1,11 +1,5 @@ package kt -// set global timing such that: -// - chaos interlaces enough with alice. -// - chaos mostly has up-to-date audits. -// - bob queries somewhere around halfway thru alice's puts. -// - before alice and bob finally check keys, the auditor has caught up. - import ( "github.com/goose-lang/primitive" "github.com/goose-lang/std" @@ -14,9 +8,8 @@ import ( ) const ( - aliceUid uint64 = 0 - bobUid uint64 = 1 - charlieUid uint64 = 2 + aliceUid uint64 = 0 + bobUid uint64 = 1 ) func testAllFull(servAddr uint64, adtrAddrs []uint64) { @@ -24,25 +17,16 @@ func testAllFull(servAddr uint64, adtrAddrs []uint64) { } func testAll(setup *setupParams) { - // run background threads. - go func() { - charlie := newClient(charlieUid, setup.servAddr, setup.servSigPk, setup.servVrfPk) - chaos(charlie) - }() - go func() { - syncAdtrs(setup.servAddr, setup.adtrAddrs) - }() - // alice does a bunch of puts. aliceCli := newClient(aliceUid, setup.servAddr, setup.servSigPk, setup.servVrfPk) alice := &alice{cli: aliceCli} - // TODO: if this works, change the other ones as well. alice.mu = new(sync.Mutex) alice.mu.Lock() go func() { alice.run() }() - // bob does a get. + + // bob does a get at some time in the middle of alice's puts. bobCli := newClient(bobUid, setup.servAddr, setup.servSigPk, setup.servVrfPk) bob := &bob{cli: bobCli} bob.mu = new(sync.Mutex) @@ -61,8 +45,8 @@ func testAll(setup *setupParams) { // this last self monitor will be our history bound. primitive.Assume(bob.epoch <= selfMonEp) - // wait for auditors to catch all updates. - primitive.Sleep(1000_000_000) + // sync auditors. in real world, this'll happen periodically. + updAdtrsAll(setup.servAddr, setup.adtrAddrs) // alice and bob audit. ordering irrelevant across clients. doAudits(alice.cli, setup.adtrAddrs, setup.adtrPks) @@ -84,7 +68,7 @@ type alice struct { func (a *alice) run() { for i := uint64(0); i < uint64(20); i++ { - primitive.Sleep(50_000_000) + primitive.Sleep(5_000_000) pk := []byte{byte(i)} epoch, err0 := a.cli.Put(pk) primitive.Assume(!err0.err) @@ -102,7 +86,7 @@ type bob struct { } func (b *bob) run() { - primitive.Sleep(550_000_000) + primitive.Sleep(120_000_000) isReg, pk, epoch, err0 := b.cli.Get(aliceUid) primitive.Assume(!err0.err) b.epoch = epoch @@ -111,31 +95,16 @@ func (b *bob) run() { b.mu.Unlock() } -// chaos from Charlie running ops. -func chaos(charlie *Client) { - for { - primitive.Sleep(40_000_000) - pk := []byte{2} - _, err0 := charlie.Put(pk) - primitive.Assume(!err0.err) - _, _, _, err1 := charlie.Get(aliceUid) - primitive.Assume(!err1.err) - _, err2 := charlie.SelfMon() - primitive.Assume(!err2.err) - } -} - -func syncAdtrs(servAddr uint64, adtrAddrs []uint64) { +func updAdtrsAll(servAddr uint64, adtrAddrs []uint64) { servCli := advrpc.Dial(servAddr) adtrs := mkRpcClients(adtrAddrs) var epoch uint64 for { - primitive.Sleep(1_000_000) upd, err := callServAudit(servCli, epoch) if err { - continue + break } - updAdtrs(upd, adtrs) + updAdtrsOnce(upd, adtrs) epoch++ } } From 4e004ae1c48ee24ed7d817df90f850314b39a017 Mon Sep 17 00:00:00 2001 From: Sanjit Bhat Date: Wed, 23 Oct 2024 09:45:58 -0400 Subject: [PATCH 6/8] switch back to waitgroup --- kt/test.go | 17 +++++++---------- 1 file changed, 7 insertions(+), 10 deletions(-) diff --git a/kt/test.go b/kt/test.go index 358af27..3793643 100644 --- a/kt/test.go +++ b/kt/test.go @@ -17,27 +17,28 @@ func testAllFull(servAddr uint64, adtrAddrs []uint64) { } func testAll(setup *setupParams) { + var wg sync.WaitGroup + wg.Add(1) + wg.Add(1) + // alice does a bunch of puts. aliceCli := newClient(aliceUid, setup.servAddr, setup.servSigPk, setup.servVrfPk) alice := &alice{cli: aliceCli} - alice.mu = new(sync.Mutex) - alice.mu.Lock() go func() { alice.run() + wg.Done() }() // bob does a get at some time in the middle of alice's puts. bobCli := newClient(bobUid, setup.servAddr, setup.servSigPk, setup.servVrfPk) bob := &bob{cli: bobCli} - bob.mu = new(sync.Mutex) - bob.mu.Lock() go func() { bob.run() + wg.Done() }() // wait for alice and bob to finish. - alice.mu.Lock() - bob.mu.Lock() + wg.Wait() // alice self monitor. in real world, she'll come on-line at times and do this. selfMonEp, err0 := alice.cli.SelfMon() @@ -61,7 +62,6 @@ func testAll(setup *setupParams) { } type alice struct { - mu *sync.Mutex cli *Client hist []*HistEntry } @@ -74,11 +74,9 @@ func (a *alice) run() { primitive.Assume(!err0.err) a.hist = append(a.hist, &HistEntry{Epoch: epoch, HistVal: pk}) } - a.mu.Unlock() } type bob struct { - mu *sync.Mutex cli *Client epoch uint64 isReg bool @@ -92,7 +90,6 @@ func (b *bob) run() { b.epoch = epoch b.isReg = isReg b.alicePk = pk - b.mu.Unlock() } func updAdtrsAll(servAddr uint64, adtrAddrs []uint64) { From 4981acbd69f3213a4925812a3c9c97065f52096a Mon Sep 17 00:00:00 2001 From: Sanjit Bhat Date: Wed, 23 Oct 2024 10:01:16 -0400 Subject: [PATCH 7/8] goose --- kt/test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/kt/test.go b/kt/test.go index 3793643..0b4a79e 100644 --- a/kt/test.go +++ b/kt/test.go @@ -17,7 +17,7 @@ func testAllFull(servAddr uint64, adtrAddrs []uint64) { } func testAll(setup *setupParams) { - var wg sync.WaitGroup + wg := new(sync.WaitGroup) wg.Add(1) wg.Add(1) From 4cbccb5ec97169d3cd4be947ee0d98fbcd0710a3 Mon Sep 17 00:00:00 2001 From: Sanjit Bhat Date: Wed, 23 Oct 2024 10:56:46 -0400 Subject: [PATCH 8/8] re-order cli makes to come before wg --- kt/test.go | 13 +++++-------- 1 file changed, 5 insertions(+), 8 deletions(-) diff --git a/kt/test.go b/kt/test.go index 0b4a79e..026cbf4 100644 --- a/kt/test.go +++ b/kt/test.go @@ -17,27 +17,24 @@ func testAllFull(servAddr uint64, adtrAddrs []uint64) { } func testAll(setup *setupParams) { + aliceCli := newClient(aliceUid, setup.servAddr, setup.servSigPk, setup.servVrfPk) + alice := &alice{cli: aliceCli} + bobCli := newClient(bobUid, setup.servAddr, setup.servSigPk, setup.servVrfPk) + bob := &bob{cli: bobCli} + wg := new(sync.WaitGroup) wg.Add(1) wg.Add(1) - // alice does a bunch of puts. - aliceCli := newClient(aliceUid, setup.servAddr, setup.servSigPk, setup.servVrfPk) - alice := &alice{cli: aliceCli} go func() { alice.run() wg.Done() }() - // bob does a get at some time in the middle of alice's puts. - bobCli := newClient(bobUid, setup.servAddr, setup.servSigPk, setup.servVrfPk) - bob := &bob{cli: bobCli} go func() { bob.run() wg.Done() }() - - // wait for alice and bob to finish. wg.Wait() // alice self monitor. in real world, she'll come on-line at times and do this.