Alfresco repository filesystem oplocks implementation

Oplock support can be switched off using the 'disableOplocks' property

git-svn-id: https://svn.alfresco.com/repos/alfresco-enterprise/alfresco/HEAD/root@18116 c4b6b30b-aa2e-2d43-bbcb-ca4b014f7261

source/java/org/alfresco/filesys/AbstractServerConfigurationBean.java

@@ -92,7 +92,7 @@ public abstract class AbstractServerConfigurationBean extends ServerConfiguratio
   
   protected static final String m_sessDbgStr[] = { "NETBIOS", "STATE", "RXDATA", "TXDATA", "DUMPDATA", "NEGOTIATE", "TREE", "SEARCH", "INFO", "FILE",
           "FILEIO", "TRANSACT", "ECHO", "ERROR", "IPC", "LOCK", "PKTTYPE", "DCERPC", "STATECACHE", "TIMING", "NOTIFY",
-          "STREAMS", "SOCKET", "PKTPOOL", "PKTSTATS", "THREADPOOL", "BENCHMARK" };
+          "STREAMS", "SOCKET", "PKTPOOL", "PKTSTATS", "THREADPOOL", "BENCHMARK", "OPLOCK" };
 
   // FTP server debug type strings
 

source/java/org/alfresco/filesys/repo/ContentContext.java

@@ -62,6 +62,9 @@ public class ContentContext extends AlfrescoContext
     
     private AccessControlListBean m_accessControlList;
         
+    // Enable/disable oplocks
+    
+    private boolean m_oplocksDisabled;
     
     // Node monitor
     
@@ -138,6 +141,15 @@ public class ContentContext extends AlfrescoContext
     }
 
 
+    /**
+     * Enable/disable oplock support
+     * 
+     * @param disableOplocks boolean
+     */
+    public void setDisableOplocks( boolean disableOplocks) {
+    	m_oplocksDisabled = disableOplocks;
+    }
+    
     @Override
     public void initialize(AlfrescoDiskDriver filesysDriver)
     {
@@ -215,6 +227,15 @@ public class ContentContext extends AlfrescoContext
         return m_disableNodeMonitor;
     }
 
+    /**
+     * Determine if oplocks support should be disabled
+     * 
+     * @return boolean
+     */
+    public boolean getDisableOplocks() {
+    	return m_oplocksDisabled;
+    }
+    
     /**
      * Gets the access control list.
      * 

source/java/org/alfresco/filesys/repo/ContentDiskDriver.java

@@ -64,6 +64,8 @@ import org.alfresco.jlan.server.filesys.pseudo.PseudoFileList;
 import org.alfresco.jlan.server.filesys.pseudo.PseudoNetworkFile;
 import org.alfresco.jlan.server.locking.FileLockingInterface;
 import org.alfresco.jlan.server.locking.LockManager;
+import org.alfresco.jlan.server.locking.OpLockInterface;
+import org.alfresco.jlan.server.locking.OpLockManager;
 import org.alfresco.jlan.smb.SharingMode;
 import org.alfresco.jlan.smb.WinNT;
 import org.alfresco.jlan.smb.server.SMBServer;
@@ -97,7 +99,7 @@ import org.apache.commons.logging.LogFactory;
  * 
  * @author Derek Hulley
  */
-public class ContentDiskDriver extends AlfrescoDiskDriver implements DiskInterface, FileLockingInterface
+public class ContentDiskDriver extends AlfrescoDiskDriver implements DiskInterface, FileLockingInterface, OpLockInterface
 {
     // Logging
     
@@ -130,7 +132,7 @@ public class ContentDiskDriver extends AlfrescoDiskDriver implements DiskInterfa
     
 	//	Lock manager
 	
-	private static LockManager _lockManager = new FileStateLockManager();
+	private static FileStateLockManager _lockManager = new FileStateLockManager();
     
     /**
      * Class constructor
@@ -430,6 +432,14 @@ public class ContentDiskDriver extends AlfrescoDiskDriver implements DiskInterfa
             context.setDisableNodeMonitor(true);
         }
         
+        // Check if oplocks are enabled, if so then enable oplocks in the lock manager
+        
+        if ( cfg.getChild("disableOplocks") != null) {
+        	context.setDisableOplocks( true);
+        }
+        
+        // Register the device context
+        
         registerContext(context);
         
         // Return the context for this shared filesystem
@@ -597,6 +607,17 @@ public class ContentDiskDriver extends AlfrescoDiskDriver implements DiskInterfa
             NodeMonitor nodeMonitor = m_nodeMonitorFactory.createNodeMonitor( this, context);
             context.setNodeMonitor( nodeMonitor);
         }
+        
+        // Check if oplocks are enabled
+        
+        if ( context.getDisableOplocks() == false) {
+                
+            // Enable oplock support
+            	
+            _lockManager.setStateTable( context.getStateTable());
+        }
+        else
+        	logger.warn("Oplock support disabled for filesystem " + ctx.getDeviceName());
     }
 
     /**
@@ -3087,4 +3108,30 @@ public class ContentDiskDriver extends AlfrescoDiskDriver implements DiskInterfa
 	public LockManager getLockManager(SrvSession sess, TreeConnection tree) {
 		return _lockManager;
 	}
+	
+	/**
+	 * Return the oplock manager implementation associated with this virtual filesystem
+	 * 
+	 * @param sess SrvSession
+	 * @param tree TreeConnection
+	 * @return OpLockManager
+	 */
+	public OpLockManager getOpLockManager(SrvSession sess, TreeConnection tree) {
+		return _lockManager;
+	}
+	
+	/**
+	 * Enable/disable oplock support
+	 * 
+	 * @param sess SrvSession
+	 * @param tree TreeConnection
+	 * @return boolean
+	 */
+	public boolean isOpLocksEnabled(SrvSession sess, TreeConnection tree) {
+		
+        // Check if oplocks are enabled
+        
+        ContentContext ctx = (ContentContext) tree.getContext();
+        return ctx.getDisableOplocks() ? false : true;
+	}
 }

