support/espeak/fifo.c

/***************************************************************************
 *   Copyright (C) 2007, Gilles Casse <gcasse@oralux.org>                  *
 *                                                                         *
 *   This program is free software; you can redistribute it and/or modify  *
 *   it under the terms of the GNU General Public License as published by  *
 *   the Free Software Foundation; either version 3 of the License, or     *
 *   (at your option) any later version.                                   *
 *                                                                         *
 *   This program 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 for more details.                          *
 *                                                                         *
 *   You should have received a copy of the GNU General Public License     *
 *   along with this program; if not, write to the                         *
 *   Free Software Foundation, Inc.,                                       *
 *   59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.             *
 ***************************************************************************/
#include "speech.h"

#ifdef USE_ASYNC
// This source file is only used for asynchronious modes


//<includes

#include <unistd.h>
#include <assert.h>
#include <string.h>
#include <stdlib.h>
#include <pthread.h>
#include <semaphore.h>
#include <wchar.h>
#include <errno.h>
#include <sys/time.h>
#include <time.h>

#include "fifo.h"
#include "wave.h"
#include "debug.h"


//>
//<decls and function prototypes

// my_mutex: protects my_thread_is_talking, 
// my_stop_is_required, and the command fifo
static pthread_mutex_t my_mutex;
static int my_command_is_running = 0;
static int my_stop_is_required = 0;
// + fifo
//

// my_thread: reads commands from the fifo, and runs them. 
static pthread_t my_thread;
static sem_t my_sem_start_is_required;
static sem_t my_sem_stop_is_acknowledged;

static void* say_thread(void*);

static espeak_ERROR push(t_espeak_command* the_command);
static t_espeak_command* pop();
static void init();
static int node_counter=0;
enum {MAX_NODE_COUNTER=400,
      INACTIVITY_TIMEOUT=50, // in ms, check that the stream is inactive
      MAX_INACTIVITY_CHECK=2
};

//>
//<fifo_init
void fifo_init()
{
  ENTER("fifo_init");

  // security
  pthread_mutex_init( &my_mutex, (const pthread_mutexattr_t *)NULL);
  init();

  assert(-1 != sem_init(&my_sem_start_is_required, 0, 0));
  assert(-1 != sem_init(&my_sem_stop_is_acknowledged, 0, 0));

  pthread_attr_t a_attrib;    
  if (pthread_attr_init (& a_attrib)
      || pthread_attr_setdetachstate(&a_attrib, PTHREAD_CREATE_JOINABLE)
      || pthread_create( &my_thread, 
                         & a_attrib, 
                         say_thread, 
                         (void*)NULL))
    {
      assert(0);
    }

  pthread_attr_destroy(&a_attrib);

  // leave once the thread is actually started
  SHOW_TIME("fifo > wait for my_sem_stop_is_acknowledged\n");
  while ((sem_wait(&my_sem_stop_is_acknowledged) == -1) && errno == EINTR)
    {
      continue; // Restart when interrupted by handler
    }
  SHOW_TIME("fifo > get my_sem_stop_is_acknowledged\n");
}
//>
//<fifo_add_command

espeak_ERROR fifo_add_command (t_espeak_command* the_command)
{
  ENTER("fifo_add_command");
          
  int a_status = pthread_mutex_lock(&my_mutex); 
  espeak_ERROR a_error = EE_OK;

  if (!a_status)
    {
      SHOW_TIME("fifo_add_command > locked\n");
      a_error = push(the_command);
      SHOW_TIME("fifo_add_command > unlocking\n");
      a_status = pthread_mutex_unlock(&my_mutex);
    }

  if (!a_status && !my_command_is_running && (a_error == EE_OK))
    {
      // quit when command is actually started 
      // (for possible forthcoming 'end of command' checks)
      SHOW_TIME("fifo_add_command > post my_sem_start_is_required\n");
      sem_post(&my_sem_start_is_required);
      int val=1;
      while (val)
        {
          usleep(50000); // TBD: event?
          sem_getvalue(&my_sem_start_is_required, &val);
        }
    }

  if (a_status != 0)
    {
      a_error = EE_INTERNAL_ERROR;
    }

  SHOW_TIME("LEAVE fifo_add_command");
  return a_error;
}

//>
//<fifo_add_commands

