Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions Makefile
Original file line number Diff line number Diff line change
Expand Up @@ -64,8 +64,8 @@ tools: download ## Install development tools

test: ## Run all tests with verbose output (logs to .log/test.log)
@mkdir -p .log
@echo "Running tests... (output in .log/test.log)"
@go test -v ./... > .log/test.log 2>&1 && echo "Tests passed! See .log/test.log for details" || (echo "Tests failed! See .log/test.log for details" && exit 1)
@echo "Running tests always (not cached)... (output in .log/test.log)"
@go test -count=1 -v ./... > .log/test.log 2>&1 && echo "Tests passed! See .log/test.log for details" || (echo "Tests failed! See .log/test.log for details" && exit 1)

short: ## Run short tests only
@go test -short ./...
Expand Down
12 changes: 7 additions & 5 deletions client/client.go
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,6 @@ import (

"github.com/relab/gorums"
"github.com/relab/hotstuff"
"github.com/relab/hotstuff/core"
"github.com/relab/hotstuff/core/eventloop"
"github.com/relab/hotstuff/core/logging"
"github.com/relab/hotstuff/internal/proto/clientpb"
Expand All @@ -25,6 +24,9 @@ import (
"google.golang.org/protobuf/types/known/emptypb"
)

// ID is the identifier for a client.
type ID uint32

type qspec struct {
faulty int
}
Expand Down Expand Up @@ -61,7 +63,7 @@ type Config struct {
type Client struct {
eventLoop *eventloop.EventLoop
logger logging.Logger
config *core.RuntimeConfig
id ID

mut sync.Mutex
mgr *clientpb.Manager
Expand All @@ -82,13 +84,13 @@ type Client struct {
func New(
eventLoop *eventloop.EventLoop,
logger logging.Logger,
config *core.RuntimeConfig,
id ID,
conf Config,
) (client *Client) {
client = &Client{
eventLoop: eventLoop,
logger: logger,
config: config,
id: id,

pendingCmds: make(chan pendingCmd, conf.MaxConcurrent),
highestCommitted: 1,
Expand Down Expand Up @@ -230,7 +232,7 @@ loop:
}

cmd := &clientpb.Command{
ClientID: uint32(c.config.ID()),
ClientID: uint32(c.id),
SequenceNumber: num,
Data: data[:n],
}
Expand Down
5 changes: 3 additions & 2 deletions internal/config/config.go
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@ import (
"time"

"github.com/relab/hotstuff"
"github.com/relab/hotstuff/client"
"github.com/relab/hotstuff/internal/proto/orchestrationpb"
"google.golang.org/protobuf/types/known/durationpb"
)
Expand Down Expand Up @@ -238,7 +239,7 @@ func (c *ExperimentConfig) lookupByzStrategy(replicaID hotstuff.ID) string {
// AssignClients assigns clients to hosts.
func (c *ExperimentConfig) AssignClients() ClientMap {
hostsToClients := make(ClientMap)
nextClientID := hotstuff.ID(1)
nextClientID := client.ID(1)

for hostIdx, host := range c.ClientHosts {
numClients := c.ClientsForHost(hostIdx)
Expand Down Expand Up @@ -320,7 +321,7 @@ func (r ReplicaMap) ReplicaIDs(host string) []uint32 {
}

// ClientMap maps from a host to a slice of client IDs.
type ClientMap map[string][]hotstuff.ID
type ClientMap map[string][]client.ID

// ClientIDs returns the IDs of the clients running on the given host.
func (c ClientMap) ClientIDs(host string) []uint32 {
Expand Down
8 changes: 4 additions & 4 deletions internal/orchestration/controller.go
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@ import (
"time"

"github.com/relab/hotstuff"
"github.com/relab/hotstuff/client"
"github.com/relab/hotstuff/core/logging"
"github.com/relab/hotstuff/internal/config"
"github.com/relab/hotstuff/internal/latency"
Expand Down Expand Up @@ -166,7 +167,7 @@ type assignmentsFileContents struct {
// the host associated with each replica.
HostsToReplicas map[string][]hotstuff.ID
// the host associated with each client.
HostsToClients map[string][]hotstuff.ID
HostsToClients map[string][]client.ID
}

func (e *Experiment) writeAssignmentsFile(replicaMap config.ReplicaMap, clientIDs config.ClientMap) (err error) {
Expand Down Expand Up @@ -257,7 +258,7 @@ func (e *Experiment) startClients(cfg *orchestrationpb.ReplicaConfiguration, src
req.Configuration = cfg.GetReplicas()
req.CertificateAuthority = keygen.CertToPEM(e.ca)
for _, id := range clientMap[host] {
clientOpts := proto.Clone(srcClientOpt).(*orchestrationpb.ClientOpts)
clientOpts := proto.CloneOf(srcClientOpt)
clientOpts.ID = uint32(id)
req.Clients[uint32(id)] = clientOpts
e.logger.Infof("client %d assigned to host %s", id, host)
Expand All @@ -272,8 +273,7 @@ func (e *Experiment) startClients(cfg *orchestrationpb.ReplicaConfiguration, src

func (e *Experiment) stopClients(clientMap config.ClientMap) error {
for host, worker := range e.workers {
req := &orchestrationpb.StopClientRequest{}
req.IDs = clientMap.ClientIDs(host)
req := &orchestrationpb.StopClientRequest{IDs: clientMap.ClientIDs(host)}
_, err := worker.StopClient(req)
if err != nil {
return err
Expand Down
35 changes: 17 additions & 18 deletions internal/orchestration/worker.go
Original file line number Diff line number Diff line change
Expand Up @@ -54,7 +54,7 @@ type Worker struct {
measurementInterval time.Duration

replicas map[hotstuff.ID]*replica.Replica
clients map[hotstuff.ID]*client.Client
clients map[client.ID]*client.Client
}

// Run runs the worker until it receives a command to quit.
Expand Down Expand Up @@ -102,7 +102,7 @@ func NewWorker(send *protostream.Writer, recv *protostream.Reader, dl metrics.Lo
metrics: metrics,
measurementInterval: measurementInterval,
replicas: make(map[hotstuff.ID]*replica.Replica),
clients: make(map[hotstuff.ID]*client.Client),
clients: make(map[client.ID]*client.Client),
}
}

Expand Down Expand Up @@ -339,8 +339,8 @@ func newTree(opts *orchestrationpb.ReplicaOpts) *tree.Tree {
}

func (w *Worker) startReplicas(req *orchestrationpb.StartReplicaRequest) (*orchestrationpb.StartReplicaResponse, error) {
for _, id := range req.GetIDs() {
replica, ok := w.replicas[hotstuff.ID(id)]
for _, id := range req.ReplicaIDs() {
replica, ok := w.replicas[id]
if !ok {
return nil, status.Errorf(codes.NotFound, "The replica with ID %d was not found.", id)
}
Expand All @@ -353,7 +353,7 @@ func (w *Worker) startReplicas(req *orchestrationpb.StartReplicaRequest) (*orche
return nil, err
}

defer func(id uint32) {
defer func(id hotstuff.ID) {
w.metricsLogger.Log(&types.StartEvent{Event: types.NewReplicaEvent(id, time.Now())})
replica.Start()
}(id)
Expand Down Expand Up @@ -387,26 +387,25 @@ func (w *Worker) startClients(req *orchestrationpb.StartClientRequest) (*orchest
w.metricsLogger.Log(opts)

c := client.Config{
TLS: opts.GetUseTLS(),
RootCAs: cp,
MaxConcurrent: opts.GetMaxConcurrent(),
PayloadSize: opts.GetPayloadSize(),
Input: io.NopCloser(rand.Reader),
TLS: opts.GetUseTLS(),
RootCAs: cp,
MaxConcurrent: opts.GetMaxConcurrent(),
PayloadSize: opts.GetPayloadSize(),
Input: io.NopCloser(rand.Reader),
RateLimit: opts.GetRateLimit(),
RateStep: opts.GetRateStep(),
RateStepInterval: opts.GetRateStepInterval().AsDuration(),
Timeout: opts.GetTimeout().AsDuration(),
}
runtimeCfg := core.NewRuntimeConfig(hotstuff.ID(opts.GetID()), nil)
logger := logging.New("cli" + strconv.Itoa(int(opts.GetID())))
logger := logging.New("cli" + opts.ClientIDString())
eventLoop := eventloop.New(logger, 1000)

if w.measurementInterval > 0 {
err := metrics.Enable(
eventLoop,
logger,
w.metricsLogger,
runtimeCfg.ID(),
opts.ClientID(),
w.measurementInterval,
w.metrics...,
)
Expand All @@ -418,7 +417,7 @@ func (w *Worker) startClients(req *orchestrationpb.StartClientRequest) (*orchest
cli := client.New(
eventLoop,
logger,
runtimeCfg,
opts.ClientID(),
c,
)
cfg, err := getConfiguration(req.GetConfiguration(), true)
Expand All @@ -430,15 +429,15 @@ func (w *Worker) startClients(req *orchestrationpb.StartClientRequest) (*orchest
return nil, err
}
cli.Start()
w.metricsLogger.Log(&types.StartEvent{Event: types.NewClientEvent(opts.GetID(), time.Now())})
w.clients[hotstuff.ID(opts.GetID())] = cli
w.metricsLogger.Log(&types.StartEvent{Event: types.NewClientEvent(opts.ClientID(), time.Now())})
w.clients[opts.ClientID()] = cli
}
return &orchestrationpb.StartClientResponse{}, nil
}

func (w *Worker) stopClients(req *orchestrationpb.StopClientRequest) (*orchestrationpb.StopClientResponse, error) {
for _, id := range req.GetIDs() {
cli, ok := w.clients[hotstuff.ID(id)]
for _, id := range req.ClientIDs() {
cli, ok := w.clients[id]
if !ok {
return nil, status.Errorf(codes.NotFound, "the client with ID %d was not found", id)
}
Expand Down
26 changes: 26 additions & 0 deletions internal/proto/orchestrationpb/client_opts.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,26 @@
package orchestrationpb

import (
"strconv"
"unsafe"

"github.com/relab/hotstuff/client"
)

// ClientID returns the client's ID.
func (x *ClientOpts) ClientID() client.ID {
return client.ID(x.GetID())
}

// ClientIDString returns the client's ID as a string.
func (x *ClientOpts) ClientIDString() string {
return strconv.Itoa(int(x.GetID()))
}

// ClientIDs returns the list of client IDs.
func (x *StopClientRequest) ClientIDs() []client.ID {
if x == nil || len(x.IDs) == 0 {
return nil
}
return unsafe.Slice((*client.ID)(unsafe.Pointer(unsafe.SliceData(x.IDs))), len(x.IDs))
}
9 changes: 9 additions & 0 deletions internal/proto/orchestrationpb/replica_opts.go
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@ import (
"strconv"
"strings"
"time"
"unsafe"

"github.com/relab/hotstuff"
"github.com/relab/hotstuff/protocol/comm"
Expand Down Expand Up @@ -69,6 +70,14 @@ func (x *ReplicaOpts) HotstuffID() hotstuff.ID {
return hotstuff.ID(x.GetID())
}

// ReplicaIDs returns the list of hotstuff IDs.
func (x *StartReplicaRequest) ReplicaIDs() []hotstuff.ID {
if x == nil || len(x.IDs) == 0 {
return nil
}
return unsafe.Slice((*hotstuff.ID)(unsafe.Pointer(unsafe.SliceData(x.IDs))), len(x.IDs))
}

func (x *ReplicaOpts) SetReplicaCertificates(host string, ca *x509.Certificate, caKey *ecdsa.PrivateKey) error {
x.CertificateAuthority = keygen.CertToPEM(ca)
validFor := []string{"localhost", "127.0.0.1", "127.0.1.1", host}
Expand Down
16 changes: 8 additions & 8 deletions metrics/clientlatency.go
Original file line number Diff line number Diff line change
Expand Up @@ -3,26 +3,25 @@ package metrics
import (
"time"

"github.com/relab/hotstuff"
"github.com/relab/hotstuff/client"
"github.com/relab/hotstuff/core/eventloop"
"github.com/relab/hotstuff/metrics/types"
)

const NameClientLatency = "client-latency"

// clientLatency processes LatencyMeasurementEvents, and writes LatencyMeasurements to the metrics logger.
// clientLatency measures the latency of client requests.
type clientLatency struct {
metricsLogger Logger
id hotstuff.ID

wf Welford
id client.ID
wf Welford
}

// enableClientLatency enables client latency measurement.
func enableClientLatency(
el *eventloop.EventLoop,
metricsLogger Logger,
id hotstuff.ID,
id client.ID,
) {
lr := &clientLatency{
id: id,
Expand All @@ -36,16 +35,17 @@ func enableClientLatency(
}, eventloop.Prioritize())
}

// AddLatency adds a latency data point to the current measurement.
// addLatency adds a latency data point to the current measurement.
func (lr *clientLatency) addLatency(latency time.Duration) {
millis := float64(latency) / float64(time.Millisecond)
lr.wf.Update(millis)
}

// tick logs the current latency measurement to the metrics logger.
func (lr *clientLatency) tick(_ types.TickEvent) {
mean, variance, count := lr.wf.Get()
event := &types.LatencyMeasurement{
Event: types.NewClientEvent(uint32(lr.id), time.Now()),
Event: types.NewClientEvent(lr.id, time.Now()),
Latency: mean,
Variance: variance,
Count: count,
Expand Down
17 changes: 9 additions & 8 deletions metrics/consensuslatency.go
Original file line number Diff line number Diff line change
Expand Up @@ -10,20 +10,20 @@ import (

const NameConsensusLatency = "consensus-latency"

// ConsensusLatency processes consensus latency measurements and writes them to the metrics logger.
type ConsensusLatency struct {
// consensusLatency measures the latency of consensus decisions.
type consensusLatency struct {
metricsLogger Logger
id hotstuff.ID
wf Welford
}

// InitModule gives the module access to the other modules.
// enableConsensusLatency enables consensus latency measurement.
func enableConsensusLatency(
el *eventloop.EventLoop,
metricsLogger Logger,
id hotstuff.ID,
) {
lr := ConsensusLatency{
lr := consensusLatency{
metricsLogger: metricsLogger,
id: id,
}
Expand All @@ -35,16 +35,17 @@ func enableConsensusLatency(
}, eventloop.Prioritize())
}

// AddLatency adds a latency data point to the current measurement.
func (lr *ConsensusLatency) addLatency(latency time.Duration) {
// addLatency adds a latency data point to the current measurement.
func (lr *consensusLatency) addLatency(latency time.Duration) {
millis := float64(latency) / float64(time.Millisecond)
lr.wf.Update(millis)
}

func (lr *ConsensusLatency) tick(_ types.TickEvent) {
// tick logs the current latency measurement to the metrics logger.
func (lr *consensusLatency) tick(_ types.TickEvent) {
mean, variance, count := lr.wf.Get()
event := &types.LatencyMeasurement{
Event: types.NewReplicaEvent(uint32(lr.id), time.Now()),
Event: types.NewReplicaEvent(lr.id, time.Now()),
Latency: mean,
Variance: variance,
Count: count,
Expand Down
Loading