source/java/org/alfresco/filesys/state/FileState.java

@@ -28,10 +28,12 @@ import org.alfresco.jlan.locking.FileLock;
 import org.alfresco.jlan.locking.FileLockList;
 import org.alfresco.jlan.locking.LockConflictException;
 import org.alfresco.jlan.locking.NotLockedException;
+import org.alfresco.jlan.server.filesys.ExistingOpLockException;
 import org.alfresco.jlan.server.filesys.FileOpenParams;
 import org.alfresco.jlan.server.filesys.FileStatus;
 import org.alfresco.jlan.server.filesys.pseudo.PseudoFile;
 import org.alfresco.jlan.server.filesys.pseudo.PseudoFileList;
+import org.alfresco.jlan.server.locking.OpLockDetails;
 import org.alfresco.jlan.smb.SharingMode;
 import org.alfresco.service.cmr.repository.NodeRef;
 import org.apache.commons.logging.Log;
@@ -84,6 +86,10 @@ public class FileState
 
     private FileLockList m_lockList;
     
+	// Oplock details
+
+	private OpLockDetails m_oplock;
+
     // Node for this file
     
     private NodeRef m_nodeRef;
@@ -704,6 +710,46 @@ public class FileState
         return writeOK;
     }
 
+	/**
+	 * Check if the file has an active oplock
+	 * 
+	 * @return boolean
+	 */
+	public final boolean hasOpLock() {
+		return m_oplock != null ? true : false;
+	}
+
+	/**
+	 * Return the oplock details
+	 * 
+	 * @return OpLockDetails
+	 */
+	public final OpLockDetails getOpLock() {
+		return m_oplock;
+	}
+
+	/**
+	 * Set the oplock for this file
+	 * 
+	 * @param oplock OpLockDetails
+	 * @exception ExistingOpLockException If there is an active oplock on this file
+	 */
+	public final synchronized void setOpLock(OpLockDetails oplock)
+		throws ExistingOpLockException {
+
+		if ( m_oplock == null)
+			m_oplock = oplock;
+		else
+			throw new ExistingOpLockException();
+	}
+
+	/**
+	 * Clear the oplock
+	 */
+	public final synchronized void clearOpLock() {
+		m_oplock = null;
+	}
+
     /**
      * Normalize the path to uppercase the directory names and keep the case of the file name.
      * 
@@ -756,6 +802,12 @@ public class FileState
             else
                 str.append(0);
         }
+  	  
+        if ( hasOpLock()) {
+        	str.append(",OpLock=");
+        	str.append(getOpLock());
+        }
+		
         str.append("]");
 
         return str.toString();

source/java/org/alfresco/filesys/state/FileStateLockManager.java

@@ -25,15 +25,25 @@
 package org.alfresco.filesys.state;
 
 import java.io.IOException;
+import java.util.Enumeration;
+import java.util.Hashtable;
 
+import org.alfresco.jlan.debug.Debug;
 import org.alfresco.jlan.locking.FileLock;
 import org.alfresco.jlan.locking.LockConflictException;
 import org.alfresco.jlan.locking.NotLockedException;
 import org.alfresco.jlan.server.SrvSession;
+import org.alfresco.jlan.server.filesys.ExistingOpLockException;
 import org.alfresco.jlan.server.filesys.NetworkFile;
 import org.alfresco.jlan.server.filesys.TreeConnection;
 import org.alfresco.jlan.server.filesys.pseudo.MemoryNetworkFile;
 import org.alfresco.jlan.server.locking.LockManager;
+import org.alfresco.jlan.server.locking.OpLockDetails;
+import org.alfresco.jlan.server.locking.OpLockManager;
+import org.alfresco.jlan.smb.OpLock;
+import org.alfresco.jlan.smb.SMBStatus;
+import org.alfresco.jlan.smb.server.SMBSrvPacket;
+import org.alfresco.jlan.smb.server.SMBSrvSession;
 import org.alfresco.filesys.alfresco.AlfrescoNetworkFile;
 
 /**
@@ -43,7 +53,24 @@ import org.alfresco.filesys.alfresco.AlfrescoNetworkFile;
  * 
  * @author gkspencer
  */
