The protocol between the server and the shuffler is quite simple, one * byte indicate the operation, which takes as argument two file descriptors. */ public class Shuffler { /** * The property giving the path of the shuffler server. * The shuffler server is an optional server helper, that deals with * serving resource contents. When resource contents can be efficiently * messaged between process boundaries (eg using sendmsg), the shuffler * server takes over the task of sending resource's content back to the * client. This property gives the path of the shuffler server binary * program. */ public static final String SHUFFLER_P = "org.w3c.jigsaw.shuffler" ; private static Process shuffler = null ; private static boolean inited = false ; private ShufflerThread waiter = null ; private int fd = -1 ; private native int initialize (String path) ; private native synchronized int shuffle (ShuffleHandler h) ; private native int getNextMessage (ShuffleMessage msg) ; ShuffleMessage getNextMessage() { ShuffleMessage msg = new ShuffleMessage () ; int duration = 2 ; while ( true ) { int ecode = getNextMessage (msg) ; if ( duration < 250 ) duration = (duration << 1) ; if ( ecode > 0 ) { return msg ; } else if ( ecode == 0 ) { // yield and retyr (yes, this *is* pooling :=( try { Thread.sleep (duration) ; } catch (InterruptedException e) { } Thread.yield() ; } else if ( ecode < 0 ) { String m = (this.getClass().getName() + "[getNextMessage]: failed (e="+ecode+")") ; throw new RuntimeException (m) ; } } } /** * Initialize this class. * This deserve a special method, since we want any exception to be * caught when invoking theinstance constructor. *
This method tries to launch the shuffler process automatically.
* @param path The driectory for UNIX socket bindings.
* @return A boolean true is every thing went fine,
* false otherwise.
*/
private synchronized boolean classInitialize (String path) {
File socket = new File (path + "/shuffler") ;
// Delete any old socket for this shuffler:
socket.delete() ;
// Load the native code:
Runtime.getRuntime().loadLibrary ("Shuffle") ;
inited = true ;
// Get the shuffler binary path:
String shuffler_bin = System.getProperty (SHUFFLER_P) ;
if ( shuffler_bin == null )
return false ;
// Run it:
try {
String args[] = new String[2] ;
args[0] = shuffler_bin ;
args[1] = path + "/shuffler" ;
// This is intended for debug only, it makes the shuffler emit traces
// in the provided (-log) log file
// args[2] = "-v";
// args[3] = "-log";
// args[4] = "/nfs/usr/abaird/Jigsaw/logs/shuffler";
shuffler = Runtime.getRuntime().exec (args) ;
} catch (Exception e) {
throw new RuntimeException (this.getClass().getName()
+ "[classInitialize]: "
+ "unable to launch shuffler.") ;
}
// Wait for the shuffler to create its listening socket:
int timeout = 10000 ;
while ( (timeout > 0) && (! socket.exists()) ) {
timeout -= 500 ;
try {
Thread.sleep (500) ;
} catch (InterruptedException e) {
}
}
if ( ! socket.exists() ) {
throw new RuntimeException (this.getClass().getName()
+ "[classInitialize]: "
+ " didn't create its socket.");
}
return true ;
}
private int shuffle (FileDescriptor in, FileDescriptor out)
throws IOException
{
ShuffleHandler handle = new ShuffleHandler (in, out) ;
// WARNING: code below shouldn't be changed it contains black magic
// The thing is that the shuffler is really fast, and can even
// finish its job before the waiter gets a chance to register the
// appropriate identifier (so it gets a reply for an identifier it
// doesn't know about).
// Synchronizing the waiter makes the 'shuffle' + 'register'
// operations atomic with regard to the waiter thread, which is
// what we want.
// If you have understood above comment, then there is no more black
// magic for you below.
synchronized (waiter) {
if ( (handle.id = shuffle (handle)) < 0 )
throw new IOException (this.getClass().getName()
+ " unable to shuffle !") ;
waiter.registerHandler (handle) ;
}
return handle.waitForCompletion () ;
}
/**
* Shuffle the given rteply body to the given client.
* This methods tries to outout the given reply
*/
public int shuffle (Client client, Reply reply)
throws IOException
{
FileDescriptor in = reply.getInputFileDescriptor () ;
if ( in == null )
return -1 ;
// client.flushOutput() ;
// FileDescriptor out = client.getOutputFileDescriptor() ;
// int written = shuffle (in, out) ;
// if ( written < 0 )
// throw new IOException ("Shuffler failure.") ;
// return written ;
return -1;
}
public synchronized void shutdown() {
// Kill the shuffler process (if needed) and waitfor its completion
if ( shuffler != null ) {
shuffler.destroy() ;
while ( true ) {
try {
shuffler.waitFor() ;
break ;
} catch (InterruptedException ex) {
}
}
shuffler = null ;
}
// Un-initialize, stop the waiter thread:
inited = false ;
waiter.stop() ;
waiter = null ;
}
/**
* Create a new data shuffler.
* The path identifies the directory in which UNIX socket will get bind.
* This should be an absloute path, eg /tmp/shuffler.
* @param path The path to the server.
*/
public Shuffler (String path) {
if ( ((! inited) && ( ! classInitialize (path)))
|| (initialize (path) < 0) ) {
throw new RuntimeException (this.getClass().getName()
+ ": unable to connect to shuffler "
+ path) ;
}
this.waiter = new ShufflerThread (this) ;
this.waiter.start() ;
}
// testing only
public static void main (String args[])
throws FileNotFoundException, IOException
{
Shuffler s = new Shuffler(args[0]) ;
FileInputStream f = new FileInputStream ("from") ;
FileOutputStream t = new FileOutputStream ("to") ;
s.shuffle (f.getFD(), t.getFD()) ;
f.close() ;
t.close() ;
}
}