Repository: ambari Updated Branches: refs/heads/branch-1.7.0 7566e570a -> dd572d354
http://git-wip-us.apache.org/repos/asf/ambari/blob/dd572d35/ambari-server/src/main/java/org/apache/ambari/server/state/svccomphost/ServiceComponentHostImpl.java ---------------------------------------------------------------------- diff --git a/ambari-server/src/main/java/org/apache/ambari/server/state/svccomphost/ServiceComponentHostImpl.java b/ambari-server/src/main/java/org/apache/ambari/server/state/svccomphost/ServiceComponentHostImpl.java index ade5792..15c16b0 100644 --- a/ambari-server/src/main/java/org/apache/ambari/server/state/svccomphost/ServiceComponentHostImpl.java +++ b/ambari-server/src/main/java/org/apache/ambari/server/state/svccomphost/ServiceComponentHostImpl.java @@ -131,12 +131,12 @@ public class ServiceComponentHostImpl implements ServiceComponentHost { State.INSTALLED, ServiceComponentHostEventType.HOST_SVCCOMP_OP_SUCCEEDED, new ServiceComponentHostOpCompletedTransition()) - + .addTransition(State.INSTALLED, State.INSTALLED, ServiceComponentHostEventType.HOST_SVCCOMP_OP_SUCCEEDED, new ServiceComponentHostOpCompletedTransition()) - + .addTransition(State.INSTALLING, State.INSTALLING, ServiceComponentHostEventType.HOST_SVCCOMP_OP_IN_PROGRESS, @@ -202,7 +202,7 @@ public class ServiceComponentHostImpl implements ServiceComponentHost { State.STARTING, ServiceComponentHostEventType.HOST_SVCCOMP_OP_IN_PROGRESS, new ServiceComponentHostOpInProgressTransition()) - + .addTransition(State.STARTING, State.STARTING, ServiceComponentHostEventType.HOST_SVCCOMP_START, @@ -211,7 +211,7 @@ public class ServiceComponentHostImpl implements ServiceComponentHost { State.STARTED, ServiceComponentHostEventType.HOST_SVCCOMP_STARTED, new ServiceComponentHostOpCompletedTransition()) - + .addTransition(State.STARTING, State.INSTALLED, ServiceComponentHostEventType.HOST_SVCCOMP_OP_FAILED, @@ -383,7 +383,7 @@ public class ServiceComponentHostImpl implements ServiceComponentHost { State.INSTALLING, ServiceComponentHostEventType.HOST_SVCCOMP_INSTALL, new ServiceComponentHostOpStartedTransition()) - + .addTransition(State.INSTALLING, State.INSTALLING, ServiceComponentHostEventType.HOST_SVCCOMP_OP_IN_PROGRESS, @@ -557,56 +557,44 @@ public class ServiceComponentHostImpl implements ServiceComponentHost { private void resetLastOpInfo() { - clusterGlobalLock.readLock().lock(); + writeLock.lock(); try { - try { - writeLock.lock(); - setLastOpStartTime(-1); - setLastOpLastUpdateTime(-1); - setLastOpEndTime(-1); - } finally { - writeLock.unlock(); - } + setLastOpStartTime(-1); + setLastOpLastUpdateTime(-1); + setLastOpEndTime(-1); } finally { - clusterGlobalLock.readLock().unlock(); + writeLock.unlock(); } - } private void updateLastOpInfo(ServiceComponentHostEventType eventType, long time) { - clusterGlobalLock.readLock().lock(); + writeLock.lock(); try { - try { - writeLock.lock(); - switch (eventType) { - case HOST_SVCCOMP_INSTALL: - case HOST_SVCCOMP_START: - case HOST_SVCCOMP_STOP: - case HOST_SVCCOMP_UNINSTALL: - case HOST_SVCCOMP_WIPEOUT: - case HOST_SVCCOMP_OP_RESTART: - resetLastOpInfo(); - setLastOpStartTime(time); - break; - case HOST_SVCCOMP_OP_FAILED: - case HOST_SVCCOMP_OP_SUCCEEDED: - case HOST_SVCCOMP_STOPPED: - case HOST_SVCCOMP_STARTED: - setLastOpLastUpdateTime(time); - setLastOpEndTime(time); - break; - case HOST_SVCCOMP_OP_IN_PROGRESS: - setLastOpLastUpdateTime(time); - break; - } - } finally { - writeLock.unlock(); + switch (eventType) { + case HOST_SVCCOMP_INSTALL: + case HOST_SVCCOMP_START: + case HOST_SVCCOMP_STOP: + case HOST_SVCCOMP_UNINSTALL: + case HOST_SVCCOMP_WIPEOUT: + case HOST_SVCCOMP_OP_RESTART: + resetLastOpInfo(); + setLastOpStartTime(time); + break; + case HOST_SVCCOMP_OP_FAILED: + case HOST_SVCCOMP_OP_SUCCEEDED: + case HOST_SVCCOMP_STOPPED: + case HOST_SVCCOMP_STARTED: + setLastOpLastUpdateTime(time); + setLastOpEndTime(time); + break; + case HOST_SVCCOMP_OP_IN_PROGRESS: + setLastOpLastUpdateTime(time); + break; } } finally { - clusterGlobalLock.readLock().unlock(); + writeLock.unlock(); } - } @AssistedInject @@ -615,13 +603,13 @@ public class ServiceComponentHostImpl implements ServiceComponentHost { injector.injectMembers(this); if (serviceComponent.isClientComponent()) { - this.stateMachine = clientStateMachineFactory.make(this); + stateMachine = clientStateMachineFactory.make(this); } else { - this.stateMachine = daemonStateMachineFactory.make(this); + stateMachine = daemonStateMachineFactory.make(this); } this.serviceComponent = serviceComponent; - this.clusterGlobalLock = serviceComponent.getClusterGlobalLock(); + clusterGlobalLock = serviceComponent.getClusterGlobalLock(); stateEntity = new HostComponentStateEntity(); stateEntity.setClusterId(serviceComponent.getClusterId()); @@ -646,14 +634,14 @@ public class ServiceComponentHostImpl implements ServiceComponentHost { } try { - this.host = clusters.getHost(hostName); + host = clusters.getHost(hostName); } catch (AmbariException e) { //TODO exception? LOG.error("Host '{}' was not found" + hostName); throw new RuntimeException(e); } - this.resetLastOpInfo(); + resetLastOpInfo(); } @AssistedInject @@ -663,21 +651,21 @@ public class ServiceComponentHostImpl implements ServiceComponentHost { Injector injector) { injector.injectMembers(this); this.serviceComponent = serviceComponent; - this.clusterGlobalLock = serviceComponent.getClusterGlobalLock(); + clusterGlobalLock = serviceComponent.getClusterGlobalLock(); this.desiredStateEntity = desiredStateEntity; this.stateEntity = stateEntity; //TODO implement State Machine init as now type choosing is hardcoded in above code if (serviceComponent.isClientComponent()) { - this.stateMachine = clientStateMachineFactory.make(this); + stateMachine = clientStateMachineFactory.make(this); } else { - this.stateMachine = daemonStateMachineFactory.make(this); + stateMachine = daemonStateMachineFactory.make(this); } - this.stateMachine.setCurrentState(stateEntity.getCurrentState()); + stateMachine.setCurrentState(stateEntity.getCurrentState()); try { - this.host = clusters.getHost(stateEntity.getHostName()); + host = clusters.getHost(stateEntity.getHostName()); } catch (AmbariException e) { //TODO exception? impossible due to database restrictions LOG.error("Host '{}' was not found " + stateEntity.getHostName()); @@ -689,18 +677,9 @@ public class ServiceComponentHostImpl implements ServiceComponentHost { @Override public State getState() { - clusterGlobalLock.readLock().lock(); - try { - readLock.lock(); - try { - return stateMachine.getCurrentState(); - } finally { - readLock.unlock(); - } - } finally { - clusterGlobalLock.readLock().unlock(); - } - + // there's no reason to lock around the state machine for this SCH since + // the state machine is synchronized + return stateMachine.getCurrentState(); } @Override @@ -743,8 +722,8 @@ public class ServiceComponentHostImpl implements ServiceComponentHost { } catch (InvalidStateTransitionException e) { LOG.debug("Can't handle ServiceComponentHostEvent event at" + " current state" - + ", serviceComponentName=" + this.getServiceComponentName() - + ", hostName=" + this.getHostName() + + ", serviceComponentName=" + getServiceComponentName() + + ", hostName=" + getHostName() + ", currentState=" + oldState + ", eventType=" + event.getType() + ", event=" + event); @@ -760,8 +739,8 @@ public class ServiceComponentHostImpl implements ServiceComponentHost { if (!oldState.equals(getState())) { if (LOG.isDebugEnabled()) { LOG.debug("ServiceComponentHost transitioned to a new state" - + ", serviceComponentName=" + this.getServiceComponentName() - + ", hostName=" + this.getHostName() + + ", serviceComponentName=" + getServiceComponentName() + + ", hostName=" + getHostName() + ", oldState=" + oldState + ", currentState=" + getState() + ", eventType=" + event.getType().name() @@ -772,48 +751,23 @@ public class ServiceComponentHostImpl implements ServiceComponentHost { @Override public String getServiceComponentName() { - clusterGlobalLock.readLock().lock(); - try { - readLock.lock(); - try { - return serviceComponent.getName(); - } finally { - readLock.unlock(); - } - } finally { - clusterGlobalLock.readLock().unlock(); - } + return serviceComponent.getName(); } @Override public String getHostName() { - clusterGlobalLock.readLock().lock(); - try { - readLock.lock(); - try { - return host.getHostName(); - } finally { - readLock.unlock(); - } - } finally { - clusterGlobalLock.readLock().unlock(); - } + return host.getHostName(); } /** * @return the lastOpStartTime */ public long getLastOpStartTime() { - clusterGlobalLock.readLock().lock(); + readLock.lock(); try { - readLock.lock(); - try { - return lastOpStartTime; - } finally { - readLock.unlock(); - } + return lastOpStartTime; } finally { - clusterGlobalLock.readLock().unlock(); + readLock.unlock(); } } @@ -821,16 +775,11 @@ public class ServiceComponentHostImpl implements ServiceComponentHost { * @param lastOpStartTime the lastOpStartTime to set */ public void setLastOpStartTime(long lastOpStartTime) { - clusterGlobalLock.readLock().lock(); + writeLock.lock(); try { - writeLock.lock(); - try { - this.lastOpStartTime = lastOpStartTime; - } finally { - writeLock.unlock(); - } + this.lastOpStartTime = lastOpStartTime; } finally { - clusterGlobalLock.readLock().unlock(); + writeLock.unlock(); } } @@ -838,16 +787,11 @@ public class ServiceComponentHostImpl implements ServiceComponentHost { * @return the lastOpEndTime */ public long getLastOpEndTime() { - clusterGlobalLock.readLock().lock(); + readLock.lock(); try { - readLock.lock(); - try { - return lastOpEndTime; - } finally { - readLock.unlock(); - } + return lastOpEndTime; } finally { - clusterGlobalLock.readLock().unlock(); + readLock.unlock(); } } @@ -855,16 +799,11 @@ public class ServiceComponentHostImpl implements ServiceComponentHost { * @param lastOpEndTime the lastOpEndTime to set */ public void setLastOpEndTime(long lastOpEndTime) { - clusterGlobalLock.readLock().lock(); + writeLock.lock(); try { - writeLock.lock(); - try { - this.lastOpEndTime = lastOpEndTime; - } finally { - writeLock.unlock(); - } + this.lastOpEndTime = lastOpEndTime; } finally { - clusterGlobalLock.readLock().unlock(); + writeLock.unlock(); } } @@ -872,16 +811,11 @@ public class ServiceComponentHostImpl implements ServiceComponentHost { * @return the lastOpLastUpdateTime */ public long getLastOpLastUpdateTime() { - clusterGlobalLock.readLock().lock(); + readLock.lock(); try { - readLock.lock(); - try { - return lastOpLastUpdateTime; - } finally { - readLock.unlock(); - } + return lastOpLastUpdateTime; } finally { - clusterGlobalLock.readLock().unlock(); + readLock.unlock(); } } @@ -889,250 +823,157 @@ public class ServiceComponentHostImpl implements ServiceComponentHost { * @param lastOpLastUpdateTime the lastOpLastUpdateTime to set */ public void setLastOpLastUpdateTime(long lastOpLastUpdateTime) { - clusterGlobalLock.readLock().lock(); + writeLock.lock(); try { - writeLock.lock(); - try { - this.lastOpLastUpdateTime = lastOpLastUpdateTime; - } finally { - writeLock.unlock(); - } + this.lastOpLastUpdateTime = lastOpLastUpdateTime; } finally { - clusterGlobalLock.readLock().unlock(); + writeLock.unlock(); } } @Override public long getClusterId() { - clusterGlobalLock.readLock().lock(); - try { - readLock.lock(); - try { - return serviceComponent.getClusterId(); - } finally { - readLock.unlock(); - } - } finally { - clusterGlobalLock.readLock().unlock(); - } + return serviceComponent.getClusterId(); } @Override public String getServiceName() { - clusterGlobalLock.readLock().lock(); - try { - readLock.lock(); - try { - return serviceComponent.getServiceName(); - } finally { - readLock.unlock(); - } - } finally { - clusterGlobalLock.readLock().unlock(); - } + return serviceComponent.getServiceName(); } @Override public StackId getStackVersion() { - clusterGlobalLock.readLock().lock(); + readLock.lock(); try { - readLock.lock(); - try { - return gson.fromJson(stateEntity.getCurrentStackVersion(), StackId.class); - } finally { - readLock.unlock(); - } + return gson.fromJson(stateEntity.getCurrentStackVersion(), StackId.class); } finally { - clusterGlobalLock.readLock().unlock(); + readLock.unlock(); } } @Override public void setStackVersion(StackId stackVersion) { - clusterGlobalLock.readLock().lock(); + writeLock.lock(); try { - writeLock.lock(); - try { - stateEntity.setCurrentStackVersion(gson.toJson(stackVersion)); - saveIfPersisted(); - } finally { - writeLock.unlock(); - } + stateEntity.setCurrentStackVersion(gson.toJson(stackVersion)); + saveIfPersisted(); } finally { - clusterGlobalLock.readLock().unlock(); + writeLock.unlock(); } } @Override public State getDesiredState() { - clusterGlobalLock.readLock().lock(); + readLock.lock(); try { - readLock.lock(); - try { - return desiredStateEntity.getDesiredState(); - } finally { - readLock.unlock(); - } + return desiredStateEntity.getDesiredState(); } finally { - clusterGlobalLock.readLock().unlock(); + readLock.unlock(); } } @Override public void setDesiredState(State state) { - clusterGlobalLock.readLock().lock(); + writeLock.lock(); try { - writeLock.lock(); - try { - desiredStateEntity.setDesiredState(state); - saveIfPersisted(); - } finally { - writeLock.unlock(); - } + desiredStateEntity.setDesiredState(state); + saveIfPersisted(); } finally { - clusterGlobalLock.readLock().unlock(); + writeLock.unlock(); } } @Override public StackId getDesiredStackVersion() { - clusterGlobalLock.readLock().lock(); + readLock.lock(); try { - readLock.lock(); - try { - return gson.fromJson(desiredStateEntity.getDesiredStackVersion(), StackId.class); - } finally { - readLock.unlock(); - } + return gson.fromJson(desiredStateEntity.getDesiredStackVersion(), + StackId.class); } finally { - clusterGlobalLock.readLock().unlock(); + readLock.unlock(); } } @Override public void setDesiredStackVersion(StackId stackVersion) { - clusterGlobalLock.readLock().lock(); + writeLock.lock(); try { - writeLock.lock(); - try { - desiredStateEntity.setDesiredStackVersion(gson.toJson(stackVersion)); - saveIfPersisted(); - } finally { - writeLock.unlock(); - } + desiredStateEntity.setDesiredStackVersion(gson.toJson(stackVersion)); + saveIfPersisted(); } finally { - clusterGlobalLock.readLock().unlock(); + writeLock.unlock(); } } @Override public HostComponentAdminState getComponentAdminState() { - clusterGlobalLock.readLock().lock(); + readLock.lock(); try { - readLock.lock(); - try { - HostComponentAdminState adminState = desiredStateEntity.getAdminState(); - if (adminState == null - && !serviceComponent.isClientComponent() && !serviceComponent.isMasterComponent()) { - adminState = HostComponentAdminState.INSERVICE; - } - return adminState; - } finally { - readLock.unlock(); + HostComponentAdminState adminState = desiredStateEntity.getAdminState(); + if (adminState == null && !serviceComponent.isClientComponent() + && !serviceComponent.isMasterComponent()) { + adminState = HostComponentAdminState.INSERVICE; } + return adminState; } finally { - clusterGlobalLock.readLock().unlock(); + readLock.unlock(); } } @Override public void setComponentAdminState(HostComponentAdminState attribute) { - clusterGlobalLock.readLock().lock(); + writeLock.lock(); try { - writeLock.lock(); - try { - desiredStateEntity.setAdminState(attribute); - saveIfPersisted(); - } finally { - writeLock.unlock(); - } + desiredStateEntity.setAdminState(attribute); + saveIfPersisted(); } finally { - clusterGlobalLock.readLock().unlock(); + writeLock.unlock(); } } @Override public ServiceComponentHostResponse convertToResponse() { - clusterGlobalLock.readLock().lock(); + readLock.lock(); try { - readLock.lock(); - try { - ServiceComponentHostResponse r = new ServiceComponentHostResponse( - serviceComponent.getClusterName(), - serviceComponent.getServiceName(), - serviceComponent.getName(), - getHostName(), - getState().toString(), - getStackVersion().getStackId(), - getDesiredState().toString(), - getDesiredStackVersion().getStackId(), - getComponentAdminState()); - - r.setActualConfigs(actualConfigs); + ServiceComponentHostResponse r = new ServiceComponentHostResponse( + serviceComponent.getClusterName(), serviceComponent.getServiceName(), + serviceComponent.getName(), getHostName(), getState().toString(), + getStackVersion().getStackId(), getDesiredState().toString(), + getDesiredStackVersion().getStackId(), getComponentAdminState()); - try { - r.setStaleConfig(helper.isStaleConfigs(this)); - } catch (Exception e) { - LOG.error("Could not determine stale config", e); - } + r.setActualConfigs(actualConfigs); - return r; - } finally { - readLock.unlock(); + try { + r.setStaleConfig(helper.isStaleConfigs(this)); + } catch (Exception e) { + LOG.error("Could not determine stale config", e); } + + return r; } finally { - clusterGlobalLock.readLock().unlock(); + readLock.unlock(); } } @Override public String getClusterName() { - clusterGlobalLock.readLock().lock(); - try { - readLock.lock(); - try { - return serviceComponent.getClusterName(); - } finally { - readLock.unlock(); - } - } finally { - clusterGlobalLock.readLock().unlock(); - } - + return serviceComponent.getClusterName(); } @Override public void debugDump(StringBuilder sb) { - clusterGlobalLock.readLock().lock(); + readLock.lock(); try { - readLock.lock(); - try { - sb.append("ServiceComponentHost={ hostname=").append(getHostName()) - .append(", serviceComponentName=").append(serviceComponent.getName()) - .append(", clusterName=").append(serviceComponent.getClusterName()) - .append(", serviceName=").append(serviceComponent.getServiceName()) - .append(", desiredStackVersion=").append(getDesiredStackVersion()) - .append(", desiredState=").append(getDesiredState()) - .append(", stackVersion=").append(getStackVersion()) - .append(", state=").append(getState()) - .append(" }"); - } finally { - readLock.unlock(); - } + sb.append("ServiceComponentHost={ hostname=").append(getHostName()).append( + ", serviceComponentName=").append(serviceComponent.getName()).append( + ", clusterName=").append(serviceComponent.getClusterName()).append( + ", serviceName=").append(serviceComponent.getServiceName()).append( + ", desiredStackVersion=").append(getDesiredStackVersion()).append( + ", desiredState=").append(getDesiredState()).append(", stackVersion=").append( + getStackVersion()).append(", state=").append(getState()).append(" }"); } finally { - clusterGlobalLock.readLock().unlock(); + readLock.unlock(); } - } @Override @@ -1153,12 +994,23 @@ public class ServiceComponentHostImpl implements ServiceComponentHost { @Override public void persist() { - clusterGlobalLock.readLock().lock(); + boolean clusterWriteLockAcquired = false; + if (!persisted) { + clusterGlobalLock.writeLock().lock(); + clusterWriteLockAcquired = true; + } + try { writeLock.lock(); try { if (!persisted) { + // persist the new cluster topology and then release the cluster lock + // as it has no more bearing on the rest of this persist() method persistEntities(); + clusterGlobalLock.writeLock().unlock(); + clusterWriteLockAcquired = false; + + // these shoudl still be done with the internal lock refresh(); host.refresh(); serviceComponent.refresh(); @@ -1170,9 +1022,10 @@ public class ServiceComponentHostImpl implements ServiceComponentHost { writeLock.unlock(); } } finally { - clusterGlobalLock.readLock().unlock(); + if (clusterWriteLockAcquired) { + clusterGlobalLock.writeLock().unlock(); + } } - } @Transactional @@ -1204,33 +1057,27 @@ public class ServiceComponentHostImpl implements ServiceComponentHost { @Override @Transactional public void refresh() { - clusterGlobalLock.readLock().lock(); + writeLock.lock(); try { - writeLock.lock(); - try { - if (isPersisted()) { - HostComponentStateEntityPK pk = new HostComponentStateEntityPK(); - HostComponentDesiredStateEntityPK dpk = new HostComponentDesiredStateEntityPK(); - pk.setClusterId(getClusterId()); - pk.setComponentName(getServiceComponentName()); - pk.setServiceName(getServiceName()); - pk.setHostName(getHostName()); - dpk.setClusterId(getClusterId()); - dpk.setComponentName(getServiceComponentName()); - dpk.setServiceName(getServiceName()); - dpk.setHostName(getHostName()); - stateEntity = hostComponentStateDAO.findByPK(pk); - desiredStateEntity = hostComponentDesiredStateDAO.findByPK(dpk); - hostComponentStateDAO.refresh(stateEntity); - hostComponentDesiredStateDAO.refresh(desiredStateEntity); - } - } finally { - writeLock.unlock(); + if (isPersisted()) { + HostComponentStateEntityPK pk = new HostComponentStateEntityPK(); + HostComponentDesiredStateEntityPK dpk = new HostComponentDesiredStateEntityPK(); + pk.setClusterId(getClusterId()); + pk.setComponentName(getServiceComponentName()); + pk.setServiceName(getServiceName()); + pk.setHostName(getHostName()); + dpk.setClusterId(getClusterId()); + dpk.setComponentName(getServiceComponentName()); + dpk.setServiceName(getServiceName()); + dpk.setHostName(getHostName()); + stateEntity = hostComponentStateDAO.findByPK(pk); + desiredStateEntity = hostComponentDesiredStateDAO.findByPK(dpk); + hostComponentStateDAO.refresh(stateEntity); + hostComponentDesiredStateDAO.refresh(desiredStateEntity); } } finally { - clusterGlobalLock.readLock().unlock(); + writeLock.unlock(); } - } @Transactional @@ -1244,19 +1091,21 @@ public class ServiceComponentHostImpl implements ServiceComponentHost { @Override public boolean canBeRemoved() { clusterGlobalLock.readLock().lock(); + boolean schLockAcquired = false; try { - readLock.lock(); - try { - - return (getState().isRemovableState()); + // if unable to read, then writers are writing; cannot remove SCH + schLockAcquired = readLock.tryLock(); + if (!schLockAcquired) { + return false; + } - } finally { + return (getState().isRemovableState()); + } finally { + if (schLockAcquired) { readLock.unlock(); } - } finally { clusterGlobalLock.readLock().unlock(); } - } @Override @@ -1269,7 +1118,7 @@ public class ServiceComponentHostImpl implements ServiceComponentHost { removeEntities(); persisted = false; } - clusters.getCluster(this.getClusterName()).removeServiceComponentHost(this); + clusters.getCluster(getClusterName()).removeServiceComponentHost(this); } catch (AmbariException ex) { if (LOG.isDebugEnabled()) { LOG.error(ex.getMessage()); @@ -1301,7 +1150,7 @@ public class ServiceComponentHostImpl implements ServiceComponentHost { hostComponentDesiredStateDAO.removeByPK(desiredPK); } - + @Override public void updateActualConfigs(Map<String, Map<String, String>> configTags) { Map<Long, ConfigGroup> configGroupMap; @@ -1314,165 +1163,120 @@ public class ServiceComponentHostImpl implements ServiceComponentHost { return; } - clusterGlobalLock.readLock().lock(); + writeLock.lock(); try { - writeLock.lock(); - try { - LOG.debug("Updating actual config tags: " + configTags); - actualConfigs = new HashMap<String, HostConfig>(); - - for (Entry<String, Map<String, String>> entry : configTags.entrySet()) { - String type = entry.getKey(); - Map<String, String> values = new HashMap<String, String>(entry.getValue()); - - String tag = values.get(ConfigHelper.CLUSTER_DEFAULT_TAG); - values.remove(ConfigHelper.CLUSTER_DEFAULT_TAG); - - HostConfig hc = new HostConfig(); - hc.setDefaultVersionTag(tag); - actualConfigs.put(type, hc); - - if (!values.isEmpty()) { - for (Entry<String, String> overrideEntry : values.entrySet()) { - Long groupId = Long.parseLong(overrideEntry.getKey()); - hc.getConfigGroupOverrides().put(groupId, overrideEntry.getValue()); - if (!configGroupMap.containsKey(groupId)) { - LOG.debug("Config group does not exist, id = " + groupId); - } + LOG.debug("Updating actual config tags: " + configTags); + actualConfigs = new HashMap<String, HostConfig>(); + + for (Entry<String, Map<String, String>> entry : configTags.entrySet()) { + String type = entry.getKey(); + Map<String, String> values = new HashMap<String, String>( + entry.getValue()); + + String tag = values.get(ConfigHelper.CLUSTER_DEFAULT_TAG); + values.remove(ConfigHelper.CLUSTER_DEFAULT_TAG); + + HostConfig hc = new HostConfig(); + hc.setDefaultVersionTag(tag); + actualConfigs.put(type, hc); + + if (!values.isEmpty()) { + for (Entry<String, String> overrideEntry : values.entrySet()) { + Long groupId = Long.parseLong(overrideEntry.getKey()); + hc.getConfigGroupOverrides().put(groupId, overrideEntry.getValue()); + if (!configGroupMap.containsKey(groupId)) { + LOG.debug("Config group does not exist, id = " + groupId); } } } - } finally { - writeLock.unlock(); } } finally { - clusterGlobalLock.readLock().unlock(); + writeLock.unlock(); } } - - - + + + @Override public Map<String, HostConfig> getActualConfigs() { - clusterGlobalLock.readLock().lock(); + readLock.lock(); try { - readLock.lock(); - try { - return actualConfigs; - } finally { - readLock.unlock(); - } + return actualConfigs; } finally { - clusterGlobalLock.readLock().unlock(); + readLock.unlock(); } - } @Override public HostState getHostState() { - clusterGlobalLock.readLock().lock(); + readLock.lock(); try { - readLock.lock(); - try { - return host.getState(); - } finally { - readLock.unlock(); - } + return host.getState(); } finally { - clusterGlobalLock.readLock().unlock(); + readLock.unlock(); } } - + @Override public void setMaintenanceState(MaintenanceState state) { - clusterGlobalLock.readLock().lock(); + writeLock.lock(); try { - writeLock.lock(); - try { - desiredStateEntity.setMaintenanceState(state); - saveIfPersisted(); - } finally { - writeLock.unlock(); - } + desiredStateEntity.setMaintenanceState(state); + saveIfPersisted(); } finally { - clusterGlobalLock.readLock().unlock(); + writeLock.unlock(); } } @Override public MaintenanceState getMaintenanceState() { - clusterGlobalLock.readLock().lock(); + readLock.lock(); try { - readLock.lock(); - try { - return desiredStateEntity.getMaintenanceState(); - } finally { - readLock.unlock(); - } + return desiredStateEntity.getMaintenanceState(); } finally { - clusterGlobalLock.readLock().unlock(); + readLock.unlock(); } } - + @Override public void setProcesses(List<Map<String, String>> procs) { - clusterGlobalLock.readLock().lock(); + writeLock.lock(); try { - writeLock.lock(); - try { - processes = Collections.unmodifiableList(procs); - } finally { - writeLock.unlock(); - } + processes = Collections.unmodifiableList(procs); } finally { - clusterGlobalLock.readLock().unlock(); + writeLock.unlock(); } } - - @Override + + @Override public List<Map<String, String>> getProcesses() { - clusterGlobalLock.readLock().lock(); + readLock.lock(); try { - readLock.lock(); - try { - return processes; - } finally { - readLock.unlock(); - } + return processes; } finally { - clusterGlobalLock.readLock().unlock(); + readLock.unlock(); } } @Override public boolean isRestartRequired() { - clusterGlobalLock.readLock().lock(); + readLock.lock(); try { - readLock.lock(); - try { - return desiredStateEntity.isRestartRequired(); - } finally { - readLock.unlock(); - } + return desiredStateEntity.isRestartRequired(); } finally { - clusterGlobalLock.readLock().unlock(); + readLock.unlock(); } } @Override public void setRestartRequired(boolean restartRequired) { - clusterGlobalLock.readLock().lock(); + writeLock.lock(); try { - writeLock.lock(); - try { - desiredStateEntity.setRestartRequired(restartRequired); - saveIfPersisted(); - helper.invalidateStaleConfigsCache(this); - } finally { - writeLock.unlock(); - } + desiredStateEntity.setRestartRequired(restartRequired); + saveIfPersisted(); + helper.invalidateStaleConfigsCache(this); } finally { - clusterGlobalLock.readLock().unlock(); + writeLock.unlock(); } } } http://git-wip-us.apache.org/repos/asf/ambari/blob/dd572d35/ambari-server/src/test/java/org/apache/ambari/server/state/cluster/ClusterDeadlockTest.java ---------------------------------------------------------------------- diff --git a/ambari-server/src/test/java/org/apache/ambari/server/state/cluster/ClusterDeadlockTest.java b/ambari-server/src/test/java/org/apache/ambari/server/state/cluster/ClusterDeadlockTest.java new file mode 100644 index 0000000..d1b5c3a --- /dev/null +++ b/ambari-server/src/test/java/org/apache/ambari/server/state/cluster/ClusterDeadlockTest.java @@ -0,0 +1,357 @@ +/** + * 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 org.apache.ambari.server.state.cluster; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.atomic.AtomicInteger; + +import org.apache.ambari.server.AmbariException; +import org.apache.ambari.server.ServiceComponentNotFoundException; +import org.apache.ambari.server.ServiceNotFoundException; +import org.apache.ambari.server.api.services.AmbariMetaInfo; +import org.apache.ambari.server.orm.GuiceJpaInitializer; +import org.apache.ambari.server.orm.InMemoryDefaultTestModule; +import org.apache.ambari.server.state.Cluster; +import org.apache.ambari.server.state.Clusters; +import org.apache.ambari.server.state.Host; +import org.apache.ambari.server.state.MaintenanceState; +import org.apache.ambari.server.state.Service; +import org.apache.ambari.server.state.ServiceComponent; +import org.apache.ambari.server.state.ServiceComponentFactory; +import org.apache.ambari.server.state.ServiceComponentHost; +import org.apache.ambari.server.state.ServiceComponentHostFactory; +import org.apache.ambari.server.state.ServiceFactory; +import org.apache.ambari.server.state.StackId; +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; + +import com.google.inject.Guice; +import com.google.inject.Inject; +import com.google.inject.Injector; +import com.google.inject.persist.PersistService; + +/** + * Tests AMBARI-9368 which produced a deadlock during read and writes of some of + * the impl classes. + */ +public class ClusterDeadlockTest { + private final AtomicInteger hostNameCounter = new AtomicInteger(0); + + @Inject + private Injector injector; + + @Inject + private Clusters clusters; + + @Inject + private ServiceFactory serviceFactory; + + @Inject + private ServiceComponentFactory serviceComponentFactory; + + @Inject + private ServiceComponentHostFactory serviceComponentHostFactory; + + @Inject + private AmbariMetaInfo metaInfo; + + + @Before + public void setup() throws Exception { + injector = Guice.createInjector(new InMemoryDefaultTestModule()); + injector.getInstance(GuiceJpaInitializer.class); + injector.injectMembers(this); + clusters.addCluster("c1"); + + StackId stackId = new StackId("HDP-0.1"); + Cluster c1 = clusters.getCluster("c1"); + c1.setDesiredStackVersion(stackId); + metaInfo.init(); + + // 100 hosts + for (int i = 0; i < 100; i++) { + String hostName = "c64-" + i; + clusters.addHost(hostName); + setOsFamily(clusters.getHost(hostName), "redhat", "6.4"); + clusters.getHost(hostName).persist(); + clusters.mapHostToCluster(hostName, "c1"); + } + + // force creation of the service and the components on the last host + createNewServiceComponentHost("HDFS", "NAMENODE", "c64-99", false); + createNewServiceComponentHost("HDFS", "HDFS_CLIENT", "c64-99", true); + } + + @After + public void teardown() { + injector.getInstance(PersistService.class).stop(); + } + + /** + * Tests that concurrent impl serialization and impl writing doesn't cause a + * deadlock. + * + * @throws Exception + */ + @Test(timeout = 30000) + public void testDeadlockBetweenImplementations() throws Exception { + Cluster cluster = clusters.getCluster("c1"); + Service service = cluster.getService("HDFS"); + ServiceComponent namenodeComponent = service.getServiceComponent("NAMENODE"); + ServiceComponent hdfsClientComponent = service.getServiceComponent("HDFS_CLIENT"); + + ServiceComponentHost namenodeSCH = createNewServiceComponentHost("HDFS", + "NAMENODE", "c64-0", false); + + ServiceComponentHost hdfsClientSCH = createNewServiceComponentHost("HDFS", + "HDFS_CLIENT", "c64-0", true); + + List<Thread> threads = new ArrayList<Thread>(); + for (int i = 0; i < 3; i++) { + DeadlockExerciserThread thread = new DeadlockExerciserThread(); + thread.setCluster(cluster); + thread.setService(service); + thread.setHdfsClientComponent(hdfsClientComponent); + thread.setNamenodeComponent(namenodeComponent); + thread.setNamenodeSCH(namenodeSCH); + thread.setHdfsClientSCH(hdfsClientSCH); + thread.start(); + threads.add(thread); + } + + for (Thread thread : threads) { + thread.join(); + } + } + + /** + * Tests that while serializing a service component, writes to that service + * component do not cause a deadlock with the global cluster lock. + * + * @throws Exception + */ + @Test(timeout = 30000) + public void testAddingHostComponentsWhileReading() throws Exception { + Cluster cluster = clusters.getCluster("c1"); + Service service = cluster.getService("HDFS"); + ServiceComponent namenodeComponent = service.getServiceComponent("NAMENODE"); + ServiceComponent hdfsClientComponent = service.getServiceComponent("HDFS_CLIENT"); + + List<Thread> threads = new ArrayList<Thread>(); + for (int i = 0; i < 5; i++) { + ServiceComponentDeadlockThread thread = new ServiceComponentDeadlockThread(); + thread.setHdfsClientComponent(hdfsClientComponent); + thread.setNamenodeComponent(namenodeComponent); + thread.start(); + threads.add(thread); + } + + for (Thread thread : threads) { + thread.join(); + } + } + + /** + * Tests AMBARI-9368 which saw a deadlock when adding a service component host + * while reading a service component. + */ + private final class ServiceComponentDeadlockThread extends Thread { + private ServiceComponent namenodeComponent; + private ServiceComponent hdfsClientComponent; + + /** + * @param namenodeComponent + * the namenodeComponent to set + */ + public void setNamenodeComponent(ServiceComponent namenodeComponent) { + this.namenodeComponent = namenodeComponent; + } + + /** + * @param hdfsClientComponent + * the hdfsClientComponent to set + */ + public void setHdfsClientComponent(ServiceComponent hdfsClientComponent) { + this.hdfsClientComponent = hdfsClientComponent; + } + + /** + * {@inheritDoc} + */ + @Override + public void run() { + try { + for (int i = 0; i < 15; i++) { + int hostNumeric = hostNameCounter.getAndIncrement(); + + namenodeComponent.convertToResponse(); + createNewServiceComponentHost("HDFS", "NAMENODE", "c64-" + + hostNumeric, false); + + hdfsClientComponent.convertToResponse(); + createNewServiceComponentHost("HDFS", "HDFS_CLIENT", "c64-" + + hostNumeric, true); + + Thread.sleep(10); + } + } catch (Exception exception) { + throw new RuntimeException(exception); + } + } + } + + /** + * Tests AMBARI-9368 which produced a deadlock during read and writes of some + * of the impl classes. + */ + private static final class DeadlockExerciserThread extends Thread { + private Cluster cluster; + private Service service; + private ServiceComponent namenodeComponent; + private ServiceComponent hdfsClientComponent; + private ServiceComponentHost namenodeSCH; + private ServiceComponentHost hdfsClientSCH; + + /** + * @param cluster + * the cluster to set + */ + public void setCluster(Cluster cluster) { + this.cluster = cluster; + } + + /** + * @param service + * the service to set + */ + public void setService(Service service) { + this.service = service; + } + + /** + * @param namenodeComponent + * the namenodeComponent to set + */ + public void setNamenodeComponent(ServiceComponent namenodeComponent) { + this.namenodeComponent = namenodeComponent; + } + + /** + * @param hdfsClientComponent + * the hdfsClientComponent to set + */ + public void setHdfsClientComponent(ServiceComponent hdfsClientComponent) { + this.hdfsClientComponent = hdfsClientComponent; + } + + /** + * @param namenodeSCH + * the namenodeSCH to set + */ + public void setNamenodeSCH(ServiceComponentHost namenodeSCH) { + this.namenodeSCH = namenodeSCH; + } + + /** + * @param hdfsClientSCH + * the hdfsClientSCH to set + */ + public void setHdfsClientSCH(ServiceComponentHost hdfsClientSCH) { + this.hdfsClientSCH = hdfsClientSCH; + } + + /** + * {@inheritDoc} + */ + @Override + public void run() { + try { + for (int i = 0; i < 10; i++) { + cluster.convertToResponse(); + service.convertToResponse(); + namenodeComponent.convertToResponse(); + hdfsClientComponent.convertToResponse(); + namenodeSCH.convertToResponse(); + hdfsClientSCH.convertToResponse(); + + cluster.setProvisioningState(org.apache.ambari.server.state.State.INIT); + service.setMaintenanceState(MaintenanceState.OFF); + namenodeComponent.setDesiredState(org.apache.ambari.server.state.State.STARTED); + hdfsClientComponent.setDesiredState(org.apache.ambari.server.state.State.INSTALLED); + + namenodeSCH.setState(org.apache.ambari.server.state.State.STARTED); + hdfsClientSCH.setState(org.apache.ambari.server.state.State.INSTALLED); + + Thread.sleep(100); + } + } catch (Exception exception) { + throw new RuntimeException(exception); + } + } + } + + private void setOsFamily(Host host, String osFamily, String osVersion) { + Map<String, String> hostAttributes = new HashMap<String, String>(2); + hostAttributes.put("os_family", osFamily); + hostAttributes.put("os_release_version", osVersion); + host.setHostAttributes(hostAttributes); + } + + private ServiceComponentHost createNewServiceComponentHost(String svc, + String svcComponent, String hostName, boolean isClient) + throws AmbariException { + Cluster c = clusters.getCluster("c1"); + Assert.assertNotNull(c.getConfigGroups()); + return createNewServiceComponentHost(c, svc, svcComponent, hostName); + } + + private ServiceComponentHost createNewServiceComponentHost(Cluster c, + String svc, String svcComponent, String hostName) throws AmbariException { + + Service s = null; + + try { + s = c.getService(svc); + } catch (ServiceNotFoundException e) { + s = serviceFactory.createNew(c, svc); + c.addService(s); + s.persist(); + } + + ServiceComponent sc = null; + try { + sc = s.getServiceComponent(svcComponent); + } catch (ServiceComponentNotFoundException e) { + sc = serviceComponentFactory.createNew(s, svcComponent); + s.addServiceComponent(sc); + sc.persist(); + } + + ServiceComponentHost impl = serviceComponentHostFactory.createNew(sc, + hostName); + + impl.persist(); + return impl; + } +} http://git-wip-us.apache.org/repos/asf/ambari/blob/dd572d35/ambari-web/package.json ---------------------------------------------------------------------- diff --git a/ambari-web/package.json b/ambari-web/package.json index be5a225..4468d36 100644 --- a/ambari-web/package.json +++ b/ambari-web/package.json @@ -19,7 +19,7 @@ "devDependencies": { "phantomjs": "^1.9.2", "mocha":"1.9.0", - "mocha-phantomjs": "^3.1.6", + "mocha-phantomjs": "~3.1.6", "chai":"~1.9.0", "sinon":"=1.7.3", "sinon-chai":"~2.5.0",
