You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@accumulo.apache.org by GitBox <gi...@apache.org> on 2021/10/26 18:58:56 UTC
[GitHub] [accumulo] dlmarion opened a new pull request #2329: Attempt at failing FaTE transactions that are blocking this FaTE transaction from the write lock
dlmarion opened a new pull request #2329:
URL: https://github.com/apache/accumulo/pull/2329
Closes #2328
--
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@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] dlmarion commented on a change in pull request #2329: Attempt at failing FaTE transactions that are blocking this FaTE transaction from the write lock
Posted by GitBox <gi...@apache.org>.
dlmarion commented on a change in pull request #2329:
URL: https://github.com/apache/accumulo/pull/2329#discussion_r753316029
##########
File path: core/src/main/java/org/apache/accumulo/fate/zookeeper/DistributedReadWriteLock.java
##########
@@ -218,22 +229,63 @@ public boolean tryLock() {
}
SortedMap<Long,byte[]> entries = qlock.getEarlierEntries(entry);
Iterator<Entry<Long,byte[]>> iterator = entries.entrySet().iterator();
- if (!iterator.hasNext())
+ if (!iterator.hasNext()) {
throw new IllegalStateException("Did not find our own lock in the queue: " + this.entry
+ " userData " + new String(this.userData, UTF_8) + " lockType " + lockType());
- return iterator.next().getKey().equals(entry);
+ }
+ if (iterator.next().getKey().equals(entry)) {
+ return true;
+ }
+ if (preemptBlockingTransactions) {
+ log.info("Preempting transactions that are blocking this transaction");
+ // Loop through all of the prior transactions that are waiting to
+ // acquire this write lock. If the transaction has not succeeded or failed,
+ // then fail it and return false from this method so that Utils.reserveX()
+ // will call this method again and will re-check the prior transactions.
+ iterator = entries.entrySet().iterator();
+ while (iterator.hasNext()) {
+ Entry<Long,byte[]> e = iterator.next();
+ Long priorEntry = e.getKey();
+ ParsedLock priorLock = new ParsedLock(e.getValue());
+ long priorTxid = Long.valueOf(new String(priorLock.getUserData(), UTF_8), 16);
+ if (!priorEntry.equals(entry)) {
+ log.info("Attempting to preempt blocking tx: {}", Long.toHexString(priorTxid));
+ // If the prior tx id is reserved, then it's running. We can't reserve it and set its
+ // status, we have to interrupt the thread.
+ if (store.isReserved(priorTxid)) {
+ store.cancel(priorTxid);
+ } else {
+ try {
+ store.reserve(priorTxid);
+ store.setStatus(priorTxid, TStatus.FAILED_IN_PROGRESS);
+ } finally {
+ store.unreserve(priorTxid, 0);
+ }
+ }
Review comment:
Addressed in 5166fe1
--
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@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] dlmarion commented on pull request #2329: Attempt at failing FaTE transactions that are blocking this FaTE transaction from the write lock
Posted by GitBox <gi...@apache.org>.
dlmarion commented on pull request #2329:
URL: https://github.com/apache/accumulo/pull/2329#issuecomment-971509648
@keith-turner - I addressed the issues you raised. Did you happen to look at the FaTE types that are taking advantage of this mechanism?
One thing to note is that if there are T FaTE threads and F FaTE transactions for a table, and F > T, then the preemption for a transaction will only occur when one of the threads starts to process the tx. So, it may not be immediate.
--
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@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] keith-turner commented on a change in pull request #2329: Attempt at failing FaTE transactions that are blocking this FaTE transaction from the write lock
Posted by GitBox <gi...@apache.org>.
keith-turner commented on a change in pull request #2329:
URL: https://github.com/apache/accumulo/pull/2329#discussion_r740333841
##########
File path: core/src/main/java/org/apache/accumulo/fate/zookeeper/DistributedReadWriteLock.java
##########
@@ -218,22 +237,54 @@ public boolean tryLock() {
}
SortedMap<Long,byte[]> entries = qlock.getEarlierEntries(entry);
Iterator<Entry<Long,byte[]>> iterator = entries.entrySet().iterator();
- if (!iterator.hasNext())
+ if (!iterator.hasNext()) {
throw new IllegalStateException("Did not find our own lock in the queue: " + this.entry
+ " userData " + new String(this.userData, UTF_8) + " lockType " + lockType());
- return iterator.next().getKey().equals(entry);
+ }
+ if (!failBlockers) {
+ return iterator.next().getKey().equals(entry);
+ } else {
+ ZooStore<DistributedReadWriteLock> zs;
+ try {
+ zs = new ZooStore<>(zooPath, zrw);
+ } catch (KeeperException | InterruptedException e1) {
+ log.error("Error creating zoo store", e1);
+ return false;
+ }
+ final AdminUtil<DistributedReadWriteLock> util = new AdminUtil<>();
+ boolean result = true;
+ while (iterator.hasNext()) {
+ Entry<Long,byte[]> e = iterator.next();
+ if (!e.getKey().equals(entry)) {
+ result &= util.prepFail(zs, zrw, zooManagerPath, Long.toString(e.getKey(), 16));
Review comment:
@dlmarion could possible do something like the following... where the code in `failPrecedingOperations()` would go through and transition operations to FAILED_IN_PROGRESS... also thinking that code could transition NEW things straight to FAILED instead of to FAILED_IN_PROGRESS but not sure. However there is not waiting on these other operations. The FATE framework will reserve a txid, take the top repo off the stack, and then call isReady(). If isReady returns non-zero it will unreserve and the txid and come back to it later. So if isReady transitions things to FAILED_IN_PROGRESS then hopefully a later call to isReady can get the lock sooner.
```java
class SomeRepo extends Repo {
public long isReady(long tid,...) {
long waitTime = tryToGetTableWriteLock(...);
if(waitTime > 0 && failBlockers) {
// we did not get the the table write lock, so lets initiate the failure of all operations that precede us in the lock queue.... hopefully after these fail they will release the lock and we will get the lock on a later call to isRead()
failPrecedingOperations(tid)
}
return waitTime;
}
}
```
##########
File path: core/src/main/java/org/apache/accumulo/fate/zookeeper/DistributedReadWriteLock.java
##########
@@ -218,22 +237,54 @@ public boolean tryLock() {
}
SortedMap<Long,byte[]> entries = qlock.getEarlierEntries(entry);
Iterator<Entry<Long,byte[]>> iterator = entries.entrySet().iterator();
- if (!iterator.hasNext())
+ if (!iterator.hasNext()) {
throw new IllegalStateException("Did not find our own lock in the queue: " + this.entry
+ " userData " + new String(this.userData, UTF_8) + " lockType " + lockType());
- return iterator.next().getKey().equals(entry);
+ }
+ if (!failBlockers) {
+ return iterator.next().getKey().equals(entry);
+ } else {
+ ZooStore<DistributedReadWriteLock> zs;
+ try {
+ zs = new ZooStore<>(zooPath, zrw);
+ } catch (KeeperException | InterruptedException e1) {
+ log.error("Error creating zoo store", e1);
+ return false;
+ }
+ final AdminUtil<DistributedReadWriteLock> util = new AdminUtil<>();
+ boolean result = true;
+ while (iterator.hasNext()) {
+ Entry<Long,byte[]> e = iterator.next();
+ if (!e.getKey().equals(entry)) {
+ result &= util.prepFail(zs, zrw, zooManagerPath, Long.toString(e.getKey(), 16));
Review comment:
@dlmarion could possibly do something like the following... where the code in `failPrecedingOperations()` would go through and transition operations to FAILED_IN_PROGRESS... also thinking that code could transition NEW things straight to FAILED instead of to FAILED_IN_PROGRESS but not sure. However there is not waiting on these other operations. The FATE framework will reserve a txid, take the top repo off the stack, and then call isReady(). If isReady() returns non-zero it will unreserve and the txid and come back to it later. So if isReady transitions things to FAILED_IN_PROGRESS then hopefully a later call to isReady can get the lock sooner.
```java
class SomeRepo extends Repo {
public long isReady(long tid,...) {
long waitTime = tryToGetTableWriteLock(...);
if(waitTime > 0 && failBlockers) {
// we did not get the the table write lock, so lets initiate the failure of all operations that precede us in the lock queue.... hopefully after these fail they will release the lock and we will get the lock on a later call to isRead()
failPrecedingOperations(tid)
}
return waitTime;
}
}
```
##########
File path: core/src/main/java/org/apache/accumulo/fate/zookeeper/DistributedReadWriteLock.java
##########
@@ -218,22 +237,54 @@ public boolean tryLock() {
}
SortedMap<Long,byte[]> entries = qlock.getEarlierEntries(entry);
Iterator<Entry<Long,byte[]>> iterator = entries.entrySet().iterator();
- if (!iterator.hasNext())
+ if (!iterator.hasNext()) {
throw new IllegalStateException("Did not find our own lock in the queue: " + this.entry
+ " userData " + new String(this.userData, UTF_8) + " lockType " + lockType());
- return iterator.next().getKey().equals(entry);
+ }
+ if (!failBlockers) {
+ return iterator.next().getKey().equals(entry);
+ } else {
+ ZooStore<DistributedReadWriteLock> zs;
+ try {
+ zs = new ZooStore<>(zooPath, zrw);
+ } catch (KeeperException | InterruptedException e1) {
+ log.error("Error creating zoo store", e1);
+ return false;
+ }
+ final AdminUtil<DistributedReadWriteLock> util = new AdminUtil<>();
+ boolean result = true;
+ while (iterator.hasNext()) {
+ Entry<Long,byte[]> e = iterator.next();
+ if (!e.getKey().equals(entry)) {
+ result &= util.prepFail(zs, zrw, zooManagerPath, Long.toString(e.getKey(), 16));
Review comment:
For the prev comment, failPrecedingOperations() code could possibly be pushed into the lock code. I teased it out in the example above to show how it could work w/ the isReady() call to avoid waiting for transitions to happen.
##########
File path: core/src/main/java/org/apache/accumulo/fate/zookeeper/DistributedReadWriteLock.java
##########
@@ -218,22 +233,82 @@ public boolean tryLock() {
}
SortedMap<Long,byte[]> entries = qlock.getEarlierEntries(entry);
Iterator<Entry<Long,byte[]>> iterator = entries.entrySet().iterator();
- if (!iterator.hasNext())
+ if (!iterator.hasNext()) {
throw new IllegalStateException("Did not find our own lock in the queue: " + this.entry
+ " userData " + new String(this.userData, UTF_8) + " lockType " + lockType());
- return iterator.next().getKey().equals(entry);
+ }
+ if (!failBlockers) {
+ return iterator.next().getKey().equals(entry);
+ } else {
+ ZooStore<DistributedReadWriteLock> zs;
+ try {
+ zs = new ZooStore<>(zooPath, zrw);
Review comment:
Need to use the same ZooStore object that the manager/Fate framework is using, or will not get mutual exclusion when reserving transactions. Look at the impl of zs.reserve(long), by creating a new object will be using a different hashset and synchronize on diff obj monitors than the manager/Fate framework.
##########
File path: core/src/main/java/org/apache/accumulo/fate/zookeeper/DistributedReadWriteLock.java
##########
@@ -218,22 +233,82 @@ public boolean tryLock() {
}
SortedMap<Long,byte[]> entries = qlock.getEarlierEntries(entry);
Iterator<Entry<Long,byte[]>> iterator = entries.entrySet().iterator();
- if (!iterator.hasNext())
+ if (!iterator.hasNext()) {
throw new IllegalStateException("Did not find our own lock in the queue: " + this.entry
+ " userData " + new String(this.userData, UTF_8) + " lockType " + lockType());
- return iterator.next().getKey().equals(entry);
+ }
+ if (!failBlockers) {
+ return iterator.next().getKey().equals(entry);
+ } else {
+ ZooStore<DistributedReadWriteLock> zs;
+ try {
+ zs = new ZooStore<>(zooPath, zrw);
+ } catch (KeeperException | InterruptedException e1) {
+ log.error("Error creating zoo store", e1);
+ return false;
+ }
+ // Loop through all of the prior transactions that are waiting to
+ // acquire this write lock. If the transaction has not succeeded or failed,
+ // then fail it and return false from this method so that Utils.reserveX()
+ // will call this method again and will re-check the prior transactions.
+ boolean result = true;
+ while (iterator.hasNext()) {
+ Entry<Long,byte[]> e = iterator.next();
+ Long txid = e.getKey();
+ if (!txid.equals(entry)) {
+ if (zs.isReserved(txid)) {
+ result &= false;
+ } else {
+ zs.reserve(txid);
+ TStatus status = zs.getStatus(txid);
+ if (status.equals(TStatus.FAILED)) {
+ result = false;
+ continue;
+ } else {
+ switch (status) {
+ case UNKNOWN:
+ // not sure what to do here, we have an invalid txid
+ break;
+ case SUCCESSFUL:
+ // already succeeded, lock should be removed
+ break;
+ case FAILED:
+ case FAILED_IN_PROGRESS:
+ // transaction failed or in process of failing
+ break;
+ default:
+ zs.setStatus(txid, TStatus.FAILED_IN_PROGRESS);
+ result &= false;
+ }
+ }
+ zs.unreserve(txid, 0);
+ }
+ } else {
+ return result && txid.equals(entry);
+ }
+ }
+ return result;
Review comment:
The following high level structure would be nice, where if we are first in the queue return true no matter what fail blockers is... also never return true unless we are the 1st in the queue. Can't really apply this suggestion, just used a suggestion to make clear the scope I was thinking about.
```suggestion
if( iterator.next().getKey().equals(entry)){
return true;
}
if (failBlockers) {
// attempt to fail anything before this lock entry in the queue.. but never return true unless ours is the 1st
}
return false;
```
##########
File path: core/src/main/java/org/apache/accumulo/fate/zookeeper/DistributedReadWriteLock.java
##########
@@ -218,22 +233,82 @@ public boolean tryLock() {
}
SortedMap<Long,byte[]> entries = qlock.getEarlierEntries(entry);
Iterator<Entry<Long,byte[]>> iterator = entries.entrySet().iterator();
- if (!iterator.hasNext())
+ if (!iterator.hasNext()) {
throw new IllegalStateException("Did not find our own lock in the queue: " + this.entry
+ " userData " + new String(this.userData, UTF_8) + " lockType " + lockType());
- return iterator.next().getKey().equals(entry);
+ }
+ if (!failBlockers) {
+ return iterator.next().getKey().equals(entry);
+ } else {
+ ZooStore<DistributedReadWriteLock> zs;
+ try {
+ zs = new ZooStore<>(zooPath, zrw);
+ } catch (KeeperException | InterruptedException e1) {
+ log.error("Error creating zoo store", e1);
+ return false;
+ }
+ // Loop through all of the prior transactions that are waiting to
+ // acquire this write lock. If the transaction has not succeeded or failed,
+ // then fail it and return false from this method so that Utils.reserveX()
+ // will call this method again and will re-check the prior transactions.
+ boolean result = true;
+ while (iterator.hasNext()) {
+ Entry<Long,byte[]> e = iterator.next();
+ Long txid = e.getKey();
+ if (!txid.equals(entry)) {
+ if (zs.isReserved(txid)) {
+ result &= false;
+ } else {
+ zs.reserve(txid);
+ TStatus status = zs.getStatus(txid);
+ if (status.equals(TStatus.FAILED)) {
+ result = false;
+ continue;
+ } else {
+ switch (status) {
+ case UNKNOWN:
+ // not sure what to do here, we have an invalid txid
+ break;
+ case SUCCESSFUL:
+ // already succeeded, lock should be removed
+ break;
+ case FAILED:
+ case FAILED_IN_PROGRESS:
Review comment:
Investigation would be needed to confirm this, but I suspect we should only expect to see IN_PROGRESS and FAILED_IN_PROGRESS transactions actually holding the lock. Anything else seems like and indication of a more general problem that we should not attempt to react to here. We could log a warn/error if we see them, but it may not be a good idea to let the lock be acquired when the system is an unexpected state. This kinda goes w/ my other comment of always returning false when we are not first in the queue.
--
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@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] keith-turner commented on a change in pull request #2329: Attempt at failing FaTE transactions that are blocking this FaTE transaction from the write lock
Posted by GitBox <gi...@apache.org>.
keith-turner commented on a change in pull request #2329:
URL: https://github.com/apache/accumulo/pull/2329#discussion_r741111823
##########
File path: core/src/main/java/org/apache/accumulo/fate/zookeeper/DistributedReadWriteLock.java
##########
@@ -218,22 +233,82 @@ public boolean tryLock() {
}
SortedMap<Long,byte[]> entries = qlock.getEarlierEntries(entry);
Iterator<Entry<Long,byte[]>> iterator = entries.entrySet().iterator();
- if (!iterator.hasNext())
+ if (!iterator.hasNext()) {
throw new IllegalStateException("Did not find our own lock in the queue: " + this.entry
+ " userData " + new String(this.userData, UTF_8) + " lockType " + lockType());
- return iterator.next().getKey().equals(entry);
+ }
+ if (!failBlockers) {
+ return iterator.next().getKey().equals(entry);
+ } else {
+ ZooStore<DistributedReadWriteLock> zs;
+ try {
+ zs = new ZooStore<>(zooPath, zrw);
Review comment:
Need to use the same ZooStore object that the manager/Fate framework is using, or will not get mutual exclusion when reserving transactions. Look at the impl of zs.reserve(long), by creating a new object will be using a different hashset and synchronize on diff obj monitors than the manager/Fate framework.
##########
File path: core/src/main/java/org/apache/accumulo/fate/zookeeper/DistributedReadWriteLock.java
##########
@@ -218,22 +233,82 @@ public boolean tryLock() {
}
SortedMap<Long,byte[]> entries = qlock.getEarlierEntries(entry);
Iterator<Entry<Long,byte[]>> iterator = entries.entrySet().iterator();
- if (!iterator.hasNext())
+ if (!iterator.hasNext()) {
throw new IllegalStateException("Did not find our own lock in the queue: " + this.entry
+ " userData " + new String(this.userData, UTF_8) + " lockType " + lockType());
- return iterator.next().getKey().equals(entry);
+ }
+ if (!failBlockers) {
+ return iterator.next().getKey().equals(entry);
+ } else {
+ ZooStore<DistributedReadWriteLock> zs;
+ try {
+ zs = new ZooStore<>(zooPath, zrw);
+ } catch (KeeperException | InterruptedException e1) {
+ log.error("Error creating zoo store", e1);
+ return false;
+ }
+ // Loop through all of the prior transactions that are waiting to
+ // acquire this write lock. If the transaction has not succeeded or failed,
+ // then fail it and return false from this method so that Utils.reserveX()
+ // will call this method again and will re-check the prior transactions.
+ boolean result = true;
+ while (iterator.hasNext()) {
+ Entry<Long,byte[]> e = iterator.next();
+ Long txid = e.getKey();
+ if (!txid.equals(entry)) {
+ if (zs.isReserved(txid)) {
+ result &= false;
+ } else {
+ zs.reserve(txid);
+ TStatus status = zs.getStatus(txid);
+ if (status.equals(TStatus.FAILED)) {
+ result = false;
+ continue;
+ } else {
+ switch (status) {
+ case UNKNOWN:
+ // not sure what to do here, we have an invalid txid
+ break;
+ case SUCCESSFUL:
+ // already succeeded, lock should be removed
+ break;
+ case FAILED:
+ case FAILED_IN_PROGRESS:
+ // transaction failed or in process of failing
+ break;
+ default:
+ zs.setStatus(txid, TStatus.FAILED_IN_PROGRESS);
+ result &= false;
+ }
+ }
+ zs.unreserve(txid, 0);
+ }
+ } else {
+ return result && txid.equals(entry);
+ }
+ }
+ return result;
Review comment:
The following high level structure would be nice, where if we are first in the queue return true no matter what fail blockers is... also never return true unless we are the 1st in the queue. Can't really apply this suggestion, just used a suggestion to make clear the scope I was thinking about.
```suggestion
if( iterator.next().getKey().equals(entry)){
return true;
}
if (failBlockers) {
// attempt to fail anything before this lock entry in the queue.. but never return true unless ours is the 1st
}
return false;
```
##########
File path: core/src/main/java/org/apache/accumulo/fate/zookeeper/DistributedReadWriteLock.java
##########
@@ -218,22 +233,82 @@ public boolean tryLock() {
}
SortedMap<Long,byte[]> entries = qlock.getEarlierEntries(entry);
Iterator<Entry<Long,byte[]>> iterator = entries.entrySet().iterator();
- if (!iterator.hasNext())
+ if (!iterator.hasNext()) {
throw new IllegalStateException("Did not find our own lock in the queue: " + this.entry
+ " userData " + new String(this.userData, UTF_8) + " lockType " + lockType());
- return iterator.next().getKey().equals(entry);
+ }
+ if (!failBlockers) {
+ return iterator.next().getKey().equals(entry);
+ } else {
+ ZooStore<DistributedReadWriteLock> zs;
+ try {
+ zs = new ZooStore<>(zooPath, zrw);
+ } catch (KeeperException | InterruptedException e1) {
+ log.error("Error creating zoo store", e1);
+ return false;
+ }
+ // Loop through all of the prior transactions that are waiting to
+ // acquire this write lock. If the transaction has not succeeded or failed,
+ // then fail it and return false from this method so that Utils.reserveX()
+ // will call this method again and will re-check the prior transactions.
+ boolean result = true;
+ while (iterator.hasNext()) {
+ Entry<Long,byte[]> e = iterator.next();
+ Long txid = e.getKey();
+ if (!txid.equals(entry)) {
+ if (zs.isReserved(txid)) {
+ result &= false;
+ } else {
+ zs.reserve(txid);
+ TStatus status = zs.getStatus(txid);
+ if (status.equals(TStatus.FAILED)) {
+ result = false;
+ continue;
+ } else {
+ switch (status) {
+ case UNKNOWN:
+ // not sure what to do here, we have an invalid txid
+ break;
+ case SUCCESSFUL:
+ // already succeeded, lock should be removed
+ break;
+ case FAILED:
+ case FAILED_IN_PROGRESS:
Review comment:
Investigation would be needed to confirm this, but I suspect we should only expect to see IN_PROGRESS and FAILED_IN_PROGRESS transactions actually holding the lock. Anything else seems like and indication of a more general problem that we should not attempt to react to here. We could log a warn/error if we see them, but it may not be a good idea to let the lock be acquired when the system is an unexpected state. This kinda goes w/ my other comment of always returning false when we are not first in the queue.
--
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@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] dlmarion commented on a change in pull request #2329: Attempt at failing FaTE transactions that are blocking this FaTE transaction from the write lock
Posted by GitBox <gi...@apache.org>.
dlmarion commented on a change in pull request #2329:
URL: https://github.com/apache/accumulo/pull/2329#discussion_r746867577
##########
File path: core/src/main/java/org/apache/accumulo/fate/zookeeper/DistributedReadWriteLock.java
##########
@@ -218,22 +228,69 @@ public boolean tryLock() {
}
SortedMap<Long,byte[]> entries = qlock.getEarlierEntries(entry);
Iterator<Entry<Long,byte[]>> iterator = entries.entrySet().iterator();
- if (!iterator.hasNext())
+ if (!iterator.hasNext()) {
throw new IllegalStateException("Did not find our own lock in the queue: " + this.entry
+ " userData " + new String(this.userData, UTF_8) + " lockType " + lockType());
- return iterator.next().getKey().equals(entry);
+ }
+ if (iterator.next().getKey().equals(entry)) {
+ return true;
+ }
+ if (failBlockers) {
+ // Loop through all of the prior transactions that are waiting to
+ // acquire this write lock. If the transaction has not succeeded or failed,
+ // then fail it and return false from this method so that Utils.reserveX()
+ // will call this method again and will re-check the prior transactions.
+ boolean result = true;
+ while (iterator.hasNext()) {
+ Entry<Long,byte[]> e = iterator.next();
+ Long txid = e.getKey();
+ if (!txid.equals(entry)) {
+ if (store.isReserved(txid)) {
+ result &= false;
+ } else {
+ store.reserve(txid);
+ TStatus status = store.getStatus(txid);
+ if (status.equals(TStatus.FAILED)) {
+ result = false;
+ continue;
+ } else {
+ switch (status) {
+ case FAILED_IN_PROGRESS:
+ case IN_PROGRESS:
+ store.setStatus(txid, TStatus.FAILED_IN_PROGRESS);
+ result &= false;
+ break;
+ default:
+ log.error("Attempting to fail a transaction in an unhandled state: {}", status);
+ }
+ }
+ store.unreserve(txid, 0);
+ }
+ } else {
+ return result && txid.equals(entry);
Review comment:
Addressed in 2b8fc42ebd19b8c9b3861ab9d5abaa3d34a6888f
--
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@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] dlmarion commented on pull request #2329: Attempt at failing FaTE transactions that are blocking this FaTE transaction from the write lock
Posted by GitBox <gi...@apache.org>.
dlmarion commented on pull request #2329:
URL: https://github.com/apache/accumulo/pull/2329#issuecomment-1027889508
I think I have found the issue with ConcurrentDeleteTableIT and I think it exposed something I had not accounted for. I'd like to track it down and fix it and then we can judge whether or not this approach may not be the right one. Maybe a different approach would be to just log the blockers for a FaTE transaction so that the user can take some action. That approach will likely reuse some of the code from this PR.
--
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@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] keith-turner commented on a change in pull request #2329: Attempt at failing FaTE transactions that are blocking this FaTE transaction from the write lock
Posted by GitBox <gi...@apache.org>.
keith-turner commented on a change in pull request #2329:
URL: https://github.com/apache/accumulo/pull/2329#discussion_r746901941
##########
File path: core/src/main/java/org/apache/accumulo/fate/zookeeper/DistributedReadWriteLock.java
##########
@@ -218,22 +228,69 @@ public boolean tryLock() {
}
SortedMap<Long,byte[]> entries = qlock.getEarlierEntries(entry);
Iterator<Entry<Long,byte[]>> iterator = entries.entrySet().iterator();
- if (!iterator.hasNext())
+ if (!iterator.hasNext()) {
throw new IllegalStateException("Did not find our own lock in the queue: " + this.entry
+ " userData " + new String(this.userData, UTF_8) + " lockType " + lockType());
- return iterator.next().getKey().equals(entry);
+ }
+ if (iterator.next().getKey().equals(entry)) {
+ return true;
+ }
+ if (failBlockers) {
+ // Loop through all of the prior transactions that are waiting to
+ // acquire this write lock. If the transaction has not succeeded or failed,
+ // then fail it and return false from this method so that Utils.reserveX()
+ // will call this method again and will re-check the prior transactions.
+ boolean result = true;
Review comment:
The changes in 2b8fc42 look good.
--
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@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] keith-turner commented on a change in pull request #2329: Attempt at failing FaTE transactions that are blocking this FaTE transaction from the write lock
Posted by GitBox <gi...@apache.org>.
keith-turner commented on a change in pull request #2329:
URL: https://github.com/apache/accumulo/pull/2329#discussion_r740333841
##########
File path: core/src/main/java/org/apache/accumulo/fate/zookeeper/DistributedReadWriteLock.java
##########
@@ -218,22 +237,54 @@ public boolean tryLock() {
}
SortedMap<Long,byte[]> entries = qlock.getEarlierEntries(entry);
Iterator<Entry<Long,byte[]>> iterator = entries.entrySet().iterator();
- if (!iterator.hasNext())
+ if (!iterator.hasNext()) {
throw new IllegalStateException("Did not find our own lock in the queue: " + this.entry
+ " userData " + new String(this.userData, UTF_8) + " lockType " + lockType());
- return iterator.next().getKey().equals(entry);
+ }
+ if (!failBlockers) {
+ return iterator.next().getKey().equals(entry);
+ } else {
+ ZooStore<DistributedReadWriteLock> zs;
+ try {
+ zs = new ZooStore<>(zooPath, zrw);
+ } catch (KeeperException | InterruptedException e1) {
+ log.error("Error creating zoo store", e1);
+ return false;
+ }
+ final AdminUtil<DistributedReadWriteLock> util = new AdminUtil<>();
+ boolean result = true;
+ while (iterator.hasNext()) {
+ Entry<Long,byte[]> e = iterator.next();
+ if (!e.getKey().equals(entry)) {
+ result &= util.prepFail(zs, zrw, zooManagerPath, Long.toString(e.getKey(), 16));
Review comment:
@dlmarion could possible do something like the following... where the code in `failPrecedingOperations()` would go through and transition operations to FAILED_IN_PROGRESS... also thinking that code could transition NEW things straight to FAILED instead of to FAILED_IN_PROGRESS but not sure. However there is not waiting on these other operations. The FATE framework will reserve a txid, take the top repo off the stack, and then call isReady(). If isReady returns non-zero it will unreserve and the txid and come back to it later. So if isReady transitions things to FAILED_IN_PROGRESS then hopefully a later call to isReady can get the lock sooner.
```java
class SomeRepo extends Repo {
public long isReady(long tid,...) {
long waitTime = tryToGetTableWriteLock(...);
if(waitTime > 0 && failBlockers) {
// we did not get the the table write lock, so lets initiate the failure of all operations that precede us in the lock queue.... hopefully after these fail they will release the lock and we will get the lock on a later call to isRead()
failPrecedingOperations(tid)
}
return waitTime;
}
}
```
--
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@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] keith-turner commented on a change in pull request #2329: Attempt at failing FaTE transactions that are blocking this FaTE transaction from the write lock
Posted by GitBox <gi...@apache.org>.
keith-turner commented on a change in pull request #2329:
URL: https://github.com/apache/accumulo/pull/2329#discussion_r740340322
##########
File path: core/src/main/java/org/apache/accumulo/fate/zookeeper/DistributedReadWriteLock.java
##########
@@ -218,22 +237,54 @@ public boolean tryLock() {
}
SortedMap<Long,byte[]> entries = qlock.getEarlierEntries(entry);
Iterator<Entry<Long,byte[]>> iterator = entries.entrySet().iterator();
- if (!iterator.hasNext())
+ if (!iterator.hasNext()) {
throw new IllegalStateException("Did not find our own lock in the queue: " + this.entry
+ " userData " + new String(this.userData, UTF_8) + " lockType " + lockType());
- return iterator.next().getKey().equals(entry);
+ }
+ if (!failBlockers) {
+ return iterator.next().getKey().equals(entry);
+ } else {
+ ZooStore<DistributedReadWriteLock> zs;
+ try {
+ zs = new ZooStore<>(zooPath, zrw);
+ } catch (KeeperException | InterruptedException e1) {
+ log.error("Error creating zoo store", e1);
+ return false;
+ }
+ final AdminUtil<DistributedReadWriteLock> util = new AdminUtil<>();
+ boolean result = true;
+ while (iterator.hasNext()) {
+ Entry<Long,byte[]> e = iterator.next();
+ if (!e.getKey().equals(entry)) {
+ result &= util.prepFail(zs, zrw, zooManagerPath, Long.toString(e.getKey(), 16));
Review comment:
For the prev comment, failPrecedingOperations() code could possibly be pushed into the lock code. I teased it out in the example above to show how it could work w/ the isReady() call to avoid waiting for transitions to happen.
--
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@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] keith-turner commented on a change in pull request #2329: Attempt at failing FaTE transactions that are blocking this FaTE transaction from the write lock
Posted by GitBox <gi...@apache.org>.
keith-turner commented on a change in pull request #2329:
URL: https://github.com/apache/accumulo/pull/2329#discussion_r746882673
##########
File path: core/src/main/java/org/apache/accumulo/fate/zookeeper/DistributedReadWriteLock.java
##########
@@ -218,22 +228,69 @@ public boolean tryLock() {
}
SortedMap<Long,byte[]> entries = qlock.getEarlierEntries(entry);
Iterator<Entry<Long,byte[]>> iterator = entries.entrySet().iterator();
- if (!iterator.hasNext())
+ if (!iterator.hasNext()) {
throw new IllegalStateException("Did not find our own lock in the queue: " + this.entry
+ " userData " + new String(this.userData, UTF_8) + " lockType " + lockType());
- return iterator.next().getKey().equals(entry);
+ }
+ if (iterator.next().getKey().equals(entry)) {
+ return true;
+ }
+ if (failBlockers) {
+ // Loop through all of the prior transactions that are waiting to
+ // acquire this write lock. If the transaction has not succeeded or failed,
+ // then fail it and return false from this method so that Utils.reserveX()
+ // will call this method again and will re-check the prior transactions.
+ boolean result = true;
Review comment:
> Is that what you are saying? Sounds good to me.
yeah that is what I was thinking. Was thinking if we are not the first in the list that maybe those other fate ops could be running undo operation as part of unwinding themselves in the case of a failure. Their undo ops could be modifying shared state in ZK and metadata. So was thinking its safest to wait until we are the first.
--
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@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] keith-turner commented on a change in pull request #2329: Attempt at failing FaTE transactions that are blocking this FaTE transaction from the write lock
Posted by GitBox <gi...@apache.org>.
keith-turner commented on a change in pull request #2329:
URL: https://github.com/apache/accumulo/pull/2329#discussion_r752497136
##########
File path: core/src/main/java/org/apache/accumulo/fate/zookeeper/DistributedReadWriteLock.java
##########
@@ -218,22 +229,63 @@ public boolean tryLock() {
}
SortedMap<Long,byte[]> entries = qlock.getEarlierEntries(entry);
Iterator<Entry<Long,byte[]>> iterator = entries.entrySet().iterator();
- if (!iterator.hasNext())
+ if (!iterator.hasNext()) {
throw new IllegalStateException("Did not find our own lock in the queue: " + this.entry
+ " userData " + new String(this.userData, UTF_8) + " lockType " + lockType());
- return iterator.next().getKey().equals(entry);
+ }
+ if (iterator.next().getKey().equals(entry)) {
+ return true;
+ }
+ if (preemptBlockingTransactions) {
+ log.info("Preempting transactions that are blocking this transaction");
+ // Loop through all of the prior transactions that are waiting to
+ // acquire this write lock. If the transaction has not succeeded or failed,
+ // then fail it and return false from this method so that Utils.reserveX()
+ // will call this method again and will re-check the prior transactions.
+ iterator = entries.entrySet().iterator();
+ while (iterator.hasNext()) {
+ Entry<Long,byte[]> e = iterator.next();
+ Long priorEntry = e.getKey();
+ ParsedLock priorLock = new ParsedLock(e.getValue());
+ long priorTxid = Long.valueOf(new String(priorLock.getUserData(), UTF_8), 16);
+ if (!priorEntry.equals(entry)) {
+ log.info("Attempting to preempt blocking tx: {}", Long.toHexString(priorTxid));
+ // If the prior tx id is reserved, then it's running. We can't reserve it and set its
+ // status, we have to interrupt the thread.
+ if (store.isReserved(priorTxid)) {
+ store.cancel(priorTxid);
+ } else {
+ try {
+ store.reserve(priorTxid);
+ store.setStatus(priorTxid, TStatus.FAILED_IN_PROGRESS);
+ } finally {
+ store.unreserve(priorTxid, 0);
+ }
+ }
Review comment:
Seeing three possible problems with this code.
1. Does not seems like it checks the current status before changing it. If the current status were FAILED or SUCCEEDED (if things change after we read from ZK before the loop) then we would not want to set it to FAILED_IN_PROGRESS. Seems like prev versions of this PR did this type of check. Could reserve, get the status, and change it depending on what it is.
2. When isReserved() returns false and then we call reserve(), it could become reserved between those two call which would lead to the reserve call blocking which could cause problems. May still want the tryReserve() call.
3. Thinking more about the cancel call I realized it may cancel/interrupt a transaction that is undoing and I am not completely sure what the implications of this are. I suspect the intention is that it only cancels transactions that are in the IN_PROGRESS state and not in the FAILED_IN_PROGRESS state. Interrupting something in the FAILED_IN_PROGRESS state (that is when unwinds the statck executing undo ops) seems counterproductive.
--
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@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] keith-turner commented on a change in pull request #2329: Attempt at failing FaTE transactions that are blocking this FaTE transaction from the write lock
Posted by GitBox <gi...@apache.org>.
keith-turner commented on a change in pull request #2329:
URL: https://github.com/apache/accumulo/pull/2329#discussion_r740263645
##########
File path: core/src/main/java/org/apache/accumulo/fate/zookeeper/DistributedReadWriteLock.java
##########
@@ -218,22 +237,54 @@ public boolean tryLock() {
}
SortedMap<Long,byte[]> entries = qlock.getEarlierEntries(entry);
Iterator<Entry<Long,byte[]>> iterator = entries.entrySet().iterator();
- if (!iterator.hasNext())
+ if (!iterator.hasNext()) {
throw new IllegalStateException("Did not find our own lock in the queue: " + this.entry
+ " userData " + new String(this.userData, UTF_8) + " lockType " + lockType());
- return iterator.next().getKey().equals(entry);
+ }
+ if (!failBlockers) {
+ return iterator.next().getKey().equals(entry);
+ } else {
+ ZooStore<DistributedReadWriteLock> zs;
+ try {
+ zs = new ZooStore<>(zooPath, zrw);
+ } catch (KeeperException | InterruptedException e1) {
+ log.error("Error creating zoo store", e1);
+ return false;
+ }
+ final AdminUtil<DistributedReadWriteLock> util = new AdminUtil<>();
+ boolean result = true;
+ while (iterator.hasNext()) {
+ Entry<Long,byte[]> e = iterator.next();
+ if (!e.getKey().equals(entry)) {
+ result &= util.prepFail(zs, zrw, zooManagerPath, Long.toString(e.getKey(), 16));
Review comment:
Looking at this some more, the call to reserve() in prepFail may address the 2nd issue I mentioned above.. but still investigating that.
--
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@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] milleruntime commented on pull request #2329: Attempt at failing FaTE transactions that are blocking this FaTE transaction from the write lock
Posted by GitBox <gi...@apache.org>.
milleruntime commented on pull request #2329:
URL: https://github.com/apache/accumulo/pull/2329#issuecomment-1025745466
I wonder if there are servers running taking up ports and not getting shut down properly.
--
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@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] dlmarion commented on pull request #2329: Attempt at failing FaTE transactions that are blocking this FaTE transaction from the write lock
Posted by GitBox <gi...@apache.org>.
dlmarion commented on pull request #2329:
URL: https://github.com/apache/accumulo/pull/2329#issuecomment-967567259
@keith-turner - can you take a look at my changes in 1755864 to see if this implementation seems sane. The high level notes are that I modified ZooStore to:
1. Keep a reference to the thread that is reserving the tx
2. Added a cancel method to call interrupt on the thread
The calling Thread in most cases should be the Fate.TransactionRunner threads which already handle the Exception and transition the tx to FAILED_IN_PROGRESS and then ultimately unreserves the tx. I also fixed up a test and added FateIT with a test for preemption. Finally I took a stab at determining which FaTE transaction types should preempt.
--
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@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] dlmarion closed pull request #2329: Attempt at failing FaTE transactions that are blocking this FaTE transaction from the write lock
Posted by GitBox <gi...@apache.org>.
dlmarion closed pull request #2329:
URL: https://github.com/apache/accumulo/pull/2329
--
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@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] milleruntime commented on a change in pull request #2329: Attempt at failing FaTE transactions that are blocking this FaTE transaction from the write lock
Posted by GitBox <gi...@apache.org>.
milleruntime commented on a change in pull request #2329:
URL: https://github.com/apache/accumulo/pull/2329#discussion_r791869863
##########
File path: core/src/main/java/org/apache/accumulo/fate/zookeeper/DistributedReadWriteLock.java
##########
@@ -218,22 +229,68 @@ public boolean tryLock() {
}
SortedMap<Long,byte[]> entries = qlock.getEarlierEntries(entry);
Iterator<Entry<Long,byte[]>> iterator = entries.entrySet().iterator();
- if (!iterator.hasNext())
+ if (!iterator.hasNext()) {
throw new IllegalStateException("Did not find our own lock in the queue: " + this.entry
+ " userData " + new String(this.userData, UTF_8) + " lockType " + lockType());
- return iterator.next().getKey().equals(entry);
+ }
+ if (iterator.next().getKey().equals(entry)) {
+ return true;
+ }
+ if (preemptBlockingTransactions) {
+ log.info("Preempting transactions that are blocking this transaction");
Review comment:
In other parts of the code you call this boolean `cancelWriteLockBlockers`. I think picking one consistent name would be better. I like `cancelWriteLockBlockers` better than `preemptBlockingTransactions`. I have a PR against your branch with some minor changes that includes this rename.
--
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@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] ctubbsii commented on pull request #2329: Attempt at failing FaTE transactions that are blocking this FaTE transaction from the write lock
Posted by GitBox <gi...@apache.org>.
ctubbsii commented on pull request #2329:
URL: https://github.com/apache/accumulo/pull/2329#issuecomment-1027367812
This already seems like it's getting pretty complicated. Is it even safe, in the general case, to fate operations this way? I feel like a human should probably be interactively responsible for cancelling fate operations in order to unblock their preferred one, so they can accept the risks directly.
--
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@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] dlmarion commented on a change in pull request #2329: Attempt at failing FaTE transactions that are blocking this FaTE transaction from the write lock
Posted by GitBox <gi...@apache.org>.
dlmarion commented on a change in pull request #2329:
URL: https://github.com/apache/accumulo/pull/2329#discussion_r751167001
##########
File path: core/src/main/java/org/apache/accumulo/fate/ZooStore.java
##########
@@ -529,4 +527,16 @@ public long timeCreated(long tid) {
return dops;
}
}
+
+ /**
+ * Attempt to cancel the reserved transaction by interrupting the Thread that is running it
+ *
+ * @param tid
+ * transaction id
+ */
+ public void cancel(long tid) {
+ if (reserved.containsKey(tid)) {
Review comment:
Addressed in d5df49354bf83d9174445c5e669936bb5cdf473e
--
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@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] keith-turner commented on a change in pull request #2329: Attempt at failing FaTE transactions that are blocking this FaTE transaction from the write lock
Posted by GitBox <gi...@apache.org>.
keith-turner commented on a change in pull request #2329:
URL: https://github.com/apache/accumulo/pull/2329#discussion_r740260185
##########
File path: core/src/main/java/org/apache/accumulo/fate/zookeeper/DistributedReadWriteLock.java
##########
@@ -218,22 +237,54 @@ public boolean tryLock() {
}
SortedMap<Long,byte[]> entries = qlock.getEarlierEntries(entry);
Iterator<Entry<Long,byte[]>> iterator = entries.entrySet().iterator();
- if (!iterator.hasNext())
+ if (!iterator.hasNext()) {
throw new IllegalStateException("Did not find our own lock in the queue: " + this.entry
+ " userData " + new String(this.userData, UTF_8) + " lockType " + lockType());
- return iterator.next().getKey().equals(entry);
+ }
+ if (!failBlockers) {
+ return iterator.next().getKey().equals(entry);
+ } else {
+ ZooStore<DistributedReadWriteLock> zs;
+ try {
+ zs = new ZooStore<>(zooPath, zrw);
+ } catch (KeeperException | InterruptedException e1) {
+ log.error("Error creating zoo store", e1);
+ return false;
+ }
+ final AdminUtil<DistributedReadWriteLock> util = new AdminUtil<>();
+ boolean result = true;
+ while (iterator.hasNext()) {
+ Entry<Long,byte[]> e = iterator.next();
+ if (!e.getKey().equals(entry)) {
+ result &= util.prepFail(zs, zrw, zooManagerPath, Long.toString(e.getKey(), 16));
Review comment:
I see two problems with the call here. First prepFail() calls checkGlobalLock() which calls System.exit() when the constuctor AdminUtil<>() is called. The reason it calles checkGlobalLock is because I think this code was only intended to be called when the manager was not running and no repos were running. The Second problem is that prepFail only initiates failure, it does not wait for any operations that may currently be actively executing and holding the lock to finish. This is why prepFail checks that the Manager is not running, because it does not deal with this concurrency issue any way. So maybe something could get the lock by failing other operations, but those failed operation could still be running and mutating the metadata table and zookeeper after the lock is acquired.
The first problem is easy to address by calling a diff constructor, but the 2nd problem would require some new mechanism to initiate failure and wait for anything that may be running. Since all fate ops are run in the manager could have something that does the following.
- Prevents repos related to failed txids from starting to run (like if it was sitting a thread pool queue)
- Waits until repos related to failed txids that are currently running complete.
--
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@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] dlmarion commented on a change in pull request #2329: Attempt at failing FaTE transactions that are blocking this FaTE transaction from the write lock
Posted by GitBox <gi...@apache.org>.
dlmarion commented on a change in pull request #2329:
URL: https://github.com/apache/accumulo/pull/2329#discussion_r740342160
##########
File path: core/src/main/java/org/apache/accumulo/fate/zookeeper/DistributedReadWriteLock.java
##########
@@ -218,22 +237,54 @@ public boolean tryLock() {
}
SortedMap<Long,byte[]> entries = qlock.getEarlierEntries(entry);
Iterator<Entry<Long,byte[]>> iterator = entries.entrySet().iterator();
- if (!iterator.hasNext())
+ if (!iterator.hasNext()) {
throw new IllegalStateException("Did not find our own lock in the queue: " + this.entry
+ " userData " + new String(this.userData, UTF_8) + " lockType " + lockType());
- return iterator.next().getKey().equals(entry);
+ }
+ if (!failBlockers) {
+ return iterator.next().getKey().equals(entry);
+ } else {
+ ZooStore<DistributedReadWriteLock> zs;
+ try {
+ zs = new ZooStore<>(zooPath, zrw);
+ } catch (KeeperException | InterruptedException e1) {
+ log.error("Error creating zoo store", e1);
+ return false;
+ }
+ final AdminUtil<DistributedReadWriteLock> util = new AdminUtil<>();
+ boolean result = true;
+ while (iterator.hasNext()) {
+ Entry<Long,byte[]> e = iterator.next();
+ if (!e.getKey().equals(entry)) {
+ result &= util.prepFail(zs, zrw, zooManagerPath, Long.toString(e.getKey(), 16));
Review comment:
If I understand what you are saying, then I think I was working on that path. PR updated.
--
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@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] keith-turner commented on a change in pull request #2329: Attempt at failing FaTE transactions that are blocking this FaTE transaction from the write lock
Posted by GitBox <gi...@apache.org>.
keith-turner commented on a change in pull request #2329:
URL: https://github.com/apache/accumulo/pull/2329#discussion_r740272026
##########
File path: core/src/main/java/org/apache/accumulo/fate/zookeeper/DistributedReadWriteLock.java
##########
@@ -218,22 +237,54 @@ public boolean tryLock() {
}
SortedMap<Long,byte[]> entries = qlock.getEarlierEntries(entry);
Iterator<Entry<Long,byte[]>> iterator = entries.entrySet().iterator();
- if (!iterator.hasNext())
+ if (!iterator.hasNext()) {
throw new IllegalStateException("Did not find our own lock in the queue: " + this.entry
+ " userData " + new String(this.userData, UTF_8) + " lockType " + lockType());
- return iterator.next().getKey().equals(entry);
+ }
+ if (!failBlockers) {
+ return iterator.next().getKey().equals(entry);
+ } else {
+ ZooStore<DistributedReadWriteLock> zs;
+ try {
+ zs = new ZooStore<>(zooPath, zrw);
+ } catch (KeeperException | InterruptedException e1) {
+ log.error("Error creating zoo store", e1);
+ return false;
+ }
+ final AdminUtil<DistributedReadWriteLock> util = new AdminUtil<>();
+ boolean result = true;
+ while (iterator.hasNext()) {
+ Entry<Long,byte[]> e = iterator.next();
+ if (!e.getKey().equals(entry)) {
+ result &= util.prepFail(zs, zrw, zooManagerPath, Long.toString(e.getKey(), 16));
Review comment:
I think as long as the manager and this code are using the same zoostore object, then the call to reserve will address the 2nd problem I mentioned above. I see a third problem, prepFail transitions FATE ops to FAILED_IN_PROGRESS. This means those fate ops will unwind an execute their undo() operations which could modify the metadata table and zookeeper. So this could lead to other FATE ops that no longer hold the lock still modifying persisted state related to the table. Would probably need to wait for these FATE ops to transition from FAILED_IN_PROGRESS to FAILED before getting the lock. Have to be careful about how this wait is done. If all threads in the fate thread pool are waiting for other operations to transition from FAILED_IN_PROGRESS to FAILED, then that would not leave any threads available to transitions those operations leading to deadlock.
--
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@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] dlmarion commented on a change in pull request #2329: Attempt at failing FaTE transactions that are blocking this FaTE transaction from the write lock
Posted by GitBox <gi...@apache.org>.
dlmarion commented on a change in pull request #2329:
URL: https://github.com/apache/accumulo/pull/2329#discussion_r740295557
##########
File path: core/src/main/java/org/apache/accumulo/fate/zookeeper/DistributedReadWriteLock.java
##########
@@ -218,22 +237,54 @@ public boolean tryLock() {
}
SortedMap<Long,byte[]> entries = qlock.getEarlierEntries(entry);
Iterator<Entry<Long,byte[]>> iterator = entries.entrySet().iterator();
- if (!iterator.hasNext())
+ if (!iterator.hasNext()) {
throw new IllegalStateException("Did not find our own lock in the queue: " + this.entry
+ " userData " + new String(this.userData, UTF_8) + " lockType " + lockType());
- return iterator.next().getKey().equals(entry);
+ }
+ if (!failBlockers) {
+ return iterator.next().getKey().equals(entry);
+ } else {
+ ZooStore<DistributedReadWriteLock> zs;
+ try {
+ zs = new ZooStore<>(zooPath, zrw);
+ } catch (KeeperException | InterruptedException e1) {
+ log.error("Error creating zoo store", e1);
+ return false;
+ }
+ final AdminUtil<DistributedReadWriteLock> util = new AdminUtil<>();
+ boolean result = true;
+ while (iterator.hasNext()) {
+ Entry<Long,byte[]> e = iterator.next();
+ if (!e.getKey().equals(entry)) {
+ result &= util.prepFail(zs, zrw, zooManagerPath, Long.toString(e.getKey(), 16));
Review comment:
@keith-turner - you probably know this code best, can you suggest an alternate implementation?
##########
File path: core/src/main/java/org/apache/accumulo/fate/zookeeper/DistributedReadWriteLock.java
##########
@@ -218,22 +237,54 @@ public boolean tryLock() {
}
SortedMap<Long,byte[]> entries = qlock.getEarlierEntries(entry);
Iterator<Entry<Long,byte[]>> iterator = entries.entrySet().iterator();
- if (!iterator.hasNext())
+ if (!iterator.hasNext()) {
throw new IllegalStateException("Did not find our own lock in the queue: " + this.entry
+ " userData " + new String(this.userData, UTF_8) + " lockType " + lockType());
- return iterator.next().getKey().equals(entry);
+ }
+ if (!failBlockers) {
+ return iterator.next().getKey().equals(entry);
+ } else {
+ ZooStore<DistributedReadWriteLock> zs;
+ try {
+ zs = new ZooStore<>(zooPath, zrw);
+ } catch (KeeperException | InterruptedException e1) {
+ log.error("Error creating zoo store", e1);
+ return false;
+ }
+ final AdminUtil<DistributedReadWriteLock> util = new AdminUtil<>();
+ boolean result = true;
+ while (iterator.hasNext()) {
+ Entry<Long,byte[]> e = iterator.next();
+ if (!e.getKey().equals(entry)) {
+ result &= util.prepFail(zs, zrw, zooManagerPath, Long.toString(e.getKey(), 16));
Review comment:
@keith-turner - you probably know this code best, can you suggest an alternate implementation?
##########
File path: core/src/main/java/org/apache/accumulo/fate/zookeeper/DistributedReadWriteLock.java
##########
@@ -218,22 +237,54 @@ public boolean tryLock() {
}
SortedMap<Long,byte[]> entries = qlock.getEarlierEntries(entry);
Iterator<Entry<Long,byte[]>> iterator = entries.entrySet().iterator();
- if (!iterator.hasNext())
+ if (!iterator.hasNext()) {
throw new IllegalStateException("Did not find our own lock in the queue: " + this.entry
+ " userData " + new String(this.userData, UTF_8) + " lockType " + lockType());
- return iterator.next().getKey().equals(entry);
+ }
+ if (!failBlockers) {
+ return iterator.next().getKey().equals(entry);
+ } else {
+ ZooStore<DistributedReadWriteLock> zs;
+ try {
+ zs = new ZooStore<>(zooPath, zrw);
+ } catch (KeeperException | InterruptedException e1) {
+ log.error("Error creating zoo store", e1);
+ return false;
+ }
+ final AdminUtil<DistributedReadWriteLock> util = new AdminUtil<>();
+ boolean result = true;
+ while (iterator.hasNext()) {
+ Entry<Long,byte[]> e = iterator.next();
+ if (!e.getKey().equals(entry)) {
+ result &= util.prepFail(zs, zrw, zooManagerPath, Long.toString(e.getKey(), 16));
Review comment:
If I understand what you are saying, then I think I was working on that path. PR updated.
--
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@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] keith-turner commented on a change in pull request #2329: Attempt at failing FaTE transactions that are blocking this FaTE transaction from the write lock
Posted by GitBox <gi...@apache.org>.
keith-turner commented on a change in pull request #2329:
URL: https://github.com/apache/accumulo/pull/2329#discussion_r750691497
##########
File path: core/src/main/java/org/apache/accumulo/fate/zookeeper/DistributedReadWriteLock.java
##########
@@ -218,22 +228,52 @@ public boolean tryLock() {
}
SortedMap<Long,byte[]> entries = qlock.getEarlierEntries(entry);
Iterator<Entry<Long,byte[]>> iterator = entries.entrySet().iterator();
- if (!iterator.hasNext())
+ if (!iterator.hasNext()) {
throw new IllegalStateException("Did not find our own lock in the queue: " + this.entry
+ " userData " + new String(this.userData, UTF_8) + " lockType " + lockType());
- return iterator.next().getKey().equals(entry);
+ }
+ if (iterator.next().getKey().equals(entry)) {
+ return true;
+ }
+ if (preemptBlockingTransactions) {
+ log.debug("Preempting transactions that are blocking this transaction");
+ // Loop through all of the prior transactions that are waiting to
+ // acquire this write lock. If the transaction has not succeeded or failed,
+ // then fail it and return false from this method so that Utils.reserveX()
+ // will call this method again and will re-check the prior transactions.
+ iterator = entries.entrySet().iterator();
+ while (iterator.hasNext()) {
+ Entry<Long,byte[]> e = iterator.next();
+ Long priorEntry = e.getKey();
+ ParsedLock priorLock = new ParsedLock(e.getValue());
+ long priorTxid = Long.valueOf(new String(priorLock.getUserData(), UTF_8), 16);
+ if (!priorEntry.equals(entry)) {
+ log.info("Attempting to preempt blocking tx: {}", Long.toHexString(priorTxid));
+ store.cancel(priorTxid);
Review comment:
If its not currently reserved, could possibly reserved it and change the state to FAILED_IN_PROGRESS.
```java
if(reserved)
interrupt()
else
changeToFailedInProgress()
```
##########
File path: core/src/main/java/org/apache/accumulo/fate/ZooStore.java
##########
@@ -529,4 +527,16 @@ public long timeCreated(long tid) {
return dops;
}
}
+
+ /**
+ * Attempt to cancel the reserved transaction by interrupting the Thread that is running it
+ *
+ * @param tid
+ * transaction id
+ */
+ public void cancel(long tid) {
+ if (reserved.containsKey(tid)) {
Review comment:
synchronization is needed in this method.
--
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@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] dlmarion commented on pull request #2329: Attempt at failing FaTE transactions that are blocking this FaTE transaction from the write lock
Posted by GitBox <gi...@apache.org>.
dlmarion commented on pull request #2329:
URL: https://github.com/apache/accumulo/pull/2329#issuecomment-1028228525
So, there are two issues currently with this:
1. The current logic in DistributedReadWriteLock.WriteLock.tryLock doesn't handle multiple concurrent fate transactions for the same table. I believe that I fixed the logic in the method, but it's still possible for multiple fate transactions to be started on a table and all of them will get cancelled except the last one.
2. Errors that occur due to the fate transactions being cancelled/interrupted are not always making it back to the client, leaving the client in a state where its waiting indefinitely.
I've spent enough time on this trying to get it to work. I'm inclined to close this without merging it at this point.
--
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@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] dlmarion commented on a change in pull request #2329: Attempt at failing FaTE transactions that are blocking this FaTE transaction from the write lock
Posted by GitBox <gi...@apache.org>.
dlmarion commented on a change in pull request #2329:
URL: https://github.com/apache/accumulo/pull/2329#discussion_r743752382
##########
File path: core/src/main/java/org/apache/accumulo/fate/zookeeper/DistributedReadWriteLock.java
##########
@@ -218,22 +233,82 @@ public boolean tryLock() {
}
SortedMap<Long,byte[]> entries = qlock.getEarlierEntries(entry);
Iterator<Entry<Long,byte[]>> iterator = entries.entrySet().iterator();
- if (!iterator.hasNext())
+ if (!iterator.hasNext()) {
throw new IllegalStateException("Did not find our own lock in the queue: " + this.entry
+ " userData " + new String(this.userData, UTF_8) + " lockType " + lockType());
- return iterator.next().getKey().equals(entry);
+ }
+ if (!failBlockers) {
+ return iterator.next().getKey().equals(entry);
+ } else {
+ ZooStore<DistributedReadWriteLock> zs;
+ try {
+ zs = new ZooStore<>(zooPath, zrw);
+ } catch (KeeperException | InterruptedException e1) {
+ log.error("Error creating zoo store", e1);
+ return false;
+ }
+ // Loop through all of the prior transactions that are waiting to
+ // acquire this write lock. If the transaction has not succeeded or failed,
+ // then fail it and return false from this method so that Utils.reserveX()
+ // will call this method again and will re-check the prior transactions.
+ boolean result = true;
+ while (iterator.hasNext()) {
+ Entry<Long,byte[]> e = iterator.next();
+ Long txid = e.getKey();
+ if (!txid.equals(entry)) {
+ if (zs.isReserved(txid)) {
+ result &= false;
+ } else {
+ zs.reserve(txid);
+ TStatus status = zs.getStatus(txid);
+ if (status.equals(TStatus.FAILED)) {
+ result = false;
+ continue;
+ } else {
+ switch (status) {
+ case UNKNOWN:
+ // not sure what to do here, we have an invalid txid
+ break;
+ case SUCCESSFUL:
+ // already succeeded, lock should be removed
+ break;
+ case FAILED:
+ case FAILED_IN_PROGRESS:
+ // transaction failed or in process of failing
+ break;
+ default:
+ zs.setStatus(txid, TStatus.FAILED_IN_PROGRESS);
+ result &= false;
+ }
+ }
+ zs.unreserve(txid, 0);
+ }
+ } else {
+ return result && txid.equals(entry);
+ }
+ }
+ return result;
Review comment:
Addressed in 2c3ea13
##########
File path: core/src/main/java/org/apache/accumulo/fate/zookeeper/DistributedReadWriteLock.java
##########
@@ -218,22 +233,82 @@ public boolean tryLock() {
}
SortedMap<Long,byte[]> entries = qlock.getEarlierEntries(entry);
Iterator<Entry<Long,byte[]>> iterator = entries.entrySet().iterator();
- if (!iterator.hasNext())
+ if (!iterator.hasNext()) {
throw new IllegalStateException("Did not find our own lock in the queue: " + this.entry
+ " userData " + new String(this.userData, UTF_8) + " lockType " + lockType());
- return iterator.next().getKey().equals(entry);
+ }
+ if (!failBlockers) {
+ return iterator.next().getKey().equals(entry);
+ } else {
+ ZooStore<DistributedReadWriteLock> zs;
+ try {
+ zs = new ZooStore<>(zooPath, zrw);
Review comment:
Addressed in 2c3ea13
--
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@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] keith-turner commented on a change in pull request #2329: Attempt at failing FaTE transactions that are blocking this FaTE transaction from the write lock
Posted by GitBox <gi...@apache.org>.
keith-turner commented on a change in pull request #2329:
URL: https://github.com/apache/accumulo/pull/2329#discussion_r740290834
##########
File path: core/src/main/java/org/apache/accumulo/fate/zookeeper/DistributedReadWriteLock.java
##########
@@ -218,22 +237,54 @@ public boolean tryLock() {
}
SortedMap<Long,byte[]> entries = qlock.getEarlierEntries(entry);
Iterator<Entry<Long,byte[]>> iterator = entries.entrySet().iterator();
- if (!iterator.hasNext())
+ if (!iterator.hasNext()) {
throw new IllegalStateException("Did not find our own lock in the queue: " + this.entry
+ " userData " + new String(this.userData, UTF_8) + " lockType " + lockType());
- return iterator.next().getKey().equals(entry);
+ }
+ if (!failBlockers) {
+ return iterator.next().getKey().equals(entry);
+ } else {
+ ZooStore<DistributedReadWriteLock> zs;
+ try {
+ zs = new ZooStore<>(zooPath, zrw);
+ } catch (KeeperException | InterruptedException e1) {
+ log.error("Error creating zoo store", e1);
+ return false;
+ }
+ final AdminUtil<DistributedReadWriteLock> util = new AdminUtil<>();
+ boolean result = true;
+ while (iterator.hasNext()) {
+ Entry<Long,byte[]> e = iterator.next();
+ if (!e.getKey().equals(entry)) {
+ result &= util.prepFail(zs, zrw, zooManagerPath, Long.toString(e.getKey(), 16));
Review comment:
A fourth issue to consider is deadlock around the reserve() calls, which I think may not happen but I am not completely sure. Calls to tryLock are made by FATE ops that have already reserved a txid. Assume FATE op txidA is reserved and it runs and calls tryLock which attempts to reserve tixdB. If txidB is reserved and it calls tryLock trying to reserved txidA that leads to deadlock. I think this change to tryLock will only attempt to reserve transactions that started earlier than itself which seems like it would avoid deadlock. Because if A started before B, then B would attempt to reserve A but A would never attempt to reserve B because it started after.
--
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@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] keith-turner commented on a change in pull request #2329: Attempt at failing FaTE transactions that are blocking this FaTE transaction from the write lock
Posted by GitBox <gi...@apache.org>.
keith-turner commented on a change in pull request #2329:
URL: https://github.com/apache/accumulo/pull/2329#discussion_r740263645
##########
File path: core/src/main/java/org/apache/accumulo/fate/zookeeper/DistributedReadWriteLock.java
##########
@@ -218,22 +237,54 @@ public boolean tryLock() {
}
SortedMap<Long,byte[]> entries = qlock.getEarlierEntries(entry);
Iterator<Entry<Long,byte[]>> iterator = entries.entrySet().iterator();
- if (!iterator.hasNext())
+ if (!iterator.hasNext()) {
throw new IllegalStateException("Did not find our own lock in the queue: " + this.entry
+ " userData " + new String(this.userData, UTF_8) + " lockType " + lockType());
- return iterator.next().getKey().equals(entry);
+ }
+ if (!failBlockers) {
+ return iterator.next().getKey().equals(entry);
+ } else {
+ ZooStore<DistributedReadWriteLock> zs;
+ try {
+ zs = new ZooStore<>(zooPath, zrw);
+ } catch (KeeperException | InterruptedException e1) {
+ log.error("Error creating zoo store", e1);
+ return false;
+ }
+ final AdminUtil<DistributedReadWriteLock> util = new AdminUtil<>();
+ boolean result = true;
+ while (iterator.hasNext()) {
+ Entry<Long,byte[]> e = iterator.next();
+ if (!e.getKey().equals(entry)) {
+ result &= util.prepFail(zs, zrw, zooManagerPath, Long.toString(e.getKey(), 16));
Review comment:
Looking at this some more, the call to reserve() in prepFail may address the 2nd issue I mentioned above
--
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@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] dlmarion commented on a change in pull request #2329: Attempt at failing FaTE transactions that are blocking this FaTE transaction from the write lock
Posted by GitBox <gi...@apache.org>.
dlmarion commented on a change in pull request #2329:
URL: https://github.com/apache/accumulo/pull/2329#discussion_r743752263
##########
File path: core/src/main/java/org/apache/accumulo/fate/zookeeper/DistributedReadWriteLock.java
##########
@@ -218,22 +233,82 @@ public boolean tryLock() {
}
SortedMap<Long,byte[]> entries = qlock.getEarlierEntries(entry);
Iterator<Entry<Long,byte[]>> iterator = entries.entrySet().iterator();
- if (!iterator.hasNext())
+ if (!iterator.hasNext()) {
throw new IllegalStateException("Did not find our own lock in the queue: " + this.entry
+ " userData " + new String(this.userData, UTF_8) + " lockType " + lockType());
- return iterator.next().getKey().equals(entry);
+ }
+ if (!failBlockers) {
+ return iterator.next().getKey().equals(entry);
+ } else {
+ ZooStore<DistributedReadWriteLock> zs;
+ try {
+ zs = new ZooStore<>(zooPath, zrw);
+ } catch (KeeperException | InterruptedException e1) {
+ log.error("Error creating zoo store", e1);
+ return false;
+ }
+ // Loop through all of the prior transactions that are waiting to
+ // acquire this write lock. If the transaction has not succeeded or failed,
+ // then fail it and return false from this method so that Utils.reserveX()
+ // will call this method again and will re-check the prior transactions.
+ boolean result = true;
+ while (iterator.hasNext()) {
+ Entry<Long,byte[]> e = iterator.next();
+ Long txid = e.getKey();
+ if (!txid.equals(entry)) {
+ if (zs.isReserved(txid)) {
+ result &= false;
+ } else {
+ zs.reserve(txid);
+ TStatus status = zs.getStatus(txid);
+ if (status.equals(TStatus.FAILED)) {
+ result = false;
+ continue;
+ } else {
+ switch (status) {
+ case UNKNOWN:
+ // not sure what to do here, we have an invalid txid
+ break;
+ case SUCCESSFUL:
+ // already succeeded, lock should be removed
+ break;
+ case FAILED:
+ case FAILED_IN_PROGRESS:
Review comment:
Addressed in 2c3ea13
--
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@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] milleruntime commented on a change in pull request #2329: Attempt at failing FaTE transactions that are blocking this FaTE transaction from the write lock
Posted by GitBox <gi...@apache.org>.
milleruntime commented on a change in pull request #2329:
URL: https://github.com/apache/accumulo/pull/2329#discussion_r738631261
##########
File path: core/src/main/java/org/apache/accumulo/fate/zookeeper/DistributedReadWriteLock.java
##########
@@ -218,22 +237,54 @@ public boolean tryLock() {
}
SortedMap<Long,byte[]> entries = qlock.getEarlierEntries(entry);
Iterator<Entry<Long,byte[]>> iterator = entries.entrySet().iterator();
- if (!iterator.hasNext())
+ if (!iterator.hasNext()) {
throw new IllegalStateException("Did not find our own lock in the queue: " + this.entry
+ " userData " + new String(this.userData, UTF_8) + " lockType " + lockType());
- return iterator.next().getKey().equals(entry);
+ }
+ if (!failBlockers) {
+ return iterator.next().getKey().equals(entry);
+ } else {
+ ZooStore<DistributedReadWriteLock> zs;
+ try {
+ zs = new ZooStore<>(zooPath, zrw);
+ } catch (KeeperException | InterruptedException e1) {
+ log.error("Error creating zoo store", e1);
+ return false;
+ }
+ final AdminUtil<DistributedReadWriteLock> util = new AdminUtil<>();
+ boolean result = true;
+ while (iterator.hasNext()) {
+ Entry<Long,byte[]> e = iterator.next();
+ if (!e.getKey().equals(entry)) {
+ result &= util.prepFail(zs, zrw, zooManagerPath, Long.toString(e.getKey(), 16));
Review comment:
This is pretty cool change so far but I do have hesitations about tinkering with the `tryLock()` method. One of those is making calls to other methods which may block. Calling `prepFail()` directly is kinda nice, as we don't have to make any changes to it. If we know for sure it doesn't block then I guess this is OK to do within the `tryLock()`
--
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@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] dlmarion commented on a change in pull request #2329: Attempt at failing FaTE transactions that are blocking this FaTE transaction from the write lock
Posted by GitBox <gi...@apache.org>.
dlmarion commented on a change in pull request #2329:
URL: https://github.com/apache/accumulo/pull/2329#discussion_r746867231
##########
File path: core/src/main/java/org/apache/accumulo/fate/zookeeper/DistributedReadWriteLock.java
##########
@@ -218,22 +228,69 @@ public boolean tryLock() {
}
SortedMap<Long,byte[]> entries = qlock.getEarlierEntries(entry);
Iterator<Entry<Long,byte[]>> iterator = entries.entrySet().iterator();
- if (!iterator.hasNext())
+ if (!iterator.hasNext()) {
throw new IllegalStateException("Did not find our own lock in the queue: " + this.entry
+ " userData " + new String(this.userData, UTF_8) + " lockType " + lockType());
- return iterator.next().getKey().equals(entry);
+ }
+ if (iterator.next().getKey().equals(entry)) {
+ return true;
+ }
+ if (failBlockers) {
+ // Loop through all of the prior transactions that are waiting to
+ // acquire this write lock. If the transaction has not succeeded or failed,
+ // then fail it and return false from this method so that Utils.reserveX()
+ // will call this method again and will re-check the prior transactions.
+ boolean result = true;
+ while (iterator.hasNext()) {
+ Entry<Long,byte[]> e = iterator.next();
+ Long txid = e.getKey();
+ if (!txid.equals(entry)) {
+ if (store.isReserved(txid)) {
+ result &= false;
+ } else {
+ store.reserve(txid);
+ TStatus status = store.getStatus(txid);
+ if (status.equals(TStatus.FAILED)) {
+ result = false;
+ continue;
+ } else {
+ switch (status) {
+ case FAILED_IN_PROGRESS:
+ case IN_PROGRESS:
+ store.setStatus(txid, TStatus.FAILED_IN_PROGRESS);
+ result &= false;
+ break;
+ default:
+ log.error("Attempting to fail a transaction in an unhandled state: {}", status);
+ }
+ }
+ store.unreserve(txid, 0);
Review comment:
Addressed in 2b8fc42ebd19b8c9b3861ab9d5abaa3d34a6888f
--
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@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] ctubbsii edited a comment on pull request #2329: Attempt at failing FaTE transactions that are blocking this FaTE transaction from the write lock
Posted by GitBox <gi...@apache.org>.
ctubbsii edited a comment on pull request #2329:
URL: https://github.com/apache/accumulo/pull/2329#issuecomment-1027367812
This already seems like it's getting pretty complicated. Is it even safe, in the general case, to fail fate operations this way? I feel like a human should probably be interactively responsible for cancelling fate operations in order to unblock their preferred one, so they can accept the risks directly.
--
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@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] dlmarion commented on pull request #2329: Attempt at failing FaTE transactions that are blocking this FaTE transaction from the write lock
Posted by GitBox <gi...@apache.org>.
dlmarion commented on pull request #2329:
URL: https://github.com/apache/accumulo/pull/2329#issuecomment-1027211091
In my rerun of the ITs, only one test class failed, ConcurrentDeleteTableIT. I traced it through and it is failing because of this change. In testConcurrentFateOpsWithDelete there is an expectation that the table is deleted at the end of the test, because a delete operation is executed as part of the test. The table is not deleted, because the delete operation is cancelled due to a subsequent merge (or other type) of operation. The right way to handle this is to *not* cancel certain types of operations in DistributedReadWriteLock.WriteLock.tryLock(), but I don't have access to the FateOperation type at this level. For example, we would want a MERGE to cancel any COMPACTION, but not a DELETE. It does not appear that the FateOperation type is encoded in the data stored in ZK. If we could pass the TableOperation to Utils.getLock() then we could append/prepend the TableOperation to the lockData.
--
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@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] dlmarion commented on pull request #2329: Attempt at failing FaTE transactions that are blocking this FaTE transaction from the write lock
Posted by GitBox <gi...@apache.org>.
dlmarion commented on pull request #2329:
URL: https://github.com/apache/accumulo/pull/2329#issuecomment-1025669662
FateIT is relevant to this PR. It failed because ZooKeeperTestingServer failed to start after getting several BindExceptions's.
--
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@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] dlmarion commented on pull request #2329: Attempt at failing FaTE transactions that are blocking this FaTE transaction from the write lock
Posted by GitBox <gi...@apache.org>.
dlmarion commented on pull request #2329:
URL: https://github.com/apache/accumulo/pull/2329#issuecomment-1026093610
@ctubbsii - I reran the ITs and FateIT passed this time. Maybe a transient port issue.
--
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@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] keith-turner commented on a change in pull request #2329: Attempt at failing FaTE transactions that are blocking this FaTE transaction from the write lock
Posted by GitBox <gi...@apache.org>.
keith-turner commented on a change in pull request #2329:
URL: https://github.com/apache/accumulo/pull/2329#discussion_r740333841
##########
File path: core/src/main/java/org/apache/accumulo/fate/zookeeper/DistributedReadWriteLock.java
##########
@@ -218,22 +237,54 @@ public boolean tryLock() {
}
SortedMap<Long,byte[]> entries = qlock.getEarlierEntries(entry);
Iterator<Entry<Long,byte[]>> iterator = entries.entrySet().iterator();
- if (!iterator.hasNext())
+ if (!iterator.hasNext()) {
throw new IllegalStateException("Did not find our own lock in the queue: " + this.entry
+ " userData " + new String(this.userData, UTF_8) + " lockType " + lockType());
- return iterator.next().getKey().equals(entry);
+ }
+ if (!failBlockers) {
+ return iterator.next().getKey().equals(entry);
+ } else {
+ ZooStore<DistributedReadWriteLock> zs;
+ try {
+ zs = new ZooStore<>(zooPath, zrw);
+ } catch (KeeperException | InterruptedException e1) {
+ log.error("Error creating zoo store", e1);
+ return false;
+ }
+ final AdminUtil<DistributedReadWriteLock> util = new AdminUtil<>();
+ boolean result = true;
+ while (iterator.hasNext()) {
+ Entry<Long,byte[]> e = iterator.next();
+ if (!e.getKey().equals(entry)) {
+ result &= util.prepFail(zs, zrw, zooManagerPath, Long.toString(e.getKey(), 16));
Review comment:
@dlmarion could possibly do something like the following... where the code in `failPrecedingOperations()` would go through and transition operations to FAILED_IN_PROGRESS... also thinking that code could transition NEW things straight to FAILED instead of to FAILED_IN_PROGRESS but not sure. However there is not waiting on these other operations. The FATE framework will reserve a txid, take the top repo off the stack, and then call isReady(). If isReady() returns non-zero it will unreserve and the txid and come back to it later. So if isReady transitions things to FAILED_IN_PROGRESS then hopefully a later call to isReady can get the lock sooner.
```java
class SomeRepo extends Repo {
public long isReady(long tid,...) {
long waitTime = tryToGetTableWriteLock(...);
if(waitTime > 0 && failBlockers) {
// we did not get the the table write lock, so lets initiate the failure of all operations that precede us in the lock queue.... hopefully after these fail they will release the lock and we will get the lock on a later call to isRead()
failPrecedingOperations(tid)
}
return waitTime;
}
}
```
--
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@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] keith-turner commented on a change in pull request #2329: Attempt at failing FaTE transactions that are blocking this FaTE transaction from the write lock
Posted by GitBox <gi...@apache.org>.
keith-turner commented on a change in pull request #2329:
URL: https://github.com/apache/accumulo/pull/2329#discussion_r740272026
##########
File path: core/src/main/java/org/apache/accumulo/fate/zookeeper/DistributedReadWriteLock.java
##########
@@ -218,22 +237,54 @@ public boolean tryLock() {
}
SortedMap<Long,byte[]> entries = qlock.getEarlierEntries(entry);
Iterator<Entry<Long,byte[]>> iterator = entries.entrySet().iterator();
- if (!iterator.hasNext())
+ if (!iterator.hasNext()) {
throw new IllegalStateException("Did not find our own lock in the queue: " + this.entry
+ " userData " + new String(this.userData, UTF_8) + " lockType " + lockType());
- return iterator.next().getKey().equals(entry);
+ }
+ if (!failBlockers) {
+ return iterator.next().getKey().equals(entry);
+ } else {
+ ZooStore<DistributedReadWriteLock> zs;
+ try {
+ zs = new ZooStore<>(zooPath, zrw);
+ } catch (KeeperException | InterruptedException e1) {
+ log.error("Error creating zoo store", e1);
+ return false;
+ }
+ final AdminUtil<DistributedReadWriteLock> util = new AdminUtil<>();
+ boolean result = true;
+ while (iterator.hasNext()) {
+ Entry<Long,byte[]> e = iterator.next();
+ if (!e.getKey().equals(entry)) {
+ result &= util.prepFail(zs, zrw, zooManagerPath, Long.toString(e.getKey(), 16));
Review comment:
I think as long as the manager and this code are using the same zoostore object, then the call to reserve will address the 2nd problem I mentioned above. I see a third problem, prepFail transitions FATE ops to FAILED_IN_PROGRESS. This means those fate ops will unwind and execute their undo() operations which could modify the metadata table and zookeeper. So this could lead to other FATE ops that no longer hold the lock still modifying persisted state related to the table. Would probably need to wait for these FATE ops to transition from FAILED_IN_PROGRESS to FAILED before getting the lock. Have to be careful about how this wait is done. If all threads in the fate thread pool are waiting for other operations to transition from FAILED_IN_PROGRESS to FAILED, then that would not leave any threads available to transitions those operations leading to deadlock.
--
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@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] dlmarion commented on a change in pull request #2329: Attempt at failing FaTE transactions that are blocking this FaTE transaction from the write lock
Posted by GitBox <gi...@apache.org>.
dlmarion commented on a change in pull request #2329:
URL: https://github.com/apache/accumulo/pull/2329#discussion_r746867097
##########
File path: core/src/main/java/org/apache/accumulo/fate/zookeeper/DistributedReadWriteLock.java
##########
@@ -218,22 +228,69 @@ public boolean tryLock() {
}
SortedMap<Long,byte[]> entries = qlock.getEarlierEntries(entry);
Iterator<Entry<Long,byte[]>> iterator = entries.entrySet().iterator();
- if (!iterator.hasNext())
+ if (!iterator.hasNext()) {
throw new IllegalStateException("Did not find our own lock in the queue: " + this.entry
+ " userData " + new String(this.userData, UTF_8) + " lockType " + lockType());
- return iterator.next().getKey().equals(entry);
+ }
+ if (iterator.next().getKey().equals(entry)) {
+ return true;
+ }
+ if (failBlockers) {
+ // Loop through all of the prior transactions that are waiting to
+ // acquire this write lock. If the transaction has not succeeded or failed,
+ // then fail it and return false from this method so that Utils.reserveX()
+ // will call this method again and will re-check the prior transactions.
+ boolean result = true;
+ while (iterator.hasNext()) {
+ Entry<Long,byte[]> e = iterator.next();
+ Long txid = e.getKey();
+ if (!txid.equals(entry)) {
+ if (store.isReserved(txid)) {
+ result &= false;
+ } else {
+ store.reserve(txid);
+ TStatus status = store.getStatus(txid);
+ if (status.equals(TStatus.FAILED)) {
+ result = false;
+ continue;
+ } else {
+ switch (status) {
+ case FAILED_IN_PROGRESS:
+ case IN_PROGRESS:
+ store.setStatus(txid, TStatus.FAILED_IN_PROGRESS);
+ result &= false;
+ break;
+ default:
+ log.error("Attempting to fail a transaction in an unhandled state: {}", status);
+ }
+ }
+ store.unreserve(txid, 0);
+ }
+ } else {
+ return result && txid.equals(entry);
+ }
+ }
+ return result;
Review comment:
Addressed in 2b8fc42ebd19b8c9b3861ab9d5abaa3d34a6888f
--
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@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] dlmarion commented on a change in pull request #2329: Attempt at failing FaTE transactions that are blocking this FaTE transaction from the write lock
Posted by GitBox <gi...@apache.org>.
dlmarion commented on a change in pull request #2329:
URL: https://github.com/apache/accumulo/pull/2329#discussion_r746862199
##########
File path: core/src/main/java/org/apache/accumulo/fate/zookeeper/DistributedReadWriteLock.java
##########
@@ -218,22 +228,69 @@ public boolean tryLock() {
}
SortedMap<Long,byte[]> entries = qlock.getEarlierEntries(entry);
Iterator<Entry<Long,byte[]>> iterator = entries.entrySet().iterator();
- if (!iterator.hasNext())
+ if (!iterator.hasNext()) {
throw new IllegalStateException("Did not find our own lock in the queue: " + this.entry
+ " userData " + new String(this.userData, UTF_8) + " lockType " + lockType());
- return iterator.next().getKey().equals(entry);
+ }
+ if (iterator.next().getKey().equals(entry)) {
+ return true;
+ }
+ if (failBlockers) {
+ // Loop through all of the prior transactions that are waiting to
+ // acquire this write lock. If the transaction has not succeeded or failed,
+ // then fail it and return false from this method so that Utils.reserveX()
+ // will call this method again and will re-check the prior transactions.
+ boolean result = true;
Review comment:
So, in the case where we are failingBlockers, WriteLock.tryLock() would always return false and then the FaTE framework would try again. When all of the blockers have been failed and removed, then this txid would be the first in the list. Is that what you are saying? Sounds good to me.
--
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@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] dlmarion commented on a change in pull request #2329: Attempt at failing FaTE transactions that are blocking this FaTE transaction from the write lock
Posted by GitBox <gi...@apache.org>.
dlmarion commented on a change in pull request #2329:
URL: https://github.com/apache/accumulo/pull/2329#discussion_r746867336
##########
File path: core/src/main/java/org/apache/accumulo/fate/zookeeper/DistributedReadWriteLock.java
##########
@@ -218,22 +228,69 @@ public boolean tryLock() {
}
SortedMap<Long,byte[]> entries = qlock.getEarlierEntries(entry);
Iterator<Entry<Long,byte[]>> iterator = entries.entrySet().iterator();
- if (!iterator.hasNext())
+ if (!iterator.hasNext()) {
throw new IllegalStateException("Did not find our own lock in the queue: " + this.entry
+ " userData " + new String(this.userData, UTF_8) + " lockType " + lockType());
- return iterator.next().getKey().equals(entry);
+ }
+ if (iterator.next().getKey().equals(entry)) {
+ return true;
+ }
+ if (failBlockers) {
+ // Loop through all of the prior transactions that are waiting to
+ // acquire this write lock. If the transaction has not succeeded or failed,
+ // then fail it and return false from this method so that Utils.reserveX()
+ // will call this method again and will re-check the prior transactions.
+ boolean result = true;
Review comment:
Addressed in 2b8fc42ebd19b8c9b3861ab9d5abaa3d34a6888f
##########
File path: core/src/main/java/org/apache/accumulo/fate/zookeeper/DistributedReadWriteLock.java
##########
@@ -218,22 +228,69 @@ public boolean tryLock() {
}
SortedMap<Long,byte[]> entries = qlock.getEarlierEntries(entry);
Iterator<Entry<Long,byte[]>> iterator = entries.entrySet().iterator();
- if (!iterator.hasNext())
+ if (!iterator.hasNext()) {
throw new IllegalStateException("Did not find our own lock in the queue: " + this.entry
+ " userData " + new String(this.userData, UTF_8) + " lockType " + lockType());
- return iterator.next().getKey().equals(entry);
+ }
+ if (iterator.next().getKey().equals(entry)) {
+ return true;
+ }
+ if (failBlockers) {
+ // Loop through all of the prior transactions that are waiting to
+ // acquire this write lock. If the transaction has not succeeded or failed,
+ // then fail it and return false from this method so that Utils.reserveX()
+ // will call this method again and will re-check the prior transactions.
+ boolean result = true;
+ while (iterator.hasNext()) {
+ Entry<Long,byte[]> e = iterator.next();
+ Long txid = e.getKey();
+ if (!txid.equals(entry)) {
+ if (store.isReserved(txid)) {
+ result &= false;
+ } else {
+ store.reserve(txid);
Review comment:
Addressed in 2b8fc42ebd19b8c9b3861ab9d5abaa3d34a6888f
--
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@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] milleruntime commented on a change in pull request #2329: Attempt at failing FaTE transactions that are blocking this FaTE transaction from the write lock
Posted by GitBox <gi...@apache.org>.
milleruntime commented on a change in pull request #2329:
URL: https://github.com/apache/accumulo/pull/2329#discussion_r791869863
##########
File path: core/src/main/java/org/apache/accumulo/fate/zookeeper/DistributedReadWriteLock.java
##########
@@ -218,22 +229,68 @@ public boolean tryLock() {
}
SortedMap<Long,byte[]> entries = qlock.getEarlierEntries(entry);
Iterator<Entry<Long,byte[]>> iterator = entries.entrySet().iterator();
- if (!iterator.hasNext())
+ if (!iterator.hasNext()) {
throw new IllegalStateException("Did not find our own lock in the queue: " + this.entry
+ " userData " + new String(this.userData, UTF_8) + " lockType " + lockType());
- return iterator.next().getKey().equals(entry);
+ }
+ if (iterator.next().getKey().equals(entry)) {
+ return true;
+ }
+ if (preemptBlockingTransactions) {
+ log.info("Preempting transactions that are blocking this transaction");
Review comment:
In other parts of the code you call this boolean `cancelWriteLockBlockers`. I think picking one consistent name would be better. I like `cancelWriteLockBlockers` better than `preemptBlockingTransactions`. I have a PR against your branch with some minor changes that includes this rename.
--
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@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] dlmarion commented on pull request #2329: Attempt at failing FaTE transactions that are blocking this FaTE transaction from the write lock
Posted by GitBox <gi...@apache.org>.
dlmarion commented on pull request #2329:
URL: https://github.com/apache/accumulo/pull/2329#issuecomment-952223372
This does not compile, just an illustration of the approach I was thinking of.
--
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@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] dlmarion commented on a change in pull request #2329: Attempt at failing FaTE transactions that are blocking this FaTE transaction from the write lock
Posted by GitBox <gi...@apache.org>.
dlmarion commented on a change in pull request #2329:
URL: https://github.com/apache/accumulo/pull/2329#discussion_r740295557
##########
File path: core/src/main/java/org/apache/accumulo/fate/zookeeper/DistributedReadWriteLock.java
##########
@@ -218,22 +237,54 @@ public boolean tryLock() {
}
SortedMap<Long,byte[]> entries = qlock.getEarlierEntries(entry);
Iterator<Entry<Long,byte[]>> iterator = entries.entrySet().iterator();
- if (!iterator.hasNext())
+ if (!iterator.hasNext()) {
throw new IllegalStateException("Did not find our own lock in the queue: " + this.entry
+ " userData " + new String(this.userData, UTF_8) + " lockType " + lockType());
- return iterator.next().getKey().equals(entry);
+ }
+ if (!failBlockers) {
+ return iterator.next().getKey().equals(entry);
+ } else {
+ ZooStore<DistributedReadWriteLock> zs;
+ try {
+ zs = new ZooStore<>(zooPath, zrw);
+ } catch (KeeperException | InterruptedException e1) {
+ log.error("Error creating zoo store", e1);
+ return false;
+ }
+ final AdminUtil<DistributedReadWriteLock> util = new AdminUtil<>();
+ boolean result = true;
+ while (iterator.hasNext()) {
+ Entry<Long,byte[]> e = iterator.next();
+ if (!e.getKey().equals(entry)) {
+ result &= util.prepFail(zs, zrw, zooManagerPath, Long.toString(e.getKey(), 16));
Review comment:
@keith-turner - you probably know this code best, can you suggest an alternate implementation?
--
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@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] dlmarion commented on a change in pull request #2329: Attempt at failing FaTE transactions that are blocking this FaTE transaction from the write lock
Posted by GitBox <gi...@apache.org>.
dlmarion commented on a change in pull request #2329:
URL: https://github.com/apache/accumulo/pull/2329#discussion_r743752263
##########
File path: core/src/main/java/org/apache/accumulo/fate/zookeeper/DistributedReadWriteLock.java
##########
@@ -218,22 +233,82 @@ public boolean tryLock() {
}
SortedMap<Long,byte[]> entries = qlock.getEarlierEntries(entry);
Iterator<Entry<Long,byte[]>> iterator = entries.entrySet().iterator();
- if (!iterator.hasNext())
+ if (!iterator.hasNext()) {
throw new IllegalStateException("Did not find our own lock in the queue: " + this.entry
+ " userData " + new String(this.userData, UTF_8) + " lockType " + lockType());
- return iterator.next().getKey().equals(entry);
+ }
+ if (!failBlockers) {
+ return iterator.next().getKey().equals(entry);
+ } else {
+ ZooStore<DistributedReadWriteLock> zs;
+ try {
+ zs = new ZooStore<>(zooPath, zrw);
+ } catch (KeeperException | InterruptedException e1) {
+ log.error("Error creating zoo store", e1);
+ return false;
+ }
+ // Loop through all of the prior transactions that are waiting to
+ // acquire this write lock. If the transaction has not succeeded or failed,
+ // then fail it and return false from this method so that Utils.reserveX()
+ // will call this method again and will re-check the prior transactions.
+ boolean result = true;
+ while (iterator.hasNext()) {
+ Entry<Long,byte[]> e = iterator.next();
+ Long txid = e.getKey();
+ if (!txid.equals(entry)) {
+ if (zs.isReserved(txid)) {
+ result &= false;
+ } else {
+ zs.reserve(txid);
+ TStatus status = zs.getStatus(txid);
+ if (status.equals(TStatus.FAILED)) {
+ result = false;
+ continue;
+ } else {
+ switch (status) {
+ case UNKNOWN:
+ // not sure what to do here, we have an invalid txid
+ break;
+ case SUCCESSFUL:
+ // already succeeded, lock should be removed
+ break;
+ case FAILED:
+ case FAILED_IN_PROGRESS:
Review comment:
Addressed in 2c3ea13
##########
File path: core/src/main/java/org/apache/accumulo/fate/zookeeper/DistributedReadWriteLock.java
##########
@@ -218,22 +233,82 @@ public boolean tryLock() {
}
SortedMap<Long,byte[]> entries = qlock.getEarlierEntries(entry);
Iterator<Entry<Long,byte[]>> iterator = entries.entrySet().iterator();
- if (!iterator.hasNext())
+ if (!iterator.hasNext()) {
throw new IllegalStateException("Did not find our own lock in the queue: " + this.entry
+ " userData " + new String(this.userData, UTF_8) + " lockType " + lockType());
- return iterator.next().getKey().equals(entry);
+ }
+ if (!failBlockers) {
+ return iterator.next().getKey().equals(entry);
+ } else {
+ ZooStore<DistributedReadWriteLock> zs;
+ try {
+ zs = new ZooStore<>(zooPath, zrw);
+ } catch (KeeperException | InterruptedException e1) {
+ log.error("Error creating zoo store", e1);
+ return false;
+ }
+ // Loop through all of the prior transactions that are waiting to
+ // acquire this write lock. If the transaction has not succeeded or failed,
+ // then fail it and return false from this method so that Utils.reserveX()
+ // will call this method again and will re-check the prior transactions.
+ boolean result = true;
+ while (iterator.hasNext()) {
+ Entry<Long,byte[]> e = iterator.next();
+ Long txid = e.getKey();
+ if (!txid.equals(entry)) {
+ if (zs.isReserved(txid)) {
+ result &= false;
+ } else {
+ zs.reserve(txid);
+ TStatus status = zs.getStatus(txid);
+ if (status.equals(TStatus.FAILED)) {
+ result = false;
+ continue;
+ } else {
+ switch (status) {
+ case UNKNOWN:
+ // not sure what to do here, we have an invalid txid
+ break;
+ case SUCCESSFUL:
+ // already succeeded, lock should be removed
+ break;
+ case FAILED:
+ case FAILED_IN_PROGRESS:
+ // transaction failed or in process of failing
+ break;
+ default:
+ zs.setStatus(txid, TStatus.FAILED_IN_PROGRESS);
+ result &= false;
+ }
+ }
+ zs.unreserve(txid, 0);
+ }
+ } else {
+ return result && txid.equals(entry);
+ }
+ }
+ return result;
Review comment:
Addressed in 2c3ea13
##########
File path: core/src/main/java/org/apache/accumulo/fate/zookeeper/DistributedReadWriteLock.java
##########
@@ -218,22 +233,82 @@ public boolean tryLock() {
}
SortedMap<Long,byte[]> entries = qlock.getEarlierEntries(entry);
Iterator<Entry<Long,byte[]>> iterator = entries.entrySet().iterator();
- if (!iterator.hasNext())
+ if (!iterator.hasNext()) {
throw new IllegalStateException("Did not find our own lock in the queue: " + this.entry
+ " userData " + new String(this.userData, UTF_8) + " lockType " + lockType());
- return iterator.next().getKey().equals(entry);
+ }
+ if (!failBlockers) {
+ return iterator.next().getKey().equals(entry);
+ } else {
+ ZooStore<DistributedReadWriteLock> zs;
+ try {
+ zs = new ZooStore<>(zooPath, zrw);
Review comment:
Addressed in 2c3ea13
--
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@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] dlmarion commented on pull request #2329: Attempt at failing FaTE transactions that are blocking this FaTE transaction from the write lock
Posted by GitBox <gi...@apache.org>.
dlmarion commented on pull request #2329:
URL: https://github.com/apache/accumulo/pull/2329#issuecomment-1030381828
I'm closing this in favor of #2467
--
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@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] dlmarion commented on a change in pull request #2329: Attempt at failing FaTE transactions that are blocking this FaTE transaction from the write lock
Posted by GitBox <gi...@apache.org>.
dlmarion commented on a change in pull request #2329:
URL: https://github.com/apache/accumulo/pull/2329#discussion_r751166851
##########
File path: core/src/main/java/org/apache/accumulo/fate/zookeeper/DistributedReadWriteLock.java
##########
@@ -218,22 +228,52 @@ public boolean tryLock() {
}
SortedMap<Long,byte[]> entries = qlock.getEarlierEntries(entry);
Iterator<Entry<Long,byte[]>> iterator = entries.entrySet().iterator();
- if (!iterator.hasNext())
+ if (!iterator.hasNext()) {
throw new IllegalStateException("Did not find our own lock in the queue: " + this.entry
+ " userData " + new String(this.userData, UTF_8) + " lockType " + lockType());
- return iterator.next().getKey().equals(entry);
+ }
+ if (iterator.next().getKey().equals(entry)) {
+ return true;
+ }
+ if (preemptBlockingTransactions) {
+ log.debug("Preempting transactions that are blocking this transaction");
+ // Loop through all of the prior transactions that are waiting to
+ // acquire this write lock. If the transaction has not succeeded or failed,
+ // then fail it and return false from this method so that Utils.reserveX()
+ // will call this method again and will re-check the prior transactions.
+ iterator = entries.entrySet().iterator();
+ while (iterator.hasNext()) {
+ Entry<Long,byte[]> e = iterator.next();
+ Long priorEntry = e.getKey();
+ ParsedLock priorLock = new ParsedLock(e.getValue());
+ long priorTxid = Long.valueOf(new String(priorLock.getUserData(), UTF_8), 16);
+ if (!priorEntry.equals(entry)) {
+ log.info("Attempting to preempt blocking tx: {}", Long.toHexString(priorTxid));
+ store.cancel(priorTxid);
Review comment:
Addressed in d5df49354bf83d9174445c5e669936bb5cdf473e
--
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@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] dlmarion commented on a change in pull request #2329: Attempt at failing FaTE transactions that are blocking this FaTE transaction from the write lock
Posted by GitBox <gi...@apache.org>.
dlmarion commented on a change in pull request #2329:
URL: https://github.com/apache/accumulo/pull/2329#discussion_r753226264
##########
File path: core/src/main/java/org/apache/accumulo/fate/zookeeper/DistributedReadWriteLock.java
##########
@@ -218,22 +229,63 @@ public boolean tryLock() {
}
SortedMap<Long,byte[]> entries = qlock.getEarlierEntries(entry);
Iterator<Entry<Long,byte[]>> iterator = entries.entrySet().iterator();
- if (!iterator.hasNext())
+ if (!iterator.hasNext()) {
throw new IllegalStateException("Did not find our own lock in the queue: " + this.entry
+ " userData " + new String(this.userData, UTF_8) + " lockType " + lockType());
- return iterator.next().getKey().equals(entry);
+ }
+ if (iterator.next().getKey().equals(entry)) {
+ return true;
+ }
+ if (preemptBlockingTransactions) {
+ log.info("Preempting transactions that are blocking this transaction");
+ // Loop through all of the prior transactions that are waiting to
+ // acquire this write lock. If the transaction has not succeeded or failed,
+ // then fail it and return false from this method so that Utils.reserveX()
+ // will call this method again and will re-check the prior transactions.
+ iterator = entries.entrySet().iterator();
+ while (iterator.hasNext()) {
+ Entry<Long,byte[]> e = iterator.next();
+ Long priorEntry = e.getKey();
+ ParsedLock priorLock = new ParsedLock(e.getValue());
+ long priorTxid = Long.valueOf(new String(priorLock.getUserData(), UTF_8), 16);
+ if (!priorEntry.equals(entry)) {
+ log.info("Attempting to preempt blocking tx: {}", Long.toHexString(priorTxid));
+ // If the prior tx id is reserved, then it's running. We can't reserve it and set its
+ // status, we have to interrupt the thread.
+ if (store.isReserved(priorTxid)) {
+ store.cancel(priorTxid);
+ } else {
+ try {
+ store.reserve(priorTxid);
+ store.setStatus(priorTxid, TStatus.FAILED_IN_PROGRESS);
+ } finally {
+ store.unreserve(priorTxid, 0);
+ }
+ }
Review comment:
Good feedback, I will implement these changes.
--
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@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] dlmarion edited a comment on pull request #2329: Attempt at failing FaTE transactions that are blocking this FaTE transaction from the write lock
Posted by GitBox <gi...@apache.org>.
dlmarion edited a comment on pull request #2329:
URL: https://github.com/apache/accumulo/pull/2329#issuecomment-1025669662
FateIT is relevant to this PR. It failed because ZooKeeperTestingServer failed to start after getting several BindExceptions's. Looking at the other two failed ITs, I'm not sure that their failure is related to the changes here. I was able to reproduce the failures locally, but there are log messages associated with this change that should exist in the Manager log that are not there.
--
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@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] keith-turner commented on a change in pull request #2329: Attempt at failing FaTE transactions that are blocking this FaTE transaction from the write lock
Posted by GitBox <gi...@apache.org>.
keith-turner commented on a change in pull request #2329:
URL: https://github.com/apache/accumulo/pull/2329#discussion_r741111823
##########
File path: core/src/main/java/org/apache/accumulo/fate/zookeeper/DistributedReadWriteLock.java
##########
@@ -218,22 +233,82 @@ public boolean tryLock() {
}
SortedMap<Long,byte[]> entries = qlock.getEarlierEntries(entry);
Iterator<Entry<Long,byte[]>> iterator = entries.entrySet().iterator();
- if (!iterator.hasNext())
+ if (!iterator.hasNext()) {
throw new IllegalStateException("Did not find our own lock in the queue: " + this.entry
+ " userData " + new String(this.userData, UTF_8) + " lockType " + lockType());
- return iterator.next().getKey().equals(entry);
+ }
+ if (!failBlockers) {
+ return iterator.next().getKey().equals(entry);
+ } else {
+ ZooStore<DistributedReadWriteLock> zs;
+ try {
+ zs = new ZooStore<>(zooPath, zrw);
Review comment:
Need to use the same ZooStore object that the manager/Fate framework is using, or will not get mutual exclusion when reserving transactions. Look at the impl of zs.reserve(long), by creating a new object will be using a different hashset and synchronize on diff obj monitors than the manager/Fate framework.
##########
File path: core/src/main/java/org/apache/accumulo/fate/zookeeper/DistributedReadWriteLock.java
##########
@@ -218,22 +233,82 @@ public boolean tryLock() {
}
SortedMap<Long,byte[]> entries = qlock.getEarlierEntries(entry);
Iterator<Entry<Long,byte[]>> iterator = entries.entrySet().iterator();
- if (!iterator.hasNext())
+ if (!iterator.hasNext()) {
throw new IllegalStateException("Did not find our own lock in the queue: " + this.entry
+ " userData " + new String(this.userData, UTF_8) + " lockType " + lockType());
- return iterator.next().getKey().equals(entry);
+ }
+ if (!failBlockers) {
+ return iterator.next().getKey().equals(entry);
+ } else {
+ ZooStore<DistributedReadWriteLock> zs;
+ try {
+ zs = new ZooStore<>(zooPath, zrw);
+ } catch (KeeperException | InterruptedException e1) {
+ log.error("Error creating zoo store", e1);
+ return false;
+ }
+ // Loop through all of the prior transactions that are waiting to
+ // acquire this write lock. If the transaction has not succeeded or failed,
+ // then fail it and return false from this method so that Utils.reserveX()
+ // will call this method again and will re-check the prior transactions.
+ boolean result = true;
+ while (iterator.hasNext()) {
+ Entry<Long,byte[]> e = iterator.next();
+ Long txid = e.getKey();
+ if (!txid.equals(entry)) {
+ if (zs.isReserved(txid)) {
+ result &= false;
+ } else {
+ zs.reserve(txid);
+ TStatus status = zs.getStatus(txid);
+ if (status.equals(TStatus.FAILED)) {
+ result = false;
+ continue;
+ } else {
+ switch (status) {
+ case UNKNOWN:
+ // not sure what to do here, we have an invalid txid
+ break;
+ case SUCCESSFUL:
+ // already succeeded, lock should be removed
+ break;
+ case FAILED:
+ case FAILED_IN_PROGRESS:
+ // transaction failed or in process of failing
+ break;
+ default:
+ zs.setStatus(txid, TStatus.FAILED_IN_PROGRESS);
+ result &= false;
+ }
+ }
+ zs.unreserve(txid, 0);
+ }
+ } else {
+ return result && txid.equals(entry);
+ }
+ }
+ return result;
Review comment:
The following high level structure would be nice, where if we are first in the queue return true no matter what fail blockers is... also never return true unless we are the 1st in the queue. Can't really apply this suggestion, just used a suggestion to make clear the scope I was thinking about.
```suggestion
if( iterator.next().getKey().equals(entry)){
return true;
}
if (failBlockers) {
// attempt to fail anything before this lock entry in the queue.. but never return true unless ours is the 1st
}
return false;
```
##########
File path: core/src/main/java/org/apache/accumulo/fate/zookeeper/DistributedReadWriteLock.java
##########
@@ -218,22 +233,82 @@ public boolean tryLock() {
}
SortedMap<Long,byte[]> entries = qlock.getEarlierEntries(entry);
Iterator<Entry<Long,byte[]>> iterator = entries.entrySet().iterator();
- if (!iterator.hasNext())
+ if (!iterator.hasNext()) {
throw new IllegalStateException("Did not find our own lock in the queue: " + this.entry
+ " userData " + new String(this.userData, UTF_8) + " lockType " + lockType());
- return iterator.next().getKey().equals(entry);
+ }
+ if (!failBlockers) {
+ return iterator.next().getKey().equals(entry);
+ } else {
+ ZooStore<DistributedReadWriteLock> zs;
+ try {
+ zs = new ZooStore<>(zooPath, zrw);
+ } catch (KeeperException | InterruptedException e1) {
+ log.error("Error creating zoo store", e1);
+ return false;
+ }
+ // Loop through all of the prior transactions that are waiting to
+ // acquire this write lock. If the transaction has not succeeded or failed,
+ // then fail it and return false from this method so that Utils.reserveX()
+ // will call this method again and will re-check the prior transactions.
+ boolean result = true;
+ while (iterator.hasNext()) {
+ Entry<Long,byte[]> e = iterator.next();
+ Long txid = e.getKey();
+ if (!txid.equals(entry)) {
+ if (zs.isReserved(txid)) {
+ result &= false;
+ } else {
+ zs.reserve(txid);
+ TStatus status = zs.getStatus(txid);
+ if (status.equals(TStatus.FAILED)) {
+ result = false;
+ continue;
+ } else {
+ switch (status) {
+ case UNKNOWN:
+ // not sure what to do here, we have an invalid txid
+ break;
+ case SUCCESSFUL:
+ // already succeeded, lock should be removed
+ break;
+ case FAILED:
+ case FAILED_IN_PROGRESS:
Review comment:
Investigation would be needed to confirm this, but I suspect we should only expect to see IN_PROGRESS and FAILED_IN_PROGRESS transactions actually holding the lock. Anything else seems like and indication of a more general problem that we should not attempt to react to here. We could log a warn/error if we see them, but it may not be a good idea to let the lock be acquired when the system is an unexpected state. This kinda goes w/ my other comment of always returning false when we are not first in the queue.
--
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@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] dlmarion commented on a change in pull request #2329: Attempt at failing FaTE transactions that are blocking this FaTE transaction from the write lock
Posted by GitBox <gi...@apache.org>.
dlmarion commented on a change in pull request #2329:
URL: https://github.com/apache/accumulo/pull/2329#discussion_r743752263
##########
File path: core/src/main/java/org/apache/accumulo/fate/zookeeper/DistributedReadWriteLock.java
##########
@@ -218,22 +233,82 @@ public boolean tryLock() {
}
SortedMap<Long,byte[]> entries = qlock.getEarlierEntries(entry);
Iterator<Entry<Long,byte[]>> iterator = entries.entrySet().iterator();
- if (!iterator.hasNext())
+ if (!iterator.hasNext()) {
throw new IllegalStateException("Did not find our own lock in the queue: " + this.entry
+ " userData " + new String(this.userData, UTF_8) + " lockType " + lockType());
- return iterator.next().getKey().equals(entry);
+ }
+ if (!failBlockers) {
+ return iterator.next().getKey().equals(entry);
+ } else {
+ ZooStore<DistributedReadWriteLock> zs;
+ try {
+ zs = new ZooStore<>(zooPath, zrw);
+ } catch (KeeperException | InterruptedException e1) {
+ log.error("Error creating zoo store", e1);
+ return false;
+ }
+ // Loop through all of the prior transactions that are waiting to
+ // acquire this write lock. If the transaction has not succeeded or failed,
+ // then fail it and return false from this method so that Utils.reserveX()
+ // will call this method again and will re-check the prior transactions.
+ boolean result = true;
+ while (iterator.hasNext()) {
+ Entry<Long,byte[]> e = iterator.next();
+ Long txid = e.getKey();
+ if (!txid.equals(entry)) {
+ if (zs.isReserved(txid)) {
+ result &= false;
+ } else {
+ zs.reserve(txid);
+ TStatus status = zs.getStatus(txid);
+ if (status.equals(TStatus.FAILED)) {
+ result = false;
+ continue;
+ } else {
+ switch (status) {
+ case UNKNOWN:
+ // not sure what to do here, we have an invalid txid
+ break;
+ case SUCCESSFUL:
+ // already succeeded, lock should be removed
+ break;
+ case FAILED:
+ case FAILED_IN_PROGRESS:
Review comment:
Addressed in 2c3ea13
##########
File path: core/src/main/java/org/apache/accumulo/fate/zookeeper/DistributedReadWriteLock.java
##########
@@ -218,22 +233,82 @@ public boolean tryLock() {
}
SortedMap<Long,byte[]> entries = qlock.getEarlierEntries(entry);
Iterator<Entry<Long,byte[]>> iterator = entries.entrySet().iterator();
- if (!iterator.hasNext())
+ if (!iterator.hasNext()) {
throw new IllegalStateException("Did not find our own lock in the queue: " + this.entry
+ " userData " + new String(this.userData, UTF_8) + " lockType " + lockType());
- return iterator.next().getKey().equals(entry);
+ }
+ if (!failBlockers) {
+ return iterator.next().getKey().equals(entry);
+ } else {
+ ZooStore<DistributedReadWriteLock> zs;
+ try {
+ zs = new ZooStore<>(zooPath, zrw);
+ } catch (KeeperException | InterruptedException e1) {
+ log.error("Error creating zoo store", e1);
+ return false;
+ }
+ // Loop through all of the prior transactions that are waiting to
+ // acquire this write lock. If the transaction has not succeeded or failed,
+ // then fail it and return false from this method so that Utils.reserveX()
+ // will call this method again and will re-check the prior transactions.
+ boolean result = true;
+ while (iterator.hasNext()) {
+ Entry<Long,byte[]> e = iterator.next();
+ Long txid = e.getKey();
+ if (!txid.equals(entry)) {
+ if (zs.isReserved(txid)) {
+ result &= false;
+ } else {
+ zs.reserve(txid);
+ TStatus status = zs.getStatus(txid);
+ if (status.equals(TStatus.FAILED)) {
+ result = false;
+ continue;
+ } else {
+ switch (status) {
+ case UNKNOWN:
+ // not sure what to do here, we have an invalid txid
+ break;
+ case SUCCESSFUL:
+ // already succeeded, lock should be removed
+ break;
+ case FAILED:
+ case FAILED_IN_PROGRESS:
+ // transaction failed or in process of failing
+ break;
+ default:
+ zs.setStatus(txid, TStatus.FAILED_IN_PROGRESS);
+ result &= false;
+ }
+ }
+ zs.unreserve(txid, 0);
+ }
+ } else {
+ return result && txid.equals(entry);
+ }
+ }
+ return result;
Review comment:
Addressed in 2c3ea13
##########
File path: core/src/main/java/org/apache/accumulo/fate/zookeeper/DistributedReadWriteLock.java
##########
@@ -218,22 +233,82 @@ public boolean tryLock() {
}
SortedMap<Long,byte[]> entries = qlock.getEarlierEntries(entry);
Iterator<Entry<Long,byte[]>> iterator = entries.entrySet().iterator();
- if (!iterator.hasNext())
+ if (!iterator.hasNext()) {
throw new IllegalStateException("Did not find our own lock in the queue: " + this.entry
+ " userData " + new String(this.userData, UTF_8) + " lockType " + lockType());
- return iterator.next().getKey().equals(entry);
+ }
+ if (!failBlockers) {
+ return iterator.next().getKey().equals(entry);
+ } else {
+ ZooStore<DistributedReadWriteLock> zs;
+ try {
+ zs = new ZooStore<>(zooPath, zrw);
Review comment:
Addressed in 2c3ea13
--
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@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [accumulo] keith-turner commented on a change in pull request #2329: Attempt at failing FaTE transactions that are blocking this FaTE transaction from the write lock
Posted by GitBox <gi...@apache.org>.
keith-turner commented on a change in pull request #2329:
URL: https://github.com/apache/accumulo/pull/2329#discussion_r745060098
##########
File path: core/src/main/java/org/apache/accumulo/fate/zookeeper/DistributedReadWriteLock.java
##########
@@ -218,22 +228,69 @@ public boolean tryLock() {
}
SortedMap<Long,byte[]> entries = qlock.getEarlierEntries(entry);
Iterator<Entry<Long,byte[]>> iterator = entries.entrySet().iterator();
- if (!iterator.hasNext())
+ if (!iterator.hasNext()) {
throw new IllegalStateException("Did not find our own lock in the queue: " + this.entry
+ " userData " + new String(this.userData, UTF_8) + " lockType " + lockType());
- return iterator.next().getKey().equals(entry);
+ }
+ if (iterator.next().getKey().equals(entry)) {
+ return true;
+ }
+ if (failBlockers) {
+ // Loop through all of the prior transactions that are waiting to
+ // acquire this write lock. If the transaction has not succeeded or failed,
+ // then fail it and return false from this method so that Utils.reserveX()
+ // will call this method again and will re-check the prior transactions.
+ boolean result = true;
+ while (iterator.hasNext()) {
+ Entry<Long,byte[]> e = iterator.next();
+ Long txid = e.getKey();
+ if (!txid.equals(entry)) {
+ if (store.isReserved(txid)) {
+ result &= false;
+ } else {
+ store.reserve(txid);
+ TStatus status = store.getStatus(txid);
+ if (status.equals(TStatus.FAILED)) {
+ result = false;
+ continue;
+ } else {
+ switch (status) {
+ case FAILED_IN_PROGRESS:
+ case IN_PROGRESS:
+ store.setStatus(txid, TStatus.FAILED_IN_PROGRESS);
+ result &= false;
+ break;
+ default:
+ log.error("Attempting to fail a transaction in an unhandled state: {}", status);
+ }
+ }
+ store.unreserve(txid, 0);
+ }
+ } else {
+ return result && txid.equals(entry);
Review comment:
```suggestion
break;
```
##########
File path: core/src/main/java/org/apache/accumulo/fate/zookeeper/DistributedReadWriteLock.java
##########
@@ -218,22 +228,69 @@ public boolean tryLock() {
}
SortedMap<Long,byte[]> entries = qlock.getEarlierEntries(entry);
Iterator<Entry<Long,byte[]>> iterator = entries.entrySet().iterator();
- if (!iterator.hasNext())
+ if (!iterator.hasNext()) {
throw new IllegalStateException("Did not find our own lock in the queue: " + this.entry
+ " userData " + new String(this.userData, UTF_8) + " lockType " + lockType());
- return iterator.next().getKey().equals(entry);
+ }
+ if (iterator.next().getKey().equals(entry)) {
+ return true;
+ }
+ if (failBlockers) {
+ // Loop through all of the prior transactions that are waiting to
+ // acquire this write lock. If the transaction has not succeeded or failed,
+ // then fail it and return false from this method so that Utils.reserveX()
+ // will call this method again and will re-check the prior transactions.
+ boolean result = true;
+ while (iterator.hasNext()) {
+ Entry<Long,byte[]> e = iterator.next();
+ Long txid = e.getKey();
+ if (!txid.equals(entry)) {
+ if (store.isReserved(txid)) {
+ result &= false;
+ } else {
+ store.reserve(txid);
+ TStatus status = store.getStatus(txid);
+ if (status.equals(TStatus.FAILED)) {
+ result = false;
+ continue;
+ } else {
+ switch (status) {
+ case FAILED_IN_PROGRESS:
+ case IN_PROGRESS:
+ store.setStatus(txid, TStatus.FAILED_IN_PROGRESS);
+ result &= false;
+ break;
+ default:
+ log.error("Attempting to fail a transaction in an unhandled state: {}", status);
+ }
+ }
+ store.unreserve(txid, 0);
+ }
+ } else {
+ return result && txid.equals(entry);
+ }
+ }
+ return result;
Review comment:
```suggestion
```
##########
File path: core/src/main/java/org/apache/accumulo/fate/zookeeper/DistributedReadWriteLock.java
##########
@@ -218,22 +228,69 @@ public boolean tryLock() {
}
SortedMap<Long,byte[]> entries = qlock.getEarlierEntries(entry);
Iterator<Entry<Long,byte[]>> iterator = entries.entrySet().iterator();
- if (!iterator.hasNext())
+ if (!iterator.hasNext()) {
throw new IllegalStateException("Did not find our own lock in the queue: " + this.entry
+ " userData " + new String(this.userData, UTF_8) + " lockType " + lockType());
- return iterator.next().getKey().equals(entry);
+ }
+ if (iterator.next().getKey().equals(entry)) {
+ return true;
+ }
+ if (failBlockers) {
+ // Loop through all of the prior transactions that are waiting to
+ // acquire this write lock. If the transaction has not succeeded or failed,
+ // then fail it and return false from this method so that Utils.reserveX()
+ // will call this method again and will re-check the prior transactions.
+ boolean result = true;
Review comment:
I was thinking we should drop this and always return false if we are not the first, this seems the safest to avoid concurrency issues.
```suggestion
```
##########
File path: core/src/main/java/org/apache/accumulo/fate/zookeeper/DistributedReadWriteLock.java
##########
@@ -218,22 +228,69 @@ public boolean tryLock() {
}
SortedMap<Long,byte[]> entries = qlock.getEarlierEntries(entry);
Iterator<Entry<Long,byte[]>> iterator = entries.entrySet().iterator();
- if (!iterator.hasNext())
+ if (!iterator.hasNext()) {
throw new IllegalStateException("Did not find our own lock in the queue: " + this.entry
+ " userData " + new String(this.userData, UTF_8) + " lockType " + lockType());
- return iterator.next().getKey().equals(entry);
+ }
+ if (iterator.next().getKey().equals(entry)) {
+ return true;
+ }
+ if (failBlockers) {
+ // Loop through all of the prior transactions that are waiting to
+ // acquire this write lock. If the transaction has not succeeded or failed,
+ // then fail it and return false from this method so that Utils.reserveX()
+ // will call this method again and will re-check the prior transactions.
+ boolean result = true;
+ while (iterator.hasNext()) {
+ Entry<Long,byte[]> e = iterator.next();
+ Long txid = e.getKey();
+ if (!txid.equals(entry)) {
+ if (store.isReserved(txid)) {
+ result &= false;
+ } else {
+ store.reserve(txid);
+ TStatus status = store.getStatus(txid);
+ if (status.equals(TStatus.FAILED)) {
+ result = false;
+ continue;
+ } else {
+ switch (status) {
+ case FAILED_IN_PROGRESS:
+ case IN_PROGRESS:
+ store.setStatus(txid, TStatus.FAILED_IN_PROGRESS);
+ result &= false;
+ break;
+ default:
+ log.error("Attempting to fail a transaction in an unhandled state: {}", status);
+ }
+ }
+ store.unreserve(txid, 0);
Review comment:
should probably do something like
```java
if(store.tryReserve(txid)){
try{
// attempt to cancel
} finally {
// always unresrve even if there is an exception
store.unreserve(txid, 0);
}
}
```
##########
File path: core/src/main/java/org/apache/accumulo/fate/zookeeper/DistributedReadWriteLock.java
##########
@@ -218,22 +228,69 @@ public boolean tryLock() {
}
SortedMap<Long,byte[]> entries = qlock.getEarlierEntries(entry);
Iterator<Entry<Long,byte[]>> iterator = entries.entrySet().iterator();
- if (!iterator.hasNext())
+ if (!iterator.hasNext()) {
throw new IllegalStateException("Did not find our own lock in the queue: " + this.entry
+ " userData " + new String(this.userData, UTF_8) + " lockType " + lockType());
- return iterator.next().getKey().equals(entry);
+ }
+ if (iterator.next().getKey().equals(entry)) {
+ return true;
+ }
+ if (failBlockers) {
+ // Loop through all of the prior transactions that are waiting to
+ // acquire this write lock. If the transaction has not succeeded or failed,
+ // then fail it and return false from this method so that Utils.reserveX()
+ // will call this method again and will re-check the prior transactions.
+ boolean result = true;
+ while (iterator.hasNext()) {
+ Entry<Long,byte[]> e = iterator.next();
+ Long txid = e.getKey();
+ if (!txid.equals(entry)) {
+ if (store.isReserved(txid)) {
+ result &= false;
+ } else {
+ store.reserve(txid);
Review comment:
The check for reserved and then reserving is a racy. Could add a new method that tries to reserve atomically.
```suggestion
if (store.tryReserve(txid)) {
```
--
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@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org