streams.c

Go to the documentation of this file.

#include "streams.h"
 
 
typedef struct stbuf {
  int   tn;
  int   fd;
  int   size;
  byte  buf[300];
} _stbuf;
 
static struct stbuf streambuf[10];
 
 
struct stream streams[2];
static byte streams_enabled;
static struct stream *get_stream(int fd);
static void sendfail(int fd, int tn, long sender, int err_no, const char *err);
static int get_max_open_streams() {
  return sizeof(streams) / sizeof(streams[0]);
}
static char errbuf[256]; // an error buffer, a bit like like errstr()
void stream_init() {
  errbuf[0] = 0;
  memset(&streams, 0, sizeof(streams));
  memset(&streambuf, 0, sizeof(streambuf));
  stream_enable();
}
static int are_streams_enabled() {
  return streams_enabled ? 1 : 0;
}
void stream_disable() {
  streams_enabled = 0;
  printf("Streams disabled\r\n");
}
void stream_enable() {
  streams_enabled = 1;
  printf("Streams enabled\r\n");
}
 
 
static void clear_tns(struct stream *stream) {
  int elements = sizeof(streambuf) / sizeof(streambuf[0]);
  struct stbuf *sb;
  int i;
 
  for (i = 0; i < elements; i++) {
    sb = &streambuf[i];
    if (sb->fd != stream->fd) {
      continue;
    }
    sb->fd = 0;
  }
}
static struct stbuf *get_tn(struct stream *stream, int tn) {
  int elements = sizeof(streambuf) / sizeof(streambuf[0]);
  struct stbuf *sb;
  int i;
 
  for (i = 0; i < elements; i++) {
    sb = &streambuf[i];
    if ((sb->fd == stream->fd) && (sb->tn == tn)) {
      return sb;
    }
  }
  return NULL;
}
 
static void save_tn(struct stream *stream, int tn, const byte *buf, const int size) {
  if (size > sizeof(streambuf[0].buf)) {
    return; // too big to cache
  }
  if (get_tn(stream, tn)) {
    return; // already saved
  }
  int elements = sizeof(streambuf) / sizeof(streambuf[0]);
  struct stbuf *sb;
  int i;
 
  for (i = 0; i < elements; i++) {
    sb = &streambuf[i];
    if ((sb->fd == 0)
        || ((sb->fd == stream->fd) && (sb->tn <= stream->ackedtn))
        ) {
      // use it
      sb->fd = stream->fd;
      sb->tn = tn;
      sb->size = size;
      for (i = 0; i < size; i++) {
        sb->buf[i] = buf[i];
      }
      return;
    }
  }
}
 
int stream_associate(int fd, const struct consumerinfo *ci, byte *buf, int bufsize, int packetsize) {
  struct stream *st;
 
  st = get_stream(fd);
  if (st == NULL) {
    return 1;
  }
  if (packetsize > bufsize) {
    return 2;
  }
  memset(buf, 0, bufsize);
  st->lastack = 0;
  st->ack_open = 0;
  st->inactivity_ack_ctr = 0;
  st->lasttn = 0;
  st->failed = 0;
  st->bytesinbuf = 0;
  st->consumer = ci;
  st->bufsize = bufsize;
  st->packetsize = packetsize;
  st->last_activity = mculib_get_seconds_since_boot();
  if (bufsize == 1) {
    st->bytebuf = &st->onebytebuf;
  } else {
    st->bytebuf = buf;
  }
  return 0;
}
 
struct stream *alloc_stream() {
  int i;
  long now = mculib_get_seconds_since_boot();
 
  for (i = 0; i < get_max_open_streams(); i++) {
    if (streams[i].fd == 0) {
      memset(&streams[i], 0, sizeof(streams[i]));
      streams[i].fd = ((i + 1) * (mculib_get_seconds_since_boot() & 0xFFF) + 1);
      // some defaults:
      streams[i].ackfreq = 4;
      streams[i].lastack = now;
      streams[i].last_activity = now;
      streams[i].target[0] = 0;
      return &streams[i];
    }
  }
  return NULL;
}
 
static struct stream *get_stream(int fd) {
  int i;
 
  if (fd == 0) {
    return NULL;
  }
 
  for (i = 0; i < get_max_open_streams(); i++) {
    if (streams[i].fd == fd) {
      return &streams[i];
    }
  }
  return NULL;
}
static struct stream *get_stream_by_target(const char *target) {
  int i;
 
  if ((target == NULL) || (strlen(target) == 0) || (strlen(target) > 16)) {
    return NULL;
  }
  for (i = 0; i < get_max_open_streams(); i++) {
    if (strcmp(target, streams[i].target) == 0) {
      printf("Found stream by target %s: fd=%i\r\n", target, streams[i].fd);
      return &streams[i];
    }
  }
  return NULL;
}
 
