[GitHub] incubator-tephra pull request #23: [Tephra-201] Store checkpoints in in-prog...
GitHub user anew opened a pull request: https://github.com/apache/incubator-tephra/pull/23 [Tephra-201] Store checkpoints in in-progress list See https://issues.apache.org/jira/browse/TEPHRA-201 for the design proposal implemented by this PR. You can merge this pull request into a Git repository by running: $ git pull https://github.com/anew/incubator-tephra tephra-201 Alternatively you can review and apply these changes as the patch at: https://github.com/apache/incubator-tephra/pull/23.patch To close this pull request, make a commit to your master/trunk branch with (at least) the following in the commit message: This closes #23 commit 0cce8c58a76e530c583d6cdc8b075edf2a4fb2b2 Author: anew Date: 2016-12-07T22:19:19Z [TEPHRA-201] Part 1: add test cases for checkpoints commit 4611f3e93de626890e7ce02763e311164f5e6dfb Author: anew Date: 2016-12-08T01:44:08Z [TEPHRA-201] Part 2: store checkpoints in in-progress list and related changes --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] incubator-tephra pull request #23: [Tephra-201] Store checkpoints in in-prog...
Github user chtyim commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/23#discussion_r91460555
--- Diff:
tephra-core/src/main/java/org/apache/tephra/TransactionManager.java ---
@@ -917,6 +933,15 @@ private void doCommit(long transactionId, long
writePointer, Set chang
invalidArray = invalid.toLongArray();
LOG.info("Tx invalid list: removed committed tx {}",
transactionId);
}
+} else {
+ long[] checkpointPointers =
previous.getCheckpointWritePointers().toLongArray();
--- End diff --
Can it be just operating on the `LongArrayList` returned by the
`getCheckpointWritePointers()` method instead of copying it to a new `long[]`?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-tephra pull request #23: [Tephra-201] Store checkpoints in in-prog...
Github user chtyim commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/23#discussion_r91460855
--- Diff:
tephra-core/src/main/java/org/apache/tephra/TransactionManager.java ---
@@ -1319,20 +1344,60 @@ public void shutdown() {
}
/**
+ * Type of in-progress transaction.
+ */
+ public enum InProgressType {
+
+/**
+ * Short transactions detect conflicts during commit.
+ */
+SHORT(TransactionType.SHORT),
+
+/**
+ * Long running transactions do not detect conflicts during commit.
+ */
+LONG(TransactionType.LONG),
+
+/**
+ * Check-pointed transactions are recorded as in-progress.
+ */
+CHECKPOINT(null);
+
+private final TransactionType transactionType;
+
+InProgressType(TransactionType transactionType) {
+ this.transactionType = transactionType;
+}
+
+public static InProgressType of(TransactionType type) {
+ switch (type) {
+case SHORT: return SHORT;
+case LONG: return LONG;
+default: throw new IllegalArgumentException("Unknown
TransactionType " + type);
+ }
+}
+
+@Nullable
+public TransactionType getTransactionType() {
+ return transactionType;
+}
+ }
+
+ /**
* Represents some of the info on in-progress tx
*/
public static final class InProgressTx {
/** the oldest in progress tx at the time of this tx start */
private final long visibilityUpperBound;
private final long expiration;
-private final TransactionType type;
+private final InProgressType type;
private LongArrayList checkpointWritePointers = new LongArrayList();
--- End diff --
This is a unnecessary `new`, since the constructors will always do it.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-tephra pull request #23: [Tephra-201] Store checkpoints in in-prog...
Github user chtyim commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/23#discussion_r91462088
--- Diff:
tephra-core/src/main/java/org/apache/tephra/TransactionManager.java ---
@@ -958,25 +983,31 @@ private void doAbort(long writePointer, long[]
checkpointWritePointers, Transact
// makes tx visible (assumes that all operations were rolled back)
// remove from in-progress set, so that it does not get excluded in
the future
InProgressTx removed = inProgress.remove(writePointer);
+boolean isInvalid = false;
if (removed == null) {
// tx was not in progress! perhaps it timed out and is invalid? try
to remove it there.
if (invalid.rem(writePointer)) {
+isInvalid = true;
// remove any invalidated checkpoint pointers
// this will only be present if the parent write pointer was also
invalidated
if (checkpointWritePointers != null) {
- for (int i = 0; i < checkpointWritePointers.length; i++) {
-invalid.rem(checkpointWritePointers[i]);
+ for (long checkpointWritePointer : checkpointWritePointers) {
+invalid.rem(checkpointWritePointer);
}
}
invalidArray = invalid.toLongArray();
LOG.info("Tx invalid list: removed aborted tx {}", writePointer);
-// removed a tx from excludes: must move read pointer
-moveReadPointerIfNeeded(writePointer);
}
-} else {
- // removed a tx from excludes: must move read pointer
- moveReadPointerIfNeeded(writePointer);
}
+if (!isInvalid) {
+ if (checkpointWritePointers != null) {
--- End diff --
These two `if`s can be combined with a `&&` to have the code less indented.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-tephra pull request #23: [Tephra-201] Store checkpoints in in-prog...
Github user chtyim commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/23#discussion_r91460629
--- Diff:
tephra-core/src/main/java/org/apache/tephra/TransactionManager.java ---
@@ -917,6 +933,15 @@ private void doCommit(long transactionId, long
writePointer, Set chang
invalidArray = invalid.toLongArray();
LOG.info("Tx invalid list: removed committed tx {}",
transactionId);
}
+} else {
+ long[] checkpointPointers =
previous.getCheckpointWritePointers().toLongArray();
+ if (checkpointPointers != null && checkpointPointers.length > 0) {
--- End diff --
I don't think it'll be null, right?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-tephra pull request #23: [Tephra-201] Store checkpoints in in-prog...
Github user chtyim commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/23#discussion_r91460925
--- Diff:
tephra-core/src/main/java/org/apache/tephra/TransactionManager.java ---
@@ -1319,20 +1344,60 @@ public void shutdown() {
}
/**
+ * Type of in-progress transaction.
+ */
+ public enum InProgressType {
+
+/**
+ * Short transactions detect conflicts during commit.
+ */
+SHORT(TransactionType.SHORT),
+
+/**
+ * Long running transactions do not detect conflicts during commit.
+ */
+LONG(TransactionType.LONG),
+
+/**
+ * Check-pointed transactions are recorded as in-progress.
+ */
+CHECKPOINT(null);
+
+private final TransactionType transactionType;
+
+InProgressType(TransactionType transactionType) {
+ this.transactionType = transactionType;
+}
+
+public static InProgressType of(TransactionType type) {
+ switch (type) {
+case SHORT: return SHORT;
+case LONG: return LONG;
+default: throw new IllegalArgumentException("Unknown
TransactionType " + type);
+ }
+}
+
+@Nullable
+public TransactionType getTransactionType() {
+ return transactionType;
+}
+ }
+
+ /**
* Represents some of the info on in-progress tx
*/
public static final class InProgressTx {
/** the oldest in progress tx at the time of this tx start */
private final long visibilityUpperBound;
private final long expiration;
-private final TransactionType type;
+private final InProgressType type;
private LongArrayList checkpointWritePointers = new LongArrayList();
--- End diff --
This should be a final field?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-tephra pull request #23: [Tephra-201] Store checkpoints in in-prog...
Github user chtyim commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/23#discussion_r91457499
--- Diff:
tephra-core/src/main/java/org/apache/tephra/TransactionManager.java ---
@@ -348,27 +348,30 @@ private void cleanupTimedOutTransactions() {
}
long currentTime = System.currentTimeMillis();
-List timedOut = Lists.newArrayList();
+Map timedOut = Maps.newHashMap();
--- End diff --
Unrelated to the change, but it seems like the locking in this method is
wrong.
First, we are holding the read lock for the entire cleanup method, which
seems unnecessary. The inProgress itself is a concurrent tree, so for the first
`for-loop`, it shouldn't need the lock at all, as it is just for collecting the
in progress tx that are timed out.
Then in the `if (!timedOut.isEmpty())`, seems like only the modification to
`invalid` and `invalidArray` needs to be protected by lock, but I think it
should be a "write lock"?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-tephra pull request #23: [Tephra-201] Store checkpoints in in-prog...
Github user chtyim commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/23#discussion_r91463607
--- Diff:
tephra-core/src/main/java/org/apache/tephra/TransactionManager.java ---
@@ -958,25 +983,31 @@ private void doAbort(long writePointer, long[]
checkpointWritePointers, Transact
// makes tx visible (assumes that all operations were rolled back)
// remove from in-progress set, so that it does not get excluded in
the future
InProgressTx removed = inProgress.remove(writePointer);
+boolean isInvalid = false;
if (removed == null) {
// tx was not in progress! perhaps it timed out and is invalid? try
to remove it there.
if (invalid.rem(writePointer)) {
+isInvalid = true;
// remove any invalidated checkpoint pointers
// this will only be present if the parent write pointer was also
invalidated
if (checkpointWritePointers != null) {
- for (int i = 0; i < checkpointWritePointers.length; i++) {
-invalid.rem(checkpointWritePointers[i]);
+ for (long checkpointWritePointer : checkpointWritePointers) {
+invalid.rem(checkpointWritePointer);
}
}
invalidArray = invalid.toLongArray();
LOG.info("Tx invalid list: removed aborted tx {}", writePointer);
-// removed a tx from excludes: must move read pointer
-moveReadPointerIfNeeded(writePointer);
}
-} else {
- // removed a tx from excludes: must move read pointer
- moveReadPointerIfNeeded(writePointer);
}
+if (!isInvalid) {
--- End diff --
I found this variable name less clear on the actual action. Probably better
call it `removeInProgressCheckpoints`? Because it is actually what the
condition is about. It seems less clear to me why "not invalid" means something
needs to be removed from the inprogress when reading this part of the code. I
have to trace back to see how this variable is being set. If the variable is
"removeInProgressCheckpoints", it's more obvious that removal needs to be
performed here and I can optionally trace the code if I am interested in how it
is being set.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-tephra pull request #23: [Tephra-201] Store checkpoints in in-prog...
Github user chtyim commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/23#discussion_r91461565
--- Diff:
tephra-core/src/main/java/org/apache/tephra/TransactionManager.java ---
@@ -917,6 +933,15 @@ private void doCommit(long transactionId, long
writePointer, Set chang
invalidArray = invalid.toLongArray();
LOG.info("Tx invalid list: removed committed tx {}",
transactionId);
}
+} else {
+ long[] checkpointPointers =
previous.getCheckpointWritePointers().toLongArray();
+ if (checkpointPointers != null && checkpointPointers.length > 0) {
+// adjust the write pointer to be the last checkpoint of the tx
and remove all checkpoints from inProgress
+writePointer = checkpointPointers[checkpointPointers.length - 1];
--- End diff --
To avoid boxing and unboxing (assuming you are using the LongArrayList
directly, without copying to an array first), you can do
`checkpointWritePointers.getLong(checkpointWritePointers.size())`.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-tephra pull request #23: [Tephra-201] Store checkpoints in in-prog...
Github user chtyim commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/23#discussion_r91461357
--- Diff:
tephra-core/src/main/java/org/apache/tephra/TransactionManager.java ---
@@ -917,6 +933,15 @@ private void doCommit(long transactionId, long
writePointer, Set chang
invalidArray = invalid.toLongArray();
LOG.info("Tx invalid list: removed committed tx {}",
transactionId);
}
+} else {
+ long[] checkpointPointers =
previous.getCheckpointWritePointers().toLongArray();
+ if (checkpointPointers != null && checkpointPointers.length > 0) {
+// adjust the write pointer to be the last checkpoint of the tx
and remove all checkpoints from inProgress
+writePointer = checkpointPointers[checkpointPointers.length - 1];
+for (long checkpointPointer : checkpointPointers) {
+ inProgress.remove(checkpointPointer);
--- End diff --
the `LongArrayList` returned by `getCheckpointWritePointers()` method is
actually an instance of `Collection` as well, so you can just call
`inProgress.keySet().removeAll(previous.getCheckpointWritePointers())`
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-tephra pull request #23: [Tephra-201] Store checkpoints in in-prog...
Github user chtyim commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/23#discussion_r91465579
--- Diff:
tephra-core/src/main/java/org/apache/tephra/TransactionManager.java ---
@@ -348,27 +348,30 @@ private void cleanupTimedOutTransactions() {
}
long currentTime = System.currentTimeMillis();
-List timedOut = Lists.newArrayList();
+Map timedOut = Maps.newHashMap();
--- End diff --
How about the "write lock" comment? Currently we are only holding the read
lock and modifying the invalid list.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-tephra pull request #23: [Tephra-201] Store checkpoints in in-prog...
Github user anew commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/23#discussion_r91465682
--- Diff:
tephra-core/src/main/java/org/apache/tephra/TransactionManager.java ---
@@ -917,6 +933,15 @@ private void doCommit(long transactionId, long
writePointer, Set chang
invalidArray = invalid.toLongArray();
LOG.info("Tx invalid list: removed committed tx {}",
transactionId);
}
+} else {
+ long[] checkpointPointers =
previous.getCheckpointWritePointers().toLongArray();
+ if (checkpointPointers != null && checkpointPointers.length > 0) {
--- End diff --
I guess not. Copied from existing code...
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-tephra pull request #23: [Tephra-201] Store checkpoints in in-prog...
Github user anew commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/23#discussion_r91466379
--- Diff:
tephra-core/src/main/java/org/apache/tephra/TransactionManager.java ---
@@ -917,6 +933,15 @@ private void doCommit(long transactionId, long
writePointer, Set chang
invalidArray = invalid.toLongArray();
LOG.info("Tx invalid list: removed committed tx {}",
transactionId);
}
+} else {
+ long[] checkpointPointers =
previous.getCheckpointWritePointers().toLongArray();
+ if (checkpointPointers != null && checkpointPointers.length > 0) {
--- End diff --
I can't find a away that this could be null. Changed it everywhere it is
used to assume it is not null
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-tephra pull request #23: [Tephra-201] Store checkpoints in in-prog...
Github user anew commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/23#discussion_r91466464
--- Diff:
tephra-core/src/main/java/org/apache/tephra/TransactionManager.java ---
@@ -917,6 +933,15 @@ private void doCommit(long transactionId, long
writePointer, Set chang
invalidArray = invalid.toLongArray();
LOG.info("Tx invalid list: removed committed tx {}",
transactionId);
}
+} else {
+ long[] checkpointPointers =
previous.getCheckpointWritePointers().toLongArray();
+ if (checkpointPointers != null && checkpointPointers.length > 0) {
+// adjust the write pointer to be the last checkpoint of the tx
and remove all checkpoints from inProgress
+writePointer = checkpointPointers[checkpointPointers.length - 1];
--- End diff --
you mean size()-1? Done
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-tephra pull request #23: [Tephra-201] Store checkpoints in in-prog...
Github user chtyim commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/23#discussion_r91466519
--- Diff:
tephra-core/src/main/java/org/apache/tephra/TransactionManager.java ---
@@ -917,6 +933,15 @@ private void doCommit(long transactionId, long
writePointer, Set chang
invalidArray = invalid.toLongArray();
LOG.info("Tx invalid list: removed committed tx {}",
transactionId);
}
+} else {
+ long[] checkpointPointers =
previous.getCheckpointWritePointers().toLongArray();
+ if (checkpointPointers != null && checkpointPointers.length > 0) {
+// adjust the write pointer to be the last checkpoint of the tx
and remove all checkpoints from inProgress
+writePointer = checkpointPointers[checkpointPointers.length - 1];
--- End diff --
yes :)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-tephra pull request #23: [Tephra-201] Store checkpoints in in-prog...
Github user anew commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/23#discussion_r91467537
--- Diff:
tephra-core/src/main/java/org/apache/tephra/TransactionManager.java ---
@@ -958,25 +983,31 @@ private void doAbort(long writePointer, long[]
checkpointWritePointers, Transact
// makes tx visible (assumes that all operations were rolled back)
// remove from in-progress set, so that it does not get excluded in
the future
InProgressTx removed = inProgress.remove(writePointer);
+boolean isInvalid = false;
if (removed == null) {
// tx was not in progress! perhaps it timed out and is invalid? try
to remove it there.
if (invalid.rem(writePointer)) {
+isInvalid = true;
// remove any invalidated checkpoint pointers
// this will only be present if the parent write pointer was also
invalidated
if (checkpointWritePointers != null) {
- for (int i = 0; i < checkpointWritePointers.length; i++) {
-invalid.rem(checkpointWritePointers[i]);
+ for (long checkpointWritePointer : checkpointWritePointers) {
+invalid.rem(checkpointWritePointer);
}
}
invalidArray = invalid.toLongArray();
LOG.info("Tx invalid list: removed aborted tx {}", writePointer);
-// removed a tx from excludes: must move read pointer
-moveReadPointerIfNeeded(writePointer);
}
-} else {
- // removed a tx from excludes: must move read pointer
- moveReadPointerIfNeeded(writePointer);
}
+if (!isInvalid) {
+ if (checkpointWritePointers != null) {
--- End diff --
good point. done
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-tephra pull request #23: [Tephra-201] Store checkpoints in in-prog...
Github user anew commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/23#discussion_r91467796
--- Diff:
tephra-core/src/main/java/org/apache/tephra/TransactionManager.java ---
@@ -1319,20 +1344,60 @@ public void shutdown() {
}
/**
+ * Type of in-progress transaction.
+ */
+ public enum InProgressType {
+
+/**
+ * Short transactions detect conflicts during commit.
+ */
+SHORT(TransactionType.SHORT),
+
+/**
+ * Long running transactions do not detect conflicts during commit.
+ */
+LONG(TransactionType.LONG),
+
+/**
+ * Check-pointed transactions are recorded as in-progress.
+ */
+CHECKPOINT(null);
+
+private final TransactionType transactionType;
+
+InProgressType(TransactionType transactionType) {
+ this.transactionType = transactionType;
+}
+
+public static InProgressType of(TransactionType type) {
+ switch (type) {
+case SHORT: return SHORT;
+case LONG: return LONG;
+default: throw new IllegalArgumentException("Unknown
TransactionType " + type);
+ }
+}
+
+@Nullable
+public TransactionType getTransactionType() {
+ return transactionType;
+}
+ }
+
+ /**
* Represents some of the info on in-progress tx
*/
public static final class InProgressTx {
/** the oldest in progress tx at the time of this tx start */
private final long visibilityUpperBound;
private final long expiration;
-private final TransactionType type;
+private final InProgressType type;
private LongArrayList checkpointWritePointers = new LongArrayList();
--- End diff --
ditto
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-tephra pull request #23: [Tephra-201] Store checkpoints in in-prog...
Github user anew commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/23#discussion_r91467770
--- Diff:
tephra-core/src/main/java/org/apache/tephra/TransactionManager.java ---
@@ -1319,20 +1344,60 @@ public void shutdown() {
}
/**
+ * Type of in-progress transaction.
+ */
+ public enum InProgressType {
+
+/**
+ * Short transactions detect conflicts during commit.
+ */
+SHORT(TransactionType.SHORT),
+
+/**
+ * Long running transactions do not detect conflicts during commit.
+ */
+LONG(TransactionType.LONG),
+
+/**
+ * Check-pointed transactions are recorded as in-progress.
+ */
+CHECKPOINT(null);
+
+private final TransactionType transactionType;
+
+InProgressType(TransactionType transactionType) {
+ this.transactionType = transactionType;
+}
+
+public static InProgressType of(TransactionType type) {
+ switch (type) {
+case SHORT: return SHORT;
+case LONG: return LONG;
+default: throw new IllegalArgumentException("Unknown
TransactionType " + type);
+ }
+}
+
+@Nullable
+public TransactionType getTransactionType() {
+ return transactionType;
+}
+ }
+
+ /**
* Represents some of the info on in-progress tx
*/
public static final class InProgressTx {
/** the oldest in progress tx at the time of this tx start */
private final long visibilityUpperBound;
private final long expiration;
-private final TransactionType type;
+private final InProgressType type;
private LongArrayList checkpointWritePointers = new LongArrayList();
--- End diff --
not introduced by this PR, but fixed now.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-tephra pull request #23: [Tephra-201] Store checkpoints in in-prog...
Github user anew commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/23#discussion_r91468098
--- Diff:
tephra-core/src/main/java/org/apache/tephra/TransactionManager.java ---
@@ -348,27 +348,30 @@ private void cleanupTimedOutTransactions() {
}
long currentTime = System.currentTimeMillis();
-List timedOut = Lists.newArrayList();
+Map timedOut = Maps.newHashMap();
--- End diff --
you are right. I can change it here, but it is really not related to this
Jira... will do anyway
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-tephra pull request #23: [Tephra-201] Store checkpoints in in-prog...
Github user chtyim commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/23#discussion_r91468570
--- Diff:
tephra-core/src/main/java/org/apache/tephra/TransactionManager.java ---
@@ -348,27 +348,30 @@ private void cleanupTimedOutTransactions() {
}
long currentTime = System.currentTimeMillis();
-List timedOut = Lists.newArrayList();
+Map timedOut = Maps.newHashMap();
--- End diff --
I am just not sure the correctness of that. Probably we can keep the
existing way, and have another PR just focus on lock (not necessarily target
for perf improvement, but rather for the correctness).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-tephra pull request #23: [Tephra-201] Store checkpoints in in-prog...
Github user anew commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/23#discussion_r91469516
--- Diff:
tephra-core/src/main/java/org/apache/tephra/TransactionManager.java ---
@@ -348,27 +348,30 @@ private void cleanupTimedOutTransactions() {
}
long currentTime = System.currentTimeMillis();
-List timedOut = Lists.newArrayList();
+Map timedOut = Maps.newHashMap();
--- End diff --
now I already changed it...
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-tephra pull request #23: [Tephra-201] Store checkpoints in in-prog...
Github user chtyim commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/23#discussion_r91470597
--- Diff:
tephra-core/src/main/java/org/apache/tephra/TransactionManager.java ---
@@ -340,48 +340,51 @@ public long getSleepMillis() {
private void cleanupTimedOutTransactions() {
List invalidEdits = null;
-this.logReadLock.lock();
-try {
- synchronized (this) {
-if (!isRunning()) {
- return;
-}
+synchronized (this) {
+ if (!isRunning()) {
+return;
+ }
-long currentTime = System.currentTimeMillis();
-List timedOut = Lists.newArrayList();
-for (Map.Entry tx : inProgress.entrySet()) {
- long expiration = tx.getValue().getExpiration();
- if (expiration >= 0L && currentTime > expiration) {
-// timed out, remember tx id (can't remove while iterating
over entries)
-timedOut.add(tx.getKey());
-LOG.info("Tx invalid list: added tx {} because of timeout",
tx.getKey());
- } else if (expiration < 0) {
-LOG.warn("Transaction {} has negative expiration time {}.
Likely cause is the transaction was not " +
- "migrated correctly, this transaction will be
expired immediately",
- tx.getKey(), expiration);
-timedOut.add(tx.getKey());
- }
+ long currentTime = System.currentTimeMillis();
+ Map timedOut = Maps.newHashMap();
+ for (Map.Entry tx : inProgress.entrySet()) {
+long expiration = tx.getValue().getExpiration();
+if (expiration >= 0L && currentTime > expiration) {
+ // timed out, remember tx id (can't remove while iterating over
entries)
+ timedOut.put(tx.getKey(), tx.getValue().getType());
+ LOG.info("Tx invalid list: added tx {} because of timeout",
tx.getKey());
+} else if (expiration < 0) {
+ LOG.warn("Transaction {} has negative expiration time {}. Likely
cause is the transaction was not " +
+ "migrated correctly, this transaction will be expired
immediately",
+ tx.getKey(), expiration);
+ timedOut.put(tx.getKey(), InProgressType.LONG);
}
-if (!timedOut.isEmpty()) {
+ }
+ if (!timedOut.isEmpty()) {
+logWriteLock.lock();
--- End diff --
Looking at the code again, my understand of the lock in this method is
wrong. The whole method itself is synchronized on `this` object (with
`synchronize(this)` covering almost the entire method). The `logReadLock` is
for locking on the transaction log writer, which apparently it only needs the
read lock in order to call `appendToLog`. It's probably safer to revert to old
way of locking first. Need further study on the locking mechanism on this class
(apparently there are three locks, one on `this` logic, one read and one write
lock on the tx log).
Sorry for my previous comment.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-tephra pull request #23: [Tephra-201] Store checkpoints in in-prog...
Github user anew commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/23#discussion_r91471138
--- Diff:
tephra-core/src/main/java/org/apache/tephra/TransactionManager.java ---
@@ -340,48 +340,51 @@ public long getSleepMillis() {
private void cleanupTimedOutTransactions() {
List invalidEdits = null;
-this.logReadLock.lock();
-try {
- synchronized (this) {
-if (!isRunning()) {
- return;
-}
+synchronized (this) {
+ if (!isRunning()) {
+return;
+ }
-long currentTime = System.currentTimeMillis();
-List timedOut = Lists.newArrayList();
-for (Map.Entry tx : inProgress.entrySet()) {
- long expiration = tx.getValue().getExpiration();
- if (expiration >= 0L && currentTime > expiration) {
-// timed out, remember tx id (can't remove while iterating
over entries)
-timedOut.add(tx.getKey());
-LOG.info("Tx invalid list: added tx {} because of timeout",
tx.getKey());
- } else if (expiration < 0) {
-LOG.warn("Transaction {} has negative expiration time {}.
Likely cause is the transaction was not " +
- "migrated correctly, this transaction will be
expired immediately",
- tx.getKey(), expiration);
-timedOut.add(tx.getKey());
- }
+ long currentTime = System.currentTimeMillis();
+ Map timedOut = Maps.newHashMap();
+ for (Map.Entry tx : inProgress.entrySet()) {
+long expiration = tx.getValue().getExpiration();
+if (expiration >= 0L && currentTime > expiration) {
+ // timed out, remember tx id (can't remove while iterating over
entries)
+ timedOut.put(tx.getKey(), tx.getValue().getType());
+ LOG.info("Tx invalid list: added tx {} because of timeout",
tx.getKey());
+} else if (expiration < 0) {
+ LOG.warn("Transaction {} has negative expiration time {}. Likely
cause is the transaction was not " +
+ "migrated correctly, this transaction will be expired
immediately",
+ tx.getKey(), expiration);
+ timedOut.put(tx.getKey(), InProgressType.LONG);
}
-if (!timedOut.isEmpty()) {
+ }
+ if (!timedOut.isEmpty()) {
+logWriteLock.lock();
--- End diff --
agreed. ok. will revert
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-tephra pull request #23: [Tephra-201] Store checkpoints in in-prog...
Github user chtyim commented on a diff in the pull request:
https://github.com/apache/incubator-tephra/pull/23#discussion_r91567173
--- Diff:
tephra-core/src/main/java/org/apache/tephra/TransactionManager.java ---
@@ -1011,10 +1038,9 @@ private boolean doInvalidate(long writePointer) {
} else {
// invalidate any checkpoint write pointers
LongArrayList childWritePointers =
previous.getCheckpointWritePointers();
-if (childWritePointers != null) {
- for (int i = 0; i < childWritePointers.size(); i++) {
-invalid.add(childWritePointers.get(i));
- }
+for (long childWritePointer : childWritePointers) {
+ invalid.add(childWritePointer);
+ inProgress.remove(childWritePointer);
--- End diff --
Missed this one in last review, we can use the
`keySet().removeAll(childWritePointers)` as well (and check if
`!childWritePointers.isEmpty()` first). The same for invalid, it can be
`invalid.addAll(childWritePointers)`
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-tephra pull request #23: [Tephra-201] Store checkpoints in in-prog...
Github user asfgit closed the pull request at: https://github.com/apache/incubator-tephra/pull/23 --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
