//.........这里部分代码省略.........

      ierr("ERROR: Unsupported read size: %d\n", len);
      return -ENOSYS;
    }

  /* Get exclusive access to the driver data structure */

  ret = sem_wait(&priv->devsem);
  if (ret < 0)
    {
      /* This should only happen if the wait was cancelled by an signal */

      ierr("ERROR: sem_wait: %d\n", errno);
      DEBUGASSERT(errno == EINTR);
      return -EINTR;
    }

  /* Try to read sample data. */

  ret = max11802_sample(priv, &sample);
  if (ret < 0)
    {
      /* Sample data is not available now.  We would ave to wait to get
       * receive sample data.  If the user has specified the O_NONBLOCK
       * option, then just return an error.
       */

      iinfo("Sample data is not available\n");
      if (filep->f_oflags & O_NONBLOCK)
        {
          ret = -EAGAIN;
          goto errout;
       }

      /* Wait for sample data */

      ret = max11802_waitsample(priv, &sample);
      if (ret < 0)
        {
          /* We might have been awakened by a signal */

          ierr("ERROR: max11802_waitsample: %d\n", ret);
          goto errout;
        }
    }

  /* In any event, we now have sampled MAX11802 data that we can report
   * to the caller.
   */

  report = (FAR struct touch_sample_s *)buffer;
  memset(report, 0, SIZEOF_TOUCH_SAMPLE_S(1));
  report->npoints            = 1;
  report->point[0].id        = sample.id;
  report->point[0].x         = sample.x;
  report->point[0].y         = sample.y;

  /* Report the appropriate flags */

  if (sample.contact == CONTACT_UP)
    {
      /* Pen is now up.  Is the positional data valid?  This is important to
       * know because the release will be sent to the window based on its
       * last positional data.
       */

      if (sample.valid)
        {
          report->point[0].flags  = TOUCH_UP | TOUCH_ID_VALID | TOUCH_POS_VALID;
        }
      else
        {
          report->point[0].flags  = TOUCH_UP | TOUCH_ID_VALID;
        }
    }
  else if (sample.contact == CONTACT_DOWN)
    {
      /* First contact */

      report->point[0].flags  = TOUCH_DOWN | TOUCH_ID_VALID | TOUCH_POS_VALID;
    }
  else /* if (sample->contact == CONTACT_MOVE) */
    {
      /* Movement of the same contact */

      report->point[0].flags  = TOUCH_MOVE | TOUCH_ID_VALID | TOUCH_POS_VALID;
    }

  iinfo("  id:      %d\n", report->point[0].id);
  iinfo("  flags:   %02x\n", report->point[0].flags);
  iinfo("  x:       %d\n", report->point[0].x);
  iinfo("  y:       %d\n", report->point[0].y);

  ret = SIZEOF_TOUCH_SAMPLE_S(1);

errout:
  sem_post(&priv->devsem);
  iinfo("Returning: %d\n", ret);
  return ret;
}

//.........这里部分代码省略.........
  int ret;

  fvdbg("Write %d bytes to offset %d\n", buflen, filep->f_pos);

  /* Sanity checks */

  DEBUGASSERT(filep->f_priv != NULL && filep->f_inode != NULL);

  /* Recover the open file state from the struct file instance */

  wrfile = (FAR struct nxffs_wrfile_s *)filep->f_priv;

  /* Recover the volume state from the open file */

  volume = (FAR struct nxffs_volume_s *)filep->f_inode->i_private;
  DEBUGASSERT(volume != NULL);

  /* Get exclusive access to the volume.  Note that the volume exclsem
   * protects the open file list.
   */

  ret = sem_wait(&volume->exclsem);
  if (ret != OK)
    {
      ret = -get_errno();
      fdbg("ERROR: sem_wait failed: %d\n", ret);
      goto errout;
    }

  /* Check if the file was opened with write access */

  if ((wrfile->ofile.oflags & O_WROK) == 0)
    {
      fdbg("ERROR: File not open for write access\n");
      ret = -EACCES;
      goto errout_with_semaphore;
    }

  /* Loop until we successfully appended all of the data to the file (or an
   * error occurs)
   */

  for (total = 0; total < buflen; )
    {
      remaining = buflen - total;

      /* Have we already allocated the data block? */

      if (wrfile->doffset == 0)
        {
          /* No, allocate the data block now, re-packing if necessary. */

          wrfile->datlen = 0;
          ret = nxffs_wralloc(volume, wrfile, remaining);
          if (ret < 0)
            {
              fdbg("ERROR: Failed to allocate a data block: %d\n", -ret);
              goto errout_with_semaphore;
            }
        }

      /* Seek to the FLASH block containing the data block */

      nxffs_ioseek(volume, wrfile->doffset);

      /* Verify that the FLASH data that was previously written is still intact */

      ret = nxffs_reverify(volume, wrfile);
      if (ret < 0)
        {
          fdbg("ERROR: Failed to verify FLASH data block: %d\n", -ret);
          goto errout_with_semaphore;
        }

      /* Append the data to the end of the data block and write the updated
       * block to flash.
       */

      nwritten = nxffs_wrappend(volume, wrfile, &buffer[total], remaining);
      if (nwritten < 0)
        {
          fdbg("ERROR: Failed to append to FLASH to a data block: %d\n", -ret);
          goto errout_with_semaphore;
        }

      /* Decrement the number of bytes remaining to be written */

      total += nwritten;
    }

  /* Success.. return the number of bytes written */

  ret           = total;
  filep->f_pos  = wrfile->datlen;

