Modified streams such that their timeouts are absolute times.

     Fixed a bug in TimedWait that was due to the recent changes.

7 files changed, 59 insertions, 19 deletions

diff --git a/java/src/Stream.java b/java/src/Stream.java
index 030114d092f..3c67efef220 100644
--- a/java/src/Stream.java
+++ b/java/src/Stream.java
@@ -62,11 +62,13 @@ public class Stream
       return 0;
   }
 
+    // Note that the timeout tv is absolute time
   public int put (MessageBlock mb, TimeValue tv)
   {
     return this.streamHead_.writer ().put (mb, tv);
   }
 
+    // Note that the timeout tv is absolute time
   public MessageBlock get (TimeValue tv) throws InterruptedException
   {
     return this.streamHead_.reader ().getq (tv);
@@ -282,9 +284,9 @@ public class Stream
 
     int result = 0;
 
-    if (this.streamHead_.writer ().put (cb, new TimeValue ()) == -1)
+    if (this.streamHead_.writer ().put (cb, null) == -1)
       result = -1;
-    else if ((cb = this.streamHead_.reader ().getq (new TimeValue ())) == null)
+    else if ((cb = this.streamHead_.reader ().getq (null)) == null)
       result = -1;
     else
       result = ((IOCntlMsg ) cb.obj ()).rval ();
diff --git a/java/src/StreamHead.java b/java/src/StreamHead.java
index dfc198f86e2..fc4419d819e 100644
--- a/java/src/StreamHead.java
+++ b/java/src/StreamHead.java
@@ -76,6 +76,13 @@ public class StreamHead extends Task
     return 0;
   }
 
+  // Will block forever to add the given MessageBlock
+  public int put (MessageBlock mb) 
+  {
+      return this.put (mb, null);
+  }
+
+  // tv is absolute time
   public int put (MessageBlock mb, TimeValue tv)
   {
     int res = 0;
diff --git a/java/src/StreamTail.java b/java/src/StreamTail.java
index 26f05d167e8..1c28c676c9c 100644
--- a/java/src/StreamTail.java
+++ b/java/src/StreamTail.java
@@ -82,6 +82,14 @@ public class StreamTail extends Task
     return 0;
   }
 
