You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cloudstack.apache.org by ah...@apache.org on 2013/06/27 23:37:07 UTC
[1/2] git commit: updated refs/heads/vmsync to 0233044
Updated Branches:
refs/heads/vmsync 3c2aa1850 -> 0233044b2
Deadlock resolved
Project: http://git-wip-us.apache.org/repos/asf/cloudstack/repo
Commit: http://git-wip-us.apache.org/repos/asf/cloudstack/commit/c6ba4a23
Tree: http://git-wip-us.apache.org/repos/asf/cloudstack/tree/c6ba4a23
Diff: http://git-wip-us.apache.org/repos/asf/cloudstack/diff/c6ba4a23
Branch: refs/heads/vmsync
Commit: c6ba4a23e88cd18d756cde9648a96b57c2606ae5
Parents: 3c2aa18
Author: Alex Huang <al...@gmail.com>
Authored: Wed Jun 26 16:53:54 2013 -0700
Committer: Alex Huang <al...@gmail.com>
Committed: Thu Jun 27 14:00:31 2013 -0700
----------------------------------------------------------------------
.../framework/jobs/AsyncJobManager.java | 2 +
.../framework/jobs/dao/AsyncJobJoinMapDao.java | 4 +-
.../jobs/dao/AsyncJobJoinMapDaoImpl.java | 113 +++++++++++--------
.../jobs/impl/AsyncJobManagerImpl.java | 72 +++++++++++-
tools/eclipse/remove-eclipse-profile.sh | 26 +++++
utils/src/com/cloud/utils/db/GenericDao.java | 4 +-
.../src/com/cloud/utils/db/GenericDaoBase.java | 1 +
.../cloud/utils/db/GenericSearchBuilder.java | 46 +++++---
8 files changed, 203 insertions(+), 65 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cloudstack/blob/c6ba4a23/framework/jobs/src/org/apache/cloudstack/framework/jobs/AsyncJobManager.java
----------------------------------------------------------------------
diff --git a/framework/jobs/src/org/apache/cloudstack/framework/jobs/AsyncJobManager.java b/framework/jobs/src/org/apache/cloudstack/framework/jobs/AsyncJobManager.java
index bc06101..440188a 100644
--- a/framework/jobs/src/org/apache/cloudstack/framework/jobs/AsyncJobManager.java
+++ b/framework/jobs/src/org/apache/cloudstack/framework/jobs/AsyncJobManager.java
@@ -37,6 +37,8 @@ public interface AsyncJobManager extends Manager {
void completeAsyncJob(long jobId, JobInfo.Status jobStatus, int resultCode, String result);
+ List<Long> wakeupByJoinedJobCompletion(long joinedJobId);
+
void updateAsyncJobStatus(long jobId, int processStatus, String resultObject);
void updateAsyncJobAttachment(long jobId, String instanceType, Long instanceId);
void logJobJournal(long jobId, AsyncJob.JournalType journalType, String
http://git-wip-us.apache.org/repos/asf/cloudstack/blob/c6ba4a23/framework/jobs/src/org/apache/cloudstack/framework/jobs/dao/AsyncJobJoinMapDao.java
----------------------------------------------------------------------
diff --git a/framework/jobs/src/org/apache/cloudstack/framework/jobs/dao/AsyncJobJoinMapDao.java b/framework/jobs/src/org/apache/cloudstack/framework/jobs/dao/AsyncJobJoinMapDao.java
index 9c993f1..4458fa2 100644
--- a/framework/jobs/src/org/apache/cloudstack/framework/jobs/dao/AsyncJobJoinMapDao.java
+++ b/framework/jobs/src/org/apache/cloudstack/framework/jobs/dao/AsyncJobJoinMapDao.java
@@ -37,5 +37,7 @@ public interface AsyncJobJoinMapDao extends GenericDao<AsyncJobJoinMapVO, Long>
void completeJoin(long joinJobId, JobInfo.Status joinStatus, String joinResult, long completeMsid);
List<Long> wakeupScan();
- List<Long> wakeupByJoinedJobCompletion(long joinedJobId);
+
+ List<Long> findJobsToWake(long joinedJobId);
+// List<Long> wakeupByJoinedJobCompletion(long joinedJobId);
}
http://git-wip-us.apache.org/repos/asf/cloudstack/blob/c6ba4a23/framework/jobs/src/org/apache/cloudstack/framework/jobs/dao/AsyncJobJoinMapDaoImpl.java
----------------------------------------------------------------------
diff --git a/framework/jobs/src/org/apache/cloudstack/framework/jobs/dao/AsyncJobJoinMapDaoImpl.java b/framework/jobs/src/org/apache/cloudstack/framework/jobs/dao/AsyncJobJoinMapDaoImpl.java
index 8ea5073..ba6cfbc 100644
--- a/framework/jobs/src/org/apache/cloudstack/framework/jobs/dao/AsyncJobJoinMapDaoImpl.java
+++ b/framework/jobs/src/org/apache/cloudstack/framework/jobs/dao/AsyncJobJoinMapDaoImpl.java
@@ -37,6 +37,7 @@ import com.cloud.utils.db.SearchCriteria;
import com.cloud.utils.db.SearchCriteria.Op;
import com.cloud.utils.db.Transaction;
import com.cloud.utils.db.UpdateBuilder;
+import com.cloud.utils.exception.CloudRuntimeException;
public class AsyncJobJoinMapDaoImpl extends GenericDaoBase<AsyncJobJoinMapVO, Long> implements AsyncJobJoinMapDao {
public static final Logger s_logger = Logger.getLogger(AsyncJobJoinMapDaoImpl.class);
@@ -45,8 +46,10 @@ public class AsyncJobJoinMapDaoImpl extends GenericDaoBase<AsyncJobJoinMapVO, Lo
private final SearchBuilder<AsyncJobJoinMapVO> RecordSearchByOwner;
private final SearchBuilder<AsyncJobJoinMapVO> CompleteJoinSearch;
private final SearchBuilder<AsyncJobJoinMapVO> WakeupSearch;
+
+// private final GenericSearchBuilder<AsyncJobJoinMapVO, Long> JoinJobSearch;
- public AsyncJobJoinMapDaoImpl() {
+ protected AsyncJobJoinMapDaoImpl() {
RecordSearch = createSearchBuilder();
RecordSearch.and("jobId", RecordSearch.entity().getJobId(), Op.EQ);
RecordSearch.and("joinJobId", RecordSearch.entity().getJoinJobId(), Op.EQ);
@@ -65,6 +68,10 @@ public class AsyncJobJoinMapDaoImpl extends GenericDaoBase<AsyncJobJoinMapVO, Lo
WakeupSearch.and("expiration", WakeupSearch.entity().getExpiration(), Op.GT);
WakeupSearch.and("joinStatus", WakeupSearch.entity().getJoinStatus(), Op.EQ);
WakeupSearch.done();
+
+// JoinJobSearch = createSearchBuilder(Long.class);
+// JoinJobSearch.and(JoinJobSearch.entity().getJoinJobId(), Op.SC, "joinJobId");
+// JoinJobSearch.done();
}
@Override
@@ -202,49 +209,67 @@ public class AsyncJobJoinMapDaoImpl extends GenericDaoBase<AsyncJobJoinMapVO, Lo
return standaloneList;
}
-
- @Override
- public List<Long> wakeupByJoinedJobCompletion(long joinedJobId) {
- List<Long> standaloneList = new ArrayList<Long>();
-
- Transaction txn = Transaction.currentTxn();
- PreparedStatement pstmt = null;
+
+ @Override
+ public List<Long> findJobsToWake(long joinedJobId) {
+ List<Long> standaloneList = new ArrayList<Long>();
+ Transaction txn = Transaction.currentTxn();
+ String sql = "SELECT job_id FROM async_job_join_map WHERE join_job_id = ? AND job_id NOT IN (SELECT content_id FROM sync_queue_item)";
try {
- txn.start();
-
- //
- // performance sensitive processing, do it in plain SQL
- //
- String sql = "UPDATE async_job SET job_pending_signals=? WHERE id IN " +
- "(SELECT job_id FROM async_job_join_map WHERE join_job_id = ?)";
- pstmt = txn.prepareStatement(sql);
- pstmt.setInt(1, AsyncJob.Contants.SIGNAL_MASK_WAKEUP);
- pstmt.setLong(2, joinedJobId);
- pstmt.executeUpdate();
- pstmt.close();
-
- sql = "UPDATE sync_queue_item SET queue_proc_msid=NULL, queue_proc_number=NULL WHERE content_id IN " +
- "(SELECT job_id FROM async_job_join_map WHERE join_job_id = ?)";
- pstmt = txn.prepareStatement(sql);
- pstmt.setLong(1, joinedJobId);
- pstmt.executeUpdate();
- pstmt.close();
-
- sql = "SELECT job_id FROM async_job_join_map WHERE join_job_id = ? AND job_id NOT IN (SELECT content_id FROM sync_queue_item)";
- pstmt = txn.prepareStatement(sql);
- pstmt.setLong(1, joinedJobId);
- ResultSet rs = pstmt.executeQuery();
- while(rs.next()) {
- standaloneList.add(rs.getLong(1));
- }
- rs.close();
- pstmt.close();
-
- txn.commit();
- } catch (SQLException e) {
- s_logger.error("Unexpected exception", e);
- }
-
+ PreparedStatement pstmt = txn.prepareStatement(sql);
+ pstmt.setLong(1, joinedJobId);
+ ResultSet rs = pstmt.executeQuery();
+ while (rs.next()) {
+ standaloneList.add(rs.getLong(1));
+ }
+ } catch (SQLException e) {
+ throw new CloudRuntimeException("Unable to execute " + sql, e);
+ }
return standaloneList;
- }
+ }
+
+// @Override
+// public List<Long> wakeupByJoinedJobCompletion(long joinedJobId) {
+// List<Long> standaloneList = new ArrayList<Long>();
+//
+// Transaction txn = Transaction.currentTxn();
+// PreparedStatement pstmt = null;
+// try {
+// txn.start();
+//
+// //
+// // performance sensitive processing, do it in plain SQL
+// //
+// String sql = "UPDATE async_job SET job_pending_signals=? WHERE id IN " +
+// "(SELECT job_id FROM async_job_join_map WHERE join_job_id = ?)";
+// pstmt = txn.prepareStatement(sql);
+// pstmt.setInt(1, AsyncJob.Contants.SIGNAL_MASK_WAKEUP);
+// pstmt.setLong(2, joinedJobId);
+// pstmt.executeUpdate();
+// pstmt.close();
+//
+// sql = "UPDATE sync_queue_item SET queue_proc_msid=NULL, queue_proc_number=NULL WHERE content_id IN " +
+// "(SELECT job_id FROM async_job_join_map WHERE join_job_id = ?)";
+// pstmt = txn.prepareStatement(sql);
+// pstmt.setLong(1, joinedJobId);
+// pstmt.executeUpdate();
+// pstmt.close();
+//
+// sql = "SELECT job_id FROM async_job_join_map WHERE join_job_id = ? AND job_id NOT IN (SELECT content_id FROM sync_queue_item)";
+// pstmt = txn.prepareStatement(sql);
+// pstmt.setLong(1, joinedJobId);
+// ResultSet rs = pstmt.executeQuery();
+// while(rs.next()) {
+// standaloneList.add(rs.getLong(1));
+// }
+// rs.close();
+// pstmt.close();
+//
+// txn.commit();
+// } catch (SQLException e) {
+// s_logger.error("Unexpected exception", e);
+// }
+//
+// return standaloneList;
+// }
}
http://git-wip-us.apache.org/repos/asf/cloudstack/blob/c6ba4a23/framework/jobs/src/org/apache/cloudstack/framework/jobs/impl/AsyncJobManagerImpl.java
----------------------------------------------------------------------
diff --git a/framework/jobs/src/org/apache/cloudstack/framework/jobs/impl/AsyncJobManagerImpl.java b/framework/jobs/src/org/apache/cloudstack/framework/jobs/impl/AsyncJobManagerImpl.java
index a59aea3..2351a10 100644
--- a/framework/jobs/src/org/apache/cloudstack/framework/jobs/impl/AsyncJobManagerImpl.java
+++ b/framework/jobs/src/org/apache/cloudstack/framework/jobs/impl/AsyncJobManagerImpl.java
@@ -48,6 +48,7 @@ import org.apache.cloudstack.framework.jobs.AsyncJobManager;
import org.apache.cloudstack.framework.jobs.dao.AsyncJobDao;
import org.apache.cloudstack.framework.jobs.dao.AsyncJobJoinMapDao;
import org.apache.cloudstack.framework.jobs.dao.AsyncJobJournalDao;
+import org.apache.cloudstack.framework.jobs.dao.SyncQueueItemDao;
import org.apache.cloudstack.framework.messagebus.MessageBus;
import org.apache.cloudstack.framework.messagebus.MessageDetector;
import org.apache.cloudstack.framework.messagebus.PublishScope;
@@ -65,12 +66,18 @@ import com.cloud.utils.concurrency.NamedThreadFactory;
import com.cloud.utils.db.DB;
import com.cloud.utils.db.GenericDao;
import com.cloud.utils.db.GenericDaoBase;
+import com.cloud.utils.db.GenericSearchBuilder;
import com.cloud.utils.db.GlobalLock;
+import com.cloud.utils.db.SearchBuilder;
+import com.cloud.utils.db.SearchCriteria;
+import com.cloud.utils.db.SearchCriteria.Op;
import com.cloud.utils.db.Transaction;
import com.cloud.utils.exception.CloudRuntimeException;
import com.cloud.utils.exception.ExceptionUtil;
import com.cloud.utils.mgmt.JmxUtil;
+import edu.emory.mathcs.backport.java.util.Collections;
+
public class AsyncJobManagerImpl extends ManagerBase implements AsyncJobManager, ClusterManagerListener, Configurable {
// Advanced
private static final ConfigKey<Long> JobExpireMinutes = new ConfigKey<Long>(Long.class, "job.expire.minutes", "Advanced", AsyncJobManager.class, "1440",
@@ -86,7 +93,9 @@ public class AsyncJobManagerImpl extends ManagerBase implements AsyncJobManager,
private static final int MAX_ONETIME_SCHEDULE_SIZE = 50;
private static final int HEARTBEAT_INTERVAL = 2000;
private static final int GC_INTERVAL = 10000; // 10 seconds
-
+
+ @Inject
+ private SyncQueueItemDao _queueItemDao;
@Inject private SyncQueueManager _queueMgr;
@Inject private AsyncJobDao _jobDao;
@Inject private AsyncJobJournalDao _journalDao;
@@ -224,7 +233,7 @@ public class AsyncJobManagerImpl extends ManagerBase implements AsyncJobManager,
job.setLastUpdated(DateUtil.currentGMTTime());
_jobDao.update(jobId, job);
- List<Long> wakeupList = _joinMapDao.wakeupByJoinedJobCompletion(jobId);
+ List<Long> wakeupList = wakeupByJoinedJobCompletion(jobId);
_joinMapDao.disjoinAllJobs(jobId);
txn.commit();
@@ -841,6 +850,26 @@ public class AsyncJobManagerImpl extends ManagerBase implements AsyncJobManager,
throw new ConfigurationException("Unable to load db.properties to configure AsyncJobManagerImpl");
}
+ JoinJobSearch = _joinMapDao.createSearchBuilder(Long.class);
+ JoinJobSearch.and(JoinJobSearch.entity().getJoinJobId(), Op.EQ, "joinJobId");
+ JoinJobSearch.selectField(JoinJobSearch.entity().getJobId());
+ JoinJobSearch.done();
+
+ JobIdsSearch = _jobDao.createSearchBuilder();
+ JobIdsSearch.and(JobIdsSearch.entity().getId(), Op.IN, "ids").done();
+
+ QueueJobIdsSearch = _queueItemDao.createSearchBuilder();
+ QueueJobIdsSearch.and(QueueJobIdsSearch.entity().getContentId(), Op.IN, "contentIds").done();
+
+ JoinJobIdsSearch = _joinMapDao.createSearchBuilder(Long.class);
+ JoinJobIdsSearch.selectField(JoinJobIdsSearch.entity().getJobId());
+ JoinJobIdsSearch.and(JoinJobIdsSearch.entity().getJoinJobId(), Op.EQ, "joinJobId");
+ JoinJobIdsSearch.and(JoinJobIdsSearch.entity().getJobId(), Op.NIN, "jobIds");
+ JoinJobIdsSearch.done();
+
+ ContentIdsSearch = _queueItemDao.createSearchBuilder(Long.class);
+ ContentIdsSearch.selectField(ContentIdsSearch.entity().getContentId()).done();
+
AsyncJobExecutionContext.init(this, _joinMapDao);
OutcomeImpl.init(this);
@@ -848,6 +877,34 @@ public class AsyncJobManagerImpl extends ManagerBase implements AsyncJobManager,
}
@Override
+ @DB
+ public List<Long> wakeupByJoinedJobCompletion(long joinedJobId) {
+ SearchCriteria<Long> joinJobSC = JoinJobSearch.create("joinJobId", joinedJobId);
+
+ List<Long> result = _joinMapDao.customSearch(joinJobSC, null);
+ if (result.size() != 0) {
+ Collections.sort(result);
+ Long[] ids = result.toArray(new Long[result.size()]);
+
+ SearchCriteria<AsyncJobVO> jobsSC = JobIdsSearch.create("ids", ids);
+ SearchCriteria<SyncQueueItemVO> queueItemsSC = QueueJobIdsSearch.create("contentIds", ids);
+
+ Transaction txn = Transaction.currentTxn();
+ txn.start();
+ AsyncJobVO job = _jobDao.createForUpdate();
+ job.setPendingSignals(1);
+ _jobDao.update(job, jobsSC);
+
+ SyncQueueItemVO item = _queueItemDao.createForUpdate();
+ item.setLastProcessNumber(null);
+ item.setLastProcessMsid(null);
+ _queueItemDao.update(item, queueItemsSC);
+ txn.commit();
+ }
+ return _joinMapDao.findJobsToWake(joinedJobId);
+ }
+
+ @Override
public void onManagementNodeJoined(List<? extends ManagementServerHost> nodeList, long selfNodeId) {
}
@@ -897,4 +954,15 @@ public class AsyncJobManagerImpl extends ManagerBase implements AsyncJobManager,
_executor.shutdown();
return true;
}
+
+ private GenericSearchBuilder<SyncQueueItemVO, Long> ContentIdsSearch;
+ private GenericSearchBuilder<AsyncJobJoinMapVO, Long> JoinJobSearch;
+ private SearchBuilder<AsyncJobVO> JobIdsSearch;
+ private SearchBuilder<SyncQueueItemVO> QueueJobIdsSearch;
+ private GenericSearchBuilder<AsyncJobJoinMapVO, Long> JoinJobIdsSearch;
+
+ protected AsyncJobManagerImpl() {
+
+ }
+
}
http://git-wip-us.apache.org/repos/asf/cloudstack/blob/c6ba4a23/tools/eclipse/remove-eclipse-profile.sh
----------------------------------------------------------------------
diff --git a/tools/eclipse/remove-eclipse-profile.sh b/tools/eclipse/remove-eclipse-profile.sh
new file mode 100644
index 0000000..6b31583
--- /dev/null
+++ b/tools/eclipse/remove-eclipse-profile.sh
@@ -0,0 +1,26 @@
+#!/bin/bash
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+for file in `find . -name org.eclipse.m2e.core.prefs | xargs`; do
+ if grep -q activeProfiles=eclipse $file; then
+ echo Replacing $file;
+ sed -i s/activeProfiles=eclipse/activeProfiles=/g $file;
+ else
+ echo Skipping $file;
+ fi;
+done
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/cloudstack/blob/c6ba4a23/utils/src/com/cloud/utils/db/GenericDao.java
----------------------------------------------------------------------
diff --git a/utils/src/com/cloud/utils/db/GenericDao.java b/utils/src/com/cloud/utils/db/GenericDao.java
index 3efc1c5..66d0f67 100755
--- a/utils/src/com/cloud/utils/db/GenericDao.java
+++ b/utils/src/com/cloud/utils/db/GenericDao.java
@@ -134,6 +134,8 @@ public interface GenericDao<T, ID extends Serializable> {
boolean update(ID id, T entity);
+ int update(T entity, SearchCriteria<T> sc);
+
/**
* Look for all active rows.
* @return list of entity beans.
@@ -213,7 +215,7 @@ public interface GenericDao<T, ID extends Serializable> {
/**
* Remove based on the search criteria. This will delete if the VO object
- * does not have a REMOVED column.
+ * does not have a REMOVED column.
* @param sc search criteria to match
* @return rows removed.
*/
http://git-wip-us.apache.org/repos/asf/cloudstack/blob/c6ba4a23/utils/src/com/cloud/utils/db/GenericDaoBase.java
----------------------------------------------------------------------
diff --git a/utils/src/com/cloud/utils/db/GenericDaoBase.java b/utils/src/com/cloud/utils/db/GenericDaoBase.java
index f593c38..3a7e584 100755
--- a/utils/src/com/cloud/utils/db/GenericDaoBase.java
+++ b/utils/src/com/cloud/utils/db/GenericDaoBase.java
@@ -1267,6 +1267,7 @@ public abstract class GenericDaoBase<T, ID extends Serializable> extends Compone
}
@DB(txn=false)
+ @Override
public int update(final T entity, final SearchCriteria<T> sc) {
final UpdateBuilder ub = getUpdateBuilder(entity);
return update(ub, sc, null);
http://git-wip-us.apache.org/repos/asf/cloudstack/blob/c6ba4a23/utils/src/com/cloud/utils/db/GenericSearchBuilder.java
----------------------------------------------------------------------
diff --git a/utils/src/com/cloud/utils/db/GenericSearchBuilder.java b/utils/src/com/cloud/utils/db/GenericSearchBuilder.java
index 4bae344..bf28144 100755
--- a/utils/src/com/cloud/utils/db/GenericSearchBuilder.java
+++ b/utils/src/com/cloud/utils/db/GenericSearchBuilder.java
@@ -34,7 +34,6 @@ import net.sf.cglib.proxy.MethodProxy;
import com.cloud.utils.db.SearchCriteria.Func;
import com.cloud.utils.db.SearchCriteria.Op;
import com.cloud.utils.db.SearchCriteria.SelectType;
-import com.cloud.utils.exception.CloudRuntimeException;
/**
* GenericSearchBuilder is used to build a search based on a VO object
@@ -75,13 +74,19 @@ public class GenericSearchBuilder<T, K> implements MethodInterceptor {
}
protected Attribute getSpecifiedAttribute() {
- assert(_entity != null && _specifiedAttrs != null && _specifiedAttrs.size() == 1) : "Now now, better specify an attribute or else we can't help you";
+ if (_entity == null || _specifiedAttrs == null || _specifiedAttrs.size() != 1) {
+ throw new RuntimeException("Now now, better specify an attribute or else we can't help you");
+ }
return _specifiedAttrs.get(0);
}
public GenericSearchBuilder<T, K> selectField(Object... useless) {
- assert _entity != null : "SearchBuilder cannot be modified once it has been setup";
- assert _specifiedAttrs.size() > 0 : "You didn't specify any attributes";
+ if (_entity == null) {
+ throw new RuntimeException("SearchBuilder cannot be modified once it has been setup");
+ }
+ if (_specifiedAttrs.size() <= 0) {
+ throw new RuntimeException("You didn't specify any attributes");
+ }
if (_selects == null) {
_selects = new ArrayList<Select>();
@@ -118,9 +123,15 @@ public class GenericSearchBuilder<T, K> implements MethodInterceptor {
* @return a SearchBuilder to build more search parts.
*/
public GenericSearchBuilder<T, K> select(String fieldName, Func func, Object useless, Object... params) {
- assert _entity != null : "SearchBuilder cannot be modified once it has been setup";
- assert _specifiedAttrs.size() <= 1 : "You can't specify more than one field to search on";
- assert func.getCount() == -1 || (func.getCount() == (params.length + 1)) : "The number of parameters does not match the function param count for " + func;
+ if (_entity == null) {
+ throw new RuntimeException("SearchBuilder cannot be modified once it has been setup");
+ }
+ if (_specifiedAttrs.size() > 1) {
+ throw new RuntimeException("You can't specify more than one field to search on");
+ }
+ if (func.getCount() != -1 && (func.getCount() != (params.length + 1))) {
+ throw new RuntimeException("The number of parameters does not match the function param count for " + func);
+ }
if (_selects == null) {
_selects = new ArrayList<Select>();
@@ -132,12 +143,15 @@ public class GenericSearchBuilder<T, K> implements MethodInterceptor {
field = _resultType.getDeclaredField(fieldName);
field.setAccessible(true);
} catch (SecurityException e) {
- throw new CloudRuntimeException("Unable to find " + fieldName, e);
+ throw new RuntimeException("Unable to find " + fieldName, e);
} catch (NoSuchFieldException e) {
- throw new CloudRuntimeException("Unable to find " + fieldName, e);
+ throw new RuntimeException("Unable to find " + fieldName, e);
}
} else {
- assert _selects.size() == 0 : "You're selecting more than one item and yet is not providing a container class to put these items in. So what do you expect me to do. Spin magic?";
+ if (_selects.size() != 0) {
+ throw new RuntimeException(
+ "You're selecting more than one item and yet is not providing a container class to put these items in. So what do you expect me to do. Spin magic?");
+ }
}
Select select = new Select(func, _specifiedAttrs.size() == 0 ? null : _specifiedAttrs.get(0), field, params);
@@ -148,10 +162,6 @@ public class GenericSearchBuilder<T, K> implements MethodInterceptor {
return this;
}
-// public GenericSearchBuilder<T, K> select(String joinName, String fieldName, Func func, Object useless, Object... params) {
-//
-// }
-
@Override
public Object intercept(Object object, Method method, Object[] args, MethodProxy methodProxy) throws Throwable {
String name = method.getName();
@@ -175,7 +185,7 @@ public class GenericSearchBuilder<T, K> implements MethodInterceptor {
}
}
}
- assert false : "Perhaps you need to make the method start with get or is?";
+ throw new RuntimeException("Perhaps you need to make the method start with get or is: " + method);
}
}
return methodProxy.invokeSuper(object, args);
@@ -479,7 +489,9 @@ public class GenericSearchBuilder<T, K> implements MethodInterceptor {
sql.delete(sql.length() - 5, sql.length());
sql.append(" IS NOT NULL ");
} else {
- assert((op.getParams() == 0 && params == null) || (params.length == op.getParams())) : "Problem with condition: " + name;
+ if ((op.getParams() != 0 || params != null) && (params.length != op.getParams())) {
+ throw new RuntimeException("Problem with condition: " + name);
+ }
}
}
@@ -527,7 +539,7 @@ public class GenericSearchBuilder<T, K> implements MethodInterceptor {
public GenericSearchBuilder<T, K> values(Object... params) {
if (condition.op.getParams() > 0 && condition.op.params != params.length) {
- throw new CloudRuntimeException("The # of parameters set " + params.length + " does not match # of parameters required by " + condition.op);
+ throw new RuntimeException("The # of parameters set " + params.length + " does not match # of parameters required by " + condition.op);
}
condition.setPresets(params);
return builder;
[2/2] git commit: updated refs/heads/vmsync to 0233044
Posted by ah...@apache.org.
Deadlock fix
Project: http://git-wip-us.apache.org/repos/asf/cloudstack/repo
Commit: http://git-wip-us.apache.org/repos/asf/cloudstack/commit/0233044b
Tree: http://git-wip-us.apache.org/repos/asf/cloudstack/tree/0233044b
Diff: http://git-wip-us.apache.org/repos/asf/cloudstack/diff/0233044b
Branch: refs/heads/vmsync
Commit: 0233044b20bab0e0f7822660aa9e08bf530409ee
Parents: c6ba4a2
Author: Alex Huang <al...@gmail.com>
Authored: Thu Jun 27 14:37:59 2013 -0700
Committer: Alex Huang <al...@gmail.com>
Committed: Thu Jun 27 14:37:59 2013 -0700
----------------------------------------------------------------------
.../cloudstack/framework/jobs/AsyncJob.java | 2 +-
.../framework/jobs/AsyncJobManager.java | 2 -
.../jobs/dao/AsyncJobJoinMapDaoImpl.java | 2 +-
.../jobs/impl/AsyncJobManagerImpl.java | 157 ++++++++++++-------
4 files changed, 103 insertions(+), 60 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cloudstack/blob/0233044b/framework/jobs/src/org/apache/cloudstack/framework/jobs/AsyncJob.java
----------------------------------------------------------------------
diff --git a/framework/jobs/src/org/apache/cloudstack/framework/jobs/AsyncJob.java b/framework/jobs/src/org/apache/cloudstack/framework/jobs/AsyncJob.java
index 2ed75a9..61fb396 100644
--- a/framework/jobs/src/org/apache/cloudstack/framework/jobs/AsyncJob.java
+++ b/framework/jobs/src/org/apache/cloudstack/framework/jobs/AsyncJob.java
@@ -32,7 +32,7 @@ public interface AsyncJob extends JobInfo {
public static final String JOB_STATE = "job.state";
}
- public static interface Contants {
+ public static interface Constants {
// Although we may have detailed masks for each individual wakeup event, i.e.
// periodical timer, matched topic from message bus, it seems that we don't
http://git-wip-us.apache.org/repos/asf/cloudstack/blob/0233044b/framework/jobs/src/org/apache/cloudstack/framework/jobs/AsyncJobManager.java
----------------------------------------------------------------------
diff --git a/framework/jobs/src/org/apache/cloudstack/framework/jobs/AsyncJobManager.java b/framework/jobs/src/org/apache/cloudstack/framework/jobs/AsyncJobManager.java
index 440188a..bc06101 100644
--- a/framework/jobs/src/org/apache/cloudstack/framework/jobs/AsyncJobManager.java
+++ b/framework/jobs/src/org/apache/cloudstack/framework/jobs/AsyncJobManager.java
@@ -37,8 +37,6 @@ public interface AsyncJobManager extends Manager {
void completeAsyncJob(long jobId, JobInfo.Status jobStatus, int resultCode, String result);
- List<Long> wakeupByJoinedJobCompletion(long joinedJobId);
-
void updateAsyncJobStatus(long jobId, int processStatus, String resultObject);
void updateAsyncJobAttachment(long jobId, String instanceType, Long instanceId);
void logJobJournal(long jobId, AsyncJob.JournalType journalType, String
http://git-wip-us.apache.org/repos/asf/cloudstack/blob/0233044b/framework/jobs/src/org/apache/cloudstack/framework/jobs/dao/AsyncJobJoinMapDaoImpl.java
----------------------------------------------------------------------
diff --git a/framework/jobs/src/org/apache/cloudstack/framework/jobs/dao/AsyncJobJoinMapDaoImpl.java b/framework/jobs/src/org/apache/cloudstack/framework/jobs/dao/AsyncJobJoinMapDaoImpl.java
index ba6cfbc..fa3b14b 100644
--- a/framework/jobs/src/org/apache/cloudstack/framework/jobs/dao/AsyncJobJoinMapDaoImpl.java
+++ b/framework/jobs/src/org/apache/cloudstack/framework/jobs/dao/AsyncJobJoinMapDaoImpl.java
@@ -169,7 +169,7 @@ public class AsyncJobJoinMapDaoImpl extends GenericDaoBase<AsyncJobJoinMapVO, Lo
String sql = "UPDATE async_job SET job_pending_signals=? WHERE id IN " +
"(SELECT job_id FROM async_job_join_map WHERE next_wakeup < ? AND expiration > ?)";
pstmt = txn.prepareStatement(sql);
- pstmt.setInt(1, AsyncJob.Contants.SIGNAL_MASK_WAKEUP);
+ pstmt.setInt(1, AsyncJob.Constants.SIGNAL_MASK_WAKEUP);
pstmt.setString(2, DateUtil.getDateDisplayString(TimeZone.getTimeZone("GMT"), cutDate));
pstmt.setString(3, DateUtil.getDateDisplayString(TimeZone.getTimeZone("GMT"), cutDate));
pstmt.executeUpdate();
http://git-wip-us.apache.org/repos/asf/cloudstack/blob/0233044b/framework/jobs/src/org/apache/cloudstack/framework/jobs/impl/AsyncJobManagerImpl.java
----------------------------------------------------------------------
diff --git a/framework/jobs/src/org/apache/cloudstack/framework/jobs/impl/AsyncJobManagerImpl.java b/framework/jobs/src/org/apache/cloudstack/framework/jobs/impl/AsyncJobManagerImpl.java
index 2351a10..9b6aa97 100644
--- a/framework/jobs/src/org/apache/cloudstack/framework/jobs/impl/AsyncJobManagerImpl.java
+++ b/framework/jobs/src/org/apache/cloudstack/framework/jobs/impl/AsyncJobManagerImpl.java
@@ -19,12 +19,17 @@ package org.apache.cloudstack.framework.jobs.impl;
import java.io.File;
import java.io.FileInputStream;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Random;
+import java.util.TimeZone;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.RejectedExecutionException;
@@ -241,7 +246,7 @@ public class AsyncJobManagerImpl extends ManagerBase implements AsyncJobManager,
for(Long id : wakeupList) {
// TODO, we assume that all jobs in this category is API job only
AsyncJobVO jobToWakeup = _jobDao.findById(id);
- if (jobToWakeup != null && (jobToWakeup.getPendingSignals() & AsyncJob.Contants.SIGNAL_MASK_WAKEUP) != 0)
+ if (jobToWakeup != null && (jobToWakeup.getPendingSignals() & AsyncJob.Constants.SIGNAL_MASK_WAKEUP) != 0)
scheduleExecution(jobToWakeup, false);
}
@@ -353,8 +358,9 @@ public class AsyncJobManagerImpl extends ManagerBase implements AsyncJobManager,
// TODO
// this is a temporary solution to solve strange MySQL deadlock issue,
// completeJoin() causes deadlock happens at async_job table
+ // I removed the temporary solution already. I think my changes should fix the deadlock.
-/*
+/*
------------------------
LATEST DETECTED DEADLOCK
------------------------
@@ -387,25 +393,9 @@ public class AsyncJobManagerImpl extends ManagerBase implements AsyncJobManager,
0: len 8; hex 0000000000000009; asc ;; 1: len 6; hex 000005d8b0d7; asc ;; 2: len 7; hex 00000009280110; asc ( ;; 3: len 8; hex 0000000000000002; asc ;; 4: len 8; hex 0000000000000002; asc ;; 5: SQL NULL; 6: SQL NULL; 7: len 30; hex 6f72672e6170616368652e636c6f7564737461636b2e6170692e636f6d6d; asc org.apache.cloudstack.api.comm;...(truncated); 8: len 30; hex 7b226964223a2233222c22706879736963616c6e6574776f726b6964223a; asc {"id":"3","physicalnetworkid":;...(truncated); 9: len 4; hex 80000000; asc ;; 10: len 4; hex 80000001; asc ;; 11: len 4; hex 80000000; asc ;; 12: len 4; hex 80000000; asc ;; 13: len 30; hex 6f72672e6170616368652e636c6f7564737461636b2e6170692e72657370; asc org.apache.cloudstack.api.resp;...(truncated); 14: len 8; hex 80001a6f7bb0d0a8; asc o{ ;; 15: len 8; hex 80001a6f7bb0d0a8; asc o{ ;; 16: len 8; hex 8000124f06cfd5b6; asc O ;; 17: len 8; hex 8000124f06cfd5b6; asc O ;; 18: SQL NULL; 19: SQ
L NULL; 20: len 30; hex 62313065306432342d336233352d343663622d386361622d623933623562; asc b10e0d24-3b35-46cb-8cab-b93b5b;...(truncated); 21: len 30; hex 39353664383563632d383336622d346663612d623738622d646238343739; asc 956d85cc-836b-4fca-b78b-db8479;...(truncated); 22: SQL NULL; 23: len 21; hex 4170694173796e634a6f6244697370617463686572; asc ApiAsyncJobDispatcher;; 24: SQL NULL; 25: len 4; hex 80000000; asc ;;
*** WE ROLL BACK TRANSACTION (2)
-*/
+*/
- //
- // TODO
- // ACQUIRE_GLOBAL_LOCK_TIMEOUT_FOR_SYNC is a hard-coded time out value, this value
- // should actually be in sync with mysql settings
- //
- // TODO
- // how to handle failures from locking?
-
- if(_jobDao.lockInLockTable(AsyncJob.Contants.SYNC_LOCK_NAME, ACQUIRE_GLOBAL_LOCK_TIMEOUT_FOR_SYNC)) {
- try {
_joinMapDao.completeJoin(joinJobId, joinStatus, joinResult, getMsid());
- } finally {
- _jobDao.unlockFromLockTable(AsyncJob.Contants.SYNC_LOCK_NAME);
- }
- } else {
- s_logger.error("If this happens, it means too bad");
- }
}
@Override
@@ -524,7 +514,7 @@ public class AsyncJobManagerImpl extends ManagerBase implements AsyncJobManager,
s_logger.debug("Executing " + job);
}
- if ((getAndResetPendingSignals(job) & AsyncJob.Contants.SIGNAL_MASK_WAKEUP) != 0) {
+ if ((getAndResetPendingSignals(job) & AsyncJob.Constants.SIGNAL_MASK_WAKEUP) != 0) {
AsyncJobDispatcher jobDispatcher = getWakeupDispatcher(job);
if(jobDispatcher != null) {
jobDispatcher.runJob(job);
@@ -709,19 +699,13 @@ public class AsyncJobManagerImpl extends ManagerBase implements AsyncJobManager,
}
}
- if(_jobDao.lockInLockTable(AsyncJob.Contants.SYNC_LOCK_NAME, ACQUIRE_GLOBAL_LOCK_TIMEOUT_FOR_SYNC)) {
- try {
- List<Long> standaloneWakeupJobs = _joinMapDao.wakeupScan();
+ List<Long> standaloneWakeupJobs = wakeupScan();
for(Long jobId : standaloneWakeupJobs) {
// TODO, we assume that all jobs in this category is API job only
AsyncJobVO job = _jobDao.findById(jobId);
- if (job != null && (job.getPendingSignals() & AsyncJob.Contants.SIGNAL_MASK_WAKEUP) != 0)
+ if (job != null && (job.getPendingSignals() & AsyncJob.Constants.SIGNAL_MASK_WAKEUP) != 0)
scheduleExecution(job, false);
}
- } finally {
- _jobDao.unlockFromLockTable(AsyncJob.Contants.SYNC_LOCK_NAME);
- }
- }
} catch(Throwable e) {
s_logger.error("Unexpected exception when trying to execute queue item, ", e);
} finally {
@@ -830,6 +814,92 @@ public class AsyncJobManagerImpl extends ManagerBase implements AsyncJobManager,
}
}
+ @DB
+ protected List<Long> wakeupByJoinedJobCompletion(long joinedJobId) {
+ SearchCriteria<Long> joinJobSC = JoinJobSearch.create("joinJobId", joinedJobId);
+
+ List<Long> result = _joinMapDao.customSearch(joinJobSC, null);
+ if (result.size() != 0) {
+ Collections.sort(result);
+ Long[] ids = result.toArray(new Long[result.size()]);
+
+ SearchCriteria<AsyncJobVO> jobsSC = JobIdsSearch.create("ids", ids);
+ SearchCriteria<SyncQueueItemVO> queueItemsSC = QueueJobIdsSearch.create("contentIds", ids);
+
+ Transaction txn = Transaction.currentTxn();
+ txn.start();
+ AsyncJobVO job = _jobDao.createForUpdate();
+ job.setPendingSignals(AsyncJob.Constants.SIGNAL_MASK_WAKEUP);
+ _jobDao.update(job, jobsSC);
+
+ SyncQueueItemVO item = _queueItemDao.createForUpdate();
+ item.setLastProcessNumber(null);
+ item.setLastProcessMsid(null);
+ _queueItemDao.update(item, queueItemsSC);
+ txn.commit();
+ }
+ return _joinMapDao.findJobsToWake(joinedJobId);
+ }
+
+ @DB
+ protected List<Long> wakeupScan() {
+ List<Long> standaloneList = new ArrayList<Long>();
+
+ Date cutDate = DateUtil.currentGMTTime();
+
+ Transaction txn = Transaction.currentTxn();
+ PreparedStatement pstmt = null;
+ try {
+ txn.start();
+
+
+ //
+ // performance sensitive processing, do it in plain SQL
+ //
+ String sql = "UPDATE async_job SET job_pending_signals=? WHERE id IN " +
+ "(SELECT job_id FROM async_job_join_map WHERE next_wakeup < ? AND expiration > ?)";
+ pstmt = txn.prepareStatement(sql);
+ pstmt.setInt(1, AsyncJob.Constants.SIGNAL_MASK_WAKEUP);
+ pstmt.setString(2, DateUtil.getDateDisplayString(TimeZone.getTimeZone("GMT"), cutDate));
+ pstmt.setString(3, DateUtil.getDateDisplayString(TimeZone.getTimeZone("GMT"), cutDate));
+ pstmt.executeUpdate();
+ pstmt.close();
+
+ sql = "UPDATE sync_queue_item SET queue_proc_msid=NULL, queue_proc_number=NULL WHERE content_id IN " +
+ "(SELECT job_id FROM async_job_join_map WHERE next_wakeup < ? AND expiration > ?)";
+ pstmt = txn.prepareStatement(sql);
+ pstmt.setString(1, DateUtil.getDateDisplayString(TimeZone.getTimeZone("GMT"), cutDate));
+ pstmt.setString(2, DateUtil.getDateDisplayString(TimeZone.getTimeZone("GMT"), cutDate));
+ pstmt.executeUpdate();
+ pstmt.close();
+
+ sql = "SELECT job_id FROM async_job_join_map WHERE next_wakeup < ? AND expiration > ? AND job_id NOT IN (SELECT content_id FROM sync_queue_item)";
+ pstmt = txn.prepareStatement(sql);
+ pstmt.setString(1, DateUtil.getDateDisplayString(TimeZone.getTimeZone("GMT"), cutDate));
+ pstmt.setString(2, DateUtil.getDateDisplayString(TimeZone.getTimeZone("GMT"), cutDate));
+ ResultSet rs = pstmt.executeQuery();
+ while (rs.next()) {
+ standaloneList.add(rs.getLong(1));
+ }
+ rs.close();
+ pstmt.close();
+
+ // update for next wake-up
+ sql = "UPDATE async_job_join_map SET next_wakeup=DATE_ADD(next_wakeup, INTERVAL wakeup_interval SECOND) WHERE next_wakeup < ? AND expiration > ?";
+ pstmt = txn.prepareStatement(sql);
+ pstmt.setString(1, DateUtil.getDateDisplayString(TimeZone.getTimeZone("GMT"), cutDate));
+ pstmt.setString(2, DateUtil.getDateDisplayString(TimeZone.getTimeZone("GMT"), cutDate));
+ pstmt.executeUpdate();
+ pstmt.close();
+
+ txn.commit();
+ } catch (SQLException e) {
+ s_logger.error("Unexpected exception", e);
+ }
+
+ return standaloneList;
+ }
+
@Override
public boolean configure(String name, Map<String, Object> params) throws ConfigurationException {
_jobExpireSeconds = _configDepot.get(JobExpireMinutes).setMultiplier(60);
@@ -855,6 +925,8 @@ public class AsyncJobManagerImpl extends ManagerBase implements AsyncJobManager,
JoinJobSearch.selectField(JoinJobSearch.entity().getJobId());
JoinJobSearch.done();
+ JoinJobTimeSearch
+
JobIdsSearch = _jobDao.createSearchBuilder();
JobIdsSearch.and(JobIdsSearch.entity().getId(), Op.IN, "ids").done();
@@ -877,34 +949,6 @@ public class AsyncJobManagerImpl extends ManagerBase implements AsyncJobManager,
}
@Override
- @DB
- public List<Long> wakeupByJoinedJobCompletion(long joinedJobId) {
- SearchCriteria<Long> joinJobSC = JoinJobSearch.create("joinJobId", joinedJobId);
-
- List<Long> result = _joinMapDao.customSearch(joinJobSC, null);
- if (result.size() != 0) {
- Collections.sort(result);
- Long[] ids = result.toArray(new Long[result.size()]);
-
- SearchCriteria<AsyncJobVO> jobsSC = JobIdsSearch.create("ids", ids);
- SearchCriteria<SyncQueueItemVO> queueItemsSC = QueueJobIdsSearch.create("contentIds", ids);
-
- Transaction txn = Transaction.currentTxn();
- txn.start();
- AsyncJobVO job = _jobDao.createForUpdate();
- job.setPendingSignals(1);
- _jobDao.update(job, jobsSC);
-
- SyncQueueItemVO item = _queueItemDao.createForUpdate();
- item.setLastProcessNumber(null);
- item.setLastProcessMsid(null);
- _queueItemDao.update(item, queueItemsSC);
- txn.commit();
- }
- return _joinMapDao.findJobsToWake(joinedJobId);
- }
-
- @Override
public void onManagementNodeJoined(List<? extends ManagementServerHost> nodeList, long selfNodeId) {
}
@@ -960,6 +1004,7 @@ public class AsyncJobManagerImpl extends ManagerBase implements AsyncJobManager,
private SearchBuilder<AsyncJobVO> JobIdsSearch;
private SearchBuilder<SyncQueueItemVO> QueueJobIdsSearch;
private GenericSearchBuilder<AsyncJobJoinMapVO, Long> JoinJobIdsSearch;
+ private GenericSearchBuilder<AsyncJobJoinMapVO, Long> JoinJobTimeSearch;
protected AsyncJobManagerImpl() {