espeak_ERROR fifo_add_commands (t_espeak_command* command1, t_espeak_command* command2)
{
  ENTER("fifo_add_command");
          
  int a_status = pthread_mutex_lock(&my_mutex); 
  espeak_ERROR a_error = EE_OK;

  if (!a_status)
    {
      SHOW_TIME("fifo_add_commands > locked\n");

      if (node_counter+1 >= MAX_NODE_COUNTER)
        {
          SHOW("push > %s\n", "EE_BUFFER_FULL");
          a_error = EE_BUFFER_FULL;
        }
      else
        {
          push(command1);
          push(command2);
        }
      SHOW_TIME("fifo_add_command > unlocking\n");
      a_status = pthread_mutex_unlock(&my_mutex);
    }

  if (!a_status && !my_command_is_running && (a_error == EE_OK))
    {
      // quit when one command is actually started 
      // (for possible forthcoming 'end of command' checks)
      SHOW_TIME("fifo_add_command > post my_sem_start_is_required\n");
      sem_post(&my_sem_start_is_required);
      int val=1;
      while (val)
        {
          usleep(50000); // TBD: event?
          sem_getvalue(&my_sem_start_is_required, &val);
        }
    }

  if (a_status != 0)
    {
      a_error = EE_INTERNAL_ERROR;
    }

  SHOW_TIME("LEAVE fifo_add_commands");
  return a_error;
}

//>
//<fifo_stop

espeak_ERROR fifo_stop ()
{
  ENTER("fifo_stop");

  int a_command_is_running = 0;
  int a_status = pthread_mutex_lock(&my_mutex);
  SHOW_TIME("fifo_stop > locked\n");
  if (a_status != 0)
    {
      return EE_INTERNAL_ERROR;
    }

  if (my_command_is_running)
    {
      a_command_is_running = 1;
      my_stop_is_required = 1;
      SHOW_TIME("fifo_stop > my_stop_is_required = 1\n");
    }
  SHOW_TIME("fifo_stop > unlocking\n");
  a_status = pthread_mutex_unlock(&my_mutex);
  if (a_status != 0)
    {
      return EE_INTERNAL_ERROR;
    }

  if (a_command_is_running)
    {
      SHOW_TIME("fifo_stop > wait for my_sem_stop_is_acknowledged\n");
      while ((sem_wait(&my_sem_stop_is_acknowledged) == -1) && errno == EINTR)
        {
          continue; // Restart when interrupted by handler
        }
      SHOW_TIME("fifo_stop > get my_sem_stop_is_acknowledged\n");
    }

  SHOW_TIME("fifo_stop > my_stop_is_required = 0\n");
  my_stop_is_required = 0;
  SHOW_TIME("LEAVE fifo_stop\n");

  return EE_OK;
}

//>

//<fifo_is_speaking
int fifo_is_busy ()
{
  //  ENTER("isSpeaking");
  //  int aResult = (int) (my_command_is_running || WaveIsPlaying());
  SHOW("fifo_is_busy > aResult = %d\n",my_command_is_running);
  return my_command_is_running;
}

// int pause ()
// {
//   ENTER("pause");
//   // TBD
//   //   if (espeakPause (espeakHandle, 1))
//   return true;
// }

// int resume ()
// {
//   ENTER("resume");
//   // TBD
//   //   if (espeakPause (espeakHandle, 0))
//   return true;
// }
//>


//<sleep_until_start_request_or_inactivity

