#include <aio.h>
#include <stdio.h>
#include <fcntl.h>
#include <errno.h>
#include <unistd.h>
#include <stdarg.h>
#include <stdlib.h>
#include <sys/ioctl.h>
#include <sys/socket.h>
#include <sys/audioio.h>

extern const char *__progname;

#include "config.h"

#include "cdif.h"
#include "blocker.h"
#include "stdioutil.h"
#include "scm-rights.h"
#include "stringgetter.h"

#include "player.h"

/*
 * Amount we tack on to allow data chunk size to grow if possible.
 */
#define DATA_CHUNK_EXCESS 256

/*
 * Size of past-data-chunk window we keep.
 */
#define DATA_CHUNK_HISTORY 10

/*
 * This number is 44100 * 2 * 2 / 75, that is,
 * bytes/frame = samples/sec * channels/sample * bytes/channel / frames/sec
 */
#define FRAME_BYTES 2352

/*
 * The type of a PQE.  PQET_TRACK is used for PQEs representing tracks,
 *  which is almost all PQEs.  PQET_FLUSH exists to handle
 *  M2P_FLUSH_PENDING messages.
 *
 * M2P_FLUSH_PENDING is a request to clear out everything after the end
 *  of the track that's currently playing.  Most of the time, there are
 *  no track boundaries in oq_audio, so doing this means waiting until
 *  the next end-of-track from the data connection, then aborting the
 *  data connection.  But if there is a track boundary in oq_audio,
 *  this means flushing the data connection immediately and, upon
 *  hitting that track boundary (the first such, if there are
 *  multiple), flushing everything then remaining in oq_audio.  (We
 *  actually don't have to flush the data connection immediately in the
 *  latter case, but it improves performance in that it means we can
 *  have the data connection recreated and data ready to read out of it
 *  once we get to the track boundary in oq_audio.)
 *
 * A PQET_FLUSH PQE exists to handle the first kind.  Such things exist
 *  only in datapend, and even there only at the head.
 *
 * Handling the second case is done with audio_specials_queued (to
 *  detect it) and audio_flushing (to indicate we are waiting to hit
 *  that special in oq_audio).
 */
typedef enum {
	  PQET_TRACK = 1,
	  PQET_FLUSH,
	  } PQETYPE;

typedef struct playqueue PLAYQUEUE;
typedef struct pqe PQE;

struct playqueue {
  PQE *head;
  PQE **tail;
  } ;

struct pqe {
  PQE *link;
  PQETYPE type;
  union {
    struct {
      char *cdname;
      int begin; /* start of track */
      int start; /* start playing here */
      int length;
      } track;
    PQE *flush;
    } ;
  } ;

int playerfd;
int playerdg;

static int audiofd;
static PLAYQUEUE playq;
static AIO_OQ oq_audio;
static int audio_specials_queued;
static int audio_flushing;
static int audio_nulpad;
static int audio_wrote;
static int open_audio_id;
static int write_audio_id;
static int play_id;
static AIO_OQ oq_main;
static int write_main_id;
static int (*process_master_input)(const void *, int);
static STRINGGETTER *pmi_sg;
static char *pmi_cdname;
static int pmi_startframe;
static int pmi_nframes;
static int pmi_next_count;
static int pmi_nb;
static int paused;
static int dataconn;
#define DC_NONE    (-1)
#define DC_PENDING (-2)
#define DC_FAILED  (-3)
static AIO_OQ oq_datareq;
static int datareq_id;
static int data_chunk_size;
static int data_chunk_history[DATA_CHUNK_HISTORY];
static int data_chunk_histhand;
static int data_chunk_histmax;
static int (*dataprocess)(const void *, int);
static DEBLOCKER *data_err_db;
static PLAYQUEUE datapend;
static int data_left;

static void plog(const char *, ...)
	__attribute__((__format__(__printf__,1,2),__used__));