errout_with_semaphore:
  sem_post(&volume->exclsem);
errout:
  return ret;
}

 /*
 This is a simulation of the process diagram
 on slide 38 module 06-Concurrency
 */
 void * p( void * arg )
 {
   unsigned k , mynum ;
   mynum = ( unsigned ) arg ;

   pthread_detach( pthread_self() );

   switch( mynum )
   {
     case 1:
	 printf("Thread %d started\n", mynum);
	 SPIN ;
	 printf("Thread %d ended\n", mynum);
	 sem_post( & s[1] /* or simply s+1 */ );
	 sem_post( s+1 );
	 sem_post( s+1 );
	 break ;

     case 2:
	 sem_wait( s+1 ) ;
	 printf("Thread %d started\n", mynum);
	 SPIN ;
	 printf("Thread %d ended\n", mynum);
	 sem_post( s+2 );
	 sem_post( s+2 );
	 break ;

     case 3:
	 sem_wait( s+2 ) ;
	 printf("Thread %d started\n", mynum);
	 SPIN ;
	 printf("Thread %d ended\n", mynum);
	 sem_post( s+3 );
	 break ;

     case 4:
	 sem_wait( s+2 ) ;
	 printf("Thread %d started\n", mynum);
	 SPIN ;
	 printf("Thread %d ended\n", mynum);
	 sem_post( s+4 );
	 break ;

     case 5:
	 sem_wait( s+1 ) ;
	 printf("Thread %d started\n", mynum);
	 SPIN ;
	 printf("Thread %d ended\n", mynum);
	 sem_post( s+5 );
	 break ;

     case 6:
	 sem_wait( s+4 ) ;
	 sem_wait( s+5 ) ;
	 printf("Thread %d started\n", mynum);
	 SPIN ;
	 printf("Thread %d ended\n", mynum);
	 sem_post( s+6 );
	 break ;

     case 7:
	 sem_wait( s+1 ) ;
	 printf("Thread %d started\n", mynum);
	 SPIN ;
	 printf("Thread %d ended\n", mynum);
	 sem_post( s+7 );
	 break ;

     case 8:
	 sem_wait( s+3 ) ;
	 sem_wait( s+6 ) ;
	 sem_wait( s+7 ) ;
	 printf("Thread %d started\n", mynum);
	 SPIN ;
	 printf("Thread %d ended\n", mynum);
	 sem_post( s+8 );
	 break ;
   };

   pthread_exit( NULL ) ;
 }

