declarations that refer to task outputs

stekaz · April 2016

Hi,

I'm having some trouble with this example code:

task test { 
  String var 
  command { 
    ./script ${var} 
  } 
  output { 
    String value = read_string(stdout()) 
  } 
} 

workflow wf { 
  call test as x {input: var="x"} 
  call test as y {input: var="y"} 

  Array[String] strs = [x.value, y.value] 
}

The errors I get (using 0.19) look like:

[error] Workflow input processing failed. 
Workflow has invalid declarations: Could not find a value for x 
Could not find a value for y 
cromwell.engine.workflow.MaterializeWorkflowDescriptorActor$$anonfun$receive$1$$anon$1: Workflow input processing failed. 
Workflow has invalid declarations: Could not find a value for x 
Could not find a value for y 
        at cromwell.engine.workflow.MaterializeWorkflowDescriptorActor$$anonfun$receive$1.applyOrElse(MaterializeWorkflowDescriptorActor.scala:69) 
        at akka.actor.Actor$class.aroundReceive(Actor.scala:467) 
        at cromwell.engine.workflow.MaterializeWorkflowDescriptorActor.aroundReceive(MaterializeWorkflowDescriptorActor.scala:59) 
        at akka.actor.ActorCell.receiveMessage(ActorCell.scala:516) 
        at akka.actor.ActorCell.invoke(ActorCell.scala:487) 
        at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:238) 
        at akka.dispatch.Mailbox.run(Mailbox.scala:220) 
        at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:397) 
        at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260) 
        at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339) 
        at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979) 
        at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107) 
Workflow input processing failed. 
Workflow has invalid declarations: Could not find a value for x 
Could not find a value for y

Is this the expected output? I also get an error if I simplify the workflow to:

workflow wf { ````

  call test {input: var="x"} 

  String str = test.value 
}

Why can't we declare the outputs of tasks? Have I missed something?

KateN · April 2016

A workflow block makes no assumptions that previous tasks have run before attempting to run another line, unless the dependency is explicitly declared. Even though you've called test prior to trying to use it's output, test.value, Cromwell sees that there are no dependencies to the line String str = test.value so it tries, and fails, to execute it.

To ensure that it will properly see the dependency, you must use a task to wrap the work in, like so:

task gatherTest{
  String val1
  String val2
  command {  }
  output {
    Array[String] strs = [val1,val2]
  }
}

Then your workflow should look like this:

workflow wf { 
  call test as x {input: var="x"} 
  call test as y {input: var="y"} 
  call gatherTest {input: val1=x.value, val2=y.value}
}

This method will work on the small scale, but if you need to run multiple test tasks on different inputs, you might consider using a scatter-gather plumbing method. In your case, it might look like this:

task test { 
  String var 
  command { 
    ./script ${var} 
  } 
  output { 
    String value = read_string(stdout()) 
  } 
} 

workflow wf { 
  Array[String] inputStrings = ["x","y"]

  scatter(input in inputStrings) {
    call test{input: var=input}
  }
}

The output of this workflow would be an Array[String] containing the value outputs from each run of the task test.

KateN · April 2016

A workflow block makes no assumptions that previous tasks have run before attempting to run another line, unless the dependency is explicitly declared. Even though you've called test prior to trying to use it's output, test.value, Cromwell sees that there are no dependencies to the line String str = test.value so it tries, and fails, to execute it.

To ensure that it will properly see the dependency, you must use a task to wrap the work in, like so:

task gatherTest{
  String val1
  String val2
  command {  }
  output {
    Array[String] strs = [val1,val2]
  }
}

Then your workflow should look like this:

workflow wf { 
  call test as x {input: var="x"} 
  call test as y {input: var="y"} 
  call gatherTest {input: val1=x.value, val2=y.value}
}

This method will work on the small scale, but if you need to run multiple test tasks on different inputs, you might consider using a scatter-gather plumbing method. In your case, it might look like this:

task test { 
  String var 
  command { 
    ./script ${var} 
  } 
  output { 
    String value = read_string(stdout()) 
  } 
} 

workflow wf { 
  Array[String] inputStrings = ["x","y"]

  scatter(input in inputStrings) {
    call test{input: var=input}
  }
}

The output of this workflow would be an Array[String] containing the value outputs from each run of the task test.

stekaz · April 2016

Thanks for clearing up my confusion, Kate. Much appreciated!

declarations that refer to task outputs

Best Answer

Answers