static void plog(const char *fmt, ...)
{
 va_list ap;
 char *s;
 int sn;
 char *ts;
 int tsn;
 static int fd = -2;
 struct timeval tv;
 time_t tt;
 struct tm *tm;
 struct iovec iov[2];

 if (! write_logs) return;
 if (fd == -2) fd = open("cdif-player.log",O_TRUNC|O_WRONLY|O_APPEND|O_CREAT,0666);
 if (fd < 0) return;
 gettimeofday(&tv,0);
 tt = tv.tv_sec;
 tm = localtime(&tt);
 tsn = asprintf( &ts,"%04d-%02d-%02d %02d:%02d:%02d.%06d ",
	tm->tm_year + 1900,
	tm->tm_mon + 1,
	tm->tm_mday,
	tm->tm_hour,
	tm->tm_min,
	tm->tm_sec,
	(int)tv.tv_usec );
 va_start(ap,fmt);
 sn = vasprintf(&s,fmt,ap);
 va_end(ap);
 iov[0].iov_base = ts;
 iov[0].iov_len = tsn;
 iov[1].iov_base = s;
 iov[1].iov_len = sn;
 writev(fd,&iov[0],2);
 free(ts);
 free(s);
}

static void send_msg_head(unsigned char opc)
{
 aio_oq_queue_copy(&oq_main,&opc,1);
}

static void tell_user(const char *, ...)
	__attribute__((__format__(__printf__,1,2)));
static void tell_user(const char *fmt, ...)
{
 va_list ap;
 char *s;
 int l;

 va_start(ap,fmt);
 l = vasprintf(&s,fmt,ap);
 va_end(ap);
 send_msg_head(PLAYER_P2M_MESSAGE);
 aio_oq_queue_free(&oq_main,s,l+1);
 plog("tell_user: %.*s\n",l,s);
}

static void playqueue_init(PLAYQUEUE *pq)
{
 pq->head = 0;
 pq->tail = &pq->head;
}

static PQE *playqueue_dequeue(PLAYQUEUE *q)
{
 PQE *e;

 e = q->head;
 if (e)
  { q->head = e->link;
    if (! q->head) q->tail = &q->head;
  }
 return(e);
}

static void playqueue_append(PLAYQUEUE *q, PQE *e)
{
 e->link = 0;
 *q->tail = e;
 q->tail = &e->link;
}

static void pqe_free(PQE *e)
{
 if (! e) return;
 switch (e->type)
  { default:
       break;
    case PQET_TRACK:
       free(e->track.cdname);
       break;
    case PQET_FLUSH:
       pqe_free(e->flush);
       break;
  }
 free(e);
}

static int wtest_audio(void *arg __attribute__((__unused__)))
{
 return(audio_nulpad||(!paused&&aio_oq_nonempty(&oq_audio)));
}

static void update_master(void)
{
 int foff;

 foff = audio_wrote / FRAME_BYTES;
 if (datapend.head && (datapend.head->type == PQET_TRACK)) foff += datapend.head->track.start - datapend.head->track.begin;
 if (aio_oq_qlen(&oq_main) > 65536)
  { plog("update_master: skipping sending %d, queue too backlogged\n",foff);
  }
 else
  { send_msg_head(PLAYER_P2M_PLAYSTATUS);
    aio_oq_queue_copy(&oq_main,&foff,sizeof(int));
    plog("update_master: sent %d\n",foff);
  }
}

static void track_ended(PQE *e)
{
 if (e)
  { if (e->type != PQET_TRACK) panic();
    plog("track_ended: end of %s %u/%u %u\n",e->track.cdname,e->track.begin,e->track.start,e->track.length);
  }
 else
  { plog("track_ended: end of live-when-flushed track\n");
  }
 send_msg_head(PLAYER_P2M_PLAYDONE);
 audio_wrote = 0;
 update_master();
 pqe_free(e);
}

static void audio_special_justfree(void *v, int i __attribute__((__unused__)))
{
 pqe_free(v);
}
    
static int wr_audio_special(void *v, int i __attribute__((__unused__)))
{
 track_ended(v);
 if (audio_flushing)
  { aio_oq_flush_special(&oq_audio,&audio_special_justfree);
    audio_flushing = 0;
  }
 audio_specials_queued --;
 return(0);
}

