[jira] [Comment Edited] (TEZ-1273) Refactor DAGAppMaster to state machine based

["Jeff Zhang (JIRA)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/TEZ-1273?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14591400#comment-14591400 ] 

Jeff Zhang edited comment on TEZ-1273 at 6/18/15 7:44 AM:
----------------------------------------------------------

bq. writeLock is used in the statemachine handle() call but readLock is not used anywhere. Doesn't all read access to data that is updated within the state transitions need to be protected by a read lock?
I think currently DAGAppMaster expose itself through RunningAppContext to be used by other components. Ideally RunningAppContext should be protected by read lock, but it would easily bring dead lock issue. One example is that DAGAppMaster#getAMState, I use a volatile field state in DAGAppMaster but not get it from the state machine object every time to avoid deadlock. Given that most of fields accessed in RunningAppContext won't change after AM is started properly, it should be safe without read lock. But in the future, we do need to consider other better ways to access RunningAppContext.

bq. would it make the code less complex if we had diff transitions for dag_succeded, dag_failed, dag_internal_error, dag_killed?
I'm afraid not. They share lots of common code in DAGAppMaster#dagComplete, only different on the following part. And this lead less code change, easy for review. 
{code}
 switch(dagFinishedEvent.getDAGState()) {
      case SUCCEEDED:
        if (!currentDAG.getName().startsWith(
            TezConstants.TEZ_PREWARM_DAG_NAME_PREFIX)) {
          successfulDAGs.incrementAndGet();
        }
        break;
      case FAILED:
        if (!currentDAG.getName().startsWith(
            TezConstants.TEZ_PREWARM_DAG_NAME_PREFIX)) {
          failedDAGs.incrementAndGet();
        }
        
        break;
      case KILLED:
        if (!currentDAG.getName().startsWith(
            TezConstants.TEZ_PREWARM_DAG_NAME_PREFIX)) {
          killedDAGs.incrementAndGet();
        }
        break;
      case ERROR:
        if (!currentDAG.getName().startsWith(
            TezConstants.TEZ_PREWARM_DAG_NAME_PREFIX)) {
          failedDAGs.incrementAndGet();
        }
{code}

bq. this transition seems a bit confusing. We should make it more clear as to what it is trying to do. Also the comment on submit dag in non-session mode does not seem right.
bq. Related to the above, we have the rpc/api layer and the state machine layer. Some of the calls need to be synchronous and others are handed off the state machine. This leads a possibility of race conditions such as 2 dags being submitted at the same time while the AM is in an idle state. Sending an event to the state machine here would likely not work. We may need to explicitly state to RUNNING from IDLE to prevent such races. Any thoughts on handling this?
Introduce a new event type DAG_SUBMITTED to represent that a new DAG is submitted through rpc and rename NEW_DAG_SUBMITTED to DAG_PREPARE to notify components that new dag is submitted. 

bq. Why is an internal error ignorable? Can this cause the AM to remain infinitely in a terminating state?
Fixed.

bq. catch alls hide bugs. We should remove the above.
Fixed.

bq. does work within serviceInit/serviceStart/serviceStop need to be within a write lock?
Previously they are in synchronized block. I think this is to avoid that case that DAGAppMasterShutdownHook is called when it is in initialing or starting. This could prevent these 3 methods invoked at the the same time in different thread. 
Also prevent shutdownTezAM & submitDAGToAppMaster invoked at the same time (they are also both in synchronized block) write lock here is just to replace the synchronized keyword. 

bq. can serviceStop be triggered by something outside of the state machine?
Yes, it is possible when DAGAppMasterShutdownHook is invoked when AM receive terminate signal. 

bq. precondition checks are not useful for a state machine. This will throw an exception from the handle function and lead to an internal error and the AM shutting down.
Fixed.

bq. For ShutdownWhenRunningTransition, shouldn't the sessionStopped flag be set? What will happen after the dag completion ( after kill ) comes back? In this case, won't the AM go back to idle state for a session?

bq. confusing comment. Not matching code.
Fixed.

bq. Also, if the version mismatch error occurs and causes AM to go into ERROR state - how does the AM unregister from RM and update diagnostics?
diagnostics has already been updated when version compares.  And it will unregister from RM
{code}
     versionMismatchDiagnostics = "Incompatible versions found"
          + ", clientVersion=" + clientVersion
          + ", AMVersion=" + dagVersionInfo.getVersion();
      addDiagnostic(versionMismatchDiagnostics);
{code}

{code}
if (versionMismatch) {
      // Short-circuit and return as no DAG should not be run
      LOG.info("Version Mismatch, shutting down AM");
      this.taskSchedulerEventHandler.setShouldUnregisterFlag();
      shutdownHandler.shutdown();
      return DAGAppMasterState.ERROR;
    }
{code}

{code}
sendEvent(new DAGAppMasterEvent(DAGAppMasterEventType.SESSION_TIMEOUT));
{code}
bq. does this introduce a race with submitDag() ?
Fix it by handle session time event synchronously. 


bq. DrainDispatcher.java in tez-dag/src/test/ - any reason why we need this?
It is just for unit test. Used in TestMockDAGAppMaster

bq. LocalClient changes?
Minor change on LocalClient because we introduce a new DAGAppMasterState RECOVERING.

bq. TestMemoryWithEvents
Due to change on MockTezClient to simulate recovering. 


was (Author: zjffdu):
bq. writeLock is used in the statemachine handle() call but readLock is not used anywhere. Doesn't all read access to data that is updated within the state transitions need to be protected by a read lock?
I think currently DAGAppMaster expose itself through RunningAppContext to be used by other components. Ideally RunningAppContext should be protected by read lock, but it would easily bring dead lock issue. One example is that DAGAppMaster#getAMState, I use a volatile field state in DAGAppMaster but not get it from the state machine object every time to avoid deadlock. Given that most of fields accessed in RunningAppContext won't change after AM is started properly, it should be safe without read lock. But in the future, we do need to consider other better ways to access RunningAppContext.

bq. would it make the code less complex if we had diff transitions for dag_succeded, dag_failed, dag_internal_error, dag_killed?
I'm afraid not. They share lots of common code in DAGAppMaster#dagComplete, only different on the following part. And this lead less code change, easy for review. 
{code}
 switch(dagFinishedEvent.getDAGState()) {
      case SUCCEEDED:
        if (!currentDAG.getName().startsWith(
            TezConstants.TEZ_PREWARM_DAG_NAME_PREFIX)) {
          successfulDAGs.incrementAndGet();
        }
        break;
      case FAILED:
        if (!currentDAG.getName().startsWith(
            TezConstants.TEZ_PREWARM_DAG_NAME_PREFIX)) {
          failedDAGs.incrementAndGet();
        }
        
        break;
      case KILLED:
        if (!currentDAG.getName().startsWith(
            TezConstants.TEZ_PREWARM_DAG_NAME_PREFIX)) {
          killedDAGs.incrementAndGet();
        }
        break;
      case ERROR:
        if (!currentDAG.getName().startsWith(
            TezConstants.TEZ_PREWARM_DAG_NAME_PREFIX)) {
          failedDAGs.incrementAndGet();
        }
{code}

bq. this transition seems a bit confusing. We should make it more clear as to what it is trying to do. Also the comment on submit dag in non-session mode does not seem right.
bq. Related to the above, we have the rpc/api layer and the state machine layer. Some of the calls need to be synchronous and others are handed off the state machine. This leads a possibility of race conditions such as 2 dags being submitted at the same time while the AM is in an idle state. Sending an event to the state machine here would likely not work. We may need to explicitly state to RUNNING from IDLE to prevent such races. Any thoughts on handling this?
Introduce a new event type DAG_SUBMITTED to represent that a new DAG is submitted through rpc and rename NEW_DAG_SUBMITTED to DAG_PREPARE to notify components that new dag is submitted. 

bq. Why is an internal error ignorable? Can this cause the AM to remain infinitely in a terminating state?
Fixed.

bq. catch alls hide bugs. We should remove the above.
Fixed.

bq. does work within serviceInit/serviceStart/serviceStop need to be within a write lock?
originally they are in synchronized block. I think this is to avoid that multiple thread access serviceStop which may cause conflicts. Also prevent shutdownTezAM & submitDAGToAppMaster invoked at the same time (they are both in synchronized block) write lock here is just to replace the synchronized keyword. 

bq. can serviceStop be triggered by something outside of the state machine?
Yes, it is possible when DAGAppMasterShutdownHook is invoked when AM receive terminate signal. 

bq. precondition checks are not useful for a state machine. This will throw an exception from the handle function and lead to an internal error and the AM shutting down.
Fixed.

bq. For ShutdownWhenRunningTransition, shouldn't the sessionStopped flag be set? What will happen after the dag completion ( after kill ) comes back? In this case, won't the AM go back to idle state for a session?

bq. confusing comment. Not matching code.
Fixed.

bq. Also, if the version mismatch error occurs and causes AM to go into ERROR state - how does the AM unregister from RM and update diagnostics?
diagnostics has already been updated when version compares.  And it will unregister from RM
{code}
     versionMismatchDiagnostics = "Incompatible versions found"
          + ", clientVersion=" + clientVersion
          + ", AMVersion=" + dagVersionInfo.getVersion();
      addDiagnostic(versionMismatchDiagnostics);
{code}

{code}
if (versionMismatch) {
      // Short-circuit and return as no DAG should not be run
      LOG.info("Version Mismatch, shutting down AM");
      this.taskSchedulerEventHandler.setShouldUnregisterFlag();
      shutdownHandler.shutdown();
      return DAGAppMasterState.ERROR;
    }
{code}

{code}
sendEvent(new DAGAppMasterEvent(DAGAppMasterEventType.SESSION_TIMEOUT));
{code}
bq. does this introduce a race with submitDag() ?
Fix it by handle session time event synchronously. 


bq. DrainDispatcher.java in tez-dag/src/test/ - any reason why we need this?
It is just for unit test. Used in TestMockDAGAppMaster

bq. LocalClient changes?
Minor change on LocalClient because we introduce a new DAGAppMasterState RECOVERING.

bq. TestMemoryWithEvents
Due to change on MockTezClient to simulate recovering. 

> Refactor DAGAppMaster to state machine based
> --------------------------------------------
>
>                 Key: TEZ-1273
>                 URL: https://issues.apache.org/jira/browse/TEZ-1273
>             Project: Apache Tez
>          Issue Type: Improvement
>            Reporter: Jeff Zhang
>            Assignee: Jeff Zhang
>         Attachments: DAGAppMaster_3.pdf, DAGAppMaster_4.pdf, DAGAppMaster_5.pdf, TEZ-1273-3.patch, TEZ-1273-4.patch, TEZ-1273-5.patch, TEZ-1273-6.patch, TEZ-1273-7.patch, TEZ-1273-8.patch, TEZ-1273-9.patch, Tez-1273-2.patch, Tez-1273.patch, dag_app_master.pdf, dag_app_master2.pdf
>
>
> Almost all our entities (Vertex, Task etc) are state machine based and written using a formal state machine. But DAGAppMaster is not written on a formal state machine even though it has a state machine based behavior. This jira is for refactoring it into state machine based



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)