DttSP_EXP void
Audio_Callback2 (float **input, float **output, unsigned int nframes)
{
	unsigned int thread;
	BOOLEAN b = reset_em;
	BOOLEAN return_empty=FALSE;
	unsigned int i;
	for(thread=0;thread<threadno;thread++)
	{
		if (top[thread].susp) return_empty = TRUE;
	}

	if (return_empty)
	{
		for(thread=0;thread<threadno;thread++) 
		{
			memset (output[2*thread], 0, nframes * sizeof (float));
			memset (output[2*thread+1], 0, nframes * sizeof (float));
		}
		return;
	}

	if (b)
	{
		//fprintf(stderr, "reset_em!\n"); fflush(stderr);
		//fprintf(stdout,"Audio_Callback2: reset_em = TRUE\n"), fflush(stdout);
		reset_system_audio(nframes);
		for(thread=0;thread<threadno;thread++) {
			memset (output[2*thread], 0, nframes * sizeof (float));
			memset (output[2*thread+1], 0, nframes * sizeof (float));
		}
		return;
    }
#if 0
	if (diversity.flag) {
		// Deal with the transmitter first
		if ((ringb_float_read_space (top[1].jack.ring.o.l) >= nframes)
			&& (ringb_float_read_space (top[1].jack.ring.o.r) >= nframes))
		{
			ringb_float_read (top[1].jack.ring.o.l, output[2], nframes);
			ringb_float_read (top[1].jack.ring.o.r, output[3], nframes);
		}
		else
		{	
			// rb pathology
			//reset_system_audio(nframes);
			for(thread=0;thread<threadno;thread++) 
			{
				memset (output[thread], 0, nframes * sizeof (float));
				memset (output[thread], 0, nframes * sizeof (float));
			}
			return;
		}

		// input: copy from port to ring
		if ((ringb_float_write_space (top[1].jack.ring.i.l) >= nframes)
			&& (ringb_float_write_space (top[1].jack.ring.i.r) >= nframes))
		{
			ringb_float_write (top[1].jack.ring.i.l, input[2], nframes);
			ringb_float_write (top[1].jack.ring.i.r, input[3], nframes);
		}
		else
		{	
			// rb pathology
			for(thread=0;thread<threadno;thread++) 
			{
				memset (output[thread], 0, nframes * sizeof (float));
				memset (output[thread], 0, nframes * sizeof (float));
			}
			return;
		}
		
		// if enough accumulated in ring, fire dsp
		if ((ringb_float_read_space (top[1].jack.ring.i.l) >= top[1].hold.size.frames) &&
			(ringb_float_read_space (top[1].jack.ring.i.r) >= top[1].hold.size.frames))
			sem_post (&top[1].sync.buf.sem);

		//		
		// Deal with the diversity channel next
		//
		if ((ringb_float_read_space (top[0].jack.ring.o.l) >= nframes)
			&& (ringb_float_read_space (top[0].jack.ring.o.r) >= nframes))
		{
			/*ringb_float_read (top[thread].jack.auxr.o.l, output[l], nframes);
			ringb_float_read (top[thread].jack.auxr.o.r, output[r], nframes);*/
			ringb_float_read (top[0].jack.ring.o.l, output[2], nframes);
			ringb_float_read (top[0].jack.ring.o.r, output[3], nframes);
		}
		else
		{	
			// rb pathology
			//reset_system_audio(nframes);
			for(thread=0;thread<threadno;thread++) 
			{
				memset (output[thread], 0, nframes * sizeof (float));
				memset (output[thread], 0, nframes * sizeof (float));
			}
			return;
		}

//.........这里部分代码省略.........

int main(int argc, char *argv[])
{
  int n_elements = 61;
  int n_processors = 10;

  // First calculate how much work to give to each process.
  // Really, this is just a dummy example. Each child will just
  // do some meaningless calculations as an example.
  int n_per_processor = n_elements / n_processors;

  int n_remaining = n_elements % n_processors;

  int node_start = 0;
  int node_end;
  printf("\n");

  // Here we read the key for the shared memory. This will be availble
  // to all clients so they can attach to shared memory.
  // Remember, each child will inherit an exact copy of the stack and
  // heap.
  //
  //
  int shm_fd = shm_open("/wikastarea1", O_RDWR, 0755);
  if (shm_fd < 0)
  {
    perror("shm_open");
    exit(0);
  }
  void *mptr = mmap(NULL, getpagesize(), PROT_READ|PROT_WRITE,
                   MAP_SHARED, shm_fd,   0x0);
  if (mptr == MAP_FAILED)
  {
    perror("mmap");
    exit(0);
  }

  shdata_t *shdata = (shdata_t *) mptr;

  sem_t *sem1 = sem_open("/wikassem1", 0);
  if (sem1 == SEM_FAILED)
  {
    perror("sem_open sem1");
    exit(0);
  }

  sem_t *sem2 = sem_open("/wikassem2", 0);
  if (sem2 == SEM_FAILED)
  {
    perror("sem_open sem2");
    exit(0);
  }


  int count = 0;
  for (int i=0; i < 100; i++)
  {
    sem_wait(sem2);

    printf("Damn! the pot is empty!\n");
    for (int j=0;  j < 10; j++)
    {
      shdata->values[j] = 1.0;
      printf("producing %d\n", ++count);
    }

    sem_post(sem1);
  }

}

static gsize get_mpeg2ts_segment (RequestData *request_data, EncoderOutput *encoder_output, gchar **buf)
{
    GstClockTime timestamp;
    gint number;
    guint64 year, month, mday, hour, sequence, duration, rap_addr, max_age;
    GstClockTime us; /* microseconds */
    struct tm tm;
    gchar date[20], *header, *path, *file, *cache_control;
    gsize buf_size;
    GError *err = NULL;
    struct timespec ts;

    number = sscanf (request_data->uri, "/%*[^/]/encoder/%*[^/]/%04lu%02lu%02lu%02lu/%lu_%lu_%lu.ts$",
            &year, &month, &mday, &hour, &us, &sequence, &duration);
    if (number != 7) {
        GST_WARNING ("uri not found: %s", request_data->uri);
        *buf = g_strdup_printf (http_404, PACKAGE_NAME, PACKAGE_VERSION);
        buf_size = strlen (*buf);
        return buf_size;
    }
    sprintf (date, "%04lu-%02lu-%02lu %02lu:00:00", year, month, mday, hour);
    memset (&tm, 0, sizeof (struct tm));
    strptime (date, "%Y-%m-%d %H:%M:%S", &tm);
    tm.tm_isdst = daylight;
    timestamp = mktime (&tm) * 1000000 + us;

    /* read from memory */
    if (clock_gettime (CLOCK_REALTIME, &ts) == -1) {
        GST_ERROR ("get_mpeg2ts_segment clock_gettime error: %s", g_strerror (errno));
        *buf = g_strdup_printf (http_500, PACKAGE_NAME, PACKAGE_VERSION);
        buf_size = strlen (*buf);
        return buf_size;
    }
    ts.tv_sec += 2;
    while (sem_timedwait (encoder_output->semaphore, &ts) == -1) {
        if (errno == EINTR) {
            continue;
        }
        GST_ERROR ("get_mpeg2ts_segment sem_timedwait failure: %s", g_strerror (errno));
        *buf = g_strdup_printf (http_500, PACKAGE_NAME, PACKAGE_VERSION);
        buf_size = strlen (*buf);
        return buf_size;
    }
    /* seek gop */
    rap_addr = encoder_output_gop_seek (encoder_output, timestamp);
    if (rap_addr != G_MAXUINT64) {
        /* segment found, send it */
        gsize gop_size;

        gop_size = encoder_output_gop_size (encoder_output, rap_addr);
        cache_control = g_strdup_printf ("max-age=%lu", encoder_output->dvr_duration);
        header = g_strdup_printf (http_200, PACKAGE_NAME, PACKAGE_VERSION, "video/mpeg", gop_size, cache_control, "");
        g_free (cache_control);
        *buf = g_malloc (strlen (header) + gop_size);
        memcpy (*buf, header, strlen(header));
        if (rap_addr + gop_size + 12 < encoder_output->cache_size) {
            memcpy (*buf + strlen (header), encoder_output->cache_addr + rap_addr + 12, gop_size);

        } else {
            gint n;

            n = encoder_output->cache_size - rap_addr - 12;
            if (n > 0) {
                memcpy (*buf + strlen (header), encoder_output->cache_addr + rap_addr + 12, n);
                memcpy (*buf + strlen (header) + n, encoder_output->cache_addr, gop_size - n);

            } else {
                GST_WARNING ("nnnnn: n < 0 %d", n);
                memcpy (*buf + strlen (header), encoder_output->cache_addr - n, gop_size);
            }
        }
        buf_size = strlen (header) + gop_size;
        g_free (header);

    } else {
        buf_size = 0;
    }
    sem_post (encoder_output->semaphore);

    /* buf_size == 0? segment not found in memory, read frome dvr directory */
    if (buf_size == 0) {
        path = g_strdup_printf ("%s/%04lu%02lu%02lu%02lu/%010lu_%lu_%lu.ts",
                encoder_output->record_path,
                year, month, mday, hour, us, sequence, duration);
        if (!g_file_get_contents (path, &file, &buf_size, &err)) {
            g_error_free (err);
            GST_WARNING ("segment not found: %s", request_data->uri);
            *buf = g_strdup_printf (http_404, PACKAGE_NAME, PACKAGE_VERSION);
            buf_size = strlen (*buf);

        } else {
            max_age = encoder_output->dvr_duration - (g_get_real_time () - timestamp) / 1000000;
            cache_control = g_strdup_printf ("max-age=%lu", max_age);
            header = g_strdup_printf (http_200,
                    PACKAGE_NAME,
                    PACKAGE_VERSION,
                    "video/mpeg",
                    buf_size,
                    cache_control,
                    "");
//.........这里部分代码省略.........

// Serves as a global method to trigger reinitialization
// and as a function that can be provided to signal().
void reinit_signal_handler(int /*signal*/) {
    sem_post(&reinit);
}

int service_engine(char *addr, int port)
{
    int i;
    if(MAX_TASK_NUM > 0)
    {
        for (i = 0; i < MAX_TASK_NUM; i++)
        {
            sem_init(&event[i], 0, 0);
            int temp = i;
           if (pthread_create(&thread_id[i], NULL, (void *)handle_requests, (void *)temp) != 0)
            {
                printf("pthread_create error, %s:%d\n",__FILE__, __LINE__);
                exit(-1);
            }

        }
    }
    /* 
     * command line console
     */

    init_mdb();

    ServiceHandler h = -1;
    initialize_service(addr,port);
    while(1)
    {
        h = service_start();
        //        handle_request(h);  
        task_node_t *tnode = (task_node_t *)malloc( sizeof(task_node_t));
        tnode->buf_size = MAX_BUF_LEN;
        if (receive_data (h, tnode->buf, &(tnode->buf_size)) == 0)
        {
            service_stop(h);
            continue;
        }

        if (MAX_TASK_NUM > 0)
        {
            tnode->req = h;
            int i = random_int(MAX_TASK_NUM);
            /* put tnode into task_list */
            if (task_list[i] == NULL)
            {
                task_list[i] = tnode;
                tnode->next = NULL;
                tnode->last = tnode;
            }
            else
            {
                task_node_t *p = task_list[i];
                p->last->next = tnode;
                p->last = tnode;
                tnode->next = NULL;
            }
            sem_post(&event[i]);

            //        service_stop(h); 
        }
        else 
        {
            handle_one_request(h, tnode->buf, tnode->buf_size);
            free(tnode);
        }
    }
    shutdown_service();
    close_mdb();
    if (MAX_TASK_NUM > 0 )
    {
        for (i = 0; i< MAX_TASK_NUM; i++)
        {
            task_node_t *p = task_list[i];
            task_node_t *q;
            while (p)
            {
                q = p->next;
                free(p);
                p = q;
            }
        }
    }
    return 0;
}

void threadfunc()
{
        printf("Hello from threadfunc!\n");
        sleep(3);
        sem_post(&g_sema);
}

 virtual void readDone(uint32_t v) {
   fprintf(stderr, "readDone %d\n", v);
   sem_post(&read_sem);
 }

/*
 * La cola RR solo la agregamos por compatibilidad,
 * siempre va a ser un puntero a NULL.
 */
uint32_t atender_peticion_orquestador(uint32_t cliente, t_stream *pedido, t_queue_rr *cola_rr){

	/*t_stream *pedido;
	recibir(cliente, (void*)&pedido);
*/
	t_header header;
	deserializar_header(pedido, &header);
	int32_t pos_payload = sizeof(t_header);
	//char personajeElegido;
	switch (header.type){

	case HANDSHAKE_NUEVO_NIVEL:{

		uint32_t nro_nivel;
		uint32_t puerto;
		t_stream *payload = get_payload(pedido);
		memcpy(&nro_nivel,&payload[0],4);
		memcpy(&puerto,&payload[4],4);

		t_planificador *nodo_planificador;
		nodo_planificador = get_nodo_nivel(nro_nivel);
		//log_debug(logger_orq, "Hand_nuevo_nivel: dir nodo_planificador = %d", nodo_planificador);
		if(nodo_planificador == NULL){
			nodo_planificador = malloc(sizeof(t_planificador));
			nodo_planificador->tid=malloc(sizeof(pthread_t));
			nodo_planificador->nivel.direccion = malloc(sizeof(struct sockaddr_in));

			log_info(logger_orq, "Handshake_nuevo_nivel: Nueva conexión, nivel %d.", nro_nivel);
			completar_datos_nivel(nro_nivel, cliente, nodo_planificador);

			nodo_planificador->planificador.descriptor = 0;
			(nodo_planificador->nivel.direccion)->sin_port=htons(puerto);
			list_mutex_add(planificadores, (void*) nodo_planificador);
			sem_post(&semaforo_nodos_nuevos);
			free(pedido);
			return EXIT_SUCCESS;
		}
		log_info(logger_orq, "Se reconecta en nivel %d.", nro_nivel);
		completar_datos_nivel(nro_nivel, cliente, nodo_planificador);
		(nodo_planificador->nivel.direccion)->sin_port=htons(puerto);
		free(pedido);
		return EXIT_SUCCESS;
		}

	case HANDSHAKE_NUEVO_PJ:{

		int8_t simbolo_pj;
		t_stream *payload =get_payload(pedido);
		simbolo_pj = (int8_t)payload[0];
		log_info(logger_orq, "Handshake_nuevo_pj: Nueva conexión, pj %c", simbolo_pj);
		t_personaje_finalizado *pj;
		int pos_pj;
		pos_pj = get_pos_pj_finalizado(simbolo_pj);

		//Si todavía no está en la lista lo metemos.
		if(pos_pj == -1){
			log_debug(logger_orq, "Handshake_nuevo_pj: El pj no existía así que lo guardo en listo personajes_finalizados.");
			pj = malloc(sizeof(t_personaje_finalizado));
			pj->simbolo_pj = simbolo_pj;
			pj->termino_plan_de_niveles = false;
			log_debug(logger_orq, "Handshake_nuevo_pj: Guardo: Simbolo = %c  bool_fin = %d", pj->simbolo_pj, pj->termino_plan_de_niveles);
			list_mutex_add(personajes_finalizados, (void*)pj);
			t_personaje_finalizado *test_pj = list_mutex_get(personajes_finalizados, 0);
			log_debug(logger_orq,"Handshake_nuevo_pj: Guardé simbolo = %c, bool_fin = %d ", test_pj->simbolo_pj, test_pj->termino_plan_de_niveles);
		}
		else{
			pthread_mutex_lock(&personajes_finalizados->mutex);
			pj = list_get(personajes_finalizados->list, pos_pj);
			pj->termino_plan_de_niveles = false;
			pthread_mutex_unlock(&personajes_finalizados->mutex);
		}
		t_stream *stream_senial_ok;
		stream_senial_ok  = serializar_senial(OK);
		log_debug(logger_orq, "Handshake_nuevo_pj: Envío OK al pj recién conectado");
		enviar(cliente, stream_senial_ok);
		free(stream_senial_ok);
		free(pedido);
		return EXIT_SUCCESS;


	}

	case SOLICITUD_DATOS_NIVEL:{

		t_stream *payload = get_payload(pedido);
		int8_t simbolo_pj = payload[0];
		int8_t nro_nivel = payload[1];
	    t_datos_nivel datos_nivel;
	    datos_nivel.direccion_nivel[0] = '\0';
	    datos_nivel.direccion_planificador[0]='\0';
	    int resultado = get_datos_nivel(nro_nivel, &datos_nivel);

	    //Si el nivel no existe o si todavía no se lanzó su planificador asociado.
	    if(resultado == EXIT_FAILURE){
	    	char cadena_vacia = 'a';
	        t_stream *stream_nivel_no_existe = serializar_solicitud(NO_EXISTE_EL_NIVEL, sizeof(cadena_vacia), (t_stream*)&cadena_vacia);
//.........这里部分代码省略.........

 virtual void writeDone(uint32_t v) {
   fprintf(stderr, "writeDone %d\n", v);
   sem_post(&write_sem);
 }

static inline void cgi_semgive(void)
{
  sem_post(&g_cgisem);
}

 virtual void bscanGet(uint64_t v) {
     printf("bscanGet: %llx\n", (long long)v);
     sem_post(&sem_bscan);
 }

static int process_callback( jack_nframes_t nframes, void *arg )
{
	int c;
	jack_handle_t* handle = (jack_handle_t*)arg;
	size_t to_read = nframes;

	for(c=0; c<handle->channels; ++c)
		handle->ports_buf[c] =
			jack_port_get_buffer(handle->ports[c], nframes);

	/* One ringbuffer to rule them all, getting interleaved data piecewise
	   and appending to non-interleaved buffers. */
	while(to_read)
	{
		/* Need to read into temporary storage, then deinterleave to JACK
		   buffers. */
		size_t got_piece;
		size_t avail_piece;
		size_t piece = to_read > handle->procbuf_frames
		?	handle->procbuf_frames
		:	to_read;
		/* Ensure we get only full PCM frames by checking available byte count
		   and reducing expectation. */
		avail_piece = jack_ringbuffer_read_space(handle->rb)/handle->framesize;
		got_piece = jack_ringbuffer_read( handle->rb
		,	handle->procbuf, (avail_piece > piece ? piece : avail_piece)
		*	handle->framesize ) / handle->framesize;
		debug2( "fetched %"SIZE_P" frames from ringbuffer (wanted %"SIZE_P")"
		,	(size_p)got_piece, (size_p)piece );
		/* If this is the last piece, fill up, not time to wait. */
		if(to_read > piece)
			piece = got_piece; /* We got further loop cycle(s) to get the rest. */
		else
		{
			if(piece > got_piece)
			{
				debug("filling up with zeros");
				bzero( handle->procbuf+got_piece*handle->framesize
				,	(piece-got_piece)*handle->framesize );
			}
		}
		/* Now extract the pieces for the channels. */
		for (c=0; c < handle->channels; ++c)
		{
			size_t n;
			jack_default_audio_sample_t *dst = handle->ports_buf[c];
			if(handle->encoding == MPG123_ENC_FLOAT_32)
			{
				float* src = (float*)handle->procbuf;
				for(n=0; n<piece; ++n)
					*(dst++) = src[(n*handle->channels)+c];
			}
			else /* MPG123_ENC_FLOAT_64 */
			{
				double* src = (double*)handle->procbuf;
				for(n=0; n<piece; ++n)
					*(dst++) = src[(n*handle->channels)+c];
			}
			/* Store output buffer offset. */
			handle->ports_buf[c] = dst;
		}
		/* Give the writer a hint about the time passed. */
		sem_post(&handle->sem);
		to_read -= piece;
	}
	/* Success*/
	return 0;
}

void threadfunc1()
{
        printf("Hello from threadfunc1\n");
        sleep(5);
        sem_post(&g_sema);
}

static GstClockTime send_chunk (EncoderOutput *encoder_output, RequestData *request_data)
{
    HTTPStreamingPrivateData *priv_data;
    gint64 current_gop_end_addr, tail_addr;
    gint32 ret;
    struct timespec ts;

    priv_data = request_data->priv_data;

    if (clock_gettime (CLOCK_REALTIME, &ts) == -1) {
        GST_ERROR ("send_chunk clock_gettime error: %s", g_strerror (errno));
        return 100 * GST_MSECOND + g_random_int_range (1, 1000000);
    }
    ts.tv_sec += 2;
    while (sem_timedwait (encoder_output->semaphore, &ts) == -1) {
        if (errno == EINTR) {
            continue;
        }
        GST_ERROR ("send_chunk sem_timedwait failure: %s", g_strerror (errno));
        return 100 * GST_MSECOND + g_random_int_range (1, 1000000);
    }
    tail_addr = *(encoder_output->tail_addr);
    current_gop_end_addr = get_current_gop_end (encoder_output, priv_data);

    if (priv_data->send_count == priv_data->chunk_size + priv_data->chunk_size_str_len + 2) {
        /* completly send a chunk, prepare next. */
        priv_data->send_position += priv_data->send_count - priv_data->chunk_size_str_len - 2;
        if (priv_data->send_position == encoder_output->cache_size) {
            priv_data->send_position = 0;
        }
        g_free (priv_data->chunk_size_str);
        priv_data->chunk_size_str_len = 0;
        priv_data->chunk_size = 0;
    }

    if (priv_data->send_position == current_gop_end_addr) {
        /* next gop. */
        priv_data->rap_addr = current_gop_end_addr;
        if (priv_data->send_position + 12 < encoder_output->cache_size) {
            priv_data->send_position += 12;

        } else {
            priv_data->send_position = priv_data->send_position + 12 - encoder_output->cache_size;
        }
        current_gop_end_addr = get_current_gop_end (encoder_output, priv_data);
    }
    sem_post (encoder_output->semaphore);

    if (priv_data->chunk_size == 0) {
        if (current_gop_end_addr == -1) {
            /* current output gop. */
            if ((tail_addr - priv_data->send_position) > 16384) {
                priv_data->chunk_size = 16384;

            } else if (tail_addr > priv_data->send_position) {
                /* send to tail. */
                priv_data->chunk_size = tail_addr - priv_data->send_position;

            } else if (tail_addr == priv_data->send_position) {
                /* no data available, wait a while. */
                return 100 * GST_MSECOND + g_random_int_range (1, 1000000);

            } else if ((encoder_output->cache_size - priv_data->send_position) > 16384) {
                priv_data->chunk_size = 16384;

            } else {
                priv_data->chunk_size = encoder_output->cache_size - priv_data->send_position;
            }

        } else {
            /* completely output gop. */
            if ((current_gop_end_addr - priv_data->send_position) > 16384) {
                priv_data->chunk_size = 16384;

            } else if (current_gop_end_addr > priv_data->send_position) {
                /* send to gop end. */
                priv_data->chunk_size = current_gop_end_addr - priv_data->send_position;

            } else if (current_gop_end_addr == priv_data->send_position) {
                /* no data available, wait a while. */
                return 100 * GST_MSECOND + g_random_int_range (1, 1000000); //FIXME FIXME

            } else {
                /* send to cache end. */
                priv_data->chunk_size = encoder_output->cache_size - priv_data->send_position;
            }
        }
        priv_data->chunk_size_str = g_strdup_printf ("%x\r\n", priv_data->chunk_size);
        priv_data->chunk_size_str_len = strlen (priv_data->chunk_size_str);
        priv_data->send_count = 0;
    }

    /* send data. */
    ret = send_data (encoder_output, request_data);
    if (ret == -1) {
        return GST_CLOCK_TIME_NONE;

    } else {
        priv_data->send_count += ret;
        request_data->bytes_send += ret;
//.........这里部分代码省略.........

int main(void)
{
	int ret, status;
	pid_t child, ctl;
	pthread_t th;

	output_init();

	ctl = getpid();

	/* Initialize the semaphore */
	sem = sem_open("/fork_21_1", O_CREAT, O_RDWR, 0);

	if (sem == SEM_FAILED)
		UNRESOLVED(errno, "Failed to open the semaphore");

	sem_unlink("/fork_21_1");

	/* Create thread */
	ret = pthread_create(&th, NULL, threaded, &ctl);

	if (ret != 0)
		UNRESOLVED(ret, "Failed to create the thread");

	/* Create the child */
	child = fork();

	if (child == -1)
		UNRESOLVED(errno, "Failed to fork");

	/* We post the semaphore twice */
	do {
		ret = sem_post(sem);
	} while (ret != 0 && errno == EINTR);

	if (ret != 0)
		UNRESOLVED(errno, "Failed to post the semaphore");

	/* child */
	if (child == 0) {
		/* sleep a little while to let the thread execute in case it exists */
		sleep(1);

		/* We're done */
		exit(PTS_PASS);
	}

	/* Parent joins the child */
	ctl = waitpid(child, &status, 0);

	if (ctl != child)
		UNRESOLVED(errno, "Waitpid returned the wrong PID");

	if (!WIFEXITED(status) || WEXITSTATUS(status) != PTS_PASS)
		FAILED("Child exited abnormally");

	/* Destroy everything */
	ret = sem_close(sem);

	if (ret != 0)
		UNRESOLVED(errno, "Failed to close the semaphore");

	ret = pthread_join(th, NULL);

	if (ret != 0)
		UNRESOLVED(ret, "Failed to join the thread in parent");

#if VERBOSE > 0
	output("Test passed\n");
#endif
	PASSED;
}

void stdout_stats_and_post_sem_cb(struct rb_mse_api *rb_mse, struct rb_mse_stats *stats, void *_sem)
{
	sem_t * sem = _sem;
	stdout_stats_cb(rb_mse,stats,NULL);
	sem_post(sem);
}

static int stream_transfer(struct stream_transfer *stream_transfer)
{

	struct dev_stream *stream_sender;
	struct dev_stream *stream_receiver;
	int size_transfer = 0;
	int ret   =0;
	int exit_flag   =0;
	int i   =0;
	short* Srcptr;
	short* Drcptr;

	stream_sender = stream_transfer->stream_sender;
	stream_receiver = stream_transfer->stream_receiver;
	size_transfer = stream_sender->buf_size;


#ifdef  START_ZERO_BUFFER
	/* 消除开头杂音 */
	memset(stream_sender->buf, 0, stream_sender->buf_size);
	pcm_write(stream_receiver->dev, stream_sender->buf, stream_sender->buf_size);
#endif

	while( 1 ){
		if ( (!stream_transfer->voice_thread_run_flag) || (exit_flag == 1)){
			break;	
		}

		ret = pcm_read(stream_sender->dev, stream_sender->buf, size_transfer);
		if (ret != 0) {
			exit_flag = 1;
			ALOGE("err: read codec err:%s, ret=%d", strerror(errno), ret);
			break;
		}

		if ( (!stream_transfer->voice_thread_run_flag) || (exit_flag == 1)){
			break;	
		}

		ret = pcm_write(stream_receiver->dev, stream_sender->buf, size_transfer);
		if (ret != 0) {
			exit_flag = 1;
			ALOGE("err: write pcm err:%s, ret=%d", strerror(errno), ret);
		}

		if ( (!stream_transfer->voice_thread_run_flag) || (exit_flag == 1)){
			break;	
		}

		if (stream_transfer->record_flag == 1){
			//是上行,还是下行.
			if (stream_transfer->voice_direction == UPSTREAM){
				Srcptr = (short*)(stream_sender->buf);
				Drcptr = (short*)(record_data.record_buf + (record_data.lenwriteup%record_data.record_lenth));
				if(record_data.lenwriteup >= record_data.lenwritedown)
				{
					memcpy(Drcptr,Srcptr,size_transfer);
				}
				else
				{
					int i;
					for(i=0;i<size_transfer/2;i++,Drcptr++)
					{
						*Drcptr = (*Drcptr + *Srcptr++)/2;
					}
					record_data.lenwrite += size_transfer;
//				        sem_post(&sem_record);
				}
				record_data.lenwriteup += size_transfer;
				//ALOGD("stream is upload");
			} else {
				Srcptr = (short*)(stream_sender->buf);
				Drcptr = (short*)(record_data.record_buf + (record_data.lenwritedown%record_data.record_lenth));
				if(record_data.lenwritedown >= record_data.lenwriteup)
				{
					memcpy(Drcptr,Srcptr,size_transfer);
				}
				else
				{

					for(i=0;i<size_transfer/2;i++,Drcptr++)
					{
						*Drcptr = ((int)*Drcptr + (int)(*Srcptr++))/2;
					}
					record_data.lenwrite += size_transfer;
//				        sem_post(&sem_record);
				}
				record_data.lenwritedown += size_transfer;
				//ALOGD("stream is download");
			}	
		        sem_post(&sem_record);
		}

		//ALOGD("pcm running ... , type=%d ",stream_sender->type);
		if ( (!stream_transfer->voice_thread_run_flag) || (exit_flag == 1)){
			break;	
		}
	}
	return 0;
}