Executing multiple machines in parallel in thread safe manner

Question

I have a list of hostname which I am executing in parallel using ExecutorService to collect all the metrics for each hostname. And then I am making a List which has all the metrics related information for all the hostnames by iterating each hostnames future. Since I am executitng multiple hostnames in parallel so I am not sure whether this code is thread safe or not.

This is my main code where I am executing multiple HOSTNAMES in parallel:

final Flows typeOfFlow = Flows.TREE;

List<Future<MachineMetrics>> machineFutureList = new ArrayList<>();
for (final String machine : HOSTNAMES) {
    machineFutureList.add(executorService.submit(new Callable<MachineMetrics>() {
        @Override
        public MachineMetrics call() throws Exception {
            MachineMetrics machineMetrics = new MachineMetrics();
            String url = "http://" + machine + ":8080/text";
            Map<String, String> metrics = getMetrics(machine, url, typeOfFlow);
            machineMetrics.setMachineName(machine.split("\\.")[0]);
            machineMetrics.setDatacenter(TestUtils.findDatacenter(machine).get().name().toLowerCase());
            machineMetrics.setMetrics(metrics);
            return machineMetrics;
        }
    }));
}
List<MachineMetrics> metricsList = new ArrayList<>();
for (Future<MachineMetrics> future : machineFutureList) {
    try {
        metricsList.add(future.get());
    } catch (InterruptedException | ExecutionException ex) {
        // log exception here
    }
}
// now print all the hostnames metrics information
System.out.println(metricsList);

And below is my getMetrics code in the same class where my above code is there:

private Map<String, String> getMetrics(final String machine, final String url, final Flows flowType) {
    Map<String, String> holder = new HashMap<String, String>();
    try {
        RestTemplate restTemplate = RestTemplateClient.getInstance().getClient();
        String response = restTemplate.getForObject(url, String.class);
        Matcher m = PATTERN.matcher(response);
        while (m.find()) {
            String key = m.group(1).trim();
            String value = m.group(2).trim();
            holder.put(key, value);
        }
    } catch (Exception ex) {
        // log here
    }

    return TestUtils.process(holder);
}

And below is my findDatacenter code in TestUtils class:

public static Optional<Datacenter> findDatacenter(final String hostname) {
    if (!TestUtils.isEmpty(hostname)) {
        for (Datacenter dc : DC_LIST) {
            String namepart = "." + dc.name().toLowerCase() + ".";
            if (hostname.indexOf(namepart) >= 0) {
                return Optional.of(dc);
            }
        }
    }
    return Optional.absent();
}

And below is my process method in TestUtils class:

public static Map<String, String> process(final Map<String, String> holder) {
    Map<String, String> tempMap = new HashMap<>();

    for (Map.Entry<String, String> entry : holder.entrySet()) {
        if (!entry.getKey().startsWith("calls_") && !entry.getValue().contains("|")) {
            continue;
        }
        String currentKey = entry.getKey();
        String currentValue = entry.getValue();
        StringTokenizer tokenizer = new StringTokenizer(currentValue, "|");

        String count = tokenizer.nextToken().trim();
        String avgData = tokenizer.nextToken().trim();
        String medianData = tokenizer.nextToken().trim();
        String n95data = tokenizer.nextToken().trim();
        String n99data = tokenizer.nextToken().trim();

        tempMap.put(generateKey(currentKey, currentKey.contains(MISS), COUNT), count);
        tempMap.put(generateKey(currentKey, currentKey.contains(MISS), AVG_IN_MS), avgData);
        tempMap.put(generateKey(currentKey, currentKey.contains(MISS), MEDIAN_IN_MS), medianData);
        tempMap.put(generateKey(currentKey, currentKey.contains(MISS), N95_IN_MS), n95data);
        tempMap.put(generateKey(currentKey, currentKey.contains(MISS), N99_IN_MS), n99data);

        holder.remove(currentKey);
    }

    tempMap.putAll(holder);

    return tempMap;
}

And below is my generateKey method in TestUtils class:

private static String generateKey(final String currentKey, final boolean hasMiss, final String constant) {
    StringBuilder newKey = new StringBuilder();

    if (hasMiss) {
        newKey.append(currentKey).append(constant);
    } else {
        String firstPart = currentKey.substring(0, currentKey.indexOf("_"));
        String secondPart = currentKey.substring(currentKey.lastIndexOf("_") + 1, currentKey.length());
        newKey.append(firstPart).append(CACHE).append(secondPart).append(constant);
    }

    return newKey.toString();
}

And below is my MachineMetrics class:

public class MachineMetrics {

    private String machineName;
    private String datacenter;
    private Map<String, String> metrics;

    // normal setters and getters here
}

Is my above code thread safe? Am I doing anything wrong by which I may be seeing wrong results because of some race conditions or thread safety issues?

score 1 · Accepted Answer · edited May 23 '17 at 12:07

1

Looks good. Your methods are stateless. Also you use immutable objects as methods params. So you will have no problem with thread safety.

One remark:

for (Future<MachineMetrics> future : machineFutureList) {
    try {
        metricsList.add(future.get());
    } catch (InterruptedException | ExecutionException ex) {
        // log exception here
    }
}

get Waits if necessary for the computation to complete, and then retrieves its result. So if first call was slow, you will not retrieve other results. Use isDone to check that you can call get without waiting.

edited May 23 '17 at 12:07

Community

1
1

answered Feb 08 '16 at 07:14

Xupypr MV

935
10
18

Thanks for your suggestion. Can you provide an example how will I use `isDone` method here along with `get` call? I am slightly confuse here. – john Feb 08 '16 at 16:36
And also in some of my methods, I do have `static` in my `TestUtils` class. Is that ok? Since I am using those methods from the code which is multithreaded. – john Feb 08 '16 at 16:49
It's ok to use `static`. Because method is not synchronized and not use any shared state. – Xupypr MV Feb 08 '16 at 19:12
Here is example with `isDone` call: http://javarevisited.blogspot.ru/2015/01/how-to-use-future-and-futuretask-in-Java.html – Xupypr MV Feb 08 '16 at 19:14

Executing multiple machines in parallel in thread safe manner

1 Answers1