static int sleep_until_start_request_or_inactivity()
{
  SHOW_TIME("fifo > sleep_until_start_request_or_inactivity > ENTER");
  int a_start_is_required=0;

  // Wait for the start request (my_sem_start_is_required).
  // Besides this, if the audio stream is still busy, 
  // check from time to time its end. 
  // The end of the stream is confirmed by several checks 
  // for filtering underflow.
  //
  int i=0;
  while((i<= MAX_INACTIVITY_CHECK) && !a_start_is_required)
    {
      if (wave_is_busy( NULL) )
        {
          i = 0;
        }
      else
        {
          i++;
        }

      int err=0;
      struct timespec ts, to;
      struct timeval tv;
      
      clock_gettime2( &ts);
      to.tv_sec = ts.tv_sec;
      to.tv_nsec = ts.tv_nsec;
      
      add_time_in_ms( &ts, INACTIVITY_TIMEOUT);
      
      SHOW("fifo > sleep_until_start_request_or_inactivity > start sem_timedwait (start_is_required) from %d.%09lu to %d.%09lu \n", 
           to.tv_sec, to.tv_nsec,
           ts.tv_sec, ts.tv_nsec);
      
      while ((err = sem_timedwait(&my_sem_start_is_required, &ts)) == -1 
             && errno == EINTR)
        {
              continue;
        }
      
      assert (gettimeofday(&tv, NULL) != -1);
      SHOW("fifo > sleep_until_start_request_or_inactivity > stop sem_timedwait (start_is_required, err=%d) %d.%09lu \n", err, 
           tv.tv_sec, tv.tv_usec*1000);
      
      if (err==0)
        {
          a_start_is_required = 1;
        }
    }
  SHOW_TIME("fifo > sleep_until_start_request_or_inactivity > LEAVE");
  return a_start_is_required;
}

//>
//<close_stream

static void close_stream()
{
  SHOW_TIME("fifo > close_stream > ENTER\n");

  // Warning: a wave_close can be already required by 
  // an external command (espeak_Cancel + fifo_stop), if so: 
  // my_stop_is_required = 1;

  int a_status = pthread_mutex_lock(&my_mutex);
  assert (!a_status);
  int a_stop_is_required = my_stop_is_required;
  if (!a_stop_is_required)
    {
      my_command_is_running = 1;
    }
  a_status = pthread_mutex_unlock(&my_mutex);

  if (!a_stop_is_required)
    {
      wave_close(NULL);

      int a_status = pthread_mutex_lock(&my_mutex);
      assert (!a_status);
      my_command_is_running = 0;

      a_stop_is_required = my_stop_is_required;
      a_status = pthread_mutex_unlock(&my_mutex);
      
      if (a_stop_is_required)
        {
          // acknowledge the stop request
          SHOW_TIME("fifo > close_stream > post my_sem_stop_is_acknowledged\n");
          int a_status = sem_post(&my_sem_stop_is_acknowledged);
          assert( a_status != -1);
        }
    }

  SHOW_TIME("fifo > close_stream > LEAVE\n");
}

//>
//<say_thread

static void* say_thread(void*)
{
  ENTER("say_thread");

  SHOW_TIME("say_thread > post my_sem_stop_is_acknowledged\n");

  // announce that thread is started
  sem_post(&my_sem_stop_is_acknowledged);

  int look_for_inactivity=0;

  while(1)
    {
      SHOW_TIME("say_thread > wait for my_sem_start_is_required\n");

      int a_start_is_required = 0;
      if (look_for_inactivity)
        {
          a_start_is_required = sleep_until_start_request_or_inactivity();
          if (!a_start_is_required)
            {
              close_stream();
            }
        }
      look_for_inactivity = 1;

      if (!a_start_is_required)
        {
          while ((sem_wait(&my_sem_start_is_required) == -1) && errno == EINTR)
            {
              continue; // Restart when interrupted by handler
            }
        }
      SHOW_TIME("say_thread > get my_sem_start_is_required\n");

      SHOW_TIME("say_thread > my_command_is_running = 1\n");
      my_command_is_running = 1;

      while( my_command_is_running)
        {
          SHOW_TIME("say_thread > locking\n");
          int a_status = pthread_mutex_lock(&my_mutex);
          assert (!a_status);
          t_espeak_command* a_command = (t_espeak_command*)pop();

          if (a_command == NULL)
            {
              SHOW_TIME("say_thread > text empty (talking=0) \n");
              a_status = pthread_mutex_unlock(&my_mutex);
              SHOW_TIME("say_thread > unlocked\n");
              SHOW_TIME("say_thread > my_command_is_running = 0\n");
              my_command_is_running = 0;
            }
          else
            {
              display_espeak_command(a_command);
              // purge start semaphore
              SHOW_TIME("say_thread > purge my_sem_start_is_required\n");
              while(0 == sem_trywait(&my_sem_start_is_required))
                {
                };

              if (my_stop_is_required)
                { 
                  SHOW_TIME("say_thread > my_command_is_running = 0\n");
                  my_command_is_running = 0;
                }
              SHOW_TIME("say_thread > unlocking\n");
              a_status = pthread_mutex_unlock(&my_mutex);

              if (my_command_is_running)
                {
                  process_espeak_command(a_command);
                }
              delete_espeak_command(a_command);
            }
        }

      if (my_stop_is_required)
        { 
          // no mutex required since the stop command is synchronous
          // and waiting for my_sem_stop_is_acknowledged
          init();

          // purge start semaphore
          SHOW_TIME("say_thread > purge my_sem_start_is_required\n");
          while(0==sem_trywait(&my_sem_start_is_required))
            {
            };
          
          // acknowledge the stop request
          SHOW_TIME("say_thread > post my_sem_stop_is_acknowledged\n");
          int a_status = sem_post(&my_sem_stop_is_acknowledged);
          assert( a_status != -1);
        }
      // and wait for the next start
      SHOW_TIME("say_thread > wait for my_sem_start_is_required\n");      
    }
    
  return NULL;
}

