Merge lp:~elambert/gearman-java/Bug400466 into lp:gearman-java

[Eric Lambert]

[Eric Lambert (elambert)]

fix and test for 400466. Removed selectUpdatedEvents method as it was the implementation of this method which was causing the leak. Users that what to receive all intermediate events from a job (which is what the selectUpdatedEvents method did) can instead register a listener to called when events are received.

=== modified file 'src/org/gearman/client/GearmanClient.java'
--- src/org/gearman/client/GearmanClient.java	2009-07-15 18:47:07 +0000
+++ src/org/gearman/client/GearmanClient.java	2009-08-05 18:34:26 +0000
@@ -6,13 +6,11 @@
 package org.gearman.client;
 
 import java.io.IOException;
-import java.util.Collection;
 import java.util.List;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Future;
 
 import org.gearman.common.GearmanJobServerConnection;
-import org.gearman.common.GearmanPacket;
 
 /**
  * This interface through which users of the Gearman Java Libaray will use to
@@ -106,22 +104,6 @@
             IllegalStateException;
 
     /**
-     * As {@link GearmanJob} are submitted and executed, they will receive
-     * notifications from the Gearman Job Server as certain packets are received
-     * (such as JOB_COMPLETE or WORK_DATA). Users may want to process these
-     * packets,for example if jobs are sending back intermediate data, a user
-     * may wish to receive that data and act upon it. This method returns to
-     * the user all events which have occured for any job submitted by the
-     * client.
-     *
-     * @return A collection of {@link GearmanPacket} that have been received.
-     *
-     * @throws IllegalStateException If the client has already been stopped.
-     */
-    Collection<GearmanPacket> selectUpdatedJobEvents()
-            throws IllegalStateException;
-
-    /**
      * Sends a WORK_STATUS request to the Gearman Job Server running the
      * {@link GearmanJob} and then returns {@link GearmanJobStatus} representing
      * the current status of the job.

=== modified file 'src/org/gearman/client/GearmanClientImpl.java'
--- src/org/gearman/client/GearmanClientImpl.java	2009-08-05 00:13:52 +0000
+++ src/org/gearman/client/GearmanClientImpl.java	2009-08-05 18:34:26 +0000
@@ -12,7 +12,6 @@
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.HashMap;
-import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Set;
@@ -70,7 +69,6 @@
     private final String DESCRIPTION;
     private HashMap<SelectionKey, GearmanJobServerSession> sessionsMap = null;
     private Selector ioAvailable = null;
-    private ArrayList<GearmanPacket> updatedJobs = null;
     private static final Logger LOG = Logger.getLogger(
             Constants.GEARMAN_CLIENT_LOGGER_NAME);
     private state runState = state.RUNNING;
@@ -125,7 +123,6 @@
      */
     public GearmanClientImpl() {
         sessionsMap = new HashMap<SelectionKey, GearmanJobServerSession>();
-        updatedJobs = new ArrayList<GearmanPacket>();
         jobsMaps = new HashMap<JobHandle, GearmanJobImpl>();
         submitJobMap = new HashMap<GearmanJobServerSession, GearmanJobImpl>();
         DESCRIPTION = new String(DESCRIPION_PREFIX + ":" +
@@ -372,27 +369,6 @@
         return handler.getResults();
     }
 
-    @SuppressWarnings(value = "unchecked")
-    public Collection<GearmanPacket> selectUpdatedJobEvents() throws
-            GearmanException, IllegalStateException {
-        if (!runState.equals(state.RUNNING)) {
-            throw new IllegalStateException("Client is not active");
-        }
-        Collection<GearmanPacket> retSet = new HashSet<GearmanPacket>();
-        try {
-            driveClientIO();
-        } catch (IOException ioe) {
-            LOG.log(Level.WARNING, "Encountered IOException while driving " +
-                    "client IO" + ioe);
-            //TODO improve ioexception handling
-        }
-        if (!updatedJobs.isEmpty()) {
-            retSet = (Collection<GearmanPacket>) updatedJobs.clone();
-            updatedJobs.clear();
-        }
-        return retSet;
-    }
-
     public int getNumberofActiveJobs() throws IllegalStateException {
         if (runState.equals(state.TERMINATED)) {
             throw new IllegalStateException("Client is not active");
@@ -410,7 +386,6 @@
                 GearmanJobImpl sjob = submitJobMap.get(s);
                 if (!sjob.isBackgroundJob()) {
                     jobsMaps.put(new JobHandle(sjob.getHandle()), sjob);
-                    updatedJobs.add(p);
                 }
                 break;
             case WORK_DATA:
@@ -419,7 +394,6 @@
             case WORK_COMPLETE:
             case WORK_FAIL:
             case WORK_EXCEPTION:
-                updatedJobs.add(p);
                 JobHandle handle = new JobHandle(p.getDataComponentValue(
                         GearmanPacket.DataComponentName.JOB_HANDLE));
                 GearmanJobImpl job = jobsMaps.get(handle);
@@ -492,8 +466,6 @@
         }
         sessionsMap.clear();
         sessionsMap = null;
-        updatedJobs.clear();
-        updatedJobs = null;
         runState = state.TERMINATED;
         LOG.log(Level.FINE, "Completed shutdown of client: " + this);
         return new ArrayList<Runnable>();

=== added file 'test/org/gearman/client/Bug400466Test.java'
--- test/org/gearman/client/Bug400466Test.java	1970-01-01 00:00:00 +0000
+++ test/org/gearman/client/Bug400466Test.java	2009-08-05 18:34:26 +0000
@@ -0,0 +1,141 @@
+/*
+ * Copyright (C) 2009 by Eric Lambert <[e-mail]>
+ * Use and distribution licensed under the BSD license.  See
+ * the COPYING file in the parent directory for full text.
+ */
+package org.gearman.client;
+
+import java.io.IOException;
+import java.util.concurrent.ExecutionException;
+import junit.framework.Assert;
+import org.gearman.common.GearmanNIOJobServerConnection;
+import org.gearman.util.ByteUtils;
+import org.gearman.worker.AbstractGearmanFunction;
+import org.gearman.worker.GearmanFunction;
+import org.gearman.worker.GearmanFunctionFactory;
+import org.gearman.worker.GearmanWorker;
+import org.gearman.worker.GearmanWorkerImpl;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+public class Bug400466Test {
+
+    GearmanClientImpl gc = null;
+    GearmanWorker worker = null;
+    Thread workerThread = null;
+    WorkerRunnable runner = null;
+    Thread wt = null;
+    byte[] data = new byte[8193];
+    Runtime rt = null;
+
+    class newReverseFunction extends AbstractGearmanFunction {
+
+        @Override
+        public GearmanJobResult executeFunction() throws Exception {
+            StringBuffer sb = null;
+            byte[] results = null;
+            if (data instanceof byte[]) {
+                sb = new StringBuffer(ByteUtils.fromUTF8Bytes((byte[]) data));
+            } else {
+                sb = new StringBuffer(data.toString());
+            }
+            results = sb.reverse().toString().getBytes();
+            return new GearmanJobResultImpl(jobHandle, true, results,
+                    new byte[0], new byte[0], 0, 0);
+        }
+    }
+
+    class newReverseFunctionFactory implements GearmanFunctionFactory {
+
+        public String getFunctionName() {
+            return newReverseFunction.class.getCanonicalName();
+        }
+
+        public GearmanFunction getFunction() {
+            return new newReverseFunction();
+        }
+    }
+
+    class WorkerRunnable implements Runnable {
+
+        GearmanWorker myWorker = null;
+        boolean isRunning = true;
+
+        public WorkerRunnable(GearmanWorker w) {
+            myWorker = w;
+        }
+
+        public void run() {
+            while (isRunning) {
+                myWorker.work();
+            }
+        }
+
+        public void stop() {
+            isRunning = false;
+        }
+    }
+
+    @Before
+    public void initTest() throws IOException {
+        for (int i = 0; i < data.length; i++) {
+            data[i] = '0';
+        }
+        rt = Runtime.getRuntime();
+        gc = new GearmanClientImpl();
+        worker = new GearmanWorkerImpl();
+        gc.addJobServer(new GearmanNIOJobServerConnection("localhost"));
+
+        //create a worker for each of the job servers
+        worker.addServer(new GearmanNIOJobServerConnection("localhost"));
+        worker.registerFunctionFactory(new newReverseFunctionFactory());
+        runner = new WorkerRunnable(worker);
+        wt = new Thread(runner, "workerThread");
+        wt.start();
+        try {
+            Thread.sleep(100);
+        } catch (InterruptedException ioe) {
+        }
+    }
+
+    @After
+    public void shutdownTest() throws IOException {
+        if (gc != null) {
+            gc.shutdownNow();
+        }
+        if (worker != null) {
+            worker.stop();
+        }
+        if (runner != null) {
+            runner.stop();
+        }
+    }
+
+    @Test
+    public void test400466()
+            throws IOException, InterruptedException, ExecutionException {
+        long heapSizeCeiling = 0;
+        long memUsed = 0;
+        for (int x = 1; x <= 1000; x++) {
+            GearmanJob job = GearmanJobImpl.createJob(
+                    newReverseFunction.class.getCanonicalName(),
+                    data, null);
+            gc.submit(job);
+            if (x % 100 == 0) {
+                memUsed = rt.totalMemory() - rt.freeMemory();
+                if (heapSizeCeiling == 0) {
+                    heapSizeCeiling = memUsed * 4;
+                } else {
+                    Assert.assertTrue("ceiling = " + heapSizeCeiling +
+                            " used = " + memUsed, memUsed < heapSizeCeiling);
+                }
+            }
+        }
+        //Submit once last job and wait on it so as to clear the server q
+        GearmanJob job = GearmanJobImpl.createJob(
+                newReverseFunction.class.getCanonicalName(),
+                data, null);
+        gc.submit(job).get();
+    }
+}

=== modified file 'test/org/gearman/client/GearmanClientJobExecTest.java'
--- test/org/gearman/client/GearmanClientJobExecTest.java	2009-08-05 00:13:52 +0000
+++ test/org/gearman/client/GearmanClientJobExecTest.java	2009-08-05 18:34:26 +0000
@@ -12,8 +12,6 @@
 import org.gearman.common.GearmanPacketType;
 import org.gearman.util.ByteUtils;
 import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collection;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
@@ -160,6 +158,23 @@
         }
     }
 
