-
Notifications
You must be signed in to change notification settings - Fork 542
/
nodehost.go
2195 lines (2067 loc) · 72.9 KB
/
nodehost.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
// Copyright 2017-2021 Lei Ni (nilei81@gmail.com) and other contributors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
/*
Package dragonboat is a feature complete and highly optimized multi-group Raft
implementation for providing consensus in distributed systems.
The NodeHost struct is the facade interface for all features provided by the
dragonboat package. Each NodeHost instance usually runs on a separate server
managing CPU, storage and network resources used for achieving consensus. Each
NodeHost manages Raft nodes from different Raft groups known as Raft shards.
Each Raft shard is identified by its ShardID, it usually consists of
multiple nodes (also known as replicas) each identified by a ReplicaID value.
Nodes from the same Raft shard suppose to be distributed on different NodeHost
instances across the network, this brings fault tolerance for machine and
network failures as application data stored in the Raft shard will be
available as long as the majority of its managing NodeHost instances (i.e. its
underlying servers) are accessible.
Arbitrary number of Raft shards can be launched across the network to
aggregate distributed processing and storage capacities. Users can also make
membership change requests to add or remove nodes from selected Raft shard.
User applications can leverage the power of the Raft protocol by implementing
the IStateMachine or IOnDiskStateMachine component, as defined in
github.com/lni/dragonboat/v4/statemachine. Known as user state machines, each
IStateMachine or IOnDiskStateMachine instance is in charge of updating, querying
and snapshotting application data with minimum exposure to the Raft protocol
itself.
Dragonboat guarantees the linearizability of your I/O when interacting with the
IStateMachine or IOnDiskStateMachine instances. In plain English, writes (via
making proposals) to your Raft shard appears to be instantaneous, once a write
is completed, all later reads (via linearizable read based on Raft's ReadIndex
protocol) should return the value of that write or a later write. Once a value
is returned by a linearizable read, all later reads should return the same value
or the result of a later write.
To strictly provide such guarantee, we need to implement the at-most-once
semantic. For a client, when it retries the proposal that failed to complete by
its deadline, it faces the risk of having the same proposal committed and
applied twice into the user state machine. Dragonboat prevents this by
implementing the client session concept described in Diego Ongaro's PhD thesis.
*/
package dragonboat // github.com/lni/dragonboat/v4
import (
"context"
"math"
"reflect"
"runtime"
"sync"
"sync/atomic"
"time"
"github.com/cockroachdb/errors"
"github.com/lni/goutils/logutil"
"github.com/lni/goutils/syncutil"
"github.com/lni/dragonboat/v4/client"
"github.com/lni/dragonboat/v4/config"
"github.com/lni/dragonboat/v4/internal/id"
"github.com/lni/dragonboat/v4/internal/invariants"
"github.com/lni/dragonboat/v4/internal/logdb"
"github.com/lni/dragonboat/v4/internal/registry"
"github.com/lni/dragonboat/v4/internal/rsm"
"github.com/lni/dragonboat/v4/internal/server"
"github.com/lni/dragonboat/v4/internal/settings"
"github.com/lni/dragonboat/v4/internal/transport"
"github.com/lni/dragonboat/v4/internal/utils"
"github.com/lni/dragonboat/v4/internal/vfs"
"github.com/lni/dragonboat/v4/raftio"
pb "github.com/lni/dragonboat/v4/raftpb"
sm "github.com/lni/dragonboat/v4/statemachine"
)
const (
// DragonboatMajor is the major version number
DragonboatMajor = 4
// DragonboatMinor is the minor version number
DragonboatMinor = 0
// DragonboatPatch is the patch version number
DragonboatPatch = 0
// DEVVersion is a boolean flag indicating whether this is a dev version
DEVVersion = true
)
var (
receiveQueueLen = settings.Soft.ReceiveQueueLength
requestPoolShards = settings.Soft.NodeHostRequestStatePoolShards
streamConnections = settings.Soft.StreamConnections
)
var (
// ErrClosed is returned when a request is made on closed NodeHost instance.
ErrClosed = errors.New("dragonboat: closed")
// ErrReplicaRemoved indictes that the requested node has been removed.
ErrReplicaRemoved = errors.New("node removed")
// ErrShardNotFound indicates that the specified shard is not found.
ErrShardNotFound = errors.New("shard not found")
// ErrShardAlreadyExist indicates that the specified shard already exist.
ErrShardAlreadyExist = errors.New("shard already exist")
// ErrShardNotStopped indicates that the specified shard is still running
// and thus prevented the requested operation to be completed.
ErrShardNotStopped = errors.New("shard not stopped")
// ErrInvalidShardSettings indicates that shard settings specified for
// the StartReplica method are invalid.
ErrInvalidShardSettings = errors.New("shard settings are invalid")
// ErrShardNotBootstrapped indicates that the specified shard has not
// been boostrapped yet. When starting this node, depending on whether this
// node is an initial member of the Raft shard, you must either specify
// all of its initial members or set the join flag to true.
// When used correctly, dragonboat only returns this error in the rare
// situation when you try to restart a node crashed during its previous
// bootstrap attempt.
ErrShardNotBootstrapped = errors.New("shard not bootstrapped")
// ErrDeadlineNotSet indicates that the context parameter provided does not
// carry a deadline.
ErrDeadlineNotSet = errors.New("deadline not set")
// ErrInvalidDeadline indicates that the specified deadline is invalid, e.g.
// time in the past.
ErrInvalidDeadline = errors.New("invalid deadline")
// ErrDirNotExist indicates that the specified dir does not exist.
ErrDirNotExist = errors.New("specified dir does not exist")
// ErrLogDBNotCreatedOrClosed indicates that the logdb is not created yet or closed already.
ErrLogDBNotCreatedOrClosed = errors.New("logdb is not created yet or closed already")
// ErrInvalidRange indicates that the specified log range is invalid.
ErrInvalidRange = errors.New("invalid log range")
)
// ShardInfo is a record for representing the state of a Raft shard based
// on the knowledge of the local NodeHost instance.
type ShardInfo = registry.ShardInfo
// ShardView is a record for representing the state of a Raft shard based
// on the knowledge of distributed NodeHost instances as shared by gossip.
type ShardView = registry.ShardView
// GossipInfo contains details of the gossip service.
type GossipInfo struct {
// AdvertiseAddress is the advertise address used by the gossip service.
AdvertiseAddress string
// NumOfKnownNodeHosts is the number of current live NodeHost instances known
// to the gossip service. Note that the gossip service always knowns the
// local NodeHost instance itself. When the NumOfKnownNodeHosts value is 1,
// it means the gossip service doesn't know any other NodeHost instance that
// is considered as live.
NumOfKnownNodeHosts int
// Enabled is a boolean flag indicating whether the gossip service is enabled.
Enabled bool
}
// NodeHostInfo provides info about the NodeHost, including its managed Raft
// shard nodes and available Raft logs saved in its local persistent storage.
type NodeHostInfo struct {
// NodeHostID is the unique identifier of the NodeHost instance.
NodeHostID string
// RaftAddress is the public address of the NodeHost used for exchanging Raft
// messages, snapshots and other metadata with other NodeHost instances.
RaftAddress string
// Gossip contains gossip service related information.
Gossip GossipInfo
// ShardInfo is a list of all Raft shards managed by the NodeHost
ShardInfoList []ShardInfo
// LogInfo is a list of raftio.NodeInfo values representing all Raft logs
// stored on the NodeHost.
LogInfo []raftio.NodeInfo
}
// NodeHostInfoOption is the option type used when querying NodeHostInfo.
type NodeHostInfoOption struct {
// SkipLogInfo is the boolean flag indicating whether Raft Log info should be
// skipped when querying the NodeHostInfo.
SkipLogInfo bool
}
// DefaultNodeHostInfoOption is the default NodeHostInfoOption value. It
// requests the GetNodeHostInfo method to return all supported info.
var DefaultNodeHostInfoOption NodeHostInfoOption
// SnapshotOption is the options supported when requesting a snapshot to be
// generated.
type SnapshotOption struct {
// ExportPath is the path where the exported snapshot should be stored, it
// must point to an existing directory for which the current user has write
// permission.
ExportPath string
// CompactionOverhead is the compaction overhead value to use for the
// requested snapshot operation when OverrideCompactionOverhead is set to
// true. This field is ignored when exporting a snapshot. ErrInvalidOption
// will be returned if both CompactionOverhead and CompactionIndex are set.
CompactionOverhead uint64
// CompactionIndex specifies the raft log index before which all log entries
// can be compacted after creating the snapshot. This option is only considered
// when OverrideCompactionOverhead is set to true, ErrInvalidOption will be
// returned if both CompactionOverhead and CompactionIndex are set.
CompactionIndex uint64
// Exported is a boolean flag indicating whether to export the requested
// snapshot. For an exported snapshot, users are responsible for managing the
// snapshot files. An exported snapshot is usually used to repair the shard
// when it permanently loses its majority quorum. See the ImportSnapshot method
// in the tools package for more details.
Exported bool
// OverrideCompactionOverhead defines whether the requested snapshot operation
// should override the compaction overhead setting specified in node's config.
// This field is ignored when exporting a snapshot.
OverrideCompactionOverhead bool
}
// Validate checks the SnapshotOption and return error when there is any
// invalid option found.
func (o SnapshotOption) Validate() error {
if o.OverrideCompactionOverhead {
if o.CompactionOverhead > 0 && o.CompactionIndex > 0 {
plog.Errorf("both CompactionOverhead and CompactionIndex are set")
return ErrInvalidOption
}
} else {
if o.CompactionOverhead > 0 || o.CompactionIndex > 0 {
plog.Warningf("CompactionOverhead and CompactionIndex will be ignored")
}
}
return nil
}
// ReadonlyLogReader provides safe readonly access to the underlying logdb.
type ReadonlyLogReader interface {
// GetRange returns the range of the entries in LogDB.
GetRange() (uint64, uint64)
// NodeState returns the state of the node persistent in LogDB.
NodeState() (pb.State, pb.Membership)
// Term returns the entry term of the specified entry.
Term(index uint64) (uint64, error)
// Entries returns entries between [low, high) with total size of entries
// limited to maxSize bytes.
Entries(low uint64, high uint64, maxSize uint64) ([]pb.Entry, error)
// Snapshot returns the metadata for the most recent snapshot known to the
// LogDB.
Snapshot() pb.Snapshot
}
// DefaultSnapshotOption is the default SnapshotOption value to use when
// requesting a snapshot to be generated. This default option causes a regular
// snapshot to be generated.
var DefaultSnapshotOption SnapshotOption
// Target is the type used to specify where a node is running. Target is remote
// NodeHost's RaftAddress value when NodeHostConfig.DefaultNodeRegistryEnabled is not
// set. Target will use NodeHost's ID value when
// NodeHostConfig.DefaultNodeRegistryEnabled is set.
type Target = string
// NodeHost manages Raft shards and enables them to share resources such as
// transport and persistent storage etc. NodeHost is also the central thread
// safe access point for accessing Dragonboat functionalities.
type NodeHost struct {
mu struct {
sync.RWMutex
cci uint64
cciCh chan struct{}
shards sync.Map
lm sync.Map
logdb raftio.ILogDB
}
events struct {
leaderInfoQ *leaderInfoQueue
raft raftio.IRaftEventListener
sys *sysEventListener
}
registry INodeHostRegistry
nodes raftio.INodeRegistry
fs vfs.IFS
transport transport.ITransport
id *id.UUID
stopper *syncutil.Stopper
msgHandler *messageHandler
env *server.Env
engine *engine
nhConfig config.NodeHostConfig
requestPools []*sync.Pool
partitioned int32
closed int32
}
var _ nodeLoader = (*NodeHost)(nil)
var dn = logutil.DescribeNode
var firstError = utils.FirstError
// NewNodeHost creates a new NodeHost instance. In a typical application, it is
// expected to have one NodeHost on each server.
func NewNodeHost(nhConfig config.NodeHostConfig) (*NodeHost, error) {
logBuildTagsAndVersion()
if err := nhConfig.Validate(); err != nil {
return nil, err
}
if err := nhConfig.Prepare(); err != nil {
return nil, err
}
env, err := server.NewEnv(nhConfig, nhConfig.Expert.FS)
if err != nil {
return nil, err
}
nh := &NodeHost{
env: env,
nhConfig: nhConfig,
stopper: syncutil.NewStopper(),
fs: nhConfig.Expert.FS,
}
// make static check happy
_ = nh.partitioned
nh.events.raft = nhConfig.RaftEventListener
nh.events.sys = newSysEventListener(nhConfig.SystemEventListener,
nh.stopper.ShouldStop())
nh.mu.cciCh = make(chan struct{}, 1)
if nhConfig.RaftEventListener != nil {
nh.events.leaderInfoQ = newLeaderInfoQueue()
}
if nhConfig.RaftEventListener != nil || nhConfig.SystemEventListener != nil {
nh.stopper.RunWorker(func() {
nh.handleListenerEvents()
})
}
nh.msgHandler = newNodeHostMessageHandler(nh)
nh.createPools()
defer func() {
if r := recover(); r != nil {
nh.Close()
if r, ok := r.(error); ok {
panicNow(r)
}
}
}()
did := nh.nhConfig.GetDeploymentID()
plog.Infof("DeploymentID set to %d", did)
if err := nh.createLogDB(); err != nil {
nh.Close()
return nil, err
}
if err := nh.loadNodeHostID(); err != nil {
nh.Close()
return nil, err
}
plog.Infof("NodeHost ID: %s", nh.id.String())
if err := nh.createNodeRegistry(); err != nil {
nh.Close()
return nil, err
}
errorInjection := false
if nhConfig.Expert.FS != nil {
_, errorInjection = nhConfig.Expert.FS.(*vfs.ErrorFS)
plog.Infof("filesystem error injection mode enabled: %t", errorInjection)
}
nh.engine = newExecEngine(nh, nhConfig.Expert.Engine,
nh.nhConfig.NotifyCommit, errorInjection, nh.env, nh.mu.logdb)
if err := nh.createTransport(); err != nil {
nh.Close()
return nil, err
}
nh.stopper.RunWorker(func() {
nh.nodeMonitorMain()
})
nh.stopper.RunWorker(func() {
nh.tickWorkerMain()
})
nh.logNodeHostDetails()
return nh, nil
}
// Close stops all managed Raft nodes and releases all resources owned by the
// NodeHost instance.
func (nh *NodeHost) Close() {
nh.events.sys.Publish(server.SystemEvent{
Type: server.NodeHostShuttingDown,
})
nh.mu.Lock()
if atomic.LoadInt32(&nh.closed) != 0 {
panic("NodeHost.Stop called twice")
}
atomic.StoreInt32(&nh.closed, 1)
nh.mu.Unlock()
nodes := make([]raftio.NodeInfo, 0)
nh.forEachShard(func(cid uint64, node *node) bool {
nodes = append(nodes, raftio.NodeInfo{
ShardID: node.shardID,
ReplicaID: node.replicaID,
})
return true
})
for _, node := range nodes {
if err := nh.stopNode(node.ShardID, node.ReplicaID, true); err != nil {
plog.Errorf("failed to remove shard %s",
logutil.ShardID(node.ShardID))
}
}
plog.Debugf("%s is stopping the nh stopper", nh.describe())
nh.stopper.Stop()
var err error
plog.Debugf("%s is stopping the tranport module", nh.describe())
if nh.transport != nil {
err = firstError(err, nh.transport.Close())
}
if nh.nodes != nil {
err = firstError(err, nh.nodes.Close())
nh.nodes = nil
}
plog.Debugf("%s is stopping the engine module", nh.describe())
if nh.engine != nil {
err = firstError(err, nh.engine.close())
nh.engine = nil
nh.transport = nil
}
plog.Debugf("%s is stopping the logdb module", nh.describe())
if nh.mu.logdb != nil {
err = firstError(err, nh.mu.logdb.Close())
nh.mu.logdb = nil
}
plog.Debugf("%s is stopping the env module", nh.describe())
err = firstError(err, nh.env.Close())
plog.Debugf("NodeHost %s stopped", nh.describe())
if err != nil {
panicNow(err)
}
}
// NodeHostConfig returns the NodeHostConfig instance used for configuring this
// NodeHost instance.
func (nh *NodeHost) NodeHostConfig() config.NodeHostConfig {
return nh.nhConfig
}
// RaftAddress returns the Raft address of the NodeHost instance, it is the
// network address by which the NodeHost can be reached by other NodeHost
// instances for exchanging Raft messages, snapshots and other metadata.
func (nh *NodeHost) RaftAddress() string {
return nh.nhConfig.RaftAddress
}
// ID returns the string representation of the NodeHost ID value. The NodeHost
// ID is assigned to each NodeHost on its initial creation and it can be used
// to uniquely identify the NodeHost instance for its entire life cycle. When
// the system is running in the AddressByNodeHost mode, it is used as the target
// value when calling the StartReplica, RequestAddReplica, RequestAddNonVoting,
// RequestAddWitness methods.
func (nh *NodeHost) ID() string {
return nh.id.String()
}
// GetNodeHostRegistry returns the NodeHostRegistry instance that can be used
// to query NodeHost details shared between NodeHost instances by gossip.
func (nh *NodeHost) GetNodeHostRegistry() (INodeHostRegistry, bool) {
return nh.registry, nh.nhConfig.DefaultNodeRegistryEnabled
}
// StartReplica adds the specified Raft replica node to the NodeHost and starts
// the node to make it ready for accepting incoming requests. The node to be
// started is backed by a regular state machine that implements the
// sm.IStateMachine interface.
//
// The input parameter initialMembers is a map of replica ID to replica target for all
// Raft shard's initial member nodes. By default, the target is the
// RaftAddress value of the NodeHost where the node will be running. When running
// in the DefaultNodeRegistryEnabled mode, target should be set to the NodeHostID value
// of the NodeHost where the node will be running. See the godoc of NodeHost's ID
// method for the full definition of NodeHostID. For the same Raft shard, the
// same initialMembers map should be specified when starting its initial member
// nodes on distributed NodeHost instances.
//
// The join flag indicates whether the node is a new node joining an existing
// shard. create is a factory function for creating the IStateMachine instance,
// cfg is the configuration instance that will be passed to the underlying Raft
// node object, the shard ID and replica ID of the involved node are specified in
// the ShardID and ReplicaID fields of the provided cfg parameter.
//
// Note that this method is not for changing the membership of the specified
// Raft shard, it launches a node that is already a member of the Raft shard.
//
// As a summary, when -
// - starting a brand new Raft shard, set join to false and specify all initial
// member node details in the initialMembers map.
// - joining a new node to an existing Raft shard, set join to true and leave
// the initialMembers map empty. This requires the joining node to have already
// been added as a member node of the Raft shard.
// - restarting a crashed or stopped node, set join to false and leave the
// initialMembers map to be empty. This applies to both initial member nodes
// and those joined later.
func (nh *NodeHost) StartReplica(initialMembers map[uint64]Target,
join bool, create sm.CreateStateMachineFunc, cfg config.Config) error {
cf := func(shardID uint64, replicaID uint64,
done <-chan struct{}) rsm.IManagedStateMachine {
sm := create(shardID, replicaID)
return rsm.NewNativeSM(cfg, rsm.NewInMemStateMachine(sm), done)
}
return nh.startShard(initialMembers, join, cf, cfg, pb.RegularStateMachine)
}
// StartConcurrentReplica is similar to the StartReplica method but it is used
// to start a Raft node backed by a concurrent state machine.
func (nh *NodeHost) StartConcurrentReplica(initialMembers map[uint64]Target,
join bool, create sm.CreateConcurrentStateMachineFunc, cfg config.Config) error {
cf := func(shardID uint64, replicaID uint64,
done <-chan struct{}) rsm.IManagedStateMachine {
sm := create(shardID, replicaID)
return rsm.NewNativeSM(cfg, rsm.NewConcurrentStateMachine(sm), done)
}
return nh.startShard(initialMembers,
join, cf, cfg, pb.ConcurrentStateMachine)
}
// StartOnDiskReplica is similar to the StartReplica method but it is used to
// start a Raft node backed by an IOnDiskStateMachine.
func (nh *NodeHost) StartOnDiskReplica(initialMembers map[uint64]Target,
join bool, create sm.CreateOnDiskStateMachineFunc, cfg config.Config) error {
cf := func(shardID uint64, replicaID uint64,
done <-chan struct{}) rsm.IManagedStateMachine {
sm := create(shardID, replicaID)
return rsm.NewNativeSM(cfg, rsm.NewOnDiskStateMachine(sm), done)
}
return nh.startShard(initialMembers,
join, cf, cfg, pb.OnDiskStateMachine)
}
// StopShard stops the local Raft replica associated with the specified Raft
// shard.
//
// Note that this is not the membership change operation required to remove the
// node from the Raft shard.
func (nh *NodeHost) StopShard(shardID uint64) error {
if atomic.LoadInt32(&nh.closed) != 0 {
return ErrClosed
}
return nh.stopNode(shardID, 0, false)
}
// StopReplica stops the specified Raft replica.
//
// Note that this is not the membership change operation required to remove the
// node from the Raft shard.
func (nh *NodeHost) StopReplica(shardID uint64, replicaID uint64) error {
if atomic.LoadInt32(&nh.closed) != 0 {
return ErrClosed
}
return nh.stopNode(shardID, replicaID, true)
}
// SyncPropose makes a synchronous proposal on the Raft shard specified by
// the input client session object. The specified context parameter must have
// the timeout value set.
//
// SyncPropose returns the result returned by IStateMachine or
// IOnDiskStateMachine's Update method, or the error encountered. The input
// byte slice can be reused for other purposes immediate after the return of
// this method.
//
// After calling SyncPropose, unless NO-OP client session is used, it is
// caller's responsibility to update the client session instance accordingly
// based on SyncPropose's outcome. Basically, when a ErrTimeout error is
// returned, application can retry the same proposal without updating the
// client session instance. When ErrInvalidSession error is returned, it
// usually means the session instance has been evicted from the server side,
// the Raft paper recommends to crash the client in this highly unlikely
// event. When the proposal completed successfully, caller must call
// client.ProposalCompleted() to get it ready to be used in future proposals.
func (nh *NodeHost) SyncPropose(ctx context.Context,
session *client.Session, cmd []byte) (sm.Result, error) {
timeout, err := getTimeoutFromContext(ctx)
if err != nil {
return sm.Result{}, err
}
rs, err := nh.Propose(session, cmd, timeout)
if err != nil {
return sm.Result{}, err
}
result, err := getRequestState(ctx, rs)
if err != nil {
return sm.Result{}, err
}
rs.Release()
return result, nil
}
// SyncRead performs a synchronous linearizable read on the specified Raft
// shard. The specified context parameter must have the timeout value set. The
// query interface{} specifies what to query, it will be passed to the Lookup
// method of the IStateMachine or IOnDiskStateMachine after the system
// determines that it is safe to perform the local read. It returns the query
// result from the Lookup method or the error encountered.
func (nh *NodeHost) SyncRead(ctx context.Context, shardID uint64,
query interface{}) (interface{}, error) {
v, err := nh.linearizableRead(ctx, shardID,
func(node *node) (interface{}, error) {
data, err := node.sm.Lookup(query)
if errors.Is(err, rsm.ErrShardClosed) {
return nil, ErrShardClosed
}
return data, err
})
if err != nil {
return nil, err
}
return v, nil
}
// GetLogReader returns a read-only LogDB reader.
func (nh *NodeHost) GetLogReader(shardID uint64) (ReadonlyLogReader, error) {
nh.mu.RLock()
defer nh.mu.RUnlock()
if nh.mu.logdb == nil {
return nil, ErrLogDBNotCreatedOrClosed
}
n, ok := nh.getShard(shardID)
if !ok {
return nil, ErrLogDBNotCreatedOrClosed
}
return n.logReader, nil
}
// Membership is the struct used to describe Raft shard membership.
type Membership struct {
// ConfigChangeID is the Raft entry index of the last applied membership
// change entry.
ConfigChangeID uint64
// Nodes is a map of ReplicaID values to NodeHost Raft addresses for all regular
// Raft nodes.
Nodes map[uint64]string
// NonVotings is a map of ReplicaID values to NodeHost Raft addresses for all
// nonVotings in the Raft shard.
NonVotings map[uint64]string
// Witnesses is a map of ReplicaID values to NodeHost Raft addresses for all
// witnesses in the Raft shard.
Witnesses map[uint64]string
// Removed is a set of ReplicaID values that have been removed from the Raft
// shard. They are not allowed to be added back to the shard.
Removed map[uint64]struct{}
}
// SyncGetShardMembership is a synchronous method that queries the membership
// information from the specified Raft shard. The specified context parameter
// must have the timeout value set.
func (nh *NodeHost) SyncGetShardMembership(ctx context.Context,
shardID uint64) (*Membership, error) {
v, err := nh.linearizableRead(ctx, shardID,
func(node *node) (interface{}, error) {
m := node.sm.GetMembership()
cm := func(input map[uint64]bool) map[uint64]struct{} {
result := make(map[uint64]struct{})
for k := range input {
result[k] = struct{}{}
}
return result
}
return &Membership{
Nodes: m.Addresses,
NonVotings: m.NonVotings,
Witnesses: m.Witnesses,
Removed: cm(m.Removed),
ConfigChangeID: m.ConfigChangeId,
}, nil
})
if err != nil {
return nil, err
}
return v.(*Membership), nil
}
// GetLeaderID returns the leader replica ID of the specified Raft shard based
// on local node's knowledge. The returned boolean value indicates whether the
// leader information is available.
func (nh *NodeHost) GetLeaderID(shardID uint64) (uint64, uint64, bool, error) {
if atomic.LoadInt32(&nh.closed) != 0 {
return 0, 0, false, ErrClosed
}
v, ok := nh.getShard(shardID)
if !ok {
return 0, 0, false, ErrShardNotFound
}
leaderID, term, valid := v.getLeaderID()
return leaderID, term, valid, nil
}
// GetNoOPSession returns a NO-OP client session ready to be used for making
// proposals. The NO-OP client session is a dummy client session that will not
// be checked or enforced. Use this No-OP client session when you want to ignore
// features provided by client sessions. A NO-OP client session is not
// registered on the server side and thus not required to be closed at the end
// of its life cycle.
//
// Returned NO-OP client session instance can be concurrently used in multiple
// goroutines.
//
// Use this NO-OP client session when your IStateMachine provides idempotence in
// its own implementation.
//
// NO-OP client session must be used for making proposals on IOnDiskStateMachine
// based user state machines.
func (nh *NodeHost) GetNoOPSession(shardID uint64) *client.Session {
return client.NewNoOPSession(shardID, nh.env.GetRandomSource())
}
// SyncGetSession starts a synchronous proposal to create, register and return
// a new client session object for the specified Raft shard. The specified
// context parameter must have the timeout value set.
//
// A client session object is used to ensure that a retried proposal, e.g.
// proposal retried after timeout, will not be applied more than once into the
// state machine.
//
// Returned client session instance is not thread safe.
//
// Client session is not supported by IOnDiskStateMachine based user state
// machines. NO-OP client session must be used on IOnDiskStateMachine based
// state machines.
func (nh *NodeHost) SyncGetSession(ctx context.Context,
shardID uint64) (*client.Session, error) {
timeout, err := getTimeoutFromContext(ctx)
if err != nil {
return nil, err
}
cs := client.NewSession(shardID, nh.env.GetRandomSource())
cs.PrepareForRegister()
rs, err := nh.ProposeSession(cs, timeout)
if err != nil {
return nil, err
}
result, err := getRequestState(ctx, rs)
if err != nil {
return nil, err
}
if result.Value != cs.ClientID {
plog.Panicf("unexpected result %d, want %d", result.Value, cs.ClientID)
}
cs.PrepareForPropose()
return cs, nil
}
// SyncCloseSession closes the specified client session by unregistering it
// from the system in a synchronous manner. The specified context parameter
// must have the timeout value set.
//
// Closed client session should not be used in future proposals.
func (nh *NodeHost) SyncCloseSession(ctx context.Context,
cs *client.Session) error {
timeout, err := getTimeoutFromContext(ctx)
if err != nil {
return err
}
cs.PrepareForUnregister()
rs, err := nh.ProposeSession(cs, timeout)
if err != nil {
return err
}
result, err := getRequestState(ctx, rs)
if err != nil {
return err
}
if result.Value != cs.ClientID {
plog.Panicf("unexpected result %d, want %d", result.Value, cs.ClientID)
}
return nil
}
// QueryRaftLog starts an asynchronous query for raft logs in the specified
// range [firstIndex, lastIndex) on the given Raft shard. The returned
// raft log entries are limited to maxSize in bytes.
//
// This method returns a RequestState instance or an error immediately. User
// can use the CompletedC channel of the returned RequestState to get notified
// when the query result becomes available.
func (nh *NodeHost) QueryRaftLog(shardID uint64, firstIndex uint64,
lastIndex uint64, maxSize uint64) (*RequestState, error) {
return nh.queryRaftLog(shardID, firstIndex, lastIndex, maxSize)
}
// Propose starts an asynchronous proposal on the Raft shard specified by the
// Session object. The input byte slice can be reused for other purposes
// immediate after the return of this method.
//
// This method returns a RequestState instance or an error immediately. User can
// wait on the ResultC() channel of the returned RequestState instance to get
// notified for the outcome of the proposal.
//
// After the proposal is completed, i.e. RequestResult is received from the
// ResultC() channel of the returned RequestState, unless NO-OP client session
// is used, it is caller's responsibility to update the Session instance
// accordingly. Basically, when RequestTimeout is returned, you can retry the
// same proposal without updating your client session instance, when a
// RequestRejected value is returned, it usually means the session instance has
// been evicted from the server side as there are too many ongoing client
// sessions, the Raft paper recommends users to crash the client in such highly
// unlikely event. When the proposal completed successfully with a
// RequestCompleted value, application must call client.ProposalCompleted() to
// get the client session ready to be used in future proposals.
func (nh *NodeHost) Propose(session *client.Session, cmd []byte,
timeout time.Duration) (*RequestState, error) {
return nh.propose(session, cmd, timeout)
}
// ProposeSession starts an asynchronous proposal on the specified shard
// for client session related operations. Depending on the state of the specified
// client session object, the supported operations are for registering or
// unregistering a client session. Application can select on the ResultC()
// channel of the returned RequestState instance to get notified for the
// completion (RequestResult.Completed() is true) of the operation.
func (nh *NodeHost) ProposeSession(session *client.Session,
timeout time.Duration) (*RequestState, error) {
n, ok := nh.getShard(session.ShardID)
if !ok {
return nil, ErrShardNotFound
}
// witness node is not expected to propose anything
if n.isWitness() {
return nil, ErrInvalidOperation
}
if !n.supportClientSession() && !session.IsNoOPSession() {
plog.Panicf("IOnDiskStateMachine based nodes must use NoOPSession")
}
defer nh.engine.setStepReady(session.ShardID)
return n.proposeSession(session, nh.getTimeoutTick(timeout))
}
// ReadIndex starts the asynchronous ReadIndex protocol used for linearizable
// read on the specified shard. This method returns a RequestState instance
// or an error immediately. Application should wait on the ResultC() channel
// of the returned RequestState object to get notified on the outcome of the
// ReadIndex operation. On a successful completion, the ReadLocalNode method
// can then be invoked to query the state of the IStateMachine or
// IOnDiskStateMachine with linearizability guarantee.
func (nh *NodeHost) ReadIndex(shardID uint64,
timeout time.Duration) (*RequestState, error) {
rs, _, err := nh.readIndex(shardID, timeout)
return rs, err
}
// ReadLocalNode queries the Raft node identified by the input RequestState
// instance. ReadLocalNode is only allowed to be called after receiving a
// RequestCompleted notification from the ReadIndex method.
func (nh *NodeHost) ReadLocalNode(rs *RequestState,
query interface{}) (interface{}, error) {
if atomic.LoadInt32(&nh.closed) != 0 {
return nil, ErrClosed
}
rs.mustBeReadyForLocalRead()
// translate the rsm.ErrShardClosed to ErrShardClosed
// internally, the IManagedStateMachine might obtain a RLock before performing
// the local read. The critical section is used to make sure we don't read
// from a destroyed C++ StateMachine object
data, err := rs.node.sm.Lookup(query)
if errors.Is(err, rsm.ErrShardClosed) {
return nil, ErrShardClosed
}
return data, err
}
// NAReadLocalNode is a no extra heap allocation variant of ReadLocalNode, it
// uses byte slice as its input and output data to avoid extra heap allocations
// caused by using interface{}. Users are recommended to use the ReadLocalNode
// method unless performance is the top priority.
//
// As an optional feature of the state machine, NAReadLocalNode returns
// statemachine.ErrNotImplemented if the underlying state machine does not
// implement the statemachine.IExtended interface.
//
// Similar to ReadLocalNode, NAReadLocalNode is only allowed to be called after
// receiving a RequestCompleted notification from the ReadIndex method.
func (nh *NodeHost) NAReadLocalNode(rs *RequestState,
query []byte) ([]byte, error) {
if atomic.LoadInt32(&nh.closed) != 0 {
return nil, ErrClosed
}
rs.mustBeReadyForLocalRead()
data, err := rs.node.sm.NALookup(query)
if errors.Is(err, rsm.ErrShardClosed) {
return nil, ErrShardClosed
}
return data, err
}
var staleReadCalled uint32
// StaleRead queries the specified Raft node directly without any
// linearizability guarantee.
func (nh *NodeHost) StaleRead(shardID uint64,
query interface{}) (interface{}, error) {
if atomic.LoadInt32(&nh.closed) != 0 {
return nil, ErrClosed
}
if atomic.CompareAndSwapUint32(&staleReadCalled, 0, 1) {
plog.Warningf("StaleRead called, linearizability not guaranteed for stale read")
}
n, ok := nh.getShard(shardID)
if !ok {
return nil, ErrShardNotFound
}
if !n.initialized() {
return nil, ErrShardNotInitialized
}
if n.isWitness() {
return nil, ErrInvalidOperation
}
data, err := n.sm.Lookup(query)
if errors.Is(err, rsm.ErrShardClosed) {
return nil, ErrShardClosed
}
return data, err
}
// SyncRequestSnapshot is the synchronous variant of the RequestSnapshot
// method. See RequestSnapshot for more details.
//
// The input context object must have deadline set.
//
// SyncRequestSnapshot returns the index of the created snapshot or the error
// encountered.
func (nh *NodeHost) SyncRequestSnapshot(ctx context.Context,
shardID uint64, opt SnapshotOption) (uint64, error) {
timeout, err := getTimeoutFromContext(ctx)
if err != nil {
return 0, err
}
rs, err := nh.RequestSnapshot(shardID, opt, timeout)
if err != nil {
return 0, err
}
v, err := getRequestState(ctx, rs)
if err != nil {
return 0, err
}
return v.Value, nil
}
// RequestSnapshot requests a snapshot to be created asynchronously for the
// specified shard node. For each node, only one ongoing snapshot operation
// is allowed.
//
// Each requested snapshot will also trigger Raft log and snapshot compactions
// similar to automatic snapshotting. Users need to subsequently call
// RequestCompaction(), which can be far more I/O intensive, at suitable time to
// actually reclaim disk spaces used by Raft log entries and snapshot metadata
// records.
//
// RequestSnapshot returns a RequestState instance or an error immediately.
// Applications can wait on the ResultC() channel of the returned RequestState
// instance to get notified for the outcome of the create snasphot operation.
// The RequestResult instance returned by the ResultC() channel tells the
// outcome of the snapshot operation, when successful, the SnapshotIndex method
// of the returned RequestResult instance reports the index of the created
// snapshot.
//
// Requested snapshot operation will be rejected if there is already an existing
// snapshot in the system at the same Raft log index.
func (nh *NodeHost) RequestSnapshot(shardID uint64,
opt SnapshotOption, timeout time.Duration) (*RequestState, error) {
if atomic.LoadInt32(&nh.closed) != 0 {
return nil, ErrClosed
}
n, ok := nh.getShard(shardID)
if !ok {
return nil, ErrShardNotFound
}
if err := opt.Validate(); err != nil {
return nil, err
}
defer nh.engine.setStepReady(shardID)
return n.requestSnapshot(opt, nh.getTimeoutTick(timeout))
}
// RequestCompaction requests a compaction operation to be asynchronously
// executed in the background to reclaim disk spaces used by Raft Log entries
// that have already been marked as removed. This includes Raft Log entries
// that have already been included in created snapshots and Raft Log entries
// that belong to nodes already permanently removed via NodeHost.RemoveData().
//
// By default, compaction is automatically issued after each snapshot is
// captured. RequestCompaction can be used to manually trigger such compaction
// when auto compaction is disabled by the DisableAutoCompactions option in
// config.Config.
//
// The returned *SysOpState instance can be used to get notified when the
// requested compaction is completed. ErrRejected is returned when there is
// nothing to be reclaimed.
func (nh *NodeHost) RequestCompaction(shardID uint64,
replicaID uint64) (*SysOpState, error) {
nh.mu.Lock()
defer nh.mu.Unlock()
if atomic.LoadInt32(&nh.closed) != 0 {
return nil, ErrClosed
}
n, ok := nh.getShard(shardID)