void close_stream(int fd, int reason) {
  struct stream *st;
 
  st = get_stream(fd);
  if (st == NULL) {
    printf("Warning attempt to close stream #%i which is not open atm.\r\n", fd);
    return;
  }
  printf("Closing stream %i, reason %i\r\n", fd, reason);
  clear_tns(st);
  if ((st->consumer != NULL) && (st->consumer->close_consumer != NULL)) {
    st->consumer->close_consumer(fd, reason);
  }
 
  memset(st, 0, sizeof(streams[0]));
  printf("Stream #%i closed.\r\n", fd);
}
 
static void repeat_sendack(struct stream *stream) {
  struct command *com;
 
  if (!are_streams_enabled()) {
    return;
  }
 
  com = alloc_command();
  if (com == NULL) {
    printf("No memory to allocate command to send ack\r\n");
    return;
  }
  command_init(com);
  com->target = stream->peer;
  com->encoding = 'a';
  com->com = 29; // streamcontrol
  com->flags = COMFLAGS_SUCCESS | COMFLAGS_ACK;
  command_add_arg(com, "fd=%i", stream->fd);
  command_add_arg(com, "acktn=%i", stream->lasttn);
  send_command(com);
  free_command(com);
}
static void sendack(struct stream *stream) {
  repeat_sendack(stream);
  stream->ackedtn = stream->lasttn;
  stream->lastack = mculib_get_seconds_since_boot();
  stream->ack_open = 1;
}
static void sendfail(int fd, int tn, long sender, int err_no, const char *err) {
  struct command *com;
 
  com = alloc_command();
  if (com == NULL) {
    printf("No memory to allocate command to send ack\r\n");
    return;
  }
  struct stream *st = get_stream(fd);
 
  if (st != NULL) {
    command_init(com);
    com->target = sender;
    com->encoding = 'a';
    com->com = 29;
    command_add_arg(com, "fd=%i", fd);
    command_add_arg(com, "fc=%i", st->failed);
    if (st->failed != 0) {
      command_add_arg(com, "msg=%s", errbuf);
    }
    send_command(com);
    free_command(com);
    com = alloc_command();
    if (com == NULL) {
      printf("No memory to allocate command to send ack\r\n");
      return;
    }
  }
  command_init(com);
  com->target = sender;
  com->encoding = 'a';
  com->com = 29;
  command_add_arg(com, "fd=%i", fd);
  command_add_arg(com, "tn=%i", tn);
  command_add_arg(com, "errno=%i", err_no);
  command_add_arg(com, "err=%s", err);
  if (st != NULL) {
    command_add_arg(com, "fc=%i", st->failed);
    if (st->failed != 0) {
      command_add_arg(com, "msg=%s", errbuf);
    }
  }
  send_command(com);
  free_command(com);
}
 
void streamsetup(struct command *com) {
  struct stream *st = NULL;
  struct command *data;
 
  data = NULL;
 
  if (!are_streams_enabled()) {
    return;
  }
  const char *tgt = namedarg(com, "tgt"); // get the unique target (idempotency - if a stream with that same tgt (target) is open already, return that instead)
 
  if (tgt != NULL) {
    if (strlen(tgt) >= 16) {
      printf("stream target too long\r\n");
      return;
    }
    st = get_stream_by_target(tgt);
  }
  if (st == NULL) {
    st = alloc_stream();
    if (st == NULL) {
      if (namedarg(com, "streaminit") != NULL) {
        stream_init();
      }
      st = alloc_stream();
    }
  }
  if (st == NULL) {
    printf("No more available streams!\r\n");
    goto fail;
  }
  if ((tgt != NULL)) {
    strncpy(st->target, tgt, 16);
  }
  st->peer = com->sender;
  data = get_data_reply(com);
  if (data == NULL) {
    goto fail;
  }
  command_add_arg(data, "fd=%i", st->fd);
  const char *reqack = namedarg(com, "ackfreq");
 
  if (reqack != NULL) {
    int ra = atoi(reqack);
    if (ra == 0) {
      goto fail;
    }
    st->ackfreq = ra;
  }
  command_add_arg(data, "ackfreq=%i", st->ackfreq);
  if (send_command(data) != 0) {
    goto fail;
  }
 
  if (data != NULL) {
    free_command(data);
  }
  send_command_reply(com, COMFLAGS_ACK | COMFLAGS_SUCCESS);
  return;
fail:
  if (data != NULL) {
    free_command(data);
  }
  if (st != NULL) {
    close_stream(st->fd, 1);
  }
  send_command_reply(com, COMFLAGS_ACK);
}
static int call_consumer(struct stream *stream, byte *newdata, int size) {
  int i;
  int z;
 
  // this should NEVER happen, but if it does bail out big time
  if (stream->bytesinbuf >= stream->packetsize) {
    stream->bytesinbuf = 0;
    return 109;
  }
  // stream bytes in packets to consumer...
  for (i = 0; i < size; i++) {
    stream->bytebuf[stream->bytesinbuf++] = newdata[i];
    if (stream->bytesinbuf >= stream->packetsize) {
      int s = stream->bytesinbuf;
      stream->bytesinbuf = 0;
      if ((z = stream->consumer->data_consumer(stream->fd, stream->bytebuf, s)) != 0) {
        return z;
      }
    }
  }
  return 0;
}
void flush(struct stream *stream) {
  // call consumer one last time with the remaining data
  int fd = stream->fd;
  int s = stream->bytesinbuf;
  int tn = stream->ackedtn;
 
  printf("Stream: flush()\r\n");
  stream->bytesinbuf = 0;
  int z;
 
  if (s != 0) {
    if ((z = stream->consumer->data_consumer(stream->fd, stream->bytebuf, s)) != 0) {
      printf("\r\nError: Stream #%i processing failure code %i\r\n", fd, z);
      stream->failed = z;
      sendfail(fd, tn, stream->peer, z, "Stream processing failure");
      return;
    }
  }
}
void check_close(struct stream *stream) {
  printf("Check_close: ackedtn: %i, closetn: %i\r\n", stream->ackedtn, stream->closetn);
  if (!stream->closetn) {
    return; // no close packet received yet
  }
  if (stream->ackedtn == (stream->closetn - 1)) {
    flush(stream);
    stream->lasttn = stream->closetn;
    int i = 0;
    for (i = 0; i < 5; i++) {
      sendack(stream);
      Delay(5);
    }
    close_stream(stream->fd, 0);
  }
}
 
