You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@ignite.apache.org by GitBox <gi...@apache.org> on 2021/08/13 09:58:08 UTC
[GitHub] [ignite] denis-chudov opened a new pull request #9327: IGNITE-13558-4
denis-chudov opened a new pull request #9327:
URL: https://github.com/apache/ignite/pull/9327
Thank you for submitting the pull request to the Apache Ignite.
In order to streamline the review of the contribution
we ask you to ensure the following steps have been taken:
### The Contribution Checklist
- [ ] There is a single JIRA ticket related to the pull request.
- [ ] The web-link to the pull request is attached to the JIRA ticket.
- [ ] The JIRA ticket has the _Patch Available_ state.
- [ ] The pull request body describes changes that have been made.
The description explains _WHAT_ and _WHY_ was made instead of _HOW_.
- [ ] The pull request title is treated as the final commit message.
The following pattern must be used: `IGNITE-XXXX Change summary` where `XXXX` - number of JIRA issue.
- [ ] A reviewer has been mentioned through the JIRA comments
(see [the Maintainers list](https://cwiki.apache.org/confluence/display/IGNITE/How+to+Contribute#HowtoContribute-ReviewProcessandMaintainers))
- [ ] The pull request has been checked by the Teamcity Bot and
the `green visa` attached to the JIRA ticket (see [TC.Bot: Check PR](https://mtcga.gridgain.com/prs.html))
### Notes
- [How to Contribute](https://cwiki.apache.org/confluence/display/IGNITE/How+to+Contribute)
- [Coding abbreviation rules](https://cwiki.apache.org/confluence/display/IGNITE/Abbreviation+Rules)
- [Coding Guidelines](https://cwiki.apache.org/confluence/display/IGNITE/Coding+Guidelines)
- [Apache Ignite Teamcity Bot](https://cwiki.apache.org/confluence/display/IGNITE/Apache+Ignite+Teamcity+Bot)
If you need any help, please email dev@ignite.apache.org or ask anу advice on http://asf.slack.com _#ignite_ channel.
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [ignite] asfgit closed pull request #9327: IGNITE-13558-4
Posted by GitBox <gi...@apache.org>.
asfgit closed pull request #9327:
URL: https://github.com/apache/ignite/pull/9327
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [ignite] denis-chudov commented on a change in pull request #9327: IGNITE-13558-4
Posted by GitBox <gi...@apache.org>.
denis-chudov commented on a change in pull request #9327:
URL: https://github.com/apache/ignite/pull/9327#discussion_r717621879
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheOffheapManager.java
##########
@@ -89,16 +88,28 @@
*/
public void stop();
+ /**
+ * Pre-create single partition that resides in page memory or WAL and restores their state.
+ *
+ * @param p Partition id.
+ * @param recoveryState Partition recovery state.
+ * @return Processing time in millis.
+ * @throws IgniteCheckedException If failed.
+ */
+ long restoreStateOfPartition(int p, @Nullable Integer recoveryState) throws IgniteCheckedException;
+
/**
* Pre-create partitions that resides in page memory or WAL and restores their state.
*
- * @param partRecoveryStates Partition recovery states.
- * @return Processed partitions: partition id -> processing time in millis.
* @throws IgniteCheckedException If failed.
*/
- Map<Integer, Long> restorePartitionStates(
- Map<GroupPartitionId, Integer> partRecoveryStates
- ) throws IgniteCheckedException;
+ void restorePartitionStates() throws IgniteCheckedException;
+
+ /**
+ * Confirm that partition states are restored. This method should be called after restoring state of all partitions
+ * in group using {@link #restoreStateOfPartition(int, Integer)}.
+ */
+ void confirmPartitionStatesRestored();
Review comment:
I am not sure - current method name completely matches with what it does
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [ignite] tkalkirill commented on a change in pull request #9327: IGNITE-13558-4
Posted by GitBox <gi...@apache.org>.
tkalkirill commented on a change in pull request #9327:
URL: https://github.com/apache/ignite/pull/9327#discussion_r705311334
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
##########
@@ -5614,17 +5642,44 @@ private void restorePartitionStates(
throw restoreStateError.get();
if (log.isInfoEnabled()) {
- SortedSet<T3<Long, Long, GroupPartitionId>> t = printTop ? topPartRef.get() : null;
+ SortedSet<T3<Long, Long, GroupPartitionId>> t =
+ printTop ? collectTopProcessedParts(threadCtxs, topPartRefLimit) : null;
+
+ long totalProcessed = threadCtxs.stream().mapToLong(c -> c.processedCnt).sum();
log.info("Finished restoring partition state for local groups [" +
"groupsProcessed=" + forGroups.size() +
- ", partitionsProcessed=" + totalProcessed.get() +
+ ", partitionsProcessed=" + totalProcessed +
", time=" + U.humanReadableDuration(U.currentTimeMillis() - startRestorePart) +
(t == null ? "" : ", topProcessedPartitions=" + toStringTopProcessingPartitions(t, forGroups)) +
"]");
}
}
+ /**
+ * Collects top processed partitions from thread local contexts of restore partition state process.
+ *
+ * @param threadCtxs Thread local contexts.
+ * @param topPartRefLimit Limit of top partitions collection size.
+ */
+ private SortedSet<T3<Long, Long, GroupPartitionId>> collectTopProcessedParts(
Review comment:
It may not work correctly in a multi-thread.
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [ignite] denis-chudov commented on a change in pull request #9327: IGNITE-13558-4
Posted by GitBox <gi...@apache.org>.
denis-chudov commented on a change in pull request #9327:
URL: https://github.com/apache/ignite/pull/9327#discussion_r717620455
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheOffheapManager.java
##########
@@ -89,16 +88,28 @@
*/
public void stop();
+ /**
+ * Pre-create single partition that resides in page memory or WAL and restores their state.
+ *
+ * @param p Partition id.
+ * @param recoveryState Partition recovery state.
+ * @return Processing time in millis.
+ * @throws IgniteCheckedException If failed.
+ */
+ long restoreStateOfPartition(int p, @Nullable Integer recoveryState) throws IgniteCheckedException;
Review comment:
too similar to `restorePartitionStates`
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheOffheapManager.java
##########
@@ -89,16 +88,28 @@
*/
public void stop();
+ /**
+ * Pre-create single partition that resides in page memory or WAL and restores their state.
+ *
+ * @param p Partition id.
+ * @param recoveryState Partition recovery state.
+ * @return Processing time in millis.
+ * @throws IgniteCheckedException If failed.
+ */
+ long restoreStateOfPartition(int p, @Nullable Integer recoveryState) throws IgniteCheckedException;
+
/**
* Pre-create partitions that resides in page memory or WAL and restores their state.
*
- * @param partRecoveryStates Partition recovery states.
- * @return Processed partitions: partition id -> processing time in millis.
* @throws IgniteCheckedException If failed.
*/
- Map<Integer, Long> restorePartitionStates(
- Map<GroupPartitionId, Integer> partRecoveryStates
- ) throws IgniteCheckedException;
+ void restorePartitionStates() throws IgniteCheckedException;
+
+ /**
+ * Confirm that partition states are restored. This method should be called after restoring state of all partitions
+ * in group using {@link #restoreStateOfPartition(int, Integer)}.
+ */
+ void confirmPartitionStatesRestored();
Review comment:
I am not sure - current method name completely matches with what it does
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [ignite] tkalkirill commented on a change in pull request #9327: IGNITE-13558-4
Posted by GitBox <gi...@apache.org>.
tkalkirill commented on a change in pull request #9327:
URL: https://github.com/apache/ignite/pull/9327#discussion_r710129931
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheOffheapManager.java
##########
@@ -89,16 +88,28 @@
*/
public void stop();
+ /**
+ * Pre-create single partition that resides in page memory or WAL and restores their state.
+ *
+ * @param p Partition id.
+ * @param recoveryState Partition recovery state.
+ * @return Processing time in millis.
+ * @throws IgniteCheckedException If failed.
+ */
+ long restoreStateOfPartition(int p, @Nullable Integer recoveryState) throws IgniteCheckedException;
Review comment:
Maybe a better name: `restorePartitionState`
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
##########
@@ -5519,66 +5519,76 @@ private void restorePartitionStates(
if (log.isInfoEnabled())
log.info("Restoring partition state for local groups.");
- AtomicLong totalProcessed = new AtomicLong();
-
AtomicReference<IgniteCheckedException> restoreStateError = new AtomicReference<>();
ExecutorService sysPool = ctx.pools().getSystemExecutorService();
- CountDownLatch completionLatch = new CountDownLatch(forGroups.size());
+ final int totalPart = forGroups.stream().mapToInt(grpCtx -> grpCtx.affinity().partitions()).sum();
- AtomicReference<SortedSet<T3<Long, Long, GroupPartitionId>>> topPartRef = new AtomicReference<>();
+ CountDownLatch completionLatch = new CountDownLatch(totalPart);
- long totalPart = forGroups.stream().mapToLong(grpCtx -> grpCtx.affinity().partitions()).sum();
+ Map<Thread, RestorePartitionStateThreadContext> threadCtxs = new ConcurrentHashMap<>();
- for (CacheGroupContext grp : forGroups) {
- sysPool.execute(() -> {
- try {
- Map<Integer, Long> processed = grp.offheap().restorePartitionStates(partStates);
+ final int topPartRefLimit = 5;
- totalProcessed.addAndGet(processed.size());
+ for (CacheGroupContext grpCtx : forGroups) {
+ for (int i = 0; i < grpCtx.affinity().partitions(); i++) {
+ final int partId = i;
- if (log.isInfoEnabled()) {
- TreeSet<T3<Long, Long, GroupPartitionId>> top =
- new TreeSet<>(processedPartitionComparator());
+ sysPool.execute(() -> {
+ GroupPartitionId grpPartId = new GroupPartitionId(grpCtx.groupId(), partId);
- long ts = System.currentTimeMillis();
+ try {
+ long time = grpCtx.offheap().restoreStateOfPartition(grpPartId.getPartitionId(),
Review comment:
You can use `partId` instead `grpPartId.getPartitionId()`
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
##########
@@ -5519,66 +5519,76 @@ private void restorePartitionStates(
if (log.isInfoEnabled())
log.info("Restoring partition state for local groups.");
- AtomicLong totalProcessed = new AtomicLong();
-
AtomicReference<IgniteCheckedException> restoreStateError = new AtomicReference<>();
ExecutorService sysPool = ctx.pools().getSystemExecutorService();
- CountDownLatch completionLatch = new CountDownLatch(forGroups.size());
+ final int totalPart = forGroups.stream().mapToInt(grpCtx -> grpCtx.affinity().partitions()).sum();
- AtomicReference<SortedSet<T3<Long, Long, GroupPartitionId>>> topPartRef = new AtomicReference<>();
+ CountDownLatch completionLatch = new CountDownLatch(totalPart);
- long totalPart = forGroups.stream().mapToLong(grpCtx -> grpCtx.affinity().partitions()).sum();
+ Map<Thread, RestorePartitionStateThreadContext> threadCtxs = new ConcurrentHashMap<>();
- for (CacheGroupContext grp : forGroups) {
- sysPool.execute(() -> {
- try {
- Map<Integer, Long> processed = grp.offheap().restorePartitionStates(partStates);
+ final int topPartRefLimit = 5;
- totalProcessed.addAndGet(processed.size());
+ for (CacheGroupContext grpCtx : forGroups) {
+ for (int i = 0; i < grpCtx.affinity().partitions(); i++) {
+ final int partId = i;
- if (log.isInfoEnabled()) {
- TreeSet<T3<Long, Long, GroupPartitionId>> top =
- new TreeSet<>(processedPartitionComparator());
+ sysPool.execute(() -> {
+ GroupPartitionId grpPartId = new GroupPartitionId(grpCtx.groupId(), partId);
- long ts = System.currentTimeMillis();
+ try {
+ long time = grpCtx.offheap().restoreStateOfPartition(grpPartId.getPartitionId(),
+ partStates.get(grpPartId));
- for (Map.Entry<Integer, Long> e : processed.entrySet()) {
- top.add(new T3<>(e.getValue(), ts, new GroupPartitionId(grp.groupId(), e.getKey())));
+ if (log.isInfoEnabled()) {
+ T3<Long, Long, GroupPartitionId> curPart = new T3<>(time, U.currentTimeMillis(), grpPartId);
- trimToSize(top, 5);
- }
+ RestorePartitionStateThreadContext threadCtx = threadCtxs.computeIfAbsent(
+ Thread.currentThread(),
+ t -> new RestorePartitionStateThreadContext()
+ );
- topPartRef.updateAndGet(top0 -> {
- if (top0 == null)
- return top;
+ Comparator<T3<Long, Long, GroupPartitionId>> cmp = processedPartitionComparator();
- for (T3<Long, Long, GroupPartitionId> t2 : top0) {
- top.add(t2);
+ threadCtx.topPartRef.updateAndGet(top0 -> {
+ if (threadCtx.topPartRef.get() == null ||
+ cmp.compare(threadCtx.topPartRef.get().last(), curPart) < 0) {
Review comment:
`threadCtx.topPartRef.get().last()` -> `prev.last()`
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
##########
@@ -5519,66 +5519,76 @@ private void restorePartitionStates(
if (log.isInfoEnabled())
log.info("Restoring partition state for local groups.");
- AtomicLong totalProcessed = new AtomicLong();
-
AtomicReference<IgniteCheckedException> restoreStateError = new AtomicReference<>();
ExecutorService sysPool = ctx.pools().getSystemExecutorService();
- CountDownLatch completionLatch = new CountDownLatch(forGroups.size());
+ final int totalPart = forGroups.stream().mapToInt(grpCtx -> grpCtx.affinity().partitions()).sum();
- AtomicReference<SortedSet<T3<Long, Long, GroupPartitionId>>> topPartRef = new AtomicReference<>();
+ CountDownLatch completionLatch = new CountDownLatch(totalPart);
- long totalPart = forGroups.stream().mapToLong(grpCtx -> grpCtx.affinity().partitions()).sum();
+ Map<Thread, RestorePartitionStateThreadContext> threadCtxs = new ConcurrentHashMap<>();
- for (CacheGroupContext grp : forGroups) {
- sysPool.execute(() -> {
- try {
- Map<Integer, Long> processed = grp.offheap().restorePartitionStates(partStates);
+ final int topPartRefLimit = 5;
- totalProcessed.addAndGet(processed.size());
+ for (CacheGroupContext grpCtx : forGroups) {
+ for (int i = 0; i < grpCtx.affinity().partitions(); i++) {
+ final int partId = i;
- if (log.isInfoEnabled()) {
- TreeSet<T3<Long, Long, GroupPartitionId>> top =
- new TreeSet<>(processedPartitionComparator());
+ sysPool.execute(() -> {
+ GroupPartitionId grpPartId = new GroupPartitionId(grpCtx.groupId(), partId);
- long ts = System.currentTimeMillis();
+ try {
+ long time = grpCtx.offheap().restoreStateOfPartition(grpPartId.getPartitionId(),
+ partStates.get(grpPartId));
- for (Map.Entry<Integer, Long> e : processed.entrySet()) {
- top.add(new T3<>(e.getValue(), ts, new GroupPartitionId(grp.groupId(), e.getKey())));
+ if (log.isInfoEnabled()) {
+ T3<Long, Long, GroupPartitionId> curPart = new T3<>(time, U.currentTimeMillis(), grpPartId);
- trimToSize(top, 5);
- }
+ RestorePartitionStateThreadContext threadCtx = threadCtxs.computeIfAbsent(
+ Thread.currentThread(),
+ t -> new RestorePartitionStateThreadContext()
+ );
- topPartRef.updateAndGet(top0 -> {
- if (top0 == null)
- return top;
+ Comparator<T3<Long, Long, GroupPartitionId>> cmp = processedPartitionComparator();
- for (T3<Long, Long, GroupPartitionId> t2 : top0) {
- top.add(t2);
+ threadCtx.topPartRef.updateAndGet(top0 -> {
+ if (threadCtx.topPartRef.get() == null ||
+ cmp.compare(threadCtx.topPartRef.get().last(), curPart) < 0) {
+ SortedSet<T3<Long, Long, GroupPartitionId>> top = new TreeSet<>(cmp);
- trimToSize(top, 5);
- }
+ top.add(curPart);
- return top;
- });
+ if (top0 != null)
+ top.addAll(top0);
+
+ trimToSize(top, topPartRefLimit);
+
+ return top;
+ }
+ else
+ return top0;
+ });
+
+ RestorePartitionStateThreadContext.PROCESSED_CNT_UPD.incrementAndGet(threadCtx);
Review comment:
Let's make it a `RestorePartitionStateThreadContext` method.
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
##########
@@ -5519,66 +5519,76 @@ private void restorePartitionStates(
if (log.isInfoEnabled())
log.info("Restoring partition state for local groups.");
- AtomicLong totalProcessed = new AtomicLong();
-
AtomicReference<IgniteCheckedException> restoreStateError = new AtomicReference<>();
ExecutorService sysPool = ctx.pools().getSystemExecutorService();
- CountDownLatch completionLatch = new CountDownLatch(forGroups.size());
+ final int totalPart = forGroups.stream().mapToInt(grpCtx -> grpCtx.affinity().partitions()).sum();
- AtomicReference<SortedSet<T3<Long, Long, GroupPartitionId>>> topPartRef = new AtomicReference<>();
+ CountDownLatch completionLatch = new CountDownLatch(totalPart);
- long totalPart = forGroups.stream().mapToLong(grpCtx -> grpCtx.affinity().partitions()).sum();
+ Map<Thread, RestorePartitionStateThreadContext> threadCtxs = new ConcurrentHashMap<>();
- for (CacheGroupContext grp : forGroups) {
- sysPool.execute(() -> {
- try {
- Map<Integer, Long> processed = grp.offheap().restorePartitionStates(partStates);
+ final int topPartRefLimit = 5;
- totalProcessed.addAndGet(processed.size());
+ for (CacheGroupContext grpCtx : forGroups) {
+ for (int i = 0; i < grpCtx.affinity().partitions(); i++) {
+ final int partId = i;
- if (log.isInfoEnabled()) {
- TreeSet<T3<Long, Long, GroupPartitionId>> top =
- new TreeSet<>(processedPartitionComparator());
+ sysPool.execute(() -> {
+ GroupPartitionId grpPartId = new GroupPartitionId(grpCtx.groupId(), partId);
- long ts = System.currentTimeMillis();
+ try {
+ long time = grpCtx.offheap().restoreStateOfPartition(grpPartId.getPartitionId(),
+ partStates.get(grpPartId));
- for (Map.Entry<Integer, Long> e : processed.entrySet()) {
- top.add(new T3<>(e.getValue(), ts, new GroupPartitionId(grp.groupId(), e.getKey())));
+ if (log.isInfoEnabled()) {
+ T3<Long, Long, GroupPartitionId> curPart = new T3<>(time, U.currentTimeMillis(), grpPartId);
- trimToSize(top, 5);
- }
+ RestorePartitionStateThreadContext threadCtx = threadCtxs.computeIfAbsent(
+ Thread.currentThread(),
+ t -> new RestorePartitionStateThreadContext()
+ );
- topPartRef.updateAndGet(top0 -> {
- if (top0 == null)
- return top;
+ Comparator<T3<Long, Long, GroupPartitionId>> cmp = processedPartitionComparator();
- for (T3<Long, Long, GroupPartitionId> t2 : top0) {
- top.add(t2);
+ threadCtx.topPartRef.updateAndGet(top0 -> {
Review comment:
Please rename `top0` -> `prev`.
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
##########
@@ -5519,66 +5519,76 @@ private void restorePartitionStates(
if (log.isInfoEnabled())
log.info("Restoring partition state for local groups.");
- AtomicLong totalProcessed = new AtomicLong();
-
AtomicReference<IgniteCheckedException> restoreStateError = new AtomicReference<>();
ExecutorService sysPool = ctx.pools().getSystemExecutorService();
- CountDownLatch completionLatch = new CountDownLatch(forGroups.size());
+ final int totalPart = forGroups.stream().mapToInt(grpCtx -> grpCtx.affinity().partitions()).sum();
- AtomicReference<SortedSet<T3<Long, Long, GroupPartitionId>>> topPartRef = new AtomicReference<>();
+ CountDownLatch completionLatch = new CountDownLatch(totalPart);
- long totalPart = forGroups.stream().mapToLong(grpCtx -> grpCtx.affinity().partitions()).sum();
+ Map<Thread, RestorePartitionStateThreadContext> threadCtxs = new ConcurrentHashMap<>();
- for (CacheGroupContext grp : forGroups) {
- sysPool.execute(() -> {
- try {
- Map<Integer, Long> processed = grp.offheap().restorePartitionStates(partStates);
+ final int topPartRefLimit = 5;
- totalProcessed.addAndGet(processed.size());
+ for (CacheGroupContext grpCtx : forGroups) {
+ for (int i = 0; i < grpCtx.affinity().partitions(); i++) {
+ final int partId = i;
- if (log.isInfoEnabled()) {
- TreeSet<T3<Long, Long, GroupPartitionId>> top =
- new TreeSet<>(processedPartitionComparator());
+ sysPool.execute(() -> {
+ GroupPartitionId grpPartId = new GroupPartitionId(grpCtx.groupId(), partId);
- long ts = System.currentTimeMillis();
+ try {
+ long time = grpCtx.offheap().restoreStateOfPartition(grpPartId.getPartitionId(),
+ partStates.get(grpPartId));
- for (Map.Entry<Integer, Long> e : processed.entrySet()) {
- top.add(new T3<>(e.getValue(), ts, new GroupPartitionId(grp.groupId(), e.getKey())));
+ if (log.isInfoEnabled()) {
+ T3<Long, Long, GroupPartitionId> curPart = new T3<>(time, U.currentTimeMillis(), grpPartId);
- trimToSize(top, 5);
- }
+ RestorePartitionStateThreadContext threadCtx = threadCtxs.computeIfAbsent(
+ Thread.currentThread(),
+ t -> new RestorePartitionStateThreadContext()
+ );
- topPartRef.updateAndGet(top0 -> {
- if (top0 == null)
- return top;
+ Comparator<T3<Long, Long, GroupPartitionId>> cmp = processedPartitionComparator();
- for (T3<Long, Long, GroupPartitionId> t2 : top0) {
- top.add(t2);
+ threadCtx.topPartRef.updateAndGet(top0 -> {
+ if (threadCtx.topPartRef.get() == null ||
Review comment:
Use `prev == null`
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheOffheapManager.java
##########
@@ -89,16 +88,28 @@
*/
public void stop();
+ /**
+ * Pre-create single partition that resides in page memory or WAL and restores their state.
+ *
+ * @param p Partition id.
+ * @param recoveryState Partition recovery state.
+ * @return Processing time in millis.
+ * @throws IgniteCheckedException If failed.
+ */
+ long restoreStateOfPartition(int p, @Nullable Integer recoveryState) throws IgniteCheckedException;
+
/**
* Pre-create partitions that resides in page memory or WAL and restores their state.
*
- * @param partRecoveryStates Partition recovery states.
- * @return Processed partitions: partition id -> processing time in millis.
* @throws IgniteCheckedException If failed.
*/
- Map<Integer, Long> restorePartitionStates(
- Map<GroupPartitionId, Integer> partRecoveryStates
- ) throws IgniteCheckedException;
+ void restorePartitionStates() throws IgniteCheckedException;
+
+ /**
+ * Confirm that partition states are restored. This method should be called after restoring state of all partitions
+ * in group using {@link #restoreStateOfPartition(int, Integer)}.
+ */
+ void confirmPartitionStatesRestored();
Review comment:
It's look like callback: `onRestorePartitionStates`
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [ignite] tkalkirill commented on a change in pull request #9327: IGNITE-13558-4
Posted by GitBox <gi...@apache.org>.
tkalkirill commented on a change in pull request #9327:
URL: https://github.com/apache/ignite/pull/9327#discussion_r705309062
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
##########
@@ -5519,66 +5520,89 @@ private void restorePartitionStates(
if (log.isInfoEnabled())
log.info("Restoring partition state for local groups.");
- AtomicLong totalProcessed = new AtomicLong();
-
AtomicReference<IgniteCheckedException> restoreStateError = new AtomicReference<>();
ExecutorService sysPool = ctx.pools().getSystemExecutorService();
- CountDownLatch completionLatch = new CountDownLatch(forGroups.size());
+ final int totalPart = forGroups.stream().mapToInt(grpCtx -> grpCtx.affinity().partitions()).sum();
- AtomicReference<SortedSet<T3<Long, Long, GroupPartitionId>>> topPartRef = new AtomicReference<>();
+ // Group id -> completed partitions counter.
+ Map<Integer, AtomicInteger> grps = new ConcurrentHashMap<>();
Review comment:
I think we can get rid of it.
We can call the `IgniteCacheOffheapManager#confirmPartitionStatesRestored` after restoring all partitions for all groups.
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
##########
@@ -5614,17 +5642,44 @@ private void restorePartitionStates(
throw restoreStateError.get();
if (log.isInfoEnabled()) {
- SortedSet<T3<Long, Long, GroupPartitionId>> t = printTop ? topPartRef.get() : null;
+ SortedSet<T3<Long, Long, GroupPartitionId>> t =
+ printTop ? collectTopProcessedParts(threadCtxs, topPartRefLimit) : null;
+
+ long totalProcessed = threadCtxs.stream().mapToLong(c -> c.processedCnt).sum();
log.info("Finished restoring partition state for local groups [" +
"groupsProcessed=" + forGroups.size() +
- ", partitionsProcessed=" + totalProcessed.get() +
+ ", partitionsProcessed=" + totalProcessed +
", time=" + U.humanReadableDuration(U.currentTimeMillis() - startRestorePart) +
(t == null ? "" : ", topProcessedPartitions=" + toStringTopProcessingPartitions(t, forGroups)) +
"]");
}
}
+ /**
+ * Collects top processed partitions from thread local contexts of restore partition state process.
+ *
+ * @param threadCtxs Thread local contexts.
+ * @param topPartRefLimit Limit of top partitions collection size.
+ */
+ private SortedSet<T3<Long, Long, GroupPartitionId>> collectTopProcessedParts(
Review comment:
It may not work correctly in a multi-thread.
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
##########
@@ -5519,66 +5520,89 @@ private void restorePartitionStates(
if (log.isInfoEnabled())
log.info("Restoring partition state for local groups.");
- AtomicLong totalProcessed = new AtomicLong();
-
AtomicReference<IgniteCheckedException> restoreStateError = new AtomicReference<>();
ExecutorService sysPool = ctx.pools().getSystemExecutorService();
- CountDownLatch completionLatch = new CountDownLatch(forGroups.size());
+ final int totalPart = forGroups.stream().mapToInt(grpCtx -> grpCtx.affinity().partitions()).sum();
- AtomicReference<SortedSet<T3<Long, Long, GroupPartitionId>>> topPartRef = new AtomicReference<>();
+ // Group id -> completed partitions counter.
+ Map<Integer, AtomicInteger> grps = new ConcurrentHashMap<>();
- long totalPart = forGroups.stream().mapToLong(grpCtx -> grpCtx.affinity().partitions()).sum();
+ CountDownLatch completionLatch = new CountDownLatch(totalPart);
- for (CacheGroupContext grp : forGroups) {
- sysPool.execute(() -> {
- try {
- Map<Integer, Long> processed = grp.offheap().restorePartitionStates(partStates);
+ Collection<RestorePartitionStateThreadContext> threadCtxs = new CopyOnWriteArrayList<>();
- totalProcessed.addAndGet(processed.size());
+ ThreadLocal<RestorePartitionStateThreadContext> threadLocalCtx =
Review comment:
I think we should get rid of her.
Let's use the striped map.
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
##########
@@ -5939,13 +5994,25 @@ static String toStringTopProcessingPartitions(
/**
* Comparator of processed partitions.
* T3 -> 1 - duration, 2 - timestamp, 3 - partition of group.
- * Sort order: duration -> timestamp -> partition of group.
+ * Sort order: duration -> timestamp (reversed order) -> partition of group.
*
* @return Comparator.
*/
static Comparator<T3<Long, Long, GroupPartitionId>> processedPartitionComparator() {
Comparator<T3<Long, Long, GroupPartitionId>> comp = Comparator.comparing(T3::get1);
- return comp.thenComparing(T3::get2).thenComparing(T3::get3);
+ return comp.thenComparing(T3::get2, Comparator.reverseOrder()).thenComparing(T3::get3);
+ }
+
+ /**
+ * Thread local context of restore partition state progress.
+ */
+ private static class RestorePartitionStateThreadContext {
+ /** Top partitions by processing time. */
+ final AtomicReference<SortedSet<T3<Long, Long, GroupPartitionId>>> topPartRef =
+ new AtomicReference<>();
+
+ /** Processed partitions count. It is always updated from the same thread. */
+ volatile long processedCnt = 0;
Review comment:
Plz use `java.util.concurrent.atomic.AtomicLongFieldUpdater`
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java
##########
@@ -5519,66 +5520,89 @@ private void restorePartitionStates(
if (log.isInfoEnabled())
log.info("Restoring partition state for local groups.");
- AtomicLong totalProcessed = new AtomicLong();
-
AtomicReference<IgniteCheckedException> restoreStateError = new AtomicReference<>();
ExecutorService sysPool = ctx.pools().getSystemExecutorService();
- CountDownLatch completionLatch = new CountDownLatch(forGroups.size());
+ final int totalPart = forGroups.stream().mapToInt(grpCtx -> grpCtx.affinity().partitions()).sum();
- AtomicReference<SortedSet<T3<Long, Long, GroupPartitionId>>> topPartRef = new AtomicReference<>();
+ // Group id -> completed partitions counter.
+ Map<Integer, AtomicInteger> grps = new ConcurrentHashMap<>();
- long totalPart = forGroups.stream().mapToLong(grpCtx -> grpCtx.affinity().partitions()).sum();
+ CountDownLatch completionLatch = new CountDownLatch(totalPart);
- for (CacheGroupContext grp : forGroups) {
- sysPool.execute(() -> {
- try {
- Map<Integer, Long> processed = grp.offheap().restorePartitionStates(partStates);
+ Collection<RestorePartitionStateThreadContext> threadCtxs = new CopyOnWriteArrayList<>();
- totalProcessed.addAndGet(processed.size());
+ ThreadLocal<RestorePartitionStateThreadContext> threadLocalCtx =
+ ThreadLocal.withInitial(() -> {
+ RestorePartitionStateThreadContext threadCtx = new RestorePartitionStateThreadContext();
- if (log.isInfoEnabled()) {
- TreeSet<T3<Long, Long, GroupPartitionId>> top =
- new TreeSet<>(processedPartitionComparator());
+ threadCtxs.add(threadCtx);
- long ts = System.currentTimeMillis();
+ return threadCtx;
+ });
- for (Map.Entry<Integer, Long> e : processed.entrySet()) {
- top.add(new T3<>(e.getValue(), ts, new GroupPartitionId(grp.groupId(), e.getKey())));
+ final int topPartRefLimit = 5;
- trimToSize(top, 5);
- }
+ for (CacheGroupContext grpCtx : forGroups) {
+ for (int i = 0; i < grpCtx.affinity().partitions(); i++) {
+ final int partId = i;
+
+ sysPool.execute(() -> {
+ GroupPartitionId grpPartId = new GroupPartitionId(grpCtx.groupId(), partId);
+
+ try {
+ long time = grpCtx.offheap().restoreStateOfPartition(grpPartId.getPartitionId(),
+ partStates.get(grpPartId));
+
+ if (log.isInfoEnabled()) {
+ T3<Long, Long, GroupPartitionId> curPart = new T3<>(time, U.currentTimeMillis(), grpPartId);
+
+ RestorePartitionStateThreadContext threadCtx = threadLocalCtx.get();
+
+ Comparator<T3<Long, Long, GroupPartitionId>> cmp = processedPartitionComparator();
+
+ if (threadCtx.topPartRef.get() == null ||
Review comment:
It may not work correctly in a multi-thread.
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [ignite] denis-chudov commented on a change in pull request #9327: IGNITE-13558-4
Posted by GitBox <gi...@apache.org>.
denis-chudov commented on a change in pull request #9327:
URL: https://github.com/apache/ignite/pull/9327#discussion_r717620455
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheOffheapManager.java
##########
@@ -89,16 +88,28 @@
*/
public void stop();
+ /**
+ * Pre-create single partition that resides in page memory or WAL and restores their state.
+ *
+ * @param p Partition id.
+ * @param recoveryState Partition recovery state.
+ * @return Processing time in millis.
+ * @throws IgniteCheckedException If failed.
+ */
+ long restoreStateOfPartition(int p, @Nullable Integer recoveryState) throws IgniteCheckedException;
Review comment:
too similar to `restorePartitionStates`
--
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: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org