#151 Specify number of threads to consume by a task.

[jirifilip]

Add an option to specify how many "threads" should a certain task consume with regards to the total number of threads set by pramen.parallel.tasks.

An example:

pramen {
  pipeline.name = "A testing pipeline"

  parallel.tasks = 4
}

pramen.operations = [
  {
    name = "easy job"
    ...
   # default is 1
    consume.threads = 1 # not a resource intensive job, so Pramen can run 4 of these at one time (if nothing else is running of course)
  },
  {
    name = "hard job"
    ...
    consume.threads = 4 # run only one instance of this operation at a time
  },
]

These are not real threads, it is more akin to an indication to pramen saying: "This is a resource-intenstive operation, so you don't want to have many of these running at once".

I tried thinking of a better name to not be as confusing but did not come up with anything useful. I was thinking about pramen.parallelism.available.slots and uses.slots but these do not sound very good 😄

Some notes

• I tested this end-to-end. Was also thinking about how to unit test this better but did not come up with anything other than is in the PR.
• the Try in the whenEnoughResourcesAreAvailable is there to ensure than the permits get released but I call .get() after to let the exception "bubble up". I was going through the code and it seems that the task run failure is already handled by the runTask() function and the Future failure is also handled. But still not sure if I had done this correctly.
• It still spawns all of the threads. The ones that don't have enough "slots" are blocking on the semaphore until there are enough "threads" so that they can run.

[yruslan]

Looks great! Just a couple of tiny suggestions.

Also, the description to the PR is so good that you can add a small section to README explaining how to use 'consume.threads'.

Regarding tests - it is always not easy to test concurrency. You can try (maybe in one of next PRs) to test methods separately (whenEnoughResourcesAreAvailable, getTruncatedResourceCount, etc).

This is just an idea. To test whenEnoughResourcesAreAvailable() you can create a fixed thread pool in the test suite itself, and a list of tasks with known execution times. And then you can assert that the wall time is as expected, and the order of execution is as expected.

[yruslan]

<= 0 ?

[jirifilip]

ah, crap, thank you for spotting this! 😄

[yruslan]

😆

[yruslan]

I'm thinking making the semaphore 'fair'. Otherwise, it could happen that easy jobs can block hard jobs for some time.

Suggested change

	private val availableResources: Semaphore = new Semaphore(maxResources)
	private val availableResources: Semaphore = new Semaphore(maxResources, true)

[yruslan]

I like the idea of wrapping the dangerous code with Try, and using Try.get at the end!

[jirifilip]

Thank you for the feedback, it was fun working on this! I added some description to the README on how to use the parallelism options :) Regarding the tests, it's a very good idea but I would do it in the next PR if that's ok (will have to transfer the tests to a different package and this PR could get too big).