-public class FileStateLockManager implements LockManager {
+public class FileStateLockManager implements LockManager, OpLockManager, Runnable {
+
+	// Oplock break timeout
+	
+	private static final long OpLockBreakTimeout	= 5000L;	// 5 seconds
+	
+	// File state cache used for byte range locks/oplocks
+	
+	private FileStateTable m_stateCache;
+	
+	// Oplock breaks in progress
+	
+	private Hashtable<String, OpLockDetails> m_oplockQueue;
+	
+	// Oplock break timeout thread
+	
+	private Thread m_expiryThread;
+	private boolean m_shutdown;
 	
 	/**
 	 * Lock a byte range within a file, or the whole file. 
@@ -176,4 +203,294 @@ public class FileStateLockManager implements LockManager {
 			}
 		}
 	}
+	
+	/**
+	 * Enable oplock support by setting the file state table
+	 * 
+	 * @param stateTable FileStateTable
+	 */
+	public final void setStateTable(FileStateTable stateTable) {
+		
+		m_stateCache = stateTable;
+		
+		// Create the oplock break queue
+		
+		m_oplockQueue = new Hashtable<String, OpLockDetails>();
+		
+		// Start the oplock break expiry thread
+		
+        m_expiryThread = new Thread(this);
+        m_expiryThread.setDaemon(true);
+        m_expiryThread.setName("OpLockExpire");
+        m_expiryThread.start();
+}
+	
+	/**
+	 * Check if there is an oplock for the specified path, return the oplock type.
+	 * 
+	 * @param path String
+	 * @return int
+	 */
+	public int hasOpLock(String path) {
+		
+		// Check if oplocks/state cache are enabled
+		
+		if ( m_stateCache == null)
+			return OpLock.TypeNone;
+		
+		// Get the file state
+		
+		FileState fstate = m_stateCache.findFileState(path);
+		if ( fstate != null && fstate.hasOpLock()) {
+		
+			// Return the oplock type
+			
+			OpLockDetails oplock = fstate.getOpLock();
+			if ( oplock != null)
+				return oplock.getLockType();
+		}
+		
+		// No oplock
+		
+		return OpLock.TypeNone;
+	}
+	
+	/**
+	 * Return the oplock details for a path, or null if there is no oplock on the path
+	 * 
+	 * @param path String
+	 * @return OpLockDetails
+	 */
+	public OpLockDetails getOpLockDetails(String path) {
+		
+		// Check if oplocks/state cache are enabled
+		
+		if ( m_stateCache == null)
+			return null;
+		
+		// Get the file state
+		
+		FileState fstate = m_stateCache.findFileState(path);
+		if ( fstate != null)
+			return fstate.getOpLock();
+
+		// No oplock
+		
+		return null;
+	}
+	
+	/**
+	 * Grant an oplock, store the oplock details
+	 * 
+	 * @param path String
+	 * @param oplock OpLockDetails
+	 * @return boolean
+	 * @exception ExistingOpLockException	If the file already has an oplock
+	 */
+	public boolean grantOpLock(String path, OpLockDetails oplock)
+		throws ExistingOpLockException {
+
+		// Check if oplocks/state cache are enabled
+		
+		if ( m_stateCache == null)
+			return false;
+		
+		// Get, or create, a file state
+		
+		FileState fstate = m_stateCache.findFileState(path, false, true);
+		
+		// Check if the file is already in use
+		
+		if ( fstate.getOpenCount() != 1)
+			return false;
+
+		// Set the oplock
+		
+		fstate.setOpLock( oplock);
+		return true;
+	}
+	
+	/**
+	 * Inform the oplock manager that an oplock break is in progress for the specified file/oplock
+	 * 
+	 * @param path String
+	 * @param oplock OpLockDetails
+	 */
+	public void informOpLockBreakInProgress(String path, OpLockDetails oplock) {
+		
+		// Check if oplocks/state cache are enabled
+		
+		if ( m_stateCache == null)
+			return;
+		
+		// Add the oplock to the break in progress queue
+		
+		synchronized ( m_oplockQueue) {
+			m_oplockQueue.put( path, oplock);
+			m_oplockQueue.notify();
+		}
+	}
+	
+	/**
+	 * Release an oplock
+	 * 
+	 * @param path String
+	 */
+	public void releaseOpLock(String path) {
+		
+		// Check if oplocks/state cache are enabled
+		
+		if ( m_stateCache == null)
+			return;
+		
+		// Get the file state
+		
+		FileState fstate = m_stateCache.findFileState(path);
+		if ( fstate != null)
+			fstate.clearOpLock();
+		
+		// Remove from the pending oplock break queue
+		
+		synchronized ( m_oplockQueue) {
+			m_oplockQueue.remove( path);
+		}
+	}
+	
+	/**
+	 * Check for expired oplock break requests
+	 * 
+	 * @return int
+	 */
+	public int checkExpiredOplockBreaks() {
+		
+		// Check if there are ny oplock breaks in progress
+		
+		if ( m_oplockQueue.size() == 0)
+			return 0;
+		
+		// Check for oplock break requests that have expired
+
+		int expireCnt = 0;
+		
+		long timeNow = System.currentTimeMillis();
+		Enumeration<String> opBreakKeys = m_oplockQueue.keys();
+		
+		while ( opBreakKeys.hasMoreElements()) {
+			
+			// Check the current oplock break
+			
+			String path = opBreakKeys.nextElement();
+			OpLockDetails opLock = m_oplockQueue.get( path);
+			if ( opLock != null) {
+				
+				// Check if the oplock break has timed out
+				
+				if ( opLock.hasDeferredSession() && (opLock.getOplockBreakTime() + OpLockBreakTimeout) <= timeNow) {
+					
+					// Get the deferred request details
+					
+					SMBSrvSession sess = opLock.getDeferredSession();
+					SMBSrvPacket  pkt  = opLock.getDeferredPacket();
+					
+					try {
+						
+						// Return an error for the deferred file open request
+						
+						if ( sess.sendAsyncErrorResponseSMB( pkt, SMBStatus.NTAccessDenied, SMBStatus.NTErr) == true) {
+						
+							// DEBUG
+							
+							if ( Debug.EnableDbg && sess.hasDebug( SMBSrvSession.DBG_OPLOCK))
+								sess.debugPrintln( "Oplock break timeout, oplock=" + opLock);
+							
+							// Release the packet back to the pool
+							
+							sess.getPacketPool().releasePacket( pkt);
+						}
+						else if ( Debug.EnableDbg && sess.hasDebug( SMBSrvSession.DBG_OPLOCK))
+							sess.debugPrintln( "Failed to send open reject, oplock break timed out, oplock=" + opLock);
+					}
+					catch ( IOException ex) {
+						
+					}
+					
+					// Remove the oplock break from the queue
+					
+					m_oplockQueue.remove( path);
+					
+					// Clear the deferred packet details
+					
+					opLock.clearDeferredSession();
+					
+					// Mark the oplock has having a failed oplock break
+					
+					opLock.setOplockBreakFailed();
+					
+					// Update the expired oplock break count
+					
+					expireCnt++;
+				}
+			}
+		}
+		
+		// Return the count of expired oplock breaks
+		
+		return expireCnt;
+	}
+	
+	/**
+	 * Run the oplock break expiry
+	 */
+    public void run()
+    {
+        // Loop forever
+
+    	m_shutdown = false;
+    	
+        while ( m_shutdown == false)
+        {
+            // Wait for an oplock break or sleep for a while if there are active oplock break requests
+
+        	try
+            {
+        		synchronized ( m_oplockQueue) {
+        			if ( m_oplockQueue.size() == 0)
+        				m_oplockQueue.wait();
+        		}
+        		
+        		// Oplock break added to the queue, wait a while before checking the queue
+        		
+        		if ( m_oplockQueue.size() > 0)
+        			Thread.sleep( OpLockBreakTimeout);
+            }
+            catch (InterruptedException ex)
+            {
+            }
+
+            //	Check for shutdown
+            
+            if ( m_shutdown == true)
+            	return;
+            
+            // Check for expired oplock break requests
+            
+            checkExpiredOplockBreaks();
+        }
+    }
+
+	/**
+	 * Request the oplock break expiry thread to shutdown
+	 */
+	public final void shutdownRequest() {
+		m_shutdown = true;
+		
+		if ( m_expiryThread != null)
+		{
+			try {
+				m_expiryThread.interrupt();
+			}
+			catch (Exception ex) {
+			}
+		}
+	}
 }