make locks expire

git-svn-id: http://google-refine.googlecode.com/svn/trunk@1014 7d457c2a-affb-35e4-300a-418c747d4874

.gitignore

@@ -10,3 +10,6 @@ tools/
 extensions/sample-extension/module/MOD-INF/classes/
 extensions/jython/module/MOD-INF/classes/
 extensions/jython/module/MOD-INF/lib/cachedir/
+broker/appengine/module/MOD-INF/classes/
+broker/core/module/MOD-INF/classes/
+broker/local/module/MOD-INF/classes/

broker/appengine/WEB-INF/cron.xml

@@ -0,0 +1,7 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<cronentries>
+  <cron>
+    <url>/expire_locks</url>
+    <schedule>every 5 seconds synchronized</schedule>
+  </cron>
+</cronentries>

broker/appengine/WEB-INF/web.xml

@@ -1,10 +1,6 @@
 <?xml version="1.0"?>
 
-<!DOCTYPE web-app
+<web-app xmlns="http://java.sun.com/xml/ns/javaee" version="2.5">
-    PUBLIC "-//Sun Microsystems, Inc.//DTD Web Application 2.2//EN"
-    "http://java.sun.com/j2ee/dtds/web-app_2_2.dtd">
-
-<web-app>
      
     <servlet>
         <servlet-name>Butterfly</servlet-name>
