Merge Updates

docs/_config.yml

@@ -0,0 +1,3 @@
+theme: jekyll-theme-minimal
+title: Docker Unified UIMA Interface
+description: A framework at the dawn of a new era for NLP processing

src/main/java/org/texttechnologylab/DockerUnifiedUIMAInterface/DUUIDockerInterface.java

@@ -20,33 +20,42 @@
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.CancellationException;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import static java.lang.String.format;
 
 //TODO: Enable different tags here
 class PullImageStdout extends PullImageResultCallback {
     private String _status;
+    private final AtomicBoolean shutdown;
 
-    PullImageStdout() {
+    PullImageStdout(AtomicBoolean shutdown) {
         _status = "";
+        this.shutdown = shutdown;
     }
 
     @Override
     public void onNext(PullResponseItem item) {
+        if (shutdown.get()) {
+            onComplete();
+            onError(new CancellationException("Pull has been cancelled."));
+            return;
+        }
+
         if (item.getStatus() != null) {
             _status = item.getStatus();
         }
 
         if (item.getProgressDetail() != null) {
             if (item.getProgressDetail().getCurrent() != null && item.getProgressDetail().getTotal() != null) {
-                System.out.printf("[DockerSwarmDriver] %s: %.2f%%\n", _status, ((float) item.getProgressDetail().getCurrent() / (float) item.getProgressDetail().getTotal()) * 100);
+                float progress = ((float) item.getProgressDetail().getCurrent() / (float) item.getProgressDetail().getTotal()) * 100;
             } else {
                 System.out.printf("[DockerSwarmDriver] %s.\n", _status);
             }
         }
     }
-
 }
 
 class PushImageStdout extends ResultCallback.Adapter<PushResponseItem> {
@@ -154,16 +163,16 @@ public class DUUIDockerInterface {
     public DUUIDockerInterface() throws IOException {
 
         // GREAT TODO!!!
-//        URI dockerClientURI;
-//        if (System.getProperty("os.name").toLowerCase().contains("windows")) {
-//            dockerClientURI = URI.create("npipe:////./pipe/docker_engine");
-//        } else {
-//            dockerClientURI = URI.create("tcp://localhost:2375");
-//        }
-//
-//        _docker = DockerClientBuilder.getInstance()
-//                .withDockerHttpClient(new ApacheDockerHttpClient.Builder()
-//                        .dockerHost(dockerClientURI).build()).build();
+        //URI dockerClientURI;
+        //if (System.getProperty("os.name").toLowerCase().contains("windows")) {
+        //    dockerClientURI = URI.create("npipe:////./pipe/docker_engine");
+        //} else {
+        //    dockerClientURI = URI.create("tcp://localhost:2375");
+        //}
+
+        //_docker = DockerClientBuilder.getInstance()
+//            .withDockerHttpClient(new ApacheDockerHttpClient.Builder()
+//                .dockerHost(dockerClientURI).build()).build();
         _docker = DockerClientBuilder.getInstance().build();
 
 
@@ -181,7 +190,7 @@ public DUUIDockerInterface() throws IOException {
      */
     public int extract_port_mapping(String containerid) throws InterruptedException {
         InspectContainerResponse container
-                = _docker.inspectContainerCmd(containerid).exec();
+            = _docker.inspectContainerCmd(containerid).exec();
 
         int innerport = 0;
         for (Map.Entry<ExposedPort, Ports.Binding[]> port : container.getNetworkSettings().getPorts().getBindings().entrySet()) {
@@ -203,7 +212,7 @@ public int extract_port_mapping(String containerid) throws InterruptedException
      */
     public int extract_port_mapping(String containerid, int portMapping) throws InterruptedException {
         InspectContainerResponse container
-                = _docker.inspectContainerCmd(containerid).exec();
+            = _docker.inspectContainerCmd(containerid).exec();
 
         int innerport = 0;
         for (Map.Entry<ExposedPort, Ports.Binding[]> port : container.getNetworkSettings().getPorts().getBindings().entrySet()) {
@@ -261,14 +270,14 @@ public String get_ip() {
     public String get_logs(String containerid) throws InterruptedException {
         final List<String> logs = new ArrayList<>();
         InspectContainerResponse container
-                = _docker.inspectContainerCmd(containerid).exec();
+            = _docker.inspectContainerCmd(containerid).exec();
         _docker.logContainerCmd(containerid).withContainerId(containerid).withStdOut(true).withStdErr(true)
-                .withTimestamps(true).withTail(5).exec(new LogContainerResultCallback() {
-                    @Override
-                    public void onNext(Frame item) {
-                        logs.add(item.toString());
-                    }
-                }).awaitCompletion();
+            .withTimestamps(true).withTail(5).exec(new LogContainerResultCallback() {
+                @Override
+                public void onNext(Frame item) {
+                    logs.add(item.toString());
+                }
+            }).awaitCompletion();
         String completelog = "";
         for (String x : logs) {
             completelog += x;
@@ -282,7 +291,7 @@ public void onNext(Frame item) {
      * @param id The id of the container to stop.
      */
     public void stop_container(String id) {
-        try{
+        try {
             _docker.stopContainerCmd(id).withTimeout(10).exec();
         } catch (NotModifiedException e) {
         } catch (Exception e) {
@@ -313,14 +322,14 @@ public void export_to_new_image(String containerid, String imagename) {
 
     public Image getLocalImage(String imageName) {
         List<Image> images = _docker.listImagesCmd()
-                .withShowAll(true)
-                .exec();
-        for(Image i : images) {
-            if(i.getId() == imageName) {
+            .withShowAll(true)
+            .exec();
+        for (Image i : images) {
+            if (i.getId() == imageName) {
                 return i;
             }
             String[] repoTags = i.getRepoTags();
-            if(repoTags!=null) {
+            if (repoTags != null) {
                 for (String repo : repoTags) {
                     if (repo.equals(imageName)) {
                         return i;
@@ -331,32 +340,32 @@ public Image getLocalImage(String imageName) {
         return null;
     }
 
-    public void push_image(String remoteName, String localName,String username, String password) throws InterruptedException {
+    public void push_image(String remoteName, String localName, String username, String password) throws InterruptedException {
         Image img = getLocalImage(localName);
-        if(img==null) {
-            throw new InvalidParameterException(format("Could not find local image %s, not attempting to upload it to a registry!",localName));
+        if (img == null) {
+            throw new InvalidParameterException(format("Could not find local image %s, not attempting to upload it to a registry!", localName));
         }
 
-        _docker.tagImageCmd(localName,remoteName,"latest").exec();
-        if(username!=null && password!=null) {
+        _docker.tagImageCmd(localName, remoteName, "latest").exec();
+        if (username != null && password != null) {
             AuthConfig cfg = new AuthConfig();
             cfg.withPassword(password);
             cfg.withUsername(username);
             _docker.pushImageCmd(remoteName)
-                    .withAuthConfig(cfg)
-                    .exec(new PushImageStdout())
-                    .awaitCompletion();
-        }
-        else {
+                .withAuthConfig(cfg)
+                .exec(new PushImageStdout())
+                .awaitCompletion();
+        } else {
             _docker.pushImageCmd(remoteName)
-                    .exec(new PushImageStdout())
-                    .awaitCompletion();
+                .exec(new PushImageStdout())
+                .awaitCompletion();
         }
     }
 
     public String run_service(String imagename, int scale) throws InterruptedException {
         return run_service(imagename, scale, new ArrayList<>(0));
     }
+
     public String run_service(String imagename, int scale, List<String> constraints) throws InterruptedException {
         ServiceSpec spec = new ServiceSpec();
         ServiceModeConfig cfg = new ServiceModeConfig();
@@ -369,7 +378,7 @@ public String run_service(String imagename, int scale, List<String> constraints)
         ContainerSpec cont = new ContainerSpec();
         cont = cont.withImage(imagename);
         task.withContainerSpec(cont);
-        if(constraints.size()>0) {
+        if (constraints.size() > 0) {
             task.withPlacement(new ServicePlacement().withConstraints(constraints));
         }
 
@@ -386,26 +395,25 @@ public String run_service(String imagename, int scale, List<String> constraints)
 
     public boolean hasLocalImage(String imageName) {
         InspectImageResponse resp = _docker.inspectImageCmd(imageName).exec();
-        if(resp!=null) {
+        if (resp != null) {
             return true;
         }
 
         List<Image> images = _docker.listImagesCmd()
-                .withShowAll(true)
-                .exec();
-        for(Image i : images) {
-            if(i.getId() == imageName) {
+            .withShowAll(true)
+            .exec();
+        for (Image i : images) {
+            if (i.getId() == imageName) {
                 return true;
             }
 
-            if(i.getRepoTags()!=null) {
+            if (i.getRepoTags() != null) {
                 for (String repo : i.getRepoTags()) {
                     if (repo.equals(imageName)) {
                         return true;
                     }
                 }
-            }
-            else if(i.getRepoDigests()!=null) {
+            } else if (i.getRepoDigests() != null) {
                 for (String dig : i.getRepoDigests()) {
                     if (dig.equals(imageName)) {
                         return true;
@@ -422,8 +430,8 @@ public DockerClient getDockerClient() {
 
     public String build(Path builddir, List<String> buildArgs) {
         BuildImageCmd buildCmd = _docker.buildImageCmd().withPull(true)
-                .withBaseDirectory(builddir.toFile())
-                .withDockerfile(Paths.get(builddir.toString(), "dockerfile").toFile());
+            .withBaseDirectory(builddir.toFile())
+            .withDockerfile(Paths.get(builddir.toString(), "dockerfile").toFile());
 
         for (String buildArg : buildArgs) {
             String[] fields = buildArg.split("=", 2);
@@ -441,44 +449,64 @@ public boolean isSwarmManagerNode() {
         return sw.getSwarm().getControlAvailable();
     }
 
+
     public String pullImage(String tag) throws InterruptedException {
         return pullImage(tag, null, null);
     }
 
-    public String pullImage(String tag,String username, String password) throws InterruptedException {
+
+    public String pullImage(String tag, String username, String password) throws InterruptedException {
+        return pullImage(tag, username, password, null);
+    }
+
+    /**
+     * Attempts to pull a docker image using its tag, and credentials if necessary.
+     *
+     * @param tag      The unique identifier of the image (uri:version).
+     * @param username Credentials username.
+     * @param password Credentials password.
+     * @param shutdown A flag to signal a pull cancel.
+     * @return The image tag.
+     * @throws InterruptedException
+     */
+    public String pullImage(String tag, String username, String password, AtomicBoolean shutdown) throws InterruptedException {
+
         try {
             if (username != null && password != null) {
                 AuthConfig cfg = new AuthConfig();
                 cfg.withUsername(username);
                 cfg.withPassword(password);
                 ResultCallbackTemplate temp = _docker.pullImageCmd(tag)
-                        .withAuthConfig(cfg)
-                        .exec(new PullImageStdout());
+                    .withAuthConfig(cfg)
+                    .exec(new PullImageStdout(shutdown));
                 temp.onError(new Exception());
                 temp.awaitCompletion();
             } else {
+
                 ResultCallbackTemplate template = _docker.pullImageCmd(tag)
-                        .exec(new PullImageStdout());
-                template.onError(new Exception());
+                    .exec(new PullImageStdout(shutdown));
+
+                // What is the purpose of throwing an Exception every time?
                 template.awaitCompletion();
             }
-        }
-        catch(Exception e) {
+        } catch (CancellationException exception) {
+            shutdown.set(true);
+        } catch (Exception e) {
             e.printStackTrace();
-            System.out.printf("Could not fetch image %s, continuing without.\n",tag);
+            System.out.printf("Could not fetch image %s, continuing without.\n", tag);
         }
         return tag;
     }
 
     public String getDigestFromImage(String imagename) {
 
-        if(imagename.split(":").length == 1) {
-            imagename = imagename+":latest";
+        if (imagename.split(":").length == 1) {
+            imagename = imagename + ":latest";
         }
         InspectImageResponse resp = _docker.inspectImageCmd(imagename).exec();
         List<String> digests = resp.getRepoDigests();
 
-        for(String i : digests) {
+        for (String i : digests) {
             return i;
         }
         return null;
@@ -501,15 +529,15 @@ public String run(String imageid, boolean gpu, boolean autoremove, int port, boo
         }
         if (gpu) {
             cfg = cfg.withDeviceRequests(ImmutableList.of(new DeviceRequest()
-                    .withCapabilities(ImmutableList.of(ImmutableList.of("gpu")))));
+                .withCapabilities(ImmutableList.of(ImmutableList.of("gpu")))));
         }
 
-        if(mapDaemon) {
+        if (mapDaemon) {
             cfg = cfg.withBinds(Bind.parse("/var/run/docker.sock:/var/run/docker.sock"));
         }
         CreateContainerCmd cmd = _docker.createContainerCmd(imageid)
-                .withHostConfig(cfg)
-                .withExposedPorts(ExposedPort.tcp(port)).withPublishAllPorts(true);
+            .withHostConfig(cfg)
+            .withExposedPorts(ExposedPort.tcp(port)).withPublishAllPorts(true);
 
         CreateContainerResponse feedback = cmd.exec();
         _docker.startContainerCmd(feedback.getId()).exec();
@@ -533,17 +561,17 @@ public String run(String imageid, boolean gpu, boolean autoremove, int portConta
         }
         if (gpu) {
             cfg = cfg.withDeviceRequests(ImmutableList.of(new DeviceRequest()
-                    .withCapabilities(ImmutableList.of(ImmutableList.of("gpu")))));
+                .withCapabilities(ImmutableList.of(ImmutableList.of("gpu")))));
         }
 
-        cfg.withPortBindings(new PortBinding(new Ports.Binding(null,String.valueOf(portHost)), new ExposedPort(portContainer)));
+        cfg.withPortBindings(new PortBinding(new Ports.Binding(null, String.valueOf(portHost)), new ExposedPort(portContainer)));
 
-        if(mapDaemon) {
+        if (mapDaemon) {
             cfg = cfg.withBinds(Bind.parse("/var/run/docker.sock:/var/run/docker.sock"));
         }
         CreateContainerCmd cmd = _docker.createContainerCmd(imageid)
-                .withHostConfig(cfg)
-                .withExposedPorts(ExposedPort.tcp(portContainer)).withPublishAllPorts(true);
+            .withHostConfig(cfg)
+            .withExposedPorts(ExposedPort.tcp(portContainer)).withPublishAllPorts(true);
 
         CreateContainerResponse feedback = cmd.exec();
         _docker.startContainerCmd(feedback.getId()).exec();