void streamdata(struct command *com) {
  const int data_arg = 3;
  const char *para;
  struct stream *stream;
  int i;
  int len = -1;
  int tn = 0;
  int fd = 0;
 
  if (!are_streams_enabled()) {
    return;
  }
  if ((para = namedarg(com, "fd")) != NULL) {
    fd = atoi(para);
  }
  if ((para = namedarg(com, "ln")) != NULL) {
    len = atoi(para);
  }
  if ((para = namedarg(com, "tn")) != NULL) {
    tn = atoi(para);
  }
 
  stream = get_stream(fd);
  if (stream == NULL) {
    printf("\r\nError: Stream #%i not open\r\n", fd);
    sendfail(fd, tn, com->sender, 1, "Stream not open");
    return;
  }
  if ((para = namedarg(com, "request")) != NULL) {
    if (strcmp("close", para) == 0) {
      printf("Closing stream due to peer's request\r\n");
      sendack(stream);
      stream->closetn = tn;
      check_close(stream);
      return;
    } else if (strcmp("syncreq", para) == 0) {
      printf("Sync requested\r\n");
      repeat_sendack(stream);
      return;
    } else {
      printf("\r\nError: Invalid stream request %s\r\n", para);
      sendfail(fd, tn, com->sender, 5, "invalid stream request");
    }
  }
 
  if (len != get_arg_size(com, data_arg)) {
    printf("\r\nWarn: length mismatch sent=%i vs received=%i\r\n", len, get_arg_size(com, data_arg));
    // ignore error, wait for resent
    //sendfail(fd, tn, com->sender, 6, "length mismatch");
    return;
  }
 
  if (stream->failed) {
    sendfail(fd, tn, com->sender, stream->failed, "Stream processing failure.");
    return;
  }
  if (com->argctr < 4) {
    printf("\r\nError: Not enough parameters %i\r\n", com->argctr);
    sendfail(fd, tn, com->sender, 2, "Not enough parameters for stream data");
    return;
  }
  if (stream->consumer == NULL) {
    printf("\r\nError: Stream #%i not associated with consumer\r\n", fd);
    sendfail(fd, tn, com->sender, 3, "Stream not associated with consumer");
    return;
  }
  if (stream->consumer->data_consumer == NULL) {
    printf("\r\nError: Stream #%i associated with invalid consumer\r\n", fd);
    sendfail(fd, tn, com->sender, 3, "Stream association invalid");
    return;
  }
  if (tn < (stream->lasttn + 1)) {
    printf("Warning: #%i received, but we're ahead at %i\r\n", tn, stream->lasttn);
    stream->outoforderctr++;
    if (stream->outoforderctr > 5) {
      stream->outoforderctr = 0;
      repeat_sendack(stream);
    }
    // ignore we had this before...
    return;
  }
  byte *cbuf = (byte *)get_arg(com, data_arg);
 
  if (tn > (stream->lasttn + 1)) {
    // too large
    printf("Out of sync: #%i received, but we're still at %i\r\n", tn, stream->lasttn);
    stream->outoforderctr++;
    save_tn(stream, tn, cbuf, len);
    if (stream->outoforderctr > 5) {
      stream->outoforderctr = 0;
      repeat_sendack(stream);
    }
    return;
  }
  stream->outoforderctr = 0;
  // got valid and good data!
  //printf("Good Streamdata received: fd=%i, ln=%i, tn=%i, p=%p\r\n", fd, len, tn, stream);
 
  i = call_consumer(stream, cbuf, len);
  if (i != 0) {
    printf("\r\nError: Stream #%i processing failure code %i\r\n", fd, i);
    stream->failed = i;
    sendfail(fd, tn, com->sender, i, "Stream processing failure");
    return;
  }
  stream->lasttn = tn;
  tn++;
 
  // do we have more in cache?
  struct stbuf *st;
 
  while ((st = get_tn(stream, tn))) {
    printf("Using cached tn %i\r\n", tn);
    cbuf = (byte *)&st->buf;
    len = st->size;
    i = call_consumer(stream, cbuf, len);
    if (i != 0) {
      printf("\r\nError: Stream #%i processing failure code %i\r\n", fd, i);
      stream->failed = i;
      sendfail(fd, tn, com->sender, i, "Stream processing failure");
      return;
    }
    tn++;
  }
  tn--; // last one was not available!
  stream->lasttn = tn;
  stream->ack_open = 0;
  stream->last_activity = mculib_get_seconds_since_boot();
  stream->inactivity_ack_ctr = 0;
  //  if ((stream->lasttn - stream->ackedtn) >= stream->ackfreq) {
  sendack(stream);
  //}
  check_close(stream);
  return;
}
 