@@ -17,4 +13,13 @@
         <url-pattern>/*</url-pattern>
     </servlet-mapping>
     
+    <security-constraint>
+        <web-resource-collection>
+            <url-pattern>/expire_locks</url-pattern>
+        </web-resource-collection>
+        <auth-constraint>
+            <role-name>admin</role-name>
+        </auth-constraint>
+    </security-constraint>
+        
 </web-app>

broker/appengine/src/com/metaweb/gridworks/broker/AppEngineGridworksBroker.java

@@ -4,6 +4,7 @@ import java.io.Writer;
 import java.util.ArrayList;
 import java.util.List;
 
+import javax.jdo.Extent;
 import javax.jdo.JDOHelper;
 import javax.jdo.PersistenceManager;
 import javax.jdo.PersistenceManagerFactory;
@@ -53,6 +54,39 @@ public class AppEngineGridworksBroker extends GridworksBroker {
     
     // ---------------------------------------------------------------------------------
     
+    protected void expireLocks(HttpServletResponse response) throws Exception {
+        
+        PersistenceManager pm = pmfInstance.getPersistenceManager();
+        
+        try {
+            Extent<Lock> extent = pm.getExtent(Lock.class, false);
+
+            try {
+                for (Lock lock : extent) {
+                    if (lock.timestamp + LOCK_DURATION < System.currentTimeMillis()) {
+                        Transaction tx = pm.currentTransaction();
+                        try {
+                            tx.begin();
+                            pm.deletePersistent(lock);
+                            tx.commit();
+                        } finally {
+                            if (tx.isActive()) {
+                                tx.rollback();
+                            }
+                        }
+                    }
+                }
+            } finally {
+                extent.closeAll();                
+            }            
+            
+            respond(response, OK);
+            
+        } finally {
+            pm.close();
+        }
+    }
+    
     protected void getLock(HttpServletResponse response, String pid) throws Exception {
         PersistenceManager pm = pmfInstance.getPersistenceManager();
         
@@ -295,6 +329,7 @@ public class AppEngineGridworksBroker extends GridworksBroker {
             o.put("lock_id", lock.id);
             o.put("project_id", lock.pid);
             o.put("user_id", lock.uid);
+            o.put("timestamp", lock.timestamp);
         }
         return o;
     }
@@ -312,10 +347,14 @@ public class AppEngineGridworksBroker extends GridworksBroker {
         @Persistent
         String uid;
         
+        @Persistent
+        long timestamp;
+        
         Lock(String id, String pid, String uid) {
             this.id = id;
             this.pid = pid;
             this.uid = uid;
+            this.timestamp = System.currentTimeMillis();
         }
     }
     

broker/core/src/com/metaweb/gridworks/broker/GridworksBroker.java

@@ -62,6 +62,9 @@ public abstract class GridworksBroker extends ButterflyModuleImpl {
         }
     }
     
+    static public final long LOCK_DURATION = 60 * 1000; // 1 minute
+    static public final long LOCK_EXPIRATION_CHECK_DELAY = 5 * 1000; // 5 seconds
+    
     protected HttpClient httpclient;
 
     @Override
@@ -98,6 +101,8 @@ public abstract class GridworksBroker extends ButterflyModuleImpl {
             
             if ("get_lock".equals(path)) {
                 getLock(response, pid);
+            } else if ("expire_locks".equals(path)) {
+                expireLocks(response);
             } else if ("obtain_lock".equals(path)) {
                 obtainLock(response, pid, uid);
             } else if ("release_lock".equals(path)) {
@@ -129,6 +134,8 @@ public abstract class GridworksBroker extends ButterflyModuleImpl {
     
     protected abstract HttpClient getHttpClient();
 
+    protected abstract void expireLocks(HttpServletResponse response) throws Exception;
+    
     protected abstract void getLock(HttpServletResponse response, String pid) throws Exception;
 
     protected abstract void obtainLock(HttpServletResponse response, String pid, String uid) throws Exception;

broker/local/src/com/metaweb/gridworks/broker/LocalGridworksBroker.java

@@ -4,6 +4,8 @@ import java.io.File;
 import java.io.Writer;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Timer;
+import java.util.TimerTask;
 
 import javax.servlet.ServletConfig;
 import javax.servlet.http.HttpServletResponse;
@@ -36,12 +38,17 @@ public class LocalGridworksBroker extends GridworksBroker {
     PrimaryIndex<String,Project> projectById;
     PrimaryIndex<String,Lock> lockByProject;
 
-    protected HttpClient httpclient;
+    Timer timer;
+    LockExpirer expirer;
     
     @Override
     public void init(ServletConfig config) throws Exception {
         super.init(config);
 
+        timer = new Timer();
+        expirer = new LockExpirer();
+        timer.schedule(expirer, LOCK_EXPIRATION_CHECK_DELAY, LOCK_EXPIRATION_CHECK_DELAY);
+        
         File dataPath = new File("data"); // FIXME: data should be configurable;
 
         EnvironmentConfig envConfig = new EnvironmentConfig();
@@ -77,6 +84,20 @@ public class LocalGridworksBroker extends GridworksBroker {
         }
     }
 
+    class LockExpirer extends TimerTask {
+        public void run() {
+            for (Lock lock : lockByProject.entities()) {
+                if (lock.timestamp + LOCK_DURATION < System.currentTimeMillis()) {
+                    try {
+                        releaseLock(null, lock.pid, lock.uid, lock.id);
+                    } catch (Exception e) {
+                        logger.error("Exception while expiring lock for project '" + lock.pid + "'", e);
+                    }
+                }
+            }
+        }
+    }
+    
     // ---------------------------------------------------------------------------------
 
     protected HttpClient getHttpClient() {
@@ -85,6 +106,11 @@ public class LocalGridworksBroker extends GridworksBroker {
     
     // ---------------------------------------------------------------------------------
     
+    protected void expireLocks(HttpServletResponse response) throws Exception {
+        expirer.run();
+        respond(response, OK);
+    }
+    
     protected void getLock(HttpServletResponse response, String pid) throws Exception {
         respond(response, lockToJSON(getLock(pid)));
     }
@@ -135,8 +161,10 @@ public class LocalGridworksBroker extends GridworksBroker {
             }
         }
                 
+        if (response != null) { // this because the expiration thread can call this method without a real response
             respond(response, OK);
         }
+    }
     
     // ----------------------------------------------------------------------------------------------------
     
@@ -283,6 +311,7 @@ public class LocalGridworksBroker extends GridworksBroker {
             o.put("lock_id", lock.id);
             o.put("project_id", lock.pid);
             o.put("user_id", lock.uid);
+            o.put("timestamp", lock.timestamp);
         }
         return o;
     }
@@ -297,10 +326,13 @@ public class LocalGridworksBroker extends GridworksBroker {
         
         String uid;
         
+        long timestamp;
+        
         Lock(String id, String pid, String uid) {
             this.id = id;
             this.pid = pid;
             this.uid = uid;
+            this.timestamp = System.currentTimeMillis();
         }
                 
         @SuppressWarnings("unused")