int fifo_is_command_enabled()
{
  SHOW("ENTER fifo_is_command_enabled=%d\n",(int)(0 == my_stop_is_required));
  return (0 == my_stop_is_required);
}

//>
//<fifo
typedef struct t_node
{
  t_espeak_command* data;
  t_node *next;
} node;

static node* head=NULL;
static node* tail=NULL;
// return 1 if ok, 0 otherwise
static espeak_ERROR push(t_espeak_command* the_command)
{
  ENTER("fifo > push");

  assert((!head && !tail) || (head && tail));

  if (the_command == NULL)
    {
      SHOW("push > command=0x%x\n", NULL);
      return EE_INTERNAL_ERROR;
    }

  if (node_counter >= MAX_NODE_COUNTER)
    {
      SHOW("push > %s\n", "EE_BUFFER_FULL");
      return EE_BUFFER_FULL;
    }

  node *n = (node *)malloc(sizeof(node));
  if (n == NULL)
    {
      return EE_INTERNAL_ERROR;
    }
  
  if (head == NULL)
    {
      head = n;
      tail = n;
    }
  else
    {
      tail->next = n;
      tail = n;
    }
  
  tail->next = NULL;
  tail->data = the_command;

  node_counter++;
  SHOW("push > counter=%d\n",node_counter);

  the_command->state = CS_PENDING;
  display_espeak_command(the_command);

  return EE_OK;
}

static t_espeak_command* pop()
{
  ENTER("fifo > pop");
  t_espeak_command* the_command = NULL;

  assert((!head && !tail) || (head && tail));

  if (head != NULL)
    {
      node* n = head;
      the_command = n->data;
      head = n->next;
      free(n);
      node_counter--;
      SHOW("pop > command=0x%x (counter=%d)\n",the_command, node_counter);
    }

  if(head == NULL)
    {
      tail = NULL;
    }

  display_espeak_command(the_command);
 
  return the_command;
}


static void init()
{
  ENTER("fifo > init");
  while (delete_espeak_command( pop() ))
    {}
  node_counter = 0;
}

//>
//<fifo_init
void fifo_terminate()
{
  ENTER("fifo_terminate");

  pthread_cancel(my_thread);
  pthread_join(my_thread,NULL);
  pthread_mutex_destroy(&my_mutex);
  sem_destroy(&my_sem_start_is_required);
  sem_destroy(&my_sem_stop_is_acknowledged);

  init(); // purge fifo
}

#endif
//>

