Yes, that should be fine. But what do you mean by re-entrant in this
context? All accesses to reader should be single-threaded.
On 9/20/19 6:11 PM, Ken Barr wrote:
Is the IO SDK re-entrant? Is it safe to call advance() from within start()?
On 2019/09/19 14:57:09, Jan Lukavský <je...@seznam.cz> wrote:
Hi Ken,
I have seen some deadlock behavior with custom sources earlier
(different runner, but that might not be important). Some lessons learned:
a) please make sure your advance() or start() methods do not block,
that will cause issues and possibly deadlocks you describe
b) if you want to limit parallelism, that should be possible in the
split() method - you can return collection containing only (this) if
there is no more readers
Hope this helps, please feel free to ask more details if needed.
Best,
Jan
On 9/19/19 4:47 PM, Ken Barr wrote:
I have a custom UnboundedSource IO that reads from a series of messaging
queues. I have implemented this such that the IO takes a list of queues and
expands a UnboundedSource/UnboundedReader for each queue.
If I use autoscaling with maxNumWorkers <= # number of queues everything works
well. For example if I have 4 queues and run in dataflow; the Dataflow process
starts with 1 worker with 4 Readers each consuming from a queue. As CPU usage
and backlog grow, Dataflow spawns more workers and moves the Readers to the new
workers. As CPU usage and backlog shrinks, the Readers are moved back to the
original worker and unused workers are deleted. This is exactly what I was hoping
for.
Problems happen if I set maxNumWorkers greater then number of queues. As
scaleup goes past the number of queues, not only are Readers moved, but for
some reason new Readers are created. This should not be too bad, new Readers
would just not receive messages as the original Reader is holding exclusive
access to ensure in-order delivery. The real problem is that the original
Readers are holding the queue and their advance() method is not being called.
The new Readers advance() method is being called, but they are not active on
the queue, hence the system is now deadlocked.
Questions are:
Why are new Readers being spawned if maxNumWorkers exceeds original number of
Readers? Is there a way of preventing this as I would like to maintain
in-order delivery?
Why is the original Readers advance() method no longer being called? This is
causing a deadlock.
