ojluni/src/main/java/sun/nio/fs/AbstractPoller.java - platform/libcore.git - Git at Google

 /*
  * Copyright (c) 2008, 2011, Oracle and/or its affiliates. All rights reserved.
  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
  *
  * This code is free software; you can redistribute it and/or modify it
  * under the terms of the GNU General Public License version 2 only, as
  * published by the Free Software Foundation.  Oracle designates this
  * particular file as subject to the "Classpath" exception as provided
  * by Oracle in the LICENSE file that accompanied this code.
  *
  * This code is distributed in the hope that it will be useful, but WITHOUT
  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
  * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
  * version 2 for more details (a copy is included in the LICENSE file that
  * accompanied this code).
  *
  * You should have received a copy of the GNU General Public License version
  * 2 along with this work; if not, write to the Free Software Foundation,
  * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
  *
  * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
  * or visit www.oracle.com if you need additional information or have any
  * questions.
  */

 package sun.nio.fs;

 import java.nio.file.*;
 import java.security.AccessController;
 import java.security.PrivilegedAction;
 import java.io.IOException;
 import java.util.*;

 /**
  * Base implementation of background poller thread used in watch service
  * implementations. A poller thread waits on events from the file system and
  * also services "requests" from clients to register for new events or cancel
  * existing registrations.
  */

 abstract class AbstractPoller implements Runnable {

     // list of requests pending to the poller thread
     private final LinkedList<Request> requestList;

     // set to true when shutdown
     private boolean shutdown;

     protected AbstractPoller() {
         this.requestList = new LinkedList<Request>();
         this.shutdown = false;
     }

     /**
      * Starts the poller thread
      */
     public void start() {
         final Runnable thisRunnable = this;
         AccessController.doPrivileged(new PrivilegedAction<Object>() {
             @Override
             public Object run() {
                 Thread thr = new Thread(thisRunnable);
                 thr.setDaemon(true);
                 thr.start();
                 return null;
             }
          });
     }

     /**
      * Wakeup poller thread so that it can service pending requests
      */
     abstract void wakeup() throws IOException;

     /**
      * Executed by poller thread to register directory for changes
      */
     abstract Object implRegister(Path path,
                                  Set<? extends WatchEvent.Kind<?>> events,
                                  WatchEvent.Modifier... modifiers);

     /**
      * Executed by poller thread to cancel key
      */
     abstract void implCancelKey(WatchKey key);

     /**
      * Executed by poller thread to shutdown and cancel all keys
      */
     abstract void implCloseAll();

     /**
      * Requests, and waits on, poller thread to register given file.
      */
     final WatchKey register(Path dir,
                             WatchEvent.Kind<?>[] events,
                             WatchEvent.Modifier... modifiers)
         throws IOException
     {
         // validate arguments before request to poller
         if (dir == null)
             throw new NullPointerException();
         if (events.length == 0)
             throw new IllegalArgumentException("No events to register");
         Set<WatchEvent.Kind<?>> eventSet = new HashSet<>(events.length);
         for (WatchEvent.Kind<?> event: events) {
             // standard events
             if (event == StandardWatchEventKinds.ENTRY_CREATE ||
                 event == StandardWatchEventKinds.ENTRY_MODIFY ||
                 event == StandardWatchEventKinds.ENTRY_DELETE)
             {
                 eventSet.add(event);
                 continue;
             }

             // OVERFLOW is ignored
             if (event == StandardWatchEventKinds.OVERFLOW) {
                 if (events.length == 1)
                     throw new IllegalArgumentException("No events to register");
                 continue;
             }

             // null/unsupported
             if (event == null)
                 throw new NullPointerException("An element in event set is 'null'");
             throw new UnsupportedOperationException(event.name());
         }
         return (WatchKey)invoke(RequestType.REGISTER, dir, eventSet, modifiers);
     }

     /**
      * Cancels, and waits on, poller thread to cancel given key.
      */
     final void cancel(WatchKey key) {
         try {
             invoke(RequestType.CANCEL, key);
         } catch (IOException x) {
             // should not happen
             throw new AssertionError(x.getMessage());
         }
     }

     /**
      * Shutdown poller thread
      */
     final void close() throws IOException {
         invoke(RequestType.CLOSE);
     }

     /**
      * Types of request that the poller thread must handle
      */
     private static enum RequestType {
         REGISTER,
         CANCEL,
         CLOSE;
     }

     /**
      * Encapsulates a request (command) to the poller thread.
      */
     private static class Request {
         private final RequestType type;
         private final Object[] params;

         private boolean completed = false;
         private Object result = null;

         Request(RequestType type, Object... params) {
             this.type = type;
             this.params = params;
         }

         RequestType type() {
             return type;
         }

         Object[] parameters() {
             return params;
         }

         void release(Object result) {
             synchronized (this) {
                 this.completed = true;
                 this.result = result;
                 notifyAll();
             }
         }

         /**
          * Await completion of the request. The return value is the result of
          * the request.
          */
         Object awaitResult() {
             synchronized (this) {
                 while (!completed) {
                     try {
                         wait();
                     } catch (InterruptedException x) {
                         // ignore
                     }
                 }
                 return result;
             }
         }
     }

     /**
      * Enqueues request to poller thread and waits for result
      */
     private Object invoke(RequestType type, Object... params) throws IOException {
         // submit request
         Request req = new Request(type, params);
         synchronized (requestList) {
             if (shutdown) {
                 throw new ClosedWatchServiceException();
             }
             requestList.add(req);
         }

         // wakeup thread
         wakeup();

         // wait for result
         Object result = req.awaitResult();

         if (result instanceof RuntimeException)
             throw (RuntimeException)result;
         if (result instanceof IOException )
             throw (IOException)result;
         return result;
     }

     /**
      * Invoked by poller thread to process all pending requests
      *
      * @return  true if poller thread should shutdown
      */
     @SuppressWarnings("unchecked")
     boolean processRequests() {
         synchronized (requestList) {
             Request req;
             while ((req = requestList.poll()) != null) {
                 // if in process of shutdown then reject request
                 if (shutdown) {
                     req.release(new ClosedWatchServiceException());
                 }

                 switch (req.type()) {
                     /**
                      * Register directory
                      */
                     case REGISTER: {
                         Object[] params = req.parameters();
                         Path path = (Path)params[0];
                         Set<? extends WatchEvent.Kind<?>> events =
                             (Set<? extends WatchEvent.Kind<?>>)params[1];
                         WatchEvent.Modifier[] modifiers =
                             (WatchEvent.Modifier[])params[2];
                         req.release(implRegister(path, events, modifiers));
                         break;
                     }
                     /**
                      * Cancel existing key
                      */
                     case CANCEL : {
                         Object[] params = req.parameters();
                         WatchKey key = (WatchKey)params[0];
                         implCancelKey(key);
                         req.release(null);
                         break;
                     }
                     /**
                      * Close watch service
                      */
                     case CLOSE: {
                         implCloseAll();
                         req.release(null);
                         shutdown = true;
                         break;
                     }

                     default:
                         req.release(new IOException("request not recognized"));
                 }
             }
         }
         return shutdown;
     }
 }
	/*
	* Copyright (c) 2008, 2011, Oracle and/or its affiliates. All rights reserved.
	* DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
	*
	* This code is free software; you can redistribute it and/or modify it
	* under the terms of the GNU General Public License version 2 only, as
	* published by the Free Software Foundation. Oracle designates this
	* particular file as subject to the "Classpath" exception as provided
	* by Oracle in the LICENSE file that accompanied this code.
	*
	* This code is distributed in the hope that it will be useful, but WITHOUT
	* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
	* FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
	* version 2 for more details (a copy is included in the LICENSE file that
	* accompanied this code).
	*
	* You should have received a copy of the GNU General Public License version
	* 2 along with this work; if not, write to the Free Software Foundation,
	* Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
	*
	* Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
	* or visit www.oracle.com if you need additional information or have any
	* questions.
	*/

	package sun.nio.fs;

	import java.nio.file.*;
	import java.security.AccessController;
	import java.security.PrivilegedAction;
	import java.io.IOException;
	import java.util.*;

	/**
	* Base implementation of background poller thread used in watch service
	* implementations. A poller thread waits on events from the file system and
	* also services "requests" from clients to register for new events or cancel
	* existing registrations.
	*/

	abstract class AbstractPoller implements Runnable {

	// list of requests pending to the poller thread
	private final LinkedList<Request> requestList;

	// set to true when shutdown
	private boolean shutdown;

	protected AbstractPoller() {
	this.requestList = new LinkedList<Request>();
	this.shutdown = false;
	}

	/**
	* Starts the poller thread
	*/
	public void start() {
	final Runnable thisRunnable = this;
	AccessController.doPrivileged(new PrivilegedAction<Object>() {
	@Override
	public Object run() {
	Thread thr = new Thread(thisRunnable);
	thr.setDaemon(true);
	thr.start();
	return null;
	}
	});
	}

	/**
	* Wakeup poller thread so that it can service pending requests
	*/
	abstract void wakeup() throws IOException;

	/**
	* Executed by poller thread to register directory for changes
	*/
	abstract Object implRegister(Path path,
	Set<? extends WatchEvent.Kind<?>> events,
	WatchEvent.Modifier... modifiers);

	/**
	* Executed by poller thread to cancel key
	*/
	abstract void implCancelKey(WatchKey key);

	/**
	* Executed by poller thread to shutdown and cancel all keys
	*/
	abstract void implCloseAll();

	/**
	* Requests, and waits on, poller thread to register given file.
	*/
	final WatchKey register(Path dir,
	WatchEvent.Kind<?>[] events,
	WatchEvent.Modifier... modifiers)
	throws IOException
	{
	// validate arguments before request to poller
	if (dir == null)
	throw new NullPointerException();
	if (events.length == 0)
	throw new IllegalArgumentException("No events to register");
	Set<WatchEvent.Kind<?>> eventSet = new HashSet<>(events.length);
	for (WatchEvent.Kind<?> event: events) {
	// standard events
	if (event == StandardWatchEventKinds.ENTRY_CREATE \|\|
	event == StandardWatchEventKinds.ENTRY_MODIFY \|\|
	event == StandardWatchEventKinds.ENTRY_DELETE)
	{
	eventSet.add(event);
	continue;
	}

	// OVERFLOW is ignored
	if (event == StandardWatchEventKinds.OVERFLOW) {
	if (events.length == 1)
	throw new IllegalArgumentException("No events to register");
	continue;
	}

	// null/unsupported
	if (event == null)
	throw new NullPointerException("An element in event set is 'null'");
	throw new UnsupportedOperationException(event.name());
	}
	return (WatchKey)invoke(RequestType.REGISTER, dir, eventSet, modifiers);
	}

	/**
	* Cancels, and waits on, poller thread to cancel given key.
	*/
	final void cancel(WatchKey key) {
	try {
	invoke(RequestType.CANCEL, key);
	} catch (IOException x) {
	// should not happen
	throw new AssertionError(x.getMessage());
	}
	}

	/**
	* Shutdown poller thread
	*/
	final void close() throws IOException {
	invoke(RequestType.CLOSE);
	}

	/**
	* Types of request that the poller thread must handle
	*/
	private static enum RequestType {
	REGISTER,
	CANCEL,
	CLOSE;
	}

	/**
	* Encapsulates a request (command) to the poller thread.
	*/
	private static class Request {
	private final RequestType type;
	private final Object[] params;

	private boolean completed = false;
	private Object result = null;

	Request(RequestType type, Object... params) {
	this.type = type;
	this.params = params;
	}

	RequestType type() {
	return type;
	}

	Object[] parameters() {
	return params;
	}

	void release(Object result) {
	synchronized (this) {
	this.completed = true;
	this.result = result;
	notifyAll();
	}
	}

	/**
	* Await completion of the request. The return value is the result of
	* the request.
	*/
	Object awaitResult() {
	synchronized (this) {
	while (!completed) {
	try {
	wait();
	} catch (InterruptedException x) {
	// ignore
	}
	}
	return result;
	}
	}
	}

	/**
	* Enqueues request to poller thread and waits for result
	*/
	private Object invoke(RequestType type, Object... params) throws IOException {
	// submit request
	Request req = new Request(type, params);
	synchronized (requestList) {
	if (shutdown) {
	throw new ClosedWatchServiceException();
	}
	requestList.add(req);
	}

	// wakeup thread
	wakeup();

	// wait for result
	Object result = req.awaitResult();

	if (result instanceof RuntimeException)
	throw (RuntimeException)result;
	if (result instanceof IOException )
	throw (IOException)result;
	return result;
	}

	/**
	* Invoked by poller thread to process all pending requests
	*
	* @return true if poller thread should shutdown
	*/
	@SuppressWarnings("unchecked")
	boolean processRequests() {
	synchronized (requestList) {
	Request req;
	while ((req = requestList.poll()) != null) {
	// if in process of shutdown then reject request
	if (shutdown) {
	req.release(new ClosedWatchServiceException());
	}

	switch (req.type()) {
	/**
	* Register directory
	*/
	case REGISTER: {
	Object[] params = req.parameters();
	Path path = (Path)params[0];
	Set<? extends WatchEvent.Kind<?>> events =
	(Set<? extends WatchEvent.Kind<?>>)params[1];
	WatchEvent.Modifier[] modifiers =
	(WatchEvent.Modifier[])params[2];
	req.release(implRegister(path, events, modifiers));
	break;
	}
	/**
	* Cancel existing key
	*/
	case CANCEL : {
	Object[] params = req.parameters();
	WatchKey key = (WatchKey)params[0];
	implCancelKey(key);
	req.release(null);
	break;
	}
	/**
	* Close watch service
	*/
	case CLOSE: {
	implCloseAll();
	req.release(null);
	shutdown = true;
	break;
	}

	default:
	req.release(new IOException("request not recognized"));
	}
	}
	}
	return shutdown;
	}
	}