func upstreamProcsForProc(proc WorkflowProcess) map[string]WorkflowProcess {
	procs := map[string]WorkflowProcess{}
	for _, inp := range proc.InPorts() {
		for _, rpt := range inp.RemotePorts {
			procs[rpt.Process().Name()] = rpt.Process()
			mergeWFMaps(procs, upstreamProcsForProc(rpt.Process()))
		}
	}
	for _, pip := range proc.InParamPorts() {
		for _, rpp := range pip.RemotePorts {
			procs[rpp.Process().Name()] = rpp.Process()
			mergeWFMaps(procs, upstreamProcsForProc(rpp.Process()))
		}
	}
	return procs
}
upstreamProcsForProc returns all processes it is connected to, either
directly or indirectly, via its in-ports and param-in-ports
package scipipe
import (
	"fmt"
	"os"
	"regexp"
	"sort"
	"strings"
	"sync"
	"time"
)
type Workflow struct {
	name              string
	procs             map[string]WorkflowProcess
	concurrentTasks   chan struct{}
	concurrentTasksMx sync.Mutex
	sink              *Sink
	driver            WorkflowProcess
	logFile           string
	PlotConf          WorkflowPlotConf
}
type WorkflowPlotConf struct {
	EdgeLabels bool
}
type WorkflowProcess interface {
	Name() string
	InPorts() map[string]*InPort
	OutPorts() map[string]*OutPort
	InParamPorts() map[string]*InParamPort
	OutParamPorts() map[string]*OutParamPort
	Ready() bool
	Run()
	Fail(interface{})
	Failf(string, ...interface{})
}
func NewWorkflow(name string, maxConcurrentTasks int) *Workflow {
	wf := newWorkflowWithoutLogging(name, maxConcurrentTasks)
	allowedCharsPtrn := regexp.MustCompile("[^a-z0-9_]")
	wfNameNormalized := allowedCharsPtrn.ReplaceAllString(strings.ToLower(name), "-")
	wf.logFile = "log/scipipe-" + time.Now().Format("20060102-150405") + "-" + wfNameNormalized + ".log"
	InitLogAuditToFile(wf.logFile)
	return wf
}
func NewWorkflowCustomLogFile(name string, maxConcurrentTasks int, logFile string) *Workflow {
	wf := newWorkflowWithoutLogging(name, maxConcurrentTasks)
	wf.logFile = logFile
	InitLogAuditToFile(logFile)
	return wf
}
func newWorkflowWithoutLogging(name string, maxConcurrentTasks int) *Workflow {
	wf := &Workflow{
		name:            name,
		procs:           map[string]WorkflowProcess{},
		concurrentTasks: make(chan struct{}, maxConcurrentTasks),
		PlotConf:        WorkflowPlotConf{EdgeLabels: true},
	}
	sink := NewSink(wf, name+"_default_sink")
	wf.sink = sink
	wf.driver = sink
	return wf
}
func (wf *Workflow) Name() string {
	return wf.name
}
func (wf *Workflow) NewProc(procName string, commandPattern string) *Process {
	proc := NewProc(wf, procName, commandPattern)
	return proc
}
func (wf *Workflow) Proc(procName string) WorkflowProcess {
	if _, ok := wf.procs[procName]; !ok {
		wf.Failf("No process named (%s)", procName)
	}
	return wf.procs[procName]
}
func (wf *Workflow) ProcsSorted() []WorkflowProcess {
	keys := []string{}
	for k := range wf.Procs() {
		keys = append(keys, k)
	}
	sort.Strings(keys)
	procs := []WorkflowProcess{}
	for _, k := range keys {
		procs = append(procs, wf.Proc(k))
	}
	return procs
}
func (wf *Workflow) Procs() map[string]WorkflowProcess {
	return wf.procs
}
func (wf *Workflow) AddProc(proc WorkflowProcess) {
	if wf.procs[proc.Name()] != nil {
		wf.Failf("A process with name (%s) already exists in the workflow! Use a more unique name!", proc.Name())
	}
	wf.procs[proc.Name()] = proc
}
func (wf *Workflow) AddProcs(procs ...WorkflowProcess) {
	for _, proc := range procs {
		wf.AddProc(proc)
	}
}
func (wf *Workflow) Sink() *Sink {
	return wf.sink
}
func (wf *Workflow) SetSink(sink *Sink) {
	if wf.sink.Ready() {
		wf.Fail("Trying to replace a sink which is already connected. Are you combining SetSink() with ConnectFinalOutPort()? That is not allowed!")
	}
	wf.sink = sink
}
func (wf *Workflow) IncConcurrentTasks(slots int) {
	wf.concurrentTasksMx.Lock()
	for i := 0; i < slots; i++ {
		wf.concurrentTasks <- struct{}{}
		Debug.Println("Increased concurrent tasks")
	}
	wf.concurrentTasksMx.Unlock()
}
func (wf *Workflow) DecConcurrentTasks(slots int) {
	for i := 0; i < slots; i++ {
		<-wf.concurrentTasks
		Debug.Println("Decreased concurrent tasks")
	}
}
func (wf *Workflow) PlotGraph(filePath string) {
	dot := wf.DotGraph()
	createDirs(filePath)
	dotFile, err := os.Create(filePath)
	CheckWithMsg(err, "Could not create dot file "+filePath)
	_, errDot := dotFile.WriteString(dot)
	if errDot != nil {
		wf.Failf("Could not write to DOT-file %s: %s", dotFile.Name(), errDot)
	}
}
func (wf *Workflow) PlotGraphPDF(filePath string) {
	wf.PlotGraph(filePath)
	ExecCmd(fmt.Sprintf("dot -Tpdf %s -o %s.pdf", filePath, filePath))
}
func (wf *Workflow) DotGraph() (dot string) {
	dot = fmt.Sprintf(`digraph "%s" {`+"\n", wf.Name())
	dot += `  rankdir=LR;` + "\n"
	dot += `  graph [fontname="Arial",fontsize=13,color="#384A52",fontcolor="#384A52"];` + "\n"
	dot += `  node  [fontname="Arial",fontsize=11,color="#384A52",fontcolor="#384A52",fillcolor="#EFF2F5",shape=box,style=filled];` + "\n"
	dot += `  edge  [fontname="Arial",fontsize=9, color="#384A52",fontcolor="#384A52"];` + "\n"
	con := ""
	remToDotPtn := regexp.MustCompile(`^.*\.`)
	for _, p := range wf.ProcsSorted() {
		dot += fmt.Sprintf(`  "%s" [shape=box];`+"\n", p.Name())
		for opname, op := range p.OutPorts() {
			for rpname, rp := range op.RemotePorts {
				if wf.PlotConf.EdgeLabels {
					con += fmt.Sprintf(`  "%s" -> "%s" [taillabel="%s", headlabel="%s"];`+"\n", op.Process().Name(), rp.Process().Name(), remToDotPtn.ReplaceAllString(opname, ""), remToDotPtn.ReplaceAllString(rpname, ""))
				} else {
					con += fmt.Sprintf(`  "%s" -> "%s";`+"\n", op.Process().Name(), rp.Process().Name())
				}
			}
		}
		for popname, pop := range p.OutParamPorts() {
			for rpname, rp := range pop.RemotePorts {
				if wf.PlotConf.EdgeLabels {
					con += fmt.Sprintf(`  "%s" -> "%s" [style="dashed", taillabel="%s", headlabel="%s"];`+"\n", pop.Process().Name(), rp.Process().Name(), remToDotPtn.ReplaceAllString(popname, ""), remToDotPtn.ReplaceAllString(rpname, ""))
				} else {
					con += fmt.Sprintf(`  "%s" -> "%s" [style="dashed"];`+"\n", pop.Process().Name(), rp.Process().Name())
				}
			}
		}
	}
	dot += con
	dot += "}\n"
	return
}
func (wf *Workflow) Run() {
	wf.runProcs(wf.procs)
}
func (wf *Workflow) RunTo(finalProcNames ...string) {
	procs := []WorkflowProcess{}
	for _, procName := range finalProcNames {
		procs = append(procs, wf.Proc(procName))
	}
	wf.RunToProcs(procs...)
}
func (wf *Workflow) RunToRegex(procNamePatterns ...string) {
	procsToRun := []WorkflowProcess{}
	for _, pattern := range procNamePatterns {
		regexpPtrn := regexp.MustCompile(pattern)
		for procName, proc := range wf.Procs() {
			matches := regexpPtrn.MatchString(procName)
			if matches {
				procsToRun = append(procsToRun, proc)
			}
		}
	}
	wf.RunToProcs(procsToRun...)
}
func (wf *Workflow) RunToProcs(finalProcs ...WorkflowProcess) {
	procsToRun := map[string]WorkflowProcess{}
	for _, finalProc := range finalProcs {
		procsToRun = mergeWFMaps(procsToRun, upstreamProcsForProc(finalProc))
		procsToRun[finalProc.Name()] = finalProc
	}
	wf.runProcs(procsToRun)
}
func (wf *Workflow) runProcs(procs map[string]WorkflowProcess) {
	wf.reconnectDeadEndConnections(procs)
	if !wf.readyToRun(procs) {
		wf.Fail("Workflow not ready to run, due to previously reported errors, so exiting.")
	}
	for _, proc := range procs {
		Debug.Printf(wf.name+": Starting process (%s) in new go-routine", proc.Name())
		go proc.Run()
	}
	Debug.Printf("%s: Starting driver process (%s) in main go-routine", wf.name, wf.driver.Name())
	wf.Auditf("Starting workflow (Writing log to %s)", wf.logFile)
	wf.driver.Run()
	wf.Auditf("Finished workflow (Log written to %s)", wf.logFile)
}
func (wf *Workflow) readyToRun(procs map[string]WorkflowProcess) bool {
	if len(procs) == 0 {
		Error.Println(wf.name + ": The workflow is empty. Did you forget to add the processes to it?")
		return false
	}
	if wf.sink == nil {
		Error.Println(wf.name + ": sink is nil!")
		return false
	}
	for _, proc := range procs {
		if !proc.Ready() {
			Error.Println(wf.name + ": Not everything connected. Workflow shutting down.")
			return false
		}
	}
	return true
}
func (wf *Workflow) reconnectDeadEndConnections(procs map[string]WorkflowProcess) {
	foundNewDriverProc := false
	for _, proc := range procs {
		for _, opt := range proc.OutPorts() {
			for iptName, ipt := range opt.RemotePorts {
				if ipt.Process() == nil {
					Debug.Printf("Disconnecting in-port (%s) from out-port (%s)", ipt.Name(), opt.Name())
					opt.Disconnect(iptName)
				} else if _, ok := procs[ipt.Process().Name()]; !ok {
					Debug.Printf("Disconnecting in-port (%s) from out-port (%s)", ipt.Name(), opt.Name())
					opt.Disconnect(iptName)
				}
			}
			if !opt.Ready() {
				Debug.Printf("Connecting disconnected out-port (%s) of process (%s) to workflow sink", opt.Name(), opt.Process().Name())
				wf.sink.From(opt)
			}
		}
		for _, pop := range proc.OutParamPorts() {
			for rppName, rpp := range pop.RemotePorts {
				if rpp.Process() == nil {
					Debug.Printf("Disconnecting in-port (%s) from out-port (%s)", rpp.Name(), pop.Name())
					pop.Disconnect(rppName)
				} else if _, ok := procs[rpp.Process().Name()]; !ok {
					Debug.Printf("Disconnecting in-port (%s) from out-port (%s)", rpp.Name(), pop.Name())
					pop.Disconnect(rppName)
				}
			}
			if !pop.Ready() {
				Debug.Printf("Connecting disconnected out-port (%s) of process (%s) to workflow sink", pop.Name(), pop.Process().Name())
				wf.sink.FromParam(pop)
			}
		}
		if len(proc.OutPorts()) == 0 && len(proc.OutParamPorts()) == 0 {
			if foundNewDriverProc {
				wf.Failf("Found more than one process without out-ports nor out-param ports. Cannot use both as drivers (One of them being '%s'). Adapt your workflow accordingly.", proc.Name())
			}
			foundNewDriverProc = true
			wf.driver = proc
		}
	}
	if foundNewDriverProc && len(procs) > 1 {  
		delete(wf.procs, wf.driver.Name())
	}
}