static void wr_audio(void *arg __attribute__((__unused__)))
{
 int w;

 if (audiofd < 0) panic();
 if (audio_nulpad)
  { w = write(audiofd,"\0\0\0",audio_nulpad);
    if (w < 0)
     { switch (errno)
	{ case EINTR:
	  case EWOULDBLOCK:
	     return;
	     break;
	}
       fprintf(stderr,"%s: audio write: %s\n",__progname,strerror(errno));
       exit(1);
     }
    audio_nulpad -= w;
    if (audio_nulpad) return;
  }
 if (paused) return;
 w = aio_oq_writev(&oq_audio,audiofd,&wr_audio_special);
 if (w < 0)
  { switch (errno)
     { case EINTR:
       case EWOULDBLOCK:
	  return;
	  break;
     }
    fprintf(stderr,"%s: audio write: %s\n",__progname,strerror(errno));
    exit(1);
  }
 audio_wrote += w;
 plog("wr_audio wrote %d\n",w);
 aio_oq_dropdata(&oq_audio,w);
 update_master();
}

static int try_open_audio(void *arg __attribute__((__unused__)))
{
 int e;
 struct audio_info ai;

 aio_remove_block(open_audio_id);
 open_audio_id = AIO_PL_NOID;
 audiofd = open(AUDIO_DEV,O_RDWR,0);
 if (audiofd < 0)
  { e = errno;
    plog("try_open_audio: open failed: %s\n",strerror(e));
    send_msg_head(PLAYER_P2M_AUDIO_FAIL_OPEN);
    aio_oq_queue_copy(&oq_main,&e,sizeof(int));
  }
 else
  { AUDIO_INITINFO(&ai);
    ai.play.sample_rate = 44100;
    ai.play.channels = 2;
    ai.play.precision = 16;
    ai.play.encoding = AUDIO_ENCODING_SLINEAR_BE;
    ai.play.pause = 0;
    ai.record.pause = 1;
    ai.mode = AUMODE_PLAY_ALL;
    if (ioctl(audiofd,AUDIO_SETINFO,&ai) < 0)
     { e = errno;
       plog("try_open_audio: setup failed: %s\n",strerror(e));
       send_msg_head(PLAYER_P2M_AUDIO_FAIL_SETUP);
       aio_oq_queue_copy(&oq_main,&e,sizeof(int));
       close(audiofd);
       audiofd = -1;
     }
    else
     { plog("try_open_audio: all good\n");
       send_msg_head(PLAYER_P2M_AUDIO_OPEN);
       set_nonblock(audiofd);
       if (write_audio_id != AIO_PL_NOID) panic();
       write_audio_id = aio_add_poll(audiofd,&aio_rwtest_never,&wtest_audio,0,&wr_audio,0);
     }
  }
 return(AIO_BLOCK_LOOP);
}

static void audio_up(void)
{
 if (audiofd >= 0)
  { plog("audio_up: short-circuiting\n");
    send_msg_head(PLAYER_P2M_AUDIO_OPEN);
    return;
  }
 if (open_audio_id == AIO_PL_NOID)
  { plog("audio_up: opening\n");
    open_audio_id = aio_add_block(&try_open_audio,0);
  }
 else
  { plog("audio_up: already in progress\n");
  }
}

static void audio_down(void)
{
 if (open_audio_id != AIO_PL_NOID)
  { plog("audio_down: in progress, cancelling\n");
    aio_remove_block(open_audio_id);
    open_audio_id = AIO_PL_NOID;
  }
 if (audiofd >= 0)
  { plog("audio_down: closing\n");
    if (write_audio_id == AIO_PL_NOID) panic();
    aio_remove_poll(write_audio_id);
    write_audio_id = AIO_PL_NOID;
    close(audiofd);
    audiofd = -1;
  }
 else
  { plog("audio_down: already closed\n");
  }
 send_msg_head(PLAYER_P2M_AUDIO_CLOSED);
}

static int wtest_main(void *arg __attribute__((__unused__)))
{
 return(aio_oq_nonempty(&oq_main));
}

static void rd_main(void *arg __attribute__((__unused__)))
{
 unsigned char buf[512];
 int r;
 int n;
 int o;

 r = read(playerfd,&buf[0],sizeof(buf));
 if (r < 0)
  { switch (errno)
     { case EINTR:
       case EWOULDBLOCK:
	  return;
	  break;
     }
    fprintf(stderr,"%s: player<-master read: %s\n",__progname,strerror(errno));
    exit(1);
  }
 if (r == 0)
  { plog("rd_main: read EOF, exiting\n");
    exit(0);
  }
 plog("rd_main: read %d\n",r);
 o = 0;
 while (o < r)
  { n = (*process_master_input)(&buf[o],r-o);
    o += n;
  }
}

