gian/cloudstack-mirror
1
0
Fork 0
mirror of https://github.com/apache/cloudstack synced 2026-08-31 01:05:57 +00:00
Code Issues Actions 2 Packages Projects Releases Wiki Activity

more job cleanup

Browse Source
This commit is contained in:
Kelven Yang 2013-04-09 11:14:49 -07:00
parent 5585b5ea6e
commit 55935f9868
9 changed files with 40 additions and 100 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
client/tomcatconf/applicationContext.xml.in
View File

@ -594,7 +594,6 @@
<bean id="apiRateLimitServiceImpl" class="org.apache.cloudstack.ratelimit.ApiRateLimitServiceImpl"/>
<bean id="alertManagerImpl" class="com.cloud.alert.AlertManagerImpl" />
<bean id="asyncJobExecutorContextImpl" class="com.cloud.async.AsyncJobExecutorContextImpl" />
<bean id="asyncJobManagerImpl" class="com.cloud.async.AsyncJobManagerImpl" />
<bean id="autoScaleManagerImpl" class="com.cloud.network.as.AutoScaleManagerImpl" />
<bean id="capacityManagerImpl" class="com.cloud.capacity.CapacityManagerImpl" />

34
server/src/com/cloud/async/BaseAsyncJobExecutor.java → server/src/com/cloud/async/AsyncJobExecutionContext.java
View File