+    class IncrementalListener implements GearmanIOEventListener {
+
+        private StringBuffer sb = new StringBuffer();
+        public void handleGearmanIOEvent(GearmanPacket event) 
+                throws IllegalArgumentException {
+            if (!event.getPacketType().equals(GearmanPacketType.WORK_DATA)) {
+                return;
+            }
+            sb.append(ByteUtils.fromUTF8Bytes((event.getDataComponentValue(
+                        GearmanPacket.DataComponentName.DATA))));
+        }
+        public String getResults() {
+            return sb.toString();
+        }
+
+    }
+
     @Before
     public void initTest() throws IOException {
         gc = new GearmanClientImpl();
@@ -205,31 +220,17 @@
     public void incrementalAttachedJob()
             throws IOException, InterruptedException, ExecutionException {
             StringBuffer text = generateData(8193, "Hello World");
-            StringBuffer resultText = new StringBuffer();
             GearmanJob job = GearmanJobImpl.createJob(
                     incrementalReverseFunction.class.getCanonicalName(),
                     ByteUtils.toAsciiBytes(text.toString()), null);
+            IncrementalListener il = new IncrementalListener();
+            job.registerEventListener(il);
             gc.submit(job);
-            while (!job.isDone()) {
-                Collection<GearmanPacket> events = gc.selectUpdatedJobEvents();
-                for (GearmanPacket event : events) {
-                    if (!event.getPacketType().equals(GearmanPacketType.WORK_DATA)) {
-                        continue;
-                    }
-                    if (Arrays.equals(event.getDataComponentValue(
-                            GearmanPacket.DataComponentName.JOB_HANDLE),
-                            job.getHandle())) {
-                        resultText.append(ByteUtils.fromUTF8Bytes((
-                                event.getDataComponentValue(
-                                GearmanPacket.DataComponentName.DATA))));
-                    }
-                }
-            }
             GearmanJobResult result = job.get();
             Assert.assertTrue(result.jobSucceeded());
             Assert.assertTrue("Client reports active jobs even though all " +
                     "jobs have completed", gc.getNumberofActiveJobs() == 0);
-            String resultString = resultText.toString();
+            String resultString = il.getResults();
             Assert.assertTrue(resultString.equals(text.reverse().toString()));
     }
 
@@ -275,7 +276,7 @@
             boolean hasHitRunning = false;
             for (int i = 0; i < 5; i++) {
                 try {
-                    Thread.sleep(100);
+                    Thread.sleep(10);
                 } catch (InterruptedException ie) {
                 }
                 status = gc.getJobStatus(job);