static void wr_main(void *arg __attribute__((__unused__)))
{
 int w;

 w = aio_oq_writev(&oq_main,playerfd,0);
 if (w < 0)
  { switch (errno)
     { case EINTR:
       case EWOULDBLOCK:
	  return;
	  break;
     }
    fprintf(stderr,"%s: player->master write: %s\n",__progname,strerror(errno));
    exit(1);
  }
 plog("wr_main: wrote %d\n",w);
 aio_oq_dropdata(&oq_main,w);
}

static int pmi_pkt_begin(const void *, int); /* forward */

static int pmi_int(int *ip, const void *data, int len, void (*done)(void))
{
 int n;

 n = sizeof(int) - pmi_nb;
 if (n > len) n = len;
 bcopy(data,((char *)ip)+pmi_nb,n);
 pmi_nb += n;
 if (pmi_nb >= sizeof(int)) (*done)();
 return(n);
}

static int play_w(void *cookie __attribute__((__unused__)), const char *data, int len)
{
 plog("play_w: queueing %d\n",len);
 aio_oq_queue_copy(&oq_datareq,data,len);
 return(len);
}

static int player(void *arg __attribute__((__unused__)))
{
 PQE *e;
 FILE *af;
 FILE *bf;
 int len;

 if (paused || !playq.head || (aio_oq_qlen(&oq_datareq) >= 32768)) return(AIO_BLOCK_NIL);
 switch (dataconn)
  { case DC_NONE:
       send_msg_head(PLAYER_P2M_WANT_DATA_CONN);
       dataconn = DC_PENDING;
       return(AIO_BLOCK_LOOP);
       break;
    case DC_PENDING:
    case DC_FAILED:
       return(AIO_BLOCK_NIL);
       break;
  }
 if (dataconn < 0) panic();
 af = funopen(0,0,&play_w,0,0);
 while (playq.head && (aio_oq_qlen(&oq_datareq) < 65536))
  { e = playqueue_dequeue(&playq);
    if (e->type != PQET_TRACK) panic();
    len = e->track.length - (e->track.start - e->track.begin);
    plog("player: requesting %s %u %u\n",e->track.cdname,e->track.start,len);
    bf = fwrap_w_blocked(af);
    fprintf(bf,"%s%c%u%c%u%c",e->track.cdname,'\0',e->track.start,'\0',len,'\0');
    fclose(bf);
    playqueue_append(&datapend,e);
  }
 fclose(af);
 return(AIO_BLOCK_LOOP);
}

static void pmi_play_queueit(void)
{
 PQE *qe;

 process_master_input = &pmi_pkt_begin;
 qe = malloc(sizeof(PQE));
 qe->type = PQET_TRACK;
 qe->track.cdname = pmi_cdname;
 qe->track.begin = pmi_startframe;
 qe->track.start = qe->track.begin;
 qe->track.length = pmi_nframes;
 playqueue_append(&playq,qe);
 plog("pmi_play_queueit: queued %s %u %u\n",qe->track.cdname,qe->track.start,qe->track.length);
}

static int pmi_play_count(const void *data, int len)
{
 return(pmi_int(&pmi_nframes,data,len,&pmi_play_queueit));
}

static void pmi_play_get_count(void)
{
 process_master_input = &pmi_play_count;
 pmi_nb = 0;
}

static int pmi_play_start(const void *data, int len)
{
 return(pmi_int(&pmi_startframe,data,len,&pmi_play_get_count));
}

static void play_have_cdname(STRINGGOT s, void *arg __attribute__((__unused__)))
{
 pmi_cdname = malloc(s.len+1);
 bcopy(s.str,pmi_cdname,s.len);
 pmi_cdname[s.len] = '\0';
 process_master_input = &pmi_play_start;
 pmi_nb = 0;
}

static int pmi_play_cdname(const void *data, int len)
{
 int n;

 for (n=0;(n<len)&&(process_master_input==&pmi_play_cdname);n++)
  { sg_put(pmi_sg,((const unsigned char *)data)[n]);
  }
 return(n);
}

static void set_paused(int p)
{
 plog("set_paused\n");
 paused = p;
 send_msg_head(paused?PLAYER_P2M_PAUSED:PLAYER_P2M_UNPAUSED);
}