@ -16,21 +16,17 @@
// under the License.
package com.cloud.async;
public abstract class BaseAsyncJobExecutor implements AsyncJobExecutor {
public class AsyncJobExecutionContext {
private SyncQueueItemVO _syncSource;
private AsyncJobVO _job;
private boolean _fromPreviousSession;
private AsyncJobManager _asyncJobMgr;
private static ThreadLocal<AsyncJobExecutor> s_currentExector = new ThreadLocal<AsyncJobExecutor>();
public AsyncJobManager getAsyncJobMgr() {
return _asyncJobMgr;
private static ThreadLocal<AsyncJobExecutionContext> s_currentExectionContext = new ThreadLocal<AsyncJobExecutionContext>();
public AsyncJobExecutionContext() {
}
public void setAsyncJobMgr(AsyncJobManager asyncMgr) {
_asyncJobMgr = asyncMgr;
public AsyncJobExecutionContext(AsyncJobVO job) {
_job = job;
}
public SyncQueueItemVO getSyncSource() {
@ -49,21 +45,11 @@ public abstract class BaseAsyncJobExecutor implements AsyncJobExecutor {
_job = job;
}
public void setFromPreviousSession(boolean value) {
_fromPreviousSession = value;
public static AsyncJobExecutionContext getCurrentExecutionContext() {
return s_currentExectionContext.get();
}
public boolean isFromPreviousSession() {
return _fromPreviousSession;
}
public abstract boolean execute();
public static AsyncJobExecutor getCurrentExecutor() {
return s_currentExector.get();
}
public static void setCurrentExecutor(AsyncJobExecutor currentExecutor) {
s_currentExector.set(currentExecutor);
public static void setCurrentExecutionContext(AsyncJobExecutionContext currentContext) {
s_currentExectionContext.set(currentContext);
}
}

39
server/src/com/cloud/async/AsyncJobExecutor.java
View File

@ -1,39 +0,0 @@
// 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.
package com.cloud.async;
public interface AsyncJobExecutor {
public AsyncJobManager getAsyncJobMgr();
public void setAsyncJobMgr(AsyncJobManager asyncMgr);
public SyncQueueItemVO getSyncSource();
public void setSyncSource(SyncQueueItemVO syncSource);
public AsyncJobVO getJob();
public void setJob(AsyncJobVO job);
public void setFromPreviousSession(boolean value);
public boolean isFromPreviousSession();
/**
*
* otherwise return false and once the executor finally has completed with the sync source,
* it needs to call AsyncJobManager.releaseSyncSource
*
* if executor does not have a sync source, always return true
*/
public boolean execute();
}

2
server/src/com/cloud/async/AsyncJobManager.java
View File

@ -36,7 +36,7 @@ public interface AsyncJobManager extends Manager {
public void updateAsyncJobStatus(long jobId, int processStatus, Object resultObject);
public void updateAsyncJobAttachment(long jobId, String instanceType, Long instanceId);
public void releaseSyncSource(AsyncJobExecutor executor);
public void releaseSyncSource();
public void syncAsyncJobExecution(AsyncJob job, String syncObjType, long syncObjId, long queueSizeLimit);

28
server/src/com/cloud/async/AsyncJobManagerImpl.java
View File

@ -325,7 +325,6 @@ public class AsyncJobManagerImpl extends ManagerBase implements AsyncJobManager,
txt.start();
AsyncJobVO job = _jobDao.findById(jobId);
if(job != null) {
jobResult.setCmdOriginator(job.getCmdOriginator());
jobResult.setJobStatus(job.getStatus());
jobResult.setProcessStatus(job.getProcessStatus());
jobResult.setResult(job.getResult());
@ -391,6 +390,8 @@ public class AsyncJobManagerImpl extends ManagerBase implements AsyncJobManager,
} catch(Exception e) {
s_logger.warn("Unable to register active job " + job.getId() + " to JMX monitoring due to exception " + ExceptionUtil.toString(e));
}
AsyncJobExecutionContext.setCurrentExecutionContext(new AsyncJobExecutionContext(job));
BaseAsyncCmd cmdObj = null;
Transaction txn = Transaction.open(Transaction.CLOUD_DB);
@ -500,6 +501,8 @@ public class AsyncJobManagerImpl extends ManagerBase implements AsyncJobManager,
s_logger.error("Caught: " + th);
} catch (Throwable th2) {
}
} finally {
AsyncJobExecutionContext.setCurrentExecutionContext(null);
}
}
};
@ -535,17 +538,20 @@ public class AsyncJobManagerImpl extends ManagerBase implements AsyncJobManager,
}
@Override
public void releaseSyncSource(AsyncJobExecutor executor) {
if(executor.getSyncSource() != null) {
public void releaseSyncSource() {
AsyncJobExecutionContext executionContext = AsyncJobExecutionContext.getCurrentExecutionContext();
assert(executionContext != null);
if(executionContext.getSyncSource() != null) {
if(s_logger.isDebugEnabled()) {
s_logger.debug("Release sync source for job-" + executor.getJob().getId() + " sync source: "
+ executor.getSyncSource().getContentType() + "-"
+ executor.getSyncSource().getContentId());
}
_queueMgr.purgeItem(executor.getSyncSource().getId());
checkQueue(executor.getSyncSource().getQueueId());
}
s_logger.debug("Release sync source for job-" + executionContext.getJob().getId() + " sync source: "
+ executionContext.getSyncSource().getContentType() + "-"
+ executionContext.getSyncSource().getContentId());
}
_queueMgr.purgeItem(executionContext.getSyncSource().getId());
checkQueue(executionContext.getSyncSource().getQueueId());
}
}
private void checkQueue(long queueId) {

9
server/src/com/cloud/async/AsyncJobResult.java
View File

@ -23,7 +23,6 @@ public class AsyncJobResult {
public static final int STATUS_SUCCEEDED = 1;
public static final int STATUS_FAILED = 2;
private String cmdOriginator;
private long jobId;
private int jobStatus;
private int processStatus;
@ -39,14 +38,6 @@ public class AsyncJobResult {
result = "";
}
public String getCmdOriginator() {
return cmdOriginator;
}
public void setCmdOriginator(String cmdOriginator) {
this.cmdOriginator = cmdOriginator;
}
public long getJobId() {
return jobId;
}

1
server/src/com/cloud/async/SyncQueueManager.java
View File

@ -20,7 +20,6 @@ import java.util.List;
import com.cloud.utils.component.Manager;
public interface SyncQueueManager extends Manager {
public SyncQueueVO queue(String syncObjType, long syncObjId, String itemType, long itemId, long queueSizeLimit);
public SyncQueueItemVO dequeueFromOne(long queueId, Long msid);

8
server/src/com/cloud/server/ManagementServerImpl.java
View File

@ -2833,9 +2833,9 @@ public class ManagementServerImpl extends ManagerBase implements ManagementServe
volume.getName(), ApiDBUtils.findAccountById(accountId).getUuid(), UploadVO.Status.COPY_IN_PROGRESS.toString(),
uploadJob.getUuid());
resultObj.setResponseName(cmd.getCommandName());
AsyncJobExecutor asyncExecutor = BaseAsyncJobExecutor.getCurrentExecutor();
if (asyncExecutor != null) {
job = asyncExecutor.getJob();
AsyncJobExecutionContext asyncExecutionContext = AsyncJobExecutionContext.getCurrentExecutionContext();
if (asyncExecutionContext != null) {
job = asyncExecutionContext.getJob();
_asyncMgr.updateAsyncJobAttachment(job.getId(), Upload.Type.VOLUME.toString(), volumeId);
_asyncMgr.updateAsyncJobStatus(job.getId(), AsyncJobResult.STATUS_IN_PROGRESS, resultObj);
}
@ -2857,7 +2857,7 @@ public class ManagementServerImpl extends ManagerBase implements ManagementServe
// Update the async job.
resultObj.setResultString(errorString);
resultObj.setUploadStatus(UploadVO.Status.COPY_ERROR.toString());
if (asyncExecutor != null) {
if (asyncExecutionContext != null) {
_asyncMgr.completeAsyncJob(job.getId(), AsyncJobResult.STATUS_FAILED, 0, resultObj);
}

18
server/src/com/cloud/storage/VolumeManagerImpl.java
View File

@ -71,10 +71,9 @@ import com.cloud.agent.api.AttachVolumeCommand;
import com.cloud.agent.api.to.VolumeTO;
import com.cloud.alert.AlertManager;
import com.cloud.api.ApiDBUtils;
import com.cloud.async.AsyncJobExecutor;
import com.cloud.async.AsyncJobManager;
import com.cloud.async.AsyncJobVO;
import com.cloud.async.BaseAsyncJobExecutor;
import com.cloud.async.AsyncJobExecutionContext;
import com.cloud.capacity.CapacityManager;
import com.cloud.capacity.dao.CapacityDao;
import com.cloud.configuration.Config;
@ -1743,10 +1742,10 @@ public class VolumeManagerImpl extends ManagerBase implements VolumeManager {
}
AsyncJobExecutor asyncExecutor = BaseAsyncJobExecutor
.getCurrentExecutor();
if (asyncExecutor != null) {
AsyncJobVO job = asyncExecutor.getJob();
AsyncJobExecutionContext asyncExecutionContext = AsyncJobExecutionContext.getCurrentExecutionContext();
if (asyncExecutionContext != null) {
AsyncJobVO job = asyncExecutionContext.getJob();
if (s_logger.isInfoEnabled()) {
s_logger.info("Trying to attaching volume " + volumeId
@ -1826,10 +1825,9 @@ public class VolumeManagerImpl extends ManagerBase implements VolumeManager {
"Please specify a VM that is either running or stopped.");
}
AsyncJobExecutor asyncExecutor = BaseAsyncJobExecutor
.getCurrentExecutor();
if (asyncExecutor != null) {
AsyncJobVO job = asyncExecutor.getJob();
AsyncJobExecutionContext asyncExecutionContext = AsyncJobExecutionContext.getCurrentExecutionContext();
if (asyncExecutionContext != null) {
AsyncJobVO job = asyncExecutionContext.getJob();
if (s_logger.isInfoEnabled()) {
s_logger.info("Trying to attaching volume " + volumeId