+ public boolean checkForDeadlock(String keyspace, String table, String lockName, LockType locktype, String owner, boolean forAcquire) throws MusicServiceException, MusicQueryException {
+ if (locktype.equals(LockType.READ)) return false;
+ if (owner==null || owner.length()==0) return false;
+
+ String lockTable = table_prepend_name + table;
+ PreparedQueryObject queryObject = new PreparedQueryObject();
+ queryObject.appendQueryString("SELECT key, acquiretime, owner FROM " + keyspace + "." + lockTable);
+ queryObject.appendQueryString(" WHERE lockType = ? ALLOW FILTERING");
+ queryObject.addValue(LockType.WRITE);
+
+ DeadlockDetectionUtil ddu = new DeadlockDetectionUtil();
+
+ ResultSet rs = dsHandle.executeQuorumConsistencyGet(queryObject);
+ logger.debug("rs has " + rs.getAvailableWithoutFetching() + (rs.isFullyFetched()?"":" (or more!)") );
+ Iterator<Row> it = rs.iterator();
+ while (it.hasNext()) {
+ Row row = it.next();
+ logger.debug("key = " + row.getString("key") + ", time = " + row.getString("acquiretime") + ", owner = " + row.getString("owner") );
+ ddu.setExisting(row.getString("key"), row.getString("owner"), ("0".equals(row.getString("acquiretime")))?OwnershipType.CREATED:OwnershipType.ACQUIRED);
+ }
+ boolean deadlock = ddu.checkForDeadlock(lockName, owner, forAcquire?OwnershipType.ACQUIRED:OwnershipType.CREATED);
+ if (deadlock) logger.warn("Deadlock detected when " + owner + " tried to create lock on " + keyspace + "." + lockTable + "." + lockName);
+ return deadlock;
+ }
+
+ public List<String> getAllLocksForOwner(String ownerId, String keyspace, String table) throws MusicServiceException, MusicQueryException {
+ List<String> toRet = new ArrayList<String>();
+ String lockTable = table_prepend_name + table;
+ PreparedQueryObject queryObject = new PreparedQueryObject();
+ queryObject.appendQueryString("SELECT key, lockreference FROM " + keyspace + "." + lockTable);
+ queryObject.appendQueryString(" WHERE owner = '" + ownerId + "' ALLOW FILTERING");
+
+ ResultSet rs = dsHandle.executeQuorumConsistencyGet(queryObject);
+ Iterator<Row> it = rs.iterator();
+ while (it.hasNext()) {
+ Row row = it.next();
+ toRet.add(row.getString("key") + "$" + row.getLong("lockreference"));
+ }
+ return toRet;
+ }
+
+ public ReturnType promoteLock(String keyspace, String table, String key, String lockRef)
+ throws MusicLockingException, MusicServiceException, MusicQueryException {
+ String lockqtable = table_prepend_name + table;
+ String selectQuery = "select * from " + keyspace + "." + lockqtable + " where key=?;";
+
+ PreparedQueryObject queryObject = new PreparedQueryObject();
+ queryObject.appendQueryString(selectQuery);
+ queryObject.addValue(key);
+ ResultSet rs = dsHandle.executeOneConsistencyGet(queryObject);
+
+ long refToPromote = Long.parseLong(lockRef);
+
+ boolean topOfQueue = true;
+ boolean readBlock = false;
+ boolean seenLockToPromote = false;
+ boolean promotionOngoing = false;
+ long readBlockStart = 0;
+ long readBlockEnd = 0;
+
+
+ for (Row row : rs) {
+ long ref = row.getLong("lockreference");
+ LockType lockType = row.get("lockType", LockType.class);
+
+ if (refToPromote==ref) {
+ if (promotionOngoing) {
+ return new ReturnType(ResultType.FAILURE, "Can't promote, already promoting another lockref.");
+ }
+ seenLockToPromote = true;
+ if (!topOfQueue) {
+ readBlockStart = ref;
+ readBlockEnd = ref;
+ break;
+ }
+ } else if (!seenLockToPromote && refToPromote<ref) {
+ return new ReturnType(ResultType.FAILURE, "Lockref does not exist.");
+ }
+
+ if (lockType==LockType.READ || lockType==LockType.PROMOTING) {
+ if (!readBlock) {
+ readBlockStart = ref;
+ readBlock = true;
+ }
+ if (readBlock) {
+ readBlockEnd = ref;
+ }
+ if (lockType==LockType.PROMOTING) {
+ promotionOngoing = true;
+ }
+ }
+
+ if (lockType==LockType.WRITE) {
+ if (refToPromote==ref) {
+ return new ReturnType(ResultType.FAILURE, "Lockref is already write.");
+ }
+ if (readBlock) {
+ readBlock = false;
+ promotionOngoing = false;
+ if (seenLockToPromote) {
+ break;
+ }
+ //can no longer be lock holder after this
+ topOfQueue = false;
+ }
+ }
+ }
+
+ if (readBlockStart<=refToPromote && refToPromote<=readBlockEnd) {
+ if (readBlockStart==refToPromote && refToPromote==readBlockEnd) {
+ promoteLockTo(keyspace, lockqtable, key, lockRef, LockType.WRITE);
+ return new ReturnType(ResultType.SUCCESS, "Lock has successfully been upgraded.");
+ }
+ promoteLockTo(keyspace, lockqtable, key, lockRef, LockType.PROMOTING);
+ return new ReturnType(ResultType.FAILURE, "Your lock upgrade is in progress. Check again to see if successful.");
+ }
+
+ //shouldn't reach here?
+ return new ReturnType(ResultType.FAILURE,"Promotion failed.");
+ }
+
+ private void promoteLockTo(String keyspace, String table, String key, String lockRef, LockType newLockType)
+ throws MusicServiceException, MusicQueryException {
+ PreparedQueryObject queryObject =
+ new PreparedQueryObject("UPDATE " + keyspace + "." + table + " SET lockType=? WHERE key='" + key
+ + "' AND lockReference = " + lockRef + " IF EXISTS;", newLockType);
+
+ //cannot use executePut because we need to ignore music timestamp adjustments for lock store
+ dsHandle.executePut(queryObject, MusicUtil.QUORUM);