static void put_back_playqueue(PLAYQUEUE *q, PLAYQUEUE *to)
{
 if (q->head)
  { if (! to->head) to->tail = q->tail;
    *q->tail = to->head;
    to->head = q->head;
    q->head = 0;
    q->tail = &q->head;
  }
}

static void lost_dataconn(void)
{
 plog("lost_dataconn\n");
 if (datareq_id == AIO_PL_NOID) panic();
 aio_remove_poll(datareq_id);
 datareq_id = AIO_PL_NOID;
 if (datapend.head && (datapend.head->type == PQET_FLUSH))
  { pqe_free(playqueue_dequeue(&datapend));
  }
 put_back_playqueue(&datapend,&playq);
 aio_oq_flush(&oq_datareq);
 close(dataconn);
 dataconn = DC_NONE;
}

static int data_errmsg(const void *data, int len)
{
 int t;
 int n;

 t = 0;
 while ((dataprocess == &data_errmsg) && (t < len) && (dataconn >= 0))
  { n = deblocker_push1(data_err_db,t+(const char *)data,len-t);
    if (n < 0) return(len);
    t += n;
  }
 return(t);
}

static void new_track_data(void)
{
 dataprocess = &data_errmsg;
}

static int data_bulk_data(const void *data, int len)
{
 int n;
 PQE *e;
 PQE *f;

 if (data_left < 1) panic();
 n = data_left;
 if (n > len) n = len;
 aio_oq_queue_copy(&oq_audio,data,n);
 plog("data_bulk_data: len %d, data_left %d, consumed %d, data_left now %d\n",len,data_left,n,data_left-n);
 data_left -= n;
 if (data_left < 1)
  { e = playqueue_dequeue(&datapend);
    switch (e->type)
     { default:
	  panic();
	  break;
       case PQET_TRACK:
	  aio_oq_queue_special(&oq_audio,e,0);
	  audio_specials_queued ++;
	  new_track_data();
	  break;
       case PQET_FLUSH:
	  f = e->flush;
	  lost_dataconn();
	  aio_oq_queue_special(&oq_audio,f,0);
	  audio_specials_queued ++;
	  e->flush = 0;
	  pqe_free(e);
	  return(len);
	  break;
     }
  }
 return(n);
}

static int rtest_data(void *arg __attribute__((__unused__)))
{
 return(!paused&&(aio_oq_qlen(&oq_audio)<1048576));
}

static int wtest_data(void *arg __attribute__((__unused__)))
{
 return(aio_oq_nonempty(&oq_datareq));
}

static void rd_data(void *arg __attribute__((__unused__)))
{
 int s;
 int r;
 unsigned char *data;
 int o;
 int n;
 int i;

 s = data_chunk_size + DATA_CHUNK_EXCESS;
 data = malloc(s);
 r = read(dataconn,&data[0],s);
 if (r < 0)
  { switch (errno)
     { case EINTR:
       case EWOULDBLOCK:
	  return;
	  break;
     }
    tell_user("Data connection read: %s",strerror(errno));
    lost_dataconn();
    return;
  }
 if (r == 0)
  { tell_user("Data connection EOF");
    lost_dataconn();
    return;
  }
 if (data_chunk_history[data_chunk_histhand] == data_chunk_histmax)
  { int m;
    data_chunk_history[data_chunk_histhand] = 0;
    m = 0;
    for (i=DATA_CHUNK_HISTORY-1;i>=0;i--)
     { if (data_chunk_history[i] > m) m = data_chunk_history[i];
       if (m == data_chunk_histmax) break;
     }
    data_chunk_histmax = m;
  }
 data_chunk_history[data_chunk_histhand] = r;
 if (r > data_chunk_histmax) data_chunk_histmax = r;
 data_chunk_histhand = (data_chunk_histhand ? : DATA_CHUNK_HISTORY) - 1;
 data_chunk_size = data_chunk_histmax;
 plog("rd_data: tried %d, read %d, max %d, chunk size now %d\n",s,r,data_chunk_histmax,data_chunk_size);
 if ( (dataprocess == &data_bulk_data) &&
      (r < data_left) )
  { aio_oq_queue_free(&oq_audio,data,r);
    data_left -= r;
    plog("rd_data: optimizing data_bulk_data, data_left now %d\n",data_left);
    return;
  }
 o = 0;
 while (o < r)
  { plog("rd_data: calling dataprocess, r=%d, o=%d\n",r,o);
    n = (*dataprocess)(data+o,r-o);
    plog("rd_data: dataprocess returned %d\n",n);
    o += n;
  }
 free(data);
}

