diff --git a/client/tomcatconf/applicationContext.xml.in b/client/tomcatconf/applicationContext.xml.in index ca6b4020364..e492f182df4 100644 --- a/client/tomcatconf/applicationContext.xml.in +++ b/client/tomcatconf/applicationContext.xml.in @@ -594,7 +594,6 @@ - diff --git a/server/src/com/cloud/async/BaseAsyncJobExecutor.java b/server/src/com/cloud/async/AsyncJobExecutionContext.java similarity index 59% rename from server/src/com/cloud/async/BaseAsyncJobExecutor.java rename to server/src/com/cloud/async/AsyncJobExecutionContext.java index 122b34bd181..d8cf3aa3b10 100644 --- a/server/src/com/cloud/async/BaseAsyncJobExecutor.java +++ b/server/src/com/cloud/async/AsyncJobExecutionContext.java @@ -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 s_currentExector = new ThreadLocal(); - - public AsyncJobManager getAsyncJobMgr() { - return _asyncJobMgr; + private static ThreadLocal s_currentExectionContext = new ThreadLocal(); + + 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); } } diff --git a/server/src/com/cloud/async/AsyncJobExecutor.java b/server/src/com/cloud/async/AsyncJobExecutor.java deleted file mode 100644 index d224c8f1dd1..00000000000 --- a/server/src/com/cloud/async/AsyncJobExecutor.java +++ /dev/null @@ -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(); -} - diff --git a/server/src/com/cloud/async/AsyncJobManager.java b/server/src/com/cloud/async/AsyncJobManager.java index 206df6a168b..f7ffc5aaef0 100644 --- a/server/src/com/cloud/async/AsyncJobManager.java +++ b/server/src/com/cloud/async/AsyncJobManager.java @@ -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); diff --git a/server/src/com/cloud/async/AsyncJobManagerImpl.java b/server/src/com/cloud/async/AsyncJobManagerImpl.java index 761ac361075..85e2a563f3e 100644 --- a/server/src/com/cloud/async/AsyncJobManagerImpl.java +++ b/server/src/com/cloud/async/AsyncJobManagerImpl.java @@ -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) { diff --git a/server/src/com/cloud/async/AsyncJobResult.java b/server/src/com/cloud/async/AsyncJobResult.java index cf343ea0402..4c379298dbd 100644 --- a/server/src/com/cloud/async/AsyncJobResult.java +++ b/server/src/com/cloud/async/AsyncJobResult.java @@ -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; } diff --git a/server/src/com/cloud/async/SyncQueueManager.java b/server/src/com/cloud/async/SyncQueueManager.java index a7032daaa47..fb4f2d36719 100644 --- a/server/src/com/cloud/async/SyncQueueManager.java +++ b/server/src/com/cloud/async/SyncQueueManager.java @@ -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); diff --git a/server/src/com/cloud/server/ManagementServerImpl.java b/server/src/com/cloud/server/ManagementServerImpl.java index d0904e1049c..b8de3659ca3 100755 --- a/server/src/com/cloud/server/ManagementServerImpl.java +++ b/server/src/com/cloud/server/ManagementServerImpl.java @@ -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); } diff --git a/server/src/com/cloud/storage/VolumeManagerImpl.java b/server/src/com/cloud/storage/VolumeManagerImpl.java index ff0235f018a..1f58a904c9c 100644 --- a/server/src/com/cloud/storage/VolumeManagerImpl.java +++ b/server/src/com/cloud/storage/VolumeManagerImpl.java @@ -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