+  // put the given MessageBlock without a timeout (block forever if
+  // necessary)
+  public int put (MessageBlock mb) 
+  {
+    return this.put (mb, null);
+  }
+
+  // tv is an absolute time timeout
   public int put (MessageBlock mb, TimeValue tv)
   {
     if (this.isWriter ())
diff --git a/java/src/Task.java b/java/src/Task.java
index 1d0f2894c8d..825254e69a8 100644
--- a/java/src/Task.java
+++ b/java/src/Task.java
@@ -230,6 +230,16 @@ public abstract class Task implements Runnable, EventHandler
   }
 
   /**
+   * Insert a message into the queue, blocking forever if necessary.
+   *@param mb Message Block to insert
+   *@exception java.lang.InterruptedException Interrupted while accessing queue
+   */
+  protected int putq (MessageBlock mb) throws InterruptedException
+  {
+      return this.putq(mb, null);
+  }
+
+  /**
    * Insert message into the message queue.
    *@param mb Message Block to insert into the Message Queue
    *@param tv time to wait until (absolute time)
@@ -241,6 +251,17 @@ public abstract class Task implements Runnable, EventHandler
     }
 
   /**
+   * Extract the first message from the queue, blocking forever if
+   * necessary.
+   *@return the first Message Block from the Message Queue.
+   *@exception InterrupteException Interrupted while accessing queue
+   */
+  protected MessageBlock getq() throws InterruptedException
+    {
+      return this.getq(null);
+    }
+
+  /**
    * Extract the first message from the queue. Note that the call is blocking.
    *@return the first Message Block from the Message Queue.
    *@param tv Latest time to wait until (absolute time)
diff --git a/java/src/TimeValue.java b/java/src/TimeValue.java
index 26391b61c6c..97334452a06 100644
--- a/java/src/TimeValue.java
+++ b/java/src/TimeValue.java
@@ -112,7 +112,7 @@ public class TimeValue
    */
   public static TimeValue relativeTimeOfDay(long sec, int nanos) 
   {
-    return new TimeValue(System.currentTimeMillis() / 1000 + sec, 
+    return new TimeValue ((System.currentTimeMillis() / 1000) + sec,
 			 nanos);
   }  
 
@@ -125,8 +125,9 @@ public class TimeValue
    */
   public static TimeValue relativeTimeOfDay(TimeValue offset)
   {
-    return new TimeValue(System.currentTimeMillis() + offset.sec(),
-                         offset.nanos());
+    return new TimeValue ((System.currentTimeMillis() / 1000) +
+			  offset.sec(),
+			  offset.nanos());
   }
 
   /**
diff --git a/java/src/TimedWait.java b/java/src/TimedWait.java
index e8a7eb2d408..a8f337c086b 100644
--- a/java/src/TimedWait.java
+++ b/java/src/TimedWait.java
@@ -92,7 +92,7 @@ public abstract class TimedWait
 	long start = System.currentTimeMillis();
 	long waitTime = tv.getMilliTime() - start;
 
-	// Safety check since there is a possibility that it is
+	// Safety check since there is a possibility that it is now
 	// exactly the same time as the tv.  That would cause
 	// waitTime to be 0, and since Java's wait(timeout) blocks
 	// when timeout is 0, it would mean trouble.
@@ -107,7 +107,6 @@ public abstract class TimedWait
 	  if (!condition ()) {
 
 	    long now = System.currentTimeMillis();
-	    long timeSoFar = now - start;
 	    
 	    // Timed out!
 	    if (now >= tv.getMilliTime ()) 
@@ -115,7 +114,7 @@ public abstract class TimedWait
 	    else 
 	      // We still have some time left to wait, so adjust the
 	      // wait_time.
-	      waitTime -= timeSoFar; 
+	      waitTime = tv.getMilliTime() - now; 
 	  }
 	  else
 	    break;  // Condition became true.
diff --git a/java/src/Token.java b/java/src/Token.java
index ec6abb42343..f8c001573e0 100644
--- a/java/src/Token.java
+++ b/java/src/Token.java
@@ -55,17 +55,21 @@ public class Token
 
   /**
    * Acquire the token. Note that this will block. The method uses
-   * synchronized blocks internally to avoid race conditions.
+   * synchronized blocks internally to avoid race conditions.  It
+   * ignores thread interrupts.
    *@return 0 if acquires without calling <sleepHook>
    * 1 if <sleepHook> is called.
-   * -1 if failure occurs
-   *@exception InterruptedException exception during wait
+   * -1 if failure occurs (should never happen)
    */
-  public int acquire () throws InterruptedException
+  public int acquire ()
     {
       try
 	{
-	  return this.acquire (null);
+	    while (true) {
+		try {
+		    return this.acquire (null);
+		} catch (InterruptedException e) { }
+	    }
 	}
       catch (TimeoutException e)
 	{
@@ -76,7 +80,7 @@ public class Token
     }
 
   /**
-   * Acquire the token.  
+   * Acquire the token.  Returns failure   
    * Throws a TimeoutException if the token isn't acquired before the
    * given absolute time timeout.
    *@param timeout time (TimeValue) to wait until before throwing a
@@ -84,12 +88,9 @@ public class Token
    * Performs a blocking acquire if the given timeout is null.
    *@return 0 if acquires without calling <sleepHook>
    * 1 if <sleepHook> is called.
-   * -1 if failure occurs
-   *@exception TimeoutException exception if timeout occurs
-   *@exception InterruptedException exception during wait
+   * -1 if failure occurs (timeout)
    */
   public int acquire (TimeValue timeout) 
-      throws InterruptedException, TimeoutException
     {
       int result = 0;
       WaitObject snl = new WaitObject ();
@@ -260,7 +261,8 @@ public class Token
             snl.condition (false);
 	    // Wait until the given absolute time (or until notified
 	    // if the timeout is null)
-	    snl.timedWait (timeout);
+	    try {
+		snl.timedWait (timeout);
         }
         // Restore the nesting level and current owner of the lock
         this.nestingLevel_ = saveNestingLevel;