Add tasks status definition + use HashedWheelTimer in waitFor #393
Conversation
| eventually { | ||
| val status = client.execute { getStatus task task from "indexToGetTaskStatus" } | ||
| whenReady(status) { result => | ||
| result.status == "published" |
There was a problem hiding this comment.
shouldn't there be a assertion here?
There was a problem hiding this comment.
Same question for the test below
There was a problem hiding this comment.
Ah you are right. Fixed.
|
|
||
| val task = taskShouldBeCreated(create) | ||
|
|
||
| eventually { |
There was a problem hiding this comment.
Why do you need the eventually?
There was a problem hiding this comment.
Same question for the test below
There was a problem hiding this comment.
We need eventually because it takes some time for a task to be actually published: the very frirst call may indicate the task as "pending" and not "published".
So eventually retries (as defined in Eventually) until the block does not throw any exception (in our case a TestFailedException if the status is not "published").
| @@ -64,6 +60,19 @@ trait WaitForTaskDsl { | |||
| implicit object WaitForTaskDefinitionExecutable | |||
| extends Executable[WaitForTaskDefinition, TaskStatus] { | |||
|
|
|||
| // Run every 50 ms, use a wheel with 512 buckets | |||
| private lazy val timer = { | |||
| val threadFactory = new ThreadFactory { | |||
There was a problem hiding this comment.
couldn't we reuse the thread pool from the execution context?
Not a big fan of hiding a theadFactory.
There was a problem hiding this comment.
This ThreadFactory is used to create the worker thread of the wheel, and nothing else. This is the thread that runs the "timed out" timer tasks. The problem of using the given ExecutionContext is that we impose the creation of a long running worker thread in the pool. So if the client passes in a single thread pool, we are left with no thread to execute the map operations over the returned Futures. Also if the created thread is not a daemon thread, it may lead to complications when terminating the application.
In a nutshell, I think it makes more sense for the library to take care of its own worker thread and prevent the user from having to care about it. (That is basically what the current implementation using Timer does!)
There was a problem hiding this comment.
Yes, you are right the current implementation was using an internal pool of threads, but I was not very big fan of it.
At least, could you expose the number of threads of this pool in the configuration? So people could tweak it.
There was a problem hiding this comment.
It will always be one thread. The real configuration is what HashedWheelTimer offers as parameters:
ThreadFactoryto determine how to create this single worker thread- the
tickDuration: how often are timed out tasks being executed by the worker thread - the size of the wheel
I am not sure users care too much about all of them but maybe the tickDuration if they are willing to go for "more often" or "more batching". I guess we can add a tickDuration as a FiniteDuration in the WaitForTaskDefinition. It exposes the internals though.
Something like:
case class WaitForTaskDefinition(taskId: Long,
index: Option[String] = None,
baseDelay: Long = 100,
maxDelay: Long = Long.MaxValue,
tickDuration: Long = 50) // using Long here for consistency with the other fields
...
new HashedWheelTimer(threadFactory, query.tickDuration, TimeUnit.MILLISECONDS, 512)
There was a problem hiding this comment.
Actually the whole point being to reuse this HashedWheelTimer for all submitted queries, there is no way to use the query itself to create it.
Given that it is used for network calls, I think 50 or 100 ms is a good number, especially considering the default base delay is 100ms and then 200, 400 etc.
What do you think?
|
Are more changes required before merging? |
This PR makes two changes:
TaskStatusDefinitionto simply query the status of task without waiting for itWaitForTaskDefinition, replaces the creation ofTimerwith aHashedWheelTimerThe motivation of the first change is to allow the caller to handle the retries (if any) instead of imposing a strategy. It also allows not to create a new
Timer(and therefore a newThread) for each call.The motivation of the second change is to allow multiple
waitForswithout creating new timerThreadsfor each one.At Flow, we ran into issues while checking the status of multiple index tasks with aggressive
baseDelayas it ended up creating several thousands ofThreads. The recursive implementation of theWaitForTaskDsl.requestallows for the creation of severalThreadsperwaitForcall. In our case, it lead tojava.lang.OutOfMemoryError: unable to create new native thread.I understand we do not want to encourage clients to issue thousand of calls at once but again in our case, a few
waitForswith shortbaseDelaylead to the creation of several thousands threads and eventually to a painful death ;-)Since extreme time precision is not a requirement, the proposed solution therefore uses a
HashedWheelTimerwith a tick duration of 50ms. This number is roughly in the order of a network round trip. That would allow to defer the calls to the underlying http client.Please feel free to modify :)