task_executor.groovy

abstract class TaskProvider {
    def abstract nextTask()
    def abstract runTask(task)

    def beforeExecution(int parallelismFactor) {}
    def afterExecution() {}

    // callbacks can be used to update statistics
    // and change behavior of nextTask() dynamically
    def onStart(task) {}
    def onFinish(task, result) {}
    def onFail(task, exception) {}
}

class TaskExecutor {
    def static execute(int parallelismFactor, final TaskProvider provider) {
        ExecutorService executor = Executors.newFixedThreadPool(parallelismFactor)
        try {
            provider.beforeExecution(parallelismFactor)
            def nextTask
            def handlers = []
            while(nextTask = provider.nextTask()) {
                final task = nextTask
                handlers.add(executor.submit({
                    provider.onStart(task)
                    try {
                        provider.onFinish(task, provider.runTask(task))
                    } catch (Exception e) {
                        provider.onFail(task, e)
                    }
                }))
            }
            handlers.each { it.get() }
            provider.afterExecution()
        } finally{
            executor.shutdown()
        }
    }
}

TaskExecutor.execute(2, new TaskProvider() {
    def freeHosts = new LinkedBlockingQueue(["ip1", "ip2"])
    def testSets = [["test11", "test12", "test13"], ["test21", "test22", "test23"], ["test31", "test32", "test33"]]
    def testPointer = 0
    def taskId = 0
    @Override
    def nextTask() {
        def host = freeHosts.take()
        synchronized(this) {
            if(testPointer < testSets.size) {
                [id : taskId++,
                 host : host,
                 testSet : testSets[testPointer++] ]
            } else {
                null
            }
        }
    }

    @Override
    def runTask(task) {
        println "RUN TASK <${task.id}> : ssh oberon@${task.host} 'mvn test -Dtest=${task.testSet.join(',')}'"
        "OK"
    }

    def onFinish(task, result) {freeHosts.put(task.host)}

    def onFail(task, exception) {freeHosts.put(task.host)}
})

// ---------------------------------------------------------------------------------------------------------------------
// EXECUTION RESULT:
// RUN TASK <1> : ssh oberon@ip2 'mvn test -Dtest=test21,test22,test23'
// RUN TASK <0> : ssh oberon@ip1 'mvn test -Dtest=test11,test12,test13'
// RUN TASK <2> : ssh oberon@ip1 'mvn test -Dtest=test31,test32,test33'