(Not) A synchronization question.

[i30817]

Ok it doesn't have anything to do with games unless you're planing to do Text to speech (an actually fairly compact way of putting really shitty spoken text in a game - a narrator for ex.)

The class is supposted to be a low latency, single thread accessible text to speech object with speech events. It kinda works, but i want to diminish the latency on cancel() of speech. Discussion bellow, relevant code ahead, complete class still bellow.

Since freetts is synchronous in its speak mode, i add each speak request in a Executors singleThreadedExecutor singleThread for processing when ready. Before adding i increment an atomic integer that is decremented in case of the executor is terminated or when the speech terminates, and is used to check is speech is ended programmatically. Since the class is expected to be used in a single thread (the EDT say) there is no synchronization when accessing the shared recourse singleThread.

The effect of singleThread is to queue all requests and to send events at the right time. It also increments a volatile variable index that changes the position for the events on each update. Since only one Runnable is executing at a time i thought volatile sufficed (not simple because executor can and will change executing threads.

The problem is cancellation, since, to cancel i have to wait for the current event to terminate, on a wait loop (hidden inside awaitTermination). Then after termination i have to recreate singleThread and put an event that will reset the index (that's why its volatile). No synchronization again because it is not thread safe(tm).

Can i do this to eliminate the latency?

1   2   3   4   5   6   7   8   9   10   11   12   13   14   15   16   17   18   19

public void cancel() {         if (queuedTasks.get() > 0) {                         final ExecutorService local = Executors.newSingleThreadExecutor();                         local.execute(new Runnable() {                 public void run() {                     killSingleThread(true);                     //this is not the original thread                    //but we waited for the tasks in                    //the other to end, and this goes                    //before any others.                    index = 0;                     singleThread = local;                 }             });         }     }

So i run 3 threads concurrently for a short time before another terminates and passes the hand off to the other ? Or is this an extremely bad idea?
By the way if anybody ever worked with the freetts api do you know if the "real" way to cancel sound output is bugged beyond recovery? Here it just never closed on time. close() put a stop on that non-sense.


All the code and if you find any problems, and if it would be too much trouble could you please report? No pressure or anything.

1   2   3   4   5   6   7   8   9   10   11   12   13   14   15   16   17   18   19   20   21   22   23   24   25   26   27   28   29   30   31   32   33   34   35   36   37   38   39   40   41   42   43   44   45   46   47   48   49   50   51   52   53   54   55   56   57   58   59   60   61   62   63   64   65   66   67   68   69   70   71   72   73   74   75   76   77   78   79   80   81   82   83   84   85   86   87   88   89   90   91   92   93   94   95   96   97   98   99   100   101   102   103   104   105   106   107   108   109   110   111   112   113   114   115   116   117   118   119   120   121   122   123   124   125   126   127   128   129   130   131   132   133   134   135   136   137   138   139   140   141   142   143   144   145   146   147   148   149   150   151   152   153   154   155   156   157   158   159   160   161   162   163   164   165   166   167   168   169   170   171   172   173   174   175   176   177   178   179   180   181   182   183   184   185   186   187   188   189   190   191   192   193   194   195   196   197   198   199   200   201   202   203   204   205   206   207   208   209   210   211   212   213

package util.speech; import com.sun.speech.freetts.VoiceManager; import com.sun.speech.freetts.audio.JavaStreamingAudioPlayer; import java.util.List; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import javax.swing.event.EventListenerList; public final class VoiceImpl extends Voice {     private final AtomicInteger queuedTasks = new AtomicInteger();     //modified only on one thread (by the singleThread executor)    //so no synchronization beyond volatile.    private volatile int index = 0;     private final AtomicBoolean paused = new AtomicBoolean();     private final AtomicBoolean closed = new AtomicBoolean();     private com.sun.speech.freetts.Voice voice;     private final EventListenerList listenerList = new EventListenerList();     //Enqueues the audio. Another thread plays it.    private ExecutorService singleThread = Executors.newSingleThreadExecutor();     public VoiceImpl(String voice) throws IllegalArgumentException {         //bug with streaming audio player requires us to recreate it on cancelation        //make the buffer size smaller.        System.setProperty("com.sun.speech.freetts.audio.AudioPlayer.bufferSize", "1024");         setVoice(voice); }     public VoiceImpl(com.sun.speech.freetts.Voice voice) {         this.voice = voice;     }     @Override     public String getName() {         return voice.getName();     }     private void fireSpeechEvent(EventType type, SpeechEvent evt) {         Object[] listeners = listenerList.getListenerList();         switch (type) {             case START:                 for (int i = listeners.length - 2; i >= 0; i -= 2) {                     if (listeners[i] == SpeechListener.class) {                         SpeechListener l = (SpeechListener) listeners[i + 1];                         l.startedSpeaking(evt);                     }                 }                 break;             case END:                 for (int i = listeners.length - 2; i >= 0; i -= 2) {                     if (listeners[i] == SpeechListener.class) {                         SpeechListener l = (SpeechListener) listeners[i + 1];                         l.endedSpeaking(evt);                     }                 }                 break;         }     }     public void addSpeechListener(SpeechListener l) {         listenerList.add(SpeechListener.class, l);     }     public void removeSpeechListener(SpeechListener l) {         listenerList.remove(SpeechListener.class, l);     }     private void setVoice(String voiceName) throws IllegalArgumentException {         if (voiceName == null) {             throw new IllegalArgumentException("null voice given");         }         com.sun.speech.freetts.Voice tmpVoice = VoiceManager.getInstance().getVoice(voiceName);         if (tmpVoice == null) {             throw new IllegalArgumentException("unrecognized or missing voice");         }         //tmpVoice.getUtteranceProcessors().add(new NotifierProcessor());        voice = tmpVoice;     }     private void prepareToSpeak() {         if (!voice.isLoaded()) {             voice.allocate();         }     }     private void testPrecondictions(String speech) {         if (speech == null) {             throw new IllegalArgumentException("Tried to speak null string.");         }         if (closed.get()) {             throw new IllegalStateException("Tried to speak on a closed voice.");         }     }     public void speak(final String speech) {         testPrecondictions(speech);         executeInSingleThread(new Runnable() {             public void run() {                 doSpeak(speech, index);             }         });     }     public void speak(final int newEventIndex, final String speech) {         testPrecondictions(speech);         executeInSingleThread(new Runnable() {             public void run() {                 doSpeak(speech, newEventIndex);             }         });     }     private void doSpeak(final String speech, int localIndex) {         prepareToSpeak();         fireSpeechEvent(EventType.START, new SpeechEvent(localIndex, speech, false));         boolean success = voice.speak(speech);         if (success) {             fireSpeechEvent(EventType.END, new SpeechEvent(localIndex, speech, false));         } else {             //terrible code inside JavaStreamingAudioPlayer forces me to this.            voice.getAudioPlayer().close();             voice.setAudioPlayer(new JavaStreamingAudioPlayer());             fireSpeechEvent(EventType.END, new SpeechEvent(localIndex, speech, true));         }         index = localIndex + speech.length();         queuedTasks.decrementAndGet();     }     private void executeInSingleThread(Runnable r) {         try {             queuedTasks.incrementAndGet();             singleThread.execute(r);         } catch (RejectedExecutionException re) {             queuedTasks.decrementAndGet();         }     }     public boolean isGeneralDomainVoice() {         return voice.getDomain().equalsIgnoreCase("general");     }     public boolean isSpeaking() {         return queuedTasks.get() != 0;     }     public boolean isPaused() {         return paused.get();     }     public void setPaused(boolean p) {         boolean pa = paused.get();         if (p && !pa) {             voice.getAudioPlayer().pause();         } else if (!p && pa) {             voice.getAudioPlayer().resume();         }         paused.set(p);     }     private void killSingleThread(boolean awaitEnd) {         List l = singleThread.shutdownNow();         queuedTasks.addAndGet(-l.size());         if (awaitEnd) {             try {                 singleThread.awaitTermination(1, TimeUnit.SECONDS);             } catch (InterruptedException ex) {                 throw new AssertionError("Interrupted while waiting for audio to end!");             }         }     }     /**      * Cancel can't be assynchronous.      * close can because it never will      * execute anymore.      */     public void cancel() {         if (queuedTasks.get() > 0) {             killSingleThread(true);             singleThread = Executors.newSingleThreadExecutor();                         singleThread.execute(new Runnable() {                 public void run() {                     //this is not the original thread                    //but we waited for the tasks in                    //the other to end, and this goes                    //before any others.                    index = 0;                 }             });         }     }     public void close() {         if (!closed.getAndSet(true)) {             killSingleThread(false);             if (voice.isLoaded()) {                 voice.deallocate();             }         }     } }

[Mr_Light]

Ok first I'm going comment on things other then your problem cause I'm weird like that....

EventListenerList is pretty obsolete in the light of generics.
Simply use CopyOnWriteSet(or List) and as the generic type use the listener type:

so:

1   2   3   4   5   6   7   8

Set<SpeechListener> listeners = new CopyOnWriteSet<SpeechListener>(); .... protected void fireFoo(...) {  for(SpeechListener listener : listeners) {   listener.onFoo(...);  } }

apart from being more efficient and faster it's probably a lot less obscure.

As to rest of your code, the quedTasks seemed to provide the same as java.util.Queue's size(); if you need to expose the number of to-execute-tasks.
your code with index is not going to work in any 'good' way. With the introduction of the queue you can 'detach'(remove the executors ability to poll()) it(thus making sure indexs don't change) then insert your stuff as needed then reattach that is the the only way you have assurance that stuff doesn't go haywire.(ie the indexes all change because the queue is polled and the method with your index is still being executed.) The other solution would be to expose some dynamic view of the list(queue) and provide a best effort approach. which might be more elegant under some circumstances - so that dependence on your requirements.

As for the cancel method: if you only queue one task to the ExecutorService and use Submit(from ExecutorService) as suppose to Execute(from Executor) you can take the Future that method returns and call cancel (after 'detaching'/stop poll()-ing the queue offcourse...) .

Can i do this to eliminate the latency?

Can you elaborate on that one?

Also are you sure you don't want your cancel method to block? - or at least have a blocking sibling? (ala awaitTermination()) Cause if you queue a task right after you call cancel right now - no behaviour is guaranteed - the task could start playing right away or never gets played. Though depending on your requirements you don't need that guaranty.

[i30817]

About the listeners - coool -.

I need something like singleThreadExecutor because the speak() freetts method is blocking, and i don't want to block the edt, or spam swingworkers (that would not be deterministically ordered anyway).
About the queuedTasks ... i introduced it since ExecutorService interface returned by Executors.newSingleThreadExecutor() has no size method (understandable in a concurrent queue), but since i'm going to use it only in a single thread, i have no such scruples.

This is used only for exposing the

1   2   3

public boolean isSpeaking() {         return queuedTasks.get() != 0;     }

Functionality. To make it "Work" i need to fiddle with it on cancellation (-cancelledTasks), successful task execution and RejectedExecutionException.

Looking at the code in Executors.newSingleThreadExecutor() it instanciates a ThreadPoolExecutor. Looking at threadPoolExecutor.getTaskCount(), doesn't do what i want (just to eventually know that there is nothing executing) but count all tasks that were executed, and threadPoolExecutor.getActiveCount() is looking at Thread counts and moreover both methods have the scary word "approximately" in the javadoc.

Do you think threadPoolExecutor.getQueue().getSize() would work? I don't think it works to just use queue directly because the speak method being synchronous as explained above.

The index is a sum of the String lengths spoken since the start, cancellation or the overiden speak method. It's there mostly to be sent in events. I'm considering just erasing it and simplifying the speak interface to
speak(String)
and
speak(Iterator)
People that need to keep indexes should just use the second and keep it internally in the iterator - i think this makes client code more organized too.

I think you're right about how cancel should be blocking, its just that
1) It runs on the edt - i guess no way around it.
2) The AssertionError i'm throwing in killSingleThread(true) worries me, since just being late should probably not kill the calling thread.
You'd recommend a cycle instead?


What do you think?

Games published by our own members! Check 'em out!

[i30817]

Here is what i've got (without altering anything about the queuedItems Atomic integer.
Also since the class is obviously not thread safe, i guess the paused and closed AtomicBooleans can be normal ones, right? 

1   2   3   4   5   6   7   8   9   10   11   12   13   14   15   16   17   18   19   20   21   22   23   24   25   26   27   28   29   30   31   32   33   34   35   36   37   38   39   40   41   42   43   44   45   46   47   48   49   50   51   52   53   54   55   56   57   58   59   60   61   62   63   64   65   66   67   68   69   70   71   72   73   74   75   76   77   78   79   80   81   82   83   84   85   86   87   88   89   90   91   92   93   94   95   96   97   98   99   100   101   102   103   104   105   106   107   108   109   110   111   112   113   114   115   116   117   118   119   120   121   122   123   124   125   126   127   128   129   130   131   132   133   134   135   136   137   138   139   140   141   142   143   144   145   146   147   148   149   150   151   152   153   154   155   156   157   158   159   160   161   162   163   164   165   166   167   168   169   170   171   172   173   174   175   176   177   178   179   180   181   182   183   184   185   186   187   188   189   190   191   192   193   194   195   196   197   198   199   200

package util.speech; import com.sun.speech.freetts.VoiceManager; import com.sun.speech.freetts.audio.JavaStreamingAudioPlayer; import java.util.Iterator; import java.util.List; import java.util.Set; import java.util.concurrent.CopyOnWriteArraySet; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import javax.swing.event.EventListenerList; //@org.openide.util.lookup.ServiceProvider(service=Voice.class) public final class VoiceImpl extends Voice {     private final AtomicInteger queuedTasks = new AtomicInteger();     private boolean paused;     private boolean closed;     private com.sun.speech.freetts.Voice voice;     private final Set<SpeechListener> listeners = new CopyOnWriteArraySet<SpeechListener>();     //Enqueues the audio. Another thread plays it.    private ExecutorService singleThread = Executors.newSingleThreadExecutor();     public VoiceImpl(String voice) throws IllegalArgumentException {         setVoice(voice);     }     public VoiceImpl(com.sun.speech.freetts.Voice voice) {         this.voice = voice;     }     @Override     public String getName() {         return voice.getName();     }     private void fireSpeechEvent(EventType type, SpeechEvent evt) {                 switch (type) {             case START:                 for (SpeechListener listener : listeners) {                     listener.startedSpeaking(evt);                 }                 break;             case END:                 for (SpeechListener listener : listeners) {                     listener.endedSpeaking(evt);                 }                 break;         }     }     public void addSpeechListener(SpeechListener l) {         listeners.add(l);     }     public void removeSpeechListener(SpeechListener l) {         listeners.remove(l);     }     private void setVoice(String voiceName) throws IllegalArgumentException {         if (voiceName == null) {             throw new IllegalArgumentException("null voice given");         }         com.sun.speech.freetts.Voice tmpVoice = VoiceManager.getInstance().getVoice(voiceName);         if (tmpVoice == null) {             throw new IllegalArgumentException("unrecognized or missing voice");         }         voice = tmpVoice;     }     private void prepareToSpeak() {         if (!voice.isLoaded()) {             voice.allocate();         }     }     private void testPrecondictions(Object speech) {         if (speech == null) {             throw new IllegalArgumentException("Tried to speak a null speech.");         }         if (closed) {             throw new IllegalStateException("Tried to speak on a closed voice.");         }     }     public void speak(final String speech) {         testPrecondictions(speech);         executeInSingleThread(new Runnable() {             public void run() {                 doSpeak(speech, 0);                 queuedTasks.decrementAndGet();             }         });     }     public void speak(final Iterator<String> speech) {         testPrecondictions(speech);         executeInSingleThread(new Runnable() {             public void run() {                 boolean success = true;                 int lengthSum = 0;                 while (success && speech.hasNext()) {                     String next = speech.next();                     success = doSpeak(next, lengthSum);                     lengthSum += next.length();                 }                 queuedTasks.decrementAndGet();             }         });     }     private boolean doSpeak(final String speech, int localIndex) {         prepareToSpeak();         fireSpeechEvent(EventType.START, new SpeechEvent(localIndex, speech, false));         boolean success = voice.speak(speech);         if (success) {             fireSpeechEvent(EventType.END, new SpeechEvent(localIndex, speech, false));         } else {             //terrible code inside JavaStreamingAudioPlayer forces me to this.            voice.getAudioPlayer().close();             voice.setAudioPlayer(new JavaStreamingAudioPlayer());             fireSpeechEvent(EventType.END, new SpeechEvent(localIndex, speech, true));         }         return success;     }     private void executeInSingleThread(Runnable r) {         try {             queuedTasks.incrementAndGet();             singleThread.execute(r);         } catch (RejectedExecutionException re) {             queuedTasks.decrementAndGet();         }     }     public boolean isGeneralDomainVoice() {         return voice.getDomain().equalsIgnoreCase("general");     }     public boolean isSpeaking() {         return queuedTasks.get() != 0;     }     public boolean isPaused() {         return paused;     }     public void setPaused(boolean p) {                 if (p && !paused) {             voice.getAudioPlayer().pause();         } else if (!p && paused) {             voice.getAudioPlayer().resume();         }         paused = p;     }     private void killSingleThread(boolean awaitEnd) {         List l = singleThread.shutdownNow();         queuedTasks.addAndGet(-l.size());         if (awaitEnd) {             try {                 singleThread.awaitTermination(1, TimeUnit.SECONDS);             } catch (InterruptedException ex) {                 throw new AssertionError("Interrupted while waiting for audio to end!");             }         }     }     /**      * Cancel can't be assynchronous.      * close can because it never will      * execute anymore.      */     public void cancel() {         if (queuedTasks.get() > 0) {             killSingleThread(true);             singleThread = Executors.newSingleThreadExecutor();         }     }     public void close() {         if (!closed) {             closed = true;             killSingleThread(false);             if (voice.isLoaded()) {                 voice.deallocate();             }         }     } }

[i30817]

Oh, Oh, OH. Finally tried this in linux. In windows it works , in linux it  gives me
LINE UNAVAILABLE: Format is PCM_SIGNED 16000.0 Hz, 16 bit, mono, 2 bytes/frame, big-endian

Caused by the extremely shitty JavaStreamingAudioPlayer class trying to access a sound dataline and evidently failing. Freetts is more or less dead, but i'm not giving up yet. Time to troll the openjdk sound mailing lists.

Edit: In another computer with the same disto (computer science lab) it works. How annoying. How am i going to present a coherent bug report?

[Mr_Light]

you can make cancel assync by seduling a cancelFinished event on the EDT you might want to keep the code outside your class though as it couples AWT with your class.

threadPoolExecutor.getQueue().getSize() would work 'as well' as your old solution (+1 of the running thread) but as long as you don't include some signaling from poll() but even then (unless you block the poll() from the queue) you can't guaranty that the value your returning is still truth by the time it returns but again it might be good "enough".

[i30817]

I don't think i understand. How can it not be good enough (either way) since
1) the inner voice.speak(String) is blocking
2) the value is only decreased after voice speak ends.
3) the value is visibility thread safe (volatile, atomic) so eventually the thread that is using the class sees the change.
4) one thread is using the class, so it can either query the value immediatly (and get the correct true) or after the runnable runs (and get false)

As you can see i already deleted the index signaling, in favor of Iterator<String> so there is no reason for me not to make cancel asynchronous i guess. Good, one less failure mode. I'm going to try to see if this http://forums.sun.com/thread.jspa?threadID=5189363 is the other dataline problem. Don't you just love neglected jdk parts? It makes me think of ... IBM.

[Mr_Light]

Perhaps I'm mistaking the context in which isSpeaking() is suppose to be called, but let me illustrate

1 -> thread 1
2 -> thread 2

1: request number of queued (queued = 1)
2: voice speak ends => actual speaking is false now.
1: comparing number with 0 (queued != 0=>true)
1: returning true.

Sure the next call returns false but.... etc.

Anyways, does that make sense?

//edit
if point 4 is really true then it should only get called by the same thread that calls the actual speak() and perhaps you should wonder if this method should actually be public. This also sounds weird in the light of speak() being blocking - how are you ever going to call that method and make it return true. That or it is named wrong and should be something like remaining tasks. Either I'm being dense or we are mis-communicating here.

[i30817]

I see what you mean. Point 4 is then completely wrong and i am officially a concurrency idiot.
The blocking method is the FreeTTS one, called in doSpeak(String, int).

Originally isSpeaking was part of another public method await(), but lately i've been using it like this:

1   2   3   4   5   6   7   8

public void toggleVoice() {         if (voice.isSpeaking()) {             voice.setPaused(!voice.isPaused());         } else {             int index = pane.getIndex();             voice.speak(pane.getParagraphIterator(index));         }     }

I wondered a time ago why i sometimes had to press pause twice for it to start speaking again. Guess now i know. Trouble is i hate having to wait for this.

Anyway i sent a message to the openjdk sound mailing list (pending moderator approval) telling them of the strange dataline behavior in linux, tested the workaround in the labs machines (it works) and am now going to try to make a patch to the FreeTTS project. There are hundreds of poor sods with the same problem in google.

[Mr_Light]

I wouldn't beat yourself over it

But if you just need the isSpeaking for pausing the thing then there are easier ways as out lined.

Something alike:

1   2   3   4   5   6   7   8   9   10   11   12   13   14   15   16   17   18   19   20   21   22   23

volatile boolean pause; // could also use atomic I suppose.. volatile BlockingQueue<String> taskQueue = ... public void run() {   while(!shutdown) {     try {       if(paused) //wait.      if(Thread.currentThread().isInterrupted()) {         // reset interrupted status        continue;       }       String word = taskQueue.poll(); //poll blocks.      speak(word);     } catch (InterruptedException e ) {         if(!paused) return;         // reset interrupted status    } } public void pause() {    paused = true;    workerThread.interrupt(); }

though you could also use Executor and provide a ThreadFactory (which would allow you to call interrupt.
ExecutorService not queue more then one task and use the future returned by submit.