void stream_loop() {
  int i;
  struct stream *st;
 
  if (!are_streams_enabled()) {
    return;
  }
  long now = mculib_get_seconds_since_boot();
 
  for (i = 0; i < get_max_open_streams(); i++) {
    st = &streams[i];
    if (st->fd == 0) {
      continue;
    }
    long diff = now - st->lastack;
    long diffms = systick_diff(st->lastackms);
    if (
      ((st->inactivity_ack_ctr < 2) && (st->ack_open) && (diff >= 2))
      || ((st->inactivity_ack_ctr < 5) && (diffms > 10))
      ) {
      st->lastackms = get_systickctr();
      st->inactivity_ack_ctr++;
      //printf("Stream %i stalled? sending ack for %i...\r\n", st->fd,st-lastack);
      //sendack(st);
    }
    if ((now - st->last_activity) > 600) {
      printf("Closing stream #%i due to inactivity\r\n", st->fd);
      close_stream(st->fd, 3);
    }
  }
}
int count_open_streams() {
  int res = 0;
  int i;
  struct stream *st;
 
  for (i = 0; i < get_max_open_streams(); i++) {
    st = &streams[i];
    if (st->fd == 0) {
      continue;
    }
    res++;
  }
  return res;
}
 
void print_stream_info() {
  int i;
  long now = mculib_get_seconds_since_boot();
  struct stream *st;
 
  printf("Stream info (at %D):\r\n", now);
  for (i = 0; i < get_max_open_streams(); i++) {
    st = &streams[i];
    if (st->fd == 0) {
      continue;
    }
    printf("Stream #%i\r\n", st->fd);
    printf("   Ack Frequency:  %i\r\n", st->ackfreq);
    printf("   Ack Open:       %i\r\n", st->ack_open);
    printf("   Inactivity ctr: %i\r\n", st->inactivity_ack_ctr);
    printf("   acked tn:       %i\r\n", st->ackedtn);
    printf("   last tn:        %i\r\n", st->lasttn);
    printf("   peer:           %N\r\n", st->peer);
    printf("   lastack secs:   %i\r\n", st->lastack);
    printf("   last activity:  %i\r\n", st->last_activity);
    printf("   byte consumer:  %s\r\n", (st->consumer == NULL) ? "NOT SET" : "SET");
    printf("   packetsize:     %i\r\n", st->packetsize);
    printf("   bufsize:        %i\r\n", st->bufsize);
    printf("   bytes in buf:   %i\r\n", st->bytesinbuf);
    printf("   closing:        %i\r\n", st->closetn);
    printf("   target:         %s\r\n", st->target);
  }
  printf("End Stream info\r\n");
}
 
void stream_seterror(int streamfd, const char *format, ...) {
  char *buf = &errbuf[0];
  size_t size = sizeof(errbuf);
  struct stream *st = get_stream(streamfd);
  va_list args;
 
  va_start(args, format);
  vsnprintf(buf, size, format, args);
  va_end(args);
  if (st == NULL) {
    return;
  }
  if (st->failed == 0) {
    st->failed = 1;
  }
  printf("stream failed: %s\r\n", errbuf);
}