edu.iu.dsc.tws.rsched.bootstrap

Class ZKWorkerController

java.lang.Object

edu.iu.dsc.tws.rsched.bootstrap.ZKWorkerController

All Implemented Interfaces:

IWorkerController

public class ZKWorkerController
extends java.lang.Object
implements IWorkerController

gets unique workerID's for each client by using DistributedAtomicInteger gets the current list of all workers from PathChildrenCache it does not keep the list of worker nodes, it is already kept in PathChildrenCache If a worker joins after failure, it gets its previous ID There is no gap in id sequence. All worker names and IDs are kept in the job node as binary encoded WorkerInfo Each worker appends its WorkerInfo Before each WorkerInfo byte array, we put the length of the byte array as 4 bytes So WorkerInfo objects are decoded by using their lengths They append to this list in synchronized fashion by acquiring a distributed lock: InterProcessMutex we use a barrier to make all workers wait until the last worker arrives at the barrier point we count the number of waiting workers by using a DistributedAtomicInteger

Field Summary

Fields

Modifier and Type	Field and Description
static java.util.logging.Logger	LOG

Constructor Summary

Constructors

Constructor and Description
ZKWorkerController(Config config, java.lang.String jobName, java.lang.String workerIpAndPort, int numberOfWorkers, NodeInfo nodeInfo, JobAPI.ComputeResource computeResource)

Method Summary

Modifier and Type	Method and Description
void	close() close the children cache close persistent node for this worker close the connection if this is the last worker to complete, delete all relevant znode for this job
java.util.List<WorkerInfo>	getAllWorkers() wait to make sure that the number of workers reached the total number of workers in the job return all joined workers in the job including the ones that have already left return null if the timeLimit is reached or en exception is thrown while waiting
java.util.List<WorkerInfo>	getCurrentWorkers() Get current list of workers from local children cache This list does not have the workers that have already left
java.util.List<WorkerInfo>	getJoinedWorkers() Get all joined workers including the ones that have already completed and left
int	getNumberOfCurrentWorkers() get number of current workers in the job as seen from this worker
int	getNumberOfWorkers() return the number of all workers in this job, including non-started ones and finished ones
WorkerInfo	getWorkerInfo() return the WorkerInfo object for this worker
WorkerInfo	getWorkerInfoForID(int id) return the WorkerInfo object for the given ID
boolean	initialize() connect to the server get a workerID for this worker append this worker info to the body of job znode create an ephemeral znode for this client
boolean	initialize(int workerID) connect to the server get a workerID for this worker append this worker info to the body of job znode create an ephemeral znode for this client
void	printWorkers(java.util.List<WorkerInfo> workers) Print all given workers
void	waitOnBarrier() we use a DistributedAtomicInteger to count the number of workers that have reached to the barrier point Last worker to call this method and to increase the DistributedAtomicInteger, removes the barrier and lets all previous waiting workers be released other workers to call this method and to increase the DistributedAtomicInteger, enables the barrier by calling setBarrier method and wait it is enough to call setBarrier method by only the first worker, however, it does not harm calling by many workers if we let only the first worker to set the barrier with setBarrier method, then, the second worker may call this method after the dai is increased but before the setBarrier method is called.

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Methods inherited from interface edu.iu.dsc.tws.api.resource.IWorkerController

getCheckpointingClient, getFailureListener, getRuntimeObject

Field Detail

LOG

public static final java.util.logging.Logger LOG

Constructor Detail

ZKWorkerController

public ZKWorkerController(Config config,
                          java.lang.String jobName,
                          java.lang.String workerIpAndPort,
                          int numberOfWorkers,
                          NodeInfo nodeInfo,
                          JobAPI.ComputeResource computeResource)

Method Detail

initialize

public boolean initialize()

connect to the server get a workerID for this worker append this worker info to the body of job znode create an ephemeral znode for this client

Returns:

initialize

public boolean initialize(int workerID)

connect to the server get a workerID for this worker append this worker info to the body of job znode create an ephemeral znode for this client

Returns:

getWorkerInfo

public WorkerInfo getWorkerInfo()

Description copied from interface: IWorkerController

return the WorkerInfo object for this worker

Specified by:

getWorkerInfo in interface IWorkerController

getWorkerInfoForID

public WorkerInfo getWorkerInfoForID(int id)

Description copied from interface: IWorkerController

return the WorkerInfo object for the given ID

Specified by:

getWorkerInfoForID in interface IWorkerController

getNumberOfWorkers

public int getNumberOfWorkers()

Description copied from interface: IWorkerController

return the number of all workers in this job, including non-started ones and finished ones

Specified by:

getNumberOfWorkers in interface IWorkerController

printWorkers

public void printWorkers(java.util.List<WorkerInfo> workers)

Print all given workers

getCurrentWorkers

public java.util.List<WorkerInfo> getCurrentWorkers()

Get current list of workers from local children cache This list does not have the workers that have already left

getNumberOfCurrentWorkers

public int getNumberOfCurrentWorkers()

get number of current workers in the job as seen from this worker

getJoinedWorkers

public java.util.List<WorkerInfo> getJoinedWorkers()

Get all joined workers including the ones that have already completed and left

Specified by:

getJoinedWorkers in interface IWorkerController

getAllWorkers

public java.util.List<WorkerInfo> getAllWorkers()
                                         throws TimeoutException

wait to make sure that the number of workers reached the total number of workers in the job return all joined workers in the job including the ones that have already left return null if the timeLimit is reached or en exception is thrown while waiting

Specified by:

getAllWorkers in interface IWorkerController

Throws:

TimeoutException

waitOnBarrier

public void waitOnBarrier()
                   throws TimeoutException

we use a DistributedAtomicInteger to count the number of workers that have reached to the barrier point Last worker to call this method and to increase the DistributedAtomicInteger, removes the barrier and lets all previous waiting workers be released other workers to call this method and to increase the DistributedAtomicInteger, enables the barrier by calling setBarrier method and wait it is enough to call setBarrier method by only the first worker, however, it does not harm calling by many workers if we let only the first worker to set the barrier with setBarrier method, then, the second worker may call this method after the dai is increased but before the setBarrier method is called. To prevent this, we may need to use a distributed InterProcessMutex. So, instead of using a distributed InterProcessMutex, we call this method many times DistributedAtomicInteger always increases. We check whether it is a multiple of numberOfWorkers in a job If so, all workers have reached the barrier this method may be called many times during a computation return true if all workers have reached the barrier and they are all released if timeout is reached, return false

Specified by:

waitOnBarrier in interface IWorkerController

Throws:

TimeoutException

close

public void close()

close the children cache close persistent node for this worker close the connection if this is the last worker to complete, delete all relevant znode for this job