Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

IDS-9547: Max Processes Per Worker #126

Merged
merged 19 commits into from
Feb 9, 2023
Merged

IDS-9547: Max Processes Per Worker #126

Show file tree
Hide file tree
Changes from 14 commits
Commits
Show all changes
19 commits
Select commit Hold shift + click to select a range
ed5e86f
add max process per worker config
voxparcxls Oct 24, 2022
535ee22
modify cws_worker in core db
voxparcxls Nov 15, 2022
4e3c980
modify core sql
voxparcxls Nov 22, 2022
da654ad
rename new proc variable worker_max_num_running_procs
voxparcxls Dec 7, 2022
8f01bc3
MaxNumForProcsOnWorker variable for current Running processes
voxparcxls Dec 14, 2022
48b337c
remove added totalNumOfProcDefKeysOnWorker var
voxparcxls Dec 14, 2022
05f32cf
clean up workerservice commit
voxparcxls Dec 14, 2022
1a81a85
set cap on queryLimit for claiming a process start
voxparcxls Dec 14, 2022
05ae700
rename variables
voxparcxls Dec 19, 2022
024860b
add upgrade script for DB
voxparcxls Jan 10, 2023
8978bed
working but needs clean and optimization
voxparcxls Jan 26, 2023
5a22eed
clean up logs
voxparcxls Feb 2, 2023
8ca402b
clean up CWSInstaller
voxparcxls Feb 2, 2023
fa61fad
Merge branch 'main' into IDS-9547-worker-process-total-cap
voxparcxls Feb 2, 2023
a394301
worker_max_num_running_procs changed to default 16
voxparcxls Feb 2, 2023
06122f3
variable description update
voxparcxls Feb 2, 2023
389561a
removed limitProcDefKey condition in claimWithCounter
voxparcxls Feb 7, 2023
cc395ba
set limit param to null. All enabled procs have opporunity to be clai…
voxparcxls Feb 8, 2023
0f4d595
claim query
voxparcxls Feb 8, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
121 changes: 105 additions & 16 deletions cws-core/src/main/java/jpl/cws/core/db/SchedulerDbService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -48,17 +48,19 @@ public class SchedulerDbService extends DbService implements InitializingBean {
public static final int DEFAULT_WORKER_PROC_DEF_MAX_INSTANCES = 1;
public static final int PROCESSES_PAGE_SIZE = 100;

// KEY FOR THIS IS: KEY `claimKey` (`status`,`proc_def_key`,`priority`,`created_time`)

public static final String FIND_CLAIMABLE_ROWS_SQL =
"SELECT uuid FROM cws_sched_worker_proc_inst " +
"WHERE " +
" status='"+PENDING+"' AND " +
" proc_def_key=? " +
" proc_def_key=?" +
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this line will cause an issue. There needs to be a space

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed: 0f4d595

"ORDER BY " +
" priority ASC, " + // lower priorities favored
" created_time ASC " + // older dates (FIFO) favored
"LIMIT ?";



public static final String UPDATE_CLAIMABLE_ROW_SQL =
"UPDATE cws_sched_worker_proc_inst " +
"SET " +
Expand Down Expand Up @@ -243,17 +245,25 @@ public int updateProcInstIdAndStartedByWorker(
return numUpdated;
}





/**
* Attempt to claim a process start request in the database.
*
* @param procDefKey -- only attempt to claim rows for this process definition
* @param workerProcsList -- attempts to claim rows for the active set of process definition(s)
* @return mappings of claimUuids and claimedRowUuids
*
*/
public Map<String,List<String>> claimHighestPriorityStartReq(String workerId, String procDefKey, int limit) {

public Map<String,List<String>> claimHighestPriorityStartReq(String workerId, Map<String,Integer> workerProcsList, Map<String,Integer> limitsPerProcs, int limit) {
List<String> claimUuids = new ArrayList<String>();
List<String> rowUuids = new ArrayList<String>();
List<String> rowUuidsPerProcDefKey = new ArrayList<String>();
List<String> claimedUuidsCountsPerProcs = new ArrayList<String>();
LinkedHashMap<String, String> uuidAndProcDefKeyPair = new LinkedHashMap<String, String>();
List<String> clearOutUnclaimedInst = new ArrayList<String>();
List<String> unfilteredRowUuids = new ArrayList<String>();
List<String> claimedRowUuids = new ArrayList<String>();
long t0 = System.currentTimeMillis();
int numClaimed = 0;
Expand All @@ -266,11 +276,59 @@ public Map<String,List<String>> claimHighestPriorityStartReq(String workerId, St
try {
// Find claimable rows
//
rowUuids = jdbcTemplate.queryForList(
FIND_CLAIMABLE_ROWS_SQL, String.class,
new Object[] {procDefKey,
limit*2}); // over-find because some workers might compete with this set
int sumOfProcsLimitSet = 0;
for (int limitValue : limitsPerProcs.values()) {
sumOfProcsLimitSet += limitValue;
}

for (Map.Entry<String, Integer> procs : limitsPerProcs.entrySet()) {

String queryRowUuids = "SELECT uuid FROM cws_sched_worker_proc_inst " +
"WHERE status='" + PENDING + "' AND" +
" proc_def_key='" + procs.getKey() + "' " +
"ORDER BY" +
" priority ASC," + // lower priorities favored
" created_time ASC " + // older dates (FIFO) favored
"LIMIT " + procs.getValue() * 2;

// get list of uuids using array of procdefkeys IN (keys)
rowUuidsPerProcDefKey = jdbcTemplate.queryForList(queryRowUuids, String.class); // over-find because some workers might compete with this set
unfilteredRowUuids.addAll(rowUuidsPerProcDefKey);
}


Collections.sort(unfilteredRowUuids);
for (String id : unfilteredRowUuids) {
String procDefKeyString = getProcDefKeyFromUuid(id);
uuidAndProcDefKeyPair.put(id, procDefKeyString);
}

for (Map.Entry<String,Integer> procLimit : limitsPerProcs.entrySet()) {
Set<String> keys = uuidAndProcDefKeyPair.keySet();
int applyPerProcsCap = 0;
for (String key : keys) {

if (uuidAndProcDefKeyPair.get(key).equals(procLimit.getKey())) {
applyPerProcsCap = applyPerProcsCap + 1;
if (applyPerProcsCap > procLimit.getValue()) {
clearOutUnclaimedInst.add(key);
}
}
}
}

for (String removeUuidFromList : clearOutUnclaimedInst) {
uuidAndProcDefKeyPair.remove(removeUuidFromList);
}

Set<String> uuidKeys = uuidAndProcDefKeyPair.keySet();
// after its filtered add the uuids to rowUuids arraylist
for (String key : uuidKeys) {
rowUuids.add(key);
}

// make query that uses multi limit per ProcDefkey (JOIN)
// iterate to grab 30
if (!rowUuids.isEmpty()) {
// Found some claimable rows, so now try to claim them..
//
Expand All @@ -283,7 +341,7 @@ public Map<String,List<String>> claimHighestPriorityStartReq(String workerId, St
numClaimed++;
claimUuids.add(claimUuid);
claimedRowUuids.add(uuid);
log.debug("CLAIMED " + claimUuid + " (uuid=" +uuid+") for procDefKey '" + procDefKey + "'");
//log.debug("CLAIMED " + claimUuid + " (uuid=" +uuid+") for procDefKey '" + procDefKeyList + "'");
}

if (numClaimed == limit) {
Expand All @@ -293,12 +351,12 @@ public Map<String,List<String>> claimHighestPriorityStartReq(String workerId, St

if (numClaimed == 0) {
// other workers beat us to claiming the rows
log.warn("Attempted to claim " + rowUuids.size() + " rows for procDefKey '" + procDefKey + "', but claimed none! " +
log.warn("Attempted to claim " + rowUuids.size() + " rows for procDefKeys '" + workerProcsList.keySet() + "', but claimed none! " +
(attempts < 10 ? "Retrying..." : "GIVING UP!"));
continue; // retry finding claimable rows
}
else {
log.debug("Claimed (" + numClaimed + " of " + rowUuids.size() + ") for procDefKey '" + procDefKey + "'");
log.debug("Claimed (" + numClaimed + " of " + rowUuids.size() + ") for procDefKeys '" + workerProcsList.keySet() + "'");
}
}
else if (log.isTraceEnabled()) {
Expand Down Expand Up @@ -335,11 +393,11 @@ else if (log.isTraceEnabled()) {
if (numClaimed != claimUuids.size()) {
log.error("numUpdated != claimUuids.size()" );
}

Map<String,List<String>> ret = new HashMap<String,List<String>>();
ret.put("claimUuids", claimUuids);
ret.put("claimedRowUuids", claimedRowUuids);

return ret;
}

Expand All @@ -356,8 +414,39 @@ public String getProcInstRowStatus(String uuid) {
return null;
}
}



public int getMaxProcsValueForWorker(String workerId) {

return jdbcTemplate.queryForObject(
"SELECT max_num_running_procs FROM cws_worker WHERE id=?",
new Object[] {workerId}, Integer.class);
}

public int getProcDefKeyPerWorker(String workerId) {
return jdbcTemplate.queryForObject(
"SELECT COUNT(*) FROM cws_worker_proc_def WHERE worker_id=?",
new Object[] {workerId}, Integer.class);
}

public int getMaxInstancesValueForProcDef(String workerId, String procDefKey) {
return jdbcTemplate.queryForObject(
"SELECT max_instances FROM cws_worker_proc_def WHERE worker_id=? AND proc_def_key=?",
new Object[] {workerId, procDefKey}, Integer.class);
}


public int getCountForClaimedProcInstPerKey(String procDefKey, List<String> claimedUuids) {
String listOfClaimUuid = "\"" + String.join("\", \"", claimedUuids) + "\"" ;
String query = "SELECT count(*) FROM cws_sched_worker_proc_inst " + "WHERE proc_def_key='" + procDefKey + "' " + "AND claim_uuid IN (" + listOfClaimUuid + ")";
return jdbcTemplate.queryForObject(query, Integer.class);
}


public String getProcDefKeyFromUuid(String uuid) {
String query = "SELECT proc_def_key FROM cws_sched_worker_proc_inst " + "WHERE uuid='" + uuid + "'";
return jdbcTemplate.queryForObject(query, String.class);
}

public Map<String,Object> getProcInstRow(String uuid) {
List<Map<String,Object>> list = jdbcTemplate.queryForList(
"SELECT * FROM cws_sched_worker_proc_inst " +
Expand Down
6 changes: 4 additions & 2 deletions cws-engine-service/src/main/java/jpl/cws/engine/EngineDbService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@ public class EngineDbService extends DbService implements InitializingBean {
@Value("${cws.install.type}") private String cwsInstallType;
@Value("${cws.worker.type}") private String cwsWorkerType;
@Value("${camunda.executor.service.max.pool.size}") private int maxExecutorServicePoolSize;
@Value("${worker.max.num.running.procs}") private int workerMaxNumRunningProcs;

private Logger log;

Expand Down Expand Up @@ -180,8 +181,8 @@ public void createOrUpdateWorkerRow(String lockOwner) {
numUpdated = jdbcTemplate.update(
"INSERT INTO cws_worker" +
" (id, lock_owner, name, install_directory, cws_install_type, cws_worker_type, " +
" status, job_executor_max_pool_size, created_time, last_heartbeat_time) " +
"VALUES (?,?,?,?,?,?,?,?,?,?)",
" status, job_executor_max_pool_size, max_num_running_procs, created_time, last_heartbeat_time) " +
"VALUES (?,?,?,?,?,?,?,?,?,?,?)",
new Object[]{
workerId,
lockOwner,
Expand All @@ -191,6 +192,7 @@ public void createOrUpdateWorkerRow(String lockOwner) {
cwsWorkerType,
null, // status will be changed to null once the worker is fully initialized
maxExecutorServicePoolSize, // changeable later via the UI..
workerMaxNumRunningProcs,
tsNow, // created_time
tsNow // last_heartbeat_time
});
Expand Down
121 changes: 80 additions & 41 deletions cws-engine-service/src/main/java/jpl/cws/engine/WorkerService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -75,7 +75,9 @@ public class WorkerService implements InitializingBean {
@Value("${cws.engine.jobexecutor.enabled}") private boolean jobExecutorEnabled;

@Value("${cws.tomcat.lib}") private String cwsTomcatLib;


@Value("${worker.max.num.running.procs}") private int workerMaxNumRunningProcs;

private Logger log;

public static AtomicInteger processorExecuteCount = new AtomicInteger(0);
Expand All @@ -84,8 +86,9 @@ public class WorkerService implements InitializingBean {

// Map of procDefKey and count of active process instances
public static Map<String,Integer> processCounters = new HashMap<String,Integer>();

private static Map<String,Integer> workerMaxProcInstances = new HashMap<String,Integer>();

private static Set<String> procStartReqUuidStartedThisWorker = new HashSet<String>();
private static Set<String> acceptingProcDefKeys = new HashSet<String>();
//private static Set<String> runningToCompleteTransitionUuids = new HashSet<String>();
Expand Down Expand Up @@ -185,7 +188,7 @@ public void initProcessCountersAndLimits() {

}

log.debug("AFTER INIT: limits: " + workerMaxProcInstances + ", counts: " + processCounters);
log.info("AFTER INIT: limits: " + workerMaxProcInstances + ", counts: " + processCounters);
}


Expand Down Expand Up @@ -312,7 +315,7 @@ public boolean updateProcessCountersAndLimits() {
//
String postConfig = "limits: " + workerMaxProcInstances + ", counts: " + processCounters;
if (lastProcCounterStatusMsg == null || !lastProcCounterStatusMsg.equals(postConfig)) {
log.debug("NEW: " + postConfig + ", OLD: " + lastProcCounterStatusMsg);
log.info("NEW: " + postConfig + ", OLD: " + lastProcCounterStatusMsg);
lastProcCounterStatusMsg = postConfig;
return true; // config changed
}
Expand Down Expand Up @@ -670,8 +673,17 @@ public List<Map<String,Object>> claimWithCounter(String limitToProcDefKey) {

synchronized (procStateLock) { // procCountsLock
t1 = System.currentTimeMillis();

int procSetSize = 0;
//int totalCurrentRunningProcsOnWorker = 0;
Map<String,Integer> currentCounts = new HashMap<String,Integer>();
Map<String,Integer> remainders = new HashMap<String,Integer>();
Map<String,Integer> queryLimitForProcSet = new HashMap<String,Integer>();
Map<String,Integer> limitToProcDefKeyObject = new HashMap<String,Integer>();

for (Entry<String,Integer> procMax : workerMaxProcInstances.entrySet()) {
String procDefKey = procMax.getKey();

if (limitToProcDefKey != null && !limitToProcDefKey.equals(procDefKey)) {
continue;
}
Expand All @@ -680,49 +692,76 @@ public List<Map<String,Object>> claimWithCounter(String limitToProcDefKey) {
//log.debug("skipping " + procDefKey + " BECAUSE IT NOT ACCEPTING RIGHT NOW!!!!");
continue;
}


currentCounts.put(procDefKey, processCounters.get(procDefKey));
remainders.put(procDefKey, procMaxNumber - currentCounts.get(procDefKey));
queryLimitForProcSet.put(procDefKey, Math.min(EXEC_SERVICE_MAX_POOL_SIZE, remainders.get(procDefKey)));

//log.trace("getting currentCount for procDefKey " + procDefKey);
int currentCount = processCounters.get(procDefKey);
//int currentCount = processCounters.get(procDefKey);
//log.trace("currentCount for " + procDefKey + " is " + currentCount);
int remainder = procMaxNumber - currentCount;
//int remainder = procMaxNumber - currentCount;
//log.trace("remainder for " + procDefKey + " is " + remainder);
int queryLimit = Math.min(EXEC_SERVICE_MAX_POOL_SIZE, remainder);
//int queryLimit = Math.min(EXEC_SERVICE_MAX_POOL_SIZE, remainder); // FIXME: needs revisit for proper min
//log.trace("queryLimit for " + procDefKey + " is " + queryLimit);

if (remainder > 0) {
// claim for remainder (marks DB rows as "claimedByWorker")
Map<String,List<String>> claimRowData =
schedulerDbService.claimHighestPriorityStartReq(
workerId, procDefKey, queryLimit);

List<String> claimed = claimRowData.get("claimUuids");

if (!claimed.isEmpty()) {
// increment counter by amount that was actually claimed
// in anticipation that the start will actually work.
// If the start turns out not to later worker, then this count will be decremented at that time.
//
processCounters.put(procDefKey, processCounters.get(procDefKey) + claimed.size());
// update uuid list
procStartReqUuidStartedThisWorker.addAll(claimRowData.get("claimedRowUuids"));
//log.debug("procStartReqUuidStartedThisWorker = " + procStartReqUuidStartedThisWorker);

log.debug("(CLAIMED " + claimed.size() + " / " + queryLimit + ", maxProcs=" + procMaxNumber + ") for procDef '" + procDefKey + "' (limitToProcDefKey="+limitToProcDefKey+")");

claimUuids.addAll(claimed);

} // end for loop

int totalCurrentRunningProcsOnWorker = 0;
for (Entry<String,Integer> entry : processCounters.entrySet()) {
totalCurrentRunningProcsOnWorker += entry.getValue().intValue();
}

// rename to workerMaxProcQueryLimit
int MaxNumForProcsOnWorker = schedulerDbService.getMaxProcsValueForWorker(workerId);
// this is for all procDefs cap
int workerMaxProcQueryLimit = MaxNumForProcsOnWorker - totalCurrentRunningProcsOnWorker;

int remaindersTotal = 0;
for (int r: remainders.values()) {
remaindersTotal += r;
}

if (remaindersTotal > 0 && workerMaxProcQueryLimit > 0) {
// claim for remainder (marks DB rows as "claimedByWorker")

int queryLimit = Math.min(MaxNumForProcsOnWorker, workerMaxProcQueryLimit);

Map<String,List<String>> claimRowData =
schedulerDbService.claimHighestPriorityStartReq(
workerId, currentCounts, queryLimitForProcSet, queryLimit); // pass list of procDefkey and a map of queryLimit per procDefKey

List<String> claimed = claimRowData.get("claimUuids");

if (!claimed.isEmpty()) {
// increment counter by amount that was actually claimed
// in anticipation that the start will actually work.
// If the start turns out not to later worker, then this count will be decremented at that time.
//
for (Map.Entry<String,Integer> procDefKey : processCounters.entrySet()) {
int claimedInstCount = schedulerDbService.getCountForClaimedProcInstPerKey(procDefKey.getKey(), claimed);
processCounters.put(procDefKey.getKey(), processCounters.get(procDefKey.getKey()) + claimedInstCount);
}
//else {
// log.debug("NONE CLAIMED (queryLimit=" + queryLimit + ", max=" + procMaxNumber + ") for procDef '" + procDefKey + "' (limitToProcDefKey="+limitToProcDefKey+")");
//}
}
else {
log.debug("[" + procDefKey + "] remainder <= 0, so not attempting claim. " +
"(remainder = " + remainder +
", procMaxNumber = " + procMaxNumber +
", currentCount = " + currentCount + ")");

// update uuid list
procStartReqUuidStartedThisWorker.addAll(claimRowData.get("claimedRowUuids"));
//log.debug("procStartReqUuidStartedThisWorker = " + procStartReqUuidStartedThisWorker);

log.debug("(CLAIMED " + claimed.size() + " / " + queryLimit + ", maxProcs=" + workerMaxProcInstances.entrySet() + ") for procDefKeys '" + workerMaxProcInstances.keySet() + "' (limitToProcDefKey="+limitToProcDefKey+")" + ", workerMaxNumRunningProcs=" + MaxNumForProcsOnWorker);
claimUuids.addAll(claimed);
}

} // end for loop
//else {
// log.debug("NONE CLAIMED (queryLimit=" + queryLimit + ", max=" + procMaxNumber + ") for procDef '" + procDefKey + "' (limitToProcDefKey="+limitToProcDefKey+")");
//}
}
else {
log.debug("Remainder for Worker Max Process Limit [" + workerMaxProcQueryLimit + "] workerMaxProcQueryLimit <= 0 OR Total of remainders [" + remaindersTotal + "] is <=0, so not attempting claim. " +
"(remainders = " + remainders +
", procMaxNumbers = " + workerMaxProcInstances.entrySet() +
", currentCounts = " + currentCounts + ")");
}



} // release lock

Expand Down
Loading