1	/***************************************************************************
2	* Copyright (C) 2007, Gilles Casse <gcasse@oralux.org> *
3	* *
4	* This program is free software; you can redistribute it and/or modify *
5	* it under the terms of the GNU General Public License as published by *
6	* the Free Software Foundation; either version 3 of the License, or *
7	* (at your option) any later version. *
8	* *
9	* This program is distributed in the hope that it will be useful, *
10	* but WITHOUT ANY WARRANTY; without even the implied warranty of *
11	* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the *
12	* GNU General Public License for more details. *
13	* *
14	* You should have received a copy of the GNU General Public License *
15	* along with this program; if not, write to the *
16	* Free Software Foundation, Inc., *
17	* 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA. *
18	***************************************************************************/
19	#include "speech.h"
20
21	#ifdef USE_ASYNC
22	// This source file is only used for asynchronious modes
23
24
25	//<includes
26
27	#include <unistd.h>
28	#include <assert.h>
29	#include <string.h>
30	#include <stdlib.h>
31	#include <pthread.h>
32	#include <semaphore.h>
33	#include <wchar.h>
34	#include <errno.h>
35	#include <sys/time.h>
36	#include <time.h>
37
38	#include "fifo.h"
39	#include "wave.h"
40	#include "debug.h"
41
42
43	//>
44	//<decls and function prototypes
45
46	// my_mutex: protects my_thread_is_talking,
47	// my_stop_is_required, and the command fifo
48	static pthread_mutex_t my_mutex;
49	static int my_command_is_running = 0;
50	static int my_stop_is_required = 0;
51	// + fifo
52	//
53
54	// my_thread: reads commands from the fifo, and runs them.
55	static pthread_t my_thread;
56	static sem_t my_sem_start_is_required;
57	static sem_t my_sem_stop_is_acknowledged;
58
59	static void* say_thread(void*);
60
61	static espeak_ERROR push(t_espeak_command* the_command);
62	static t_espeak_command* pop();
63	static void init();
64	static int node_counter=0;
65	enum {MAX_NODE_COUNTER=400,
66	INACTIVITY_TIMEOUT=50, // in ms, check that the stream is inactive
67	MAX_INACTIVITY_CHECK=2
68	};
69
70	//>
71	//<fifo_init
72	void fifo_init()
73	{
74	ENTER("fifo_init");
75
76	// security
77	pthread_mutex_init( &my_mutex, (const pthread_mutexattr_t *)NULL);
78	init();
79
80	assert(-1 != sem_init(&my_sem_start_is_required, 0, 0));
81	assert(-1 != sem_init(&my_sem_stop_is_acknowledged, 0, 0));
82
83	pthread_attr_t a_attrib;
84	if (pthread_attr_init (& a_attrib)
85	\|\| pthread_attr_setdetachstate(&a_attrib, PTHREAD_CREATE_JOINABLE)
86	\|\| pthread_create( &my_thread,
87	& a_attrib,
88	say_thread,
89	(void*)NULL))
90	{
91	assert(0);
92	}
93
94	pthread_attr_destroy(&a_attrib);
95
96	// leave once the thread is actually started
97	SHOW_TIME("fifo > wait for my_sem_stop_is_acknowledged\n");
98	while ((sem_wait(&my_sem_stop_is_acknowledged) == -1) && errno == EINTR)
99	{
100	continue; // Restart when interrupted by handler
101	}
102	SHOW_TIME("fifo > get my_sem_stop_is_acknowledged\n");
103	}
104	//>
105	//<fifo_add_command
106
107	espeak_ERROR fifo_add_command (t_espeak_command* the_command)
108	{
109	ENTER("fifo_add_command");
110
111	int a_status = pthread_mutex_lock(&my_mutex);
112	espeak_ERROR a_error = EE_OK;
113
114	if (!a_status)
115	{
116	SHOW_TIME("fifo_add_command > locked\n");
117	a_error = push(the_command);
118	SHOW_TIME("fifo_add_command > unlocking\n");
119	a_status = pthread_mutex_unlock(&my_mutex);
120	}
121
122	if (!a_status && !my_command_is_running && (a_error == EE_OK))
123	{
124	// quit when command is actually started
125	// (for possible forthcoming 'end of command' checks)
126	SHOW_TIME("fifo_add_command > post my_sem_start_is_required\n");
127	sem_post(&my_sem_start_is_required);
128	int val=1;
129	while (val)
130	{
131	usleep(50000); // TBD: event?
132	sem_getvalue(&my_sem_start_is_required, &val);
133	}
134	}
135
136	if (a_status != 0)
137	{
138	a_error = EE_INTERNAL_ERROR;
139	}
140
141	SHOW_TIME("LEAVE fifo_add_command");
142	return a_error;
143	}
144
145	//>
146	//<fifo_add_commands
147
148	espeak_ERROR fifo_add_commands (t_espeak_command* command1, t_espeak_command* command2)
149	{
150	ENTER("fifo_add_command");
151
152	int a_status = pthread_mutex_lock(&my_mutex);
153	espeak_ERROR a_error = EE_OK;
154
155	if (!a_status)
156	{
157	SHOW_TIME("fifo_add_commands > locked\n");
158
159	if (node_counter+1 >= MAX_NODE_COUNTER)
160	{
161	SHOW("push > %s\n", "EE_BUFFER_FULL");
162	a_error = EE_BUFFER_FULL;
163	}
164	else
165	{
166	push(command1);
167	push(command2);
168	}
169	SHOW_TIME("fifo_add_command > unlocking\n");
170	a_status = pthread_mutex_unlock(&my_mutex);
171	}
172
173	if (!a_status && !my_command_is_running && (a_error == EE_OK))
174	{
175	// quit when one command is actually started
176	// (for possible forthcoming 'end of command' checks)
177	SHOW_TIME("fifo_add_command > post my_sem_start_is_required\n");
178	sem_post(&my_sem_start_is_required);
179	int val=1;
180	while (val)
181	{
182	usleep(50000); // TBD: event?
183	sem_getvalue(&my_sem_start_is_required, &val);
184	}
185	}
186
187	if (a_status != 0)
188	{
189	a_error = EE_INTERNAL_ERROR;
190	}
191
192	SHOW_TIME("LEAVE fifo_add_commands");
193	return a_error;
194	}
195
196	//>
197	//<fifo_stop
198
199	espeak_ERROR fifo_stop ()
200	{
201	ENTER("fifo_stop");
202
203	int a_command_is_running = 0;
204	int a_status = pthread_mutex_lock(&my_mutex);
205	SHOW_TIME("fifo_stop > locked\n");
206	if (a_status != 0)
207	{
208	return EE_INTERNAL_ERROR;
209	}
210
211	if (my_command_is_running)
212	{
213	a_command_is_running = 1;
214	my_stop_is_required = 1;
215	SHOW_TIME("fifo_stop > my_stop_is_required = 1\n");
216	}
217	SHOW_TIME("fifo_stop > unlocking\n");
218	a_status = pthread_mutex_unlock(&my_mutex);
219	if (a_status != 0)
220	{
221	return EE_INTERNAL_ERROR;
222	}
223
224	if (a_command_is_running)
225	{
226	SHOW_TIME("fifo_stop > wait for my_sem_stop_is_acknowledged\n");
227	while ((sem_wait(&my_sem_stop_is_acknowledged) == -1) && errno == EINTR)
228	{
229	continue; // Restart when interrupted by handler
230	}
231	SHOW_TIME("fifo_stop > get my_sem_stop_is_acknowledged\n");
232	}
233
234	SHOW_TIME("fifo_stop > my_stop_is_required = 0\n");
235	my_stop_is_required = 0;
236	SHOW_TIME("LEAVE fifo_stop\n");
237
238	return EE_OK;
239	}
240
241	//>
242
243	//<fifo_is_speaking
244	int fifo_is_busy ()
245	{
246	// ENTER("isSpeaking");
247	// int aResult = (int) (my_command_is_running \|\| WaveIsPlaying());
248	SHOW("fifo_is_busy > aResult = %d\n",my_command_is_running);
249	return my_command_is_running;
250	}
251
252	// int pause ()
253	// {
254	// ENTER("pause");
255	// // TBD
256	// // if (espeakPause (espeakHandle, 1))
257	// return true;
258	// }
259
260	// int resume ()
261	// {
262	// ENTER("resume");
263	// // TBD
264	// // if (espeakPause (espeakHandle, 0))
265	// return true;
266	// }
267	//>
268
269
270	//<sleep_until_start_request_or_inactivity
271
272	static int sleep_until_start_request_or_inactivity()
273	{
274	SHOW_TIME("fifo > sleep_until_start_request_or_inactivity > ENTER");
275	int a_start_is_required=0;
276
277	// Wait for the start request (my_sem_start_is_required).
278	// Besides this, if the audio stream is still busy,
279	// check from time to time its end.
280	// The end of the stream is confirmed by several checks
281	// for filtering underflow.
282	//
283	int i=0;
284	while((i<= MAX_INACTIVITY_CHECK) && !a_start_is_required)
285	{
286	if (wave_is_busy( NULL) )
287	{
288	i = 0;
289	}
290	else
291	{
292	i++;
293	}
294
295	int err=0;
296	struct timespec ts, to;
297	struct timeval tv;
298
299	clock_gettime2( &ts);
300	to.tv_sec = ts.tv_sec;
301	to.tv_nsec = ts.tv_nsec;
302
303	add_time_in_ms( &ts, INACTIVITY_TIMEOUT);
304
305	SHOW("fifo > sleep_until_start_request_or_inactivity > start sem_timedwait (start_is_required) from %d.%09lu to %d.%09lu \n",
306	to.tv_sec, to.tv_nsec,
307	ts.tv_sec, ts.tv_nsec);
308
309	while ((err = sem_timedwait(&my_sem_start_is_required, &ts)) == -1
310	&& errno == EINTR)
311	{
312	continue;
313	}
314
315	assert (gettimeofday(&tv, NULL) != -1);
316	SHOW("fifo > sleep_until_start_request_or_inactivity > stop sem_timedwait (start_is_required, err=%d) %d.%09lu \n", err,
317	tv.tv_sec, tv.tv_usec*1000);
318
319	if (err==0)
320	{
321	a_start_is_required = 1;
322	}
323	}
324	SHOW_TIME("fifo > sleep_until_start_request_or_inactivity > LEAVE");
325	return a_start_is_required;
326	}
327
328	//>
329	//<close_stream
330
331	static void close_stream()
332	{
333	SHOW_TIME("fifo > close_stream > ENTER\n");
334
335	// Warning: a wave_close can be already required by
336	// an external command (espeak_Cancel + fifo_stop), if so:
337	// my_stop_is_required = 1;
338
339	int a_status = pthread_mutex_lock(&my_mutex);
340	assert (!a_status);
341	int a_stop_is_required = my_stop_is_required;
342	if (!a_stop_is_required)
343	{
344	my_command_is_running = 1;
345	}
346	a_status = pthread_mutex_unlock(&my_mutex);
347
348	if (!a_stop_is_required)
349	{
350	wave_close(NULL);
351
352	int a_status = pthread_mutex_lock(&my_mutex);
353	assert (!a_status);
354	my_command_is_running = 0;
355
356	a_stop_is_required = my_stop_is_required;
357	a_status = pthread_mutex_unlock(&my_mutex);
358
359	if (a_stop_is_required)
360	{
361	// acknowledge the stop request
362	SHOW_TIME("fifo > close_stream > post my_sem_stop_is_acknowledged\n");
363	int a_status = sem_post(&my_sem_stop_is_acknowledged);
364	assert( a_status != -1);
365	}
366	}
367
368	SHOW_TIME("fifo > close_stream > LEAVE\n");
369	}
370
371	//>
372	//<say_thread
373
374	static void* say_thread(void*)
375	{
376	ENTER("say_thread");
377
378	SHOW_TIME("say_thread > post my_sem_stop_is_acknowledged\n");
379
380	// announce that thread is started
381	sem_post(&my_sem_stop_is_acknowledged);
382
383	int look_for_inactivity=0;
384
385	while(1)
386	{
387	SHOW_TIME("say_thread > wait for my_sem_start_is_required\n");
388
389	int a_start_is_required = 0;
390	if (look_for_inactivity)
391	{
392	a_start_is_required = sleep_until_start_request_or_inactivity();
393	if (!a_start_is_required)
394	{
395	close_stream();
396	}
397	}
398	look_for_inactivity = 1;
399
400	if (!a_start_is_required)
401	{
402	while ((sem_wait(&my_sem_start_is_required) == -1) && errno == EINTR)
403	{
404	continue; // Restart when interrupted by handler
405	}
406	}
407	SHOW_TIME("say_thread > get my_sem_start_is_required\n");
408
409	SHOW_TIME("say_thread > my_command_is_running = 1\n");
410	my_command_is_running = 1;
411
412	while( my_command_is_running)
413	{
414	SHOW_TIME("say_thread > locking\n");
415	int a_status = pthread_mutex_lock(&my_mutex);
416	assert (!a_status);
417	t_espeak_command* a_command = (t_espeak_command*)pop();
418
419	if (a_command == NULL)
420	{
421	SHOW_TIME("say_thread > text empty (talking=0) \n");
422	a_status = pthread_mutex_unlock(&my_mutex);
423	SHOW_TIME("say_thread > unlocked\n");
424	SHOW_TIME("say_thread > my_command_is_running = 0\n");
425	my_command_is_running = 0;
426	}
427	else
428	{
429	display_espeak_command(a_command);
430	// purge start semaphore
431	SHOW_TIME("say_thread > purge my_sem_start_is_required\n");
432	while(0 == sem_trywait(&my_sem_start_is_required))
433	{
434	};
435
436	if (my_stop_is_required)
437	{
438	SHOW_TIME("say_thread > my_command_is_running = 0\n");
439	my_command_is_running = 0;
440	}
441	SHOW_TIME("say_thread > unlocking\n");
442	a_status = pthread_mutex_unlock(&my_mutex);
443
444	if (my_command_is_running)
445	{
446	process_espeak_command(a_command);
447	}
448	delete_espeak_command(a_command);
449	}
450	}
451
452	if (my_stop_is_required)
453	{
454	// no mutex required since the stop command is synchronous
455	// and waiting for my_sem_stop_is_acknowledged
456	init();
457
458	// purge start semaphore
459	SHOW_TIME("say_thread > purge my_sem_start_is_required\n");
460	while(0==sem_trywait(&my_sem_start_is_required))
461	{
462	};
463
464	// acknowledge the stop request
465	SHOW_TIME("say_thread > post my_sem_stop_is_acknowledged\n");
466	int a_status = sem_post(&my_sem_stop_is_acknowledged);
467	assert( a_status != -1);
468	}
469	// and wait for the next start
470	SHOW_TIME("say_thread > wait for my_sem_start_is_required\n");
471	}
472
473	return NULL;
474	}
475
476	int fifo_is_command_enabled()
477	{
478	SHOW("ENTER fifo_is_command_enabled=%d\n",(int)(0 == my_stop_is_required));
479	return (0 == my_stop_is_required);
480	}
481
482	//>
483	//<fifo
484	typedef struct t_node
485	{
486	t_espeak_command* data;
487	t_node *next;
488	} node;
489
490	static node* head=NULL;
491	static node* tail=NULL;
492	// return 1 if ok, 0 otherwise
493	static espeak_ERROR push(t_espeak_command* the_command)
494	{
495	ENTER("fifo > push");
496
497	assert((!head && !tail) \|\| (head && tail));
498
499	if (the_command == NULL)
500	{
501	SHOW("push > command=0x%x\n", NULL);
502	return EE_INTERNAL_ERROR;
503	}
504
505	if (node_counter >= MAX_NODE_COUNTER)
506	{
507	SHOW("push > %s\n", "EE_BUFFER_FULL");
508	return EE_BUFFER_FULL;
509	}
510
511	node n = (node )malloc(sizeof(node));
512	if (n == NULL)
513	{
514	return EE_INTERNAL_ERROR;
515	}
516
517	if (head == NULL)
518	{
519	head = n;
520	tail = n;
521	}
522	else
523	{
524	tail->next = n;
525	tail = n;
526	}
527
528	tail->next = NULL;
529	tail->data = the_command;
530
531	node_counter++;
532	SHOW("push > counter=%d\n",node_counter);
533
534	the_command->state = CS_PENDING;
535	display_espeak_command(the_command);
536
537	return EE_OK;
538	}
539
540	static t_espeak_command* pop()
541	{
542	ENTER("fifo > pop");
543	t_espeak_command* the_command = NULL;
544
545	assert((!head && !tail) \|\| (head && tail));
546
547	if (head != NULL)
548	{
549	node* n = head;
550	the_command = n->data;
551	head = n->next;
552	free(n);
553	node_counter--;
554	SHOW("pop > command=0x%x (counter=%d)\n",the_command, node_counter);
555	}
556
557	if(head == NULL)
558	{
559	tail = NULL;
560	}
561
562	display_espeak_command(the_command);
563
564	return the_command;
565	}
566
567
568	static void init()
569	{
570	ENTER("fifo > init");
571	while (delete_espeak_command( pop() ))
572	{}
573	node_counter = 0;
574	}
575
576	//>
577	//<fifo_init
578	void fifo_terminate()
579	{
580	ENTER("fifo_terminate");
581
582	pthread_cancel(my_thread);
583	pthread_join(my_thread,NULL);
584	pthread_mutex_destroy(&my_mutex);
585	sem_destroy(&my_sem_start_is_required);
586	sem_destroy(&my_sem_stop_is_acknowledged);
587
588	init(); // purge fifo
589	}
590
591	#endif
592	//>
593