static void wr_data(void *arg __attribute__((__unused__)))
{
 int w;

 w = aio_oq_writev(&oq_datareq,dataconn,0);
 if (w < 0)
  { switch (errno)
     { case EINTR:
       case EWOULDBLOCK:
	  return;
	  break;
     }
    fprintf(stderr,"%s: data connection write: %s\n",__progname,strerror(errno));
    exit(1);
  }
 plog("wr_data: wrote %d\n",w);
 aio_oq_dropdata(&oq_datareq,w);
}

static void data_got_errmsg(const void *msg, int msglen, void *arg __attribute__((__unused__)))
{
 PQE *e;

 if (msglen > 0)
  { tell_user("Data fetch error: %.*s",msglen,(const char *)msg);
    e = playqueue_dequeue(&datapend);
    if (! e) panic();
    switch (e->type)
     { default:
	  panic();
	  break;
       case PQET_TRACK:
	  plog("data_got_errmsg: %s %u %u\n",e->track.cdname,e->track.start,e->track.length);
	  plog("data_got_errmsg: error: %.*s\n",msglen,(const char *)msg);
	  send_msg_head(PLAYER_P2M_PLAYDONE);
	  break;
     }
    pqe_free(e);
  }
 else
  { if (! datapend.head) panic();
    switch (datapend.head->type)
     { default:
	  panic();
	  break;
       case PQET_TRACK:
	  data_left = datapend.head->track.length * FRAME_BYTES;
	  plog("data_got_errmsg: no error for %s %u %u, data_left now %d\n",
		datapend.head->track.cdname,
		datapend.head->track.start,
		datapend.head->track.length,
		data_left );
	  dataprocess = &data_bulk_data;
	  break;
     }
  }
}

static void data_proto_err(void *arg __attribute__((__unused__)), const char *fmt, ...)
{
 va_list ap;
 char *s;

 va_start(ap,fmt);
 vasprintf(&s,fmt,ap);
 va_end(ap);
 tell_user("Data protocol error: %s",s);
 free(s);
 lost_dataconn();
 dataconn = DC_FAILED;
}

static void receive_data_conn(void)
{
 struct msghdr mh;
 struct cmsghdr cmh;
 struct iovec iov;
 unsigned char ctlbuf[CMSPACE(sizeof(int))];
 unsigned char junk;
 int fd;

 iov.iov_base = &junk;
 iov.iov_len = 1;
 mh.msg_name = 0;
 mh.msg_namelen = 0;
 mh.msg_iov = &iov;
 mh.msg_iovlen = 1;
 mh.msg_control = (void *) &ctlbuf[0];
 mh.msg_controllen = CMSPACE(sizeof(int));
 mh.msg_flags = 0;
 if (recvmsg(playerdg,&mh,0) < 0)
  { fprintf(stderr,"%s: data conn recvmsg: %s\n",__progname,strerror(errno));
    exit(1);
  }
 if (mh.msg_controllen < sizeof(struct cmsghdr))
  { fprintf(stderr,"%s: data conn: control length (%d) too small\n",__progname,(int)mh.msg_controllen);
    exit(1);
  }
 bcopy(&ctlbuf[0],&cmh,sizeof(struct cmsghdr));
 if ((cmh.cmsg_level != SOL_SOCKET) || (cmh.cmsg_type != SCM_RIGHTS))
  { fprintf(stderr,"%s: data conn: level/type wrong\n",__progname);
    exit(1);
  }
 if (cmh.cmsg_len < CMLEN(sizeof(int)))
  { fprintf(stderr,"%s: data conn: cmsg_len (%d) too short\n",__progname,(int)cmh.cmsg_len);
    exit(1);
  }
 bcopy(&ctlbuf[CMSKIP(&cmh)],&fd,sizeof(int));
 if (fd < 0)
  { fprintf(stderr,"%s: received data conn %d < 0\n",__progname,fd);
    exit(1);
  }
 dataconn = fd;
 plog("received dataconn, fd %d\n",fd);
 if (datareq_id != AIO_PL_NOID) panic();
 datareq_id = aio_add_poll(dataconn,&rtest_data,&wtest_data,&rd_data,&wr_data,0);
 new_track_data();
}

