kgusakov commented on code in PR #936:
URL: https://github.com/apache/ignite-3/pull/936#discussion_r930387918
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/server/impl/JraftServerImpl.java:
##########
@@ -326,89 +331,113 @@ public synchronized boolean startRaftGroup(
/** {@inheritDoc} */
@Override
- public synchronized boolean startRaftGroup(
- String groupId,
+ public boolean startRaftGroup(
+ String grpId,
RaftGroupEventsListener evLsnr,
RaftGroupListener lsnr,
@Nullable List<Peer> initialConf,
RaftGroupOptions groupOptions
) {
- if (groups.containsKey(groupId)) {
+ // fast track to check if group with the same name is already created.
+ if (groups.containsKey(grpId)) {
return false;
}
- // Thread pools are shared by all raft groups.
- NodeOptions nodeOptions = opts.copy();
+ Lock lock = groupLock(grpId);
- // TODO: IGNITE-17083 - Do not create paths for volatile stores at all
when we get rid of snapshot storage on FS.
- Path serverDataPath = getServerDataPath(groupId);
+ lock.lock();
try {
- Files.createDirectories(serverDataPath);
- } catch (IOException e) {
- throw new IgniteInternalException(e);
- }
+ // double check if group wasn't created before receiving the lock.
+ if (groups.containsKey(grpId)) {
+ return false;
+ }
- nodeOptions.setLogUri(groupId);
+ // Thread pools are shared by all raft groups.
+ NodeOptions nodeOptions = opts.copy();
- nodeOptions.setRaftMetaUri(serverDataPath.resolve("meta").toString());
+ // TODO: IGNITE-17083 - Do not create paths for volatile stores at
all when we get rid of snapshot storage on FS.
+ Path serverDataPath = getServerDataPath(grpId);
-
nodeOptions.setSnapshotUri(serverDataPath.resolve("snapshot").toString());
+ try {
+ Files.createDirectories(serverDataPath);
+ } catch (IOException e) {
+ throw new IgniteInternalException(e);
+ }
- nodeOptions.setFsm(new DelegatingStateMachine(lsnr));
+ nodeOptions.setLogUri(grpId);
- nodeOptions.setRaftGrpEvtsLsnr(evLsnr);
+
nodeOptions.setRaftMetaUri(serverDataPath.resolve("meta").toString());
- LogStorageFactory logStorageFactory =
groupOptions.getLogStorageFactory() == null
- ? this.logStorageFactory : groupOptions.getLogStorageFactory();
+
nodeOptions.setSnapshotUri(serverDataPath.resolve("snapshot").toString());
- IgniteJraftServiceFactory serviceFactory = new
IgniteJraftServiceFactory(logStorageFactory);
+ nodeOptions.setFsm(new DelegatingStateMachine(lsnr));
- if (groupOptions.snapshotStorageFactory() != null) {
-
serviceFactory.setSnapshotStorageFactory(groupOptions.snapshotStorageFactory());
- }
+ nodeOptions.setRaftGrpEvtsLsnr(evLsnr);
- if (groupOptions.raftMetaStorageFactory() != null) {
-
serviceFactory.setRaftMetaStorageFactory(groupOptions.raftMetaStorageFactory());
- }
+ LogStorageFactory logStorageFactory =
groupOptions.getLogStorageFactory() == null
+ ? this.logStorageFactory :
groupOptions.getLogStorageFactory();
- nodeOptions.setServiceFactory(serviceFactory);
+ IgniteJraftServiceFactory serviceFactory = new
IgniteJraftServiceFactory(logStorageFactory);
- if (initialConf != null) {
- List<PeerId> mapped =
initialConf.stream().map(PeerId::fromPeer).collect(Collectors.toList());
+ if (groupOptions.snapshotStorageFactory() != null) {
+
serviceFactory.setSnapshotStorageFactory(groupOptions.snapshotStorageFactory());
+ }
- nodeOptions.setInitialConf(new Configuration(mapped, null));
- }
+ if (groupOptions.raftMetaStorageFactory() != null) {
+
serviceFactory.setRaftMetaStorageFactory(groupOptions.raftMetaStorageFactory());
+ }
+
+ nodeOptions.setServiceFactory(serviceFactory);
- IgniteRpcClient client = new IgniteRpcClient(service);
+ if (initialConf != null) {
+ List<PeerId> mapped =
initialConf.stream().map(PeerId::fromPeer).collect(Collectors.toList());
- nodeOptions.setRpcClient(client);
+ nodeOptions.setInitialConf(new Configuration(mapped, null));
+ }
+
+ IgniteRpcClient client = new IgniteRpcClient(service);
- NetworkAddress addr =
service.topologyService().localMember().address();
+ nodeOptions.setRpcClient(client);
- var peerId = new PeerId(addr.host(), addr.port(), 0,
ElectionPriority.DISABLED);
+ NetworkAddress addr =
service.topologyService().localMember().address();
- var server = new RaftGroupService(groupId, peerId, nodeOptions,
rpcServer, nodeManager);
+ var peerId = new PeerId(addr.host(), addr.port(), 0,
ElectionPriority.DISABLED);
- server.start();
+ var server = new RaftGroupService(grpId, peerId, nodeOptions,
rpcServer, nodeManager);
- groups.put(groupId, server);
+ server.start();
- return true;
+ groups.put(grpId, server);
+
+ return true;
+ } finally {
+ lock.unlock();
+ }
}
/** {@inheritDoc} */
@Override
- public boolean stopRaftGroup(String groupId) {
- RaftGroupService svc = groups.remove(groupId);
+ public boolean stopRaftGroup(String grpId) {
Review Comment:
Reworked the whole locking mechanism - now it is simplier and without these
issues
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/server/impl/JraftServerImpl.java:
##########
@@ -466,6 +495,25 @@ public void stopBlockMessages(String groupId) {
client.stopBlock();
}
+ /**
+ * Initialize (if needed) and receive per raft group lock,
+ * to prevent concurrent start of the same raft group.
+ *
+ * @param grpId Group id.
+ * @return lock
+ */
+ private Lock groupLock(String grpId) {
+ var newGroupLock = new ReentrantLock();
Review Comment:
Reworked the whole locking mechanism - now it is simplier and without these
issues
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
