core pcm: use the same model for writing and reading audio

Reading sound frames relies on usage of poll before calling snd_pcm_writei
Recent seeks for bugs and CPU load issues have shown that the same thing
must be done on the write side.
Thus this commit uses the same mechanism, including the dedicated event
pipe for the thread termination.

Change-Id: Ifd5407c9e27da7801623583a6f7d4a2e845b43ea
Signed-off-by: Thierry Bultel <thierry.bultel@iot.bzh>

2 files changed, 177 insertions, 67 deletions

diff --git a/plugins/alsa/alsa-core-pcm.c b/plugins/alsa/alsa-core-pcm.c
index 3ef6012..e03c320 100644
--- a/plugins/alsa/alsa-core-pcm.c
+++ b/plugins/alsa/alsa-core-pcm.c
@@ -169,7 +169,7 @@ PUBLIC int AlsaPcmConf(SoftMixerT *mixer, AlsaPcmCtlT *pcm, int mode) {
 
     /* The following code, that
      * 1) sets period time/size; buffer time/size hardware params
-     * 2) sets start and stop threashold in software params
+     * 2) sets start and stop threshold in software params
      * ... is taken as such from 'aplay' in alsa-utils */
 
     unsigned buffer_time = 0;
@@ -275,6 +275,8 @@ PUBLIC int AlsaPcmConf(SoftMixerT *mixer, AlsaPcmCtlT *pcm, int mode) {
     // available_min is the minimum number of frames available to read,
     // when the call to poll returns. Assume this is the same for playback (not checked that)
 
+    AFB_API_DEBUG(mixer->api, "(%s) try AVAIL_MIN: %ld", card, n);
+
     if ((error = snd_pcm_sw_params_set_avail_min(pcm->handle, pxmSwParams, n)) < 0) {
         AFB_API_ERROR(mixer->api,
 					 "%s (%s): mixer=%s Fail set_buffersize error=%s",
@@ -283,11 +285,12 @@ PUBLIC int AlsaPcmConf(SoftMixerT *mixer, AlsaPcmCtlT *pcm, int mode) {
     };
 
     snd_pcm_sw_params_get_avail_min(pxmSwParams, &pcm->avail_min);
+    AFB_API_DEBUG(mixer->api, "(%s) AVAIL_MIN set to : %ld", card, pcm->avail_min);
 
     int start_delay = 0;
 	snd_pcm_uframes_t start_threshold;
 
-    /* round up to closest transfer boundary */
+    /* round up to closest transfer boundary. Start delay is in microseconds */
     n = buffer_size;
     if (start_delay <= 0) {
     	start_threshold = n + (size_t)((double)rate * start_delay / 1000000);
@@ -340,13 +343,14 @@ STATIC int AlsaPcmReadCB( AlsaPcmCopyHandleT * pcmCopyHandle) {
 	snd_pcm_uframes_t bufSize = alsa_ringbuf_buffer_size(rbuf);
 
 	int err;
+	const char * uid = ALSA_PCM_UID(pcmIn, string);
 
 	// do we have waiting frames ?
 	availIn = snd_pcm_avail_update(pcmIn);
 	if (availIn <= 0) {
 		if (availIn == -EPIPE) {
 			int ret = xrun(pcmIn, (int)availIn);
-			AFB_API_DEBUG(pcmCopyHandle->api, "XXX read EPIPE (recov=%d) {%s}!", ret, ALSA_PCM_UID(pcmIn, string));
+			AFB_API_DEBUG(pcmCopyHandle->api, "XXX read EPIPE (recov=%d) {%s}!", ret, uid);
 
 			// For some (undocumented...) reason, a start is mandatory.
 			snd_pcm_start(pcmIn);
@@ -376,14 +380,13 @@ STATIC int AlsaPcmReadCB( AlsaPcmCopyHandleT * pcmCopyHandle) {
 		pthread_mutex_unlock(&pcmCopyHandle->mutex);
 
 		nbRead = snd_pcm_readi(pcmIn, buf, remain);
-
 		if (nbRead == 0) {
 			break;
 		}
 		if (nbRead < 0) {
 			if (nbRead== -EPIPE) {
 				err = xrun(pcmIn, (int)nbRead);
-				AFB_API_DEBUG(pcmCopyHandle->api, "read EPIPE (%d), recov %d", ++pcmCopyHandle->read_err_count, err);
+				AFB_API_DEBUG(pcmCopyHandle->api, "read EPIPE (%d), recov %d {%s}", ++pcmCopyHandle->read_err_count, err, uid);
 				goto ExitOnSuccess;
 			} else if (nbRead== -ESTRPIPE) {
 				AFB_API_DEBUG(pcmCopyHandle->api, "read ESTRPIPE");
@@ -391,6 +394,7 @@ STATIC int AlsaPcmReadCB( AlsaPcmCopyHandleT * pcmCopyHandle) {
 					goto ExitOnSuccess;
 				nbRead = 0;
 			} else {
+				AFB_API_DEBUG(pcmCopyHandle->api, "read error: %ld (%s) {%s}", -nbRead, strerror((int)-nbRead), uid);
 				goto ExitOnSuccess;
 			}
 		}
@@ -422,7 +426,6 @@ ExitOnSuccess:
 static int xrun( snd_pcm_t * pcm, int error)
 {
 	int err;
-
 	if ((err = snd_pcm_recover(pcm, error, 1)) < 0) {
 		return err;
 	}
@@ -445,8 +448,8 @@ static int suspend( snd_pcm_t * pcm, int error)
 static void readSuspend(AlsaPcmCopyHandleT * pcmCopyHandle) {
 
 	// will be deaf
-	pcmCopyHandle->saveFd = pcmCopyHandle->pollFds[1].fd;
-	pcmCopyHandle->pollFds[1].fd = -1;
+	pcmCopyHandle->saveFd = pcmCopyHandle->pollFdsIn[1].fd;
+	pcmCopyHandle->pollFdsIn[1].fd = -1;
 
 	AFB_API_NOTICE(pcmCopyHandle->api, "capture muted");
 }
@@ -454,7 +457,7 @@ static void readSuspend(AlsaPcmCopyHandleT * pcmCopyHandle) {
 static void readResume(AlsaPcmCopyHandleT * pcmCopyHandle) {
 
 	// undeaf it
-	pcmCopyHandle->pollFds[1].fd = pcmCopyHandle->saveFd;
+	pcmCopyHandle->pollFdsIn[1].fd = pcmCopyHandle->saveFd;
 	snd_pcm_prepare(pcmCopyHandle->pcmIn->handle);
 	snd_pcm_start(pcmCopyHandle->pcmIn->handle);
 	AFB_API_NOTICE(pcmCopyHandle->api, "capture unmuted");
@@ -466,32 +469,26 @@ static void *readThreadEntry(void *handle) {
 
     AlsaPcmCopyHandleT *pcmCopyHandle = (AlsaPcmCopyHandleT*) handle;
     pcmCopyHandle->tid = (int) syscall(SYS_gettid);
-	int ix;
 
     AFB_API_NOTICE(pcmCopyHandle->api,
                   "%s :%s/%d Started, muted=%d",
                   __func__, pcmCopyHandle->info, pcmCopyHandle->tid, pcmCopyHandle->pcmIn->mute);
 
-	struct pollfd * eventFd =  &pcmCopyHandle->pollFds[0];
-	struct pollfd * framePfds = &pcmCopyHandle->pollFds[1];
-
-	eventFd->events  = POLLIN | POLLHUP;
+	struct pollfd * eventFd =  &pcmCopyHandle->pollFdsIn[0];
+	struct pollfd * framePfds = &pcmCopyHandle->pollFdsIn[1];
 
-	for (ix = 0; ix <pcmCopyHandle->nbPcmFds-1; ix++) {
-		framePfds[ix].events = POLLIN | POLLHUP;
-	}
+	eventFd->events  = POLLIN;
 
    	bool muted = pcmCopyHandle->pcmIn->mute;
 
   	if (muted)
    		readSuspend(pcmCopyHandle);
 
-
     /* loop until end */
     for (;;) {
 
-  	   	int err = poll(pcmCopyHandle->pollFds, pcmCopyHandle->nbPcmFds, LOOP_TIMEOUT_MSEC);
-    	if (err < 0) {
+		int err = poll(pcmCopyHandle->pollFdsIn, pcmCopyHandle->nbPcmFdsIn, LOOP_TIMEOUT_MSEC);
+		if (err < 0) {
     		AFB_API_ERROR(pcmCopyHandle->api, "%s: poll err %s", __func__, strerror(errno));
     		continue;
     	}
@@ -503,7 +500,7 @@ static void *readThreadEntry(void *handle) {
     	}
 
 		// handle the incoming events/mute order
-		if ((eventFd->revents & EPOLLIN) != 0) {
+		if ((eventFd->revents & POLLIN) != 0) {
 			PcmCopyEvent event;
 
 			size_t ret = read(eventFd->fd, &event, sizeof(event));
@@ -537,8 +534,7 @@ static void *readThreadEntry(void *handle) {
 
 		unsigned short revents = 0;
 
-		int res = snd_pcm_poll_descriptors_revents(pcmCopyHandle->pcmIn->handle, framePfds, pcmCopyHandle->nbPcmFds, &revents);
-
+		int res = snd_pcm_poll_descriptors_revents(pcmCopyHandle->pcmIn->handle, framePfds, pcmCopyHandle->nbPcmFdsIn-1, &revents);
 		if (res == -ENODEV) {
     		sleep(1);
     		continue;
@@ -549,6 +545,11 @@ static void *readThreadEntry(void *handle) {
     		continue;
     	}
 
+		if (!(revents & POLLIN)) {
+			AFB_API_DEBUG(pcmCopyHandle->api, "no POLLIN. try again !");
+			continue;
+		}
+
 		AlsaPcmReadCB(pcmCopyHandle);
     }
 done:
@@ -564,23 +565,14 @@ static void *writeThreadEntry(void *handle) {
 
 	alsa_ringbuf_t * rbuf = pcmCopyHandle->rbuf;
 
-	snd_pcm_status_t *pcmOutStatus;
-	snd_pcm_uframes_t pcmOutSize;
-
-	snd_pcm_sframes_t threshold;
-
-	snd_pcm_status_alloca(&pcmOutStatus);
-	snd_pcm_status(pcmOut, pcmOutStatus);
-	pcmOutSize = snd_pcm_status_get_avail_max(pcmOutStatus);
-
 	const char * cardid = pcmCopyHandle->pcmOut->cid.cardid;
 
-	/* This threshold is the expected space available in the hw output buffer
-	 * The aim is to wait to have a significant amount of space, in order to
-	 * avoid to write to the device too often, or take a very small amount of
-	 * frames from the ring buffer. So basically, this saves some CPU load */
+	struct pollfd * eventFd =  &pcmCopyHandle->pollFdsOut[0];
+	struct pollfd * framePfds = &pcmCopyHandle->pollFdsOut[1];
 
-	threshold = pcmOutSize / 3;
+	eventFd->events  = POLLIN;
+
+	const char * name = pcmCopyHandle->pcmOut->name;
 
 	for (;;) {
 
@@ -593,6 +585,62 @@ static void *writeThreadEntry(void *handle) {
 				goto done;
 			}
 
+			int err = poll(pcmCopyHandle->pollFdsOut, pcmCopyHandle->nbPcmFdsOut, LOOP_TIMEOUT_MSEC);
+			if (err < 0) {
+				AFB_API_ERROR(pcmCopyHandle->api, "%s: %s poll out err %s", __func__, name, strerror(errno));
+				continue;
+			}
+
+			// timeout
+			if (err == 0)
+				continue;
+
+			// handle the incoming events/mute order
+			if ((eventFd->revents & POLLIN) != 0) {
+				PcmCopyEvent event;
+
+				size_t ret = read(eventFd->fd, &event, sizeof(event));
+				if (ret <= 0)
+					continue;
+
+				switch (event.eventType) {
+				case PCM_COPY_MUTE:
+				case PCM_COPY_UNMUTE:
+					break;
+				case PCM_COPY_END:
+					AFB_API_DEBUG(pcmCopyHandle->api, "%s ending -> EXIT", __func__);
+					goto done;
+					break;
+				case PCM_COPY_LAST:
+				default:
+					AFB_API_ERROR(pcmCopyHandle->api, "%s: Unexpected event 0x%x", __func__, event.eventType);
+					break;
+				}
+				continue;
+			}
+
+			unsigned short revents = 0;
+
+			int res = snd_pcm_poll_descriptors_revents(pcmCopyHandle->pcmOut->handle, framePfds, pcmCopyHandle->nbPcmFdsOut-1, &revents);
+			if (res == -ENODEV) {
+				sleep(1);
+				continue;
+			}
+
+			if (revents & POLLHUP) {
+				AFB_API_NOTICE(pcmCopyHandle->api, "Frame POLLHUP");
+				continue;
+			}
+
+			// not ready ... wait again
+			if ((revents & POLLOUT) == 0) {
+				// When that quirk is set, continuing to poll again would block too much time
+				// and led to a EPIPE
+				// Instead, we ignore the "non ready state" and perform the write directly
+				if ((pcmCopyHandle->pcmOut->quirks & QUIRK_BOGUS_POLL_REVENTS_DEMANGLING) == 0)
+					continue;
+			}
+
 			snd_pcm_sframes_t used, nbWritten;
 			snd_pcm_sframes_t availOut = snd_pcm_avail(pcmOut);
 
@@ -609,15 +657,9 @@ static void *writeThreadEntry(void *handle) {
 				}
 			}
 
-			// no space for output
-			if (availOut <= threshold) {
-				usleep(500);
-				continue;
-			}
-
 			pthread_mutex_lock(&pcmCopyHandle->mutex);
 			used = alsa_ringbuf_frames_used(rbuf);
-			if (used <= 0) {
+			if (used <= 0) { /* no more frames*/
 				pthread_mutex_unlock(&pcmCopyHandle->mutex);
 				break; // will wait again
 			}
@@ -640,8 +682,11 @@ static void *writeThreadEntry(void *handle) {
 				} else if (nbWritten == -ESTRPIPE) {
 					AFB_API_DEBUG(pcmCopyHandle->api, "XXX write ESTRPIPE");
 					break;
+				} else if (nbWritten == 0) {
+					AFB_API_DEBUG(pcmCopyHandle->api, "%s: nothing was written", __func__);
+					break;
 				}
-				AFB_API_DEBUG(pcmCopyHandle->api, "Unhandled error %s", strerror(errno));
+				AFB_API_DEBUG(pcmCopyHandle->api, "%s: Unhandled error %s", __func__, strerror((int)-nbWritten));
 				break;
 			}
 
@@ -676,8 +721,11 @@ static int AlsaPcmCopyEnd(SoftMixerT *mixer, AlsaPcmCopyHandleT * handle) {
 	handle->ending = true;
 	// wake up the reader through the eventfd
 	AlsaPcmCopyEndSignal(mixer, handle->pcmIn);
+
 	// wake up the writer through the wait semaphore
 	sem_post(&handle->sem);
+	// and make in exit from poll
+	AlsaPcmCopyEndSignal(mixer, handle->pcmOut);
 
 	return 0;
 }
@@ -700,7 +748,8 @@ PUBLIC int AlsaPcmCopyStop(SoftMixerT *mixer, AlsaPcmCopyHandleT * handle) {
 
 	alsa_ringbuf_free(handle->rbuf);
 
-	free(handle->pollFds);
+	free(handle->pollFdsIn);
+	free(handle->pollFdsOut);
 	free(handle);
 
 	return 0;
@@ -791,7 +840,7 @@ PUBLIC int AlsaPcmCopyStart(SoftMixerT *mixer, AlsaStreamAudioT *stream, AlsaPcm
 	cHandle->stream = stream;
 
 	cHandle->frame_size = (snd_pcm_format_physical_width(opts->format) / 8) * opts->channels;
-	AFB_API_DEBUG(mixer->api, "%s: Frame size is %zu", __func__, cHandle->frame_size);
+	AFB_API_DEBUG(mixer->api, "%s: Frame size is %zu (%d channels)", __func__, cHandle->frame_size, opts->channels);
 	AFB_API_DEBUG(mixer->api, "%s: Buffer delay is %d ms", __func__, stream->delayms);
 
 	snd_pcm_uframes_t nbFrames = (stream->delayms * opts->rate)/1000;
@@ -807,6 +856,29 @@ PUBLIC int AlsaPcmCopyStart(SoftMixerT *mixer, AlsaStreamAudioT *stream, AlsaPcm
 
 	AFB_API_DEBUG(mixer->api, "%s Copy buffer: nbframes is %zu", __func__, nbFrames);
 
+	// get FD poll descriptor for playback PCM
+	int pcmOutCount = snd_pcm_poll_descriptors_count(pcmOut->handle);
+	if (pcmOutCount <= 0) {
+	    AFB_API_ERROR(mixer->api,
+	                 "%s: Fail pcmOut=%s get fds count error=%s",
+	                 __func__, ALSA_PCM_UID(pcmOut->handle, string), snd_strerror(error));
+	    goto OnErrorExit;
+	}
+
+	cHandle->nbPcmFdsOut = pcmOutCount+1;
+	cHandle->pollFdsOut = (struct pollfd *) malloc((cHandle->nbPcmFdsOut)*sizeof(struct pollfd));
+	if (cHandle->pollFdsOut == NULL){
+		SOFTMIXER_NOMEM(mixer->api);
+	    goto OnErrorExit;
+	}
+
+	if ((error = snd_pcm_poll_descriptors(pcmOut->handle, cHandle->pollFdsOut+1, pcmOutCount)) < 0) {
+		AFB_API_ERROR(mixer->api,
+	                 "%s: Fail pcmOut=%s get pollfds error=%s",
+	                 __func__, ALSA_PCM_UID(pcmOut->handle, string), snd_strerror(error));
+	    goto OnErrorExit;
+	};
+
     // get FD poll descriptor for capture PCM
     int pcmInCount = snd_pcm_poll_descriptors_count(pcmIn->handle);
     if (pcmInCount <= 0) {
@@ -816,43 +888,61 @@ PUBLIC int AlsaPcmCopyStart(SoftMixerT *mixer, AlsaStreamAudioT *stream, AlsaPcm
         goto OnErrorExit;
 	}
 
-	cHandle->nbPcmFds = pcmInCount+1;
-	cHandle->pollFds = (struct pollfd *) malloc((cHandle->nbPcmFds)*sizeof(struct pollfd));
-	if (cHandle->pollFds == NULL){
+	cHandle->nbPcmFdsIn = pcmInCount+1;
+	cHandle->pollFdsIn = (struct pollfd *) malloc((cHandle->nbPcmFdsIn)*sizeof(struct pollfd));
+	if (cHandle->pollFdsIn == NULL){
 		SOFTMIXER_NOMEM(mixer->api);
     	goto OnErrorExit;
     }
 
-	if ((error = snd_pcm_poll_descriptors(pcmIn->handle, cHandle->pollFds+1, pcmInCount)) < 0) {
+	if ((error = snd_pcm_poll_descriptors(pcmIn->handle, cHandle->pollFdsIn+1, pcmInCount)) < 0) {
         AFB_API_ERROR(mixer->api,
                      "%s: Fail pcmIn=%s get pollfds error=%s",
                      __func__, ALSA_PCM_UID(pcmOut->handle, string), snd_strerror(error));
         goto OnErrorExit;
     };
 
-    // create the mute pipe
-	int eventFdPipe[2];
-	error = pipe(eventFdPipe);
+    // create the event pipe for capture
+	int eventFdPipeIn[2];
+	error = pipe(eventFdPipeIn);
     if (error < 0) {
         AFB_API_ERROR(mixer->api,
-                     "Unable to create the mute signaling pipe");
+                     "Unable to create the mute signaling pipe for capture");
         goto OnErrorExit;
     }
 
-	struct pollfd * eventPollFd = &cHandle->pollFds[0];
+	struct pollfd * eventPollFdIn = &cHandle->pollFdsIn[0];
     // read end
-	eventPollFd->fd = eventFdPipe[0];
-	eventPollFd->events = POLLIN;
-	eventPollFd->revents = 0;
+	eventPollFdIn->fd = eventFdPipeIn[0];
+	eventPollFdIn->events = POLLIN;
+	eventPollFdIn->revents = 0;
 
     // write end
-	pcmIn->eventFd = eventFdPipe[1];
+	pcmIn->eventFd = eventFdPipeIn[1];
+
+	// create the event pipe for playback
+	int eventFdPipeOut[2];
+	error = pipe(eventFdPipeOut);
+	if (error < 0) {
+	    AFB_API_ERROR(mixer->api,
+	                "Unable to create the signaling pipe for playback");
+	    goto OnErrorExit;
+	 }
+
+	struct pollfd * eventPollFdOut = &cHandle->pollFdsOut[0];
+	// read end
+	eventPollFdOut->fd = eventFdPipeOut[0];
+	eventPollFdOut->events = POLLIN;
+	eventPollFdOut->revents = 0;
+
+	// write end
+	pcmOut->eventFd = eventFdPipeOut[1];
 
     error = sem_init(&cHandle->sem, 0 , 0);
     if (error < 0) {
     	AFB_API_ERROR(mixer->api,
-    	                     "%s Fail initialize loop semaphore pcmIn=%s err=%d",
-    	                     __func__, ALSA_PCM_UID(pcmIn->handle, string), error);
+					 "%s Fail to initialize the loop semaphore pcmIn=%s err=%d",
+					 __func__, ALSA_PCM_UID(pcmIn->handle, string), error);
     	goto OnErrorExit;
     }
 
@@ -910,11 +1000,28 @@ OnErrorExit:
     AFB_API_ERROR(mixer->api, "%s: - pcmIn=%s" , __func__, ALSA_PCM_UID(pcmIn->handle, string));
     AFB_API_ERROR(mixer->api, "%s: - pcmOut=%s", __func__, ALSA_PCM_UID(pcmOut->handle, string));
 
-	if (cHandle &&cHandle->pollFds) {
-		free (cHandle->pollFds);
-		cHandle->pollFds = NULL;
+	if (cHandle &&cHandle->pollFdsIn) {
+		free (cHandle->pollFdsIn);
+		cHandle->pollFdsIn = NULL;
+	}
+
+	if (cHandle && cHandle->pollFdsOut) {
+		free (cHandle->pollFdsOut);
+		cHandle->pollFdsOut = NULL;
 	}
 
+	if (eventFdPipeIn[0])
+		close(eventFdPipeIn[0]);
+
+	if (eventFdPipeIn[1])
+		close(eventFdPipeIn[1]);
+
+	if (eventFdPipeOut[0])
+		close(eventFdPipeOut[0]);
+
+	if (eventFdPipeOut[1])
+		close(eventFdPipeOut[1]);
+
 	if (cHandle)
 		free(cHandle);
 
diff --git a/plugins/alsa/alsa-softmixer.h b/plugins/alsa/alsa-softmixer.h
index 9710dd3..384d1b0 100644
--- a/plugins/alsa/alsa-softmixer.h
+++ b/plugins/alsa/alsa-softmixer.h
@@ -158,8 +158,11 @@ typedef struct {
     int tid;
     char* info;
 
-    int nbPcmFds;
-	struct pollfd * pollFds;
+    int nbPcmFdsIn;
+	struct pollfd * pollFdsIn;
+
+	int nbPcmFdsOut;
+	struct pollfd * pollFdsOut;
 
     sem_t sem;
     pthread_mutex_t mutex;