static void skip_next(int count)
{
 int v;
 PQE *e;

 if (audio_wrote % 4)
  { v = audio_nulpad;
    audio_nulpad += 4 - (audio_wrote % 4);
    audio_nulpad &= 3;
    plog("skip_next: audio_nulpad changed from %d to %d\n",v,audio_nulpad);
  }
 if (audio_flushing)
  { aio_oq_flush_special(&oq_audio,&audio_special_justfree);
    audio_specials_queued = 0;
    audio_flushing = 0;
  }
 else
  { while <"flushq"> (1)
     { v = aio_oq_headlen(&oq_audio);
       switch (v)
	{ case AIO_OQ_HL_EMPTY:
	     plog("skip_next: audio queue now empty\n");
	     break <"flushq">;
	  case AIO_OQ_HL_SPECIAL:
	     plog("skip_next: dropping special\n");
	     aio_oq_writev(&oq_audio,-1,&wr_audio_special);
	     if (--count < 1) return;
	     break;
	  default:
	     if (v < 0) panic();
	     plog("skip_next: dropping %d data\n",v);
	     audio_wrote += v;
	     aio_oq_dropdata(&oq_audio,v);
	     break;
	}
     }
  }
 if (datapend.head && (datapend.head->type == PQET_FLUSH))
  { pqe_free(playqueue_dequeue(&datapend));
  }
 while (count > 0)
  { e = playqueue_dequeue(&datapend);
    if (! e) e = playqueue_dequeue(&playq);
    if (e) track_ended(e);
    /*
     * Don't do anything here if !e; this can happen if an
     *	M2P_PLAY_NEXT crosses in transit with a P2M_PLAYDONE.
     */
    count --;
  }
 if (dataconn >= 0) lost_dataconn();
}

static void do_skip_next(void)
{
 process_master_input = &pmi_pkt_begin;
 plog("do_skip_next: count = %d\n",pmi_next_count);
 if (pmi_next_count < 1) return;
 skip_next(pmi_next_count);
}

static int pmi_play_next_count(const void *data, int len)
{
 return(pmi_int(&pmi_next_count,data,len,&do_skip_next));
}

static void seek_to(int frameoff)
{
 PQE *nqe;
 PQE *oqe;
 PLAYQUEUE fpq;

 plog("seek_to: offset = %d\n",frameoff);
 nqe = 0;
 if (datapend.head)
  { plog("seek_to: have datapend\n");
    oqe = datapend.head;
    if (oqe->type == PQET_FLUSH)
     { plog("seek_to: have FLUSH\n");
       oqe = oqe->link;
     }
    if (oqe)
     { if (oqe->type != PQET_TRACK) panic();
       plog("seek_to: copying datapend head\n");
       nqe = malloc(sizeof(PQE));
       *nqe = *oqe;
       nqe->track.cdname = strdup(oqe->track.cdname);
     }
    plog("seek_to: calling skip_next(1)\n");
    skip_next(1);
  }
 if (nqe)
  { plog("seek_to: pushing back new PQE\n");
    fpq.head = nqe;
    fpq.tail = &nqe->link;
    nqe->link = 0;
    put_back_playqueue(&fpq,&playq);
  }
 if (playq.head)
  { plog("seek_to: adjusting playq head\n");
    oqe = playq.head;
    if (frameoff < 0) frameoff = 0;
    if (frameoff >= oqe->track.length) frameoff = oqe->track.length - 1;
    oqe->track.start = oqe->track.begin + frameoff;
  }
 else
  { plog("seek_to: no playq - nothing to do\n");
  }
}

static void do_seek_to(void)
{
 process_master_input = &pmi_pkt_begin;
 seek_to(pmi_nframes);
}

static int pmi_seek_offset(const void *data, int len)
{
 return(pmi_int(&pmi_nframes,data,len,&do_seek_to));
}

static void flush_pending(void)
{
 PQE *e;
 PQE *dph;

 dph = datapend.head ? playqueue_dequeue(&datapend) : 0;
 while (datapend.head) pqe_free(playqueue_dequeue(&datapend));
 while (playq.head) pqe_free(playqueue_dequeue(&playq));
 if (audio_specials_queued)
  { audio_flushing = 1;
    if (dph) pqe_free(dph);
    lost_dataconn();
  }
 else if (dph)
  { e = malloc(sizeof(PQE));
    e->type = PQET_FLUSH;
    e->flush = dph;
    playqueue_append(&datapend,e);
  }
 send_msg_head(PLAYER_P2M_FLUSHDONE);
}

static int pmi_pkt_begin(const void *data, int len)
{
 if (len < 1) panic();
 switch (*(const unsigned char *)data)
  { case PLAYER_M2P_PLAY:
       plog("PLAYER_M2P_PLAY\n");
       pmi_sg = sg_init_push(&play_have_cdname,&sg_term_nul,0);
       process_master_input = &pmi_play_cdname;
       break;
    case PLAYER_M2P_DATA_CONN:
       plog("PLAYER_M2P_DATA_CONN\n");
       if (dataconn != DC_PENDING) panic();
       receive_data_conn();
       break;
    case PLAYER_M2P_DATA_CONN_FAIL:
       plog("PLAYER_M2P_DATA_CONN_FAIL\n");
       if (dataconn != DC_PENDING) panic();
       dataconn = DC_FAILED;
       break;
    case PLAYER_M2P_OPEN_AUDIO:
       plog("PLAYER_M2P_OPEN_AUDIO\n");
       audio_up();
       break;
    case PLAYER_M2P_CLOSE_AUDIO:
       plog("PLAYER_M2P_CLOSE_AUDIO\n");
       audio_down();
       break;
    case PLAYER_M2P_PAUSE:
       plog("PLAYER_M2P_PAUSE\n");
       set_paused(1);
       break;
    case PLAYER_M2P_UNPAUSE:
       plog("PLAYER_M2P_UNPAUSE\n");
       set_paused(0);
       break;
    case PLAYER_M2P_DATA_RETRY:
       plog("PLAYER_M2P_DATA_RETRY\n");
       if (dataconn == DC_FAILED) dataconn = DC_NONE;
       break;
    case PLAYER_M2P_PLAY_NEXT:
       plog("PLAYER_M2P_PLAY_NEXT\n");
       process_master_input = &pmi_play_next_count;
       pmi_nb = 0;
       break;
    case PLAYER_M2P_FLUSH_PENDING:
       plog("PLAYER_M2P_FLUSH_PENDING\n");
       flush_pending();
       break;
    case PLAYER_M2P_SEEK:
       plog("PLAYER_M2P_SEEK\n");
       process_master_input = &pmi_seek_offset;
       pmi_nb = 0;
       break;
    default:
       panic();
       break;
  }
 return(1);
}

static void setup(void)
{
 int i;

 set_nonblock(playerfd);
 set_nonblock(playerdg);
 audiofd = -1;
 dataconn = DC_NONE;
 aio_oq_init(&oq_datareq);
 datareq_id = AIO_PL_NOID;
 paused = 0;
 audio_wrote = 0;
 open_audio_id = AIO_PL_NOID;
 play_id = aio_add_block(&player,0);
 aio_oq_init(&oq_main);
 aio_oq_init(&oq_audio);
 audio_specials_queued = 0;
 audio_flushing = 0;
 audio_nulpad = 0;
 playqueue_init(&playq);
 playqueue_init(&datapend);
 audio_up();
 write_main_id = aio_add_poll(playerfd,&aio_rwtest_always,&wtest_main,&rd_main,&wr_main,0);
 write_audio_id = AIO_PL_NOID;
 process_master_input = &pmi_pkt_begin;
 data_err_db = deblocker_open(&data_got_errmsg,&data_proto_err,0);
 data_chunk_size = 8192;
 data_chunk_histhand = 0;
 for (i=DATA_CHUNK_HISTORY-1;i>=0;i--) data_chunk_history[i] = 8192;
 data_chunk_histmax = 8192;
 plog("startup done\n");
}

void player_main(void)
{
 setproctitle("player [%d]",getppid());
 setup();
 while (1)
  { aio_pre_poll();
    if (aio_do_poll() < 0)
     { if (errno == EINTR) continue;
       fprintf(stderr,"%s: poll: %s\n",__progname,strerror(errno));
       exit(1);
     }
    aio_post_poll();
  }
}